feat(database): add infra-error classifier and read-retry util

Adds a shared classifier (isInfrastructureError / looksLikeConnectivityError)
recognising connection-blip failures (P1001/P1002/P1008/P1017, ECONNRESET,
"connection terminated", "server has closed the connection"), and
withInfraRetry — a retry helper gated by an enabled kill-switch (default off)
and the existing TokenBucketRetryBudget.

Only for operations safe to run more than once (reads, or writes made
idempotent); it never authorises retrying a bare non-idempotent write.

TRI-13553

Author: d-cs

internal-packages/database/src/index.ts

@@ -1,3 +1,5 @@
 export * from "../generated/prisma";
 export * from "./boundedIn";
+export * from "./infraError";
+export * from "./infraRetry";
 export * from "./transaction";

internal-packages/database/src/infraError.test.ts

@@ -0,0 +1,44 @@
+import { describe, expect, it } from "vitest";
+import { Prisma } from "../generated/prisma";
+import { isInfrastructureError, looksLikeConnectivityError } from "./infraError";
+
+const known = (code: string, message = "") =>
+  new Prisma.PrismaClientKnownRequestError(message, { code, clientVersion: "6.14.0" });
+
+describe("isInfrastructureError", () => {
+  it("treats connection-level Prisma codes as infrastructure errors", () => {
+    for (const code of ["P1001", "P1002", "P1008", "P1017"]) {
+      expect(isInfrastructureError(known(code, "boom"))).toBe(true);
+    }
+  });
+
+  it("does not treat query/validation errors as infrastructure errors", () => {
+    expect(isInfrastructureError(known("P2025", "record not found"))).toBe(false);
+    expect(isInfrastructureError(known("P2002", "unique constraint"))).toBe(false);
+  });
+
+  it("treats P2010 as infrastructure only when the message looks like connectivity loss", () => {
+    expect(isInfrastructureError(known("P2010", "Connection terminated unexpectedly"))).toBe(true);
+    expect(isInfrastructureError(known("P2010", "syntax error at or near"))).toBe(false);
+  });
+
+  it("treats init / panic / unknown request errors as infrastructure errors", () => {
+    expect(
+      isInfrastructureError(new Prisma.PrismaClientInitializationError("no db", "6.14.0"))
+    ).toBe(true);
+  });
+
+  it("recognises raw connectivity errno / messages", () => {
+    expect(isInfrastructureError({ code: "ECONNRESET" })).toBe(true);
+    expect(isInfrastructureError(new Error("server has closed the connection"))).toBe(true);
+    expect(isInfrastructureError(new Error("column does not exist"))).toBe(false);
+  });
+});
+
+describe("looksLikeConnectivityError", () => {
+  it("matches known errno codes and message fragments", () => {
+    expect(looksLikeConnectivityError({ code: "EHOSTUNREACH" })).toBe(true);
+    expect(looksLikeConnectivityError(new Error("Can't reach database server"))).toBe(true);
+    expect(looksLikeConnectivityError(new Error("relation does not exist"))).toBe(false);
+  });
+});

internal-packages/database/src/infraError.ts

@@ -0,0 +1,50 @@
+import { Prisma } from "../generated/prisma";
+
+// Prisma connectivity / infrastructure error codes — connection-level failures,
+// not query- or validation-level ones (e.g. P1001 "Can't reach database server").
+const INFRASTRUCTURE_PRISMA_CODES = new Set(["P1001", "P1002", "P1008", "P1017"]);
+
+const CONNECTIVITY_ERRNO = new Set([
+  "ECONNREFUSED",
+  "ENOTFOUND",
+  "ETIMEDOUT",
+  "ECONNRESET",
+  "EHOSTUNREACH",
+  "EPIPE",
+]);
+
+const CONNECTIVITY_MESSAGE =
+  /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|ECONNRESET|EHOSTUNREACH|database not reachable|can't reach database|connection terminated|server has closed the connection|timed out fetching a new connection/i;
+
+/** True for an errno/message that looks like a lost or unreachable connection. */
+export function looksLikeConnectivityError(error: unknown): boolean {
+  const e = error as { code?: unknown; message?: unknown };
+  if (typeof e?.code === "string" && CONNECTIVITY_ERRNO.has(e.code)) {
+    return true;
+  }
+  return typeof e?.message === "string" && CONNECTIVITY_MESSAGE.test(e.message);
+}
+
+/**
+ * True when `error` is a Prisma infrastructure/connectivity failure (DB
+ * unreachable, timed out, connection dropped) rather than a query- or
+ * validation-level error. This is the retry gate for connection blips.
+ */
+export function isInfrastructureError(error: unknown): boolean {
+  if (
+    error instanceof Prisma.PrismaClientInitializationError ||
+    error instanceof Prisma.PrismaClientRustPanicError ||
+    error instanceof Prisma.PrismaClientUnknownRequestError
+  ) {
+    return true;
+  }
+
+  if (error instanceof Prisma.PrismaClientKnownRequestError) {
+    if (INFRASTRUCTURE_PRISMA_CODES.has(error.code)) {
+      return true;
+    }
+    return error.code === "P2010" && looksLikeConnectivityError(error);
+  }
+
+  return looksLikeConnectivityError(error);
+}

