diff --git a/test/lib.streams.utest.ts b/test/lib.streams.utest.ts index 364c1ac..fe629c6 100644 --- a/test/lib.streams.utest.ts +++ b/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 = []; diff --git a/wrappers/fanout-read-position.utest.ts b/wrappers/fanout-read-position.utest.ts new file mode 100644 index 0000000..95c2d42 --- /dev/null +++ b/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()); + }); +}); diff --git a/wrappers/fanout-read-tracker.ts b/wrappers/fanout-read-tracker.ts new file mode 100644 index 0000000..f559474 --- /dev/null +++ b/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; + } +} diff --git a/wrappers/fanout-read-tracker.utest.ts b/wrappers/fanout-read-tracker.utest.ts new file mode 100644 index 0000000..495ba2b --- /dev/null +++ b/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)); + }); +}); diff --git a/wrappers/fanout-reduce-guard.utest.ts b/wrappers/fanout-reduce-guard.utest.ts new file mode 100644 index 0000000..dcf2101 --- /dev/null +++ b/wrappers/fanout-reduce-guard.utest.ts @@ -0,0 +1,45 @@ +import { assert } from "chai"; + +const { reduceCheckpoints } = require("./fanout"); + +const eid = (n: number, ms = 1790773397549) => `z/2026/09/30/13/03/${ms}-${String(n).padStart(7, "0")}`; +// 0923 was written at 13:03:09, the batch (398-411) at 13:03:17 +const early = (n: number) => eid(n, 1790773389433); +const report = (iid: number, checkpoint?: string) => ({ + iid, + checkpoints: checkpoint ? { + bot: { read: { q: { records: 1, checkpoint } }, write: {} } + } : {} +}); + +describe("Fanout reduceCheckpoints silent instances", () => { + it("takes the min when every instance reports", () => { + const result = reduceCheckpoints([report(3, early(923)), report(2, early(923)), report(1, early(923)), report(0, eid(411))]); + assert.equal(result[0].bot.read.q.eid, early(923)); + }); + + it("throws if an instance did not report a queue another instance reported", () => { + assert.throws( + () => reduceCheckpoints([report(1), report(0, eid(409))]), + Error, + "fanout instance(s) [1] did not report a read position for bot/q" + ); + }); + + it("does not throw for silent instances when strictCheckpoints is false", () => { + const result = reduceCheckpoints([report(1), report(0, eid(409))], { strictCheckpoints: false }); + assert.equal(result[0].bot.read.q.eid, eid(409)); + }); + + it("does not throw when nobody reported (nothing was read)", () => { + assert.deepEqual(reduceCheckpoints([report(1), report(0)]), [{}]); + }); + + it("merges a queue only a later response reported (fixes checkpointData.read typo)", () => { + const first = { iid: 1, checkpoints: { bot: { read: { a: { checkpoint: eid(5) } }, write: {} } } }; + const second = { iid: 0, checkpoints: { bot: { read: { a: { checkpoint: eid(7) }, b: { checkpoint: eid(9) } }, write: {} } } }; + assert.throws(() => reduceCheckpoints([first, second]), Error, "did not report a read position for bot/b"); + const result = reduceCheckpoints([first, second], { strictCheckpoints: false }); + assert.equal(result[0].bot.read.b.eid, eid(9)); + }); +}); diff --git a/wrappers/fanout.js b/wrappers/fanout.js index 0e63673..d709699 100644 --- a/wrappers/fanout.js +++ b/wrappers/fanout.js @@ -5,6 +5,7 @@ const async = require("async"); const eventIdFormat = "[z/]YYYY/MM/DD/HH/mm/"; const moment = require("moment"); const logger = require("leo-logger")("======= [ leo-fanout ] ======="); +const { ReadTracker } = require("./fanout-read-tracker"); /** * @param {function(BotEvent, LambdaContext, Callback)} handler function to the code handler * @param {function(QueueEvent): any} eventPartition function to return the value representing what partition for the event @@ -28,6 +29,8 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { let leoBotReportComplete = leo.bot.reportComplete; let cronData; let checkpoints = {}; + // Per bot/queue read position of this instance, see fanout-read-tracker.js + let readTrackers = {}; function fixInstanceForLocal(cronData) { // Get fanout data from process env if running locally @@ -38,15 +41,64 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { } } + function getReadTracker(ID, queue) { + const id = String(ID).replace(/_reader$/, ""); + const key = id + "\u0000" + queue; + if (!(key in readTrackers)) { + // Every instance starts reading after the bot's stored checkpoint, which the master passes along in __cron + const read = (cronData.checkpoints || {}).read || {}; + const start = (read["queue:" + queue] || {}).checkpoint; + readTrackers[key] = { id, queue, tracker: new ReadTracker(start) }; + } + return readTrackers[key].tracker; + } + + // Every instance must vote on how far it read, even if it owned nothing. Otherwise the master + // only counts the instances that checkpointed and advances past events the silent owners never saw. + function reportCheckpoints() { + if (opts.allowCheckpoint === true) { + return checkpoints; + } + const now = Date.now(); + Object.keys(readTrackers).forEach(key => { + const { id, queue, tracker } = readTrackers[key]; + const position = tracker.position(); + if (!position) { + return; + } + if (!(id in checkpoints)) { + checkpoints[id] = { + read: {}, + write: {} + }; + } + const read = checkpoints[id].read; + if (!(queue in read)) { + read[queue] = { + records: 0, + started_timestamp: now, + ended_timestamp: now + }; + } + if (!read[queue].checkpoint || position > read[queue].checkpoint) { + read[queue].checkpoint = position; + } + }); + return checkpoints; + } + // Override reading from leo to only read up to the max event id send from the master. leo.streams.fromLeo = leo.read = (ID, queue, opts = {}) => { opts.maxOverride = min(opts.maxOverride, cronData.maxeid); logger.log(`Reading Queue Wrapper. Bot: ${ID}, IID: ${cronData.iid}, Queue: ${queue}, Max: ${opts.maxOverride}`); let reader = leoStreamsFromLeo.call(leo.streams, ID, queue, opts); + const tracker = getReadTracker(ID, queue); let stream = leo.streams.pipe(reader, leo.streams.through((obj, done) => { let partition = Math.abs(hashCode(eventPartition(obj))) % cronData.icount; logger.log("[partition]", partition, "[iid]", cronData.iid, "[eid]", obj.eid, "[icount]", cronData.icount); - if (partition == cronData.iid) { + const owned = partition == cronData.iid; + tracker.read(obj.eid, owned); + if (owned) { logger.log("------ PROCESSING: matched on partition ------"); done(null, obj); } else { @@ -83,6 +135,12 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { botData[event].source_timestamp = params.source_timestamp; botData[event].started_timestamp = params.started_timestamp; botData[event].ended_timestamp = params.ended_timestamp; + if ((params.type || "read") === "read" && params.eid) { + const entry = readTrackers[id + "\u0000" + event]; + if (entry) { + entry.tracker.checkpoint(params.eid); + } + } logger.log(`Checkpoint Wrapper. Bot: ${id}:${cronData.iid}, Queue: ${event}, data: ${JSON.stringify(params)}`); done(); }; @@ -121,6 +179,7 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { logger.debug("Fanout Handler Event", JSON.stringify(event, null, 2)); logger.debug("Fanout Handler Context", JSON.stringify(context, null, 2)); checkpoints = {}; + readTrackers = {}; // If this is a worker then report back the checkpoints or error if (cronData.iid && cronData.icount) { @@ -135,7 +194,7 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { wasCalled = true; let response = { error: err, - checkpoints: checkpoints, + checkpoints: reportCheckpoints(), data: data, iid: cronData.iid }; @@ -153,8 +212,7 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { } return handlerResponse; } else { - let timestamp = moment.utc(); - cronData.maxeid = timestamp.format(eventIdFormat) + timestamp.valueOf(); + cronData.maxeid = getMaxEid(cronData); cronData.iid = 0; logger.log("Fanout Master", cronData.iid); // If this is the master start the workers needed Workers @@ -175,7 +233,7 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { logger.log(`Done with instance 1/${instances}`); resolve({ error: err, - checkpoints: checkpoints, + checkpoints: reportCheckpoints(), data: data, iid: 0 }); @@ -190,12 +248,12 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { }) ]; for (let i = 1; i < instances; i++) { - workers.unshift(invokeSelf(event, i, instances, context)); + workers.unshift(invokeSelf(event, i, instances, context, cronData)); } // Wait for all workers to return and figure out what checkpoint to persist logger.debug(`Waiting on all Fanout workers: count ${workers.length}`); - Promise.all(workers).then(callCheckpointOnResponses(leoBotCheckpoint, callback)).catch((err) => { + Promise.all(workers).then(callCheckpointOnResponses(leoBotCheckpoint, callback, opts)).catch((err) => { logger.error("[err]", err); return callback(err); }); @@ -203,10 +261,10 @@ const fanoutFactory = (handler, eventPartition, opts = {}) => { }); }; -function callCheckpointOnResponses(leoBotCheckpoint, callback) { +function callCheckpointOnResponses(leoBotCheckpoint, callback, opts = {}) { return function(responses) { logger.log("Return from all workers, reducing checkpoints"); - let checkpoints = reduceCheckpoints(responses).map((data) => { + let checkpoints = reduceCheckpoints(responses, opts).map((data) => { logger.log("[data]", data); return Object.keys(data).map((botId) => { logger.log("[botId]", botId); @@ -251,13 +309,18 @@ function callCheckpointOnResponses(leoBotCheckpoint, callback) { * @param {number} iid The instance id of this worker * @param {number} count The total number of workers * @param {*} context Lambda context object + * @param {*} cronData The master's cron data (max eid and start checkpoints for the workers) * @param {function(BotEvent, LambdaContext, Callback)} handler */ -function invokeSelf(event, iid, count, context) { +function invokeSelf(event, iid, count, context, cronData = {}) { + // Workers must get the master's max eid and the start checkpoints, otherwise they read an unbounded window + // and can't report where they started let newEvent = { __cron: { iid, - icount: count + icount: count, + maxeid: cronData.maxeid, + checkpoints: cronData.checkpoints } }; try { @@ -339,9 +402,14 @@ function invokeSelf(event, iid, count, context) { /** * * @param {[checkpoint]} responses Array of responses from the workers + * @param {{strictCheckpoints: boolean}} [opts] strictCheckpoints (default true) throws if an instance did not report a read position for a queue another instance reported * @returns {[checkpoint]} Consolidated checkpoint */ -function reduceCheckpoints(responses) { +function reduceCheckpoints(responses, opts = {}) { + // Check before reducing: the reduce below aliases and mutates the response objects. Errors are reported by the reduce. + if (opts.strictCheckpoints !== false && !responses.some(response => response && response.error)) { + assertAllReported(responses); + } logger.log("[responses]", JSON.stringify(responses, null, 2)); let checkpoints = responses.reduce((agg, curr) => { if (curr && curr.error) { @@ -359,7 +427,7 @@ function reduceCheckpoints(responses) { Object.keys(curr.checkpoints[botId].read || {}).map(queue => { if (!(queue in checkpointData)) { checkpointData[queue] = curr.checkpoints[botId].read[queue]; - checkpointData.read[queue].eid = curr.checkpoints[botId].read[queue].checkpoint; + checkpointData[queue].eid = curr.checkpoints[botId].read[queue].checkpoint; } else { let minCheckpoint = min(checkpointData[queue].checkpoint, curr.checkpoints[botId].read[queue].checkpoint); if (minCheckpoint && minCheckpoint == curr.checkpoints[botId].read[queue].checkpoint) { @@ -393,6 +461,46 @@ function reduceCheckpoints(responses) { } } +/** + * The max eid every instance reads up to. Pinned to the cron invocation token (__cron.ts, in ms) so the window is + * fixed by the trigger rather than by when the master happened to start. Falls back to now if the token is + * missing or in the future. + * @param {{ts?: number|string}} cronData The master's cron data + * @param {moment.Moment} [now] Current time, for testing + * @returns {string} Max eid in the form z/YYYY/MM/DD/HH/mm/ + */ +function getMaxEid(cronData, now = moment.utc()) { + const ts = Number(cronData.ts); + const timestamp = Number.isFinite(ts) && ts > 0 && ts <= now.valueOf() ? moment.utc(ts) : now; + return timestamp.format(eventIdFormat) + timestamp.valueOf(); +} + +/** + * A silent instance must not be counted as a vote to advance. Every instance has to report a read position + * for every bot/queue any instance reported, otherwise the persisted checkpoint could pass events that + * instance never saw. + * @param {[checkpoint]} responses Array of responses from the workers + */ +function assertAllReported(responses) { + const reported = {}; + responses.forEach(response => { + const data = (response && response.checkpoints) || {}; + Object.keys(data).forEach(botId => { + Object.keys(data[botId].read || {}).forEach(queue => { + const key = botId + "/" + queue; + reported[key] = reported[key] || new Set(); + reported[key].add(response.iid); + }); + }); + }); + Object.keys(reported).forEach(key => { + const missing = responses.filter(response => !(response && reported[key].has(response.iid))).map(response => response && response.iid); + if (missing.length) { + throw new Error(`fanout instance(s) [${missing}] did not report a read position for ${key}`); + } + }); +} + /** * @param {string|number} str Converts {str} to a hash code value */ @@ -430,6 +538,7 @@ function min(...args) { } fanoutFactory.reduceCheckpoints = reduceCheckpoints; +fanoutFactory.getMaxEid = getMaxEid; fanoutFactory.callCheckpointOnResponses = callCheckpointOnResponses; module.exports = fanoutFactory;