Event Ordering and Duplication

The previous lesson closed out the publisher side: the outbox lost no message, but delivered 24 twice. This is no longer a problem the publisher can solve. As long as the relay’s “I sent it” record and the consumer’s “I processed it” record sit in separate stores, the two cannot enter the same transaction. The cost passes to the consumer.

On the consumer side there are two separate kinds of corruption, and their effects differ. The first is duplication: the same message arrives twice. The second is reordering: a message that is running late arrives after a message that follows it. M19/K05’s Distributed Correctness topic built and measured the idempotency key and the key ledger; that measurement is not repeated here. The key ledger cuts off duplication but does nothing against reordering — this lesson separates the two and measures the cost of an implementation that handles both at once.

Mechanism

DC7. Network corruption is modeled with a seeded generator: seed 20260731, delay rate 0.20, duplication rate 0.15. At the same rates, the delivery stream is identical across every run; so the difference between runs comes only from the consumer’s implementation.

DC8. The consumer is an in-process module and writes to its own store. The query count is read from the code’s own counter, not from the store; what gets measured is the amount of work the consumer does.

# setup.sh — the consumer side's own store: view table and side-effect log
rm -f view.db*
sqlite3 view.db "PRAGMA journal_mode=WAL;
  CREATE TABLE book_view (book_id INTEGER PRIMARY KEY, status TEXT, last_version INTEGER DEFAULT 0);
  CREATE TABLE notification (id INTEGER PRIMARY KEY, book_id INTEGER, status TEXT);" >/dev/null

The catalog produces events with an increasing version for each book: odd versions are loans, even versions are returns. So the correct order can be read from a single number, and so can reordering.

// stream.mjs — produces the event log, then breaks delivery order and duplicates messages
export function generator(seed) {                  // seed is visible, sequence is the same on every run
  let d = seed >>> 0;
  return () => { d = (Math.imul(d, 1103515245) + 12345) >>> 0; return (d >>> 8) / 16777216; };
}

export function journal(books, versionCount) {      // the catalog's correctly ordered journal
  const g = [];
  for (let s = 1; s <= versionCount; s++)
    for (let k = 1; k <= books; k++)
      g.push({ message_id: g.length + 1, book: k, version: s, status: s % 2 ? "on_loan" : "on_shelf" });
  return g;
}

export function deliveryOrder(log, seed, delay, duplication) {
  const roll = generator(seed), output = [], held = [];
  for (const o of log) {
    if (roll() < delay) { held.push(o); continue; }             // this message will be late
    output.push(o);
    if (held.length && roll() < 0.10) output.push(held.shift());
    if (roll() < duplication) output.push(o);                   // at-least-once delivery: duplicate
  }
  output.push(...held);                                         // the latest stragglers go last
  return output;
}

The consumer is a single file and contains two implementations. --plain applies whatever arrives, as is. --versioned stores, for each book, the last version it processed, and discards any message that is not greater than that version.

// consumer.mjs — same stream, two consumer implementations: --plain and --versioned
import { DatabaseSync } from "node:sqlite";
import { journal, deliveryOrder } from "./stream.mjs";

const versioned = process.argv.includes("--versioned");
const BOOKS = 10, VERSIONS = 300, SEED = 20260731;
const delay = Number(process.env.DELAY ?? 0.20);        // fraction of messages that break the order
const duplication = Number(process.env.DUPLICATION ?? 0.15);  // fraction delivered a second time
const correct = journal(BOOKS, VERSIONS);
const stream = deliveryOrder(correct, SEED, delay, duplication);

const db = new DatabaseSync("view.db");
const readVersion = db.prepare("SELECT last_version FROM book_view WHERE book_id = ?");
const write = db.prepare(
  "INSERT INTO book_view (book_id, status, last_version) VALUES (?,?,?)" +
  " ON CONFLICT(book_id) DO UPDATE SET status = excluded.status, last_version = excluded.last_version");
