09 - Idempotenza nei Consumer: Deduplication e Idempotency Key

Perché la mail di conferma parte due volte

Il sistema funziona in sviluppo e in produzione, finché un cliente non riceve due volte la mail di conferma, o una carta viene addebitata due volte. Non c'è un bug da trovare: i broker che usi, a partire da SQS, garantiscono la consegna almeno una volta, e le copie sono previste. Un consumatore che sbaglia le copie è un consumatore scritto per un mondo che non esiste. La difesa è l'idempotenza: elaborare lo stesso messaggio una, due o dieci volte deve produrre lo stesso risultato che elaborarlo una sola volta.

Ci sono tre strumenti, e questo articolo li esegue tutti: una chiave di idempotenza scritta insieme all'effetto, un'operazione idempotente per natura, e il pattern inbox, che separa il ricevere dall'elaborare. Per ogni strumento mostro dove tiene e dove si rompe (la finestra di una chiave con scadenza, un effetto esterno che la transazione non copre). I programmi sono in memoria o su SQLite, senza broker: dove un broker vero cambia le cose, lo dico. Il contesto sta nell'articolo sulle dead letter queue, che spiega da dove vengono molti dei messaggi riconsegnati.

Da dove vengono i duplicati

Il caso tipico: il consumatore addebita il pagamento, poi muore prima di confermare il messaggio al broker. Per il broker il messaggio non è stato elaborato, e lo riconsegna. Il secondo caso è il timeout: in SQS il messaggio ricevuto diventa invisibile per il visibility timeout, 30 secondi per default secondo la guida; se l'elaborazione ne richiede 40, il messaggio torna visibile e un altro consumatore lo prende mentre il primo sta ancora lavorando (la prova è nell'articolo su SQS, SNS ed EventBridge).

Un terzo caso è il produttore che ritenta un invio di cui non ha ricevuto conferma. In nessuno dei tre casi c'è un errore da correggere; il rimedio sta nel consumatore, come per i consumatori di Kafka, trattati nella serie dedicata. Come un consumatore Kafka salva la posizione e che cosa succede a un record che non passa è nell'articolo sulla dead letter queue in Kafka; dove finisce la garanzia delle transazioni lo spiega l'articolo sull'exactly-once; i record riletti dopo un failover tra cluster sono il caso della guida alla produzione.

La chiave e l'effetto nella stessa transazione

L'idea è semplice: ogni messaggio ha un identificativo, e prima di elaborarlo il consumatore controlla se l'ha già visto. Il punto delicato è dove scrivi la chiave. Se la scrivi prima dell'effetto, un crash a metà lascia la chiave senza l'effetto, e il retry viene scartato come duplicato. Se la scrivi dopo, un crash fra i due lascia l'effetto senza chiave e il retry lo ripete.

