feat(redis-worker): migrate mollifier queue from LIST to ZSET (Phase B1)

Per-env queue `mollifier:queue:{envId}` switches from a Redis LIST
(LPUSH/RPOP) to a sorted set keyed by `createdAtMicros`. Pop semantics
are unchanged (FIFO by creation time, now via ZPOPMIN). Entry hashes
carry a new `createdAtMicros` field equal to the score.

Requeue keeps the original score — createdAt is immutable across
retries, so a retried entry continues to pop next by virtue of being
the oldest. `maxAttempts` in the drainer bounds the retry loop. The
inverted "FIFO retry" test reflects the new (correct) semantics under
the score-equals-createdAt invariant.

`listEntriesForEnv` reads via ZREVRANGE (newest-first). Orphan-handling
tests that injected via LPUSH now use ZADD; queue-depth assertions
switch from LLEN to ZCARD.

This is the substrate for the listing pagination work in Phase E and
for the snapshot-mutate work in Phase B3.

Author: d-cs

.changeset/mollifier-buffer-zset-migration.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Migrate the mollifier per-env queue from a Redis LIST to a ZSET scored by `createdAtMicros`. Internal change; the public `MollifierBuffer` API is unchanged. Entry hashes now carry a `createdAtMicros` field matching the ZSET score; `accept` uses `ZADD`, `pop` uses `ZPOPMIN`, `requeue` reuses the original score so retries do not advance the entry's creation timestamp. Listing (`listEntriesForEnv`) reads via `ZREVRANGE`. This unlocks O(log N + pageSize) paginated listing of buffered runs without changing FIFO drain semantics.

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

