diff --git a/doc/idle-approval-queue-seeding.md b/doc/idle-approval-queue-seeding.md index 14642e8ed97..b724522a12e 100644 --- a/doc/idle-approval-queue-seeding.md +++ b/doc/idle-approval-queue-seeding.md @@ -92,7 +92,7 @@ other source owners. - Before the fix, the PostgreSQL regression found no PR queue after the normal maintenance path excluded approvals. After the fix, the queue API finds it without a full attention read. -- Twenty real-database worker tests cover empty-to-active-to-quiet behavior, +- Twenty-one real-database worker tests cover empty-to-active-to-quiet behavior, startup and completed history, actual approval/link/work-product producers, batch import, first-link selection, company isolation, hidden targets, queue rule enablement, company activation, resubmission, outer commit and rollback, @@ -112,3 +112,9 @@ omitted required creator attribution, and a failure test expected no idle hold while committed projection work was still unresolved. The corrected test waits for the active attempt to finish and checks passive SQL directly while preserving that required hold. Neither required weakening the worker's recovery contract. + +Review found a queue/item lock inversion between seeded insertion and manual +membership changes. A two-connection PostgreSQL regression reproduced `deadlock +detected`. Manual add/remove now lock the queue before touching its items, matching +the seed writer. The regression and the existing queue route suite pass (27 cases). +Server typecheck and build pass after the repair. Fresh-head CI/review is required. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 968a2d6fe72..ca388a5f3b9 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -914,3 +914,20 @@ without preserving queue population. Other source seeding remains to be migrated threads. Its unchanged login-cleanup consume test failed in CI; all 16 local login-cleanup tests pass. One justified retry of failed run 40139410144 was requested. Do not mark that PR green until the retry completes successfully. + + +Slice 16 is [#17510](https://github.com/paperclipai/paperclip/pull/17510). +Greptile identified a queue/item lock-order inversion. A real PostgreSQL test +reproduced a deadlock; the common queue-before-item order repairs manual add/remove +and seeding. The worker and queue route suites pass 27 cases after the fix, with +server typecheck/build passing. Other selected consumers passed on the initial +head (136 unique local cases including the new regression). Fresh-head checks are +required; the initial head also hit an unchanged environment cleanup paging test. +Embedded PostgreSQL is unsupported on this host, so its first local run skipped; +this is not counted as a pass. + +#17508 now has all checks accepted at `9c5c4fbf65628589b466ce5650e8969388213716` +after its single justified retry. The duplicate cancelled workflows for #17490 +and #17497 were each rerun once to clear stale cancelled contexts. The #17490 +retry has a new E2E shard 3 failure under investigation; do not call it green or +rerun again without a repair. #17497's retry was still running at this observation. diff --git a/server/src/__tests__/approval-queue-seeding-work.test.ts b/server/src/__tests__/approval-queue-seeding-work.test.ts index 93a688b0afb..375dcb724b9 100644 --- a/server/src/__tests__/approval-queue-seeding-work.test.ts +++ b/server/src/__tests__/approval-queue-seeding-work.test.ts @@ -37,6 +37,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu afterEach(async () => { await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists approval_queue_delete_gate on decision_queue_items`); + await db.execute(sql`drop function if exists approval_queue_delete_gate()`); await db.execute(sql`drop trigger if exists approval_queue_fail on decision_queue_items`); await db.execute(sql`drop function if exists approval_queue_fail()`); await db.delete(decisionTriageEvents); await db.delete(decisionQueueItems); await db.delete(decisionQueues); @@ -290,4 +292,37 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await start(); await quiet(); expect(await memberships()).toHaveLength(2); }); + it("serializes manual removal and automatic seeding without a queue/item lock cycle", async () => { + const approval = await pending(); + const [link] = await db.select().from(issueApprovals).where(eq(issueApprovals.approvalId, approval.id)); + const queues = decisionQueueService(db); + const seed = () => queues.materializeSeedSources(companyId, [{ sourceKind: "approval", sourceId: approval.id, issueId: link.issueId }]); + await seed(); + await db.execute(sql`create function approval_queue_delete_gate() returns trigger language plpgsql as $$ begin + perform pg_advisory_xact_lock(hashtext(TG_ARGV[0])); return old; end $$`); + await db.execute(sql`create trigger approval_queue_delete_gate after delete on decision_queue_items for each row execute function approval_queue_delete_gate(${sql.raw("'" + companyId + "'")})`); + const blocker = await db.$client.reserve(); + await blocker`select pg_advisory_lock(hashtext(${companyId}))`; + let complete: Promise[]> | undefined; + try { + const removed = queues.removeItem({ ...mutation(), sourceKind: "approval", sourceId: approval.id }); + await vi.waitFor(async () => { + const [row] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and state = 'active' and wait_event = 'advisory'`); + expect(row.count).toBeGreaterThan(0); + }, { timeout: 5000 }); + complete = Promise.allSettled([removed, seed()]); + await vi.waitFor(async () => { + const [row] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and state = 'active' and wait_event_type = 'Lock' and wait_event <> 'advisory'`); + expect(row.count).toBeGreaterThan(0); + }, { timeout: 5000 }); + } finally { + await blocker`select pg_advisory_unlock_all()`; blocker.release(); + } + const results = await complete!; + expect(results.map(result => result.status), results.filter(result => result.status === "rejected").map(result => String(result.reason?.cause?.message ?? result.reason)).join("; ")).toEqual(["fulfilled", "fulfilled"]); + expect(await memberships()).toHaveLength(1); + }); + }); diff --git a/server/src/services/decision-queues.ts b/server/src/services/decision-queues.ts index db39daac1a7..1468d0444cc 100644 --- a/server/src/services/decision-queues.ts +++ b/server/src/services/decision-queues.ts @@ -376,6 +376,12 @@ export function decisionQueueService(db: Db) { .then((rows) => rows[0] ?? null); } + async function lockQueue(database: Db, companyId: string, id: string) { + const [queue] = await database.select({ id: decisionQueues.id }).from(decisionQueues) + .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.id, id))).for("update"); + if (!queue) throw notFound("Decision queue not found"); + } + async function visibleItems(companyId: string, queueId: string, authActor: AuthorizationActor) { const rows = await db.select().from(decisionQueueItems) .where(and(eq(decisionQueueItems.companyId, companyId), eq(decisionQueueItems.queueId, queueId))) @@ -600,6 +606,7 @@ export function decisionQueueService(db: Db) { await requireSourceRead(db, input.authActor, input.companyId, input.sourceKind, input.sourceId); return db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await lockQueue(txDb, input.companyId, queue.id); const inserted = await txDb.insert(decisionQueueItems).values({ companyId: input.companyId, queueId: queue.id, @@ -665,6 +672,7 @@ export function decisionQueueService(db: Db) { } return db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await lockQueue(txDb, input.companyId, queue.id); if (input.key === "prs" && input.sourceKind === "approval") await notifyApprovalQueueWork(txDb); const removed = await txDb.delete(decisionQueueItems).where(and( eq(decisionQueueItems.companyId, input.companyId),