From 22dbbc90fac5f6aa5c8f5268fc4ab70c11e5c3ed Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Wed, 20 May 2026 15:03:37 +0100 Subject: [PATCH] feat(redis-worker): mollifier ack marks materialised + grace TTL (Phase B2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MollifierBuffer.ack` previously deleted the entry hash. It now sets `materialised=true` and resets the TTL to a 30s grace window via a new atomic `ackMollifierEntry` Lua script. The entry hash persists past materialisation as a read-fallback safety net for the brief PG replica lag window between drainer-side write and reader-side visibility (Q1 D2). `BufferEntrySchema` gains an optional `materialised` boolean (string "true"/"false" in Redis → boolean in JS). Accept still refuses while *any* entry exists for the runId — including materialised ones — as defense-in-depth against runId reuse. The drainer's "drains one queued entry … and acks" test now asserts `materialised=true` instead of entry deletion. The "re-accept after ack works" test is inverted to "accept refused while a previously-acked entry is still inside its grace TTL". --- .changeset/mollifier-buffer-ack-grace-ttl.md | 5 ++ .../redis-worker/src/mollifier/buffer.test.ts | 62 ++++++++++++++++--- packages/redis-worker/src/mollifier/buffer.ts | 37 ++++++++++- .../src/mollifier/drainer.test.ts | 5 +- .../redis-worker/src/mollifier/schemas.ts | 9 +++ 5 files changed, 106 insertions(+), 12 deletions(-) create mode 100644 .changeset/mollifier-buffer-ack-grace-ttl.md diff --git a/.changeset/mollifier-buffer-ack-grace-ttl.md b/.changeset/mollifier-buffer-ack-grace-ttl.md new file mode 100644 index 000000000..f893d102b --- /dev/null +++ b/.changeset/mollifier-buffer-ack-grace-ttl.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/redis-worker": patch +--- + +Mollifier drainer ack no longer deletes the entry hash. Instead, `MollifierBuffer.ack` sets `materialised=true` on the entry and resets its TTL to a 30s grace window. Entry hashes persist past materialisation as a read-fallback safety net for the brief PG replica-lag window between drainer-side write and reader-side visibility. `BufferEntrySchema` gains an optional `materialised` boolean. diff --git a/packages/redis-worker/src/mollifier/buffer.test.ts b/packages/redis-worker/src/mollifier/buffer.test.ts index b4d7e1a53..86b6c3240 100644 --- a/packages/redis-worker/src/mollifier/buffer.test.ts +++ b/packages/redis-worker/src/mollifier/buffer.test.ts @@ -172,7 +172,41 @@ describe("MollifierBuffer.pop", () => { }); describe("MollifierBuffer.ack", () => { - redisTest("ack deletes the entry", { timeout: 20_000 }, async ({ redisContainer }) => { + redisTest( + "ack marks entry materialised and applies the grace TTL — entry persists as a read-fallback safety net", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + entryTtlSeconds: 600, + logger: new Logger("test", "log"), + }); + + try { + await buffer.accept({ runId: "run_x", envId: "env_a", orgId: "org_1", payload: "{}" }); + await buffer.pop("env_a"); + await buffer.ack("run_x"); + + const after = await buffer.getEntry("run_x"); + expect(after).not.toBeNull(); + expect(after!.materialised).toBe(true); + + // TTL was reset to the grace window — should be at most 30s, well + // under the original 600s entryTtlSeconds. + const ttl = await buffer.getEntryTtlSeconds("run_x"); + expect(ttl).toBeGreaterThan(0); + expect(ttl).toBeLessThanOrEqual(30); + } finally { + await buffer.close(); + } + }, + ); + + redisTest("ack on missing entry is a no-op", { timeout: 20_000 }, async ({ redisContainer }) => { const buffer = new MollifierBuffer({ redisOptions: { host: redisContainer.getHost(), @@ -184,12 +218,12 @@ describe("MollifierBuffer.ack", () => { }); try { - await buffer.accept({ runId: "run_x", envId: "env_a", orgId: "org_1", payload: "{}" }); - await buffer.pop("env_a"); - await buffer.ack("run_x"); - - const after = await buffer.getEntry("run_x"); - expect(after).toBeNull(); + await buffer.ack("run_ghost"); + const stored = await buffer.getEntry("run_ghost"); + expect(stored).toBeNull(); + // Critical: no partial hash created. + const raw = await buffer["redis"].hgetall("mollifier:entries:run_ghost"); + expect(Object.keys(raw)).toHaveLength(0); } finally { await buffer.close(); } @@ -909,9 +943,15 @@ describe("MollifierBuffer.accept idempotency", () => { ); redisTest( - "re-accept after ack works (terminal entry can be re-accepted)", + "accept refused while a previously-acked (materialised) entry is still inside its grace TTL", { timeout: 20_000 }, async ({ redisContainer }) => { + // After ack, the entry hash persists for the grace window as a + // read-fallback safety net (Q1 D2). RunIds are server-generated and + // never collide in practice, but defense-in-depth: accept refuses + // while *any* entry exists for the runId, including materialised + // ones. The entry hash's TTL is now ~30s instead of the original + // entryTtlSeconds. const buffer = new MollifierBuffer({ redisOptions: { host: redisContainer.getHost(), @@ -932,7 +972,6 @@ describe("MollifierBuffer.accept idempotency", () => { await buffer.pop("env_a"); await buffer.ack("run_x"); - // Entry is gone — re-accept should succeed. const reAccept = await buffer.accept({ runId: "run_x", envId: "env_a", @@ -941,7 +980,10 @@ describe("MollifierBuffer.accept idempotency", () => { }); expect(first).toBe(true); - expect(reAccept).toBe(true); + expect(reAccept).toBe(false); + + const stored = await buffer.getEntry("run_x"); + expect(stored!.materialised).toBe(true); } finally { await buffer.close(); } diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index 869975dc8..7b8325958 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/packages/redis-worker/src/mollifier/buffer.ts @@ -14,6 +14,11 @@ export type MollifierBufferOptions = { logger?: Logger; }; +// Grace TTL applied to the entry hash on drainer ack. The entry survives +// this long after materialisation so direct reads (retrieve, trace, etc.) +// have a safety net while PG replica lag settles. Q1 D2. +const ACK_GRACE_TTL_SECONDS = 30; + export class MollifierBuffer { private readonly redis: Redis; private readonly entryTtlSeconds: number; @@ -158,8 +163,15 @@ export class MollifierBuffer { return entries; } + // Marks the entry as materialised (PG row written) and resets its TTL to + // the grace window. Entry hash persists past ack as a read-fallback + // safety net for the brief PG replica-lag window between drainer-side + // write and reader-side visibility (Q1 D2). async ack(runId: string): Promise { - await this.redis.del(`mollifier:entries:${runId}`); + await this.redis.ackMollifierEntry( + `mollifier:entries:${runId}`, + String(ACK_GRACE_TTL_SECONDS), + ); } async requeue(runId: string): Promise { @@ -353,6 +365,24 @@ export class MollifierBuffer { `, }); + this.redis.defineCommand("ackMollifierEntry", { + numberOfKeys: 1, + lua: ` + local entryKey = KEYS[1] + local graceTtlSeconds = tonumber(ARGV[1]) + + -- Guard: never create a partial entry. If the hash expired between + -- pop and ack, the run is gone — nothing to mark materialised. + if redis.call('EXISTS', entryKey) == 0 then + return 0 + end + + redis.call('HSET', entryKey, 'materialised', 'true') + redis.call('EXPIRE', entryKey, graceTtlSeconds) + return 1 + `, + }); + this.redis.defineCommand("failMollifierEntry", { numberOfKeys: 1, lua: ` @@ -427,6 +457,11 @@ declare module "@internal/redis" { orgEnvsPrefix: string, callback?: Callback, ): Result; + ackMollifierEntry( + entryKey: string, + graceTtlSeconds: string, + callback?: Callback, + ): Result; failMollifierEntry( entryKey: string, errorPayload: string, diff --git a/packages/redis-worker/src/mollifier/drainer.test.ts b/packages/redis-worker/src/mollifier/drainer.test.ts index c8f68977f..db70bd9c3 100644 --- a/packages/redis-worker/src/mollifier/drainer.test.ts +++ b/packages/redis-worker/src/mollifier/drainer.test.ts @@ -87,8 +87,11 @@ describe("MollifierDrainer.runOnce", () => { payload: { foo: 1 }, }); + // After ack the entry persists as a read-fallback safety net with + // materialised=true and a fresh grace TTL (Q1 D2 / Phase B2). const entry = await buffer.getEntry("run_1"); - expect(entry).toBeNull(); + expect(entry).not.toBeNull(); + expect(entry!.materialised).toBe(true); } finally { await buffer.close(); } diff --git a/packages/redis-worker/src/mollifier/schemas.ts b/packages/redis-worker/src/mollifier/schemas.ts index 23a71a816..b67a9ed79 100644 --- a/packages/redis-worker/src/mollifier/schemas.ts +++ b/packages/redis-worker/src/mollifier/schemas.ts @@ -27,6 +27,10 @@ const stringToDate = z.string().transform((v, ctx) => { return d; }); +const stringToBool = z + .union([z.literal("true"), z.literal("false")]) + .transform((v) => v === "true"); + const stringToError = z.string().transform((v, ctx) => { try { return BufferEntryError.parse(JSON.parse(v)); @@ -47,6 +51,11 @@ export const BufferEntrySchema = z.object({ // Microsecond epoch matching the ZSET queue score. Stable across // requeues — the score never moves once set at accept time. createdAtMicros: stringToInt, + // Drainer-ack flag: `true` once the drainer has materialised this run + // into PG. The hash persists for a short grace TTL after ack so direct + // reads (retrieve, trace, etc.) still resolve while PG replica lag + // settles. Absent on pre-ack entries. + materialised: stringToBool.default("false"), lastError: stringToError.optional(), });