La soluzione è scrivere chiave ed effetto nella stessa transazione: o entrambi o nessuno. Il programma lo fa con una tabella processed_messages e INSERT OR IGNORE, che su SQLite fa la stessa cosa del INSERT ... ON CONFLICT DO NOTHING di PostgreSQL (la guida di PostgreSQL lo definisce come l'alternativa che evita semplicemente di inserire la riga).

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE processed_messages (message_id TEXT PRIMARY KEY, consumer TEXT NOT NULL);
  CREATE TABLE inventory (product_id TEXT PRIMARY KEY, quantity INTEGER NOT NULL);
  CREATE TABLE orders (id TEXT PRIMARY KEY, payment_status TEXT NOT NULL);
  INSERT INTO inventory VALUES ('penna', 100);
  INSERT INTO orders VALUES ('o-1', 'pending');
`);
const quantita = () => (db.prepare("SELECT quantity FROM inventory WHERE product_id = 'penna'").get() as { quantity: number }).quantity;

// Il consumatore scrive la chiave e l'effetto nella STESSA transazione: o entrambi o nessuno.
function elabora(messageId: string, handler: () => void): 'elaborato' | 'duplicato' {
  db.exec('BEGIN');
  try {
    const r = db.prepare('INSERT OR IGNORE INTO processed_messages VALUES (?, ?)').run(messageId, 'inventario');
    if (r.changes === 0) { db.exec('ROLLBACK'); return 'duplicato'; }
    handler();
    db.exec('COMMIT');
    return 'elaborato';
  } catch (e) {
    db.exec('ROLLBACK');                 // la chiave sparisce insieme all'effetto: il retry e' possibile
    throw e;
  }
}
const scala = (n: number) => () => db.prepare("UPDATE inventory SET quantity = quantity - ? WHERE product_id = 'penna'").run(n);

// 1) Stesso messaggio due volte.
console.log('1) prima consegna :', elabora('msg-1', scala(5)), '| quantita', quantita());
console.log('1) seconda consegna:', elabora('msg-1', scala(5)), '| quantita', quantita());

// 2) L'handler fallisce a meta': niente chiave, niente effetto, e il retry funziona.
try { elabora('msg-2', () => { scala(5)(); throw new Error('crash dopo l\'UPDATE'); }); } catch (e) { console.log('2) errore:', (e as Error).message, '| quantita', quantita()); }
console.log('2) retry          :', elabora('msg-2', scala(5)), '| quantita', quantita());

// 3) Effetto esterno al database: la transazione non lo copre.
let addebiti = 0;
const gateway = { visti: new Set<string>(), addebita(chiave?: string) { if (chiave && this.visti.has(chiave)) return; if (chiave) this.visti.add(chiave); addebiti++; } };
function pagaSenzaChiave(id: string) {
  try { elabora(id, () => { gateway.addebita(); throw new Error('commit fallito dopo l\'addebito'); }); } catch { /* il broker riconsegna */ }
  elabora(id, () => gateway.addebita());
}
pagaSenzaChiave('pay-1');
console.log('3) senza chiave al gateway: addebiti =', addebiti);
addebiti = 0;
function pagaConChiave(id: string) {
  try { elabora(id, () => { gateway.addebita(id); throw new Error('commit fallito dopo l\'addebito'); }); } catch { /* il broker riconsegna */ }
  elabora(id, () => gateway.addebita(id));
}
pagaConChiave('pay-2');
console.log('3) con la chiave al gateway: addebiti =', addebiti);

// 4) Idempotenza naturale: stato assoluto invece di delta, transizione che non fa nulla la seconda volta.
const segnaPagato = () => db.prepare("UPDATE orders SET payment_status = 'paid' WHERE id = 'o-1' AND payment_status = 'pending'").run().changes;
console.log('4) segna pagato: modifiche', segnaPagato(), 'poi', segnaPagato());

// Output:
// 1) prima consegna : elaborato | quantita 95
// 1) seconda consegna: duplicato | quantita 95
// 2) errore: crash dopo l'UPDATE | quantita 95
// 2) retry          : elaborato | quantita 90
// 3) senza chiave al gateway: addebiti = 2
// 3) con la chiave al gateway: addebiti = 1
// 4) segna pagato: modifiche 1 poi 0

Nel primo scenario la seconda consegna di msg-1 viene scartata e la quantità resta 95: il changes === 0 della INSERT OR IGNORE è il segnale di duplicato, e il ROLLBACK chiude la transazione senza eseguire l'handler. Il secondo scenario è il motivo per cui la chiave sta nella transazione: l'handler fallisce dopo l'UPDATE, la ROLLBACK annulla sia l'UPDATE sia la chiave (quantità ancora 95), e il retry riesce (90).

Il terzo scenario è la trappola più costosa: l'addebito sul gateway non fa parte della transazione del database. Se il commit fallisce dopo l'addebito, il retry addebita di nuovo (addebiti = 2). L'unico rimedio è passare la chiave anche al gateway: Stripe, per esempio, accetta l'intestazione Idempotency-Key sulle richieste e riconosce una richiesta ripetuta; nel programma il gateway finto si comporta così e gli addebiti restano 1. Una transazione del database garantisce solo ciò che sta nel database.

Su PostgreSQL il comportamento sotto concorrenza (due consumatori che inseriscono la stessa chiave nello stesso istante) è un vincolo di unicità che il database fa rispettare; non l'ho eseguito qui, e va coperto da un test sul database vero. Una precisazione sulla chiave: usa l'identificatore che identifica l'evento di business (per esempio l'id dell'evento o del pagamento), non l'identificatore che il broker assegna al singolo invio. Se il produttore ritenta e rimanda lo stesso evento, il broker può assegnargli un nuovo identificativo di messaggio, e la chiave del broker non riconosce la copia.

Idempotenza per natura: lo stato, non il delta

La soluzione più robusta è non averne bisogno. Un comando come «sottrai 5 al magazzino» cambia il risultato a ogni ripetizione; un comando come «segna l'ordine come pagato se è in attesa» no. L'ultima riga dell'uscita lo mostra: la prima esecuzione modifica una riga, la seconda zero, perché la clausola WHERE payment_status = 'pending' non trova più nulla da cambiare.

Lo stesso vale per un UPSERT che scrive il valore assoluto («la prenotazione dell'ordine o-1 per la penna è 5») invece di un incremento. Non sempre si può: un addebito è per sua natura un'azione, e lì serve la chiave. Ma ogni operazione che riesci a rendere un'asserzione sullo stato è un meccanismo in meno da gestire, da scadenzare e da far sopravvivere a un riavvio.

Chiavi con scadenza: la finestra e il riavvio

Se i messaggi sono molti, una tabella di chiavi per sempre non è praticabile e si usa un archivio con scadenza. Redis ha il comando SET con le opzioni NX (la guida: imposta la chiave solo se non esiste già) ed EX (la scadenza in secondi): in un solo comando atomico dice se la chiave era nuova, senza bisogno degli script Lua.

Il modello sotto lo riproduce con un tempo virtuale e mostra due limiti: la chiave scade, e un archivio senza persistenza perde le chiavi al riavvio (la guida di Redis avverte che uno snapshot RDB può far perdere gli ultimi minuti di dati). Il secondo pezzo riguarda il lato produttore: le code FIFO di SQS scartano un invio con lo stesso identificativo entro cinque minuti.

// Tempo virtuale in secondi.
const clock = { t: 0 };

// Modello di SET key value NX EX ttl: crea la chiave solo se manca e la fa scadere dopo ttl secondi.
class StoreConScadenza {
  private chiavi = new Map<string, number>();            // chiave -> istante di scadenza
  setNxEx(chiave: string, ttl: number): boolean {
    const scadenza = this.chiavi.get(chiave);
    if (scadenza !== undefined && scadenza > clock.t) return false;   // esiste e non e' scaduta: duplicato
    this.chiavi.set(chiave, clock.t + ttl);
    return true;
  }
  riavvioSenzaPersistenza(): void { this.chiavi.clear(); }
}

const TTL = 24 * 3600;
const store = new StoreConScadenza();
let effetti = 0;
const consuma = (chiave: string) => { if (store.setNxEx(chiave, TTL)) effetti++; };

consuma('orders-0-17');
clock.t = 60;           consuma('orders-0-17');
console.log('1) duplicato dopo 1 minuto  : effetti =', effetti);
clock.t = 25 * 3600;    consuma('orders-0-17');
console.log('2) duplicato dopo 25 ore    : effetti =', effetti, '(la chiave e\' scaduta)');
store.riavvioSenzaPersistenza(); consuma('orders-0-17');
console.log('3) dopo un riavvio dello store: effetti =', effetti, '(la chiave e\' stata persa)');

// Deduplicazione lato produttore: la chiave deve descrivere l'evento, non l'istante dell'invio.
const FINESTRA = 300;
class CodaConDedup {
  private visti = new Map<string, number>();
  accettati: string[] = [];
  invia(dedupId: string): void {
    const prima = this.visti.get(dedupId);
    if (prima !== undefined && clock.t - prima < FINESTRA) return;
    this.visti.set(dedupId, clock.t);
    this.accettati.push(dedupId);
  }
}
clock.t = 0;
const sbagliata = new CodaConDedup(), giusta = new CodaConDedup();
for (const tentativo of [0, 1]) {                       // il produttore ritenta l'invio dopo 1 secondo
  clock.t = tentativo;
  sbagliata.invia(`OrdineCreato-o-9-${1_700_000_000 + clock.t}`);   // con l'istante nella chiave
  giusta.invia('OrdineCreato-o-9');                                 // con l'identita' dell'evento
}
console.log('4) chiave con istante :', sbagliata.accettati.length, 'messaggi accettati');
console.log('4) chiave dell\'evento :', giusta.accettati.length, 'messaggio accettato');

// Output:
// 1) duplicato dopo 1 minuto  : effetti = 1
// 2) duplicato dopo 25 ore    : effetti = 2 (la chiave e' scaduta)
// 3) dopo un riavvio dello store: effetti = 3 (la chiave e' stata persa)
// 4) chiave con istante : 2 messaggi accettati
// 4) chiave dell'evento : 1 messaggio accettato

Il duplicato dopo un minuto è scartato (effetti = 1); quello dopo 25 ore passa, perché il TTL era di 24 ore, e il terzo passa dopo il riavvio. La conclusione operativa è che il TTL dev'essere più lungo del più lungo ritardo plausibile di una riconsegna (per SQS la conservazione di default è 4 giorni, quindi un TTL di 24 ore può non bastare) e che per le operazioni irreversibili, come i pagamenti, la chiave sta nel database transazionale e non solo in Redis.

Le ultime due righe sono l'errore classico del produttore: se nella chiave di deduplicazione metti l'istante dell'invio, ogni ritentativo ha una chiave nuova e passa; una chiave che descrive l'evento (tipo e id dell'ordine) fa scartare il secondo invio.

Inbox: ricevere e elaborare sono due passi

Il pattern inbox sposta la chiave in una tabella che contiene anche il messaggio. Il primo passo salva il messaggio e restituisce subito il consenso al broker: se lo stesso messaggio arriva due volte, il vincolo sulla chiave primaria scarta la copia. Il secondo passo, un worker, legge i messaggi non ancora elaborati e li esegue in una transazione che include l'effetto e la marcatura «elaborato». Rispetto alla sola chiave ottieni due vantaggi: il broker non vede mai un fallimento di elaborazione (il messaggio è già al sicuro) e hai uno storico con errori e numero di tentativi.

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE inbox (id TEXT PRIMARY KEY, event_type TEXT NOT NULL, payload TEXT NOT NULL,
                      processed_at TEXT, error TEXT, retry_count INTEGER NOT NULL DEFAULT 0);
  CREATE TABLE inventory (product_id TEXT PRIMARY KEY, reserved INTEGER NOT NULL);
  INSERT INTO inventory VALUES ('penna', 0);
`);

