feat(run-store): add an execution-snapshot store decorator behind an off-by-default dial

internal-packages/redis/src/index.ts

@@ -1,7 +1,24 @@
-import { Redis, type RedisOptions } from "ioredis";
+import { type Cluster, Redis, type RedisOptions } from "ioredis";
 import { Logger } from "@trigger.dev/core/logger";
 
-export { Redis, type Callback, type RedisOptions, type Result, type RedisCommander } from "ioredis";
+export {
+  Redis,
+  Cluster,
+  type Callback,
+  type RedisOptions,
+  type ClusterNode,
+  type ClusterOptions,
+  type Result,
+  type RedisCommander,
+} from "ioredis";
+
+/**
+ * Either endpoint shape. A component that only issues key-addressed commands works against both, so
+ * it should accept this rather than pin itself to a standalone connection. Commands with no key —
+ * SCAN above all — do NOT fan out across a cluster, so anything that issues one must iterate
+ * `cluster.nodes("master")` itself.
+ */
+export type RedisClient = Redis | Cluster;
 
 /**
  * Reply-error -> reconnect mapping. Without this hook, an ElastiCache

internal-packages/run-engine/src/engine/tests/finalRunStatusParity.test.ts

@@ -0,0 +1,20 @@
+// The snapshot sweeper needs to know which run statuses are terminal, and it cannot import that
+// list: run-engine depends on run-store, not the other way round. So the list is duplicated, and
+// this is the only thing that keeps the copy honest.
+//
+// Without it, a status added here and not there makes the sweeper treat a finished run as live and
+// never apply its completion expiry. A status removed here and not there makes it treat a live run
+// as finished. The second one reaps state a run is still using.
+import { describe, expect, it } from "vitest";
+import { FINAL_RUN_STATUSES } from "@internal/run-store";
+import { getFinalRunStatuses } from "../statuses.js";
+
+describe("terminal run statuses", () => {
+  it("match between the engine and the snapshot sweeper", () => {
+    expect([...FINAL_RUN_STATUSES].sort()).toEqual([...getFinalRunStatuses()].sort());
+  });
+
+  it("are not empty, so the comparison cannot pass vacuously", () => {
+    expect(FINAL_RUN_STATUSES.length).toBeGreaterThan(0);
+  });
+});

internal-packages/run-engine/src/engine/tests/helpers/decoratedStore.ts

@@ -0,0 +1,75 @@
+// Builds the snapshot-store decorator over a real PostgresRunStore, for injection through the
+// engine's `store` option — the seam runStoreInjectability.test.ts already proves.
+//
+// The point of injecting it is that the engine suites keep their own assertions: the same flows,
+// the same expectations, a different store underneath.
+import {
+  PostgresRunStore,
+  RedisSnapshotStore,
+  TaskRunExecutionSnapshotStore,
+  type SnapshotFaultInjector,
+  type SnapshotRepairEnqueuer,
+  type SnapshotStoreMode,
+} from "@internal/run-store";
+import type { PrismaClient } from "@trigger.dev/database";
+import type { RedisOptions } from "@internal/redis";
+
+const COMPLETED_TTL_MS = 72 * 60 * 60 * 1000;
+
+export type DecoratedStoreHarness = {
+  store: TaskRunExecutionSnapshotStore;
+  redis: RedisSnapshotStore;
+  /** Every read the decorator served, and which store answered it. */
+  reads: { method: string; source: "redis" | "postgres" }[];
+  /** Every append outcome, keyed by the write site that produced it. */
+  writes: { site: string; outcome: string }[];
+  /** Runs handed to the repair job because their append was lost. */
+  repairs: { runId: string; snapshotId: string; executionStatus: string }[];
+  quit(): Promise<void>;
+};
+
+export function buildDecoratedStore(opts: {
+  prisma: PrismaClient;
+  redisOptions: RedisOptions;
+  mode: SnapshotStoreMode;
+  readPercent?: number;
+  faults?: SnapshotFaultInjector;
+  onAppendFailure?: SnapshotRepairEnqueuer;
+}): DecoratedStoreHarness {
+  const redis = new RedisSnapshotStore({
+    redisOptions: opts.redisOptions,
+    completedTtlMs: COMPLETED_TTL_MS,
+  });
+
+  const reads: DecoratedStoreHarness["reads"] = [];
+  const writes: DecoratedStoreHarness["writes"] = [];
+  const repairs: DecoratedStoreHarness["repairs"] = [];
+
+  const store = new TaskRunExecutionSnapshotStore(
+    new PostgresRunStore({ prisma: opts.prisma as never, readOnlyPrisma: opts.prisma as never }),
+    {
+      store: redis,
+      mode: opts.mode,
+      readPercent: opts.readPercent ?? 100,
+      ...(opts.faults && { faults: opts.faults }),
+      onAppendFailure: async (args) => {
+        repairs.push(args);
+        await opts.onAppendFailure?.(args);
+      },
+      metrics: {
+        recordWrite: (site, outcome) => writes.push({ site, outcome }),
+        recordAppendFailed: () => {},
+        recordRead: (method, source) => reads.push({ method, source }),
+      },
+    }
+  );
+
+  return {
+    store,
+    redis,
+    reads,
+    writes,
+    repairs,
+    quit: () => redis.quit(),
+  };
+}