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

## Summary

Adds a `RunStore` decorator that mirrors execution snapshots into Redis
alongside Postgres, plus the orphan-key sweep and the fault-injection
suite that prove the write protocol converges after a crash. Nothing
constructs it, so merging this changes no behaviour: the configuration,
the production wiring and the Redis client all arrive in later work.

The execution-state log is the hottest table in the run graph, and
moving it out of Postgres has to happen without a big-bang cutover. This
is the attachment point for that: a decorator that wraps the existing
storage interface and intercepts only the methods that touch snapshots,
so none of the many callers change.

## Design

Write order is the correctness property, and the two orders differ on
purpose.

A transition writes Postgres first and Redis second. A crash in the gap
leaves a run whose latest snapshot is stale, which is the state the
heartbeat stall watchdog already heals in production today.

A birth writes Redis first and Postgres second. A crash there leaves an
unreachable key for a run that does not exist. Postgres first would
instead leave a run with no snapshot at all, which the engine treats as
a hard error, so the run would be stuck.

Each order is chosen so the state a crash leaves behind is the harmless
one. A lost cross-store write is never recovered by a transaction or an
outbox; recovery is always the existing stall and repair job. A failed
append retries, then hands the run to that job, and never rethrows,
because Postgres has already committed and a throw would turn a healable
gap into a caller-visible error.

Inside a transaction the Redis half is staged and flushed only after the
commit, so a rollback cannot leave Redis holding a transition that never
happened.

Reads are shape matched. Two of the snapshot reads take arbitrary Prisma
arguments, and a key-value store cannot answer an arbitrary query, so
the decorator recognises exactly the shapes the engine sends and
delegates everything else. A miss falls back to Postgres, which is also
how runs created before any cutover keep working.

The sweep reaps under two rules, because neither can see what the other
leaves behind. A finished run whose keyspace never received its
completion expiry gets one applied. A keyspace with no run row at all,
past an age threshold, is deleted; that is a crashed birth, which is
non-terminal so it carries no expiry and has no run row, so the first
rule can never match it.

## Inertness

Three independent reasons this is a no-op if merged alone:

- Nothing constructs the decorator or the Redis store outside tests.
- No configuration reaches it, so the dial stays at its off position,
which is a pass-through that makes no Redis call.
- The existing Postgres store gains an off-by-default flag and two
optional input fields. Both default to today's behaviour, and only the
decorator would ever supply them.

## Notes for review

The snapshot id and the creation instant are both minted by the
decorator and written into both stores, so one snapshot has one identity
and one timestamp wherever it is read. Without that, the two stores
disagree on values that later tooling has to compare, and the cursor for
a snapshot window resolved from one store misfilters the window walked
in the other.

Three defects in this work passed the full existing test suites before
being found by review rather than by a test: the decorator wrote no wait
cycle at all, the snapshot window dropped the ordering used to give each
completed waitpoint its position in a batch, and the two stores stamped
different creation times. The common cause was that no test drove a
snapshot that actually carried waitpoints, and that the parity suite
compared a timestamp against a value it had just read back from the row
it was checking. Both gaps now have tests.

Author: d-cs

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