10 - Outbox Pattern: Pubblicazione Atomica di Eventi con CDC

Salvare l'ordine e annunciarlo: due sistemi, nessuna transazione

Un servizio salva un ordine nel database e deve pubblicare OrderPlaced su un broker. Sono due scritture su due sistemi, e nessuna transazione le copre insieme. Se scrivi prima il database e poi il broker, un guasto nel mezzo lascia un ordine di cui nessuno saprà mai nulla.

Se scrivi prima il broker, un guasto nel mezzo annuncia un ordine che non esiste. Riprovare non risolve: il problema, detto dual write, è strutturale. Il pattern outbox lo aggira scrivendo l'evento in una tabella dello stesso database, nella stessa transazione dell'ordine, e lasciando a un processo separato, il relay, il compito di spedirlo al broker.

L'atomicità la dà il database, che ce l'ha già; il prezzo è che il relay può spedire un evento due volte. Eseguo il meccanismo su SQLite con un broker finto che puoi far cadere, provo prima le due versioni ingenue, poi l'outbox, poi un relay che muore nel punto peggiore, e riporto dalla documentazione di Debezium come si fa senza polling. Mi servono i concetti del consumatore idempotente, perché la consegna ripetuta è inevitabile, e dei fondamenti degli eventi.

Le tre versioni, eseguite

Il broker finto ha un interruttore giu e un registro di ciò che ha davvero ricevuto. L'ordine ha un vincolo CHECK (total >= 0), che permette di far fallire la scrittura nel database a comando con un totale negativo. I casi A e B sono le due versioni ingenue; C è l'outbox con la scrittura che fallisce; da D a H sono il relay.

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE orders (id TEXT PRIMARY KEY, total INTEGER NOT NULL CHECK (total >= 0));
  CREATE TABLE outbox (id TEXT PRIMARY KEY, aggregate_id TEXT NOT NULL, event_type TEXT NOT NULL,
                       payload TEXT NOT NULL, published_at TEXT);
