Skip to content

Commit 2821b05

Browse files
d-csd-cs
committed
fix(run-store): write cross-DB waitpoint blocking edges via unnest
blockRunWithWaitpointEdges' legacy branch joined FROM "Waitpoint", so a LEGACY run blocking on a NEW-resident token (whose Waitpoint row lives on the run-ops database) matched no rows and wrote no edge, silently stranding the run. Source the edge rows from the id array via unnest, matching the dedicated branch. Two migrations drop the now-cross-DB TaskRunWaitpoint and _WaitpointRunConnections foreign keys to Waitpoint; integrity is app-enforced, matching the split's existing control-plane FK-removal pattern.
1 parent 9bf3749 commit 2821b05

4 files changed

Lines changed: 161 additions & 10 deletions

File tree

‎internal-packages/database/prisma/migrations/20260705210000_drop_waitpoint_run_connections_waitpoint_fk/migration.sql‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
-- Run-ops split: allow a cross-DB token connection (a LEGACY run blocking on a NEW-resident token,
2+
-- whose Waitpoint row lives on the other database). Matches the split's control-plane FK-removal pattern.
3+
ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey";

‎internal-packages/database/prisma/migrations/20260705220000_drop_task_run_waitpoint_waitpoint_fk/migration.sql‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
-- Run-ops split: drop the TaskRunWaitpoint -> Waitpoint FK so a LEGACY run's blocking edge can point
2+
-- at a NEW-resident (cross-DB) token. The #new dedicated schema is already FK-free here; this aligns
3+
-- #legacy. Referential integrity is app-enforced, matching the split's control-plane FK-removal.
4+
ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_waitpointId_fkey";

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1709,9 +1709,11 @@ export class PostgresRunStore implements RunStore {
17091709
return;
17101710
}
17111711

1712-
// Insert the blocking connections and the historical run connections.
1713-
// We use a CTE to do both inserts atomically. Data-modifying CTEs are
1714-
// always executed regardless of whether they're referenced in the outer query.
1712+
// Source edges from the id array via `unnest` (like the dedicated branch), NOT `FROM "Waitpoint"`:
1713+
// a cross-DB token (LEGACY run -> NEW-resident token) lives on the other DB, so the join matched 0
1714+
// rows and the run hung. Needs the _WaitpointRunConnections -> Waitpoint FK dropped (migration);
1715+
// casts are required because `unnest` gives no column types for the nullable params.
1716+
const ids = waitpointIds;
17151717
await prisma.$queryRaw`
17161718
WITH inserted AS (
17171719
INSERT INTO "TaskRunWaitpoint" ("id", "taskRunId", "waitpointId", "projectId", "createdAt", "updatedAt", "spanIdToComplete", "batchId", "batchIndex")
@@ -1722,19 +1724,17 @@ export class PostgresRunStore implements RunStore {
17221724
${projectId},
17231725
NOW(),
17241726
NOW(),
1725-
${spanIdToComplete ?? null},
1726-
${batchId ?? null},
1727-
${batchIndex ?? null}
1728-
FROM "Waitpoint" w
1729-
WHERE w.id IN (${Prisma.join(waitpointIds)})
1727+
${spanIdToComplete ?? null}::text,
1728+
${batchId ?? null}::text,
1729+
${batchIndex ?? null}::int
1730+
FROM unnest(${ids}::text[]) AS w(id)
17301731
ON CONFLICT DO NOTHING
17311732
RETURNING "waitpointId"
17321733
),
17331734
connected_runs AS (
17341735
INSERT INTO "_WaitpointRunConnections" ("A", "B")
17351736
SELECT ${runId}, w.id
1736-
FROM "Waitpoint" w
1737-
WHERE w.id IN (${Prisma.join(waitpointIds)})
1737+
FROM unnest(${ids}::text[]) AS w(id)
17381738
ON CONFLICT DO NOTHING
17391739
)
17401740
SELECT COUNT(*) FROM inserted`;

