diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/scripts.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/scripts.ts index 964924945..6818fd029 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/scripts.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/scripts.ts @@ -9,8 +9,14 @@ import type { Callback, Redis, Result } from "@internal/redis"; * declared key gives the caller's single-slot assertion nothing to compare. * 2. Lua never parses JSON. Each script branches only on a short status string and moves * opaque blobs, so every encoding decision stays in TypeScript. - * 3. Every returned slot is coerced with `or ''`. A Lua false or nil TRUNCATES the reply - * array at that position, silently shortening it. + * 3. A missing HGET returns Lua `false`, not `nil` — measured directly against a live + * Redis: `EVAL "return {'a', false, 'c'}"` and a table holding a missing-field HGET + * result both come back as 3 elements; only `EVAL "return {'a', nil, 'c'}"` comes back + * as 1. A `false` element converts to a reply-array null and does NOT shorten anything + * after it — only a genuine Lua nil truncates. Every returned slot is still coerced + * with `or ''` regardless, not to prevent truncation, but so an absent value arrives + * as `''` rather than `null`, giving the TypeScript one shape to decode instead of + * two. * * STORED_COMPLETED is the value written into the record's `status` field and is * UPPERCASE. The outcome tokens below are lowercase and are a separate vocabulary: they 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 0ad4f6e0f..2a62b491a 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts @@ -361,7 +361,7 @@ describe("complete", () => { // No coordinator method calls runAbsorbBlockers/runClear/wpIdemReserve yet — a later task // wires those in. Registered directly on a raw client so the Lua itself is exercised now. -describe("runAbsorbBlockers (direct Lua)", () => { +describe("runAbsorbBlockers, runClear and wpIdemReserve (direct Lua)", () => { const envelope = JSON.stringify(completion()); redisTest( @@ -432,6 +432,126 @@ describe("runAbsorbBlockers (direct Lua)", () => { } ); + redisTest("counts two distinct unreported ids as fully pending", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + registerWaitpointCommands(client); + try { + const keys = runBlockKeys("run_1"); + + const reply = await client.runAbsorbBlockers( + keys.pend, + keys.done, + keys.edge, + "2", + "w_a", + edgeField("w_a", 0), + "{}", + "", + "w_b", + edgeField("w_b", 0), + "{}", + "" + ); + + expect(reply).toEqual(["2", "2"]); + expect(await client.scard(keys.pend)).toBe(2); + } finally { + client.disconnect(); + } + }); + + redisTest( + "counts one reported and one unreported id as one pending, one delivered", + async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + registerWaitpointCommands(client); + try { + const keys = runBlockKeys("run_1"); + + const reply = await client.runAbsorbBlockers( + keys.pend, + keys.done, + keys.edge, + "2", + "w_a", + edgeField("w_a", 0), + "{}", + "", + "w_b", + edgeField("w_b", 0), + "{}", + envelope + ); + + expect(reply).toEqual(["1", "1", "w_b", envelope]); + expect(await client.scard(keys.pend)).toBe(1); + } finally { + client.disconnect(); + } + } + ); + + redisTest( + "counts the same unreported id passed twice as one pending, not two", + async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + registerWaitpointCommands(client); + try { + const keys = runBlockKeys("run_1"); + + const reply = await client.runAbsorbBlockers( + keys.pend, + keys.done, + keys.edge, + "2", + "w_a", + edgeField("w_a", 0), + "{}", + "", + "w_a", + edgeField("w_a", 1), + "{}", + "" + ); + + expect(reply).toEqual(["1", "1"]); + expect(await client.scard(keys.pend)).toBe(1); + } finally { + client.disconnect(); + } + } + ); + + redisTest( + "counts an id already in done, passed unreported, as delivered rather than pending", + async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + registerWaitpointCommands(client); + try { + const keys = runBlockKeys("run_1"); + // A completion that landed between register and absorb — the delivered set + // already has this id before the absorb call ever sees it. + await client.hset(keys.done, "w_a", envelope); + + const reply = await client.runAbsorbBlockers( + keys.pend, + keys.done, + keys.edge, + "1", + "w_a", + edgeField("w_a", 0), + "{}", + "" + ); + + expect(reply).toEqual(["0", "0", "w_a", envelope]); + expect(await client.scard(keys.pend)).toBe(0); + } finally { + client.disconnect(); + } + } + ); + redisTest("rejects an arity mismatch before writing anything", async ({ redisOptions }) => { const client = createRedisClient(redisOptions); registerWaitpointCommands(client); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts index b8d4f8178..506849fd8 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts @@ -151,9 +151,15 @@ export class WaitpointStoreCoordinator { * Exposed for the guard's own test. Delegates through #call rather than calling * assertSingleSlot directly, so a mutation to the guard inside #call fails this test too * — not only the tests that happen to exercise a real script. + * + * With cross-tag (invalid) keys, assertSingleSlot throws synchronously inside #call, + * before any promise exists, and that throw propagates straight out of this method. With + * same-tag (valid) keys, #call would go on to dispatch a real script call; this method + * never returns or awaits that promise, and swallows whatever it eventually settles to, + * so a valid-key call here can never surface as an unhandled rejection in the caller. */ - assertKeysForTest(operation: string, keys: string[]) { - return this.#call(operation as ScriptName, keys); + assertKeysForTest(operation: string, keys: string[]): void { + this.#call(operation as ScriptName, keys).catch(() => undefined); } async createIfAbsent(args: {