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..7f93e0d54 --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -0,0 +1,48 @@ +// 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 } from "vitest"; +import { snapshotKeys, deriveOrder, isValidFor } 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); + }); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts new file mode 100644 index 000000000..a5a2e6092 --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -0,0 +1,92 @@ +import { createRedisClient, type Redis, type RedisOptions } 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 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 }; + +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; +}; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7d74cbbe9..297cb7897 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