`);

// Un broker finto: puo' essere "giu'" e tiene il registro di cio' che ha davvero ricevuto.
const broker = { giu: false, ricevuti: [] as string[], publish(m: string) { if (this.giu) throw new Error('broker non raggiungibile'); this.ricevuti.push(m); } };
const ordini = () => (db.prepare('SELECT COUNT(*) AS n FROM orders').get() as { n: number }).n;

// --- Doppia scrittura ingenua: database e broker non condividono una transazione.
function ordineIngenuoDbPoiBroker(id: string, totale: number): void {
  db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, totale);
  broker.publish(`OrderPlaced:${id}`);                  // se fallisce, l'ordine esiste e nessuno lo sa
}
function ordineIngenuoBrokerPoiDb(id: string, totale: number): void {
  broker.publish(`OrderPlaced:${id}`);
  db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, totale);   // se fallisce, l'evento descrive un ordine che non esiste
}
broker.giu = true;
try { ordineIngenuoDbPoiBroker('o-1', 100); } catch (e) { console.log('A) db poi broker, broker giu: ordini nel db =', ordini(), '| eventi =', broker.ricevuti.length); }
broker.giu = false;
try { ordineIngenuoBrokerPoiDb('o-2', -5); } catch (e) { console.log('B) broker poi db, totale non valido: ordini nel db =', ordini(), '| eventi =', broker.ricevuti.length, broker.ricevuti); }

// --- Outbox: ordine ed evento nella stessa transazione.
function ordineConOutbox(id: string, eventId: string, totale = 100): void {
  db.exec('BEGIN');
  try {
    db.prepare('INSERT INTO orders VALUES (?, ?)').run(id, totale);
    db.prepare('INSERT INTO outbox (id, aggregate_id, event_type, payload) VALUES (?, ?, ?, ?)').run(eventId, id, 'OrderPlaced', JSON.stringify({ orderId: id }));
    db.exec('COMMIT');
  } catch (e) { db.exec('ROLLBACK'); throw e; }
}
ordineConOutbox('o-3', 'evt-3');
try { ordineConOutbox('o-6', 'evt-6', -5); } catch { console.log('C) totale non valido: rollback, eventi in outbox =', (db.prepare('SELECT COUNT(*) AS n FROM outbox').get() as { n: number }).n, '(solo quello di o-3)'); }
ordineConOutbox('o-4', 'evt-4');

// --- Relay: legge i non pubblicati in ordine, pubblica, poi marca. Puo' morire fra le due azioni.
function relay(morireDopoPublish = false): void {
  const righe = db.prepare('SELECT id, aggregate_id, event_type FROM outbox WHERE published_at IS NULL ORDER BY rowid LIMIT 100').all() as
    { id: string; aggregate_id: string; event_type: string }[];
  for (const r of righe) {
    broker.publish(`${r.event_type}:${r.aggregate_id}#${r.id}`);
    if (morireDopoPublish) throw new Error('relay morto dopo la pubblicazione, prima di marcare');
    db.prepare("UPDATE outbox SET published_at = 'ora' WHERE id = ?").run(r.id);
  }
}
broker.ricevuti.length = 0;
broker.giu = true;
try { relay(); } catch { console.log('D) broker giu: eventi consegnati =', broker.ricevuti.length, '| non pubblicati in outbox =', (db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE published_at IS NULL').get() as { n: number }).n); }
broker.giu = false;
try { relay(true); } catch { console.log('E) relay morto dopo il primo publish: consegnati =', broker.ricevuti.length); }
relay();
console.log('F) dopo il riavvio del relay: consegnati =', broker.ricevuti.length, broker.ricevuti);

// Il consumatore scarta le copie usando l'id dell'evento.
const visti = new Set<string>();
const effettivi = broker.ricevuti.filter((m) => { const id = m.split('#')[1]; if (visti.has(id)) return false; visti.add(id); return true; });
console.log('G) effetti dopo la deduplicazione per id evento =', effettivi.length);

// Pulizia: si cancellano solo le righe pubblicate.
const cancellate = db.prepare("DELETE FROM outbox WHERE published_at IS NOT NULL").run().changes;
console.log('H) righe di outbox pulite =', cancellate, '| rimaste =', (db.prepare('SELECT COUNT(*) AS n FROM outbox').get() as { n: number }).n);

// Output:
// A) db poi broker, broker giu: ordini nel db = 1 | eventi = 0
// B) broker poi db, totale non valido: ordini nel db = 1 | eventi = 1 [ 'OrderPlaced:o-2' ]
// C) totale non valido: rollback, eventi in outbox = 1 (solo quello di o-3)
// D) broker giu: eventi consegnati = 0 | non pubblicati in outbox = 2
// E) relay morto dopo il primo publish: consegnati = 1
// F) dopo il riavvio del relay: consegnati = 3 [
//   'OrderPlaced:o-3#evt-3',
//   'OrderPlaced:o-3#evt-3',
//   'OrderPlaced:o-4#evt-4'
// ]
// G) effetti dopo la deduplicazione per id evento = 2
// H) righe di outbox pulite = 2 | rimaste = 0

Il caso A mostra l'evento perso: l'ordine o-1 c'è nel database, nessun evento è stato pubblicato. Il caso B mostra l'evento fantasma: OrderPlaced:o-2 è nel broker, ma gli ordini nel database sono ancora uno. Nel caso C la scrittura dell'ordine fallisce, la transazione annulla anche l'evento, e l'outbox contiene solo quello di o-3: non esiste un evento senza il suo ordine.

Nel caso D il broker è giù, il relay non consegna niente e i due eventi restano nell'outbox con published_at nullo, quindi nessuna perdita. Il caso E è il momento peggiore: il relay pubblica e muore prima di marcare la riga; al riavvio (F) la ripubblica, e il broker ha tre messaggi di cui due identici. Il caso G mostra il rimedio, il consumatore che scarta le copie per id evento e ottiene due effetti. Nel caso H si cancellano solo le righe già pubblicate.

Il relay a polling

Il relay più semplice è un ciclo: legge le righe non pubblicate in ordine di inserimento, le pubblica, le marca. È facile da capire e da far girare senza altri componenti, ma ha due costi: una query continua sul database e una latenza pari all'intervallo di polling. Quando i relay sono più d'uno, due istanze non devono prendere le stesse righe: su PostgreSQL si usa SELECT ... FOR UPDATE SKIP LOCKED, che la guida indica per le tabelle di tipo coda, e la lettura deve stare nella transazione che termina dopo la marcatura, altrimenti il blocco sparisce subito. Il programma ha un solo relay, quindi non ha questo problema; non l'ho potuto provare con più connessioni. Resta il caso E: se il relay marca dopo la pubblicazione, produce copie; se marcasse prima, perderebbe eventi. La scelta giusta è produrre copie e farle assorbire dal consumatore.

Senza polling: Change Data Capture con Debezium

Debezium evita il polling leggendo il log delle transazioni del database e pubblicando le modifiche su Kafka; la sua guida per PostgreSQL richiede wal_level=logical e un massimo di slot di replicazione. Per l'outbox fornisce una trasformazione, l'EventRouter, che legge la tabella e instrada ogni riga verso un topic derivato da una colonna. Per default si aspetta una tabella con colonne dai nomi precisi: id, aggregatetype, aggregateid, type, payload.

Non è la tabella del programma sopra (che usa aggregate_id e event_type): o la ribattezzi, o configuri i nomi con le opzioni table.field.event.* che la guida elenca (id, key, payload, schema.version, timestamp). Il connettore sotto è lo schema minimo da adattare, scritto sulla documentazione e non eseguito: serve un'istanza di Kafka Connect e un PostgreSQL con i permessi di replicazione.

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "db.interno",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "<dal gestore dei segreti>",
    "database.dbname": "orders_db",
    "topic.prefix": "orders",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter"
  }
}

topic.prefix è obbligatorio nelle versioni recenti e sostituisce database.server.name. L'EventRouter è dichiarato come trasformazione con transforms.outbox.type; il nome del topic è per default derivato dal valore di aggregatetype, cioè un evento di tipo Order va su un topic che contiene Order. Due avvertenze dalla guida.

La prima: Debezium, come Kafka, garantisce la consegna almeno una volta anche in condizioni anomale, quindi l'idempotenza del consumatore non è facoltativa nemmeno con il CDC; la guida stessa suggerisce di usare l'id dell'evento per eliminare i duplicati. La seconda: gli slot di replicazione trattengono il log finché il connettore non lo consuma, e la guida raccomanda di monitorarli per evitare un consumo di disco eccessivo se uno slot resta inattivo troppo a lungo. Il lato Kafka Connect e lo slot sono approfonditi nell'articolo su Kafka Connect e Debezium, e il confine della garanzia delle transazioni Kafka, che non copre la scrittura nel database, in quello sull'exactly-once.

Pulizia e scelta

La tabella cresce: va ripulita, ma solo delle righe già pubblicate, come fa il caso H, e a lotti se è grande, per non tenere lock lunghi. Per decidere, parti dal polling: ha un componente in meno e basta finché la latenza accettabile supera l'intervallo di polling e il carico sulla tabella è basso. Passa al CDC quando il polling diventa un costo (volume alto, latenza stretta) e sei pronto a gestire Kafka Connect e gli slot. Non riporto cifre di latenza: dipendono dal tuo carico e vanno misurate.

Che cosa sai fare adesso

Sai spiegare perché la doppia scrittura perde o inventa eventi, sai scrivere un ordine e il suo evento nella stessa transazione e sai progettare un relay che sopravvive alla morte nel punto peggiore, producendo copie anziché perdite. Sai quale colonna nome cambia nel passaggio a Debezium e perché il consumatore deve restare idempotente. Il passo successivo è lo scenario in cui l'evento arriva e la lettura è ancora indietro: la consistenza eventuale e l'esperienza d'uso. Per chi coordina più servizi con questi eventi, c'è la saga. Se il lato scrittura è fatto di eventi, l'event store è già un log da cui leggere: vedi l'event sourcing e la sua combinazione con CQRS; se le viste di lettura sono alimentate dall'outbox, il modello è quello del CQRS.