diff --git a/internal-packages/run-store/package.json b/internal-packages/run-store/package.json index 110c3b49058..7263a6de05c 100644 --- a/internal-packages/run-store/package.json +++ b/internal-packages/run-store/package.json @@ -14,6 +14,7 @@ } }, "dependencies": { + "@internal/redis": "workspace:*", "@trigger.dev/core": "workspace:*", "@trigger.dev/database": "workspace:*" }, diff --git a/internal-packages/run-store/src/index.ts b/internal-packages/run-store/src/index.ts index 160f9cdada2..3717dc01527 100644 --- a/internal-packages/run-store/src/index.ts +++ b/internal-packages/run-store/src/index.ts @@ -2,3 +2,4 @@ export * from "./types.js"; export * from "./PostgresRunStore.js"; export * from "./runOpsStore.js"; export * from "./readReplicaClient.js"; +export * from "./redisSnapshotStore.js"; diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts new file mode 100644 index 00000000000..369d79e5338 --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -0,0 +1,1226 @@ +// Unit suite for the raw Redis execution-snapshot store. Redis-only: the store holds no Prisma +// reference, so no Postgres container is needed. +import { expect, describe, vi } from "vitest"; +import { redisTest } from "@internal/testcontainers"; +import { createRedisClient } from "@internal/redis"; +import { Logger } from "@trigger.dev/core/logger"; +import { + snapshotKeys, + deriveOrder, + isValidFor, + RedisSnapshotStore, + type SnapshotEntryInput, +} from "./redisSnapshotStore.js"; + +describe("snapshotKeys", () => { + it("puts every core key under one hash tag", () => { + const k = snapshotKeys("run_abc123"); + expect(k.e).toBe("snap:{run_abc123}:e"); + expect(k.idx).toBe("snap:{run_abc123}:idx"); + expect(k.cur).toBe("snap:{run_abc123}:cur"); + expect(k.seq).toBe("snap:{run_abc123}:seq"); + }); +}); + +describe("deriveOrder", () => { + it("drops entries with no index, sorts by index, and maps to id", () => { + expect( + deriveOrder([ + { id: "w_c", index: 2 }, + { id: "w_a", index: 0 }, + { id: "w_no" }, + { id: "w_b", index: 1 }, + ]) + ).toEqual(["w_a", "w_b", "w_c"]); + }); + + it("preserves a repeated id at each of its positions", () => { + expect( + deriveOrder([ + { id: "w_x", index: 0 }, + { id: "w_x", index: 1 }, + ]) + ).toEqual(["w_x", "w_x"]); + }); + + it("returns an empty list when nothing carries an index", () => { + expect(deriveOrder([{ id: "w_a" }, { id: "w_b" }])).toEqual([]); + }); +}); + +describe("isValidFor", () => { + it("is false when the entry carries an error and true otherwise", () => { + expect(isValidFor({ error: "boom" })).toBe(false); + expect(isValidFor({})).toBe(true); + expect(isValidFor({ error: undefined })).toBe(true); + }); +}); + +function entry(over: Partial = {}): SnapshotEntryInput { + return { + id: "snap_1", + engine: "V2", + executionStatus: "RUN_CREATED", + description: "created", + runId: "run_1", + runStatus: "PENDING", + createdAt: "2026-08-21T00:00:00.000Z", + environmentId: "env_1", + environmentType: "PRODUCTION", + projectId: "proj_1", + organizationId: "org_1", + ...over, + }; +} + +describe("append", () => { + redisTest("assigns a monotonic seq and reads the entry back by id", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 72 * 3600 * 1000 }); + try { + const a = await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + }); + const b = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + }); + expect(a).toMatchObject({ outcome: "written", seq: 1 }); + expect(b).toMatchObject({ outcome: "written", seq: 2 }); + + const read = await store.getById("run_1", "snap_2"); + expect(read?.seq).toBe(2); + expect(read?.isValid).toBe(true); + expect(read?.entry.description).toBe("created"); + } finally { + await store.quit(); + } + }); + + redisTest("preserves the entry JSON byte for byte", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + const e = entry({ id: "snap_1", metadata: { empty: [], nested: { a: 1 } } }); + await store.append({ entry: e, kind: "birth", isTerminal: false }); + const read = await store.getById("run_1", "snap_1"); + expect(read?.raw).toBe(JSON.stringify(e)); + expect(read?.entry).toEqual(e); + } finally { + await store.quit(); + } + }); + + redisTest("advances cur only for a valid entry", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "snap_bad", error: "nope" }), + kind: "transition", + isTerminal: false, + }); + const latest = await store.getLatest("run_1"); + expect(latest?.id).toBe("snap_1"); + + const invalid = await store.getById("run_1", "snap_bad"); + expect(invalid?.isValid).toBe(false); + } finally { + await store.quit(); + } + }); + + redisTest("skips a transition against an absent keyspace", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + const r = await store.append({ + entry: entry({ id: "snap_1", runId: "run_never" }), + kind: "transition", + isTerminal: false, + }); + expect(r).toEqual({ outcome: "skippedNoKeyspace" }); + expect(await store.getLatest("run_never")).toBeNull(); + + const k = snapshotKeys("run_never"); + const raw = createRedisClient(redisOptions); + try { + expect(await raw.exists(k.e, k.idx, k.cur, k.seq)).toBe(0); + } finally { + await raw.quit(); + } + } finally { + await store.quit(); + } + }); + + redisTest("skips a transition when only the seq key has expired", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + + const k = snapshotKeys("run_1"); + const raw = createRedisClient(redisOptions); + try { + await raw.del(k.seq); + } finally { + await raw.quit(); + } + + const r = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + }); + expect(r).toEqual({ outcome: "skippedNoKeyspace" }); + } finally { + await store.quit(); + } + }); + + // Pairs with "skips a transition when only the seq key has expired" above: liveness is checked + // against BOTH anchors, so either one missing alone must skip. + redisTest("skips a transition when only the e key has expired", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + + const k = snapshotKeys("run_1"); + const raw = createRedisClient(redisOptions); + try { + await raw.del(k.e); + } finally { + await raw.quit(); + } + + const r = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + }); + expect(r).toEqual({ outcome: "skippedNoKeyspace" }); + } finally { + await store.quit(); + } + }); + + redisTest( + "carries the original count forward on a carryForward append", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [ + { id: "w_a", index: 0 }, + { id: "w_b", index: 1 }, + ], + }, + }); + await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + const read = await store.getById("run_1", "snap_2"); + expect(read?.cycle).toEqual({ cycleSeq: 1, count: 2 }); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "reports a duplicate id without overwriting the original entry", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + const first = await store.append({ + entry: entry({ id: "snap_1", description: "created" }), + kind: "birth", + isTerminal: false, + }); + expect(first).toMatchObject({ outcome: "written", seq: 1 }); + + const dup = await store.append({ + entry: entry({ id: "snap_1", description: "different" }), + kind: "transition", + isTerminal: false, + }); + expect(dup).toEqual({ outcome: "duplicate", seq: 1 }); + + const read = await store.getById("run_1", "snap_1"); + expect(read?.entry.description).toBe("created"); + + const next = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + }); + expect(next).toMatchObject({ outcome: "written", seq: 2 }); + } finally { + await store.quit(); + } + } + ); +}); + +describe("cycle keys", () => { + redisTest( + "mints an increasing cycleSeq across successive new cycles", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + const a = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + const b = await store.append({ + entry: entry({ id: "snap_3" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + expect(a).toMatchObject({ cycleSeq: 1 }); + expect(b).toMatchObject({ cycleSeq: 2 }); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "a carry-forward reuses the cycle and does not rewrite it", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [ + { id: "w_a", index: 0 }, + { id: "w_a", index: 1 }, + ], + }, + }); + const carried = await store.append({ + entry: entry({ id: "snap_3" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + expect(carried).toMatchObject({ cycleSeq: 1, cycleMismatch: false }); + + // Both entries resolve to the SAME cycle contents, written once. + const first = await store.getSnapshotWaitpointIds("run_1", "snap_2"); + const second = await store.getSnapshotWaitpointIds("run_1", "snap_3"); + expect(first.order).toEqual(["w_a", "w_a"]); + expect(first.distinctIds).toEqual(["w_a"]); + expect(second).toEqual(first); + } finally { + await store.quit(); + } + } + ); + + redisTest("a carry-forward naming a missing cycle still appends", async ({ redisOptions }) => { + const calls: string[] = []; + const metrics = { + recordAppend: () => {}, + recordEntryBytes: () => {}, + recordCycleKeyBytes: () => {}, + recordCycleCount: () => {}, + recordSkippedNoKeyspace: () => {}, + recordCycleMismatch: () => calls.push("mismatch"), + recordLatency: () => {}, + }; + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, metrics }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + const r = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 99 }, + }); + expect(r).toMatchObject({ outcome: "written", cycleMismatch: true }); + // recordCycleMismatch is required by the spec and was previously stubbed but never checked. + expect(calls).toEqual(["mismatch"]); + } finally { + await store.quit(); + } + }); + + redisTest("reports presence and emptiness separately", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false }); + expect(await store.getSnapshotWaitpointIds("run_1", "nope")).toEqual({ + present: false, + distinctIds: [], + order: [], + }); + expect(await store.getSnapshotWaitpointIds("run_1", "snap_1")).toEqual({ + present: true, + distinctIds: [], + order: [], + }); + } finally { + await store.quit(); + } + }); +}); + +describe("read-side cycle mismatch", () => { + redisTest( + "warns and records a metric when a cycle's count disagrees with its order", + async ({ redisOptions }) => { + const calls: string[] = []; + const metrics = { + recordAppend: () => {}, + recordEntryBytes: () => {}, + recordCycleKeyBytes: () => {}, + recordCycleCount: () => {}, + recordSkippedNoKeyspace: () => {}, + recordCycleMismatch: () => calls.push("mismatch"), + recordLatency: () => {}, + }; + const logger = new Logger("test", "debug"); + const warnSpy = vi.spyOn(logger, "warn"); + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, metrics, logger }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + // The pointer's count field (written at append time) survives; only the cycle key's order + // field is wiped, so a read must catch the disagreement instead of reporting count 1. + await raw.hdel("snap:{run_1}:wp:1", "order"); + + const read = await store.getById("run_1", "s2"); + expect(read?.cycle).toEqual({ cycleSeq: 1, count: 1 }); + expect(read?.completedWaitpointIds?.order).toEqual([]); + expect(calls).toEqual(["mismatch"]); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining("cycle"), + expect.objectContaining({ runId: "run_1" }) + ); + } finally { + await raw.quit(); + await store.quit(); + } + } + ); +}); + +describe("TTL rule", () => { + redisTest("a non-terminal append leaves every key unexpiring", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + for (const key of [ + "snap:{run_1}:e", + "snap:{run_1}:idx", + "snap:{run_1}:cur", + "snap:{run_1}:seq", + "snap:{run_1}:wp:1", + ]) { + expect(await raw.pttl(key)).toBe(-1); + } + } finally { + await raw.quit(); + await store.quit(); + } + }); + + redisTest( + "a terminal append expires every key, cycle keys included", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + // Second cycle, so the terminal PEXPIRE loop runs past its first iteration. + await store.append({ + entry: entry({ id: "s1b" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + const r = await store.append({ + entry: entry({ id: "s2", executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + expect(r).toMatchObject({ ttl: "completion" }); + for (const key of [ + "snap:{run_1}:e", + "snap:{run_1}:idx", + "snap:{run_1}:cur", + "snap:{run_1}:seq", + "snap:{run_1}:wp:1", + "snap:{run_1}:wp:2", + ]) { + const ttl = await raw.pttl(key); + expect(ttl).toBeGreaterThan(0); + expect(ttl).toBeLessThanOrEqual(60_000); + } + } finally { + await raw.quit(); + await store.quit(); + } + } + ); + + redisTest("a post-completion append re-applies the completion TTL", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s1b" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s2", executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + + const keys = [ + "snap:{run_1}:e", + "snap:{run_1}:idx", + "snap:{run_1}:cur", + "snap:{run_1}:seq", + "snap:{run_1}:wp:1", + "snap:{run_1}:wp:2", + ]; + // Shrink first: a re-apply is then the only way the TTL can go back up. + for (const key of keys) { + await raw.pexpire(key, 5_000); + } + + // A stale client appends a non-terminal, invalid row after FINISHED. + const late = await store.append({ + entry: entry({ id: "s3", error: "stale" }), + kind: "transition", + isTerminal: false, + }); + expect(late).toMatchObject({ outcome: "written", ttl: "reapplied" }); + for (const key of keys) { + const ttl = await raw.pttl(key); + expect(ttl).toBeGreaterThan(55_000); + expect(ttl).toBeLessThanOrEqual(60_000); + } + } finally { + await raw.quit(); + await store.quit(); + } + }); + + redisTest("a transition after the keyspace expired writes nothing", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s2", executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + // Simulate the completion TTL firing. + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + const after = await store.append({ + entry: entry({ id: "s4" }), + kind: "transition", + isTerminal: false, + }); + expect(after).toEqual({ outcome: "skippedNoKeyspace" }); + expect(await raw.exists("snap:{run_1}:e")).toBe(0); + } finally { + await raw.quit(); + await store.quit(); + } + }); +}); + +describe("getSince", () => { + redisTest("misses on an unknown since id", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + expect(await store.getSince("run_1", "unknown")).toEqual({ kind: "miss" }); + } finally { + await store.quit(); + } + }); + + redisTest("resolves an INVALID since id through its own seq field", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s_bad", error: "x" }), + kind: "transition", + isTerminal: false, + }); + await store.append({ entry: entry({ id: "s3" }), kind: "transition", isTerminal: false }); + + // s_bad is not in the valid-only index, so ZSCORE misses and the '#s' field answers instead. + const r = await store.getSince("run_1", "s_bad"); + expect(r.kind).toBe("hit"); + if (r.kind !== "hit") throw new Error("unreachable"); + expect(r.entries.map((e) => e.id)).toEqual(["s3"]); + } finally { + await store.quit(); + } + }); + + redisTest("returns the NEWEST N ascending, not the oldest", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, sinceLimit: 5 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + for (let i = 1; i <= 12; i++) { + await store.append({ + entry: entry({ id: `s${i}` }), + kind: "transition", + isTerminal: false, + }); + } + const r = await store.getSince("run_1", "s0"); + if (r.kind !== "hit") throw new Error("expected a hit"); + // The engine reads createdAt desc / take N / reverse, so the window is the newest N ascending. + expect(r.entries.map((e) => e.id)).toEqual(["s8", "s9", "s10", "s11", "s12"]); + } finally { + await store.quit(); + } + }); + + redisTest("excludes invalid entries from the window", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s_bad", error: "x" }), + kind: "transition", + isTerminal: false, + }); + await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false }); + const r = await store.getSince("run_1", "s0"); + if (r.kind !== "hit") throw new Error("expected a hit"); + expect(r.entries.map((e) => e.id)).toEqual(["s2"]); + } finally { + await store.quit(); + } + }); + + redisTest("resolves waitpoint ids for the HEAD only", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s1" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_old", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_new", index: 0 }] }, + }); + const r = await store.getSince("run_1", "s0"); + if (r.kind !== "hit") throw new Error("expected a hit"); + // The head is the NEWEST entry, and only it carries resolved ids. + expect(r.headWaitpointIds.order).toEqual(["w_new"]); + expect(r.entries.at(-1)?.id).toBe("s2"); + expect(r.entries[0]?.completedWaitpointIds).toBeUndefined(); + } finally { + await store.quit(); + } + }); + + redisTest("misses for a foreign environment", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false }); + expect(await store.getSince("run_1", "s0", { environmentId: "env_other" })).toEqual({ + kind: "miss", + }); + } finally { + await store.quit(); + } + }); + + redisTest("misses for a foreign environment even at the newest id", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false }); + // The window here is empty (s1 is the newest), so this is the case the old reply.length > 1 + // guard could never catch: an empty window must not silently coerce a foreign miss into a hit. + expect(await store.getSince("run_1", "s1", { environmentId: "env_other" })).toEqual({ + kind: "miss", + }); + } finally { + await store.quit(); + } + }); + + redisTest( + "hits with zero entries when nothing follows the since id", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + // Resolves, nothing after it: "nothing new", NOT "not found". + expect(await store.getSince("run_1", "s0")).toEqual({ + kind: "hit", + entries: [], + headWaitpointIds: { present: false, distinctIds: [], order: [] }, + }); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "skips an entry whose body was evicted rather than throwing", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false }); + await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false }); + + // The mirror of the case the append script documents: idx survives while the entry body in + // `e` is gone. The seq field is left in place so the id still resolves. + await raw.hdel("snap:{run_1}:e", "s1"); + + const r = await store.getSince("run_1", "s0"); + expect(r.kind).toBe("hit"); + if (r.kind !== "hit") throw new Error("unreachable"); + expect(r.entries.map((e) => e.id)).toEqual(["s2"]); + } finally { + await raw.quit(); + await store.quit(); + } + } + ); + + redisTest( + "does not donate the evicted head's waitpoints to the surviving head", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s1" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_old", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_new", index: 0 }] }, + }); + + // s2 is the newest and its body is gone. s1 must come back with ITS OWN waitpoints, + // never s2's -- a dropped row must not donate its cycle data to the next one. + await raw.hdel("snap:{run_1}:e", "s2"); + + const r = await store.getSince("run_1", "s0"); + expect(r.kind).toBe("hit"); + if (r.kind !== "hit") throw new Error("unreachable"); + expect(r.entries.map((e) => e.id)).toEqual(["s1"]); + expect(r.headWaitpointIds.order).toEqual(["w_old"]); + } finally { + await raw.quit(); + await store.quit(); + } + } + ); + + redisTest( + "does not donate a foreign-environment head's waitpoints to the query's window", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ + entry: entry({ id: "s1", environmentId: "env_a" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + // Same run, a different environment -- unreachable in production, but exercises the branch + // where the Lua-chosen head is dropped by the TS-side environment filter. + await store.append({ + entry: entry({ id: "s2", environmentId: "env_b" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + + const r = await store.getSince("run_1", "s1", { environmentId: "env_a" }); + expect(r.kind).toBe("hit"); + if (r.kind !== "hit") throw new Error("unreachable"); + expect(r.entries).toEqual([]); + expect(r.headWaitpointIds.order).toEqual([]); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "hits with zero entries when scoped to the since entry's own environment", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + // Matching environment, nothing after it: pins that an empty window resolves via sinceRaw, + // not by falling through to the "sinceRaw missing" miss path. + expect(await store.getSince("run_1", "s0", { environmentId: "env_1" })).toEqual({ + kind: "hit", + entries: [], + headWaitpointIds: { present: false, distinctIds: [], order: [] }, + }); + } finally { + await store.quit(); + } + } + ); +}); + +describe("environment scoping", () => { + redisTest("getLatest and getById return null for a foreign env", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + + expect(await store.getLatest("run_1", { environmentId: "env_1" })).not.toBeNull(); + expect(await store.getLatest("run_1", { environmentId: "env_other" })).toBeNull(); + expect(await store.getById("run_1", "s1", { environmentId: "env_1" })).not.toBeNull(); + expect(await store.getById("run_1", "s1", { environmentId: "env_other" })).toBeNull(); + } finally { + await store.quit(); + } + }); + + redisTest("getLatest returns null for a run with no keys", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + expect(await store.getLatest("run_absent")).toBeNull(); + expect(await store.getById("run_absent", "nope")).toBeNull(); + } finally { + await store.quit(); + } + }); + + redisTest( + "getSince returns entries when scoped to a matching, non-empty environment", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false }); + await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false }); + // Every existing matching-env getSince test used an EMPTY window, so the per-row compare + // in #decode never ran in the passing direction. This is the first to exercise it with rows. + const r = await store.getSince("run_1", "s0", { environmentId: "env_1" }); + if (r.kind !== "hit") throw new Error("expected a hit"); + expect(r.entries.map((e) => e.id)).toEqual(["s1", "s2"]); + } finally { + await store.quit(); + } + } + ); +}); + +describe("expectedCur compare-and-set", () => { + redisTest("absent by default: cur advances unconditionally", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + const r = await store.append({ + entry: entry({ id: "s2", previousSnapshotId: "stale" }), + kind: "transition", + isTerminal: false, + }); + expect(r).toMatchObject({ outcome: "written" }); + expect((await store.getLatest("run_1"))?.id).toBe("s2"); + } finally { + await store.quit(); + } + }); + + redisTest("supplied and matching: the append proceeds", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + const r = await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + expectedCur: "s1", + }); + expect(r).toMatchObject({ outcome: "written", seq: 2 }); + } finally { + await store.quit(); + } + }); + + redisTest("supplied and stale: writes NOTHING and reports the fork", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false }); + + // A second concurrent transition that read cur = s1 before s2 landed. + const r = await store.append({ + entry: entry({ id: "s3" }), + kind: "transition", + isTerminal: false, + expectedCur: "s1", + }); + expect(r).toEqual({ outcome: "forked", actualCur: "s2" }); + + // Nothing was written: no entry, cur is still s2 (not overwritten by s3, and not cleared), + // and the seq counter did not move. + expect(await store.getById("run_1", "s3")).toBeNull(); + expect((await store.getLatest("run_1"))?.id).toBe("s2"); + const next = await store.append({ + entry: entry({ id: "s4" }), + kind: "transition", + isTerminal: false, + }); + expect(next).toMatchObject({ seq: 3 }); + } finally { + await store.quit(); + } + }); + + redisTest( + "supplied as empty string: still enforces a check against an unset cur", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + // The birth sets cur to "s1", so a caller claiming cur is UNSET (expectedCur: "") must + // fork rather than have "" silently treated as "no compare-and-set requested". + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + const r = await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + expectedCur: "", + }); + expect(r).toEqual({ outcome: "forked", actualCur: "s1" }); + expect(await store.getById("run_1", "s2")).toBeNull(); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "a duplicate id wins over a stale CAS: retrying your own successful write is not a fork", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + expectedCur: "s1", + }); + + // Retry of the same append: cur has since moved to s2, so a naive CAS-first check would + // see actual=s2 != expected=s1 and report a fork -- but s2 is THIS write, not a rival's. + const retry = await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + expectedCur: "s1", + }); + expect(retry).toEqual({ outcome: "duplicate", seq: 2 }); + } finally { + await store.quit(); + } + } + ); + + redisTest( + "supplied as empty string against a genuinely unset cur: the append proceeds", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + try { + // The load-bearing succeeding direction: expectedCur: "" asserts "cur is unset", and on a + // fresh keyspace that assertion is TRUE, so the append must proceed, not fork. + const r = await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + expectedCur: "", + }); + expect(r).toMatchObject({ outcome: "written", seq: 1 }); + } finally { + await store.quit(); + } + } + ); +}); + +// CRC16/XMODEM over a key's hash tag, per Redis's cluster hashing rule. CLUSTER KEYSLOT is +// unavailable on this standalone container ("cluster support disabled"), so the slot is computed +// here instead. Verified against the `cluster-key-slot` package's output for our key shapes. +function crc16(str: string): number { + let crc = 0; + for (let i = 0; i < str.length; i++) { + crc ^= str.charCodeAt(i) << 8; + for (let j = 0; j < 8; j++) { + crc = crc & 0x8000 ? (crc << 1) ^ 0x1021 : crc << 1; + crc &= 0xffff; + } + } + return crc; +} + +function hashSlot(key: string): number { + const start = key.indexOf("{"); + const end = start === -1 ? -1 : key.indexOf("}", start + 1); + const tag = start !== -1 && end !== -1 && end > start + 1 ? key.slice(start + 1, end) : key; + return crc16(tag) % 16384; +} + +describe("hash tag and keyPrefix", () => { + it("every key for one run lands in one cluster slot", () => { + // Keys come from snapshotKeys() plus the wp: suffix the Lua prelude derives the same way, + // with a keyPrefix prepended by hand as ioredis would. A dropped hash tag would split the slots. + // Pin the helper itself before trusting it: the published XMODEM check value, and two known + // slots (one matching cluster-key-slot, one a different run's tag as a negative control -- + // otherwise a constant-valued crc16 would satisfy slots.size === 1 for the wrong reason). + expect(crc16("123456789")).toBe(0x31c3); + expect(hashSlot("engine:snap:{run_1}:e")).toBe(8108); + expect(hashSlot("engine:snap:{run_2}:e")).toBe(12239); + + const k = snapshotKeys("run_1"); + const base = k.e.slice(0, -2); + const keys = [k.e, k.idx, k.cur, k.seq, `${base}:wp:1`, `${base}:wp:2`].map( + (key) => `engine:${key}` + ); + const slots = new Set(keys.map(hashSlot)); + expect(slots.size).toBe(1); + }); + + redisTest("the terminal append expires the PREFIXED cycle keys", async ({ redisOptions }) => { + // This is the guard for the trap: ioredis prefixes only the KEYS array, so a cycle key minted + // inside Lua would be UNPREFIXED while the client wrote a prefixed one. Deriving it from KEYS[1] + // inherits both the prefix and the hash tag. If someone later mints it in Lua, this fails. + const prefixed = { ...redisOptions, keyPrefix: "engine:" }; + const store = new RedisSnapshotStore({ redisOptions: prefixed, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + expect(await raw.exists("engine:snap:{run_1}:wp:1")).toBe(1); + expect(await raw.exists("snap:{run_1}:wp:1")).toBe(0); + + await store.append({ + entry: entry({ id: "s2", executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + const ttl = await raw.pttl("engine:snap:{run_1}:wp:1"); + expect(ttl).toBeGreaterThan(50_000); + expect(ttl).toBeLessThanOrEqual(60_000); + } finally { + raw.disconnect(); + await store.quit(); + } + }); + + redisTest("reads work through a keyPrefix", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ + redisOptions: { ...redisOptions, keyPrefix: "engine:" }, + completedTtlMs: 60_000, + }); + try { + await store.append({ + entry: entry({ id: "s1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + expect((await store.getLatest("run_1"))?.id).toBe("s1"); + expect((await store.getSnapshotWaitpointIds("run_1", "s1")).order).toEqual(["w_a"]); + } finally { + await store.quit(); + } + }); +}); + +describe("observability", () => { + redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => { + const calls: unknown[][] = []; + const metrics = { + recordAppend: (o: string, t: string) => calls.push(["append", o, t]), + recordEntryBytes: (b: number) => calls.push(["entryBytes", b]), + recordCycleKeyBytes: (b: number) => calls.push(["cycleBytes", b]), + recordCycleCount: (c: number) => calls.push(["cycleCount", c]), + recordSkippedNoKeyspace: () => calls.push(["skipped"]), + recordCycleMismatch: () => calls.push(["mismatch"]), + recordLatency: (op: string) => calls.push(["latency", op]), + }; + const store = new RedisSnapshotStore({ + redisOptions, + completedTtlMs: 60_000, + metrics, + highWater: { entryBytes: 1 }, + }); + try { + // A huge inline value is observed, never rejected or truncated: Postgres had no cap either. + const big = "x".repeat(20_000); + const bigEntry = entry({ id: "s1", description: big }); + const rawBytes = Buffer.byteLength(JSON.stringify(bigEntry), "utf8"); + const orderBytes = Buffer.byteLength(JSON.stringify(["w_a"]), "utf8"); + + const r = await store.append({ + entry: bigEntry, + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + expect(r).toMatchObject({ outcome: "written" }); + expect((await store.getById("run_1", "s1"))?.entry.description).toBe(big); + + // Exact values, not just `b > 0`: a swapped recordEntryBytes/recordCycleKeyBytes wiring + // would still pass a `b > 0` check but fails this, since the two sizes are wildly different. + expect(calls).toEqual([ + ["entryBytes", rawBytes], + ["cycleBytes", orderBytes], + ["cycleCount", 1], + ["append", "written", "none"], + ["latency", "append"], + ["latency", "getById"], + ]); + calls.length = 0; + + await store.append({ + entry: entry({ id: "s2", runId: "run_absent" }), + kind: "transition", + isTerminal: false, + }); + + // Partitioned from the first append's calls: proves recordSkippedNoKeyspace fires ONLY on + // this skip, not (also, harmlessly) on the earlier successful append. + expect(calls).toEqual([ + ["skipped"], + ["append", "skippedNoKeyspace", "none"], + ["latency", "append"], + ]); + } finally { + await store.quit(); + } + }); + + redisTest( + "names the run in a high-water warning, and stays silent under a high threshold", + async ({ redisOptions }) => { + const loudLogger = new Logger("test", "debug"); + const loudWarn = vi.spyOn(loudLogger, "warn"); + const loud = new RedisSnapshotStore({ + redisOptions, + completedTtlMs: 1000, + logger: loudLogger, + highWater: { entryBytes: 1, cycleKeyBytes: 1, cycleCount: 0 }, + }); + + const quietLogger = new Logger("test", "debug"); + const quietWarn = vi.spyOn(quietLogger, "warn"); + const quiet = new RedisSnapshotStore({ + redisOptions, + completedTtlMs: 1000, + logger: quietLogger, + highWater: { entryBytes: 1_000_000, cycleKeyBytes: 1_000_000, cycleCount: 1_000_000 }, + }); + + try { + await loud.append({ + entry: entry({ id: "s1", runId: "run_loud" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + expect(loudWarn).toHaveBeenCalledTimes(3); + for (const [, payload] of loudWarn.mock.calls) { + expect(payload).toMatchObject({ runId: "run_loud" }); + } + + // Same shape of append, high thresholds: proves the mark is respected, not just logged. + await quiet.append({ + entry: entry({ id: "s1", runId: "run_quiet" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] }, + }); + expect(quietWarn).not.toHaveBeenCalled(); + } finally { + await loud.quit(); + await quiet.quit(); + } + } + ); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts new file mode 100644 index 00000000000..7b60843e2b4 --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -0,0 +1,667 @@ +import { + createRedisClient, + type Callback, + type Redis, + type RedisOptions, + type Result, +} from "@internal/redis"; +import { Logger } from "@trigger.dev/core/logger"; + +export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string }; + +// All four core keys plus every snap:{runId}:wp: key share the {runId} hash tag, so a run's whole +// state sits in one cluster slot and every mutation is one atomic script. +export function snapshotKeys(runId: string): SnapshotKeys { + const base = `snap:{${runId}}`; + return { e: `${base}:e`, idx: `${base}:idx`, cur: `${base}:cur`, seq: `${base}:seq` }; +} + +export type CompletedWaitpointRef = { id: string; index?: number }; + +// Reproduces PostgresRunStore.#createExecutionSnapshot's completedWaitpointOrder derivation exactly: +// drop anything without an index, sort ascending by index, map to id. Repeats are preserved, because +// the same run can sit in one batch more than once under a single idempotency key. +export function deriveOrder(completedWaitpoints: CompletedWaitpointRef[]): string[] { + return completedWaitpoints + .filter((w) => w.index !== undefined) + .sort((a, b) => a.index! - b.index!) + .map((w) => w.id); +} + +// isValid is derived, never stored, so the entry JSON stays byte-identical to the caller's document. +export function isValidFor(entry: { error?: unknown }): boolean { + return !entry.error; +} + +export type SnapshotEntryInput = { + id: string; + engine: "V2"; + executionStatus: string; + description: string; + runId: string; + runStatus: string; + createdAt: string; + attemptNumber?: number | null; + previousSnapshotId?: string; + batchId?: string; + environmentId: string; + environmentType: string; + projectId: string; + organizationId: string; + checkpointId?: string; + workerId?: string; + runnerId?: string; + metadata?: unknown; + error?: string; +}; + +export type WaitpointIds = { present: boolean; distinctIds: string[]; order: string[] }; + +export type GetSinceResult = + | { kind: "miss" } + | { kind: "hit"; entries: SnapshotRead[]; headWaitpointIds: WaitpointIds }; + +export type SnapshotRead = { + id: string; + seq: number; + isValid: boolean; + entry: Record; + raw: string; + cycle?: { cycleSeq: number; count: number }; + completedWaitpointIds?: WaitpointIds; +}; + +export type AppendResult = + | { + outcome: "written"; + seq: number; + cycleSeq?: number; + ttl: "none" | "completion" | "reapplied"; + cycleMismatch: boolean; + } + | { outcome: "skippedNoKeyspace" } + | { outcome: "forked"; actualCur: string } + | { outcome: "duplicate"; seq: number }; + +export type SnapshotStoreMetrics = { + recordAppend(outcome: string, ttl: string): void; + recordEntryBytes(bytes: number): void; + recordCycleKeyBytes(bytes: number): void; + recordCycleCount(count: number): void; + recordSkippedNoKeyspace(): void; + recordCycleMismatch(): void; + recordLatency(op: string, ms: number): void; +}; + +export type RedisSnapshotStoreOptions = { + redisOptions: RedisOptions; + completedTtlMs: number; + sinceLimit?: number; + highWater?: { entryBytes?: number; cycleKeyBytes?: number; cycleCount?: number }; + metrics?: SnapshotStoreMetrics; + logger?: Logger; +}; + +const SKIPPED = "skipped"; +const FORKED = "forked"; +const WRITTEN = "written"; +const DUPLICATE = "duplicate"; + +export class RedisSnapshotStore { + private readonly redis: Redis; + private readonly logger: Logger; + private readonly completedTtlMs: number; + private readonly sinceLimit: number; + private readonly metrics?: SnapshotStoreMetrics; + private readonly highWater: NonNullable; + #quit?: Promise; + + constructor(options: RedisSnapshotStoreOptions) { + this.logger = options.logger ?? new Logger("RedisSnapshotStore", "debug"); + this.completedTtlMs = options.completedTtlMs; + this.sinceLimit = options.sinceLimit ?? 50; + this.metrics = options.metrics; + this.highWater = options.highWater ?? {}; + this.redis = createRedisClient(options.redisOptions, { + onError: (error) => this.logger.error("RedisSnapshotStore redis client error", { error }), + }); + this.#registerCommands(); + } + + async quit(): Promise { + // Idempotent and error-swallowing: every test calls this in a `finally`, and a double quit() + // (or one after a failed connect) must never mask the real assertion failure. + if (!this.#quit) { + this.#quit = this.redis.quit().then( + () => undefined, + () => undefined + ); + } + await this.#quit; + } + + async #timed(op: string, fn: () => Promise): Promise { + const started = Date.now(); + try { + return await fn(); + } finally { + this.metrics?.recordLatency(op, Date.now() - started); + } + } + + async append(args: { + entry: SnapshotEntryInput; + kind: "birth" | "transition"; + isTerminal: boolean; + expectedCur?: string; + cycle?: + | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string } + | { kind: "carryForward"; cycleSeq: number }; + }): Promise { + return this.#timed("append", async () => { + const k = snapshotKeys(args.entry.runId); + const raw = JSON.stringify(args.entry); + const valid = isValidFor(args.entry); + + let cycleMode = "none"; + let cycleSeqIn = "0"; + let orderJson = ""; + let records = ""; + let orderCount = "0"; + if (args.cycle?.kind === "new") { + const order = deriveOrder(args.cycle.completedWaitpoints); + cycleMode = "new"; + orderJson = JSON.stringify(order); + records = args.cycle.records ?? ""; + orderCount = String(order.length); + } else if (args.cycle?.kind === "carryForward") { + cycleMode = "carry"; + cycleSeqIn = String(args.cycle.cycleSeq); + } + + const reply = (await this.redis.appendSnapshotEntry( + k.e, + k.idx, + k.cur, + k.seq, + args.kind, + args.entry.id, + raw, + valid ? "1" : "0", + args.isTerminal ? "1" : "0", + String(this.completedTtlMs), + cycleMode, + cycleSeqIn, + orderJson, + records, + orderCount, + args.expectedCur ?? "", + args.expectedCur !== undefined ? "1" : "0" + )) as string[]; + + return this.#interpretAppend(reply, raw, orderJson, args.entry.runId); + }); + } + + #interpretAppend(reply: string[], raw: string, orderJson: string, runId: string): AppendResult { + if (reply[0] === SKIPPED) { + this.metrics?.recordSkippedNoKeyspace(); + this.metrics?.recordAppend("skippedNoKeyspace", "none"); + return { outcome: "skippedNoKeyspace" }; + } + if (reply[0] === FORKED) { + this.metrics?.recordAppend("forked", "none"); + return { outcome: "forked", actualCur: reply[1] ?? "" }; + } + if (reply[0] === DUPLICATE) { + this.metrics?.recordAppend("duplicate", "none"); + return { outcome: "duplicate", seq: Number(reply[1]) }; + } + const seq = Number(reply[1]); + const cycleSeq = Number(reply[2]); + const ttl = reply[3] as "none" | "completion" | "reapplied"; + const cycleMismatch = reply[4] === "1"; + if (cycleMismatch) { + this.metrics?.recordCycleMismatch(); + } + this.#observeSizes(raw, orderJson, cycleSeq, runId); + this.metrics?.recordAppend("written", ttl); + return { + outcome: "written", + seq, + ...(cycleSeq > 0 ? { cycleSeq } : {}), + ttl, + cycleMismatch, + }; + } + + #observeSizes(raw: string, orderJson: string, cycleSeq: number, runId: string): void { + const entryBytes = Buffer.byteLength(raw, "utf8"); + this.metrics?.recordEntryBytes(entryBytes); + if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) { + this.logger.warn("RedisSnapshotStore entry above high-water mark", { runId, entryBytes }); + } + if (orderJson !== "") { + const cycleBytes = Buffer.byteLength(orderJson, "utf8"); + this.metrics?.recordCycleKeyBytes(cycleBytes); + if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) { + this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { + runId, + cycleBytes, + }); + } + } + if (cycleSeq > 0) { + this.metrics?.recordCycleCount(cycleSeq); + if (this.highWater.cycleCount !== undefined && cycleSeq > this.highWater.cycleCount) { + this.logger.warn("RedisSnapshotStore cycle count above high-water mark", { + runId, + cycleSeq, + }); + } + } + } + + async getById( + runId: string, + snapshotId: string, + opts?: { environmentId?: string } + ): Promise { + return this.#timed("getById", async () => { + const k = snapshotKeys(runId); + const reply = await this.redis.readSnapshotById(k.e, k.idx, k.cur, k.seq, snapshotId); + return this.#decode(reply, opts?.environmentId, runId, true); + }); + } + + async getLatest(runId: string, opts?: { environmentId?: string }): Promise { + return this.#timed("getLatest", async () => { + const k = snapshotKeys(runId); + const reply = await this.redis.readLatestSnapshot(k.e, k.idx, k.cur, k.seq); + return this.#decode(reply, opts?.environmentId, runId, true); + }); + } + + // Returns all three shapes the Postgres surface needs from one read: `distinctIds` matches the + // deduped join that findSnapshotCompletedWaitpointIds returns, `present` serves the WithPresence + // variant (which distinguishes "no waitpoints" from "snapshot not visible"), and `order` keeps the + // repeats that the engine expands into one CompletedWaitpoint per position. + async getSnapshotWaitpointIds(runId: string, snapshotId: string): Promise { + return this.#timed("getSnapshotWaitpointIds", async () => { + const k = snapshotKeys(runId); + const reply = await this.redis.readSnapshotWaitpointIds(k.e, k.idx, k.cur, k.seq, snapshotId); + return decodeWaitpointIds(reply[0] === "1", reply[1] ?? ""); + }); + } + + // A miss is not an error. It is the coexistence path: a pre-cutover snapshot id, expired history, + // or an org not yet enabled. The caller falls back to Postgres. + async getSince( + runId: string, + sinceId: string, + opts?: { environmentId?: string; limit?: number } + ): Promise { + return this.#timed("getSince", async () => { + const k = snapshotKeys(runId); + const limit = opts?.limit ?? this.sinceLimit; + const reply = await this.redis.readSnapshotsSince( + k.e, + k.idx, + k.cur, + k.seq, + sinceId, + String(limit) + ); + if (reply === null) return { kind: "miss" }; + + const sinceRaw = reply[0] ?? ""; + if (opts?.environmentId !== undefined) { + // Scoped by the since entry itself, same as Postgres's step-1 lookup: a foreign since id + // is NOT FOUND regardless of what follows it, never an empty "nothing new" hit. + if (sinceRaw === "") return { kind: "miss" }; + const since = JSON.parse(sinceRaw) as { environmentId?: string }; + if (since.environmentId !== opts.environmentId) return { kind: "miss" }; + } + + const headOrder = reply[1] ?? ""; + const rows: SnapshotRead[] = []; + // Tracks whether the Lua-chosen head row (always the first, i === 2) itself survives the + // env filter below -- headOrder must never be attributed to a different, surviving row. + let headSurvived = false; + for (let i = 2; i + 3 < reply.length; i += 4) { + // orderKnown is false here: headOrder covers only the head row, resolved separately below. + const decoded = this.#decode( + [reply[i], reply[i + 1], reply[i + 2], reply[i + 3], ""], + opts?.environmentId, + runId, + false + ); + if (decoded) { + rows.push(decoded); + if (i === 2) headSurvived = true; + } + } + + rows.reverse(); + const head = headSurvived ? rows[rows.length - 1] : undefined; + const headWaitpointIds = decodeWaitpointIds(head !== undefined, head ? headOrder : ""); + if (head) { + head.completedWaitpointIds = headWaitpointIds; + if (head.cycle) { + this.#checkCycleMismatch(runId, head.cycle.count, headWaitpointIds.order.length); + } + } + return { kind: "hit", entries: rows, headWaitpointIds }; + }); + } + + #checkCycleMismatch(runId: string, count: number, orderLength: number): void { + if (orderLength === count) return; + this.metrics?.recordCycleMismatch(); + this.logger.warn("RedisSnapshotStore cycle count disagrees with its order", { + runId, + count, + orderLength, + }); + } + + // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the + // plan: the store returns null for a foreign environment and the 404 throw stays in the engine. + // orderKnown distinguishes "order field is genuinely empty" from "order was not read for this + // row" (getSince's tail rows use the same empty string for the latter) -- the mismatch check and + // completedWaitpointIds must both be skipped when the order was never read. + #decode( + reply: string[] | null, + environmentId: string | undefined, + runId: string, + orderKnown: boolean + ): SnapshotRead | null { + if (!reply || reply.length === 0) return null; + const [id, raw, seqStr, pointer, orderJson] = reply; + const entry = JSON.parse(raw) as Record; + if (environmentId !== undefined && entry.environmentId !== environmentId) return null; + const read: SnapshotRead = { + id, + seq: Number(seqStr), + isValid: isValidFor(entry as { error?: unknown }), + entry, + raw, + }; + if (pointer) { + const [cs, count] = pointer.split(":"); + read.cycle = { cycleSeq: Number(cs), count: Number(count) }; + if (orderKnown) { + const ids = decodeWaitpointIds(true, orderJson); + read.completedWaitpointIds = ids; + this.#checkCycleMismatch(runId, Number(count), ids.order.length); + } + } + return read; + } + + #registerCommands() { + // Every script declares exactly these four keys and derives snap:{runId}:wp: from KEYS[1] by + // string surgery. ioredis prefixes only the KEYS array, so a key minted inside Lua would be + // UNPREFIXED while the client wrote a prefixed one. + const PRELUDE = ` + local eKey, idxKey, curKey, seqKey = KEYS[1], KEYS[2], KEYS[3], KEYS[4] + local base = string.sub(eKey, 1, #eKey - 2) + local function wpKey(n) return base .. ':wp:' .. n end + local function orderFor(pointer) + if not pointer then return '' end + local cs = string.match(pointer, '^(%d+):') + if not cs then return '' end + return redis.call('HGET', wpKey(cs), 'order') or '' + end + `; + + this.redis.defineCommand("appendSnapshotEntry", { + numberOfKeys: 4, + lua: ` + ${PRELUDE} + local kind = ARGV[1] + local id = ARGV[2] + local raw = ARGV[3] + local isValid = ARGV[4] == '1' + local isTerminal = ARGV[5] == '1' + local ttlMs = tonumber(ARGV[6]) + local cycleMode = ARGV[7] + local cycleSeqIn = tonumber(ARGV[8]) + local orderJson = ARGV[9] + local records = ARGV[10] + local orderCount = ARGV[11] + local expectedCur = ARGV[12] + local casEnabled = ARGV[13] == '1' + + -- Liveness is TWO anchors: e and seq. All keys get the same PEXPIRE but expire independently + -- (or seq can vanish under maxmemory eviction while e survives), so checking e alone lets a + -- late transition recreate seq with no TTL and restart it at 1 beside a surviving idx. A + -- birth always creates both in this same script, so this never rejects a live keyspace. + if kind == 'transition' and (redis.call('EXISTS', eKey) == 0 or redis.call('EXISTS', seqKey) == 0) then + return { '${SKIPPED}' } + end + + -- Append-only: a retried append must not overwrite an existing entry. Checked BEFORE the + -- CAS below -- a present id can only be this same retry, never a competitor's write. + local prior = redis.call('HGET', eKey, id .. '#s') + if prior then + return { '${DUPLICATE}', prior } + end + + -- Optional compare-and-set on cur, checked BEFORE any mutation. Gated on an explicit flag + -- (not on expectedCur ~= ''), so a caller asserting cur is unset (expectedCur = '') still + -- gets a real check instead of silently skipping it. + if casEnabled then + local actual = redis.call('GET', curKey) + if (actual or '') ~= expectedCur then + return { '${FORKED}', actual or '' } + end + end + + local seq = redis.call('HINCRBY', seqKey, 'e', 1) + + local cycleSeq = 0 + local mismatch = 0 + if cycleMode == 'new' then + -- The STORE mints cycleSeq, so the sequence is dense by construction and the terminal + -- PEXPIRE loop from 1..c is correct. + cycleSeq = redis.call('HINCRBY', seqKey, 'c', 1) + redis.call('HSET', wpKey(cycleSeq), 'order', orderJson, 'count', orderCount) + if records ~= '' then + redis.call('HSET', wpKey(cycleSeq), 'records', records) + end + elseif cycleMode == 'carry' then + cycleSeq = cycleSeqIn + local c = redis.call('HGET', wpKey(cycleSeq), 'count') + if not c then + mismatch = 1 + else + orderCount = c + end + end + + redis.call('HSET', eKey, id, raw, id .. '#s', seq) + if cycleSeq > 0 then + redis.call('HSET', eKey, id .. '#c', cycleSeq .. ':' .. orderCount) + end + + -- idx indexes VALID entries only, which makes the since-cap exact. An invalid entry is still + -- reachable by id, and its seq is still readable from its own '#s' field. ZADD before SET cur + -- because Redis never rolls back a partially applied script: if a later call in this script + -- errored, having idx already written is the recoverable half of the pair. + if isValid then + redis.call('ZADD', idxKey, seq, id) + redis.call('SET', curKey, id) + end + + local wasTerminal = redis.call('HGET', seqKey, 't') == '1' + local ttl = 'none' + if isTerminal then + redis.call('HSET', seqKey, 't', '1') + end + if isTerminal or wasTerminal then + redis.call('PEXPIRE', eKey, ttlMs) + redis.call('PEXPIRE', idxKey, ttlMs) + redis.call('PEXPIRE', curKey, ttlMs) + redis.call('PEXPIRE', seqKey, ttlMs) + local high = tonumber(redis.call('HGET', seqKey, 'c') or '0') + for i = 1, high do + redis.call('PEXPIRE', wpKey(i), ttlMs) + end + if isTerminal and not wasTerminal then + ttl = 'completion' + else + ttl = 'reapplied' + end + end + + return { '${WRITTEN}', tostring(seq), tostring(cycleSeq), ttl, tostring(mismatch) } + `, + }); + + this.redis.defineCommand("readSnapshotById", { + numberOfKeys: 4, + lua: ` + ${PRELUDE} + local id = ARGV[1] + local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c') + if not vals[1] then return nil end + -- Coerce every element: a Lua false TRUNCATES the returned array at that position. + return { id, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) } + `, + }); + + this.redis.defineCommand("readLatestSnapshot", { + numberOfKeys: 4, + lua: ` + ${PRELUDE} + local cur = redis.call('GET', curKey) + if not cur then return nil end + local vals = redis.call('HMGET', eKey, cur, cur .. '#s', cur .. '#c') + if not vals[1] then return nil end + return { cur, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) } + `, + }); + + this.redis.defineCommand("readSnapshotWaitpointIds", { + numberOfKeys: 4, + lua: ` + ${PRELUDE} + local id = ARGV[1] + if redis.call('HEXISTS', eKey, id) == 0 then + return { '0', '' } + end + local pointer = redis.call('HGET', eKey, id .. '#c') + return { '1', orderFor(pointer) } + `, + }); + + this.redis.defineCommand("readSnapshotsSince", { + numberOfKeys: 4, + lua: ` + ${PRELUDE} + local sinceId = ARGV[1] + local limit = tonumber(ARGV[2]) + + -- The index holds valid entries only, so an invalid since id misses ZSCORE. Its seq is still + -- on its own '#s' field, which keeps the id resolvable without indexing invalid rows. + local score = redis.call('ZSCORE', idxKey, sinceId) + if not score then + score = redis.call('HGET', eKey, sinceId .. '#s') + if not score then return nil end + end + + -- Env scoping is decided from the since entry itself, not from the window it produces. + local sinceRaw = redis.call('HGET', eKey, sinceId) or '' + + -- NEWEST-first with a limit, then reversed app-side. The engine reads createdAt desc / + -- take N / reverse, so the oldest-first form would return the wrong window entirely. + local ids = redis.call('ZREVRANGEBYSCORE', idxKey, '+inf', '(' .. score, 'LIMIT', 0, limit) + if #ids == 0 then return { sinceRaw, '' } end + + -- The head is the newest SURVIVING entry, and it is the only one whose cycle key is read. + -- Deriving the order after the loop keeps it paired with the row it is attached to: a row + -- dropped for a missing body must not donate its cycle data to the next one. + local out = { sinceRaw, '' } + local headId = nil + for i = 1, #ids do + local id = ids[i] + local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c') + if vals[1] then + if not headId then headId = id end + out[#out + 1] = id + out[#out + 1] = vals[1] + out[#out + 1] = vals[2] or '' + out[#out + 1] = vals[3] or '' + end + end + if headId then + out[2] = orderFor(redis.call('HGET', eKey, headId .. '#c')) + end + return out + `, + }); + } +} + +export function decodeWaitpointIds(present: boolean, orderJson: string): WaitpointIds { + const order: string[] = orderJson === "" ? [] : (JSON.parse(orderJson) as string[]); + return { present, distinctIds: [...new Set(order)], order }; +} + +declare module "@internal/redis" { + interface RedisCommander { + appendSnapshotEntry( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + kind: string, + id: string, + raw: string, + isValid: string, + isTerminal: string, + ttlMs: string, + cycleMode: string, + cycleSeqIn: string, + orderJson: string, + records: string, + orderCount: string, + expectedCur: string, + casEnabled: string, + callback?: Callback + ): Result; + readSnapshotById( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + id: string, + callback?: Callback + ): Result; + readLatestSnapshot( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + callback?: Callback + ): Result; + readSnapshotWaitpointIds( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + id: string, + callback?: Callback + ): Result; + readSnapshotsSince( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + sinceId: string, + limit: string, + callback?: Callback + ): Result; + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf4fea17902..a7a0a3e5d54 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1290,6 +1290,9 @@ importers: internal-packages/run-store: dependencies: + '@internal/redis': + specifier: workspace:* + version: link:../redis '@trigger.dev/core': specifier: workspace:* version: link:../../packages/core