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

On the run-ops store, a run create with no associated waitpoint was wrapped in an interactive transaction whose default 5s budget is wall-clock from BEGIN, so a brief write stall could expire it and throw before the create ran, losing the run. Single-write creates now run as an implicit transaction with no app-held budget; the triggerAndWait path (run plus associated waitpoint) keeps an interactive transaction with headroom.

Author: ericallam

.server-changes/run-create-no-interactive-tx.md

@@ -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.

internal-packages/run-store/src/PostgresRunStore.createRunNoTx.test.ts

@@ -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();
+    }
+  );
+});

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

@@ -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({