diff --git a/server/src/__tests__/routines-service.test.ts b/server/src/__tests__/routines-service.test.ts index 4770f843406..e86d8997ee4 100644 --- a/server/src/__tests__/routines-service.test.ts +++ b/server/src/__tests__/routines-service.test.ts @@ -869,6 +869,80 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => { ).resolves.toHaveLength(0); }); + it("rejects a concurrent reparent 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 orphan = await issueSvc.create(companyId, { + projectId: routine.projectId, + title: "Live issue being reparented", + status: "in_progress", + 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 reparentPromise = issueSvc.update(orphan.id, { parentId: candidate.issue.id }); + await expect(waitForBlockedQuery("%ancestor_ids%", "%for key share%")).resolves.toBe(true); + + releaseCandidateLock(); + await lockTransaction; + const [sweepOutcome, reparentOutcome] = await Promise.allSettled([ + sweepPromise, + reparentPromise, + ]); + + expect(sweepOutcome.status).toBe("fulfilled"); + expect(reparentOutcome.status).toBe("rejected"); + await expect(issueSvc.getById(candidate.issue.id)).resolves.toMatchObject({ + status: "cancelled", + supersededByIssueId: winner.issue.id, + }); + await expect(issueSvc.getById(orphan.id)).resolves.toMatchObject({ + parentId: null, + status: "in_progress", + }); + }); + 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" }, {}); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 79a15cb8470..8723d4c9b31 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -4455,6 +4455,62 @@ export function issueService(db: Db) { return title.trim().replace(/\s+/g, " ").toLowerCase(); } + async function lockAndValidateIssueAncestors( + tx: any, + companyId: string, + parentId: string, + descendantIssueId?: string, + ) { + // Lock the complete ancestor chain so create and reparent operations both + // conflict with a supersession sweep's root FOR UPDATE lock. When this + // mutation 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} = ${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 === parentId) ?? null; + if (!parent) throw notFound("Parent issue not found"); + if (descendantIssueId && ancestors.some((ancestor) => ancestor.id === descendantIssueId)) { + throw unprocessable("Issue parent cannot be the issue or one of its descendants", { + issueId: descendantIssueId, + parentIssueId: parentId, + }); + } + const supersededAncestor = ancestors.find( + (ancestor) => ancestor.issue_status === "cancelled" && ancestor.superseded_by_issue_id, + ); + if (supersededAncestor) { + throw conflict("Cannot place a descendant issue under a superseded issue", { + parentIssueId: parent.id, + ancestorIssueId: supersededAncestor.id, + supersededByIssueId: supersededAncestor.superseded_by_issue_id, + }); + } + return ancestors; + } + async function getIssueByUuid(id: string) { const row = await db .select() @@ -7142,47 +7198,7 @@ export function issueService(db: Db) { } if (issueData.parentId) { - // 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"); - 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, - ancestorIssueId: supersededAncestor.id, - supersededByIssueId: supersededAncestor.superseded_by_issue_id, - }); - } + await lockAndValidateIssueAncestors(tx, companyId, issueData.parentId); } const defaultCompanyGoal = await getDefaultCompanyGoal(tx, companyId); @@ -7856,6 +7872,18 @@ export function issueService(db: Db) { .for("update") .then((rows: Array) => rows[0] ?? null); if (!receiptExisting) return null; + if ( + issueData.parentId !== undefined && + issueData.parentId !== receiptExisting.parentId && + issueData.parentId !== null + ) { + await lockAndValidateIssueAncestors( + tx, + receiptExisting.companyId, + issueData.parentId, + receiptExisting.id, + ); + } const [previousLabelsByIssueId, previousRelationSummaries] = await Promise.all([ nextLabelIds !== undefined ? labelMapForIssues(tx, [id])