feat(redis-worker): mollifier ack marks materialised + grace TTL (Phase B2)

`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

@@ -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.

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();
       }

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<void> {
-    await this.redis.del(`mollifier:entries:${runId}`);
+    await this.redis.ackMollifierEntry(
+      `mollifier:entries:${runId}`,
+      String(ACK_GRACE_TTL_SECONDS),
+    );
   }
 
   async requeue(runId: string): Promise<void> {
@@ -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<number>,
     ): Result<number, Context>;
+    ackMollifierEntry(
+      entryKey: string,
+      graceTtlSeconds: string,
+      callback?: Callback<number>,
+    ): Result<number, Context>;
     failMollifierEntry(
       entryKey: string,
       errorPayload: string,

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();
     }

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(),
 });