Diffuser le CDC PostgreSQL avec NATS : architecture temps réel

La plupart des pipelines CDC Postgres passent par pgoutput, le protocole de réplication logique binaire, puis par un connecteur comme Debezium qui le traduit vers Kafka. Aurabase saute cette étape : le service aura-realtime décode directement le WAL en réplication logique avec le plugin wal2json, qui produit du JSON exploitable sans étage de traduction intermédiaire.

-- Créé une fois si absent, recréé si le WAL a expiré
SELECT pg_create_logical_replication_slot(
  'aura_cdc_slot', 'wal2json');
-- Poll répété (backoff 100ms quand le lot revient vide)
SELECT data FROM pg_logical_slot_get_changes(
  'aura_cdc_slot', NULL, 1000,
  'include-xids', '0', 'format-version', '2',
  'include-schemas', '1', 'include-types', '0');

Trois prérequis Postgres, vérifiés dans le code : le rôle applicatif doit porter REPLICATION, wal_level = logical doit être actif, et max_slot_wal_keep_size doit borner la rétention WAL. Sans cette dernière borne, un consommateur lent fait grossir le volume disque indéfiniment. Le service surveille aussi wal_status du slot : s'il passe à lost (WAL purgé au-delà de cette limite), le slot est recréé automatiquement. La perte des événements non consommés entre-temps est assumée, et loguée.

Un lot de 1000 changements par poll, avec un filtre de tables dynamique rafraîchi toutes les 3 secondes. Seules les tables où un projet a explicitement activé le temps réel entrent dans add-tables — ça évite de décoder le WAL de tables qui n'intéressent aucun abonné.

Un slot de réplication logique Postgres ne tolère qu'un seul lecteur actif à la fois. Faire tourner plusieurs consommateurs du même aura_cdc_slot en parallèle romprait l'ordre des changements, pas seulement les dupliquerait. Aurabase a scindé aura-realtime en deux binaires distincts pour résoudre cette contrainte sans sacrifier le scale horizontal du WebSocket.

# cdc-worker : leader-élu, lance le CDC + publie sur NATS. Pas de serveur WS.
[[bin]]
name = "aura-realtime-cdc-worker"
# ws-front : N réplicas (HPA), consomme NATS, filtre RLS, sert WS/SSE.
[[bin]]
name = "aura-realtime-ws-front"

cdc-worker ne gagne le droit de lire le slot qu'en détenant un objet Lease Kubernetes (coordination.k8s.io/v1). Il tente de le créer, ou de le voler s'il a expiré, puis le renouvelle au tiers de sa durée de vie. À la perte du lease (renew échoué, holder changé), un CancellationToken coupe le travail en cours et le processus retourne en boucle d'acquisition. Hors Kubernetes — en docker-compose local, par exemple — le client K8s échoue à se connecter, et le service bascule en mode "toujours leader". Utile en dev, incorrect si on l'oublie en prod avec plusieurs réplicas.

pub async fn run_with_lease<F, Fut>(cfg: LeaseConfig, work: F) -> anyhow::Result<()>
where F: Fn(CancellationToken) -> Fut, ...
{
  // renouvelle toutes les lease_duration_secs / 3
  // hors K8s : work(token) une seule fois, token jamais annulé
}

C'est le point le plus souvent mal compris dans ce genre de pipeline : NATS et JetStream sont deux choses distinctes, et les événements CDC ne transitent pas par un stream JetStream persistant. cdc-worker publie chaque événement avec Client::publish_with_headers — l'API pub/sub cœur de NATS, celle livrée at-most-once sans persistance ni relecture, distincte de JetStream.

// cdc-worker publie (ChannelManager::publish_and_relay)
nc.publish_with_headers(subject, headers, payload).await
// ws-front écoute tout le sous-arbre (NatsRelay::start_relay)
client.subscribe("aura.realtime.>").await

Chaque réplica ws-front souscrit au sous-arbre aura.realtime.>, décode le sujet vers le nom de canal d'origine. Il republie ensuite l'événement sur son tokio::sync::broadcast local, pour les WebSocket qui y sont connectées. Un en-tête Aura-Origin porte l'identifiant de l'instance émettrice : chaque réplica ignore les messages qu'il a lui-même publiés, ce qui évite les doublons sans coordination centrale.

Ce choix a une conséquence directe : le NATS cœur livre au mieux une fois (at-most-once), sans file d'attente durable. Si un réplica ws-front est brièvement déconnecté de NATS au moment où un événement passe, il ne le rattrape pas. L'historique par canal (Channel.history, un buffer en mémoire borné par HISTORY_SIZE) vit localement à chaque réplica, pas au niveau du cluster. C'est un compromis assumable précisément parce que le slot de réplication logique Postgres reste la source de vérité durable en amont. C'est wal2json qui garantit qu'aucun changement DB n'est perdu avant consommation, pas NATS.

