feat(run-engine, run-store): durable execution-snapshot store protocol

No user-facing change. This adds dormant execution-snapshot storage
primitives that are not connected to a production execution path.

Mono-RevId: 131c951f78bca8818c3f7c9f11ba75f7a7f2db3e

Author: d-cs

docker/config/redis-cluster-init.sh

@@ -0,0 +1,60 @@
+#!/bin/sh
+# Brings up a six-node Redis cluster (3 masters, 3 replicas) inside ONE container.
+#
+# One container, not six, for two reasons that both come from Docker on macOS:
+#
+#   1. Cluster nodes gossip with each other using the address they ADVERTISE. Six separate
+#      containers each advertising 127.0.0.1 would each be talking to themselves, and the cluster
+#      never forms. Sharing one network namespace makes 127.0.0.1 mean the same thing to all six.
+#   2. A client on the host follows MOVED redirects to that same advertised address, so the ports
+#      published below resolve correctly from outside.
+#
+# Each node gets its own directory: six nodes sharing one working directory fight over nodes.conf
+# and all but the first fail to start, silently, because they are daemonised.
+set -e
+
+NODES="1 2 3 4 5 6"
+
+for i in $NODES; do
+  PORT=$((7000 + i))
+  BUS=$((17000 + i))
+  mkdir -p "/data/$PORT"
+  redis-server \
+    --port "$PORT" \
+    --cluster-enabled yes \
+    --cluster-node-timeout 5000 \
+    --cluster-announce-ip 127.0.0.1 \
+    --cluster-announce-port "$PORT" \
+    --cluster-announce-bus-port "$BUS" \
+    --cluster-config-file "nodes-$PORT.conf" \
+    --dir "/data/$PORT" \
+    --protected-mode no \
+    --appendonly no \
+    --save '' \
+    --logfile "/data/$PORT/redis.log" \
+    --daemonize yes
+done
+
+# Daemonised servers report failures only to their own log, so check before forming the cluster.
+sleep 5
+for i in $NODES; do
+  PORT=$((7000 + i))
+  if ! redis-cli -p "$PORT" ping >/dev/null 2>&1; then
+    echo "node $PORT failed to start:"
+    tail -20 "/data/$PORT/redis.log"
+    exit 1
+  fi
+done
+
+# Idempotent: a restart with a populated /data already has slots assigned, so skip the create.
+if redis-cli -p 7001 cluster info | grep -q "cluster_state:ok"; then
+  echo "cluster already formed"
+else
+  redis-cli --cluster create \
+    127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 \
+    127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 \
+    --cluster-replicas 1 --cluster-yes
+fi
+
+redis-cli -p 7001 cluster info | grep -E "cluster_state|cluster_size"
+exec tail -f /dev/null

docker/docker-compose.extras.yml

@@ -14,6 +14,7 @@
 name: triggerdotdev-docker
 
 volumes:
+  redis-cluster-data:
   prometheus-data:
   grafana-data:
 
@@ -95,6 +96,7 @@ services:
     restart: always
     volumes:
       - ./config/prometheus.yml:/etc/prometheus/prometheus.yml:ro
+      - ./config/alerts:/etc/prometheus/alerts:ro
       - prometheus-data:/prometheus
     ports:
       - "${PROMETHEUS_HOST_PORT:-9090}:9090"
@@ -122,3 +124,39 @@ services:
       - app_network
     depends_on:
       - prometheus
+
+  # Six-node Redis cluster (3 masters, 3 replicas) for validating the execution-snapshot store.
+  # The store's single-slot guarantee only means anything against a cluster-mode endpoint, and the
+  # recovery worker's per-master fan-out is unreachable on a single node.
+  #
+  #   RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST=127.0.0.1
+  #   RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT=7001
+  #   RUN_ENGINE_SNAPSHOT_STORE_REDIS_CLUSTER_MODE_ENABLED=1
+  #
+  # ioredis discovers the other five nodes from that one seed.
+  redis-cluster:
+    container_name: ${CONTAINER_PREFIX:-}redis-cluster
+    image: redis:7.2@sha256:74566c6910d13ae61e7ce73ebd3127438a1fe805b309b097c323142719ec8a5b
+    restart: always
+    volumes:
+      - ./config/redis-cluster-init.sh:/init.sh:ro
+      - redis-cluster-data:/data
+    command: sh /init.sh
+    ports:
+      - "${REDIS_CLUSTER_PORT_1:-7001}:7001"
+      - "${REDIS_CLUSTER_PORT_2:-7002}:7002"
+      - "${REDIS_CLUSTER_PORT_3:-7003}:7003"
+      - "${REDIS_CLUSTER_PORT_4:-7004}:7004"
+      - "${REDIS_CLUSTER_PORT_5:-7005}:7005"
+      - "${REDIS_CLUSTER_PORT_6:-7006}:7006"
+      - "17001:17001"
+      - "17002:17002"
+      - "17003:17003"
+      - "17004:17004"
+      - "17005:17005"
+      - "17006:17006"
+    healthcheck:
+      test: ["CMD-SHELL", "redis-cli -p 7001 cluster info | grep -q cluster_state:ok"]
+      interval: 5s
+      timeout: 3s
+      retries: 20

