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, 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/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) { 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"]); + } + ); +}); 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..aca9ceb1ba --- /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 0000000000..640de2d709 --- /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 0000000000..bb4ef35774 --- /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 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/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 344a0c1d8c..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; } @@ -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 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(); + } + } + ); +}); 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]); + } + ); +}); 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..d3ade923cf --- /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 0000000000..6cbf3c3322 --- /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 b797fae9b9..94630cdb2c 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 031d575a3e..fb81eb90c1 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 0000000000..b59b711bd8 --- /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 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]); + } + ); +}); 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); + } + ); +}); 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..102f56ea86 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);