From dd3a1c0c541182c264670fc874fabf0252b8ef89 Mon Sep 17 00:00:00 2001 From: Daniel Sutton <45313566+d-cs@users.noreply.github.com> Date: Fri, 21 Aug 2026 18:01:58 +0100 Subject: [PATCH] feat(run-store): Redis-backed store for the run execution-state log (#4754) Adds `RedisSnapshotStore` to `@internal/run-store`: a Redis-backed, append-only store for a run's execution-state log, as an alternative to keeping that log in Postgres. Nothing constructs it. No existing code path can reach it, so merging this changes no behaviour. The store, the wiring that would use it, and the switch that would enable it are deliberately separate changes. ## Design Four keys per run, plus one key per wait cycle, all sharing a `{runId}` hash tag. Every mutation for a run therefore lands in one cluster slot, and each operation is a single Lua script. No script mints a key name. Dynamic keys are derived from `KEYS[1]` by string surgery, because ioredis applies `keyPrefix` only to the KEYS array: a key built inside Lua would be unprefixed while the client wrote a prefixed one. Retention is keyed to run completion. A non-terminal run's keys carry no expiry at all, since a suspended run can wait indefinitely with nothing left to refresh a TTL. The terminal transition sets the completion expiry once, and a write arriving after completion re-applies that same expiry rather than a live one, so a stale client cannot resurrect a key. Entry JSON round-trips byte for byte. No script calls `cjson`, and the values the store assigns itself live in their own hash fields instead of being patched into the caller's document. Sizes are observed, never enforced. Entry and cycle-key bytes are recorded, with a warning above a configurable mark. Nothing rejects, truncates, or spills. `append` takes an optional expected-current-snapshot argument. Left out, it advances the pointer unconditionally, matching the Postgres behaviour it replaces. Supplied, it advances only on a match and otherwise reports the conflict without writing. Covered by 48 tests against a real Redis container, including the retention transitions, the single-slot guarantee under a key prefix, and tenant-scoped reads. --- internal-packages/run-store/package.json | 1 + internal-packages/run-store/src/index.ts | 1 + .../run-store/src/redisSnapshotStore.test.ts | 1226 +++++++++++++++++ .../run-store/src/redisSnapshotStore.ts | 667 +++++++++ pnpm-lock.yaml | 3 + 5 files changed, 1898 insertions(+) create mode 100644 internal-packages/run-store/src/redisSnapshotStore.test.ts create mode 100644 internal-packages/run-store/src/redisSnapshotStore.ts diff --git a/internal-packages/run-store/package.json b/internal-packages/run-store/package.json index 110c3b490..7263a6de0 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 160f9cdad..3717dc015 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 000000000..369d79e53 --- /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 000000000..7b60843e2 --- /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 bf4fea179..a7a0a3e5d 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