From f101983a70dd510a7b6cf962cf2a6a70dcf0a72a Mon Sep 17 00:00:00 2001 From: Daniel Sutton <45313566+d-cs@users.noreply.github.com> Date: Sun, 5 Jul 2026 23:39:40 +0100 Subject: [PATCH] fix(run-store,run-engine): fix run-ops split hangs from wrong-store reads on the resume path (#4163) ## Summary On the run-ops split, NEW-residency runs could hang. Time-based waits (`wait.for`, `wait.until`, `delay`, waitpoint tokens), `batchTriggerAndWait`, and attempt starts stalled and never resumed. Each was a run-ops read or update that hit the wrong database: either the owning store's read replica when it needed read-your-writes, or the wrong store entirely because it routed by an id that does not encode residency. ## Fixes **Waitpoint resume (the main hang).** The managed resume path reads a run's completed waitpoints by snapshot id (`findSnapshotCompletedWaitpointIds`). Snapshot ids are cuids, which always classify to the legacy store, so a NEW run's join rows (which live on the new store) were never found. The resumed run saw zero completed waitpoints and hung. It now fans out across both stores and merges, like its sibling readers. **Batch completion.** Batch item completion (`updateManyBatchTaskRunItems`) routed by the item id, which is also a cuid, so a NEW batch's items were updated on the wrong store, matched zero rows, and the batch was treated as already complete (its parent's `batchTriggerAndWait` then hung). It now routes by the batch id, which does encode residency, matching the sibling `countBatchTaskRunItems`. **Read-your-writes on the resume path.** The block-time pending-waitpoint check (`countPendingWaitpoints`) and the attempt-start lock check (`findRun` in `startRunAttempt`) both read the owning store's replica with no read-your-writes guarantee, so a just-committed waitpoint completion or dequeue lock could be missed under replica lag and strand the run. Both now read the owning primary. Each fix ships with a two-database store or engine test that reproduces the hang and passes with the fix. --- .../app/presenters/v3/SpanPresenter.server.ts | 7 +- .../v3/WaitpointPresenter.server.ts | 68 ++++- .../routes/api.v1.runs.$runParam.result.ts | 7 +- ...dedicatedConnectedRuns.readthrough.test.ts | 253 ++++++++++++++++ .../migration.sql | 7 + .../migration.sql | 8 + .../migration.sql | 8 + .../src/engine/systems/runAttemptSystem.ts | 4 +- .../src/engine/systems/waitpointSystem.ts | 24 +- .../clearBlockingWaitpointsResidency.test.ts | 182 +++++++++++ .../tests/getSnapshotsSinceResidency.test.ts | 161 ++++++++++ .../startRunAttemptReadResidency.test.ts | 228 ++++++++++++++ .../engine/tests/waitpointReplicaLag.test.ts | 285 ++++++++++++++++++ .../run-store/src/PostgresRunStore.ts | 76 +++-- .../PostgresRunStore.writeAtomicity.test.ts | 145 +++++++++ .../src/runOpsStore.batchItemMisroute.test.ts | 95 ++++++ ...OpsStore.crossDbCompletedWaitpoint.test.ts | 162 ++++++++++ .../src/runOpsStore.crossDbTokenBlock.test.ts | 144 +++++++++ ...sStore.snapshotCompletedWaitpoints.test.ts | 47 +++ .../run-store/src/runOpsStore.ts | 43 ++- 20 files changed, 1884 insertions(+), 70 deletions(-) create mode 100644 apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts 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/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql create mode 100644 internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts create mode 100644 internal-packages/run-engine/src/engine/tests/getSnapshotsSinceResidency.test.ts create mode 100644 internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts create mode 100644 internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts create mode 100644 internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts create mode 100644 internal-packages/run-store/src/runOpsStore.crossDbCompletedWaitpoint.test.ts create mode 100644 internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts create mode 100644 internal-packages/run-store/src/runOpsStore.snapshotCompletedWaitpoints.test.ts diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index ea2a89eaa..d5d1c8c66 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, diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts index a56c5e76a..74f0c06fd 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/app/routes/api.v1.runs.$runParam.result.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts index 4cbf27d32..7444f9b55 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) { 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 000000000..a71a47d53 --- /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"]); + } + ); +}); 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 000000000..aca9ceb1b --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql @@ -0,0 +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 new file mode 100644 index 000000000..640de2d70 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql @@ -0,0 +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 new file mode 100644 index 000000000..bb4ef3577 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260705230000_drop_completed_waitpoints_waitpoint_fk/migration.sql @@ -0,0 +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"; diff --git a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts index a384d37f8..7f53c7648 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/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 344a0c1d8..948346fca 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; } @@ -511,8 +506,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/clearBlockingWaitpointsResidency.test.ts b/internal-packages/run-engine/src/engine/tests/clearBlockingWaitpointsResidency.test.ts new file mode 100644 index 000000000..f989beb47 --- /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(); + } + } + ); +}); 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 000000000..adaefaed2 --- /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]); + } + ); +}); 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 000000000..d3ade923c --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/startRunAttemptReadResidency.test.ts @@ -0,0 +1,228 @@ +// 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; + }; + } + // 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) { + 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 }; +} + +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(); + } + } + ); +}); 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 000000000..6cbf3c332 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/waitpointReplicaLag.test.ts @@ -0,0 +1,285 @@ +// 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); + }; + } + // 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 }; +} + +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(); + } + } + ); +}); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index b797fae9b..94630cdb2 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: RunOpsCapableClient, + 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; @@ -1709,9 +1719,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 +1734,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/PostgresRunStore.writeAtomicity.test.ts b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts index 031d575a3..fb81eb90c 100644 --- a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts @@ -312,3 +312,148 @@ 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); + } + ); +}); + +// 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.batchItemMisroute.test.ts b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts new file mode 100644 index 000000000..b59b711bd --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.batchItemMisroute.test.ts @@ -0,0 +1,95 @@ +// 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"); + // Legacy was never touched: no phantom/double-routed item update on #legacy. + expect(await prisma14.batchTaskRunItem.count({ where: { batchTaskRunId: batchId } })).toBe(0); + } + ); +}); 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 000000000..c500c1d93 --- /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]); + } + ); +}); 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 000000000..071a37a2a --- /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); + } + ); +}); 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 000000000..b0a5a99e2 --- /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 9b7f45383..102f56ea8 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -827,16 +827,30 @@ 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); } - // 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 @@ -1291,11 +1305,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 }; } @@ -1484,9 +1499,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);