// Fase 1: ricevere = salvare. Il broker puo' ora ricevere l'ack: il messaggio e' al sicuro, anche se arriva due volte.
function ricevi(id: string, eventType: string, payload: object): boolean {
  return db.prepare('INSERT OR IGNORE INTO inbox (id, event_type, payload) VALUES (?, ?, ?)').run(id, eventType, JSON.stringify(payload)).changes === 1;
}

let guasto = true; // il primo tentativo di elaborare "OrderPlaced" fallisce
function gestisci(tipo: string, p: { productId: string; quantity: number }): void {
  if (tipo !== 'OrderPlaced') throw new Error('Tipo sconosciuto: ' + tipo);
  db.prepare('UPDATE inventory SET reserved = reserved + ? WHERE product_id = ?').run(p.quantity, p.productId);
  if (guasto) { guasto = false; throw new Error('guasto dopo l\'UPDATE'); }
}

// Fase 2: elaborare. Effetto e marcatura "processato" nella stessa transazione.
function elaboraProssimo(): string {
  const riga = db.prepare('SELECT id, event_type, payload FROM inbox WHERE processed_at IS NULL AND retry_count < 3 ORDER BY rowid LIMIT 1').get() as
    { id: string; event_type: string; payload: string } | undefined;
  if (!riga) return 'niente da fare';
  db.exec('BEGIN');
  try {
    gestisci(riga.event_type, JSON.parse(riga.payload));
    db.prepare("UPDATE inbox SET processed_at = 'ora', error = NULL WHERE id = ?").run(riga.id);
    db.exec('COMMIT');
    return `${riga.id} elaborato`;
  } catch (e) {
    db.exec('ROLLBACK');
    db.prepare('UPDATE inbox SET retry_count = retry_count + 1, error = ? WHERE id = ?').run((e as Error).message, riga.id);
    return `${riga.id} fallito: ${(e as Error).message}`;
  }
}