JetStream, lui, existe bien dans aura-realtime — mais pour un usage entièrement différent. Il sert le KV store de présence cross-instance (aura_presence, stockage mémoire, max_age de 60 secondes), qui synchronise qui est en ligne sur quel canal entre tous les réplicas ws-front. Un état partagé de courte durée, pas un flux d'événements CDC à rejouer.

Un événement CDC ne part pas brut vers tous les abonnés d'un canal. Le poller route chaque changement vers l'un de CDC_NUM_SHARDS workers (4 par défaut, hashés par project_id), qui interroge d'abord pg_policies. Une table sans policy RLS autorise tous les abonnés — le comportement Supabase — tandis qu'une table avec policies déclenche une vérification par abonné.

BEGIN;
SET LOCAL ROLE "aura_authenticated";
-- claims JWT du subscriber injectés en GUC
SELECT set_config('request.jwt.claims', $1, true);
SELECT EXISTS(SELECT 1 FROM "schema"."table" WHERE id::text = $2);
ROLLBACK;

La transaction endosse le rôle Postgres physique de l'abonné et injecte ses claims JWT dans request.jwt.claims — celui que lisent auth.uid() et auth.role() côté policy. Elle vérifie ensuite que la ligne reste visible sous ce rôle, puis s'annule : aucune écriture, une lecture RLS réelle. Seuls les sub_id validés atterrissent dans allowed_sub_ids de l'événement ; un vec vide signifie personne, en fail-closed.

Un détail qui piège souvent en migration : wal2json ne capture le contenu complet d'un UPDATE ou d'un DELETE que si la table porte REPLICA IDENTITY FULL. Sans elle, seuls les INSERT remontent. C'est pour ça que l'endpoint qui active le temps réel sur une table pose cette ALTER TABLE automatiquement — un correctif réel du dépôt, pas une case à cocher à part.

L'historique du dépôt documente un vrai piège, pas un cas théorique. L'image Postgres utilisée par défaut sur un cluster Kubernetes (l'image communautaire pgvector/pgvector stock) n'embarque pas le plugin wal2json. Sans lui, pg_create_logical_replication_slot('aura_cdc_slot', 'wal2json') échoue, et rien côté client ne le signale explicitement. Les WebSocket restent ouvertes, les abonnements sont acceptés, mais aucun événement DB n'arrive jamais.

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
      postgresql-16-wal2json=2.6-4.pgdg12+1 \
      ...

Le correctif réel a consisté à construire et publier une image Postgres dédiée avec ce paquet installé, puis à pointer le StatefulSet Kubernetes dessus au lieu de l'image stock. Il a aussi posé explicitement wal_level=logical, max_replication_slots et max_slot_wal_keep_size en arguments de démarrage — absents des images génériques.

C'est précisément pour rendre ce genre de panne observable que cdc-worker expose un drapeau cdc_active (faux tant que le slot n'est pas confirmé sain) sur son endpoint /health dédié. C'est un signal binaire à surveiller — plutôt que de déduire un CDC mort de la seule absence d'événements côté client.

Ce mécanisme lit une seule variable POSTGRES_REPLICATION_URL, donc un seul serveur Postgres et un seul slot logique. C'est cohérent avec le modèle multi-tenant à schéma par projet (project_<uuid>) sur cluster partagé. Sur ce modèle, tout fonctionne : un seul cdc-worker voit passer les changements de tous les projets d'un même cluster, et route par schéma.

Pour un projet sur un cluster CNPG dédié — sa propre instance Postgres isolée — l'image utilisée aujourd'hui (construite à partir de l'image stock CloudNativePG) n'ajoute que pg_graphql, pas wal2json. Rien, non plus, ne fait tourner un cdc-worker par cluster dédié. En l'état actuel du code, le CDC temps réel décrit ici reste donc scopé au tier partagé — un boundary d'architecture réel, pas une fonctionnalité en feuille de route.

Tout ce mécanisme reste invisible depuis le SDK. L'API chainable d'aurabase-js ne change pas selon que l'événement vient du cdc-worker via NATS ou, en dev local, du binaire aura-realtime non scindé.

const channel = aura.realtime.channel('room-1')
  .on('postgres_changes', {
    event: 'INSERT', schema: 'public', table: 'messages'
  }, (payload) => { /* … */ })
  .subscribe()

Mais avant qu'un seul événement puisse arriver, la table doit être inscrite au CDC — ce n'est pas automatique à la création. Un appel service_role sur l'endpoint dédié (ou le toggle équivalent dans Studio) écrit la table dans realtime_tables du schéma platform du projet, et pose REPLICA IDENTITY FULL pour vous.

curl -X PUT \
  https://api.aurabase.cloud/v1/db/<project_id>/schema/tables/messages/realtime \
  -H "apikey: <service_role_key>" \
  -H 'Content-Type: application/json' \
  -d '{"enabled": true}'

Pour le reste de la surface SDK compatible Supabase — auth, storage, policies RLS —, voir le guide de migration. Pour la façon dont aura-realtime se découpe en plusieurs binaires dans un seul crate du workspace, voir le détail de l'architecture Cargo.