Exactly-once processing — can Kafka or SQS guarantee it?

// Delivery semantics under a crash, simulated in-process (no real broker).
// The topic is an array; "offset" is the consumer's committed position; the
// sink is a wallet balance. The consumer crashes once, at offset 6.
const events = [];
for (let i = 0; i < 10; i++) events.push({ id: "pay-" + (101 + i), amount: 10 * (i + 1) });
const expected = events.reduce((s, e) => s + e.amount, 0);
const CRASH_AT = 6;

function run(mode) {
  const db = { balance: 0, applied: new Set(), offset: 0 }; // durable state
  let committed = 0, crashed = false, applies = 0;
  while (committed < events.length) {
    const pos = mode === "transactional" ? db.offset : committed; // where we resume
    const e = events[pos];
    const crashNow = !crashed && pos === CRASH_AT;
    if (mode === "at-most-once") {
      committed = pos + 1;                               // 1. commit offset
      if (crashNow) { crashed = true; continue; }        // crash: effect never happens
      db.balance += e.amount; applies++;                 // 2. side effect
    } else if (mode === "at-least-once") {
      db.balance += e.amount; applies++;                 // 1. side effect
      if (crashNow) { crashed = true; continue; }        // crash before commit
      committed = pos + 1;                               // 2. commit offset
    } else if (mode === "idempotent sink") {
      if (!db.applied.has(e.id)) { db.applied.add(e.id); db.balance += e.amount; applies++; } // keyed upsert
      if (crashNow) { crashed = true; continue; }
      committed = pos + 1;
    } else { // transactional: effect and offset in ONE database transaction
      const tx = { balance: db.balance + e.amount, offset: pos + 1 };
      if (crashNow) { crashed = true; continue; }        // crash before COMMIT: tx discarded
      db.balance = tx.balance; db.offset = tx.offset; applies++; committed = db.offset;
    }
  }
  const diff = db.balance - expected;
  console.log(mode.padEnd(16) + " balance=" + String(db.balance).padStart(3) + " expected=" + expected +
    " applies=" + String(applies).padStart(2) + "  " + (diff === 0 ? "correct" : diff > 0 ? "DUPLICATE +" + diff : "LOST " + -diff));
}
console.log("== 1. consumer crash between side effect and offset commit (at offset " + CRASH_AT + ") ==");
for (const m of ["at-most-once", "at-least-once", "idempotent sink", "transactional"]) run(m);

console.log("");
console.log("== 2. producer retry after a lost ack ==");
function produce(idempotent) {
  const log = [], seen = new Map(); // broker-side: producerId -> last sequence number
  const send = (pid, seq, v) => {
    if (idempotent && seen.has(pid) && seen.get(pid) >= seq) return "duplicate ignored";
    seen.set(pid, seq); log.push(v); return "appended";
  };
  const r = [send("P1", 0, "a"), send("P1", 1, "b"), send("P1", 1, "b"), send("P1", 2, "c")]; // ack for seq 1 lost, retried
  console.log((idempotent ? "idempotent producer " : "plain producer      ") + "log=[" + log.join(",") + "] retry -> " + r[2]);
}
produce(false); produce(true);

console.log("");
console.log("== 3. read-process-write transaction that aborts ==");
const out = [{ v: "r1", tx: "T1" }, { v: "r2", tx: "T1" }, { v: "r3", tx: "T2" }, { v: "r4", tx: "T2" }];
const status = { T1: "committed", T2: "aborted" };
const uncommitted = out.map((r) => r.v);
const committedOnly = out.filter((r) => status[r.tx] === "committed").map((r) => r.v);
console.log("read_uncommitted consumer sees [" + uncommitted.join(",") + "]");
console.log("read_committed   consumer sees [" + committedOnly.join(",") + "]");

console.log("");
console.log("== 4. SQS FIFO deduplication window (5 minutes) ==");
const WINDOW = 300, firstSeen = new Map(), fifo = [];
function fifoSend(dedupId, t) {
  if (firstSeen.has(dedupId) && t - firstSeen.get(dedupId) < WINDOW) return "accepted, not enqueued again";
  firstSeen.set(dedupId, t); fifo.push(dedupId); return "enqueued";
}
for (const [id, t] of [["order-42", 0], ["order-42", 60], ["order-42", 299], ["order-42", 360]])
  console.log("t=" + String(t).padStart(3) + "s send " + id + " -> " + fifoSend(id, t));
console.log("queue now holds " + fifo.length + " copies of order-42");