feat(webapp,run-store,database): env-configurable transaction resilience (maxWait + tx-start retry)

.server-changes/transaction-resilience-during-db-interruptions.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Triggering tasks is now more resilient to brief, transient service interruptions, so short stalls are less likely to surface as errors.

apps/webapp/app/db.server.ts

@@ -32,6 +32,13 @@ import {
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
 import { assertControlPlaneCoresidencyAdvisory } from "./v3/runOpsMigration/controlPlaneCoresidencySentinel.server";
 import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server";
+import {
+  controlPlaneTransactionResilience,
+  registerTransactionResilience,
+  resilienceForClient,
+  runOpsLegacyTransactionResilience,
+  runOpsTransactionResilience,
+} from "./v3/transactionResilience.server";
 import type { Span } from "@opentelemetry/api";
 import { context, trace } from "@opentelemetry/api";
 import { queryPerformanceMonitor } from "./utils/queryPerformanceMonitor.server";
@@ -59,6 +66,18 @@ function logTransactionPrismaError(error: Prisma.PrismaClientKnownRequestError)
   });
 }
 
+function withTransactionDefaults(
+  client: PrismaClientOrTransaction,
+  options?: PrismaTransactionOptions
+): PrismaTransactionOptions {
+  const resilience = resilienceForClient(client as object);
+  return {
+    maxWait: resilience.maxWait,
+    ...options,
+    startRetry: options?.startRetry ?? resilience.startRetry,
+  };
+}
+
 export async function $transaction<R>(
   prisma: PrismaClientOrTransaction,
   name: string,
@@ -93,35 +112,41 @@ async function $transactionInner<R>(
   options?: PrismaTransactionOptions
 ): Promise<R | undefined> {
   if (typeof fnOrName === "string") {
+    const effectiveOptions = withTransactionDefaults(prisma, options);
     return await startActiveSpan(fnOrName, async (span) => {
       span.setAttribute("$transaction", true);
 
-      if (options?.isolationLevel) {
-        span.setAttribute("isolation_level", options.isolationLevel);
+      if (effectiveOptions.isolationLevel) {
+        span.setAttribute("isolation_level", effectiveOptions.isolationLevel);
       }
 
-      if (options?.timeout) {
-        span.setAttribute("timeout", options.timeout);
+      if (effectiveOptions.timeout) {
+        span.setAttribute("timeout", effectiveOptions.timeout);
       }
 
-      if (options?.maxWait) {
-        span.setAttribute("max_wait", options.maxWait);
+      if (effectiveOptions.maxWait) {
+        span.setAttribute("max_wait", effectiveOptions.maxWait);
       }
 
-      if (options?.swallowPrismaErrors) {
-        span.setAttribute("swallow_prisma_errors", options.swallowPrismaErrors);
+      if (effectiveOptions.swallowPrismaErrors) {
+        span.setAttribute("swallow_prisma_errors", effectiveOptions.swallowPrismaErrors);
       }
 
       const fn = fnOrOptions as (prisma: PrismaTransactionClient, span: Span) => Promise<R>;
 
-      return transac(prisma, (client) => fn(client, span), logTransactionPrismaError, options);
+      return transac(
+        prisma,
+        (client) => fn(client, span),
+        logTransactionPrismaError,
+        effectiveOptions
+      );
     });
   } else {
     return transac(
       prisma,
       fnOrName,
       logTransactionPrismaError,
-      typeof fnOrOptions === "function" ? undefined : fnOrOptions
+      withTransactionDefaults(prisma, typeof fnOrOptions === "function" ? undefined : fnOrOptions)
     );
   }
 }
@@ -180,7 +205,10 @@ function captureInfraErrorsRunOps(client: RunOpsPrismaClient): RunOpsPrismaClien
 }
 
 export const prisma = singleton("prisma", () =>
-  captureInfrastructureErrors(tagDatasource("control-plane-writer", getClient()))
+  registerTransactionResilience(
+    captureInfrastructureErrors(tagDatasource("control-plane-writer", getClient())),
+    controlPlaneTransactionResilience
+  )
 );
 
 export const $replica: PrismaReplicaClient = singleton("replica", () => {
@@ -309,15 +337,18 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
     {
       controlPlane: { writer: prisma, replica: $replica },
       buildNewWriter: (url, clientType) =>
-        captureInfraErrorsRunOps(
-          tagDatasourceRunOps(
-            "run-ops-writer",
-            buildRunOpsWriterClient({
-              url,
-              clientType,
-              useDriverAdapter: env.RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER === "1",
-            })
-          )
+        registerTransactionResilience(
+          captureInfraErrorsRunOps(
+            tagDatasourceRunOps(
+              "run-ops-writer",
+              buildRunOpsWriterClient({
+                url,
+                clientType,
+                useDriverAdapter: env.RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER === "1",
+              })
+            )
+          ),
+          runOpsTransactionResilience
         ),
       // Brand the run-ops replica (only built for a real replica URL) so routed replica reads stay
       // off the primary. When no replica URL is set, selectRunOpsTopology reuses the writer here —
@@ -338,17 +369,20 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
       // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full
       // control-plane schema); markReadReplicaClient only on a real replica URL, as with the NEW replica.
       buildLegacyWriter: (url, clientType) =>
-        captureInfrastructureErrors(
-          tagDatasource(
-            "legacy-run-ops-writer",
-            buildWriterClient({
-              url,
-              clientType,
-              poolTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_POOL_TIMEOUT,
-              connectTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_CONNECTION_TIMEOUT,
-              useDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER === "1",
-            })
-          )
+        registerTransactionResilience(
+          captureInfrastructureErrors(
+            tagDatasource(
+              "legacy-run-ops-writer",
+              buildWriterClient({
+                url,
+                clientType,
+                poolTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_POOL_TIMEOUT,
+                connectTimeout: env.RUN_OPS_LEGACY_DATABASE_WRITER_CONNECTION_TIMEOUT,
+                useDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER === "1",
+              })
+            )
+          ),
+          runOpsLegacyTransactionResilience
         ),
       buildLegacyReplica: (url, clientType) =>
         markReadReplicaClient(

apps/webapp/app/env.server.ts

@@ -114,6 +114,26 @@ const OptionalIntEnv = z.preprocess(
   z.coerce.number().int().optional()
 );
 
+/** Optional boolean env var; blank/whitespace/unset normalises to undefined (so it falls back). */
+const OptionalBoolEnv = z.preprocess((v) => {
+  if (typeof v !== "string" || v.trim() === "") return undefined;
+  return ["true", "1"].includes(v.toLowerCase().trim());
+}, z.boolean().optional());
+
+/** Boolean env var with a default where blank/whitespace falls back to the default instead of parsing as false. */
+const BoolEnvWithDefault = (defaultValue: boolean) =>
+  z.preprocess((v) => {
+    if (typeof v !== "string" || v.trim() === "") return undefined;
+    return ["true", "1"].includes(v.toLowerCase().trim());
+  }, z.boolean().default(defaultValue));
+
+/** Int env var with a default where a blank/whitespace value falls back to the default instead of coercing to 0. */
+const IntEnvWithDefault = (defaultValue: number) =>
+  z.preprocess(
+    (v) => (typeof v === "string" && v.trim() === "" ? undefined : v),
+    z.coerce.number().int().default(defaultValue)
+  );
+
 /**
  * Optional int env var for a limit that can be switched off. Blank, whitespace and `0` all mean
  * "no limit" and normalise to undefined; anything else that is set must be greater than zero.
@@ -142,6 +162,27 @@ const EnvironmentSchema = z
     DATABASE_WRITER_CONNECTION_TIMEOUT: OptionalIntEnv,
     DATABASE_READ_REPLICA_POOL_TIMEOUT: OptionalIntEnv,
     DATABASE_READ_REPLICA_CONNECTION_TIMEOUT: OptionalIntEnv,
+    DATABASE_TRANSACTION_MAX_WAIT_MS: IntEnvWithDefault(10000),
+    DATABASE_TRANSACTION_START_RETRY_ENABLED: BoolEnvWithDefault(true),
+    DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: IntEnvWithDefault(3),
+    DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: IntEnvWithDefault(50),
+    DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: IntEnvWithDefault(250),
+    DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: IntEnvWithDefault(50),
+    DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: IntEnvWithDefault(100),
+    RUN_OPS_DATABASE_TRANSACTION_MAX_WAIT_MS: OptionalIntEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_ENABLED: OptionalBoolEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: OptionalIntEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: OptionalIntEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: OptionalIntEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: OptionalIntEnv,
+    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_MAX_WAIT_MS: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_ENABLED: OptionalBoolEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: OptionalIntEnv,
+    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: OptionalIntEnv,
     // Dashboard-agent conversation store. Cloud points this at a dedicated
     // database; when unset it falls back to DATABASE_URL (OSS), where
     // the tables live in the isolated `trigger_dashboard_agent` schema.

apps/webapp/app/v3/runStore.server.ts

@@ -12,6 +12,10 @@ import {
 } from "~/db.server";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
+import {
+  resilienceForClient,
+  type TransactionResilienceConfig,
+} from "./transactionResilience.server";
 
 type BuildRunStoreDeps = {
   /** Boot constant: true only when both run-ops DBs are configured and the split flag is on. */
@@ -27,6 +31,10 @@ type BuildRunStoreDeps = {
   singleReplica: PrismaReplicaClient;
   /** Residency classifier; defaults to ownerEngine inside RoutingRunStore. */
   classify?: (id: string) => Residency;
+  /** Per-pool transaction-resilience configs threaded into the store(s) this builds (IoC). */
+  singleResilience?: TransactionResilienceConfig;
+  newResilience?: TransactionResilienceConfig;
+  legacyResilience?: TransactionResilienceConfig;
 };
 
 /**
@@ -46,6 +54,8 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
     return new PostgresRunStore({
       prisma: deps.singleWriter,
       readOnlyPrisma: deps.singleReplica,
+      maxWait: deps.singleResilience?.maxWait,
+      transactionStartRetry: deps.singleResilience?.startRetry,
     });
   }
 
@@ -59,10 +69,14 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
     prisma: deps.newWriter,
     readOnlyPrisma: deps.newReplica,
     schemaVariant: "dedicated",
+    maxWait: deps.newResilience?.maxWait,
+    transactionStartRetry: deps.newResilience?.startRetry,
   });
   const legacyStore = new PostgresRunStore({
     prisma: deps.legacyWriter,
     readOnlyPrisma: deps.legacyReplica,
+    maxWait: deps.legacyResilience?.maxWait,
+    transactionStartRetry: deps.legacyResilience?.startRetry,
   });
 
   return new RoutingRunStore({
@@ -110,12 +124,16 @@ export const runStore: RunStore = singleton("RunStore", () => {
       splitEnabled: false,
       singleWriter: prisma,
       singleReplica: $replica,
+      singleResilience: resilienceForClient(prisma),
     });
   }
   return buildRunStore({
     splitEnabled: true,
     ...handles,
     singleWriter: prisma,
     singleReplica: $replica,
+    singleResilience: resilienceForClient(prisma),
+    newResilience: resilienceForClient(handles.newWriter),
+    legacyResilience: resilienceForClient(handles.legacyWriter),
   });
 });

apps/webapp/app/v3/transactionResilience.server.ts

@@ -0,0 +1,100 @@
+import { TokenBucketRetryBudget, type TransactionStartRetryConfig } from "@trigger.dev/database";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+
+/**
+ * Resolved transaction-resilience config for one writer pool. Each pool gets its own
+ * {@link TransactionStartRetryConfig} (with its OWN token bucket, so a storm on one pool cannot
+ * drain another's retry budget) plus the `maxWait` applied when that pool opens a transaction.
+ * Env is read here at the app boundary (IoC); the library never reads env.
+ *
+ * Kept out of `db.server` on purpose: `db.server` is mocked wholesale by ~150 tests, and a new
+ * export there breaks every mock that does not list it. Both `db.server` and `runStore.server`
+ * import these from here instead.
+ */
+export type TransactionResilienceConfig = {
+  maxWait: number;
+  startRetry: TransactionStartRetryConfig;
+};
+
+function resolveTransactionResilience(
+  pool: "control-plane" | "run-ops" | "run-ops-legacy",
+  overrides: {
+    maxWaitMs?: number;
+    enabled?: boolean;
+    maxAttempts?: number;
+    backoffMinMs?: number;
+    backoffMaxMs?: number;
+    budgetPerSec?: number;
+    budgetBurst?: number;
+  }
+): TransactionResilienceConfig {
+  const budgetPerSec =
+    overrides.budgetPerSec ?? env.DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC;
+  const budgetBurst = overrides.budgetBurst ?? env.DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST;
+  return {
+    maxWait: Math.max(0, overrides.maxWaitMs ?? env.DATABASE_TRANSACTION_MAX_WAIT_MS),
+    startRetry: {
+      options: {
+        enabled: overrides.enabled ?? env.DATABASE_TRANSACTION_START_RETRY_ENABLED,
+        maxAttempts: overrides.maxAttempts ?? env.DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS,
+        backoffMinMs: overrides.backoffMinMs ?? env.DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS,
+        backoffMaxMs: overrides.backoffMaxMs ?? env.DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS,
+      },
+      budget: new TokenBucketRetryBudget({ ratePerSec: budgetPerSec, burst: budgetBurst }),
+      onRetry: ({ attempt, delayMs }) =>
+        logger.warn("retrying transaction start after acquisition failure", {
+          pool,
+          attempt,
+          delayMs,
+        }),
+    },
+  };
+}
+
+export const controlPlaneTransactionResilience = resolveTransactionResilience("control-plane", {});
+
+export const runOpsTransactionResilience = resolveTransactionResilience("run-ops", {
+  maxWaitMs: env.RUN_OPS_DATABASE_TRANSACTION_MAX_WAIT_MS,
+  enabled: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_ENABLED,
+  maxAttempts: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS,
+  backoffMinMs: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS,
+  backoffMaxMs: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS,
+  budgetPerSec: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC,
+  budgetBurst: env.RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST,
+});
+
+export const runOpsLegacyTransactionResilience = resolveTransactionResilience("run-ops-legacy", {
+  maxWaitMs: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_MAX_WAIT_MS,
+  enabled: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_ENABLED,
+  maxAttempts: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS,
+  backoffMinMs: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS,
+  backoffMaxMs: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS,
+  budgetPerSec: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC,
+  budgetBurst: env.RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST,
+});
+
+const transactionResilienceByClient = new WeakMap<object, TransactionResilienceConfig>();
+
+/**
+ * Associate a writer client with its pool's resilience config. Returns the client for inline use at
+ * construction. Kept here (not in db.server) so nothing new lands on db.server's wholesale-mocked
+ * export surface.
+ */
+export function registerTransactionResilience<T extends object>(
+  client: T,
+  resilience: TransactionResilienceConfig
+): T {
+  transactionResilienceByClient.set(client, resilience);
+  return client;
+}
+
+/**
+ * The resilience config registered for a writer client, or the control-plane config as a safe
+ * fallback. Derives resilience from the ACTUAL client identity rather than an assumed routing role,
+ * so run-ops clients aliased onto the control-plane pool (split flag off) correctly get the
+ * control-plane config instead of a run-ops override.
+ */
+export function resilienceForClient(client: object): TransactionResilienceConfig {
+  return transactionResilienceByClient.get(client) ?? controlPlaneTransactionResilience;
+}