From bedce8b763ecbcdb11d17cf94c0444fbf24b2cf7 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 21:15:02 +0100 Subject: [PATCH] fix(run-engine): replace the reported-envelope symbol with a JSON-safe box Review fix round 1 on registerBlocks: swap EMPTY_REPORTED_MARKER for reported?: { completion?: WaitpointCompletion }, which round-trips through JSON.stringify/parse instead of silently dropping. Tightens the no-envelope regression test to check for a fabricated completion, pins the new reportedFlag='1'/empty-envelope wire shape at the direct-Lua level, documents and tests the safe partial-failure residue when registerBlocks throws mid-loop, and asserts the watcher fan-out in the multi-index merge test. --- .../storeCoordinator.test.ts | 84 ++++++++++++++++++- .../waitpointCoordinator/storeCoordinator.ts | 29 +++---- 2 files changed, 93 insertions(+), 20 deletions(-) diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts index 5a1258b06..f8b02381e 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts @@ -501,6 +501,36 @@ describe("reply framing (direct Lua — pins the wire shape the coordinator deco } ); + redisTest( + "reported flag '1' with an empty envelope still delivers, not pends", + async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + registerWaitpointCommands(client); + try { + const keys = runBlockKeys("run_1"); + + // The bug this task fixed: COMPLETED-with-no-envelope must take the reported + // branch on the flag alone, not on the envelope being non-empty. + const reply = await client.runAbsorbBlockers( + keys.pend, + keys.done, + keys.edge, + "1", + "w_a", + edgeField("w_a", 0), + "{}", + "1", + "" + ); + + expect(reply).toEqual(["0", "0", "w_a", ""]); + expect(await client.scard(keys.pend)).toBe(0); + } finally { + client.disconnect(); + } + } + ); + redisTest( "counts the same unreported id passed twice as one pending, not two", async ({ redisOptions }) => { @@ -819,7 +849,7 @@ describe("absorbBlockers", () => { try { const result = await store.absorbBlockers({ runId: RUN_ID, - edges: [edge("w_a", { reported: completion() }), edge("w_b")], + edges: [edge("w_a", { reported: { completion: completion() } }), edge("w_b")], }); expect(result.pendingOfRequested).toBe(1); @@ -837,8 +867,8 @@ describe("absorbBlockers", () => { const result = await store.absorbBlockers({ runId: RUN_ID, edges: [ - edge("w_a", { batchIndex: 0, reported: completion() }), - edge("w_a", { batchIndex: 1, reported: completion() }), + edge("w_a", { batchIndex: 0, reported: { completion: completion() } }), + edge("w_a", { batchIndex: 1, reported: { completion: completion() } }), ], }); @@ -926,7 +956,7 @@ describe("absorbBlockers", () => { const result = await store.absorbBlockers({ runId: RUN_ID, - edges: [edge("w_a", { reported: completion() })], + edges: [edge("w_a", { reported: { completion: completion() } })], }); // Nothing THIS call requested is pending (w_a arrived already delivered), but the @@ -1200,6 +1230,10 @@ describe("registerBlocks: a COMPLETED waitpoint with no envelope never blocks (r expect(result.pendingOfRequested).toBe(0); expect(result.storePendingTotal).toBe(0); expect(result.alreadyDelivered.map((d) => d.waitpointId)).toEqual(["w_a"]); + // The whole point: no fabricated envelope, and the delivery is real on the run + // shard, not just absent from pending. + expect(result.alreadyDelivered[0]!.completion).toBeUndefined(); + expect((await store.readBlockState(RUN_ID)).deliveredIds).toEqual(["w_a"]); } finally { await store.quit(); } @@ -1284,6 +1318,43 @@ describe("registerBlocks: the two orderings", () => { } }); + redisTest( + "a throw mid-loop leaves the earlier watcher registered, and that residue is safe", + async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + await store.createIfAbsent({ record: record("w_ok"), status: "PENDING" }); + + await expect( + store.registerBlocks({ runId: RUN_ID, edges: [edge("w_ok"), edge("w_missing")] }) + ).rejects.toThrow(WaitpointNotFoundError); + + // registerBlocks throws before absorbBlockers ever runs, so the run's own shard + // is untouched. + const state = await store.readBlockState(RUN_ID); + expect(state.pendingIds).toEqual([]); + expect(state.edges).toEqual([]); + + // But w_ok's watcher WAS registered on w_ok's own shard before the throw. + const completed = await store.complete({ waitpointId: "w_ok", completion: completion() }); + expect(completed.watchers.map((w) => w.runId)).toEqual([RUN_ID]); + + // Delivering it writes a `done` entry for a run that was never blocked on it — + // inert residue, not a false resume: no edge ever named it, and clearBlockState's + // reconcile would drop it the moment this run's block state is next drained. + const delivered = await store.deliverCompletion({ + runId: RUN_ID, + waitpointId: "w_ok", + completion: completed.completion!, + }); + expect(delivered.storePendingTotal).toBe(0); + expect((await store.readBlockState(RUN_ID)).deliveredIds).toEqual(["w_ok"]); + } finally { + await store.quit(); + } + } + ); + redisTest("is idempotent when run twice", async ({ redisOptions }) => { const store = coordinator(redisOptions); try { @@ -1347,6 +1418,11 @@ describe("multi-index merge, end to end into the executor shape", () => { waitpointId: "w_child", completion: completion({ output: null }), }); + // The cross-shard fact this test claims to prove: two registers for the same + // waitpoint at different indexes fanned out into two distinct watcher entries. + expect( + completed.watchers.map((w) => w.batchIndex).sort((a, b) => (a ?? 0) - (b ?? 0)) + ).toEqual([0, 2]); await store.deliverCompletion({ runId: RUN_ID, waitpointId: "w_child", diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts index 58724f9ff..da79d3c51 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts @@ -77,13 +77,6 @@ export type WatcherEntry = { createdAt: string; }; -/** - * Marks an edge as reported COMPLETED with no completion envelope — the shape a - * FINISHED-healing create can produce. Distinct from `undefined` (never reported), so the - * pending decision can key on outcome alone rather than on envelope presence. - */ -export const EMPTY_REPORTED_MARKER = Symbol("waitpoint-reported-no-envelope"); - export type CreateIfAbsentResult = | { outcome: "created" } | { @@ -115,8 +108,10 @@ export type BlockEdge = { createdAt: string; type: WaitpointRecordInput["type"]; completedAfter?: string; - /** Set when the register step already reported this waitpoint COMPLETED. */ - reported?: WaitpointCompletion | typeof EMPTY_REPORTED_MARKER; + // Set when the register step already reported this waitpoint COMPLETED. The box, not + // `completion`, carries the "reported" fact: box present + no completion means + // COMPLETED-with-no-envelope, box absent means never reported. + reported?: { completion?: WaitpointCompletion }; }; export type AbsorbResult = { @@ -375,10 +370,7 @@ export class WaitpointStoreCoordinator { for (const item of args.edges) { const { reported, ...stored } = item; const reportedFlag = reported !== undefined ? "1" : "0"; - const reportedJson = - reported !== undefined && reported !== EMPTY_REPORTED_MARKER - ? JSON.stringify(reported) - : ""; + const reportedJson = reported?.completion ? JSON.stringify(reported.completion) : ""; argv.push( item.waitpointId, edgeField(item.waitpointId, item.batchIndex), @@ -411,11 +403,16 @@ export class WaitpointStoreCoordinator { * Register on every waitpoint's own shard FIRST, then absorb on the run's shard. The * order is the protocol: a completion that lands in between finds the watcher already * registered, so it delivers onto the run's shard, and the absorb sees that delivery and - * never marks the waitpoint pending. Reversing the two would open the window where a - * completion is missed by both steps. + * never marks the waitpoint pending. * * The register keys the decision to skip the pending set on OUTCOME, never on whether a * completion envelope came back — a waitpoint can be reported COMPLETED with none. + * + * A throw partway through (a missing waitpoint) intentionally leaves any + * already-registered watchers in place rather than unwinding them. That's safe: a later + * `complete` on one of those waitpoints still delivers correctly, and if it lands before + * this run ever retries `registerBlocks`, the stray `done` entry it writes is inert until + * a future absorb or `clearBlockState`'s reconcile reads it — never a false resume. */ async registerBlocks(args: { runId: string; edges: BlockEdge[] }): Promise { const registered: BlockEdge[] = []; @@ -431,7 +428,7 @@ export class WaitpointStoreCoordinator { registered.push( result.outcome === "completed" - ? { ...item, reported: result.completion ?? EMPTY_REPORTED_MARKER } + ? { ...item, reported: { completion: result.completion } } : item ); }