From 6fe7952062f250cd96b556c74101531dfa330880 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 20:53:40 +0100 Subject: [PATCH 01/16] fix(run-engine): route the block-time pending waitpoint check to the primary blockRunWithWaitpoint confirms whether a run is still blocked with a separate countPendingWaitpoints query. Under the run-ops split that read passed no client, so it resolved to the owning store's read replica. When a waitpoint completes on the primary just before the run blocks on it (the wait, token, or batchTriggerAndWait-child race), a lagging replica still reports it PENDING, the run is marked blocked, and no continue job is enqueued, so the run never resumes. Pass the writer so the pending re-read is read-your-writes on the owning primary. Adds a two-database engine test that reproduces the strand and passes with the fix. --- .../src/engine/systems/waitpointSystem.ts | 5 +- .../engine/tests/waitpointReplicaLag.test.ts | 283 ++++++++++++++++++ 2 files changed, 286 insertions(+), 2 deletions(-) create mode 100644 internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 344a0c1d8c..6969363d1a 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -511,8 +511,9 @@ export class WaitpointSystem { batchIndex: batch?.index, }); - // Check if the run is actually blocked using a separate query (see above). - const pendingCount = await this.$.runStore.countPendingWaitpoints($waitpoints); + // Check if the run is actually blocked using a separate query (see above). Pass the writer so the + // pending re-read is read-your-writes on the owning PRIMARY (a lagging replica can strand the run). + const pendingCount = await this.$.runStore.countPendingWaitpoints($waitpoints, prisma); const isRunBlocked = pendingCount > 0; diff --git a/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts b/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts new file mode 100644 index 0000000000..dac6ffeca4 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts @@ -0,0 +1,283 @@ +// blockRunWithWaitpoint's block-time pending check is UNROUTED — countPendingWaitpoints($waitpoints) +// with no client falls to the owning store's REPLICA. When a waitpoint completes on the PRIMARY just +// before the parent blocks (the wait/token/batch-child race), a lagging replica still reports it +// PENDING → isRunBlocked=true → no continueRunIfUnblocked is enqueued → the run is stranded in +// EXECUTING_WITH_WAITPOINTS forever (the production hang under the run-ops split). +// GREEN fix: countPendingWaitpoints($waitpoints, prisma) — routed to the owning primary via #ownPrimary. +// Real two-DB topology (#new=prisma17 dedicated, #legacy=prisma14); replica lag simulated by a +// recording proxy, as in runOpsStore.readAfterWrite.test.ts. + +import { + heteroRunOpsPostgresTest, + network, + redisContainer, + redisOptions, +} from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { PostgresRunStore, RoutingRunStore, type CreateRunInput } from "@internal/run-store"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect, vi } from "vitest"; +import { RunEngine } from "../index.js"; + +const twoDbEngineTest = heteroRunOpsPostgresTest.extend<{ + redisContainer: any; + redisOptions: any; +}>({ + network, + redisContainer, + redisOptions, +}); + +// run-ops id (version "1" at index 25) → classified NEW → routed to the run-ops (#new) store. +const RUN_OPS_A = "n".repeat(24) + "01"; + +function baseEngineOptions(redisOptions: any, prisma: any) { + return { + prisma, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { redis: redisOptions }, + machines: { + defaultMachine: "small-1x" as const, + machines: { + "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }; +} + +async function seedControlPlaneEnv(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: `prod-${suffix}`, + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + maximumConcurrencyLimit: 10, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(params: { + runId: string; + friendlyId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}): CreateRunInput { + return { + data: { + id: params.runId, + engine: "V2", + status: "EXECUTING", + friendlyId: params.friendlyId, + runtimeEnvironmentId: params.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: params.organizationId, + projectId: params.projectId, + taskIdentifier: "parent-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `trace_${params.runId}`, + spanId: `span_${params.runId}`, + runTags: [], + queue: "task/parent-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: params.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: params.projectId, + organizationId: params.organizationId, + }, + }; +} + +// Seed an EXECUTING run-ops parent on #new (prisma17) via the routed store, plus a run-ops id PENDING +// RUN waitpoint co-resident on #new. +async function seedExecutingRunOpsParent( + prisma14: PrismaClient, + prisma17: RunOpsPrismaClient, + router: RoutingRunStore, + parentRunId: string, + waitpointId: string, + suffix: string +) { + const env = await seedControlPlaneEnv(prisma14, suffix); + + await router.createRun( + buildCreateRunInput({ + runId: parentRunId, + friendlyId: `run_${suffix}_parent`, + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + + const created = await router.findLatestExecutionSnapshot(parentRunId); + await router.createExecutionSnapshot( + { + run: { id: parentRunId, status: "EXECUTING", attemptNumber: 1 }, + snapshot: { executionStatus: "EXECUTING", description: "parent executing" }, + previousSnapshotId: created!.id, + environmentId: env.environment.id, + environmentType: "PRODUCTION", + projectId: env.project.id, + organizationId: env.organization.id, + }, + prisma14 + ); + + await prisma17.waitpoint.create({ + data: { + id: waitpointId, + friendlyId: `wp_${suffix}`, + type: "RUN", + status: "PENDING", + completedByTaskRunId: parentRunId, + idempotencyKey: `idem_${waitpointId}`, + userProvidedIdempotencyKey: false, + projectId: env.project.id, + environmentId: env.environment.id, + }, + }); + + return env; +} + +// A lagging NEW replica that has NOT yet applied the waitpoint completion: its pending-count query +// (the only $queryRaw countPendingWaitpoints issues) reports every queried waitpoint as still PENDING. +// Everything else forwards to the real client. `wasHit` flips true iff the pending-count query ran here. +function laggingPendingReplica( + real: C +): { client: C; wasHit: () => boolean } { + let hit = false; + const client = new Proxy(real as any, { + get(target, prop) { + if (prop === "$queryRaw") { + return (strings: TemplateStringsArray, ...values: any[]) => { + const sql = Array.isArray(strings) ? strings.join(" ") : String(strings); + if (sql.includes("pending_count")) { + hit = true; + const ids = values[0]; + const stalePending = Array.isArray(ids) ? ids.length : 1; + // stale replica: the just-completed waitpoint(s) still look PENDING + return Promise.resolve([{ pending_count: BigInt(stalePending) }]); + } + return (target as any).$queryRaw(strings, ...values); + }; + } + return (target as any)[prop]; + }, + }) as C; + return { client, wasHit: () => hit }; +} + +describe("RunEngine blockRunWithWaitpoint — block-time pending check must read the owning primary, not a lagging replica", () => { + twoDbEngineTest( + "a waitpoint completed on the primary just before block does not strand the run (continueRunIfUnblocked is enqueued)", + async ({ prisma14, prisma17, redisOptions }) => { + // #new reads go to a LAGGING replica whose pending-count still reports PENDING. + const newReplica = laggingPendingReplica(prisma17); + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: newReplica.client as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14 as unknown as PrismaClient, + readOnlyPrisma: prisma14 as unknown as PrismaClient, + schemaVariant: "legacy", + }); + const router = new RoutingRunStore({ new: newStore, legacy: legacyStore }); + + const engine = new RunEngine({ + store: router, + ...baseEngineOptions(redisOptions, prisma14), + }); + + try { + const parentRunId = `run_${RUN_OPS_A}`; + const waitpointId = `waitpoint_${RUN_OPS_A}`; + const env = await seedExecutingRunOpsParent( + prisma14 as unknown as PrismaClient, + prisma17, + router, + parentRunId, + waitpointId, + "replag" + ); + + // THE RACE: the waitpoint completes on the PRIMARY just before the parent blocks on it. + await prisma17.waitpoint.update({ + where: { id: waitpointId }, + data: { status: "COMPLETED", completedAt: new Date() }, + }); + + // The primary shows COMPLETED (0 pending); the lagging replica will still report PENDING. + expect((await prisma17.waitpoint.findFirst({ where: { id: waitpointId } }))?.status).toBe( + "COMPLETED" + ); + + const enqueueSpy = vi.spyOn((engine as any).worker, "enqueue"); + + // Block the parent on the already-completed waitpoint. The block-time pending check should see + // 0 pending and enqueue continueRunIfUnblocked. Under the unrouted replica read it sees stale + // PENDING and enqueues nothing → the run hangs. + await engine.blockRunWithWaitpoint({ + runId: parentRunId, + waitpoints: waitpointId, + projectId: env.project.id, + organizationId: env.organization.id, + tx: prisma14 as unknown as PrismaClient, + }); + + const continueEnqueued = enqueueSpy.mock.calls.some( + ([arg]) => + (arg as any)?.job === "continueRunIfUnblocked" && + (arg as any)?.payload?.runId === parentRunId + ); + // RED: not enqueued (block-time check read the lagging replica → stale PENDING → stranded). + // GREEN: enqueued (block-time check routed to the owning primary → sees COMPLETED → 0 pending). + expect(continueEnqueued).toBe(true); + + // And the fix means the pending check no longer touches the lagging replica at all. + expect(newReplica.wasHit()).toBe(false); + } finally { + await engine.quit(); + } + } + ); +}); From d79bdf0bbc4e0427b415d2a8a1f409314c2a799d Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:20:05 +0100 Subject: [PATCH 02/16] fix(run-store): route run-ops snapshot and batch-item lookups to the owning store Two run-ops split reads routed by an id that does not encode residency, so NEW-residency data on the new store was queried on the legacy store and came back empty: - findSnapshotCompletedWaitpointIds routed by the snapshot id, which is a cuid (always classifies legacy). A resuming NEW run saw zero completed waitpoints and hung. It now fans out across both stores and merges. - updateManyBatchTaskRunItems routed by the item id, also a cuid, so a NEW batch's items were updated on the wrong store and matched zero rows, leaving the batch stuck and its batchTriggerAndWait parent hanging. It now routes by the batch id. Both covered by two-database store tests that reproduce the miss. --- .../src/runOpsStore.batchItemMisroute.test.ts | 93 +++++++++++++++++++ ...sStore.snapshotCompletedWaitpoints.test.ts | 47 ++++++++++ .../run-store/src/runOpsStore.ts | 28 ++++-- 3 files changed, 159 insertions(+), 9 deletions(-) create mode 100644 internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts create mode 100644 internal-packages/run-store/src/runOpsStore.snapshotCompletedWaitpoints.test.ts diff --git a/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts new file mode 100644 index 0000000000..edacf01f9b --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts @@ -0,0 +1,93 @@ +// updateManyBatchTaskRunItems routed by where.id first, but BatchTaskRunItem.id is a cuid (always +// classifies LEGACY). For a NEW-residency batch the items live on #new, so completing an item by +// {id, batchTaskRunId, status} updated #legacy, matched 0 rows, and the caller treated count===0 as +// "already completed" -> tryCompleteBatchV3 never fired -> the item stayed PENDING and the parent's +// batchTriggerAndWait hung. Fix: route by batchTaskRunId (residency-encoding) first, like +// countBatchTaskRunItems. Real two-DB topology; never mocked. + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { PostgresRunStore } from "./PostgresRunStore.js"; +import { RoutingRunStore } from "./runOpsStore.js"; + +const NEW_ID_26 = "k".repeat(24) + "01"; // run-ops id -> NEW (#new / prisma17) + +function makeSplitRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + return new RoutingRunStore({ new: newStore, legacy: legacyStore }); +} + +// BatchTaskRunItem.taskRunId FKs into TaskRun on the dedicated schema, so seed the run first. +async function seedDedicatedRun(prisma17: RunOpsPrismaClient, envId: string, runId: string) { + await prisma17.taskRun.create({ + data: { + id: runId, + engine: "V2", + status: "PENDING", + friendlyId: `run_${runId}`, + runtimeEnvironmentId: envId, + environmentType: "DEVELOPMENT", + organizationId: "org_batchitem", + projectId: "proj_batchitem", + taskIdentifier: "batch-task", + payload: "{}", + payloadType: "application/json", + traceContext: {}, + traceId: `t_${runId}`, + spanId: `s_${runId}`, + queue: "task/batch-task", + isTest: false, + taskEventStore: "taskEvent", + depth: 0, + }, + }); +} + +describe("run-ops split — completing a batch item routes by the batch id, not the item cuid", () => { + heteroRunOpsPostgresTest( + "updateManyBatchTaskRunItems completes a NEW-resident item addressed by {id, batchTaskRunId}", + async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => { + const router = makeSplitRouter(prisma14, prisma17); + const envId = "env_batchitem"; + const batchId = `batch_${NEW_ID_26}`; // run-ops id -> #new + const runId = `run_${NEW_ID_26.slice(0, -3)}ri1`; + + await prisma17.batchTaskRun.create({ + data: { + id: batchId, + friendlyId: "batch_item_new", + runtimeEnvironmentId: envId, + runCount: 1, + status: "PROCESSING", + }, + }); + await seedDedicatedRun(prisma17, envId, runId); + // item.id is an auto cuid (classifies LEGACY) — the mis-routing key, exactly as in production. + const item = await prisma17.batchTaskRunItem.create({ + data: { batchTaskRunId: batchId, taskRunId: runId, status: "PENDING" }, + }); + + const result = await router.updateManyBatchTaskRunItems({ + where: { id: item.id, batchTaskRunId: batchId, status: "PENDING" }, + data: { status: "COMPLETED" }, + }); + + // RED: routed by the item cuid -> #legacy -> 0 rows -> batch never completes -> parent hangs. + // GREEN: routed by batchTaskRunId (NEW) -> #new -> the item completes. + expect(result.count).toBe(1); + const onNew = await prisma17.batchTaskRunItem.findUnique({ where: { id: item.id } }); + expect(onNew?.status).toBe("COMPLETED"); + } + ); +}); diff --git a/internal-packages/run-store/src/runOpsStore.snapshotCompletedWaitpoints.test.ts b/internal-packages/run-store/src/runOpsStore.snapshotCompletedWaitpoints.test.ts new file mode 100644 index 0000000000..b0a5a99e2f --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.snapshotCompletedWaitpoints.test.ts @@ -0,0 +1,47 @@ +// Managed resume reads a run's completed waitpoints by SNAPSHOT id +// (getExecutionSnapshotsSince -> getSnapshotWaitpointIds -> findSnapshotCompletedWaitpointIds). +// Snapshot ids are @default(cuid()), so #routeOrNew(snapshotId) always classifies LEGACY. For a +// NEW-residency run the snapshot's CompletedWaitpoint join rows live on #new, so routing to #legacy +// returns [] and the resumed run sees zero completed waitpoints and hangs. The fix fans out across +// both stores and merges, like findWaitpointCompletedSnapshotIds. Real two-DB topology; never mocked. + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { PostgresRunStore } from "./PostgresRunStore.js"; +import { RoutingRunStore } from "./runOpsStore.js"; + +// cuid-shaped snapshot id -> classifies LEGACY (the always-wrong routing key). +const SNAPSHOT_CUID = "c".repeat(25); +const WAITPOINT_ID = "waitpoint_" + "n".repeat(20); + +describe("run-ops split — completed waitpoints for a cuid snapshot are found on the owning store, not misrouted to legacy", () => { + heteroRunOpsPostgresTest( + "findSnapshotCompletedWaitpointIds finds a NEW-resident join despite the cuid snapshot id", + async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + const router = new RoutingRunStore({ new: newStore, legacy: legacyStore }); + + // A NEW-residency run's completed-waitpoint join lives on #new; its snapshot id is a cuid. + await prisma17.completedWaitpoint.create({ + data: { snapshotId: SNAPSHOT_CUID, waitpointId: WAITPOINT_ID }, + }); + + const ids = await router.findSnapshotCompletedWaitpointIds(SNAPSHOT_CUID); + + // RED: the cuid snapshot id routes to #legacy -> [] -> resumed run never completes its waitpoints. + // GREEN: fan-out finds the #new-resident join. + expect(ids).toEqual([WAITPOINT_ID]); + } + ); +}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 9b7f453837..9e21e1b380 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -830,13 +830,21 @@ export class RoutingRunStore implements RunStore { return (await this.#routeOrNewForWrite(input.run.id)).createExecutionSnapshot(input); } - // A snapshot lives with its run; route by the snapshot id's residency. - findSnapshotCompletedWaitpointIds(snapshotId: string, client?: ReadClient): Promise { - const store = this.#routeOrNew(snapshotId); - return store.findSnapshotCompletedWaitpointIds( - snapshotId, - RoutingRunStore.#ownPrimary(store, client) - ); + // Snapshot ids are cuids (they always classify LEGACY), and a snapshot's CompletedWaitpoint join + // co-locates with its run, which may live on either store, so fan out to BOTH and merge (like + // findWaitpointCompletedSnapshotIds) rather than route by the un-classifiable snapshot id. + async findSnapshotCompletedWaitpointIds(snapshotId: string, client?: ReadClient): Promise { + const [fromNew, fromLegacy] = await Promise.all([ + this.#new.findSnapshotCompletedWaitpointIds( + snapshotId, + RoutingRunStore.#ownPrimary(this.#new, client) + ), + this.#legacy.findSnapshotCompletedWaitpointIds( + snapshotId, + RoutingRunStore.#ownPrimary(this.#legacy, client) + ), + ]); + return uniqueStrings([...fromNew, ...fromLegacy]); } // Keyed by waitpointId, but the WaitpointRunConnection / CompletedWaitpoint join co-locates with the @@ -1484,9 +1492,11 @@ export class RoutingRunStore implements RunStore { args: Prisma.BatchTaskRunItemUpdateManyArgs, tx?: PrismaClientOrTransaction ): Promise { + // Items co-reside with their batch; route by batchTaskRunId (residency-encoding) first. The item + // id is a cuid that always classifies LEGACY, so leading with it misroutes a NEW batch's items. const id = - RoutingRunStore.#scalarId(args.where) ?? - RoutingRunStore.#scalarField(args.where, "batchTaskRunId"); + RoutingRunStore.#scalarField(args.where, "batchTaskRunId") ?? + RoutingRunStore.#scalarId(args.where); if (id !== undefined) { const store = this.#routeOrNew(id); return store.updateManyBatchTaskRunItems(args, store === this.#legacy ? tx : undefined); From 38daf4929d6207de90cf64a012c65d6977e9f111 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:20:05 +0100 Subject: [PATCH 03/16] fix(run-engine): read the run row on the owning primary in startRunAttempt The attempt-start lock check read the run with no client, so under the split it hit the owning store's replica. Dequeue had just written lockedById on the primary, so a lagging replica reported the run unlocked and rejected the start with "Task run is not locked". It now threads the writer so the read is read-your-writes on the owning primary, matching the sibling snapshot read. Covered by a two-database engine test. --- .../src/engine/systems/runAttemptSystem.ts | 4 +- .../startRunAttemptReadResidency.test.ts | 223 ++++++++++++++++++ 2 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts diff --git a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts index a384d37f83..7f53c7648d 100644 --- a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts @@ -347,7 +347,9 @@ export class RunAttemptSystem { lockedById: true, ttl: true, }, - } + }, + // read-your-writes on the owning primary (dequeue just wrote lockedById; a replica lags). + prisma ); this.$.logger.debug("Creating a task run attempt", { taskRun }); diff --git a/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts b/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts new file mode 100644 index 0000000000..4896201514 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts @@ -0,0 +1,223 @@ +// startRunAttempt gates on the run row: `if (!taskRun.lockedById) throw "Task run is not locked"`. +// Dequeue (lockRunToWorker) writes lockedById on the owning PRIMARY, then the worker issues a +// SEPARATE start-attempt request. That run-row read (findRun, waitpointless) passes no client, so +// under the split it resolves to the owning store's REPLICA. A lagging replica still shows the +// pre-lock row (lockedById null), so the just-dequeued run is rejected with a spurious 400 and never +// starts. The fix threads the writer so the read is read-your-writes on the owning primary (matching +// the sibling getLatestExecutionSnapshot call). Real two-DB topology; replica lag simulated by a proxy. + +import { + heteroRunOpsPostgresTest, + network, + redisContainer, + redisOptions, +} from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { PostgresRunStore, RoutingRunStore, type CreateRunInput } from "@internal/run-store"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; + +const twoDbEngineTest = heteroRunOpsPostgresTest.extend<{ + redisContainer: any; + redisOptions: any; +}>({ + network, + redisContainer, + redisOptions, +}); + +const RUN_OPS_A = "n".repeat(24) + "01"; // run-ops id -> classified NEW -> #new + +function baseEngineOptions(redisOptions: any, prisma: any) { + return { + prisma, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { redis: redisOptions }, + machines: { + defaultMachine: "small-1x" as const, + machines: { + "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }; +} + +async function seedControlPlaneEnv(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: `prod-${suffix}`, + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + maximumConcurrencyLimit: 10, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(params: { + runId: string; + friendlyId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}): CreateRunInput { + return { + data: { + id: params.runId, + engine: "V2", + status: "PENDING", + friendlyId: params.friendlyId, + runtimeEnvironmentId: params.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: params.organizationId, + projectId: params.projectId, + taskIdentifier: "attempt-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `trace_${params.runId}`, + spanId: `span_${params.runId}`, + runTags: [], + queue: "task/attempt-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: params.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: params.projectId, + organizationId: params.organizationId, + }, + }; +} + +// A lagging replica that has not applied the dequeue lock: taskRun reads still show lockedById null. +// Everything else forwards to the real client. `wasHit` flips true iff a taskRun read ran here. +function laggingLockReplica( + real: C +): { client: C; wasHit: () => boolean } { + let hit = false; + const laggingTaskRun = new Proxy((real as any).taskRun, { + get(target, prop) { + if (prop === "findFirst" || prop === "findUnique" || prop === "findFirstOrThrow") { + return async (...args: any[]) => { + hit = true; + const row = await (target as any)[prop](...args); + return row ? { ...row, lockedById: null } : row; + }; + } + return (target as any)[prop]; + }, + }); + const client = new Proxy(real as any, { + get(target, prop) { + return prop === "taskRun" ? laggingTaskRun : (target as any)[prop]; + }, + }) as C; + return { client, wasHit: () => hit }; +} + +describe("RunEngine startRunAttempt — the run-row lock check must read the owning primary, not a lagging replica", () => { + twoDbEngineTest( + "a just-dequeued NEW run starts despite a lagging replica that still shows it unlocked", + async ({ prisma14, prisma17, redisOptions }) => { + const newReplica = laggingLockReplica(prisma17); + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: newReplica.client as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14 as unknown as PrismaClient, + readOnlyPrisma: prisma14 as unknown as PrismaClient, + schemaVariant: "legacy", + }); + const router = new RoutingRunStore({ new: newStore, legacy: legacyStore }); + + const engine = new RunEngine({ + store: router, + ...baseEngineOptions(redisOptions, prisma14), + }); + + try { + const runId = `run_${RUN_OPS_A}`; + const env = await seedControlPlaneEnv(prisma14 as unknown as PrismaClient, "startlock"); + + await router.createRun( + buildCreateRunInput({ + runId, + friendlyId: "run_startlock", + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + + // Dequeue: write a PENDING_EXECUTING snapshot + lock the run on the owning primary. + const created = await router.findLatestExecutionSnapshot(runId); + const dequeued = await router.createExecutionSnapshot( + { + run: { id: runId, status: "DEQUEUED", attemptNumber: null }, + snapshot: { executionStatus: "PENDING_EXECUTING", description: "dequeued" }, + previousSnapshotId: created!.id, + environmentId: env.environment.id, + environmentType: "PRODUCTION", + projectId: env.project.id, + organizationId: env.organization.id, + }, + prisma14 as unknown as PrismaClient + ); + await prisma17.taskRun.update({ + where: { id: runId }, + data: { status: "DEQUEUED", lockedById: "worker_startlock", lockedAt: new Date() }, + }); + + // The run IS locked on the primary; the lagging replica shows it unlocked. RED: findRun (no + // client) reads the replica -> lockedById null -> ServiceValidationError "Task run is not + // locked". GREEN: the read routes to the owning primary, so the lock check passes (it then + // fails later on the unseeded worker-task config, which is out of scope for this gate). + let errorMessage: string | undefined; + try { + await engine.startRunAttempt({ runId, snapshotId: dequeued.id }); + } catch (error) { + errorMessage = error instanceof Error ? error.message : String(error); + } + expect(errorMessage ?? "").not.toContain("Task run is not locked"); + expect(newReplica.wasHit()).toBe(false); + } finally { + await engine.quit(); + } + } + ); +}); From fee35d5ca1ba584d1bf372a367ad49fbe0313812 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:25:52 +0100 Subject: [PATCH 04/16] test(run-engine): assert the resume path delivers a NEW run's completed waitpoints Covers getExecutionSnapshotsSince end to end for a NEW-residency run: the resumed snapshot must carry the completed waitpoint whose join lives on the new store, which the snapshot-id-routed lookup used to miss. --- .../tests/getSnapshotsSinceResidency.test.ts | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 internal-packages/run-engine/src/engine/tests/getSnapshotsSinceResidency.test.ts diff --git a/internal-packages/run-engine/src/engine/tests/getSnapshotsSinceResidency.test.ts b/internal-packages/run-engine/src/engine/tests/getSnapshotsSinceResidency.test.ts new file mode 100644 index 0000000000..adaefaed23 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/getSnapshotsSinceResidency.test.ts @@ -0,0 +1,161 @@ +// End-to-end resume-read path (managed worker `snapshots.since`): getExecutionSnapshotsSince -> +// getSnapshotWaitpointIds -> runStore.findSnapshotCompletedWaitpointIds(latestSnapshot.id). The +// latest snapshot id is a cuid, so the pre-fix router misrouted it to #legacy and the resumed NEW +// run received an empty completedWaitpoints list and hung. This asserts the whole path delivers the +// NEW-resident completed waitpoint. Real two-DB topology; never mocked. + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import { PostgresRunStore, RoutingRunStore, type CreateRunInput } from "@internal/run-store"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { setTimeout as sleep } from "timers/promises"; +import { getExecutionSnapshotsSince } from "../systems/executionSnapshotSystem.js"; + +const RUN_OPS_A = "n".repeat(24) + "01"; // run-ops id -> NEW (#new / prisma17) + +function makeRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + return new RoutingRunStore({ new: newStore, legacy: legacyStore }); +} + +async function seedControlPlaneEnv(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: `prod-${suffix}`, + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + maximumConcurrencyLimit: 10, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(p: { + runId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}): CreateRunInput { + return { + data: { + id: p.runId, + engine: "V2", + status: "EXECUTING", + friendlyId: "run_since", + runtimeEnvironmentId: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: p.organizationId, + projectId: p.projectId, + taskIdentifier: "since-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `trace_${p.runId}`, + spanId: `span_${p.runId}`, + runTags: [], + queue: "task/since-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: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: p.projectId, + organizationId: p.organizationId, + }, + }; +} + +describe("run-ops split — getExecutionSnapshotsSince delivers a NEW run's completed waitpoints on resume", () => { + heteroRunOpsPostgresTest( + "the resumed snapshot carries the NEW-resident completed waitpoint (managed snapshots.since path)", + async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => { + const router = makeRouter(prisma14, prisma17); + const runId = `run_${RUN_OPS_A}`; + const env = await seedControlPlaneEnv(prisma14, "sincepath"); + + await router.createRun( + buildCreateRunInput({ + runId, + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + const since = await router.findLatestExecutionSnapshot(runId); // RUN_CREATED, the "since" + + // The resumed PENDING_EXECUTING snapshot (its id is a cuid). Space it past `since` in time. + await sleep(10); + const latest = await router.createExecutionSnapshot( + { + run: { id: runId, status: "EXECUTING", attemptNumber: 1 }, + snapshot: { executionStatus: "PENDING_EXECUTING", description: "resumed" }, + previousSnapshotId: since!.id, + environmentId: env.environment.id, + environmentType: "PRODUCTION", + projectId: env.project.id, + organizationId: env.organization.id, + }, + prisma14 + ); + + // A completed waitpoint on #new, joined to the resumed snapshot (as completeWaitpoint would). + const waitpointId = `waitpoint_${RUN_OPS_A}`; + await prisma17.waitpoint.create({ + data: { + id: waitpointId, + friendlyId: "wp_since", + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date(), + idempotencyKey: `idem_${waitpointId}`, + userProvidedIdempotencyKey: false, + projectId: env.project.id, + environmentId: env.environment.id, + }, + }); + await prisma17.completedWaitpoint.create({ + data: { snapshotId: latest.id, waitpointId }, + }); + + const snapshots = await getExecutionSnapshotsSince(prisma14, runId, since!.id, router); + const resumed = snapshots.find((s) => s.id === latest.id); + + // RED: findSnapshotCompletedWaitpointIds(latest.id cuid) -> #legacy -> [] -> resumed run hangs. + // GREEN: fan-out finds the #new join -> the resumed snapshot carries the completed waitpoint. + expect(resumed?.completedWaitpoints.map((w) => w.id)).toEqual([waitpointId]); + } + ); +}); From 62247f47e35afa8cca2bec4b507ffc5b74c38604 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:41:53 +0100 Subject: [PATCH 05/16] chore(run-store,run-engine): fix formatting and bind forwarded prisma methods in test proxies oxfmt the wrapped method signature, and make the lagging-replica test proxies bind any forwarded method to the real client. Prisma delegates are proxy-based and not pre-bound, so an unbound forwarded method could trip a this/private-field brand check when called. --- .../engine/tests/startRunAttemptReadResidency.test.ts | 9 +++++++-- .../src/engine/tests/waitpointReplicaLag.test.ts | 4 +++- internal-packages/run-store/src/runOpsStore.ts | 5 ++++- 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts b/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts index 4896201514..d3ade923cf 100644 --- a/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts +++ b/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts @@ -137,12 +137,17 @@ function laggingLockReplica( return row ? { ...row, lockedById: null } : row; }; } - return (target as any)[prop]; + // Bind forwarded methods to the real client: Prisma delegates are proxy-based and not + // pre-bound, so an unbound method would trip a this/private-field brand check when called. + const forwarded = (target as any)[prop]; + return typeof forwarded === "function" ? forwarded.bind(target) : forwarded; }, }); const client = new Proxy(real as any, { get(target, prop) { - return prop === "taskRun" ? laggingTaskRun : (target as any)[prop]; + if (prop === "taskRun") return laggingTaskRun; + const forwarded = (target as any)[prop]; + return typeof forwarded === "function" ? forwarded.bind(target) : forwarded; }, }) as C; return { client, wasHit: () => hit }; diff --git a/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts b/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts index dac6ffeca4..6cbf3c3322 100644 --- a/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts +++ b/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts @@ -199,7 +199,9 @@ function laggingPendingReplica( return (target as any).$queryRaw(strings, ...values); }; } - return (target as any)[prop]; + // Bind forwarded methods to the real client (Prisma delegates are proxy-based, not pre-bound). + const forwarded = (target as any)[prop]; + return typeof forwarded === "function" ? forwarded.bind(target) : forwarded; }, }) as C; return { client, wasHit: () => hit }; diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 9e21e1b380..70967a1ab7 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -833,7 +833,10 @@ export class RoutingRunStore implements RunStore { // Snapshot ids are cuids (they always classify LEGACY), and a snapshot's CompletedWaitpoint join // co-locates with its run, which may live on either store, so fan out to BOTH and merge (like // findWaitpointCompletedSnapshotIds) rather than route by the un-classifiable snapshot id. - async findSnapshotCompletedWaitpointIds(snapshotId: string, client?: ReadClient): Promise { + async findSnapshotCompletedWaitpointIds( + snapshotId: string, + client?: ReadClient + ): Promise { const [fromNew, fromLegacy] = await Promise.all([ this.#new.findSnapshotCompletedWaitpointIds( snapshotId, From 3adba9a23e38f2e070d929f73adb51365d964091 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:45:33 +0100 Subject: [PATCH 06/16] test(run-store): assert the legacy store is untouched when completing a NEW batch item Broadens the batch-item misroute regression guard: completing a NEW-resident item must update #new and leave #legacy with zero matching rows. --- .../run-store/src/runOpsStore.batchItemMisroute.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts index edacf01f9b..b59b711bd8 100644 --- a/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts +++ b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts @@ -88,6 +88,8 @@ describe("run-ops split — completing a batch item routes by the batch id, not expect(result.count).toBe(1); const onNew = await prisma17.batchTaskRunItem.findUnique({ where: { id: item.id } }); expect(onNew?.status).toBe("COMPLETED"); + // Legacy was never touched: no phantom/double-routed item update on #legacy. + expect(await prisma14.batchTaskRunItem.count({ where: { batchTaskRunId: batchId } })).toBe(0); } ); }); From b7936c5c0200a2255ab884a00a6bbffecab5e621 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 21:52:37 +0100 Subject: [PATCH 07/16] fix(webapp): read run results from the owning store so triggerAndWait resolves for NEW runs The run-result route built ApiRunResultPresenter with no run-ops read clients, so under the split it read a NEW-residency run's row on the control-plane database, found nothing, and returned 404. triggerAndWait and runs.retrieve().result then never resolved for NEW runs, stalling the parent. Wire the run-ops read clients the same way the batch-results route does. --- apps/webapp/app/routes/api.v1.runs.$runParam.result.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts index 4cbf27d327..7444f9b554 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts @@ -2,6 +2,7 @@ import type { LoaderFunctionArgs } from "@remix-run/server-runtime"; import { json } from "@remix-run/server-runtime"; import { z } from "zod"; import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server"; +import { $replica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server"; import { authenticateApiRequest } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; @@ -27,7 +28,11 @@ export async function loader({ request, params }: LoaderFunctionArgs) { const { runParam } = parsed.data; try { - const presenter = new ApiRunResultPresenter(); + const presenter = new ApiRunResultPresenter(undefined, undefined, { + newClient: runOpsNewReplica, + legacyReplica: $replica, + splitEnabled: runOpsSplitReadEnabled, + }); const result = await presenter.call(runParam, authenticationResult.environment); if (!result) { From 9bf374971f78eaf51ca56d82968c73ea8165a9d4 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:20:36 +0100 Subject: [PATCH 08/16] fix(run-engine): clear a NEW run's blocking waitpoints on the owning store clearBlockingWaitpoints deleted edges via the caller's control-plane tx, so a NEW run's TaskRunWaitpoint edges (on the run-ops database) were orphaned and re-blocked the run after a retry. Route the delete through the store, which fans across both databases and applies the tx only to the legacy leg. --- .../src/engine/systems/waitpointSystem.ts | 19 +- .../clearBlockingWaitpointsResidency.test.ts | 182 ++++++++++++++++++ 2 files changed, 189 insertions(+), 12 deletions(-) create mode 100644 internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 6969363d1a..948346fca4 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -59,18 +59,13 @@ export class WaitpointSystem { runId: string; tx?: PrismaClientOrTransaction; }) { - // A tx pins a specific client and must not be re-routed through the store. - const deleted = tx - ? await tx.taskRunWaitpoint.deleteMany({ - where: { - taskRunId: runId, - }, - }) - : await this.$.runStore.deleteManyTaskRunWaitpoints({ - where: { - taskRunId: runId, - }, - }); + // Route the delete: a run's edges may live on #new and/or #legacy (mid-drain), so it must fan + // across both stores. The caller's control-plane tx would only clear #legacy, orphaning #new + // edges that then re-block the run after a retry. The router applies tx to the #legacy leg only. + const deleted = await this.$.runStore.deleteManyTaskRunWaitpoints( + { where: { taskRunId: runId } }, + tx + ); return deleted.count; } diff --git a/internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts b/internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts new file mode 100644 index 0000000000..f989beb475 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts @@ -0,0 +1,182 @@ +// clearBlockingWaitpoints (called by attemptFailed) deleted edges via the caller's control-plane tx. +// A NEW run's TaskRunWaitpoint edges live on #new, so the control-plane deleteMany matched 0 rows and +// left them orphaned; on a retry+re-block those stale PENDING edges re-block the run forever. The fix +// routes the delete through the store (which fans across both DBs, dropping the tx for the cross-DB +// path). Real two-DB topology; never mocked. + +import { + heteroRunOpsPostgresTest, + network, + redisContainer, + redisOptions, +} from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { PostgresRunStore, RoutingRunStore, type CreateRunInput } from "@internal/run-store"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; + +const twoDbEngineTest = heteroRunOpsPostgresTest.extend<{ + redisContainer: any; + redisOptions: any; +}>({ + network, + redisContainer, + redisOptions, +}); + +const RUN_OPS_A = "n".repeat(24) + "01"; // run-ops id -> NEW (#new / prisma17) + +function baseEngineOptions(redisOptions: any, prisma: any) { + return { + prisma, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { redis: redisOptions }, + machines: { + defaultMachine: "small-1x" as const, + machines: { + "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }; +} + +function makeRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + return new RoutingRunStore({ new: newStore, legacy: legacyStore }); +} + +async function seedControlPlaneEnv(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: `prod-${suffix}`, + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + maximumConcurrencyLimit: 10, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(p: { + runId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}): CreateRunInput { + return { + data: { + id: p.runId, + engine: "V2", + status: "EXECUTING", + friendlyId: "run_clearwp", + runtimeEnvironmentId: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: p.organizationId, + projectId: p.projectId, + taskIdentifier: "clearwp-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `trace_${p.runId}`, + spanId: `span_${p.runId}`, + runTags: [], + queue: "task/clearwp-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: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: p.projectId, + organizationId: p.organizationId, + }, + }; +} + +describe("RunEngine clearBlockingWaitpoints — clears a NEW run's edges even with a control-plane tx", () => { + twoDbEngineTest( + "a control-plane tx does not leave a NEW run's #new-resident blocking edge orphaned", + async ({ prisma14, prisma17, redisOptions }) => { + const router = makeRouter(prisma14 as unknown as PrismaClient, prisma17); + const engine = new RunEngine({ + store: router, + ...baseEngineOptions(redisOptions, prisma14), + }); + + try { + const runId = `run_${RUN_OPS_A}`; + const env = await seedControlPlaneEnv(prisma14 as unknown as PrismaClient, "clearwp"); + await router.createRun( + buildCreateRunInput({ + runId, + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + // A NEW run's blocking edge lives on #new. + await prisma17.taskRunWaitpoint.create({ + data: { + id: "trw_clearwp_0000000000001", + taskRunId: runId, + waitpointId: "waitpoint_clearwp_00000001", + projectId: env.project.id, + }, + }); + + // attemptFailed passes the control-plane tx; the #new edge must still be cleared. + const count = await engine.waitpointSystem.clearBlockingWaitpoints({ + runId, + tx: prisma14 as unknown as PrismaClient, + }); + + // RED: the tx deletes on #legacy -> 0; the #new edge is orphaned. + // GREEN: the routed delete fans out -> the #new edge is cleared. + expect(count).toBe(1); + expect(await prisma17.taskRunWaitpoint.count({ where: { taskRunId: runId } })).toBe(0); + } finally { + await engine.quit(); + } + } + ); +}); From 2821b05649130ceea88b1a830ed2abc4d2e97a0b Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:20:36 +0100 Subject: [PATCH 09/16] fix(run-store): write cross-DB waitpoint blocking edges via unnest blockRunWithWaitpointEdges' legacy branch joined FROM "Waitpoint", so a LEGACY run blocking on a NEW-resident token (whose Waitpoint row lives on the run-ops database) matched no rows and wrote no edge, silently stranding the run. Source the edge rows from the id array via unnest, matching the dedicated branch. Two migrations drop the now-cross-DB TaskRunWaitpoint and _WaitpointRunConnections foreign keys to Waitpoint; integrity is app-enforced, matching the split's existing control-plane FK-removal pattern. --- .../migration.sql | 3 + .../migration.sql | 4 + .../run-store/src/PostgresRunStore.ts | 20 +-- .../src/runOpsStore.crossDbTokenBlock.test.ts | 144 ++++++++++++++++++ 4 files changed, 161 insertions(+), 10 deletions(-) create mode 100644 internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql create mode 100644 internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql create mode 100644 internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts diff --git a/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql new file mode 100644 index 0000000000..526b44444e --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql @@ -0,0 +1,3 @@ +-- Run-ops split: allow a cross-DB token connection (a LEGACY run blocking on a NEW-resident token, +-- whose Waitpoint row lives on the other database). Matches the split's control-plane FK-removal pattern. +ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey"; diff --git a/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql new file mode 100644 index 0000000000..69f1ba3c2d --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql @@ -0,0 +1,4 @@ +-- Run-ops split: drop the TaskRunWaitpoint -> Waitpoint FK so a LEGACY run's blocking edge can point +-- at a NEW-resident (cross-DB) token. The #new dedicated schema is already FK-free here; this aligns +-- #legacy. Referential integrity is app-enforced, matching the split's control-plane FK-removal. +ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_waitpointId_fkey"; diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index b797fae9b9..2befca0269 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -1709,9 +1709,11 @@ export class PostgresRunStore implements RunStore { return; } - // Insert the blocking connections and the historical run connections. - // We use a CTE to do both inserts atomically. Data-modifying CTEs are - // always executed regardless of whether they're referenced in the outer query. + // Source edges from the id array via `unnest` (like the dedicated branch), NOT `FROM "Waitpoint"`: + // a cross-DB token (LEGACY run -> NEW-resident token) lives on the other DB, so the join matched 0 + // rows and the run hung. Needs the _WaitpointRunConnections -> Waitpoint FK dropped (migration); + // casts are required because `unnest` gives no column types for the nullable params. + const ids = waitpointIds; await prisma.$queryRaw` WITH inserted AS ( INSERT INTO "TaskRunWaitpoint" ("id", "taskRunId", "waitpointId", "projectId", "createdAt", "updatedAt", "spanIdToComplete", "batchId", "batchIndex") @@ -1722,19 +1724,17 @@ export class PostgresRunStore implements RunStore { ${projectId}, NOW(), NOW(), - ${spanIdToComplete ?? null}, - ${batchId ?? null}, - ${batchIndex ?? null} - FROM "Waitpoint" w - WHERE w.id IN (${Prisma.join(waitpointIds)}) + ${spanIdToComplete ?? null}::text, + ${batchId ?? null}::text, + ${batchIndex ?? null}::int + FROM unnest(${ids}::text[]) AS w(id) ON CONFLICT DO NOTHING RETURNING "waitpointId" ), connected_runs AS ( INSERT INTO "_WaitpointRunConnections" ("A", "B") SELECT ${runId}, w.id - FROM "Waitpoint" w - WHERE w.id IN (${Prisma.join(waitpointIds)}) + FROM unnest(${ids}::text[]) AS w(id) ON CONFLICT DO NOTHING ) SELECT COUNT(*) FROM inserted`; diff --git a/internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts b/internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts new file mode 100644 index 0000000000..071a37a2a5 --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts @@ -0,0 +1,144 @@ +// blockRunWithWaitpointEdges' legacy branch joined `FROM "Waitpoint" w`, so a LEGACY run blocking on +// a NEW-resident (run-ops) token — whose Waitpoint row lives on #new — matched 0 rows and wrote NO +// blocking edge. countPendingWaitpoints then fans out, still sees the token PENDING, so the run is +// suspended believing it's blocked while nothing will ever resume it -> silent hang. The fix sources +// the edge rows from the id array via `unnest` (FK-free, mirroring the dedicated branch); a migration +// drops the _WaitpointRunConnections -> Waitpoint FK so the cross-DB connection can be recorded too. +// Real two-DB topology; never mocked. + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { PostgresRunStore } from "./PostgresRunStore.js"; +import { RoutingRunStore } from "./runOpsStore.js"; + +const CUID_25 = "c".repeat(25); // LEGACY run -> #legacy (prisma14, full schema) + +function makeRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + return new RoutingRunStore({ new: newStore, legacy: legacyStore }); +} + +async function seedEnvironmentLegacy(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "DEVELOPMENT", + slug: "dev", + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_dev_${suffix}`, + pkApiKey: `pk_dev_${suffix}`, + shortcode: `short_${suffix}`, + }, + }); + return { organization, project, environment }; +} + +function taskRunData(opts: { + id: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}) { + return { + id: opts.id, + engine: "V2" as const, + status: "EXECUTING" as const, + friendlyId: `run_${opts.id}`, + runtimeEnvironmentId: opts.runtimeEnvironmentId, + environmentType: "DEVELOPMENT" as const, + organizationId: opts.organizationId, + projectId: opts.projectId, + taskIdentifier: "xdb-task", + payload: "{}", + payloadType: "application/json", + traceContext: {}, + traceId: `trace_${opts.id}`, + spanId: `span_${opts.id}`, + queue: "task/xdb-task", + isTest: false, + taskEventStore: "taskEvent", + depth: 0, + }; +} + +describe("run-ops split — a LEGACY run blocking on a NEW-resident token gets its blocking edge (cross-DB)", () => { + heteroRunOpsPostgresTest( + "blockRunWithWaitpointEdges writes the edge for a cross-DB token instead of stranding the run", + async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => { + const router = makeRouter(prisma14, prisma17); + // The harness builds the schema with `prisma db push`, which re-creates the FKs that the + // run-ops split migrations (20260705210000 / 20260705220000) drop in prod. Mirror those drops on + // this clone so the cross-DB insert exercises the real prod state (FKs gone, app-enforced). + await prisma14.$executeRawUnsafe( + `ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_waitpointId_fkey"` + ); + await prisma14.$executeRawUnsafe( + `ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey"` + ); + const seed = await seedEnvironmentLegacy(prisma14, "xdbblock"); + const runId = `run_${CUID_25}`; // LEGACY run, resident on #legacy + await prisma14.taskRun.create({ + data: taskRunData({ + id: runId, + organizationId: seed.organization.id, + projectId: seed.project.id, + runtimeEnvironmentId: seed.environment.id, + }), + }); + + // The token was minted co-located with a run-ops run, so its Waitpoint row lives on #new. + const tokenId = "waitpoint_" + "x".repeat(20); + await prisma17.waitpoint.create({ + data: { + id: tokenId, + friendlyId: "wp_xdb", + type: "MANUAL", + status: "PENDING", + idempotencyKey: `idem_${tokenId}`, + userProvidedIdempotencyKey: false, + projectId: seed.project.id, + environmentId: seed.environment.id, + }, + }); + + await router.blockRunWithWaitpointEdges({ + runId, + waitpointIds: [tokenId], + projectId: seed.project.id, + }); + + // RED: the legacy `FROM "Waitpoint"` join matches 0 rows -> no edge -> the run is stranded. + // GREEN: `unnest` writes the edge from the id directly. + expect( + await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId, waitpointId: tokenId } }) + ).toBe(1); + // The historical connection is recorded too (needs the dropped B-fkey migration). + const conn = (await prisma14.$queryRaw` + SELECT "B" FROM "_WaitpointRunConnections" WHERE "A" = ${runId} + `) as unknown[]; + expect(conn.length).toBe(1); + } + ); +}); From 31a632a202bb2bcc8f6a1d69cbd7685d5f8d85b3 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:28:22 +0100 Subject: [PATCH 10/16] fix(run-store): record cross-DB completed waitpoints via an FK-free insert createExecutionSnapshot and lockRunToWorker recorded completed waitpoints with a Prisma connect on the implicit _completedWaitpoints M2M, which ORM-validates the Waitpoint exists locally and rejects a cross-DB (NEW-resident) token. A LEGACY parent that triggerAndWaits a NEW child then hangs when the resume snapshot connects the NEW token. Insert the join rows FK-free after create, mirroring the dedicated schema, and drop the _completedWaitpoints to Waitpoint FK by migration. --- .../migration.sql | 4 + .../run-store/src/PostgresRunStore.ts | 56 +++--- ...OpsStore.crossDbCompletedWaitpoint.test.ts | 162 ++++++++++++++++++ 3 files changed, 199 insertions(+), 23 deletions(-) create mode 100644 internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql create mode 100644 internal-packages/run-store/src/runOpsStore.crossDbCompletedWaitpoint.test.ts diff --git a/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql new file mode 100644 index 0000000000..cb9e631ee3 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql @@ -0,0 +1,4 @@ +-- Run-ops split: drop the _completedWaitpoints -> Waitpoint FK so a LEGACY snapshot can record a +-- cross-DB completed token (NEW-resident). Third of the split's waitpoint-FK drops; the A (snapshot) +-- side stays same-DB. Referential integrity is app-enforced, matching the sibling FK-removals. +ALTER TABLE "_completedWaitpoints" DROP CONSTRAINT IF EXISTS "_completedWaitpoints_B_fkey"; diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 2befca0269..ae1a9eca64 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -929,6 +929,26 @@ export class PostgresRunStore implements RunStore { }); } + // Legacy implicit M2M equivalent of #connectCompletedWaitpoints: raw-insert the join rows FK-free. + // Prisma `connect` ORM-validates the Waitpoint exists locally, which fails for a cross-DB + // (NEW-resident) token; the raw insert + dropped _completedWaitpoints_B_fkey records it. A = + // TaskRunExecutionSnapshot.id, B = Waitpoint.id (implicit M2M alphabetical order). + async #connectCompletedWaitpointsLegacy( + client: PrismaClientOrTransaction, + snapshotId: string, + waitpointIds: string[] + ): Promise { + if (waitpointIds.length === 0) { + return; + } + + await client.$executeRaw` + INSERT INTO "_completedWaitpoints" ("A", "B") + SELECT ${snapshotId}, w.id + FROM unnest(${waitpointIds}::text[]) AS w(id) + ON CONFLICT DO NOTHING`; + } + async lockRunToWorker( runId: string, data: LockRunData, @@ -970,15 +990,7 @@ export class PostgresRunStore implements RunStore { organizationId: data.snapshot.organizationId, checkpointId: data.snapshot.checkpointId ?? undefined, batchId: data.snapshot.batchId ?? undefined, - // Legacy: connect the implicit M2M. Dedicated: links inserted below into the - // CompletedWaitpoint join model (no such relation field exists on that schema). - ...(dedicated - ? {} - : { - completedWaitpoints: { - connect: data.snapshot.completedWaitpointIds.map((id) => ({ id })), - }, - }), + // Completed-waitpoint links are inserted FK-free after create (below) for BOTH schemas. completedWaitpointOrder: data.snapshot.completedWaitpointOrder, workerId: data.snapshot.workerId ?? undefined, runnerId: data.snapshot.runnerId ?? undefined, @@ -993,6 +1005,12 @@ export class PostgresRunStore implements RunStore { data.snapshot.id, data.snapshot.completedWaitpointIds ); + } else { + await this.#connectCompletedWaitpointsLegacy( + prisma, + data.snapshot.id, + data.snapshot.completedWaitpointIds + ); } return result; @@ -1554,15 +1572,8 @@ export class PostgresRunStore implements RunStore { workerId, runnerId, metadata: snapshot.metadata ?? undefined, - // Legacy: connect the implicit M2M. Dedicated: links inserted below into the - // CompletedWaitpoint join model (no such relation field exists on that schema). - ...(dedicated - ? {} - : { - completedWaitpoints: { - connect: completedWaitpoints?.map((w) => ({ id: w.id })), - }, - }), + // Completed-waitpoint links are inserted FK-free after create (below) for BOTH schemas, so a + // cross-DB (NEW-resident) token can be recorded without a Prisma `connect` existence check. completedWaitpointOrder: completedWaitpoints ?.filter((c) => c.index !== undefined) .sort((a, b) => a.index! - b.index!) @@ -1573,12 +1584,11 @@ export class PostgresRunStore implements RunStore { include: { checkpoint: true }, }); + const completedWaitpointIds = completedWaitpoints?.map((w) => w.id) ?? []; if (dedicated) { - await this.#connectCompletedWaitpoints( - prisma, - newSnapshot.id, - completedWaitpoints?.map((w) => w.id) ?? [] - ); + await this.#connectCompletedWaitpoints(prisma, newSnapshot.id, completedWaitpointIds); + } else { + await this.#connectCompletedWaitpointsLegacy(prisma, newSnapshot.id, completedWaitpointIds); } return newSnapshot; diff --git a/internal-packages/run-store/src/runOpsStore.crossDbCompletedWaitpoint.test.ts b/internal-packages/run-store/src/runOpsStore.crossDbCompletedWaitpoint.test.ts new file mode 100644 index 0000000000..c500c1d934 --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.crossDbCompletedWaitpoint.test.ts @@ -0,0 +1,162 @@ +// createExecutionSnapshot's legacy branch records completed waitpoints via Prisma `connect` on the +// implicit _completedWaitpoints M2M, whose FK to Waitpoint rejects a cross-DB (NEW-resident) token. +// A LEGACY parent doing triggerAndWait on a NEW child completes on the NEW token; the resume snapshot +// (routed to #legacy) then connects that NEW token -> FK violation -> the legacy parent hangs forever. +// Fix: drop _completedWaitpoints_B_fkey (migration); with the FK gone the connect records the cross-DB +// link (app-enforced integrity). Real two-DB topology; never mocked. + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { expect } from "vitest"; +import { PostgresRunStore } from "./PostgresRunStore.js"; +import { RoutingRunStore } from "./runOpsStore.js"; +import type { CreateRunInput } from "./types.js"; + +const CUID_25 = "c".repeat(25); // LEGACY run -> #legacy (prisma14) + +function makeRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) { + const newStore = new PostgresRunStore({ + prisma: prisma17 as never, + readOnlyPrisma: prisma17 as never, + schemaVariant: "dedicated", + }); + const legacyStore = new PostgresRunStore({ + prisma: prisma14, + readOnlyPrisma: prisma14, + schemaVariant: "legacy", + }); + return new RoutingRunStore({ new: newStore, legacy: legacyStore }); +} + +async function seedEnvironmentLegacy(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "PRODUCTION", + slug: `prod-${suffix}`, + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_prod_${suffix}`, + pkApiKey: `pk_prod_${suffix}`, + shortcode: `short_${suffix}`, + maximumConcurrencyLimit: 10, + }, + }); + return { organization, project, environment }; +} + +function buildCreateRunInput(p: { + runId: string; + organizationId: string; + projectId: string; + runtimeEnvironmentId: string; +}): CreateRunInput { + return { + data: { + id: p.runId, + engine: "V2", + status: "EXECUTING", + friendlyId: "run_cwc", + runtimeEnvironmentId: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + organizationId: p.organizationId, + projectId: p.projectId, + taskIdentifier: "cwc-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: `trace_${p.runId}`, + spanId: `span_${p.runId}`, + runTags: [], + queue: "task/cwc-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: p.runtimeEnvironmentId, + environmentType: "PRODUCTION", + projectId: p.projectId, + organizationId: p.organizationId, + }, + }; +} + +describe("run-ops split — a LEGACY snapshot can record a cross-DB completed waitpoint (NEW-resident token)", () => { + heteroRunOpsPostgresTest( + "createExecutionSnapshot connects a NEW-resident completed token to a LEGACY run's snapshot", + async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => { + const router = makeRouter(prisma14, prisma17); + // The harness builds the schema with `prisma db push`, which re-creates the FK that migration + // 20260705230000 drops in prod. Mirror the drop on this clone so the connect exercises prod state. + await prisma14.$executeRawUnsafe( + `ALTER TABLE "_completedWaitpoints" DROP CONSTRAINT IF EXISTS "_completedWaitpoints_B_fkey"` + ); + const env = await seedEnvironmentLegacy(prisma14, "cwc"); + const runId = `run_${CUID_25}`; // LEGACY run -> #legacy + await router.createRun( + buildCreateRunInput({ + runId, + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + const created = await router.findLatestExecutionSnapshot(runId); + + // A completed token minted co-located with a NEW child -> its Waitpoint row lives on #new. + const tokenId = "waitpoint_" + "y".repeat(20); + await prisma17.waitpoint.create({ + data: { + id: tokenId, + friendlyId: "wp_cwc", + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date(), + idempotencyKey: `idem_${tokenId}`, + userProvidedIdempotencyKey: false, + projectId: env.project.id, + environmentId: env.environment.id, + }, + }); + + const snap = await router.createExecutionSnapshot( + { + run: { id: runId, status: "EXECUTING", attemptNumber: 1 }, + snapshot: { executionStatus: "EXECUTING_WITH_WAITPOINTS", description: "resumed" }, + previousSnapshotId: created!.id, + completedWaitpoints: [{ id: tokenId, index: 0 }], + environmentId: env.environment.id, + environmentType: "PRODUCTION", + projectId: env.project.id, + organizationId: env.organization.id, + }, + prisma14 + ); + + // RED: the connect hits _completedWaitpoints_B_fkey (token not on #legacy) -> throws -> parent hangs. + // GREEN (FK dropped): the cross-DB completed-waitpoint link is recorded. + const link = (await prisma14.$queryRaw` + SELECT "B" FROM "_completedWaitpoints" WHERE "A" = ${snap.id} + `) as { B: string }[]; + expect(link.map((r) => r.B)).toEqual([tokenId]); + } + ); +}); From c5dd23c8f8bbc738f53c2755370ba77d9458b453 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:28:22 +0100 Subject: [PATCH 11/16] fix(webapp): read the span-panel waitpoint from the owning store SpanPresenter built WaitpointPresenter with no run-ops read clients, so the trace panel read a NEW-residency waitpoint on the control-plane database and showed "Waitpoint not found". Wire the run-ops read clients like the other split-aware routes. --- apps/webapp/app/presenters/v3/SpanPresenter.server.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index ea2a89eaa1..d5d1c8c660 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts @@ -10,6 +10,7 @@ import { } from "@trigger.dev/core/v3"; import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic"; +import { $replica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server"; import { extractIdempotencyKeyScope, getUserProvidedIdempotencyKey, @@ -722,7 +723,11 @@ export class SpanPresenter extends BasePresenter { return { ...data, entity: null }; } - const presenter = new WaitpointPresenter(undefined, undefined, {}); + const presenter = new WaitpointPresenter(undefined, undefined, { + newClient: runOpsNewReplica, + legacyReplica: $replica, + splitEnabled: runOpsSplitReadEnabled, + }); const waitpoint = await presenter.call({ friendlyId: span.entity.id, environmentId, From 12c839c7f5e1403aa12981000fd79789ac84ab73 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:31:52 +0100 Subject: [PATCH 12/16] chore(database): add lock_timeout to the run-ops waitpoint FK-drop migrations Match the split's other FK-drop migrations: fail fast on the ACCESS EXCLUSIVE lock instead of queueing behind a long transaction or VACUUM. --- .../migration.sql | 4 ++++ .../migration.sql | 4 ++++ .../migration.sql | 4 ++++ 3 files changed, 12 insertions(+) diff --git a/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql index 526b44444e..aca9ceb1ba 100644 --- a/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql +++ b/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql @@ -1,3 +1,7 @@ -- Run-ops split: allow a cross-DB token connection (a LEGACY run blocking on a NEW-resident token, -- whose Waitpoint row lives on the other database). Matches the split's control-plane FK-removal pattern. + +-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock. +SET lock_timeout = '5s'; + ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey"; diff --git a/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql index 69f1ba3c2d..640de2d709 100644 --- a/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql +++ b/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql @@ -1,4 +1,8 @@ -- Run-ops split: drop the TaskRunWaitpoint -> Waitpoint FK so a LEGACY run's blocking edge can point -- at a NEW-resident (cross-DB) token. The #new dedicated schema is already FK-free here; this aligns -- #legacy. Referential integrity is app-enforced, matching the split's control-plane FK-removal. + +-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock. +SET lock_timeout = '5s'; + ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_waitpointId_fkey"; diff --git a/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql b/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql index cb9e631ee3..bb4ef35774 100644 --- a/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql +++ b/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql @@ -1,4 +1,8 @@ -- Run-ops split: drop the _completedWaitpoints -> Waitpoint FK so a LEGACY snapshot can record a -- cross-DB completed token (NEW-resident). Third of the split's waitpoint-FK drops; the A (snapshot) -- side stays same-DB. Referential integrity is app-enforced, matching the sibling FK-removals. + +-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock. +SET lock_timeout = '5s'; + ALTER TABLE "_completedWaitpoints" DROP CONSTRAINT IF EXISTS "_completedWaitpoints_B_fkey"; From 1f714091af8bd09c8b56f6473e4c470f57eee091 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:41:28 +0100 Subject: [PATCH 13/16] fix(run-store): accept the run-ops client in the legacy completed-waitpoint connect The legacy raw-insert helper only calls $executeRaw but typed its client parameter narrower than its dedicated sibling, so callers passing the store's own client failed the build. Widen it to the same client type. --- internal-packages/run-store/src/PostgresRunStore.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index ae1a9eca64..94630cdb2c 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -934,7 +934,7 @@ export class PostgresRunStore implements RunStore { // (NEW-resident) token; the raw insert + dropped _completedWaitpoints_B_fkey records it. A = // TaskRunExecutionSnapshot.id, B = Waitpoint.id (implicit M2M alphabetical order). async #connectCompletedWaitpointsLegacy( - client: PrismaClientOrTransaction, + client: RunOpsCapableClient, snapshotId: string, waitpointIds: string[] ): Promise { From f0efad3fb52ccc1aaeb1e06172e24fc289ffc036 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 22:47:17 +0100 Subject: [PATCH 14/16] fix(webapp): gather waitpoint connected runs across both run-ops stores The span-panel waitpoint presenter selected connectedRuns as a Prisma relation. That field does not exist on the dedicated run-ops Waitpoint model, so with the split read enabled the lookup threw a validation error; it also could only ever see connections whose join row lived on the waitpoint's own store, missing any run connected across the two databases. Read the run<->waitpoint join from each store instead (the explicit WaitpointRunConnection table on the dedicated schema, the implicit _WaitpointRunConnections M2M on the control plane), resolve each run's friendlyId on its own store, and union the results. --- .../v3/WaitpointPresenter.server.ts | 68 ++++- ...dedicatedConnectedRuns.readthrough.test.ts | 253 ++++++++++++++++++ 2 files changed, 314 insertions(+), 7 deletions(-) create mode 100644 apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts index a56c5e76ad..74f0c06fdd 100644 --- a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts @@ -45,12 +45,6 @@ export class WaitpointPresenter extends BasePresenter { completedAfter: true, completedAt: true, createdAt: true, - connectedRuns: { - select: { - friendlyId: true, - }, - take: 5, - }, tags: true, environmentId: true, } as const; @@ -80,6 +74,66 @@ export class WaitpointPresenter extends BasePresenter { return result.source === "new" || result.source === "legacy-replica" ? result.value : null; } + // Connected-run friendlyIds gathered across BOTH stores. The run<->waitpoint join co-locates with + // the RUN (written on the run's DB), so the waitpoint's own store misses a cross-DB connection; we + // read the join on each client and resolve the run's friendlyId on that same client, then union. + // We never relation-select `connectedRuns`: it is not a field on the dedicated subset `Waitpoint`. + async #connectedRunFriendlyIds(waitpointId: string): Promise { + const replica = this._replica as unknown as PrismaReplicaClient; + const rawClients: PrismaReplicaClient[] = + this.readThroughDeps?.splitEnabled === true + ? [ + (this.readThroughDeps.newClient as PrismaReplicaClient | undefined) ?? replica, + (this.readThroughDeps.legacyReplica as PrismaReplicaClient | undefined) ?? replica, + ] + : [replica]; + const clients = [...new Set(rawClients)]; + + const friendlyIds = new Set(); + for (const client of clients) { + const runIds = await this.#connectedRunIdsOn(client, waitpointId); + if (runIds.length === 0) { + continue; + } + const runs = await client.taskRun.findMany({ + where: { id: { in: runIds } }, + select: { friendlyId: true }, + take: 5, + }); + for (const run of runs) { + friendlyIds.add(run.friendlyId); + } + if (friendlyIds.size >= 5) { + break; + } + } + return Array.from(friendlyIds).slice(0, 5); + } + + // Schema-aware read of the run ids linked to a waitpoint: the dedicated subset uses the explicit + // `WaitpointRunConnection` model, the control-plane full schema the implicit `_WaitpointRunConnections` + // M2M (A = TaskRun.id, B = Waitpoint.id). The dedicated join delegate is absent on the full client. + async #connectedRunIdsOn(client: PrismaReplicaClient, waitpointId: string): Promise { + const joinDelegate = ( + client as unknown as { + waitpointRunConnection?: { + findMany: (args: unknown) => Promise<{ taskRunId: string }[]>; + }; + } + ).waitpointRunConnection; + if (joinDelegate && typeof joinDelegate.findMany === "function") { + const links = await joinDelegate.findMany({ + where: { waitpointId }, + select: { taskRunId: true }, + }); + return links.map((link) => link.taskRunId); + } + const rows = await client.$queryRaw<{ A: string }[]>` + SELECT "A" FROM "_WaitpointRunConnections" WHERE "B" = ${waitpointId} + `; + return rows.map((row) => row.A); + } + public async call({ friendlyId, environmentId, @@ -119,7 +173,7 @@ export class WaitpointPresenter extends BasePresenter { } } - const connectedRunIds = waitpoint.connectedRuns.map((run) => run.friendlyId); + const connectedRunIds = await this.#connectedRunFriendlyIds(waitpoint.id); const connectedRuns: NextRunListItem[] = []; if (connectedRunIds.length > 0) { diff --git a/apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts b/apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts new file mode 100644 index 0000000000..a71a47d53a --- /dev/null +++ b/apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts @@ -0,0 +1,253 @@ +import { describe, expect, vi } from "vitest"; + +// Uses the REAL dedicated run-ops client (RunOpsPrismaClient / SUBSET schema) as the new-DB handle, +// whose `Waitpoint` model has NO `connectedRuns` relation — so a relation-select of it throws rather +// than missing. The existing suite can't catch that: it wires a full-schema PG17 as the "new" client. +// NextRunListPresenter is stubbed to echo its `runId` set back as `runs`, so `result.connectedRuns` is +// exactly the friendlyIds the presenter gathered cross-DB (isolating the gather from the CH hydrate). +const legacyReplicaHolder = vi.hoisted(() => ({ client: undefined as any })); +const newClientHolder = vi.hoisted(() => ({ client: undefined as any })); + +vi.mock("~/db.server", async () => { + const { Prisma } = await import("@trigger.dev/database"); + const lazyProxy = (holder: { client: any }, label: string) => + new Proxy( + {}, + { + get(_t, prop) { + if (!holder.client) { + throw new Error(`${label} not set for this test`); + } + return holder.client[prop]; + }, + } + ); + const replicaProxy = lazyProxy(legacyReplicaHolder, "legacyReplicaHolder.client"); + return { + prisma: replicaProxy, + $replica: replicaProxy, + runOpsNewPrisma: lazyProxy(newClientHolder, "newClientHolder.client"), + runOpsNewReplica: lazyProxy(newClientHolder, "newClientHolder.client"), + runOpsLegacyPrisma: replicaProxy, + runOpsLegacyReplica: replicaProxy, + sqlDatabaseSchema: Prisma.sql([`public`]), + }; +}); + +vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({ + clickhouseFactory: { + getClickhouseForOrganization: async () => ({}), + }, +})); + +// Echo the runId set back as runs so `result.connectedRuns` == the friendlyIds the presenter gathered. +vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({ + NextRunListPresenter: class { + // eslint-disable-next-line @typescript-eslint/no-unused-vars + constructor(..._args: unknown[]) {} + async call(_organizationId: string, _environmentId: string, opts: { runId?: string[] }) { + return { + runs: (opts.runId ?? []).map((friendlyId) => ({ + friendlyId, + taskIdentifier: "echoed", + })), + }; + } + }, +})); + +import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; +import { WaitpointPresenter } from "~/presenters/v3/WaitpointPresenter.server"; + +vi.setConfig({ testTimeout: 90_000 }); + +type SeedContext = { + organizationId: string; + projectId: string; + environmentId: string; +}; + +// Parents (org/project/env) only exist on the full control-plane schema; the dedicated subset has no +// such models, so we always seed them on the legacy (PG14) client and let the resolver read them there. +async function seedParents(prisma: PrismaClient, slug: string): Promise { + const organization = await prisma.organization.create({ + data: { title: `org-${slug}`, slug: `org-${slug}` }, + }); + const project = await prisma.project.create({ + data: { + name: `proj-${slug}`, + slug: `proj-${slug}`, + organizationId: organization.id, + externalRef: `proj-${slug}`, + }, + }); + const runtimeEnvironment = await prisma.runtimeEnvironment.create({ + data: { + slug: `env-${slug}`, + type: "DEVELOPMENT", + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_dev_${slug}`, + pkApiKey: `pk_dev_${slug}`, + shortcode: `sc-${slug}`, + }, + }); + return { + organizationId: organization.id, + projectId: project.id, + environmentId: runtimeEnvironment.id, + }; +} + +async function seedWaitpoint( + prisma: PrismaClient | RunOpsPrismaClient, + ctx: SeedContext, + friendlyId: string +) { + return (prisma as PrismaClient).waitpoint.create({ + data: { + friendlyId, + type: "MANUAL", + status: "COMPLETED", + idempotencyKey: `idem-${friendlyId}`, + userProvidedIdempotencyKey: false, + output: JSON.stringify({ hello: "world" }), + outputType: "application/json", + outputIsError: false, + completedAt: new Date(), + tags: ["a", "b"], + projectId: ctx.projectId, + environmentId: ctx.environmentId, + }, + }); +} + +async function seedRun( + prisma: PrismaClient | RunOpsPrismaClient, + ctx: SeedContext, + friendlyId: string +) { + return (prisma as PrismaClient).taskRun.create({ + data: { + friendlyId, + taskIdentifier: "my-task", + status: "PENDING", + payload: JSON.stringify({ foo: friendlyId }), + payloadType: "application/json", + traceId: friendlyId, + spanId: friendlyId, + queue: "test", + runtimeEnvironmentId: ctx.environmentId, + projectId: ctx.projectId, + organizationId: ctx.organizationId, + environmentType: "DEVELOPMENT", + engine: "V2", + }, + }); +} + +const callArgs = (ctx: SeedContext, friendlyId: string) => ({ + friendlyId, + environmentId: ctx.environmentId, + projectId: ctx.projectId, +}); + +describe("WaitpointPresenter against the REAL dedicated run-ops client", () => { + // A NEW-resident waitpoint (on the dedicated subset schema) with no connected runs. The current + // relation-select of `connectedRuns` is invalid on the dedicated Waitpoint model, so the read + // throws PrismaClientValidationError. Desired: resolves the waitpoint, connectedRuns empty. + heteroRunOpsPostgresTest( + "resolves a new-resident waitpoint without a connectedRuns relation-select (no throw)", + async ({ prisma14, prisma17 }) => { + const ctx = await seedParents(prisma14, "dedself"); + const seeded = await seedWaitpoint(prisma17, ctx, "waitpoint_dedself"); + + legacyReplicaHolder.client = prisma14; + newClientHolder.client = prisma17; + + const presenter = new WaitpointPresenter(undefined, undefined, { + splitEnabled: true, + newClient: prisma17 as unknown as PrismaClient, + legacyReplica: prisma14, + }); + + const result = await presenter.call(callArgs(ctx, seeded.friendlyId)); + + expect(result?.id).toBe(seeded.friendlyId); + expect(result?.connectedRuns).toEqual([]); + } + ); + + // Cross-DB connection: waitpoint on LEGACY (PG14), the connected run + its WaitpointRunConnection + // join on the NEW dedicated DB (PG17). A single-DB gather off the waitpoint's own store misses the + // run entirely; the fix reads the join from BOTH stores (dedicated `waitpointRunConnection` + // delegate + legacy raw `_WaitpointRunConnections`) and unions the friendlyIds. + heteroRunOpsPostgresTest( + "gathers a cross-DB connected run whose join lives on the other database", + async ({ prisma14, prisma17 }) => { + const ctx = await seedParents(prisma14, "crossdb"); + const waitpoint = await seedWaitpoint(prisma14, ctx, "waitpoint_crossdb"); + + // The connected run + join live only on the NEW dedicated DB (co-resident with the run). + const run = await seedRun(prisma17, ctx, "run_crossnew"); + await prisma17.waitpointRunConnection.create({ + data: { taskRunId: run.id, waitpointId: waitpoint.id }, + }); + + legacyReplicaHolder.client = prisma14; + newClientHolder.client = prisma17; + + const presenter = new WaitpointPresenter(undefined, undefined, { + splitEnabled: true, + newClient: prisma17 as unknown as PrismaClient, + legacyReplica: prisma14, + }); + + const result = await presenter.call(callArgs(ctx, waitpoint.friendlyId)); + + expect(result?.id).toBe(waitpoint.friendlyId); + expect(result?.connectedRuns.map((r) => r.friendlyId)).toEqual(["run_crossnew"]); + } + ); + + // Same-DB legacy connection (no regression): waitpoint + connected run both on LEGACY, joined via + // the implicit `_WaitpointRunConnections` M2M. The gather must read the legacy raw join path. + heteroRunOpsPostgresTest( + "still gathers a same-DB legacy connected run via the implicit M2M", + async ({ prisma14, prisma17 }) => { + const ctx = await seedParents(prisma14, "legsame"); + const run = await seedRun(prisma14, ctx, "run_legsame"); + const waitpoint = await prisma14.waitpoint.create({ + data: { + friendlyId: "waitpoint_legsame", + type: "MANUAL", + status: "COMPLETED", + idempotencyKey: "idem-waitpoint_legsame", + userProvidedIdempotencyKey: false, + outputType: "application/json", + outputIsError: false, + completedAt: new Date(), + tags: [], + projectId: ctx.projectId, + environmentId: ctx.environmentId, + connectedRuns: { connect: [{ id: run.id }] }, + }, + }); + + legacyReplicaHolder.client = prisma14; + newClientHolder.client = prisma17; + + const presenter = new WaitpointPresenter(undefined, undefined, { + splitEnabled: true, + newClient: prisma17 as unknown as PrismaClient, + legacyReplica: prisma14, + }); + + const result = await presenter.call(callArgs(ctx, waitpoint.friendlyId)); + + expect(result?.connectedRuns.map((r) => r.friendlyId)).toEqual(["run_legsame"]); + } + ); +}); From 0a2ed96dc650a63716dc985a0131b5c9832623c7 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 23:06:37 +0100 Subject: [PATCH 15/16] fix(run-store): keep the caller transaction on the legacy leg of the edge-delete fan-out The taskRunId-keyed deleteManyTaskRunWaitpoints fan-out dropped the caller's transaction for both legs. The new (dedicated) leg can't join a control-plane transaction, but the legacy leg can, so pass it through: a legacy run's blocking edges are again deleted atomically with the caller's operation (e.g. an attempt failure) instead of auto-committing, matching the waitpointId-keyed path. --- .../PostgresRunStore.writeAtomicity.test.ts | 83 +++++++++++++++++++ .../run-store/src/runOpsStore.ts | 7 +- 2 files changed, 87 insertions(+), 3 deletions(-) diff --git a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts index 031d575a3e..94a7312c21 100644 --- a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts @@ -312,3 +312,86 @@ describe("cross-DB write atomicity (startAttempt + createExecutionSnapshot)", () } ); }); + +// A run's blocking edges may straddle both DBs mid-drain, so clearBlockingWaitpoints routes the +// taskRunId-keyed delete through the both-stores fan-out. The #new leg can't join a control-plane +// tx, but the #legacy leg CAN — so the caller's tx (e.g. attemptFailed) must still be honored for +// the legacy edges, keeping them atomic with the caller's operation instead of auto-committing. +async function seedLegacyBlockingEdge( + prisma14: PrismaClient, + env: { project: { id: string }; environment: { id: string } }, + runId: string, + suffix: string +): Promise { + const waitpoint = await prisma14.waitpoint.create({ + data: { + friendlyId: `wp_${suffix}`, + type: "MANUAL", + status: "PENDING", + idempotencyKey: `idem_${suffix}`, + userProvidedIdempotencyKey: false, + projectId: env.project.id, + environmentId: env.environment.id, + }, + }); + await prisma14.taskRunWaitpoint.create({ + data: { taskRunId: runId, waitpointId: waitpoint.id, projectId: env.project.id }, + }); +} + +describe("fan-out deleteManyTaskRunWaitpoints honors the caller's tx on the #legacy leg", () => { + heteroRunOpsPostgresTest( + "rolls the #legacy edge delete back when the caller's control-plane tx rolls back", + async ({ prisma14, prisma17 }) => { + const { router } = makeSplitRouter(prisma14, prisma17); + const env = await seedEnvironment(prisma14, "legacy", "del_tx_rb"); + const runId = `run_${CUID_25}`; + await router.createRun( + buildCreateRunInput({ + runId, + friendlyId: "run_del_tx_rb", + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + await seedLegacyBlockingEdge(prisma14, env, runId, "del_tx_rb"); + + await expect( + prisma14.$transaction(async (tx) => { + await router.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, tx); + throw new Error("rollback"); + }) + ).rejects.toThrow("rollback"); + + const remaining = await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId } }); + expect(remaining).toBe(1); + } + ); + + heteroRunOpsPostgresTest( + "still deletes the #legacy edge when the caller's tx commits", + async ({ prisma14, prisma17 }) => { + const { router } = makeSplitRouter(prisma14, prisma17); + const env = await seedEnvironment(prisma14, "legacy", "del_tx_commit"); + const runId = `run_${CUID_25}`; + await router.createRun( + buildCreateRunInput({ + runId, + friendlyId: "run_del_tx_commit", + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + await seedLegacyBlockingEdge(prisma14, env, runId, "del_tx_commit"); + + await prisma14.$transaction(async (tx) => { + await router.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, tx); + }); + + const remaining = await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId } }); + expect(remaining).toBe(0); + } + ); +}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 70967a1ab7..737e61e385 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -1302,11 +1302,12 @@ export class RoutingRunStore implements RunStore { const store = await this.#resolveWaitpointStore(waitpointId); return store.deleteManyTaskRunWaitpoints(args, store === this.#legacy ? tx : undefined); } - // Keyed by taskRunId (or other): a run's edges may straddle DBs mid-drain, so delete from - // both. Can't span one tx across two DBs, so it's dropped for the both-stores path. + // Keyed by taskRunId (or other): a run's edges may straddle DBs mid-drain, so delete from both. + // One tx can't span two DBs, so the #new leg is auto-commit; the caller's tx is control-plane, so + // the #legacy leg keeps it (atomic with the caller's op, matching the waitpointId-keyed path). const [fromNew, fromLegacy] = await Promise.all([ this.#new.deleteManyTaskRunWaitpoints(args), - this.#legacy.deleteManyTaskRunWaitpoints(args), + this.#legacy.deleteManyTaskRunWaitpoints(args, tx), ]); return { count: fromNew.count + fromLegacy.count }; } From 19c27594e76c0f9d46358fb807758ad20fd6bc5e Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 5 Jul 2026 23:16:07 +0100 Subject: [PATCH 16/16] fix(run-store): forward the caller transaction to the legacy store in createExecutionSnapshot RoutingRunStore.createExecutionSnapshot accepted a caller transaction but never forwarded it to the routed store. Forward it when the owning store is legacy so a legacy-resident snapshot stays atomic with the caller's operation; a new (cross-DB) write still cannot join a control-plane transaction, so it is dropped there and relies on runInTransaction for atomicity. --- .../PostgresRunStore.writeAtomicity.test.ts | 62 +++++++++++++++++++ .../run-store/src/runOpsStore.ts | 5 +- 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts index 94a7312c21..fb81eb90c1 100644 --- a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts @@ -395,3 +395,65 @@ describe("fan-out deleteManyTaskRunWaitpoints honors the caller's tx on the #leg } ); }); + +// RoutingRunStore.createExecutionSnapshot accepts a caller tx but must forward it to the OWNING store +// only when that store is #legacy: a control-plane tx can't wrap a #new (cross-DB) write, but it can +// (and should) wrap a legacy-resident snapshot so it stays atomic with the caller's operation. +describe("createExecutionSnapshot honors the caller's tx on the #legacy owning store", () => { + heteroRunOpsPostgresTest( + "rolls the snapshot back when a legacy run's caller tx rolls back", + async ({ prisma14, prisma17 }) => { + const { router } = makeSplitRouter(prisma14, prisma17); + const env = await seedEnvironment(prisma14, "legacy", "ces_rb"); + const runId = `run_${CUID_25}`; + await router.createRun( + buildCreateRunInput({ + runId, + friendlyId: "run_ces_rb", + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + + await expect( + prisma14.$transaction(async (tx) => { + await router.createExecutionSnapshot(snapshotInput(runId, env), tx); + throw new Error("rollback"); + }) + ).rejects.toThrow("rollback"); + + const snap = await prisma14.taskRunExecutionSnapshot.findFirst({ + where: { runId, executionStatus: "EXECUTING" }, + }); + expect(snap).toBeNull(); + } + ); + + heteroRunOpsPostgresTest( + "persists the snapshot when the legacy caller tx commits", + async ({ prisma14, prisma17 }) => { + const { router } = makeSplitRouter(prisma14, prisma17); + const env = await seedEnvironment(prisma14, "legacy", "ces_commit"); + const runId = `run_${CUID_25}`; + await router.createRun( + buildCreateRunInput({ + runId, + friendlyId: "run_ces_commit", + organizationId: env.organization.id, + projectId: env.project.id, + runtimeEnvironmentId: env.environment.id, + }) + ); + + await prisma14.$transaction(async (tx) => { + await router.createExecutionSnapshot(snapshotInput(runId, env), tx); + }); + + const snap = await prisma14.taskRunExecutionSnapshot.findFirst({ + where: { runId, executionStatus: "EXECUTING" }, + }); + expect(snap).not.toBeNull(); + } + ); +}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 737e61e385..102f56ea86 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -827,7 +827,10 @@ export class RoutingRunStore implements RunStore { input: CreateExecutionSnapshotInput, tx?: PrismaClientOrTransaction ): Promise> { - return (await this.#routeOrNewForWrite(input.run.id)).createExecutionSnapshot(input); + // Forward the caller's control-plane tx only to the #legacy store; a #new (cross-DB) write can't + // join it, so it's dropped there (the atomic #new path uses runInTransaction instead). + const store = await this.#routeOrNewForWrite(input.run.id); + return store.createExecutionSnapshot(input, store === this.#legacy ? tx : undefined); } // Snapshot ids are cuids (they always classify LEGACY), and a snapshot's CompletedWaitpoint join