@@ -20,12 +20,14 @@ describe("schemas", () => {
       status: "QUEUED",
       attempts: "0",
       createdAt: "2026-05-11T10:00:00.000Z",
+      createdAtMicros: "1747044000000000",
     };
     const parsed = BufferEntrySchema.parse(raw);
     expect(parsed.runId).toBe("run_abc");
     expect(parsed.status).toBe("QUEUED");
     expect(parsed.attempts).toBe(0);
     expect(parsed.createdAt).toBeInstanceOf(Date);
+    expect(parsed.createdAtMicros).toBe(1747044000000000);
   });
 
   it("BufferEntrySchema parses a FAILED entry with lastError", () => {
@@ -37,6 +39,7 @@ describe("schemas", () => {
       status: "FAILED",
       attempts: "3",
       createdAt: "2026-05-11T10:00:00.000Z",
+      createdAtMicros: "1747044000000000",
       lastError: JSON.stringify({ code: "P2024", message: "connection lost" }),
     };
     const parsed = BufferEntrySchema.parse(raw);
@@ -210,7 +213,7 @@ describe("MollifierBuffer.pop orphan handling", () => {
 
       try {
         // Simulate a TTL-expired orphan: queue ref exists, entry hash does not.
-        await buffer["redis"].lpush("mollifier:queue:env_a", "run_orphan");
+        await buffer["redis"].zadd("mollifier:queue:env_a", 1, "run_orphan");
 
         const popped = await buffer.pop("env_a");
         expect(popped).toBeNull();
@@ -220,7 +223,7 @@ describe("MollifierBuffer.pop orphan handling", () => {
         expect(Object.keys(raw)).toHaveLength(0);
 
         // Queue is drained — the loop pops orphans until empty.
-        const qLen = await buffer["redis"].llen("mollifier:queue:env_a");
+        const qLen = await buffer["redis"].zcard("mollifier:queue:env_a");
         expect(qLen).toBe(0);
       } finally {
         await buffer.close();
@@ -243,20 +246,20 @@ describe("MollifierBuffer.pop orphan handling", () => {
       });
 
       try {
-        // Layout (oldest-first, since RPOP takes from tail): orphan, valid, orphan.
-        // LPUSH puts items at the head, so to get RPOP order [orphan_a, valid, orphan_b]
-        // we LPUSH in reverse: orphan_b first, then valid, then orphan_a.
-        await buffer["redis"].lpush("mollifier:queue:env_a", "orphan_b");
+        // Layout by score (lowest-first, since ZPOPMIN takes the min):
+        // orphan_a (score 1) → valid (score = its createdAtMicros, large) → orphan_b (score 1e18).
+        // First pop skips orphan_a, returns valid; orphan_b remains.
+        await buffer["redis"].zadd("mollifier:queue:env_a", 1, "orphan_a");
         await buffer.accept({ runId: "valid", envId: "env_a", orgId: "org_1", payload: "{}" });
-        await buffer["redis"].lpush("mollifier:queue:env_a", "orphan_a");
+        await buffer["redis"].zadd("mollifier:queue:env_a", 1e18, "orphan_b");
 
         const popped = await buffer.pop("env_a");
         expect(popped).not.toBeNull();
         expect(popped!.runId).toBe("valid");
         expect(popped!.status).toBe("DRAINING");
 
         // The trailing orphan_b is still in the queue (single pop call).
-        const remaining = await buffer["redis"].llen("mollifier:queue:env_a");
+        const remaining = await buffer["redis"].zcard("mollifier:queue:env_a");
         expect(remaining).toBe(1);
 
         // A second pop drains the trailing orphan_b. The queue is now
@@ -458,9 +461,13 @@ describe("MollifierBuffer.requeue on missing entry", () => {
 
 describe("MollifierBuffer.requeue ordering", () => {
   redisTest(
-    "requeued entry is popped AFTER other queued entries on the same env (FIFO retry)",
+    "requeued entry retains its original createdAt and pops next (oldest-first by createdAt)",
     { timeout: 20_000 },
     async ({ redisContainer }) => {
+      // Score == createdAtMicros; requeue does not bump the score. The
+      // oldest entry continues to pop first across retries. `maxAttempts`
+      // in the drainer bounds the retry loop for a persistently failing
+      // entry (after which it goes to the `fail` path, not requeue).
       const buffer = new MollifierBuffer({
         redisOptions: {
           host: redisContainer.getHost(),
@@ -473,20 +480,23 @@ describe("MollifierBuffer.requeue ordering", () => {
 
       try {
         await buffer.accept({ runId: "a", envId: "env_a", orgId: "org_1", payload: "{}" });
+        await new Promise((r) => setTimeout(r, 2));
         await buffer.accept({ runId: "b", envId: "env_a", orgId: "org_1", payload: "{}" });
+        await new Promise((r) => setTimeout(r, 2));
         await buffer.accept({ runId: "c", envId: "env_a", orgId: "org_1", payload: "{}" });
 
         const first = await buffer.pop("env_a");
         expect(first!.runId).toBe("a");
 
         await buffer.requeue("a");
 
+        // a still has the smallest createdAtMicros → pops next.
         const next = await buffer.pop("env_a");
-        expect(next!.runId).toBe("b");
+        expect(next!.runId).toBe("a");
         const after = await buffer.pop("env_a");
-        expect(after!.runId).toBe("c");
+        expect(after!.runId).toBe("b");
         const last = await buffer.pop("env_a");
-        expect(last!.runId).toBe("a");
+        expect(last!.runId).toBe("c");
       } finally {
         await buffer.close();
       }
@@ -1026,6 +1036,124 @@ describe("MollifierBuffer envs set lifecycle", () => {
   );
 });
 
+describe("MollifierBuffer ZSET storage", () => {
+  redisTest(
+    "queue key is a ZSET scored by entry's createdAtMicros",
+    { 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: "z1", envId: "env_z", orgId: "org_1", payload: "{}" });
+
+        // ZSET-only commands must succeed against the queue key.
+        const card = await buffer["redis"].zcard("mollifier:queue:env_z");
+        expect(card).toBe(1);
+
+        const score = await buffer["redis"].zscore("mollifier:queue:env_z", "z1");
+        expect(score).not.toBeNull();
+        const scoreNum = Number(score);
+        expect(Number.isFinite(scoreNum)).toBe(true);
+
+        // Score matches the entry hash's createdAtMicros field.
+        const micros = await buffer["redis"].hget("mollifier:entries:z1", "createdAtMicros");
+        expect(micros).not.toBeNull();
+        expect(Number(micros)).toBe(scoreNum);
+
+        // Score is plausibly recent (within last minute as microseconds).
+        const nowMicros = Date.now() * 1000;
+        expect(scoreNum).toBeGreaterThan(nowMicros - 60_000_000);
+        expect(scoreNum).toBeLessThanOrEqual(nowMicros + 1_000_000);
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "pop returns entries in ascending createdAtMicros order (FIFO by time, not by member)",
+    { 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 {
+        // Insert runIds in reverse-lex order to prove ordering is by score, not member.
+        await buffer.accept({ runId: "zzz", envId: "env_o", orgId: "org_1", payload: "{}" });
+        await new Promise((r) => setTimeout(r, 5));
+        await buffer.accept({ runId: "mmm", envId: "env_o", orgId: "org_1", payload: "{}" });
+        await new Promise((r) => setTimeout(r, 5));
+        await buffer.accept({ runId: "aaa", envId: "env_o", orgId: "org_1", payload: "{}" });
+
+        const first = await buffer.pop("env_o");
+        expect(first!.runId).toBe("zzz");
+        const second = await buffer.pop("env_o");
+        expect(second!.runId).toBe("mmm");
+        const third = await buffer.pop("env_o");
+        expect(third!.runId).toBe("aaa");
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "requeue keeps original score; createdAt is immutable across retries",
+    { 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: "rq", envId: "env_rq", orgId: "org_1", payload: "{}" });
+        const originalScore = Number(
+          await buffer["redis"].zscore("mollifier:queue:env_rq", "rq"),
+        );
+        const originalMicros = Number(
+          await buffer["redis"].hget("mollifier:entries:rq", "createdAtMicros"),
+        );
+
+        await buffer.pop("env_rq");
+        await new Promise((r) => setTimeout(r, 5));
+        await buffer.requeue("rq");
+
+        const newScore = Number(
+          await buffer["redis"].zscore("mollifier:queue:env_rq", "rq"),
+        );
+        const newMicros = Number(
+          await buffer["redis"].hget("mollifier:entries:rq", "createdAtMicros"),
+        );
+        expect(newScore).toBe(originalScore);
+        expect(newMicros).toBe(originalMicros);
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+});
+
 describe("MollifierBuffer.listEntriesForEnv", () => {
   redisTest(
     "returns up to maxCount entries from the queue without consuming them",

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

@@ -53,7 +53,15 @@ export class MollifierBuffer {
     const entryKey = `mollifier:entries:${input.runId}`;
     const queueKey = `mollifier:queue:${input.envId}`;
     const orgsKey = "mollifier:orgs";
-    const createdAt = new Date().toISOString();
+    const nowMs = Date.now();
+    const createdAt = new Date(nowMs).toISOString();
+    // Microsecond epoch. JS only has millisecond precision, so multiple
+    // accepts in the same ms share a score; ZSET ties resolve by member
+    // (runId) lex order, which is deterministic and acceptable for FIFO
+    // pop. The hash carries the same value as `createdAtMicros` so the
+    // listing helper (Phase E) can read a stable per-run timestamp
+    // without re-fetching the score.
+    const createdAtMicros = nowMs * 1000;
     const result = await this.redis.acceptMollifierEntry(
       entryKey,
       queueKey,
@@ -63,6 +71,7 @@ export class MollifierBuffer {
       input.orgId,
       input.payload,
       createdAt,
+      String(createdAtMicros),
       String(this.entryTtlSeconds),
       "mollifier:org-envs:",
     );
@@ -129,14 +138,18 @@ export class MollifierBuffer {
   }
 
   // Read-only listing of currently-queued entries for a single env. Used by
-  // the dashboard's "Recently queued" surface — LRANGE is non-destructive,
-  // so the drainer still pops these entries in order. Returns up to
-  // `maxCount` entries (the most-recently-queued ones, since accept LPUSHes
-  // onto the head). Each entry hash is fetched separately; a `null` from
-  // getEntry (TTL expired between LRANGE and HGETALL) is skipped.
+  // the dashboard's "Recently queued" surface — non-destructive, so the
+  // drainer still pops these entries in order. Returns up to `maxCount`
+  // entries newest-first (highest score, which is `createdAtMicros`).
+  // Each entry hash is fetched separately; a `null` from getEntry (TTL
+  // expired between ZREVRANGE and HGETALL) is skipped.
   async listEntriesForEnv(envId: string, maxCount: number): Promise<BufferEntry[]> {
     if (maxCount <= 0) return [];
-    const runIds = await this.redis.lrange(`mollifier:queue:${envId}`, 0, maxCount - 1);
+    const runIds = await this.redis.zrevrange(
+      `mollifier:queue:${envId}`,
+      0,
+      maxCount - 1,
+    );
     const entries: BufferEntry[] = [];
     for (const runId of runIds) {
       const entry = await this.getEntry(runId);
@@ -207,8 +220,9 @@ export class MollifierBuffer {
         local orgId = ARGV[3]
         local payload = ARGV[4]
         local createdAt = ARGV[5]
-        local ttlSeconds = tonumber(ARGV[6])
-        local orgEnvsPrefix = ARGV[7]
+        local createdAtMicros = ARGV[6]
+        local ttlSeconds = tonumber(ARGV[7])
+        local orgEnvsPrefix = ARGV[8]
 
         -- Idempotent: refuse if an entry for this runId already exists in any
         -- state. Caller-side dedup is also enforced via API idempotency keys,
@@ -224,9 +238,15 @@ export class MollifierBuffer {
           'payload', payload,
           'status', 'QUEUED',
           'attempts', '0',
-          'createdAt', createdAt)
+          'createdAt', createdAt,
+          'createdAtMicros', createdAtMicros)
         redis.call('EXPIRE', entryKey, ttlSeconds)
-        redis.call('LPUSH', queueKey, runId)
+        -- ZSET keyed by createdAtMicros: ZPOPMIN drains oldest-first
+        -- (FIFO); listing pagination uses ZREVRANGEBYSCORE with a
+        -- (createdAt, runId) cursor anchor. Score is stable across the
+        -- entry's lifecycle — requeue does not bump it (see Phase 3b /
+        -- Q1 design).
+        redis.call('ZADD', queueKey, createdAtMicros, runId)
         -- Org-level membership: maintained atomically with the per-env
         -- queue so the drainer can walk orgs → envs-for-org and
         -- schedule one env per org per tick. SADDs are idempotent if the
@@ -248,15 +268,20 @@ export class MollifierBuffer {
 
         local envId = redis.call('HGET', entryKey, 'envId')
         local orgId = redis.call('HGET', entryKey, 'orgId')
-        if not envId then
+        local createdAtMicros = redis.call('HGET', entryKey, 'createdAtMicros')
+        if not envId or not createdAtMicros then
           return 0
         end
 
         local currentAttempts = redis.call('HGET', entryKey, 'attempts')
         local nextAttempts = tonumber(currentAttempts or '0') + 1
 
         redis.call('HSET', entryKey, 'status', 'QUEUED', 'attempts', tostring(nextAttempts))
-        redis.call('LPUSH', queuePrefix .. envId, runId)
+        -- Requeue re-adds with the ORIGINAL createdAtMicros score.
+        -- createdAt is immutable across retries (Phase 3b decision).
+        -- The drainer's maxAttempts caps the retry loop so a poisoned
+        -- entry doesn't head-of-line forever.
+        redis.call('ZADD', queuePrefix .. envId, tonumber(createdAtMicros), runId)
         -- Re-track the org/env: pop may have SREM'd them when the queue
         -- last emptied. SADDs are idempotent if the values are still
         -- present.
@@ -296,7 +321,9 @@ export class MollifierBuffer {
         -- hash without a TTL, leaking memory. The loop is bounded by queue
         -- length; entire Lua script remains atomic.
         while true do
-          local runId = redis.call('RPOP', queueKey)
+          -- ZPOPMIN returns {member, score} as a flat array, or {} when empty.
+          local popped = redis.call('ZPOPMIN', queueKey)
+          local runId = popped[1]
           if not runId then
             -- Queue is empty AND we have no entry to read orgId from, so
             -- skip org-level cleanup. Stale org-envs entries are bounded
@@ -313,9 +340,9 @@ export class MollifierBuffer {
               result[raw[i]] = raw[i + 1]
             end
             -- Prune org-level membership if this pop drained the queue.
-            -- Atomic with the RPOP above — a concurrent accept AFTER this
-            -- script will SADD both back along with its LPUSH.
-            if redis.call('LLEN', queueKey) == 0 then
+            -- Atomic with the ZPOPMIN above — a concurrent accept AFTER
+            -- this script will SADD both back along with its ZADD.
+            if redis.call('ZCARD', queueKey) == 0 then
               pruneOrgMembership(result['orgId'])
             end
             return cjson.encode(result)
@@ -379,6 +406,7 @@ declare module "@internal/redis" {
       orgId: string,
       payload: string,
       createdAt: string,
+      createdAtMicros: string,
       ttlSeconds: string,
       orgEnvsPrefix: string,
       callback?: Callback<number>,

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

@@ -44,6 +44,9 @@ 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,
   lastError: stringToError.optional(),
 });