diff --git a/.server-changes/run-create-no-interactive-tx.md b/.server-changes/run-create-no-interactive-tx.md new file mode 100644 index 00000000000..a8194c2f5bb --- /dev/null +++ b/.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. diff --git a/internal-packages/run-store/src/PostgresRunStore.createRunNoTx.test.ts b/internal-packages/run-store/src/PostgresRunStore.createRunNoTx.test.ts new file mode 100644 index 00000000000..5586721f17e --- /dev/null +++ b/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(); + } + ); +}); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 676e345ad5a..6a0cef55c44 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/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: (fn: (tx: RunOpsCapableClient) => Promise) => Promise; + $transaction: ( + fn: (tx: RunOpsCapableClient) => Promise, + options?: { timeout?: number; maxWait?: number; isolationLevel?: unknown } + ) => Promise; } /** @@ -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( tx: PrismaClientOrTransaction | undefined, - fn: (client: PrismaClientOrTransaction) => Promise + fn: (client: PrismaClientOrTransaction) => Promise, + options?: { timeout?: number; maxWait?: number } ): Promise { 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({