Outbox pattern vs polling for CDC — how do you avoid dual-write failure?

// Dual write vs transactional outbox vs updated_at polling, simulated in-process.
// "db" is a database with atomic transactions, "broker" is an append-only array (Kafka-like).
const N = 1000;

console.log("== 1. Dual write: commit to DB, then publish to the broker");
{
  const db = [], broker = [];
  const crashBetween = new Set([417]); // process killed after COMMIT, before publish
  const brokerDown = new Set([733]); // publish failed after the commit; error only logged
  for (let id = 1; id <= N; id++) {
    db.push(id);
    if (crashBetween.has(id) || brokerDown.has(id)) continue;
    broker.push({ eventId: id, orderId: id });
  }
  console.log(`  orders in DB ${db.length}, OrderPlaced events ${broker.length}, lost events ${db.length - broker.length}`);
}

console.log("\n== 2. Dual write the other way: publish first, then commit");
{
  const db = [], broker = [];
  const commitFails = new Set([120, 555, 901]); // deadlock victim, unique violation, failover
  for (let id = 1; id <= N; id++) {
    broker.push({ eventId: id, orderId: id });
    if (!commitFails.has(id)) db.push(id);
  }
  const inDb = new Set(db);
  const phantom = broker.filter((e) => !inDb.has(e.orderId)).length;
  console.log(`  orders in DB ${db.length}, events ${broker.length}, events for orders that do not exist ${phantom}`);
}

console.log("\n== 3. Transactional outbox + relay (batch 100), relay crashes once");
{
  const orders = [], outbox = [], broker = [];
  for (let id = 1; id <= N; id++) {
    if (id === 417) continue; // crash inside the transaction: neither row is written
    orders.push(id); // same transaction:
    outbox.push({ id, type: "OrderPlaced", orderId: id, published: false });
  }
  let crashed = false;
  for (;;) {
    const batch = outbox.filter((r) => !r.published).slice(0, 100); // SELECT ... ORDER BY id LIMIT 100
    if (batch.length === 0) break;
    for (const r of batch) broker.push({ eventId: r.id, orderId: r.orderId });
    if (!crashed && batch[0].id > 300) { crashed = true; console.log(`  relay crashed after sending ids ${batch[0].id}-${batch[batch.length - 1].id}, before marking them`); continue; }
    for (const r of batch) r.published = true; // UPDATE outbox SET published = true
  }
  const seen = new Set();
  let applied = 0, skipped = 0;
  for (const e of broker) { if (seen.has(e.eventId)) { skipped++; continue; } seen.add(e.eventId); applied++; }
  console.log(`  orders ${orders.length}, outbox rows ${outbox.length}, broker messages ${broker.length}`);
  console.log(`  consumer applied ${applied}, duplicates skipped by event id ${skipped}, lost ${orders.length - applied}`);
}

console.log("\n== 4. Query-based polling: SELECT * FROM orders WHERE updated_at > :watermark");
{
  const table = new Map(); // id -> {status, updatedAt}
  const changes = [ // [commitTime, id, status or null for delete, updatedAt written by the app]
    [50, "o2", "NEW", 50], [100, "o1", "NEW", 100], [150, "o1", "PAID", 150], [180, "o1", "SHIPPED", 180],
    [250, "o2", null, 250], [310, "o4", "NEW", 300], [420, "o3", "NEW", 290], // o3: long transaction, commits late
  ];
  const polls = [75, 200, 400, 600];
  let watermark = -1, ci = 0;
  const delivered = [];
  for (const p of polls) {
    while (ci < changes.length && changes[ci][0] <= p) {
      const [, id, status, at] = changes[ci++];
      if (status === null) table.delete(id); else table.set(id, { status, updatedAt: at });
    }
    const rows = [...table.entries()].filter(([, r]) => r.updatedAt > watermark).sort((a, b) => a[1].updatedAt - b[1].updatedAt);
    for (const [id, r] of rows) { delivered.push(`${id}:${r.status}`); watermark = Math.max(watermark, r.updatedAt); }
    console.log(`  poll t=${p}: got [${rows.map(([id, r]) => `${id}:${r.status}`).join(", ")}] watermark=${watermark}`);
  }
  const all = changes.map(([, id, s]) => `${id}:${s ?? "DELETED"}`);
  const missed = all.filter((c) => !delivered.includes(c));
  console.log(`  committed changes ${all.length}, delivered ${delivered.length}, missed ${missed.length}: ${missed.join(", ")}`);
  console.log(`  a log-based reader of the same commits (WAL order) delivers ${changes.length} of ${changes.length}`);
}