internal-packages/redis/package.json

@@ -10,6 +10,8 @@
     "@trigger.dev/core": "workspace:*"
   },
   "scripts": {
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "test": "vitest",
+    "test:watch": "vitest"
   }
 }

internal-packages/redis/src/cluster.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect, it } from "vitest";
+import { Cluster, createRedisClusterClient, defaultReconnectOnError } from "./index.js";
+
+// Port 6399 is deliberately closed: these assertions are about the client the factory builds, not
+// about a connection.
+const NODES = [{ host: "127.0.0.1", port: 6399 }];
+
+type InnerOptions = {
+  options: {
+    redisOptions?: {
+      reconnectOnError?: unknown;
+      maxRetriesPerRequest?: number;
+      retryStrategy?: unknown;
+      keyPrefix?: string;
+    };
+  };
+};
+
+function innerOptionsOf(client: Cluster) {
+  return (client as unknown as InnerOptions).options.redisOptions;
+}
+
+describe("createRedisClusterClient", () => {
+  it("returns a Cluster instance", async () => {
+    const client = createRedisClusterClient({ nodes: NODES });
+    try {
+      expect(client).toBeInstanceOf(Cluster);
+    } finally {
+      await client.quit().catch(() => undefined);
+    }
+  });
+
+  it("installs defaultReconnectOnError on the inner per-node options", async () => {
+    const client = createRedisClusterClient({ nodes: NODES });
+    try {
+      expect(innerOptionsOf(client)?.reconnectOnError).toBe(defaultReconnectOnError);
+    } finally {
+      await client.quit().catch(() => undefined);
+    }
+  });
+
+  it("carries the retry defaults onto the inner options", async () => {
+    const client = createRedisClusterClient({ nodes: NODES });
+    try {
+      const inner = innerOptionsOf(client);
+      expect(inner?.maxRetriesPerRequest).toBeTypeOf("number");
+      expect(inner?.retryStrategy).toBeTypeOf("function");
+    } finally {
+      await client.quit().catch(() => undefined);
+    }
+  });
+
+  it("lets caller redisOptions override the defaults", async () => {
+    const client = createRedisClusterClient({
+      nodes: NODES,
+      redisOptions: { keyPrefix: "engine:", maxRetriesPerRequest: 3 },
+    });
+    try {
+      const inner = innerOptionsOf(client);
+      expect(inner?.keyPrefix).toBe("engine:");
+      expect(inner?.maxRetriesPerRequest).toBe(3);
+    } finally {
+      await client.quit().catch(() => undefined);
+    }
+  });
+
+  it("keeps mapping READONLY, LOADING and UNBLOCKED to a reconnect-and-retry", () => {
+    expect(defaultReconnectOnError(new Error("READONLY against a read only replica"))).toBe(2);
+    expect(defaultReconnectOnError(new Error("LOADING Redis is loading the dataset"))).toBe(2);
+    expect(defaultReconnectOnError(new Error("UNBLOCKED force unblock"))).toBe(2);
+    expect(defaultReconnectOnError(new Error("ERR unknown command"))).toBe(false);
+  });
+});

internal-packages/redis/src/clusterFailFast.test.ts

