fix(run-store): stop run-create failing on a brief write stall (#4514)

## Summary On the run-ops store, creating a run could intermittently fail with a "Transaction already closed" error, and the run would never be created. Single-write run creates no longer run inside an interactive transaction, so a brief database write stall can't blow the transaction budget and drop the run. ## Fix The dedicated run-ops `createRun` / `createFailedRun` wrapped a single nested `taskRun.create` in an interactive `$transaction`. Its default 5s budget is wall-clock from `BEGIN`, so when a write briefly stalls the transaction expires before the create completes and throws, even though the statement itself is fast at the database. A single-write create does not need an interactive transaction: Prisma's implicit nested create is already atomic and holds no app-side budget, so it now runs directly. Only the `triggerAndWait` path (run plus its associated waitpoint, two writes that must commit together) keeps an interactive transaction, now with headroom over the default. Verified with a red/green test against the real split topology (reproduces the exact expiry on the unchanged code, green after) and an end-to-end run created and completed through the dedicated store.
2026-08-05 17:35:46 +01:00
parent 58bf4e2833
commit b20806247f
3 changed files with 156 additions and 17 deletions
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Triggering a task no longer intermittently fails to create the run when a database write briefly stalls.
@@ -0,0 +1,96 @@
+import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
+import type { RunOpsPrismaClient } from "@internal/run-ops-database";
+import { describe, expect } from "vitest";
+import { PostgresRunStore } from "./PostgresRunStore.js";
+import type { CreateRunInput } from "./types.js";
+
+const NEW_ID_26 = "k".repeat(24) + "01";
+
+function makeDedicatedStore(prisma17: RunOpsPrismaClient) {
+  return new PostgresRunStore({
+    prisma: prisma17 as never,
+    readOnlyPrisma: prisma17 as never,
+    schemaVariant: "dedicated",
+  });
+}
+
+function trackInteractiveTx(prisma17: RunOpsPrismaClient) {
+  const original = prisma17.$transaction.bind(prisma17);
+  const state = { interactiveCalls: 0 };
+  (prisma17 as { $transaction: unknown }).$transaction = (
+    arg: unknown,
+    options?: { timeout?: number; maxWait?: number }
+  ) => {
+    if (typeof arg === "function") {
+      state.interactiveCalls += 1;
+      return (original as (fn: unknown, o?: unknown) => unknown)(arg, { ...options, timeout: 1 });
+    }
+    return (original as (a: unknown, o?: unknown) => unknown)(arg, options);
+  };
+  return state;
+}
+
+function buildCreateRunInput(params: {
+  runId: string;
+  friendlyId: string;
+  suffix: string;
+}): CreateRunInput {
+  return {
+    data: {
+      id: params.runId,
+      engine: "V2",
+      status: "PENDING",
+      friendlyId: params.friendlyId,
+      runtimeEnvironmentId: `env_${params.suffix}`,
+      environmentType: "DEVELOPMENT",
+      organizationId: `org_${params.suffix}`,
+      projectId: `proj_${params.suffix}`,
+      taskIdentifier: "my-task",
+      payload: '{"hello":"world"}',
+      payloadType: "application/json",
+      traceContext: { trace: "ctx" },
+      traceId: `trace_${params.runId}`,
+      spanId: `span_${params.runId}`,
+      runTags: [],
+      queue: "task/my-task",
+      isTest: false,
+      taskEventStore: "taskEvent",
+      depth: 0,
+      createdAt: new Date("2024-01-01T00:00:00.000Z"),
+    },
+    snapshot: {
+      engine: "V2",
+      executionStatus: "RUN_CREATED",
+      description: "Run was created",
+      runStatus: "PENDING",
+      environmentId: `env_${params.suffix}`,
+      environmentType: "DEVELOPMENT",
+      projectId: `proj_${params.suffix}`,
+      organizationId: `org_${params.suffix}`,
+    },
+  };
+}
+
+describe("createRun on the dedicated store does not wrap a single-write create in an interactive transaction", () => {
+  heteroRunOpsPostgresTest(
+    "a create with no associated waitpoint survives an interactive-tx budget of 1ms (run + snapshot persist)",
+    async ({ prisma17 }) => {
+      const tx = trackInteractiveTx(prisma17);
+      const store = makeDedicatedStore(prisma17);
+      const runId = `run_${NEW_ID_26}`;
+
+      await store.createRun(
+        buildCreateRunInput({ runId, friendlyId: "run_no_tx", suffix: "no_tx" })
+      );
+
+      expect(tx.interactiveCalls).toBe(0);
+
+      const run = await prisma17.taskRun.findFirstOrThrow({ where: { id: runId } });
+      expect(run.status).toBe("PENDING");
+      const snap = await prisma17.taskRunExecutionSnapshot.findFirst({
+        where: { runId, executionStatus: "RUN_CREATED" },
+      });
+      expect(snap).not.toBeNull();
+    }
+  );
+});
@@ -84,7 +84,10 @@ export interface RunOpsCapableClient {
 * per-call `tx` so they share one transaction (see `runInTransaction`).
 */
 export interface RunOpsTransactionalClient extends RunOpsCapableClient {
-  $transaction: <R>(fn: (tx: RunOpsCapableClient) => Promise<R>) => Promise<R>;
+  $transaction: <R>(
+    fn: (tx: RunOpsCapableClient) => Promise<R>,
+    options?: { timeout?: number; maxWait?: number; isolationLevel?: unknown }
+  ) => Promise<R>;
 }

 /**
@@ -99,6 +102,8 @@ export type RunStoreSchemaVariant = "legacy" | "dedicated";
 // (apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts) — keep the values in sync.
 export const CONNECTED_RUNS_LIMIT = 5;

+export const RUN_OPS_WRITE_TX_TIMEOUT_MS = 15_000;
+
 export type PostgresRunStoreOptions = {
  prisma: RunOpsCapableClient;
  readOnlyPrisma: RunOpsCapableClient;
@@ -661,18 +666,28 @@ export class PostgresRunStore implements RunStore {
  // (snapshot + completed-waitpoints, run + associated-waitpoint) which must commit together.
  #withOptionalTransaction<R>(
    tx: PrismaClientOrTransaction | undefined,
-    fn: (client: PrismaClientOrTransaction) => Promise<R>
+    fn: (client: PrismaClientOrTransaction) => Promise<R>,
+    options?: { timeout?: number; maxWait?: number }
  ): Promise<R> {
    const alreadyInTransaction =
      tx !== undefined && typeof (tx as { $transaction?: unknown }).$transaction !== "function";
    if (alreadyInTransaction) {
      return fn(tx);
    }
-    return (this.prisma as RunOpsTransactionalClient).$transaction((t) =>
-      fn(t as unknown as PrismaClientOrTransaction)
+    return (this.prisma as RunOpsTransactionalClient).$transaction(
+      (t) => fn(t as unknown as PrismaClientOrTransaction),
+      options
    );
  }

+  #writeClientWithoutTransaction(
+    tx: PrismaClientOrTransaction | undefined
+  ): PrismaClientOrTransaction {
+    const alreadyInTransaction =
+      tx !== undefined && typeof (tx as { $transaction?: unknown }).$transaction !== "function";
+    return (alreadyInTransaction ? tx : this.prisma) as PrismaClientOrTransaction;
+  }
+
  async createRun(
    params: CreateRunInput,
    tx?: PrismaClientOrTransaction
@@ -694,19 +709,31 @@ export class PostgresRunStore implements RunStore {
    };

    if (this.schemaVariant === "dedicated") {
-      // The run + its associated RUN-type waitpoint are two writes here (the legacy branch below nests
-      // them). Commit them together so a crash / lagging read never leaves a run without its waitpoint.
-      return this.#withOptionalTransaction(tx, async (c) => {
-        const run = (await c.taskRun.create({
+      if (!params.associatedWaitpoint) {
+        const run = (await this.#writeClientWithoutTransaction(tx).taskRun.create({
          data: {
            ...params.data,
            executionSnapshots: { create: snapshotCreate },
          },
        })) as TaskRun;
+        return { ...run, associatedWaitpoint: null };
+      }

-        const associatedWaitpoint = await this.#createAssociatedWaitpoint(c, run.id, params);
-        return { ...run, associatedWaitpoint };
-      });
+      return this.#withOptionalTransaction(
+        tx,
+        async (c) => {
+          const run = (await c.taskRun.create({
+            data: {
+              ...params.data,
+              executionSnapshots: { create: snapshotCreate },
+            },
+          })) as TaskRun;
+
+          const associatedWaitpoint = await this.#createAssociatedWaitpoint(c, run.id, params);
+          return { ...run, associatedWaitpoint };
+        },
+        { timeout: RUN_OPS_WRITE_TX_TIMEOUT_MS }
+      );
    }

    return client.taskRun.create({
@@ -784,15 +811,25 @@ export class PostgresRunStore implements RunStore {
    const client = tx ?? this.prisma;

    if (this.schemaVariant === "dedicated") {
-      // Run + associated RUN-type waitpoint are two writes here; commit them together (see createRun).
-      return this.#withOptionalTransaction(tx, async (c) => {
-        const run = (await c.taskRun.create({
+      if (!params.associatedWaitpoint) {
+        const run = (await this.#writeClientWithoutTransaction(tx).taskRun.create({
          data: { ...params.data },
        })) as TaskRun;
+        return { ...run, associatedWaitpoint: null };
+      }

-        const associatedWaitpoint = await this.#createAssociatedWaitpoint(c, run.id, params);
-        return { ...run, associatedWaitpoint };
-      });
+      return this.#withOptionalTransaction(
+        tx,
+        async (c) => {
+          const run = (await c.taskRun.create({
+            data: { ...params.data },
+          })) as TaskRun;
+
+          const associatedWaitpoint = await this.#createAssociatedWaitpoint(c, run.id, params);
+          return { ...run, associatedWaitpoint };
+        },
+        { timeout: RUN_OPS_WRITE_TX_TIMEOUT_MS }
+      );
    }

    return client.taskRun.create({