fix(run-store): fall back to Postgres when the snapshot store stops answering

Two faults under one cause. The read paths fell back on a miss and on a
dangling cycle but not on an error, so a Redis that stopped answering
turned an engine read into a throw once the command timed out. Postgres
holds every row below redis-only, so falling back is strictly better
than failing. At redis-only it still throws, because nothing else holds
the rows and an empty history served as real is worse than an error.

The second is the cost of the first probe for a run a process has not
seen. The residency cache removes the steady-state round trip but not
that one, and under a brownout it costs the whole retry budget. A
per-process breaker opens after a short run of connectivity failures and
refuses later calls locally, so the endpoint takes itself off the run
path with no operator and no deploy.

Script errors never count toward it. A wrong type or a missing script
fails identically on every retry against every node, so counting those
would open the circuit on a defect and stop mirroring runs that are
healthy.

Author: d-cs

internal-packages/run-store/src/circuitBreaker.test.ts

@@ -0,0 +1,95 @@
+// The residency cache removes the steady-state round trip, but a process that has not yet seen a run
+// still probes once, and under a brownout that one probe costs the full retry budget (measured: about
+// 2.07s, from 4 attempts at a 500ms command timeout plus backoff). The breaker bounds that: after a
+// few connectivity failures the store stops trying at all, so a sick Redis takes itself off the run
+// path without an operator.
+import { describe, expect, it } from "vitest";
+import { CircuitBreaker, SnapshotStoreUnavailableError } from "./circuitBreaker.js";
+
+const OPTS = { failureThreshold: 3, openDurationMs: 1_000 };
+
+function connectivityError(message = "Command timed out"): Error {
+  return new Error(message);
+}
+
+describe("CircuitBreaker", () => {
+  it("passes calls through while closed", async () => {
+    const breaker = new CircuitBreaker(OPTS);
+    expect(await breaker.run(async () => "ok")).toBe("ok");
+  });
+
+  it("stays closed while failures are below the threshold", async () => {
+    const breaker = new CircuitBreaker(OPTS);
+    for (let i = 0; i < 2; i++) {
+      await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+    }
+    expect(breaker.state).toBe("closed");
+  });
+
+  it("opens on the threshold and then refuses without calling through", async () => {
+    const breaker = new CircuitBreaker(OPTS);
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+    }
+    expect(breaker.state).toBe("open");
+
+    let called = false;
+    await expect(
+      breaker.run(async () => {
+        called = true;
+        return "ok";
+      })
+    ).rejects.toBeInstanceOf(SnapshotStoreUnavailableError);
+    // The point of the breaker: the doomed call is not made, so it costs nothing rather than a
+    // timeout.
+    expect(called).toBe(false);
+  });
+
+  it("a success resets the count, so intermittent failures never accumulate to an open", async () => {
+    const breaker = new CircuitBreaker(OPTS);
+    for (let i = 0; i < 5; i++) {
+      await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+      await breaker.run(async () => "ok");
+    }
+    expect(breaker.state).toBe("closed");
+  });
+
+  it("half-opens after the open window and closes on a successful trial", async () => {
+    const breaker = new CircuitBreaker({ failureThreshold: 3, openDurationMs: 20 });
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+    }
+    expect(breaker.state).toBe("open");
+
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    expect(await breaker.run(async () => "recovered")).toBe("recovered");
+    expect(breaker.state).toBe("closed");
+  });
+
+  it("re-opens when the trial call fails again", async () => {
+    const breaker = new CircuitBreaker({ failureThreshold: 3, openDurationMs: 20 });
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+    }
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    await expect(breaker.run(async () => Promise.reject(connectivityError()))).rejects.toThrow();
+    expect(breaker.state).toBe("open");
+  });
+
+  it("ignores a script error, because a bug is not an outage", async () => {
+    // WRONGTYPE and other Lua errors mean the data or the script is wrong, and every retry against
+    // every node will fail the same way. Counting them would open the breaker on a defect and take
+    // the mirror down for runs that are perfectly healthy.
+    const breaker = new CircuitBreaker(OPTS);
+    for (let i = 0; i < 10; i++) {
+      await expect(
+        breaker.run(async () =>
+          Promise.reject(
+            new Error("WRONGTYPE Operation against a key holding the wrong kind of value")
+          )
+        )
+      ).rejects.toThrow();
+    }
+    expect(breaker.state).toBe("closed");
+  });
+});

internal-packages/run-store/src/circuitBreaker.ts

