mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 02:00:52 +01:00
test(routines): prove child-first sweep ordering
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
0cb2f1ee02
commit
4b74e4047d
3 files changed
+91
No files matched your search
@@ -768,6 +768,90 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
||||
}
|
||||
});
|
||||
|
||||
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" }, {});
|
||||
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 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" });
|
||||
|
||||
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;
|
||||
});
|
||||
let markCompanyLocked!: () => void;
|
||||
const companyLocked = new Promise<void>((resolve) => {
|
||||
markCompanyLocked = resolve;
|
||||
});
|
||||
const lockTransaction = db.transaction(async (tx) => {
|
||||
await tx.select({ id: companies.id }).from(companies).where(eq(companies.id, companyId)).for("update");
|
||||
markCompanyLocked();
|
||||
await companyLockReleased;
|
||||
});
|
||||
await companyLocked;
|
||||
|
||||
const childPromise = issueSvc.create(companyId, {
|
||||
projectId: routine.projectId,
|
||||
parentId: candidate.issue.id,
|
||||
title: "Child that wins the parent lock",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await expect(waitForBlockedQuery("%update%companies%")).resolves.toBe(true);
|
||||
|
||||
const sweepPromise = svc.syncRunStatusForIssue(winner.issue.id);
|
||||
await expect(waitForBlockedQuery("%issues%", "%for update%")).resolves.toBe(true);
|
||||
|
||||
releaseCompanyLock();
|
||||
await lockTransaction;
|
||||
const [child] = await Promise.all([childPromise, sweepPromise]);
|
||||
|
||||
await expect(issueSvc.getById(candidate.issue.id)).resolves.toMatchObject({
|
||||
status: "todo",
|
||||
supersededByIssueId: null,
|
||||
});
|
||||
await expect(issueSvc.getById(child.id)).resolves.toMatchObject({
|
||||
parentId: candidate.issue.id,
|
||||
status: "in_progress",
|
||||
});
|
||||
await expect(
|
||||
db.select().from(activityLog).where(eq(activityLog.action, "issue.supersession_deferred")),
|
||||
).resolves.toHaveLength(1);
|
||||
});
|
||||
|
||||
it("ignores descendants from another company during the supersession sweep", async () => {
|
||||
const firstCompany = await seedFixture();
|
||||
const secondCompany = await seedFixture();
|
||||
|
||||
@@ -7142,6 +7142,10 @@ 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,
|
||||
|
||||
@@ -1679,6 +1679,9 @@ export function routineService(
|
||||
sql`, `,
|
||||
);
|
||||
const liveStatusValues = sql.join(OPEN_ISSUE_STATUSES.map((status) => sql`${status}`), sql`, `);
|
||||
// Keep this as a statement after the candidate locks. Under READ
|
||||
// COMMITTED, a child transaction that held KEY SHARE first must commit
|
||||
// before those locks are granted, and this new snapshot sees that child.
|
||||
const liveDescendants = Array.from(await txDb.execute(sql`
|
||||
with recursive roots(root_id) as (
|
||||
values ${rootValues}
|
||||
|
||||
Reference in new issue
Block a user