Skip to content

Commit dbf6dfa

Browse files
pcncTrigger.dev RepoOps
authored andcommitted
fix(webapp): stop concurrent task identifier syncs from deadlocking
Fix a deadlock when two task identifier syncs for the same environment run at the same time, for example two `trigger dev` sessions or a deployment promote racing a rollback. Syncs for one environment now run one after another. Mono-RevId: 85fa5ce67a9fff44d5e07575369fada1e5215206
1 parent 2a12eaf commit dbf6dfa

3 files changed

Lines changed: 177 additions & 1 deletion

File tree

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: fix
4+
---
5+
6+
Fix the dashboard's task list and task filters occasionally going stale when several `trigger dev` sessions register at the same time.

‎apps/webapp/app/services/taskIdentifierRegistry.server.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,10 @@ export async function syncTaskIdentifiers(
4545

4646
// Batch: insert new rows, update existing rows per source group, archive removed tasks
4747
await db.$transaction([
48+
// Serialize syncs per environment. Each sync locks its own tasks and then
49+
// archives everything else, so two concurrent syncs with different task
50+
// lists can each hold rows the other needs and deadlock.
51+
db.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${`task_identifier_sync:${environmentId}`}))`,
4852
// Insert any new task identifiers (skips rows that already exist)
4953
db.taskIdentifier.createMany({
5054
data: tasks.map((task) => ({

‎apps/webapp/test/engine/taskIdentifierRegistry.test.ts‎

Lines changed: 167 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { setTimeout } from "node:timers/promises";
12
import { describe, expect, vi } from "vitest";
23

34
vi.mock("~/db.server", () => ({
@@ -30,14 +31,119 @@ import type { AuthenticatedEnvironment } from "@internal/run-engine/tests";
3031

3132
vi.setConfig({ testTimeout: 30_000 });
3233

34+
// Polls until `count` backends in this test's database are blocked on a lock.
35+
async function waitForLockWaiters(prisma: PrismaClient, count: number, what: string) {
36+
const deadline = performance.now() + 5_000;
37+
let waiting = 0;
38+
while (performance.now() < deadline) {
39+
const [row] = await prisma.$queryRaw<{ n: number }[]>`
40+
SELECT count(*)::int AS n FROM pg_stat_activity
41+
WHERE datname = current_database() AND wait_event_type = 'Lock'`;
42+
waiting = row.n;
43+
if (waiting >= count) return;
44+
await setTimeout(20);
45+
}
46+
expect(waiting, `${what} never blocked on a lock`).toBeGreaterThanOrEqual(count);
47+
}
48+
49+
type SyncCall = { workerId: string; tasks: string[] };
50+
51+
// Runs two syncs in a fixed interleaving. A separate transaction holds the row
52+
// lock on `gateSlug`; `first` starts and blocks (directly or not) on it, then
53+
// `second` starts and blocks, then the gate is released. Returns the error of
54+
// each sync that failed.
55+
async function runSyncsBehindRowGate(
56+
prisma: PrismaClient,
57+
env: AuthenticatedEnvironment,
58+
gateSlug: string,
59+
first: SyncCall,
60+
second: SyncCall
61+
): Promise<string[]> {
62+
const sync = ({ workerId, tasks }: SyncCall) =>
63+
Promise.allSettled([
64+
syncTaskIdentifiers(
65+
env.id,
66+
env.project.id,
67+
workerId,
68+
tasks.map((id) => ({ id })),
69+
prisma
70+
),
71+
]);
72+
73+
let releaseGate!: () => void;
74+
const gateReleased = new Promise<void>((resolve) => (releaseGate = resolve));
75+
let gateHeld!: () => void;
76+
const gateHeldPromise = new Promise<void>((resolve) => (gateHeld = resolve));
77+
const gate = prisma.$transaction(
78+
async (tx) => {
79+
await tx.$queryRaw`
80+
SELECT id FROM "TaskIdentifier"
81+
WHERE "runtimeEnvironmentId" = ${env.id} AND slug = ${gateSlug}
82+
FOR UPDATE`;
83+
gateHeld();
84+
await gateReleased;
85+
},
86+
// Shorter than the test timeout, so a hung test still closes the gate.
87+
{ timeout: 15_000 }
88+
);
89+
90+
let firstResult: ReturnType<typeof sync> | undefined;
91+
let secondResult: ReturnType<typeof sync> | undefined;
92+
try {
93+
await Promise.race([gateHeldPromise, gate]);
94+
95+
firstResult = sync(first);
96+
await waitForLockWaiters(prisma, 1, "first sync");
97+
98+
secondResult = sync(second);
99+
await waitForLockWaiters(prisma, 2, "second sync");
100+
101+
releaseGate();
102+
await gate;
103+
104+
const results = (await Promise.all([firstResult, secondResult])).flat();
105+
return results
106+
.filter((r): r is PromiseRejectedResult => r.status === "rejected")
107+
.map((r) => String(r.reason));
108+
} finally {
109+
releaseGate();
110+
await gate.catch(() => {});
111+
// Don't leave a sync running into fixture teardown if a wait above failed.
112+
await Promise.all([firstResult, secondResult]);
113+
}
114+
}
115+
116+
async function seedSyncedTasks(
117+
prisma: PrismaClient,
118+
env: AuthenticatedEnvironment,
119+
slugs: string[]
120+
) {
121+
const worker = await createWorker(prisma, env);
122+
await syncTaskIdentifiers(
123+
env.id,
124+
env.project.id,
125+
worker.id,
126+
slugs.map((id) => ({ id })),
127+
prisma
128+
);
129+
}
130+
131+
async function taskIdentifierState(prisma: PrismaClient, env: AuthenticatedEnvironment) {
132+
const rows = await prisma.taskIdentifier.findMany({
133+
where: { runtimeEnvironmentId: env.id },
134+
orderBy: { slug: "asc" },
135+
});
136+
return rows.map((r) => [r.slug, r.isInLatestDeployment, r.currentWorkerId]);
137+
}
138+
33139
async function createWorker(prisma: PrismaClient, env: AuthenticatedEnvironment) {
34140
return prisma.backgroundWorker.create({
35141
data: {
36142
friendlyId: generateFriendlyId("worker"),
37143
contentHash: `hash-${Date.now()}-${Math.random()}`,
38144
projectId: env.project.id,
39145
runtimeEnvironmentId: env.id,
40-
version: `${Date.now()}`,
146+
version: `${Date.now()}-${Math.random()}`,
41147
metadata: {},
42148
engine: "V2",
43149
},
@@ -268,4 +374,64 @@ describe("TaskIdentifierRegistry", () => {
268374
expect(rows[1].slug).toBe("standard-task");
269375
expect(rows[1].currentTriggerSource).toBe("STANDARD");
270376
});
377+
378+
// Two dev sessions (or a promote racing a rollback) sync the same environment
379+
// with different task lists. Each sync updates its own tasks and then archives
380+
// everything else, so each can hold rows the other's next statement needs.
381+
postgresTest(
382+
"should not deadlock when concurrent syncs update and archive each other's tasks",
383+
async ({ prisma }) => {
384+
const env = await setupAuthenticatedEnvironment(prisma, "DEVELOPMENT");
385+
await seedSyncedTasks(prisma, env, ["task-a", "task-b", "task-c"]);
386+
const firstWorker = await createWorker(prisma, env);
387+
const secondWorker = await createWorker(prisma, env);
388+
389+
// The first sync locks a and waits on c. The second locks b and queues
390+
// behind the first for c. When the gate opens, the first takes c and its
391+
// archive then needs b, held by the second.
392+
const failures = await runSyncsBehindRowGate(
393+
prisma,
394+
env,
395+
"task-c",
396+
{ workerId: firstWorker.id, tasks: ["task-a", "task-c"] },
397+
{ workerId: secondWorker.id, tasks: ["task-b", "task-c"] }
398+
);
399+
400+
expect(failures, failures.join("\n")).toEqual([]);
401+
// The first sync commits, then the second runs, so its task list wins.
402+
expect(await taskIdentifierState(prisma, env)).toEqual([
403+
["task-a", false, firstWorker.id],
404+
["task-b", true, secondWorker.id],
405+
["task-c", true, secondWorker.id],
406+
]);
407+
}
408+
);
409+
410+
postgresTest(
411+
"should not deadlock when a sync's archive step waits on a concurrent sync",
412+
async ({ prisma }) => {
413+
const env = await setupAuthenticatedEnvironment(prisma, "DEVELOPMENT");
414+
await seedSyncedTasks(prisma, env, ["task-a", "task-b", "task-c"]);
415+
const firstWorker = await createWorker(prisma, env);
416+
const secondWorker = await createWorker(prisma, env);
417+
418+
// The first sync locks a and waits on b. The second locks c, then its
419+
// archive waits on a. When the gate opens, the first takes b and then
420+
// needs c, held by the second.
421+
const failures = await runSyncsBehindRowGate(
422+
prisma,
423+
env,
424+
"task-b",
425+
{ workerId: firstWorker.id, tasks: ["task-a", "task-b", "task-c"] },
426+
{ workerId: secondWorker.id, tasks: ["task-c"] }
427+
);
428+
429+
expect(failures, failures.join("\n")).toEqual([]);
430+
expect(await taskIdentifierState(prisma, env)).toEqual([
431+
["task-a", false, firstWorker.id],
432+
["task-b", false, firstWorker.id],
433+
["task-c", true, secondWorker.id],
434+
]);
435+
}
436+
);
271437
});

0 commit comments

Comments
 (0)