github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

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

Browse Source 
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.
This commit is contained in:
Dan Sutton
2026-05-20 14:52:41 +01:00
parent 015787cf62
commit 709d2f5afb
4 changed files with 193 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/mollifier-buffer-zset-migration.md
+5
View File
@@ -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
+140 -12
View File
@@ -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",
packages/redis-worker/src/mollifier/buffer.ts
+45 -17
View File
@@ -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,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<number>,
packages/redis-worker/src/mollifier/schemas.ts
+3
View File
@@ -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(),
});