How does CDC with Debezium keep search and cache in sync with Postgres?

// Log-based CDC in miniature: a Postgres-like WAL with LSNs, a Debezium-like connector
// (snapshot, then stream, offsets committed every 2 events), a Kafka-like topic (array),
// a search indexer and a cache. All in-process; no real Postgres, Kafka or Elasticsearch.
const db = new Map([["p1", { name: "lamp", price: 40 }], ["p2", { name: "desk", price: 120 }], ["p3", { name: "chair", price: 80 }]]);
const wal = []; // committed changes in commit order
let lsn = 100; // snapshot is consistent as of LSN 100 (the slot is created here)
function write(op, id, row) {
  lsn++;
  const before = db.has(id) ? { ...db.get(id) } : null;
  if (op === "d") db.delete(id); else db.set(id, { ...(db.get(id) ?? {}), ...row });
  wal.push({ lsn, op, id, before, after: op === "d" ? null : { ...db.get(id) } });
}

const topic = [];
console.log("== 1. Snapshot (op=r) then streaming (op=c/u/d)");
for (const [id, row] of db) topic.push({ op: "r", id, after: { ...row }, lsn: 100 });
write("u", "p1", { price: 35 });
write("c", "p4", { name: "shelf", price: 60 });
write("u", "p2", { price: 110 });
write("d", "p3");
write("u", "p1", { price: 30 });

// Connector: emits from the slot, commits its offset every 2 events, crashes once.
let committed = 100, sinceCommit = 0, crashed = false;
for (let i = 0; i < wal.length; i++) {
  const e = wal[i];
  if (e.lsn <= committed) continue;
  topic.push({ op: e.op, id: e.id, before: e.before, after: e.after, lsn: e.lsn });
  if (e.op === "d") topic.push({ op: "tombstone", id: e.id, lsn: e.lsn });
  if (!crashed && e.lsn === 104) { // crash after sending, before the offset flush
    crashed = true; sinceCommit = 0;
    console.log(`  connector crashed after sending LSN 104; last committed offset LSN ${committed}; restart replays from there`);
    i = wal.findIndex((x) => x.lsn > committed) - 1;
    continue;
  }
  if (++sinceCommit === 2) { committed = e.lsn; sinceCommit = 0; }
}
console.log("  topic: " + topic.map((m) => `${m.op}:${m.id}@${m.lsn}`).join(" "));

console.log("\n== 2. Consumers: search index (versioned by LSN) and a naive counter");
const index = new Map(); // id -> {doc, version}
let applied = 0, stale = 0, naivePriceChanges = 0;
for (const m of topic) {
  if (m.op === "tombstone") continue; // used by Kafka log compaction, not by the indexer
  if (m.op === "u" && m.before.price !== m.after.price) naivePriceChanges++;
  const cur = index.get(m.id);
  if (cur && cur.version >= m.lsn) { stale++; continue; } // replayed or out-of-order event
  if (m.op === "d") index.set(m.id, { doc: null, version: m.lsn }); // keep a deleted marker with its version
  else index.set(m.id, { doc: m.after, version: m.lsn });
  applied++;
}
const live = [...index.entries()].filter(([, v]) => v.doc).map(([id, v]) => `${id}=${v.doc.name}:${v.doc.price}`);
const dbView = [...db.entries()].map(([id, r]) => `${id}=${r.name}:${r.price}`);
console.log(`  messages ${topic.length}, applied ${applied}, skipped as already applied ${stale}`);
console.log(`  index [${live.join(", ")}]`);
console.log(`  db    [${dbView.join(", ")}] -> match ${live.join() === dbView.join()}`);
console.log(`  naive "price changed" counter: ${naivePriceChanges} (truth ${wal.filter((e) => e.op === "u").length})`);

console.log("\n== 3. Cache race: reader fills the cache with a value read before the update");
function race(versioned) {
  const cache = new Map();
  const readerSaw = { value: 30, version: 105 }; // t1: cache miss, reader reads the DB row
  const change = { value: 25, version: 106 }; // t2: price update commits; t3: CDC event arrives
  if (versioned) cache.set("p1", change); else cache.delete("p1");
  const cur = cache.get("p1"); // t4: slow reader writes what it read at t1
  if (!versioned || !cur || cur.version < readerSaw.version) cache.set("p1", readerSaw);
  return cache.get("p1").value;
}
console.log(`  CDC deletes the key:          cache p1 price ${race(false)} (db 25) -> stale until TTL`);
console.log(`  CDC writes value + LSN guard: cache p1 price ${race(true)} (db 25)`);

console.log("\n== 4. Replication slot retention when the slot stops advancing");
const walMbPerMin = 50;
for (const hours of [1, 6, 24]) console.log(`  stalled ${String(hours).padStart(2)} h at ${walMbPerMin} MB/min of WAL -> ${(walMbPerMin * 60 * hours / 1000).toFixed(1)} GB kept on the primary disk`);