const notify = db.prepare("INSERT INTO notification (book_id, status) VALUES (?,?)");

let processed = 0, discarded = 0, queries = 0, staleArrival = 0, discardedDuplicate = 0, discardedStale = 0;
const seen = new Map(), seenMessages = new Set();
const t0 = performance.now();
db.exec("BEGIN IMMEDIATE");
for (const o of stream) {
  if ((seen.get(o.book) ?? 0) > o.version) staleArrival += 1;   // an out-of-order arrival
  seen.set(o.book, Math.max(seen.get(o.book) ?? 0, o.version));
  if (versioned) {
    const s = readVersion.get(o.book); queries += 1;            // extra read: version stamp
    if (s && s.last_version >= o.version) {
      discarded += 1; seenMessages.has(o.message_id) ? (discardedDuplicate += 1) : (discardedStale += 1);
      continue;
    }
  }
  seenMessages.add(o.message_id);
  write.run(o.book, o.status, o.version); queries += 1;
  notify.run(o.book, o.status); queries += 1;
  processed += 1;
}
db.exec("COMMIT");
const duration = performance.now() - t0;

const finalState = new Map(correct.map((o) => [o.book, o.status]));
let wrong = 0;
for (const r of db.prepare("SELECT book_id, status FROM book_view").all())
  if (finalState.get(r.book_id) !== r.status) wrong += 1;
const sideEffects = db.prepare("SELECT COUNT(*) c FROM notification").get().c;

console.log(`${versioned ? "versioned" : "plain    "} (delay=${delay} duplication=${duplication})` +
  ` | log=${correct.length} delivered=${stream.length}` +
  ` out-of-order arrivals=${staleArrival} | processed=${processed} discarded=${discarded} (duplicate=${discardedDuplicate} stale=${discardedStale})` +
  ` side effects=${sideEffects} wrong-final books=${wrong}/${BOOKS}` +
  ` | queries=${queries} duration=${duration.toFixed(1)} ms (in this run)`);

Two Kinds of Corruption, Two Different Harms

The first two runs apply the corruptions one at a time; the third gives both together; the fourth feeds the same stream to the version-stamped consumer.

sh setup.sh; DELAY=0 node consumer.mjs --plain
sh setup.sh; DUPLICATION=0 node consumer.mjs --plain
sh setup.sh; node consumer.mjs --plain
sh setup.sh; node consumer.mjs --versioned

plain     (delay=0 duplication=0.15) | log=3000 delivered=3452 out-of-order arrivals=0 | processed=3452 discarded=0 (duplicate=0 stale=0) side effects=3452 wrong-final books=0/10 | queries=6904 duration=2.8 ms (in this run)
plain     (delay=0.2 duplication=0) | log=3000 delivered=3000 out-of-order arrivals=585 | processed=3000 discarded=0 (duplicate=0 stale=0) side effects=3000 wrong-final books=4/10 | queries=6000 duration=2.3 ms (in this run)
plain     (delay=0.2 duplication=0.15) | log=3000 delivered=3359 out-of-order arrivals=585 | processed=3359 discarded=0 (duplicate=0 stale=0) side effects=3359 wrong-final books=4/10 | queries=6718 duration=2.5 ms (in this run)
versioned (delay=0.2 duplication=0.15) | log=3000 delivered=3359 out-of-order arrivals=585 | processed=2415 discarded=944 (duplicate=359 stale=585) side effects=2415 wrong-final books=0/10 | queries=8189 duration=3.4 ms (in this run)

The first row is duplication only. Three thousand events were delivered 3452 times, and the consumer produced 3452 side effects: 452 extra notifications. The final state, by contrast, is correct — because applying the same message a second time does not change the view, it only repeats the notification. The harm of duplication is in the side effect, not in the state.

The second row is reordering only. The delivery count equals the log, the side effect count is correct (3000), but 585 messages arrived after the one that followed them, and 4 of the 10 books ended up in the wrong state. The harm of reordering is in the state, not in the side effect count.