console.log('ricevuto 1a volta :', ricevi('evt-1', 'OrderPlaced', { productId: 'penna', quantity: 3 }));
console.log('ricevuto 2a volta :', ricevi('evt-1', 'OrderPlaced', { productId: 'penna', quantity: 3 }));
console.log('giro 1            :', elaboraProssimo());
console.log('giro 2            :', elaboraProssimo());
console.log('giro 3            :', elaboraProssimo());
console.log('prenotate         :', (db.prepare("SELECT reserved FROM inventory WHERE product_id = 'penna'").get() as { reserved: number }).reserved);
console.log('stato inbox       :', JSON.stringify(db.prepare('SELECT id, processed_at, retry_count, error FROM inbox').all()));

// Output:
// ricevuto 1a volta : true
// ricevuto 2a volta : false
// giro 1            : evt-1 fallito: guasto dopo l'UPDATE
// giro 2            : evt-1 elaborato
// giro 3            : niente da fare
// prenotate         : 3
// stato inbox       : [{"id":"evt-1","processed_at":"ora","retry_count":1,"error":null}]

La seconda ricezione restituisce false: il messaggio non viene salvato due volte. Il primo giro fallisce dopo l'UPDATE dell'inventario, il ROLLBACK annulla l'UPDATE e il worker incrementa retry_count; il secondo giro riesce e le unità prenotate sono 3, non 6. Il limite retry_count < 3 evita che un messaggio velenoso giri per sempre, ma poi serve qualcuno che guardi le righe fallite, come per una DLQ.