@@ -0,0 +1,104 @@
+/**
+ * Thrown instead of attempting a call the breaker has decided will fail. Named so callers can tell
+ * "Redis is not answering" apart from "Redis answered and said no", which need different handling:
+ * the first falls back, the second is a real result.
+ */
+export class SnapshotStoreUnavailableError extends Error {
+  constructor(message = "snapshot store is unavailable (circuit open)") {
+    super(message);
+    this.name = "SnapshotStoreUnavailableError";
+  }
+}
+
+export type CircuitBreakerOptions = {
+  /** Consecutive connectivity failures that open the circuit. */
+  failureThreshold?: number;
+  /** How long the circuit stays open before one trial call is allowed through. */
+  openDurationMs?: number;
+  /** Injectable for tests. Defaults to Date.now. */
+  now?: () => number;
+};
+
+export type CircuitState = "closed" | "open" | "half-open";
+
+const DEFAULT_FAILURE_THRESHOLD = 3;
+const DEFAULT_OPEN_DURATION_MS = 10_000;
+
+/**
+ * Connectivity, not correctness.
+ *
+ * A Lua error means the script or the data is wrong: every retry, against every node, fails the same
+ * way, and no amount of waiting helps. Counting those would open the circuit on a defect and stop
+ * mirroring runs that are perfectly healthy. A timeout, a closed connection or a down cluster is the
+ * opposite: nothing about the request is wrong, the server is simply not answering.
+ */
+function isConnectivityFailure(error: unknown): boolean {
+  const message = error instanceof Error ? error.message : String(error);
+  if (/WRONGTYPE|NOSCRIPT|ERR |Lua |user_script/i.test(message)) {
+    return false;
+  }
+  return /timed? ?out|ETIMEDOUT|ECONNREFUSED|ECONNRESET|EPIPE|Connection is closed|Stream isn't writeable|CLUSTERDOWN|max retries|Failed to refresh slots/i.test(
+    message
+  );
+}
+
+/**
+ * Stops a process paying the retry budget over and over for a Redis that is not answering.
+ *
+ * The residency cache already removes the steady-state round trip; this bounds what is left. One
+ * transition's worth of failed attempts opens the circuit, and every later call returns immediately
+ * instead of waiting out another timeout. A sick Redis therefore takes ITSELF off the run path,
+ * which is the property that makes the low dial positions inert under fault without an operator
+ * setting a second control.
+ *
+ * Per process and per client. The sweep holds its own connection and must not trip this one: a long
+ * scan failing is not evidence that the hot path cannot write.
+ */
+export class CircuitBreaker {
+  readonly #failureThreshold: number;
+  readonly #openDurationMs: number;
+  readonly #now: () => number;
+  #consecutiveFailures = 0;
+  #openedAt?: number;
+
+  constructor(options: CircuitBreakerOptions = {}) {
+    this.#failureThreshold = options.failureThreshold ?? DEFAULT_FAILURE_THRESHOLD;
+    this.#openDurationMs = options.openDurationMs ?? DEFAULT_OPEN_DURATION_MS;
+    this.#now = options.now ?? Date.now;
+  }
+
+  get state(): CircuitState {
+    if (this.#openedAt === undefined) {
+      return "closed";
+    }
+    return this.#now() - this.#openedAt >= this.#openDurationMs ? "half-open" : "open";
+  }
+
+  async run<T>(fn: () => Promise<T>): Promise<T> {
+    if (this.state === "open") {
+      throw new SnapshotStoreUnavailableError();
+    }
+
+    // half-open lets exactly this one call through. It either closes the circuit or re-opens it.
+    try {
+      const result = await fn();
+      this.#consecutiveFailures = 0;
+      this.#openedAt = undefined;
+      return result;
+    } catch (error) {
+      if (!isConnectivityFailure(error)) {
+        // A success resets the counter and so does an unrelated error: only an unbroken run of
+        // connectivity failures is evidence the endpoint is gone.
+        this.#consecutiveFailures = 0;
+        throw error;
+      }
+
+      this.#consecutiveFailures += 1;
+      if (this.state === "half-open" || this.#consecutiveFailures >= this.#failureThreshold) {
+        this.#openedAt = this.#now();
+        this.#consecutiveFailures = 0;
+      }
+      throw error;
+    }
+  }
+}

internal-packages/run-store/src/redisSnapshotStore.ts

@@ -7,6 +7,7 @@ import {
 } from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
 import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas";
+import { CircuitBreaker, type CircuitBreakerOptions } from "./circuitBreaker.js";
 import { ResidencyCache } from "./residencyCache.js";
 
 export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string };
