feat(redis-worker): add MollifierBuffer.mutateSnapshot (Phase B3)
Atomic Lua-driven snapshot mutation for the burst-buffer entry hash. Returns one of three result codes per Q3: - applied_to_snapshot: entry was QUEUED + not materialised; the drainer will see the patched payload on its next pop. - not_found: no entry hash for this runId. - busy: entry is DRAINING / FAILED / materialised — the caller wait-and-bounces through PG (helper lands in B5). Four patch types: - append_tags: union-merges into payload.tags, dedupes against existing values. - set_metadata: replaces metadata + metadataType (last-write-wins). - set_delay: replaces payload.delayUntil. - mark_cancelled: stamps cancelledAt + cancelReason; the drainer bifurcation in Q4 reads these on next pop. 10 new tests cover: not_found, all four patch types (success + absent-field handling), each busy state (DRAINING, FAILED, materialised), and per-runId atomicity under 50-way concurrent appends.
This commit is contained in:
Dan Sutton
@@ -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).
|
||||
@@ -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",
|
||||
|
||||
@@ -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<MutateSnapshotResult> {
|
||||
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<number>,
|
||||
): Result<number, Context>;
|
||||
mutateMollifierSnapshot(
|
||||
entryKey: string,
|
||||
patchJson: string,
|
||||
callback?: Callback<string>,
|
||||
): Result<string, Context>;
|
||||
ackMollifierEntry(
|
||||
entryKey: string,
|
||||
graceTtlSeconds: string,
|
||||
|
||||
Reference in New Issue
Block a user