diff --git a/server/src/__tests__/routines-service.test.ts b/server/src/__tests__/routines-service.test.ts index 7bdb69023d4..4770f843406 100644 --- a/server/src/__tests__/routines-service.test.ts +++ b/server/src/__tests__/routines-service.test.ts @@ -301,6 +301,23 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => { return { issue, runId, heartbeatRunId }; } + async function waitForBlockedQuery(...patterns: string[]) { + for (let attempt = 0; attempt < 80; attempt += 1) { + const [waiting] = await db.execute<{ waiting: boolean }>(sql` + select exists ( + select 1 + from pg_stat_activity + where state = 'active' + and wait_event_type = 'Lock' + and ${sql.join(patterns.map((pattern) => sql`query ilike ${pattern}`), sql` and `)} + ) as waiting + `); + if (waiting?.waiting) return true; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return false; + } + it("supersedes and links every older unfinished instance after an opted-in newer success", async () => { const { agentId, companyId, issueSvc, routine, svc } = await seedFixture(); const optedInRoutine = await svc.update(routine.id, { lifecyclePolicy: "latest_success_wins" }, {}); @@ -768,6 +785,90 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => { } }); + it("rejects a concurrent nested child when the supersession sweep locks the root first", async () => { + const { agentId, companyId, issueSvc, routine, svc } = await seedFixture(); + const optedInRoutine = await svc.update(routine.id, { lifecyclePolicy: "latest_success_wins" }, {}); + const revisionId = optedInRoutine!.latestRevisionId!; + const candidate = await createExecutionInstance({ + companyId, + agentId, + routine: optedInRoutine!, + issueSvc, + triggeredAt: new Date("2026-08-23T09:00:00.000Z"), + routineRevisionId: revisionId, + }); + const terminalChild = await issueSvc.create(companyId, { + projectId: routine.projectId, + parentId: candidate.issue.id, + title: "Completed direct child", + status: "done", + priority: "medium", + assigneeAgentId: agentId, + }); + const winner = await createExecutionInstance({ + companyId, + agentId, + routine: optedInRoutine!, + issueSvc, + triggeredAt: new Date("2026-08-23T10:00:00.000Z"), + routineRevisionId: revisionId, + }); + await issueSvc.update(winner.issue.id, { status: "done" }); + + let releaseCandidateLock!: () => void; + const candidateLockReleased = new Promise((resolve) => { + releaseCandidateLock = resolve; + }); + let markCandidateLocked!: () => void; + const candidateLocked = new Promise((resolve) => { + markCandidateLocked = resolve; + }); + const lockTransaction = db.transaction(async (tx) => { + await tx.execute(sql` + select ${issues.id} + from ${issues} + where ${issues.companyId} = ${companyId} + and ${issues.id} = ${candidate.issue.id} + for update + `); + markCandidateLocked(); + await candidateLockReleased; + }); + await candidateLocked; + + const sweepPromise = svc.syncRunStatusForIssue(winner.issue.id); + await expect(waitForBlockedQuery("%issues%", "%for update%")).resolves.toBe(true); + const grandchildPromise = issueSvc.create(companyId, { + projectId: routine.projectId, + parentId: terminalChild.id, + title: "Nested child created during supersession", + status: "in_progress", + priority: "medium", + assigneeAgentId: agentId, + }); + await expect(waitForBlockedQuery("%ancestor_ids%", "%for key share%")).resolves.toBe(true); + + releaseCandidateLock(); + await lockTransaction; + const [sweepOutcome, grandchildOutcome] = await Promise.allSettled([ + sweepPromise, + grandchildPromise, + ]); + + expect(sweepOutcome.status).toBe("fulfilled"); + expect(grandchildOutcome.status).toBe("rejected"); + await expect(issueSvc.getById(candidate.issue.id)).resolves.toMatchObject({ + status: "cancelled", + supersededByIssueId: winner.issue.id, + }); + await expect( + db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.parentId, terminalChild.id))), + ).resolves.toHaveLength(0); + }); + it("defers supersession when a concurrent direct child commits before the sweep lock", async () => { const { agentId, companyId, issueSvc, routine, svc } = await seedFixture(); const optedInRoutine = await svc.update(routine.id, { lifecyclePolicy: "latest_success_wins" }, {}); @@ -790,23 +891,6 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => { }); await issueSvc.update(winner.issue.id, { status: "done" }); - const waitForBlockedQuery = async (...patterns: string[]) => { - for (let attempt = 0; attempt < 80; attempt += 1) { - const [waiting] = await db.execute<{ waiting: boolean }>(sql` - select exists ( - select 1 - from pg_stat_activity - where state = 'active' - and wait_event_type = 'Lock' - and ${sql.join(patterns.map((pattern) => sql`query ilike ${pattern}`), sql` and `)} - ) as waiting - `); - if (waiting?.waiting) return true; - await new Promise((resolve) => setTimeout(resolve, 25)); - } - return false; - }; - let releaseCompanyLock!: () => void; const companyLockReleased = new Promise((resolve) => { releaseCompanyLock = resolve; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 483657e0db6..79a15cb8470 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -7142,25 +7142,45 @@ export function issueService(db: Db) { } if (issueData.parentId) { - // This conflicts with the supersession sweep's FOR UPDATE lock. When - // creation wins the lock order, it commits before the sweep proceeds; - // the sweep's following descendant query then sees the new child in - // its fresh READ COMMITTED snapshot. - const parent = await tx - .select({ - id: issues.id, - status: issues.status, - supersededByIssueId: issues.supersededByIssueId, - }) - .from(issues) - .where(and(eq(issues.companyId, companyId), eq(issues.id, issueData.parentId))) - .for("key share") - .then((rows) => rows[0] ?? null); + // Lock the complete ancestor chain so nested child creation also + // conflicts with a supersession sweep's root FOR UPDATE lock. When + // creation wins, it commits before the sweep proceeds and the + // sweep's fresh READ COMMITTED descendant snapshot sees the child. + const ancestors = Array.from(await tx.execute(sql` + with recursive ancestor_ids(id, parent_id, depth) as ( + select ${issues.id}, ${issues.parentId}, 0 + from ${issues} + where ${issues.companyId} = ${companyId} + and ${issues.id} = ${issueData.parentId} + union all + select ancestor_issue.id, ancestor_issue.parent_id, ancestor_ids.depth + 1 + from ${issues} ancestor_issue + join ancestor_ids on ancestor_issue.id = ancestor_ids.parent_id + where ancestor_issue.company_id = ${companyId} + ) + select ancestor_issue.id, + ancestor_issue.status as issue_status, + ancestor_issue.superseded_by_issue_id + from ${issues} ancestor_issue + join ancestor_ids on ancestor_ids.id = ancestor_issue.id + where ancestor_issue.company_id = ${companyId} + order by ancestor_ids.depth desc + for key share of ancestor_issue + `)) as Array<{ + id: string; + issue_status: string; + superseded_by_issue_id: string | null; + }>; + const parent = ancestors.find((ancestor) => ancestor.id === issueData.parentId) ?? null; if (!parent) throw notFound("Parent issue not found"); - if (parent.status === "cancelled" && parent.supersededByIssueId) { - throw conflict("Cannot create a child issue under a superseded issue", { + const supersededAncestor = ancestors.find( + (ancestor) => ancestor.issue_status === "cancelled" && ancestor.superseded_by_issue_id, + ); + if (supersededAncestor) { + throw conflict("Cannot create a descendant issue under a superseded issue", { parentIssueId: parent.id, - supersededByIssueId: parent.supersededByIssueId, + ancestorIssueId: supersededAncestor.id, + supersededByIssueId: supersededAncestor.superseded_by_issue_id, }); } }