diff --git a/.changeset/mollifier-buffer-mutate-snapshot.md b/.changeset/mollifier-buffer-mutate-snapshot.md new file mode 100644 index 000000000..6456b7bbe --- /dev/null +++ b/.changeset/mollifier-buffer-mutate-snapshot.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/redis-worker": patch +--- + +Add `MollifierBuffer.mutateSnapshot(runId, patch)` — atomic Lua-driven snapshot mutation for the burst-buffer entry hash. Supports four patch types: `append_tags` (with dedup), `set_metadata`, `set_delay`, `mark_cancelled`. Returns one of three result codes: `applied_to_snapshot` (entry was QUEUED and not materialised), `not_found` (no entry hash), or `busy` (DRAINING / FAILED / materialised — caller wait-and-bounces through PG per Q3 design). diff --git a/packages/redis-worker/src/mollifier/buffer.test.ts b/packages/redis-worker/src/mollifier/buffer.test.ts index 86b6c3240..5bbec9a95 100644 --- a/packages/redis-worker/src/mollifier/buffer.test.ts +++ b/packages/redis-worker/src/mollifier/buffer.test.ts @@ -1078,6 +1078,357 @@ describe("MollifierBuffer envs set lifecycle", () => { ); }); +describe("MollifierBuffer.mutateSnapshot", () => { + redisTest( + "returns not_found when no entry exists for the runId", + { 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 { + const result = await buffer.mutateSnapshot("nope", { + type: "append_tags", + tags: ["x"], + }); + expect(result).toBe("not_found"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "append_tags on QUEUED entry appends and dedupes", + { 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: "r1", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: ["existing"] }), + }); + const first = await buffer.mutateSnapshot("r1", { + type: "append_tags", + tags: ["existing", "new"], + }); + expect(first).toBe("applied_to_snapshot"); + + const entry = await buffer.getEntry("r1"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags).toEqual(["existing", "new"]); + + // Second mutation appends without duplicating + const second = await buffer.mutateSnapshot("r1", { + type: "append_tags", + tags: ["new", "third"], + }); + expect(second).toBe("applied_to_snapshot"); + const e2 = await buffer.getEntry("r1"); + const p2 = JSON.parse(e2!.payload) as { tags: string[] }; + expect(p2.tags).toEqual(["existing", "new", "third"]); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "append_tags creates payload.tags when absent", + { 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: "r2", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ taskId: "t" }), + }); + const result = await buffer.mutateSnapshot("r2", { + type: "append_tags", + tags: ["a", "b"], + }); + expect(result).toBe("applied_to_snapshot"); + const entry = await buffer.getEntry("r2"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags).toEqual(["a", "b"]); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "set_metadata replaces metadata + metadataType (last-write-wins)", + { 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: "r3", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ metadata: '{"v":1}', metadataType: "application/json" }), + }); + const result = await buffer.mutateSnapshot("r3", { + type: "set_metadata", + metadata: '{"v":2}', + metadataType: "application/json", + }); + expect(result).toBe("applied_to_snapshot"); + const entry = await buffer.getEntry("r3"); + const payload = JSON.parse(entry!.payload) as { + metadata: string; + metadataType: string; + }; + expect(payload.metadata).toBe('{"v":2}'); + expect(payload.metadataType).toBe("application/json"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "set_delay sets payload.delayUntil", + { 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: "r4", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ taskId: "t" }), + }); + const result = await buffer.mutateSnapshot("r4", { + type: "set_delay", + delayUntil: "2026-06-01T00:00:00.000Z", + }); + expect(result).toBe("applied_to_snapshot"); + const entry = await buffer.getEntry("r4"); + const payload = JSON.parse(entry!.payload) as { delayUntil: string }; + expect(payload.delayUntil).toBe("2026-06-01T00:00:00.000Z"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "mark_cancelled stamps cancelledAt + cancelReason", + { 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: "r5", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ taskId: "t" }), + }); + const result = await buffer.mutateSnapshot("r5", { + type: "mark_cancelled", + cancelledAt: "2026-05-19T12:00:00.000Z", + cancelReason: "user-initiated", + }); + expect(result).toBe("applied_to_snapshot"); + const entry = await buffer.getEntry("r5"); + const payload = JSON.parse(entry!.payload) as { + cancelledAt: string; + cancelReason: string; + }; + expect(payload.cancelledAt).toBe("2026-05-19T12:00:00.000Z"); + expect(payload.cancelReason).toBe("user-initiated"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "returns busy when entry is DRAINING", + { 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: "rd", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: [] }), + }); + await buffer.pop("env_m"); + const result = await buffer.mutateSnapshot("rd", { + type: "append_tags", + tags: ["x"], + }); + expect(result).toBe("busy"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "returns busy when entry is FAILED", + { 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: "rf", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: [] }), + }); + await buffer.pop("env_m"); + await buffer.fail("rf", { code: "X", message: "boom" }); + const result = await buffer.mutateSnapshot("rf", { + type: "append_tags", + tags: ["x"], + }); + expect(result).toBe("busy"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "returns busy when entry is materialised (post-ack grace window)", + { 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: "rm", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: [] }), + }); + await buffer.pop("env_m"); + await buffer.ack("rm"); + const result = await buffer.mutateSnapshot("rm", { + type: "append_tags", + tags: ["x"], + }); + expect(result).toBe("busy"); + } finally { + await buffer.close(); + } + }, + ); + + redisTest( + "Lua atomicity serialises concurrent mutations per-runId", + { 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: "rcc", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: [] }), + }); + + const tagsToAdd = Array.from({ length: 50 }, (_, i) => `t${i}`); + await Promise.all( + tagsToAdd.map((t) => buffer.mutateSnapshot("rcc", { type: "append_tags", tags: [t] })), + ); + + const entry = await buffer.getEntry("rcc"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags.sort()).toEqual(tagsToAdd.sort()); + } finally { + await buffer.close(); + } + }, + ); +}); + describe("MollifierBuffer ZSET storage", () => { redisTest( "queue key is a ZSET scored by entry's createdAtMicros", diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index 7b8325958..22483bbc4 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/packages/redis-worker/src/mollifier/buffer.ts @@ -19,6 +19,14 @@ export type MollifierBufferOptions = { // have a safety net while PG replica lag settles. Q1 D2. const ACK_GRACE_TTL_SECONDS = 30; +export type SnapshotPatch = + | { type: "append_tags"; tags: string[] } + | { type: "set_metadata"; metadata: string; metadataType: string } + | { type: "set_delay"; delayUntil: string } + | { type: "mark_cancelled"; cancelledAt: string; cancelReason?: string }; + +export type MutateSnapshotResult = "applied_to_snapshot" | "not_found" | "busy"; + export class MollifierBuffer { private readonly redis: Redis; private readonly entryTtlSeconds: number; @@ -163,6 +171,29 @@ export class MollifierBuffer { return entries; } + // Atomic snapshot mutation. Used by customer-mutation API endpoints + // (tags, metadata-put, reschedule, cancel) when the run is still in + // the buffer. Three outcomes: + // - "applied_to_snapshot": entry was QUEUED + not materialised; the + // drainer will read the patched payload on its next pop. + // - "not_found": no entry hash exists for this runId. + // - "busy": entry is DRAINING / FAILED / materialised. The API + // wait-and-bounces through PG (Q3 design). + async mutateSnapshot(runId: string, patch: SnapshotPatch): Promise { + const result = (await this.redis.mutateMollifierSnapshot( + `mollifier:entries:${runId}`, + JSON.stringify(patch), + )) as string; + if ( + result === "applied_to_snapshot" || + result === "not_found" || + result === "busy" + ) { + return result; + } + throw new Error(`MollifierBuffer.mutateSnapshot: unexpected Lua return value: ${result}`); + } + // 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 @@ -365,6 +396,65 @@ export class MollifierBuffer { `, }); + this.redis.defineCommand("mutateMollifierSnapshot", { + numberOfKeys: 1, + lua: ` + local entryKey = KEYS[1] + local patchJson = ARGV[1] + + if redis.call('EXISTS', entryKey) == 0 then + return 'not_found' + end + + local status = redis.call('HGET', entryKey, 'status') + local materialised = redis.call('HGET', entryKey, 'materialised') + if status ~= 'QUEUED' or materialised == 'true' then + return 'busy' + end + + local payloadJson = redis.call('HGET', entryKey, 'payload') + local ok, payload = pcall(cjson.decode, payloadJson) + if not ok then return 'busy' end + + local patch = cjson.decode(patchJson) + + if patch.type == 'append_tags' then + -- cjson decode of an absent or empty-array field gives nil or + -- an empty table; we rebuild as a dense array. Existing tags + -- are preserved; new tags are appended only if not present. + local existing = payload.tags or {} + local seen = {} + local merged = {} + for _, t in ipairs(existing) do + if not seen[t] then + seen[t] = true + table.insert(merged, t) + end + end + for _, t in ipairs(patch.tags or {}) do + if not seen[t] then + seen[t] = true + table.insert(merged, t) + end + end + payload.tags = merged + elseif patch.type == 'set_metadata' then + payload.metadata = patch.metadata + payload.metadataType = patch.metadataType + elseif patch.type == 'set_delay' then + payload.delayUntil = patch.delayUntil + elseif patch.type == 'mark_cancelled' then + payload.cancelledAt = patch.cancelledAt + payload.cancelReason = patch.cancelReason + else + return 'busy' + end + + redis.call('HSET', entryKey, 'payload', cjson.encode(payload)) + return 'applied_to_snapshot' + `, + }); + this.redis.defineCommand("ackMollifierEntry", { numberOfKeys: 1, lua: ` @@ -457,6 +547,11 @@ declare module "@internal/redis" { orgEnvsPrefix: string, callback?: Callback, ): Result; + mutateMollifierSnapshot( + entryKey: string, + patchJson: string, + callback?: Callback, + ): Result; ackMollifierEntry( entryKey: string, graceTtlSeconds: string,