fix(redis-worker): self-heal stale idempotency lookups + default createdAtMicros

Address AI review findings on PR #3752:

- accept(): if an idempotency lookup survives its entry hash being evicted
  (maxmemory), the lookup is stale — rebind to the new run instead of
  returning a dead existingRunId that blocks the key forever. Mirrors the
  self-heal lookupIdempotency already does. (CodeRabbit)
- lookupIdempotency(): clear a stale lookup with a compare-and-delete
  (delMollifierKeyIfEquals Lua) so a concurrent accept that rebinds the
  key between our GET and DEL isn't clobbered. (CodeRabbit)
- schemas: default createdAtMicros to "0" so an entry written before the
  field existed (or surviving across the deploy that added it) still
  parses on pop instead of being silently dropped. (Devin)
- rename the requeue-ordering test to "retry priority" — RPUSH-to-tail
  pops the requeued entry ahead of newer items; that's deliberate retry
  priority, not FIFO relative to the rest of the queue. (CodeRabbit nit)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

packages/redis-worker/src/mollifier/buffer.test.ts

@@ -30,6 +30,24 @@ describe("schemas", () => {
     expect(parsed.createdAtMicros).toBe(1747044000000000);
   });
 
+  it("BufferEntrySchema defaults createdAtMicros for entries written before the field existed", () => {
+    // Backward compat: an entry written by an accept Lua predating
+    // createdAtMicros (only the original 7 fields) must still parse on
+    // pop rather than being silently dropped.
+    const raw = {
+      runId: "run_old",
+      envId: "env_1",
+      orgId: "org_1",
+      payload: serialiseSnapshot({}),
+      status: "QUEUED",
+      attempts: "0",
+      createdAt: "2026-05-11T10:00:00.000Z",
+      // no createdAtMicros
+    };
+    const parsed = BufferEntrySchema.parse(raw);
+    expect(parsed.createdAtMicros).toBe(0);
+  });
+
   it("BufferEntrySchema parses a FAILED entry with lastError", () => {
     const raw = {
       runId: "run_abc",
@@ -560,13 +578,15 @@ describe("MollifierBuffer.requeue on missing entry", () => {
 
 describe("MollifierBuffer.requeue ordering", () => {
   redisTest(
-    "requeued entry pops next (RPUSH to the RPOP/tail end), preserving FIFO",
+    "requeued entry gets retry priority (RPUSH to the RPOP/tail end), popping ahead of newer items",
     { timeout: 20_000 },
     async ({ redisContainer }) => {
-      // LIST FIFO: accept LPUSHes at the head, pop RPOPs from the tail, so
-      // the first-accepted entry pops first. requeue RPUSHes back to the
-      // tail, so a transiently failed entry pops next rather than going to
-      // the back. `maxAttempts` in the drainer bounds the retry loop for a
+      // LIST: accept LPUSHes at the head, pop RPOPs from the tail, so the
+      // first-accepted entry pops first. requeue RPUSHes back to the tail,
+      // giving a transiently failed entry *retry priority* — it pops next,
+      // ahead of newer queued items, rather than going to the back. (This
+      // is deliberately not FIFO relative to the rest of the queue.)
+      // `maxAttempts` in the drainer bounds the retry loop for a
       // persistently failing entry (after which it goes to `fail`, not requeue).
       const buffer = new MollifierBuffer({
         redisOptions: {
@@ -1486,6 +1506,78 @@ describe("MollifierBuffer idempotency lookup", () => {
       }
     },
   );
+
+  redisTest(
+    "accept self-heals a stale lookup: a new run rebinds when the bound entry was evicted",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      // If an entry hash is evicted (maxmemory) but its idempotency lookup
+      // survives, a fresh accept with the same key must NOT return the dead
+      // runId (which would block the key forever) — it should rebind to the
+      // new run and accept it.
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        logger: new Logger("test", "log"),
+      });
+      const idem = { idempotencyKey: "kheal", taskIdentifier: "t" };
+      try {
+        await buffer.accept({ runId: "heal_old", envId: "env_h", orgId: "org_1", payload: "{}", ...idem });
+        // Simulate eviction of the entry hash while the lookup survives.
+        await buffer["redis"].del("mollifier:entries:heal_old");
+        const lookupKey = idempotencyLookupKeyFor({ envId: "env_h", ...idem });
+        expect(await buffer["redis"].get(lookupKey)).toBe("heal_old");
+
+        // A fresh accept with the same key rebinds rather than deduping
+        // onto the dead run.
+        const result = await buffer.accept({
+          runId: "heal_new",
+          envId: "env_h",
+          orgId: "org_1",
+          payload: "{}",
+          ...idem,
+        });
+        expect(result).toEqual({ kind: "accepted" });
+        expect(await buffer["redis"].get(lookupKey)).toBe("heal_new");
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "accept still dedups when the bound entry is live",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      // The self-heal must not weaken normal dedup: a live bound entry
+      // still wins, and the loser gets its runId back.
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        logger: new Logger("test", "log"),
+      });
+      const idem = { idempotencyKey: "klive", taskIdentifier: "t" };
+      try {
+        await buffer.accept({ runId: "live_win", envId: "env_h", orgId: "org_1", payload: "{}", ...idem });
+        const result = await buffer.accept({
+          runId: "live_lose",
+          envId: "env_h",
+          orgId: "org_1",
+          payload: "{}",
+          ...idem,
+        });
+        expect(result).toEqual({ kind: "duplicate_idempotency", existingRunId: "live_win" });
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
 });
 
 describe("MollifierBuffer.casSetMetadata", () => {

packages/redis-worker/src/mollifier/buffer.ts

@@ -160,6 +160,7 @@ export class MollifierBuffer {
       String(createdAtMicros),
       "mollifier:org-envs:",
       idempotencyLookupKey,
+      "mollifier:entries:",
     );
     // Lua returns 1 (accepted), 0 (duplicate runId), or a string runId
     // (duplicate idempotency — value is the existing winner's runId).
@@ -391,15 +392,17 @@ export class MollifierBuffer {
   // Resolve a buffered run by (env, task, idempotencyKey) tuple. Used by
   // `IdempotencyKeyConcern.handleTriggerRequest` after the PG check
   // misses — same key may belong to a buffered run waiting to drain. The
-  // lookup self-heals: if the lookup points at an entry hash that's
-  // expired, we DEL the lookup and report a miss.
+  // lookup self-heals: if the lookup points at an entry hash that's gone,
+  // we clear the lookup and report a miss. The clear is a compare-and-
+  // delete (only if the key still holds the stale runId we observed) so a
+  // fresh accept that rebinds the key between our GET and DEL isn't wiped.
   async lookupIdempotency(input: IdempotencyLookupInput): Promise<string | null> {
     const lookupKey = idempotencyLookupKeyFor(input);
     const runId = await this.redis.get(lookupKey);
     if (!runId) return null;
     const entry = await this.getEntry(runId);
     if (!entry) {
-      await this.redis.del(lookupKey);
+      await this.redis.delMollifierKeyIfEquals(lookupKey, runId);
       return null;
     }
     return runId;
@@ -502,6 +505,7 @@ export class MollifierBuffer {
         local createdAtMicros = ARGV[6]
         local orgEnvsPrefix = ARGV[7]
         local idempotencyLookupKey = ARGV[8] or ''
+        local entryPrefix = ARGV[9]
 
         -- Idempotent: refuse if an entry for this runId already exists in any
         -- state. Caller-side dedup is also enforced via API idempotency keys,
@@ -519,7 +523,14 @@ export class MollifierBuffer {
         if idempotencyLookupKey ~= '' then
           local existing = redis.call('GET', idempotencyLookupKey)
           if existing then
-            return existing
+            -- Self-heal: only honour the binding if its entry hash still
+            -- exists. If the entry was evicted (maxmemory) but the lookup
+            -- survived, the binding is stale — fall through and rebind to
+            -- this run rather than returning a dead runId that would block
+            -- the key indefinitely. Mirrors lookupIdempotency's self-heal.
+            if redis.call('EXISTS', entryPrefix .. existing) == 1 then
+              return existing
+            end
           end
           redis.call('SET', idempotencyLookupKey, runId)
         end
@@ -935,6 +946,20 @@ export class MollifierBuffer {
       `,
     });
 
+    // Compare-and-delete: DEL the key only if it still holds the expected
+    // value. Used by lookupIdempotency's stale-lookup self-heal so a
+    // concurrent accept that rebinds the key between the reader's GET and
+    // this DEL isn't clobbered.
+    this.redis.defineCommand("delMollifierKeyIfEquals", {
+      numberOfKeys: 1,
+      lua: `
+        if redis.call('GET', KEYS[1]) == ARGV[1] then
+          return redis.call('DEL', KEYS[1])
+        end
+        return 0
+      `,
+    });
+
     this.redis.defineCommand("mollifierEvaluateTrip", {
       numberOfKeys: 2,
       lua: `
@@ -974,6 +999,7 @@ declare module "@internal/redis" {
       createdAtMicros: string,
       orgEnvsPrefix: string,
       idempotencyLookupKey: string,
+      entryPrefix: string,
       callback?: Callback<number | string>,
     ): Result<number | string, Context>;
     popAndMarkDraining(
@@ -1039,6 +1065,11 @@ declare module "@internal/redis" {
       errorPayload: string,
       callback?: Callback<number>,
     ): Result<number, Context>;
+    delMollifierKeyIfEquals(
+      key: string,
+      expected: string,
+      callback?: Callback<number>,
+    ): Result<number, Context>;
     mollifierEvaluateTrip(
       rateKey: string,
       trippedKey: string,

packages/redis-worker/src/mollifier/schemas.ts

@@ -48,9 +48,12 @@ export const BufferEntrySchema = z.object({
   status: BufferEntryStatus,
   attempts: stringToInt,
   createdAt: stringToDate,
-  // Microsecond epoch matching the ZSET queue score. Stable across
-  // requeues — the score never moves once set at accept time.
-  createdAtMicros: stringToInt,
+  // Microsecond epoch of accept time, kept as a hash field for dwell
+  // metrics. Not a queue sort key (the queue is a FIFO LIST). Defaulted
+  // so an entry written by an accept Lua predating this field — or one
+  // surviving across the deploy that introduced it — still parses instead
+  // of being silently dropped on pop.
+  createdAtMicros: stringToInt.default("0"),
   // 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