fix(server): order queue membership locks consistently

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-24 16:06:59 -07:00
1 parent e04a0375dc
commit 4d15c08428
4 files changed
+67 -1

No files matched your search

+7 -1
View File
@@ -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.
@@ -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.
@@ -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<PromiseSettledResult<unknown>[]> | 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);
});
});
+8
View File
@@ -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),