@@ -242,6 +243,8 @@ export type RedisSnapshotStoreOptions = RedisSnapshotStoreConnection & {
   completedTtlMs: number;
   /** Entries in the per-process residency cache. See {@link ResidencyCache}. */
   residencyCacheMax?: number;
+  /** Tuning for the per-process breaker. See {@link CircuitBreaker}. */
+  breaker?: CircuitBreakerOptions;
   sinceLimit?: number;
   highWater?: { entryBytes?: number; cycleKeyBytes?: number; cycleCount?: number };
   metrics?: SnapshotStoreMetrics;
@@ -279,6 +282,12 @@ export class RedisSnapshotStore {
    * replies that are the cache's only ground truth.
    */
   readonly #residency: ResidencyCache;
+  /**
+   * Bounds what the residency cache cannot: the first probe for a run this process has not seen,
+   * which under a brownout costs the whole retry budget. After a few connectivity failures the store
+   * stops calling out at all, so a sick Redis removes itself from the run path with no operator.
+   */
+  readonly #breaker: CircuitBreaker;
   #quit?: Promise<void>;
 
   constructor(options: RedisSnapshotStoreOptions) {
@@ -290,6 +299,7 @@ export class RedisSnapshotStore {
     this.#residency = new ResidencyCache({
       ...(options.residencyCacheMax !== undefined && { max: options.residencyCacheMax }),
     });
+    this.#breaker = new CircuitBreaker(options.breaker ?? {});
     this.ownsClient = options.client === undefined;
     this.redis =
       options.client ??
@@ -315,15 +325,25 @@ export class RedisSnapshotStore {
     await this.#quit;
   }
 
+  /**
+   * Every command goes through here, so the breaker sits on the one seam rather than on each method.
+   * Latency is still recorded for a refused call: a call that cost nothing because the circuit was
+   * open is exactly the thing an operator wants to see in the latency series.
+   */
   async #timed<T>(op: string, fn: () => Promise<T>): Promise<T> {
     const started = Date.now();
     try {
-      return await fn();
+      return await this.#breaker.run(fn);
     } finally {
       this.metrics?.recordLatency(op, Date.now() - started);
     }
   }
 
+  /** Test seam. */
+  get breakerState(): "closed" | "open" | "half-open" {
+    return this.#breaker.state;
+  }
+
   /** Test seam. */
   residencyFor(runId: string): "resident" | "non-resident" | undefined {
     return this.#residency.get(runId);

internal-packages/run-store/src/taskRunExecutionSnapshotStore.hotPath.test.ts

@@ -184,3 +184,71 @@ describe("Redis stays off the hot path for a non-resident run", () => {
     }
   );
 });
+
+describe("a Redis that stops answering", () => {
+  containerTest(
+    "serves reads from Postgres instead of throwing into the engine",
+    async ({ prisma, redisOptions }) => {
+      // The read paths fell back on a miss and on a dangling cycle, but not on an ERROR, so a
+      // brownout at redis-read turned an engine read into a throw once the command timed out.
+      const redis = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      const decorated = new TaskRunExecutionSnapshotStore(
+        new PostgresRunStore({ prisma, readOnlyPrisma: prisma }) as unknown as RunStore,
+        { store: redis, mode: "redis-read", readPercent: 100 }
+      );
+      try {
+        const env = await seedSnapshotEnvironment(prisma);
+        const runId = generateInternalId();
+        await decorated.createRun({
+          data: buildCreateRunData(runId, env),
+          snapshot: birthSnapshot(env),
+        });
+
+        // Reads work while Redis answers.
+        expect(await decorated.findLatestExecutionSnapshot(runId)).not.toBeNull();
+
+        // Now take Redis away underneath it, the way a brownout does.
+        await redis.quit();
+
+        const served = await decorated.findLatestExecutionSnapshot(runId);
+        expect(served).not.toBeNull();
+        expect(served!.runId).toBe(runId);
+      } finally {
+        await redis.quit();
+      }
+    }
+  );
+
+  containerTest("stops calling out once the circuit opens", async ({ prisma, redisOptions }) => {
+    const redis = new RedisSnapshotStore({
+      redisOptions,
+      completedTtlMs: COMPLETED_TTL_MS,
+      breaker: { failureThreshold: 2, openDurationMs: 60_000 },
+    });
+    const decorated = new TaskRunExecutionSnapshotStore(
+      new PostgresRunStore({ prisma, readOnlyPrisma: prisma }) as unknown as RunStore,
+      { store: redis, mode: "redis-read", readPercent: 100 }
+    );
+    try {
+      const env = await seedSnapshotEnvironment(prisma);
+      const runId = generateInternalId();
+      await decorated.createRun({
+        data: buildCreateRunData(runId, env),
+        snapshot: birthSnapshot(env),
+      });
+
+      await redis.quit();
+      expect(redis.breakerState).toBe("closed");
+
+      for (let i = 0; i < 4; i++) {
+        expect(await decorated.findLatestExecutionSnapshot(runId)).not.toBeNull();
+      }
+
+      // Open, so every later call is refused locally rather than waiting out another timeout. This
+      // is what bounds the cold-cache probe under a brownout.
+      expect(redis.breakerState).toBe("open");
+    } finally {
+      await redis.quit();
+    }
+  });
+});