When both corruptions are given together, both show up: 3359 side effects and, again, 4 wrong books. Even if the key ledger cut off duplication, these 4 books would stay wrong; the ledger recognizes the same message arriving a second time, it does not recognize a different, late-arriving message.

Version Stamp

The fourth row feeds the same stream to the version-stamped consumer. The result: wrong-final books 0, side effects drop from 3359 to 2415, and 944 messages are discarded.

The breakdown of the discards matters: 944 = 359 duplicates + 585 stale. The duplicates are the same message arriving a second time; discarding them is pure gain. The stale ones, though, are real events — a newer version was written over them purely because they arrived late. The version stamp discards these too, because rolling a version backward has no meaning.

This is why the side effect count is 2415, not 3000. The view ended up correct, but for 585 real state changes, no notification was ever sent. This is the cost of handling both corruptions with a single mechanism: the rule that fixes reordering also erases late-arriving information.

Which behavior is right depends on what the consumer does. For a consumer that carries only the final state, like a view table, this is the correct behavior; the discarded message costs nothing. For a consumer that has to react to every transition — a service computing late fees, or one that logs every movement — the 585 losses are a real loss, and that consumer has to work with a ledger that recognizes each message individually, not a version stamp.

The Idempotent Consumer’s Cost

The run-independent measure of this cost is the query count. The plain consumer makes two writes per delivery: 3359 × 2 = 6718. The versioned consumer makes one read first for every delivery (3359), then two writes only for what it processes (2415 × 2 = 4830); total 8189. The difference is +1471 queries, about 0.44 extra queries per delivery. The duration difference points the same way (2.5 ms versus 3.4 ms), but the real fixed cost is the read: every message is queried first, even the ones that end up discarded.

The cost on the store side is visible too: a last_version column was added to the book_view table and updated on every write. The version stamp is a permanent field that enters the consumer’s schema; the service producing the message also has to make publishing an increasing version on every event part of its contract.

What it made cheaper: the version stamp derived the correct final state from a duplicated, out-of-order stream — 4 wrong books dropped to 0 — and zeroed out the extra notifications. It did this without raising any guarantee on the publisher side; the outbox’s “at least once” delivery stayed exactly as it was.

What it made more expensive: a column was added to the consumer schema, and a read was added for every message (+1471 queries); on top of that, the publisher is now obligated to publish an increasing version on every event.

Which new failure mode was born: silent discarding. The consumer swallowed 585 real events without raising a single error. Because the view is correct, this loss cannot be seen from the state; it only shows up if the side effect count is compared against the log length. If the version field is produced incorrectly — for instance if two services write to the same record with separate counters — discarded messages increase and no one notices.

Summary

• Duplication and reordering cause different harms: with duplication alone, side effects came out to 3452 instead of 3000 but the final state was correct; with reordering alone, the side effect count was correct but 4 of 10 books ended up in the wrong state.
• So a ledger that only cuts off duplication is not enough on its own; a late message is a different message and cannot be recognized.
• The version-stamped consumer derived the correct final state from the same stream (0/10 wrong) and discarded 944 messages: 359 duplicates, 585 stale.
• The discarded stale messages are real events; side effects stayed at 2415 instead of 3000. For a consumer that carries only the final state this is correct; for a consumer that reacts to every transition it is a loss.
• The cost is about 0.44 extra queries per delivery (8189 instead of 6718), a column in the consumer schema, and an increasing-version obligation in the publisher’s contract.

Next Step

This lesson’s solution silently made a promise: the publisher will publish an increasing version field on every event, and the consumer will read that field. The same kind of promise was made in earlier lessons — message_id will stay permanent, the book field will always carry the same meaning. These fields are now a contract two separate teams hold together, and the system’s correctness depends on it. Because the services are separate deployment units, the contract cannot change at a single instant: for a while, two versions run together. The next lesson takes up how the contract can change — the order of adding and removing a field, two versions running together, and protecting the consumer.