internal-packages/database/src/infraRetry.test.ts

@@ -0,0 +1,197 @@
+import { describe, expect, it } from "vitest";
+import { Prisma } from "../generated/prisma";
+import { type RetryBudget } from "./transaction";
+import { withInfraRetry } from "./infraRetry";
+
+const options = { enabled: true, maxAttempts: 4, backoffMinMs: 0, backoffMaxMs: 0 };
+const noSleep = () => Promise.resolve();
+const retryable = () => true;
+
+// A thunk that throws `error` for its first `failTimes` calls, then resolves.
+function makeThunk(failTimes: number, error: unknown = new Error("infra"), value = "ok") {
+  let calls = 0;
+  return {
+    run: async () => {
+      calls++;
+      if (calls <= failTimes) throw error;
+      return value;
+    },
+    calls: () => calls,
+  };
+}
+
+describe("withInfraRetry", () => {
+  it("runs the thunk exactly once when disabled", async () => {
+    const t = makeThunk(Infinity);
+    await expect(
+      withInfraRetry(t.run, { options: { ...options, enabled: false }, isRetryable: retryable })
+    ).rejects.toThrow("infra");
+    expect(t.calls()).toBe(1);
+  });
+
+  it("retries a retryable error until it succeeds", async () => {
+    const t = makeThunk(2);
+    const result = await withInfraRetry(t.run, { options, isRetryable: retryable, sleep: noSleep });
+    expect(result).toBe("ok");
+    expect(t.calls()).toBe(3);
+  });
+
+  it("does not retry a non-retryable error", async () => {
+    const t = makeThunk(Infinity, new Error("query bug"));
+    await expect(
+      withInfraRetry(t.run, { options, isRetryable: () => false, sleep: noSleep })
+    ).rejects.toThrow("query bug");
+    expect(t.calls()).toBe(1);
+  });
+
+  it("gives up after maxAttempts", async () => {
+    const t = makeThunk(Infinity);
+    await expect(
+      withInfraRetry(t.run, { options, isRetryable: retryable, sleep: noSleep })
+    ).rejects.toThrow("infra");
+    expect(t.calls()).toBe(4);
+  });
+
+  it("stops retrying once the shared budget is exhausted", async () => {
+    const t = makeThunk(Infinity);
+    let tokens = 1;
+    const budget: RetryBudget = {
+      tryConsume: () => {
+        if (tokens > 0) {
+          tokens--;
+          return true;
+        }
+        return false;
+      },
+    };
+    await expect(
+      withInfraRetry(t.run, { options, isRetryable: retryable, budget, sleep: noSleep })
+    ).rejects.toThrow("infra");
+    // first attempt + one budgeted retry, then the budget denies the next
+    expect(t.calls()).toBe(2);
+  });
+
+  it("runs the thunk exactly once when maxAttempts <= 1", async () => {
+    const t = makeThunk(Infinity);
+    await expect(
+      withInfraRetry(t.run, {
+        options: { ...options, maxAttempts: 1 },
+        isRetryable: retryable,
+        sleep: noSleep,
+      })
+    ).rejects.toThrow("infra");
+    expect(t.calls()).toBe(1);
+  });
+
+  it("runs the thunk once (never skipped, never unbounded) for a non-finite maxAttempts", async () => {
+    for (const maxAttempts of [NaN, Infinity]) {
+      const t = makeThunk(Infinity);
+      await expect(
+        withInfraRetry(t.run, {
+          options: { ...options, maxAttempts },
+          isRetryable: retryable,
+          sleep: noSleep,
+        })
+      ).rejects.toThrow("infra");
+      expect(t.calls()).toBe(1);
+    }
+  });
+
+  it("floors a fractional maxAttempts to whole attempts", async () => {
+    const t = makeThunk(Infinity);
+    await expect(
+      withInfraRetry(t.run, {
+        options: { ...options, maxAttempts: 3.9 },
+        isRetryable: retryable,
+        sleep: noSleep,
+      })
+    ).rejects.toThrow("infra");
+    expect(t.calls()).toBe(3);
+  });
+
+  it("defaults to isInfrastructureError when no isRetryable is provided", async () => {
+    const infra = new Prisma.PrismaClientKnownRequestError("boom", {
+      code: "P1001",
+      clientVersion: "6.14.0",
+    });
+    const t = makeThunk(1, infra);
+    // no isRetryable in the config — the module must default to the classifier
+    const result = await withInfraRetry(t.run, { options, sleep: noSleep });
+    expect(result).toBe("ok");
+    expect(t.calls()).toBe(2);
+  });
+
+  it("with the default classifier, does not retry a real query error", async () => {
+    const queryError = new Prisma.PrismaClientKnownRequestError("not found", {
+      code: "P2025",
+      clientVersion: "6.14.0",
+    });
+    const t = makeThunk(Infinity, queryError);
+    await expect(withInfraRetry(t.run, { options, sleep: noSleep })).rejects.toBe(queryError);
+    expect(t.calls()).toBe(1);
+  });
+
+  it("clamps a swapped min/max backoff", async () => {
+    const t = makeThunk(1);
+    const delays: number[] = [];
+    await withInfraRetry(t.run, {
+      options: { enabled: true, maxAttempts: 2, backoffMinMs: 300, backoffMaxMs: 100 },
+      isRetryable: retryable,
+      sleep: noSleep,
+      random: () => 0.9,
+      onRetry: ({ delayMs }) => delays.push(delayMs),
+    });
+    // min > max collapses to [100, 100], so the delay is 100 regardless of random
+    expect(delays).toEqual([100]);
+  });
+
+  it("computes a jittered delay within bounds and reports it", async () => {
+    const t = makeThunk(1);
+    const delays: number[] = [];
+    await withInfraRetry(t.run, {
+      options: { enabled: true, maxAttempts: 3, backoffMinMs: 100, backoffMaxMs: 300 },
+      isRetryable: retryable,
+      sleep: noSleep,
+      random: () => 0.5,
+      onRetry: ({ delayMs }) => delays.push(delayMs),
+    });
+    expect(delays[0]).toBe(200);
+  });
+
+  it("keeps the delay finite and in range for bad backoff bounds and out-of-range random", async () => {
+    // NaN backoff bound must not produce a NaN delay.
+    const t1 = makeThunk(1);
+    const nanDelays: number[] = [];
+    await withInfraRetry(t1.run, {
+      options: { enabled: true, maxAttempts: 2, backoffMinMs: 0, backoffMaxMs: NaN },
+      isRetryable: retryable,
+      sleep: noSleep,
+      onRetry: ({ delayMs }) => nanDelays.push(delayMs),
+    });
+    expect(nanDelays[0]).toBe(0);
+
+    // random() above 1 clamps to the high bound.
+    const t2 = makeThunk(1);
+    const highDelays: number[] = [];
+    await withInfraRetry(t2.run, {
+      options: { enabled: true, maxAttempts: 2, backoffMinMs: 100, backoffMaxMs: 300 },
+      isRetryable: retryable,
+      sleep: noSleep,
+      random: () => 2,
+      onRetry: ({ delayMs }) => highDelays.push(delayMs),
+    });
+    expect(highDelays[0]).toBe(300);
+
+    // random() below 0 clamps to the low bound.
+    const t3 = makeThunk(1);
+    const lowDelays: number[] = [];
+    await withInfraRetry(t3.run, {
+      options: { enabled: true, maxAttempts: 2, backoffMinMs: 100, backoffMaxMs: 300 },
+      isRetryable: retryable,
+      sleep: noSleep,
+      random: () => -1,
+      onRetry: ({ delayMs }) => lowDelays.push(delayMs),
+    });
+    expect(lowDelays[0]).toBe(100);
+  });
+});