@@ -0,0 +1,24 @@
+import { describe, expect, it } from "vitest";
+import { createRedisClusterClient } from "./index.js";
+
+// The offline queue on a Cluster client is a CLUSTER-level option. Setting it on the inner
+// per-node redisOptions leaves the cluster queueing commands while it cannot refresh its slot
+// cache, so a command issued during an outage waits instead of failing and the caller hangs.
+describe("cluster client against an unreachable cluster", () => {
+  it("rejects a command rather than queueing it", async () => {
+    const client = createRedisClusterClient({
+      // Nothing listens here.
+      nodes: [{ host: "127.0.0.1", port: 6391 }],
+      redisOptions: { commandTimeout: 300 },
+      failFast: true,
+    });
+
+    const started = Date.now();
+    await expect(client.get("anything")).rejects.toThrow();
+    const elapsed = Date.now() - started;
+
+    expect(elapsed).toBeLessThan(5000);
+
+    client.disconnect();
+  });
+});

internal-packages/redis/src/failFast.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, it } from "vitest";
+import { createRedisClient } from "./index.js";
+
+// A snapshot-store append sits on a request path. With the offline queue enabled and no command
+// timeout, a command issued while the endpoint is unreachable waits for a reconnect instead of
+// failing, so the request hangs rather than falling back to Postgres.
+describe("fail-fast options against an unreachable endpoint", () => {
+  it("rejects rather than hanging when the offline queue is off and a timeout is set", async () => {
+    const client = createRedisClient({
+      // Nothing listens here.
+      host: "127.0.0.1",
+      port: 6390,
+      enableOfflineQueue: false,
+      commandTimeout: 300,
+      lazyConnect: true,
+      retryStrategy: () => null,
+    });
+
+    const started = Date.now();
+    await expect(client.get("anything")).rejects.toThrow();
+    const elapsed = Date.now() - started;
+
+    // Generous, but far below the indefinite wait the offline queue produces.
+    expect(elapsed).toBeLessThan(3000);
+
+    client.disconnect();
+  });
+});

internal-packages/redis/src/index.ts

@@ -1,4 +1,10 @@
-import { type Cluster, Redis, type RedisOptions } from "ioredis";
+import {
+  Redis,
+  type Cluster,
+  type ClusterNode,
+  type ClusterOptions,
+  type RedisOptions,
+} from "ioredis";
 import { Logger } from "@trigger.dev/core/logger";
 
 export {
@@ -85,3 +91,55 @@ export function createRedisClient(
 
   return client;
 }
+
+export type RedisClusterClientOptions = {
+  nodes: ClusterNode[];
+  clusterOptions?: Omit<ClusterOptions, "redisOptions">;
+  redisOptions?: RedisOptions;
+  /**
+   * Fail a command while the cluster is unreachable instead of queueing it. For a caller on a
+   * request path, where a queued command means a hung request rather than a slow one.
+   */
+  failFast?: boolean;
+};
+
+/**
+ * Cluster-mode client. `defaultOptions` go on the INNER per-node options, so a role swap gets the
+ * same reconnect-and-retry treatment a single-node client already gets.
+ */
+export function createRedisClusterClient(
+  options: RedisClusterClientOptions,
+  handlers?: { onError?: (err: Error) => void }
+): Cluster {
+  const client = new Redis.Cluster(options.nodes, {
+    // The offline queue is a CLUSTER-level setting, separate from the per-node one below. While a
+    // cluster cannot refresh its slot cache it queues commands here, so a caller that wants a
+    // failure during an outage rather than a wait has to turn THIS one off. Default stays `true`,
+    // matching ioredis, so only a caller that asks for it changes behaviour.
+    ...(options.failFast && { enableOfflineQueue: false }),
+    ...options.clusterOptions,
+    redisOptions: {
+      ...defaultOptions,
+      ...(options.failFast && { enableOfflineQueue: false }),
+      ...options.redisOptions,
+    },
+  });
+
+  if (process.env.VITEST) {
+    client.on("error", () => {});
+    return client;
+  }
+
+  client.on("error", (error) => {
+    if (handlers?.onError) {
+      handlers.onError(error);
+    } else {
+      logger.error(`Redis cluster client error:`, {
+        error,
+        keyPrefix: options.redisOptions?.keyPrefix,
+      });
+    }
+  });
+
+  return client;
+}

internal-packages/redis/vitest.config.ts

@@ -0,0 +1,10 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+  test: {
+    include: ["**/*.test.ts"],
+    globals: true,
+    isolate: true,
+    testTimeout: 10_000,
+  },
+});