fix(routines): serialize nested child creation

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

No files matched your search

+101 -17
View File
@@ -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<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 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<void>((resolve) => {
releaseCompanyLock = resolve;
+37 -17
View File
@@ -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,
});
}
}