From 709d2f5afb969300b065626be3af1fad2f288618 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Wed, 20 May 2026 14:52:41 +0100 Subject: [PATCH] feat(redis-worker): migrate mollifier queue from LIST to ZSET (Phase B1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .changeset/mollifier-buffer-zset-migration.md | 5 + .../redis-worker/src/mollifier/buffer.test.ts | 152 ++++++++++++++++-- packages/redis-worker/src/mollifier/buffer.ts | 62 +++++-- .../redis-worker/src/mollifier/schemas.ts | 3 + 4 files changed, 193 insertions(+), 29 deletions(-) create mode 100644 .changeset/mollifier-buffer-zset-migration.md diff --git a/.changeset/mollifier-buffer-zset-migration.md b/.changeset/mollifier-buffer-zset-migration.md new file mode 100644 index 000000000..011bc2be2 --- /dev/null +++ b/.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. diff --git a/packages/redis-worker/src/mollifier/buffer.test.ts b/packages/redis-worker/src/mollifier/buffer.test.ts index 71138a915..b4d7e1a53 100644 --- a/packages/redis-worker/src/mollifier/buffer.test.ts +++ b/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,12 +246,12 @@ 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(); @@ -256,7 +259,7 @@ describe("MollifierBuffer.pop orphan handling", () => { 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,7 +480,9 @@ 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"); @@ -481,12 +490,13 @@ describe("MollifierBuffer.requeue ordering", () => { 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", diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index 6c0fbc453..869975dc8 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/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 { 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,7 +268,8 @@ 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 @@ -256,7 +277,11 @@ export class MollifierBuffer { 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, diff --git a/packages/redis-worker/src/mollifier/schemas.ts b/packages/redis-worker/src/mollifier/schemas.ts index f93b0f0a3..23a71a816 100644 --- a/packages/redis-worker/src/mollifier/schemas.ts +++ b/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(), });