feat(run-store): keyspace and entry helpers for the Redis snapshot store

internal-packages/run-store/package.json

@@ -14,6 +14,7 @@
     }
   },
   "dependencies": {
+    "@internal/redis": "workspace:*",
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/database": "workspace:*"
   },

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";

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);
+  });
+});

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:<n> 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<string, unknown>;
+  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;
+};

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