fix(routines): guard concurrent issue reparenting

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-23 09:52:23 +00:00
1 parent df08868aab
commit 38fe025b16
2 files changed
+143 -41

No files matched your search

@@ -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<void>((resolve) => {
releaseCandidateLock = resolve;
});
let markCandidateLocked!: () => void;
const candidateLocked = new Promise<void>((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" }, {});
+69 -41
View File
@@ -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<typeof issues.$inferSelect>) => 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])