internal-packages/database/src/infraRetry.ts

@@ -0,0 +1,90 @@
+import { isInfrastructureError } from "./infraError";
+import { type RetryBudget, UNLIMITED_RETRY_BUDGET } from "./transaction";
+
+/** Retry tuning for {@link withInfraRetry}. */
+export type InfraRetryOptions = {
+  /** Kill switch. When false, the thunk runs exactly once. Default off at call sites. */
+  enabled: boolean;
+  /** Total attempts including the first. `1` (or less) disables retrying. */
+  maxAttempts: number;
+  /** Lower bound of the jittered backoff between attempts, in ms. */
+  backoffMinMs: number;
+  /** Upper bound of the jittered backoff between attempts, in ms. */
+  backoffMaxMs: number;
+};
+
+export type InfraRetryConfig = {
+  options: InfraRetryOptions;
+  /** Shared budget; defaults to {@link UNLIMITED_RETRY_BUDGET}. */
+  budget?: RetryBudget;
+  /** Predicate for a retryable error; defaults to {@link isInfrastructureError}. */
+  isRetryable?: (error: unknown) => boolean;
+  sleep?: (ms: number) => Promise<void>;
+  random?: () => number;
+  onRetry?: (info: { attempt: number; delayMs: number; error: unknown }) => void;
+};
+
+function defaultSleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/**
+ * Runs `run` and retries it on a connection-blip error ({@link isInfrastructureError}
+ * by default), up to `maxAttempts` with jittered backoff, gated by a shared
+ * `budget` so a mass freeze can't amplify into a retry storm. Any other error
+ * (or an exhausted budget) rethrows immediately.
+ *
+ * ONLY wrap operations that are safe to run more than once: reads, or writes
+ * that have been made idempotent. Never wrap a bare non-idempotent write.
+ */
+export async function withInfraRetry<R>(
+  run: () => Promise<R>,
+  config?: InfraRetryConfig
+): Promise<R> {
+  if (!config || !config.options.enabled) {
+    return run();
+  }
+
+  const { backoffMinMs, backoffMaxMs } = config.options;
+  // Bound the loop defensively: a non-finite maxAttempts (NaN/Infinity from a bad
+  // env parse) or a value below 2 means "run once, no retry" — never a skipped run
+  // or an unbounded loop. Fractional values floor to whole attempts.
+  const maxAttempts = Number.isFinite(config.options.maxAttempts)
+    ? Math.floor(config.options.maxAttempts)
+    : 1;
+  if (maxAttempts <= 1) {
+    return run();
+  }
+
+  const budget = config.budget ?? UNLIMITED_RETRY_BUDGET;
+  const isRetryable = config.isRetryable ?? isInfrastructureError;
+  const sleep = config.sleep ?? defaultSleep;
+  const random = config.random ?? Math.random;
+
+  let lastError: unknown;
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    try {
+      return await run();
+    } catch (error) {
+      lastError = error;
+      if (attempt >= maxAttempts || !isRetryable(error) || !budget.tryConsume()) {
+        throw error;
+      }
+      // Defend the delay math: non-finite bounds fall back to 0, and an injected
+      // random() outside [0, 1] is clamped, so delayMs is always finite and within
+      // [low, high]. A swapped min/max still collapses to the smaller bound.
+      const min = Number.isFinite(backoffMinMs) ? backoffMinMs : 0;
+      const max = Number.isFinite(backoffMaxMs) ? backoffMaxMs : min;
+      const low = Math.max(0, Math.min(min, max));
+      const high = Math.max(low, max);
+      const r = Math.min(1, Math.max(0, random()));
+      const delayMs = Math.round(low + r * (high - low));
+      config.onRetry?.({ attempt, delayMs, error });
+      await sleep(delayMs);
+    }
+  }
+
+  // Unreachable: the final attempt always returns or throws above. Present so the
+  // function is statically known to return `R` or throw on every path.
+  throw lastError;
+}