Su PostgreSQL con più worker si usa tipicamente SELECT ... FOR UPDATE SKIP LOCKED (la guida lo indica per le tabelle di tipo coda), ma la selezione e l'elaborazione devono stare nella stessa transazione: se la lettura avviene in autocommit, il blocco sparisce subito e due worker possono prendere la stessa riga. Questo è il punto in cui si sbaglia più spesso e che SQLite, con un solo scrittore, non permette di provare.

Come testare l'idempotenza

Il test normale non basta: bisogna forzare le riconsegne. I casi minimi sono questi:

• lo stesso messaggio consegnato due volte di fila: l'effetto compare una volta sola;
• un crash dopo l'effetto e prima della conferma: il retry non lo duplica e non lo perde;
• due copie contemporanee: una sola elabora, l'altra viene scartata;
• una copia dopo la scadenza della chiave: il comportamento è deciso e documentato, non casuale.

I primi due si scrivono con un database in memoria, come negli esempi sopra. Il terzo richiede il database vero con due connessioni, perché dipende da come quel database serializza i conflitti: è il test che ti dice se la tua chiave regge. Il quarto è quello che di solito manca, e che trasforma la «finestra» da bug occasionale a limite dichiarato.

Che cosa sai fare adesso

Sai perché i duplicati sono attesi, sai scrivere chiave ed effetto nella stessa transazione e sai riconoscere gli effetti che la transazione non copre. Sai scegliere tra chiave con scadenza, chiave durevole e inbox in base al costo di un duplicato, e hai i quattro test che dimostrano che il consumatore regge. Il passo successivo riguarda il lato opposto del problema: pubblicare l'evento senza duplicarlo né perderlo, con l'outbox pattern. Per il lato che pubblica gli eventi vedi anche EventBridge; per ciò che l'utente vede mentre i dati si allineano, la consistenza eventuale.