Fix fanout read position tracking so silent instances can't advance the checkpoint

test/lib.streams.utest.ts

@@ -474,6 +474,8 @@ describe("Streams", function () {
 			Date.now = now;
 		});
 		afterEach(() => {
+			console.log = log;
+			Date.now = now;
 		});
 		it("counter with label and 1", async function () {
 			let logs = [];
@@ -613,10 +615,13 @@ describe("Streams", function () {
 
 	describe("devnull", function () {
 		let log = console.log;
+		let now = Date.now;
 		beforeEach(() => {
 			console.log = log;
 		});
 		afterEach(() => {
+			console.log = log;
+			Date.now = now;
 		});
 		it("log with label", async function () {
 			let logs = [];

wrappers/fanout-read-position.utest.ts

@@ -0,0 +1,193 @@
+import path from "path";
+import moment from "moment";
+import { assert } from "chai";
+import { Readable, Transform } from "stream";
+
+/**
+ * Replays the 2026-09-30 13:03 lost-events incident through the real fanout wrapper.
+ * Instances are invoked in-process (the AWS Lambda client is faked) and the queue is a fake that lets a
+ * batch land after the workers have read but before the master (which starts 200ms later) reads.
+ */
+const MODULE = process.env.FANOUT_UNDER_TEST || "./fanout";
+const QUEUE = "notification-intent";
+const BOT = "notification-prod-intent-to-recipient-intent";
+
+const indexPath = path.resolve(__dirname, "../index.js");
+const lambdaPath = require.resolve("@aws-sdk/client-lambda");
+const modulePath = require.resolve(MODULE);
+
+const eid = (ms: number, seq: number) => moment.utc(ms).format("[z/]YYYY/MM/DD/HH/mm/") + ms + "-" + String(seq).padStart(7, "0");
+
+interface TestEvent {
+	eid: string;
+	partition: number;
+}
+
+describe("Fanout read positions (13:03 incident replay)", function () {
+	this.timeout(5000);
+
+	const saved: any = {};
+	let T0: number, START: string, E923: TestEvent, batch: TestEvent[], world: { events: TestEvent[] }, persisted: any[];
+
+	function makeInstance() {
+		const leo: any = {
+			bot: {
+				checkpoint: (id: string, queue: string, params: any, cb: () => void) => {
+					persisted.push({ id, queue, params });
+					cb();
+				},
+				checkLock: (cron: any, runid: any, remaining: any, cb: Function) => cb(null, {}),
+				reportComplete: (cron: any, runid: any, status: any, log: any, opts: any, cb: Function) => cb(null, {})
+			},
+			streams: {
+				fromLeo: (ID: string, queue: string, opts: any) => {
+					const events = world.events
+						.filter(e => e.eid > START && (!opts.maxOverride || e.eid <= opts.maxOverride))
+						.map(e => Object.assign({}, e));
+					const reader: any = Readable.from(events);
+					reader.checkpoint = () => { };
+					reader.get = () => { };
+					return reader;
+				},
+				pipe: (a: Readable, b: Transform) => a.pipe(b),
+				through: (fn: Function) => new Transform({
+					objectMode: true,
+					transform(obj, enc, cb) {
+						fn(obj, (err: any, out: any) => cb(err, out));
+					}
+				})
+			}
+		};
+		leo.read = leo.streams.fromLeo;
+		require.cache[indexPath] = { id: indexPath, filename: indexPath, loaded: true, exports: leo } as any;
+
+		// The "bot": process what this instance owns and checkpoint at the last event it processed
+		const handler = async () => {
+			const processed: string[] = [];
+			await new Promise((resolve, reject) => {
+				leo.read(BOT + "_reader", QUEUE, {})
+					.on("data", (obj: TestEvent) => processed.push(obj.eid))
+					.on("end", resolve)
+					.on("error", reject);
+			});
+			if (processed.length) {
+				await new Promise(resolve => leo.bot.checkpoint(BOT + "_reader", QUEUE, {
+					eid: processed[processed.length - 1],
+					records: processed.length,
+					source_timestamp: 1,
+					started_timestamp: 1,
+					ended_timestamp: 1
+				}, resolve));
+			}
+			return processed;
+		};
+		const fanout = require(MODULE);
+		return fanout(handler, (e: TestEvent) => e.partition, { instances: 4 });
+	}
+
+	const context = { getRemainingTimeInMillis: () => 60000 };
+	const masterEvent = (ts?: number) => ({
+		__cron: { id: BOT, ts, checkpoints: { read: { ["queue:" + QUEUE]: { checkpoint: START } } } }
+	});
+
+	before(() => {
+		["AWS_LAMBDA_FUNCTION_NAME", "AWS_DEFAULT_REGION", "AWS_LAMBDA_FUNCTION_VERSION"].forEach(k => saved[k] = process.env[k]);
+		process.env.AWS_LAMBDA_FUNCTION_NAME = "fanout-under-test";
+		process.env.AWS_DEFAULT_REGION = "us-east-1";
+		process.env.AWS_LAMBDA_FUNCTION_VERSION = "$LATEST";
+		require("@aws-sdk/client-lambda");
+		saved.lambdaExports = require.cache[lambdaPath]!.exports;
+		class FakeLambda {
+			invoke(params: any, cb: Function) {
+				makeInstance()(JSON.parse(params.Payload), context)
+					.then((res: any) => cb(null, { StatusCode: 200, Payload: JSON.stringify(res) }), (err: any) => cb(err));
+			}
+		}
+		require.cache[lambdaPath]!.exports = { Lambda: FakeLambda };
+		delete require.cache[modulePath];
+		saved.index = require.cache[indexPath];
+	});
+
+	after(() => {
+		["AWS_LAMBDA_FUNCTION_NAME", "AWS_DEFAULT_REGION", "AWS_LAMBDA_FUNCTION_VERSION"].forEach(k => {
+			if (saved[k] === undefined) {
+				delete process.env[k];
+			} else {
+				process.env[k] = saved[k];
+			}
+		});
+		require.cache[lambdaPath]!.exports = saved.lambdaExports;
+		if (saved.index) {
+			require.cache[indexPath] = saved.index;
+		} else {
+			delete require.cache[indexPath];
+		}
+		delete require.cache[modulePath];
+	});
+
+	beforeEach(() => {
+		T0 = Date.now() - 20000;
+		START = eid(T0, 922);
+		E923 = { eid: eid(T0, 923), partition: 0 };
+		const owners = [1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3, 0, 2, 3]; // 398..411
+		batch = owners.map((partition, i) => ({ eid: eid(T0 + 8000, 398 + i), partition }));
+		world = { events: [E923] };
+		persisted = [];
+	});
+
+	it("does not advance past events the owners have not read (batch lands between the workers and the master)", async () => {
+		// the batch lands after the workers read (t~0) and before the master reads (t=200ms)
+		setTimeout(() => world.events.push(...batch), 100);
+		await makeInstance()(masterEvent(), context);
+		assert.lengthOf(persisted, 1);
+		// every owner still has to read the batch, so the stored checkpoint must stay at 0923
+		assert.equal(persisted[0].params.eid, E923.eid);
+	});
+
+	it("advances to the last event read when every instance saw the whole batch", async () => {
+		world.events.push(...batch);
+		await makeInstance()(masterEvent(), context);
+		assert.lengthOf(persisted, 1);
+		assert.equal(persisted[0].params.eid, batch[batch.length - 1].eid);
+	});
+
+	it("leaves events assigned after __cron.ts for the next run, even if every instance could read them", async () => {
+		world.events.push(...batch);
+		// the trigger fired before the batch's eids were assigned (T0+8000)
+		await makeInstance()(masterEvent(T0 + 5000), context);
+		assert.lengthOf(persisted, 1);
+		assert.equal(persisted[0].params.eid, E923.eid);
+	});
+
+	it("reads events assigned before __cron.ts", async () => {
+		world.events.push(...batch);
+		await makeInstance()(masterEvent(T0 + 9000), context);
+		assert.lengthOf(persisted, 1);
+		assert.equal(persisted[0].params.eid, batch[batch.length - 1].eid);
+	});
+});
+
+describe("Fanout getMaxEid", () => {
+	const { getMaxEid } = require("./fanout");
+	const now = moment.utc("2026-09-30T13:03:19.394Z");
+
+	it("pins to __cron.ts", () => {
+		assert.equal(getMaxEid({ ts: 1790773391546 }, now), "z/2026/09/30/13/03/1790773391546");
+	});
+
+	it("accepts a numeric string", () => {
+		assert.equal(getMaxEid({ ts: "1790773391546" }, now), "z/2026/09/30/13/03/1790773391546");
+	});
+
+	it("falls back to now if there is no ts", () => {
+		assert.equal(getMaxEid({}, now), "z/2026/09/30/13/03/" + now.valueOf());
+	});
+
+	it("falls back to now if ts is not a number", () => {
+		assert.equal(getMaxEid({ ts: "abc" }, now), "z/2026/09/30/13/03/" + now.valueOf());
+	});
+
+	it("falls back to now if ts is in the future", () => {
+		assert.equal(getMaxEid({ ts: now.valueOf() + 60000 }, now), "z/2026/09/30/13/03/" + now.valueOf());
+	});
+});

wrappers/fanout-read-tracker.ts

@@ -0,0 +1,67 @@
+interface ReadEntry {
+	key: string;
+	value: string | undefined;
+}
+
+/**
+ * Tracks, for one bot/queue within one fanout instance, the furthest eid this instance can
+ * safely report as its read position: everything at or before it is either not owned by this
+ * instance or has already been checkpointed.
+ *
+ * entries[i] = { key, value }
+ *   key   - an owned eid (entry 0 is the anchor: the start position, or the last checkpointed owned eid)
+ *   value - the latest eid read before the next owned eid
+ *
+ * Entries before the checkpointed one are dropped on every checkpoint, so memory is bounded by the
+ * owned events read since the last checkpoint. The checkpointed entry itself is kept as the anchor
+ * so unowned events read after it keep extending its value.
+ *
+ * eids are compared as strings, the same way fanout's min() does.
+ */
+export class ReadTracker {
+	entries: ReadEntry[];
+
+	/**
+	 * @param start eid this instance starts reading after (the bot's stored checkpoint)
+	 */
+	constructor(start?: string) {
+		this.entries = [{ key: start || "", value: start }];
+	}
+
+	/**
+	 * Call for every event that passes through the owner filter, in eid order.
+	 * @param eid the event id
+	 * @param owned true if this instance owns (will process) the event
+	 */
+	read(eid: string, owned: boolean): void {
+		if (owned) {
+			this.entries.push({ key: eid, value: eid });
+		} else {
+			this.entries[this.entries.length - 1].value = eid;
+		}
+	}
+
+	/**
+	 * Call when the bot checkpoints at eid. Keeps the entry for the largest key <= eid as the anchor
+	 * and drops everything before it.
+	 */
+	checkpoint(eid?: string): void {
+		if (!eid) {
+			return;
+		}
+		let i = 0;
+		while (i + 1 < this.entries.length && this.entries[i + 1].key <= eid) {
+			i++;
+		}
+		if (i > 0) {
+			this.entries.splice(0, i);
+		}
+	}
+
+	/**
+	 * The position to report to the master. Undefined if nothing is known (no start, nothing read).
+	 */
+	position(): string | undefined {
+		return this.entries[0].value;
+	}
+}

wrappers/fanout-read-tracker.utest.ts

@@ -0,0 +1,112 @@
+import { assert } from "chai";
+import { ReadTracker } from "./fanout-read-tracker";
+
+const eid = (n: number | string, ms = 1790773397549) => `z/2026/09/30/13/03/${ms}-${String(n).padStart(7, "0")}`;
+// the incident: 0922/0923 were written at 13:03:09, the batch (398-411) at 13:03:17
+const early = (n: number) => eid(n, 1790773389433);
+
+describe("Fanout ReadTracker", () => {
+	it("reports the start position if nothing was read", () => {
+		assert.equal(new ReadTracker(eid(922)).position(), eid(922));
+	});
+
+	it("reports nothing if there is no start and nothing was read", () => {
+		assert.isUndefined(new ReadTracker().position());
+	});
+
+	it("reports the last read eid if no owned events were read", () => {
+		const t = new ReadTracker(eid(922));
+		[923, 924, 925].forEach(n => t.read(eid(n), false));
+		assert.equal(t.position(), eid(925));
+	});
+
+	it("does not claim an owned event that was read but never checkpointed", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), false);
+		t.read(eid(924), true);
+		t.read(eid(925), false);
+		assert.equal(t.position(), eid(923));
+	});
+
+	it("reports through the unowned events read after the checkpointed owned event", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), true);
+		[924, 925, 926].forEach(n => t.read(eid(n), false));
+		t.checkpoint(eid(923));
+		assert.equal(t.position(), eid(926));
+		// and keeps extending as more unowned events arrive, with no further checkpoint calls
+		t.read(eid(927), false);
+		assert.equal(t.position(), eid(927));
+	});
+
+	it("freezes the value at the next owned event", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), true);
+		t.read(eid(924), false);
+		t.read(eid(925), true);
+		t.read(eid(926), false);
+		t.checkpoint(eid(923));
+		assert.equal(t.position(), eid(924));
+		t.checkpoint(eid(925));
+		assert.equal(t.position(), eid(926));
+	});
+
+	it("holds at the last checkpoint when the run stops with owned events still unflushed", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), true);
+		t.read(eid(924), false);
+		t.read(eid(925), true);
+		t.read(eid(926), true);
+		t.read(eid(927), false);
+		t.checkpoint(eid(923));
+		assert.equal(t.position(), eid(924));
+	});
+
+	it("uses the largest key <= the checkpoint when it is not an exact key", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), true);
+		t.read(eid(925), true);
+		t.read(eid(927), true);
+		t.checkpoint(eid(926));
+		assert.equal(t.position(), eid(925));
+	});
+
+	it("ignores a checkpoint older than the anchor", () => {
+		const t = new ReadTracker(eid(922));
+		t.read(eid(923), true);
+		t.checkpoint(eid(923));
+		t.read(eid(924), false);
+		t.checkpoint(eid(900));
+		assert.equal(t.position(), eid(924));
+	});
+
+	it("matches the 13:03 incident: workers that read only an unowned event vote 0923", () => {
+		const owners: { [seq: number]: number } = { 398: 1, 399: 2, 400: 3, 401: 0, 402: 1, 403: 2, 404: 3, 405: 0, 406: 1, 407: 2, 408: 3, 409: 0, 410: 2, 411: 3 };
+		// the workers ran before the batch landed and only saw 0923, which partition 0 owns
+		const worker = new ReadTracker(early(922));
+		worker.read(early(923), false);
+		assert.equal(worker.position(), early(923));
+		// the master ran after the batch landed
+		const master = new ReadTracker(early(922));
+		master.read(early(923), true);
+		Object.keys(owners).forEach(n => master.read(eid(n), owners[Number(n)] === 0));
+		master.checkpoint(eid(409));
+		assert.equal(master.position(), eid(411));
+		// the master's position is past the worker's, so the min stays at 0923 and the batch is re-read
+		assert.isTrue(worker.position()! < master.position()!);
+	});
+
+	it("keeps memory bounded by the owned events since the last checkpoint", () => {
+		const t = new ReadTracker(eid(0));
+		let max = 0;
+		for (let n = 1; n <= 1000000; n++) {
+			t.read(eid(n), n % 4 === 0);
+			if (n % 4000 === 0) {
+				t.checkpoint(eid(n));
+			}
+			max = Math.max(max, t.entries.length);
+		}
+		assert.isBelow(max, 1100);
+		assert.equal(t.position(), eid(1000000));
+	});
+});