‎internal-packages/run-store/src/runOpsStore.crossDbTokenBlock.test.ts‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
// blockRunWithWaitpointEdges' legacy branch joined `FROM "Waitpoint" w`, so a LEGACY run blocking on
2+
// a NEW-resident (run-ops) token — whose Waitpoint row lives on #new — matched 0 rows and wrote NO
3+
// blocking edge. countPendingWaitpoints then fans out, still sees the token PENDING, so the run is
4+
// suspended believing it's blocked while nothing will ever resume it -> silent hang. The fix sources
5+
// the edge rows from the id array via `unnest` (FK-free, mirroring the dedicated branch); a migration
6+
// drops the _WaitpointRunConnections -> Waitpoint FK so the cross-DB connection can be recorded too.
7+
// Real two-DB topology; never mocked.
8+
9+
import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
10+
import type { PrismaClient } from "@trigger.dev/database";
11+
import type { RunOpsPrismaClient } from "@internal/run-ops-database";
12+
import { expect } from "vitest";
13+
import { PostgresRunStore } from "./PostgresRunStore.js";
14+
import { RoutingRunStore } from "./runOpsStore.js";
15+
16+
const CUID_25 = "c".repeat(25); // LEGACY run -> #legacy (prisma14, full schema)
17+
18+
function makeRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) {
19+
const newStore = new PostgresRunStore({
20+
prisma: prisma17 as never,
21+
readOnlyPrisma: prisma17 as never,
22+
schemaVariant: "dedicated",
23+
});
24+
const legacyStore = new PostgresRunStore({
25+
prisma: prisma14,
26+
readOnlyPrisma: prisma14,
27+
schemaVariant: "legacy",
28+
});
29+
return new RoutingRunStore({ new: newStore, legacy: legacyStore });
30+
}
31+
32+
async function seedEnvironmentLegacy(prisma: PrismaClient, suffix: string) {
33+
const organization = await prisma.organization.create({
34+
data: { title: `Org ${suffix}`, slug: `org-${suffix}` },
35+
});
36+
const project = await prisma.project.create({
37+
data: {
38+
name: `Project ${suffix}`,
39+
slug: `project-${suffix}`,
40+
externalRef: `proj_${suffix}`,
41+
organizationId: organization.id,
42+
},
43+
});
44+
const environment = await prisma.runtimeEnvironment.create({
45+
data: {
46+
type: "DEVELOPMENT",
47+
slug: "dev",
48+
projectId: project.id,
49+
organizationId: organization.id,
50+
apiKey: `tr_dev_${suffix}`,
51+
pkApiKey: `pk_dev_${suffix}`,
52+
shortcode: `short_${suffix}`,
53+
},
54+
});
55+
return { organization, project, environment };
56+
}
57+
58+
function taskRunData(opts: {
59+
id: string;
60+
organizationId: string;
61+
projectId: string;
62+
runtimeEnvironmentId: string;
63+
}) {
64+
return {
65+
id: opts.id,
66+
engine: "V2" as const,
67+
status: "EXECUTING" as const,
68+
friendlyId: `run_${opts.id}`,
69+
runtimeEnvironmentId: opts.runtimeEnvironmentId,
70+
environmentType: "DEVELOPMENT" as const,
71+
organizationId: opts.organizationId,
72+
projectId: opts.projectId,
73+
taskIdentifier: "xdb-task",
74+
payload: "{}",
75+
payloadType: "application/json",
76+
traceContext: {},
77+
traceId: `trace_${opts.id}`,
78+
spanId: `span_${opts.id}`,
79+
queue: "task/xdb-task",
80+
isTest: false,
81+
taskEventStore: "taskEvent",
82+
depth: 0,
83+
};
84+
}
85+
86+
describe("run-ops split — a LEGACY run blocking on a NEW-resident token gets its blocking edge (cross-DB)", () => {
87+
heteroRunOpsPostgresTest(
88+
"blockRunWithWaitpointEdges writes the edge for a cross-DB token instead of stranding the run",
89+
async ({ prisma14, prisma17 }: { prisma14: PrismaClient; prisma17: RunOpsPrismaClient }) => {
90+
const router = makeRouter(prisma14, prisma17);
91+
// The harness builds the schema with `prisma db push`, which re-creates the FKs that the
92+
// run-ops split migrations (20260705210000 / 20260705220000) drop in prod. Mirror those drops on
93+
// this clone so the cross-DB insert exercises the real prod state (FKs gone, app-enforced).
94+
await prisma14.$executeRawUnsafe(
95+
`ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_waitpointId_fkey"`
96+
);
97+
await prisma14.$executeRawUnsafe(
98+
`ALTER TABLE "_WaitpointRunConnections" DROP CONSTRAINT IF EXISTS "_WaitpointRunConnections_B_fkey"`
99+
);
100+
const seed = await seedEnvironmentLegacy(prisma14, "xdbblock");
101+
const runId = `run_${CUID_25}`; // LEGACY run, resident on #legacy
102+
await prisma14.taskRun.create({
103+
data: taskRunData({
104+
id: runId,
105+
organizationId: seed.organization.id,
106+
projectId: seed.project.id,
107+
runtimeEnvironmentId: seed.environment.id,
108+
}),
109+
});
110+
111+
// The token was minted co-located with a run-ops run, so its Waitpoint row lives on #new.
112+
const tokenId = "waitpoint_" + "x".repeat(20);
113+
await prisma17.waitpoint.create({
114+
data: {
115+
id: tokenId,
116+
friendlyId: "wp_xdb",
117+
type: "MANUAL",
118+
status: "PENDING",
119+
idempotencyKey: `idem_${tokenId}`,
120+
userProvidedIdempotencyKey: false,
121+
projectId: seed.project.id,
122+
environmentId: seed.environment.id,
123+
},
124+
});
125+
126+
await router.blockRunWithWaitpointEdges({
127+
runId,
128+
waitpointIds: [tokenId],
129+
projectId: seed.project.id,
130+
});
131+
132+
// RED: the legacy `FROM "Waitpoint"` join matches 0 rows -> no edge -> the run is stranded.
133+
// GREEN: `unnest` writes the edge from the id directly.
134+
expect(
135+
await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId, waitpointId: tokenId } })
136+
).toBe(1);
137+
// The historical connection is recorded too (needs the dropped B-fkey migration).
138+
const conn = (await prisma14.$queryRaw`
139+
SELECT "B" FROM "_WaitpointRunConnections" WHERE "A" = ${runId}
140+
`) as unknown[];
141+
expect(conn.length).toBe(1);
142+
}
143+
);
144+
});

0 commit comments

Comments
 (0)