diff --git a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md index 1c691c19509..c22c826f22e 100644 --- a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -104,6 +104,20 @@ The full existing 17-test file passes with these failure controls. Production configuration validation remains unchanged. Current-head required CI remains the completion gate. +## Cancellation during acceptance + +The registered worker passes its optional cancellation signal into the canonical +confirmation acceptance transaction. It rechecks after the issue lock, before +mutation and before commit. Cancellation while an awaited write is blocked rolls +back the interaction, audit and transactional delivery effects. Manual callers +without a signal retain their existing acceptance and rollback behavior. + +A PostgreSQL lock wait itself remains unabortable by this signal: stop waits for +the query to unwind. Once the lock is released, cancelled acceptance leaves the +card pending for restart. The existing HTTP request timeout and prompt stop abort +remain unchanged. Actual lock and write-trigger controls verify pending state, +no post-stop wake, and exactly one acceptance/audit/wake on restart. + ## Existing limitations outside this migration Direct non-chat continuation still uses the existing postcommit `wakeup` call. diff --git a/server/src/__tests__/merged-pr-confirmation-work.test.ts b/server/src/__tests__/merged-pr-confirmation-work.test.ts index 95565fd96e7..60d1a5ffcb7 100644 --- a/server/src/__tests__/merged-pr-confirmation-work.test.ts +++ b/server/src/__tests__/merged-pr-confirmation-work.test.ts @@ -22,7 +22,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let coordinator: ReturnType | undefined; let draining = false, standby = false; const errors: unknown[] = []; - const provider = vi.fn(async (_companyId: string, _reference: { number: number }): Promise<"open" | "merged" | "unknown"> => "merged"); + const provider = vi.fn(async (_companyId: string, _reference: { number: number }, _signal?: AbortSignal): Promise<"open" | "merged" | "unknown"> => "merged"); const wakeup = vi.fn(async () => ({ id: "simulated-durable-wake" })); beforeAll(async () => { if (!external) temporary = await startEmbeddedPostgresTestDatabase("merged-pr-work-"); @@ -248,6 +248,84 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(errors).toEqual([]); }); + it("preserves manual acceptance without cancellation and rolls back a failed caller mutation", async () => { + const card = await create(); + const issue = { id: issueId, companyId, projectId: null, goalId: null }; + await expect(service().acceptInteraction(issue, card.id, {}, { userId: "board" }, { + afterResolveInTransaction: async () => { throw new Error("manual acceptance rollback"); }, + })).rejects.toThrow("manual acceptance rollback"); + expect((await stored(card.id)).status).toBe("pending"); + const accepted = await service().acceptInteraction(issue, card.id, {}, { userId: "board" }); + expect(accepted.interaction.status).toBe("accepted"); + expect((await stored(card.id)).status).toBe("accepted"); + }); + + it.each(["issue lock", "acceptance write"] as const)("keeps a stopped %s acceptance pending and accepts once after restart", async phase => { + vi.useRealTimers(); + const card = await create(); + let signal: AbortSignal | undefined; + provider.mockImplementation(async (_companyId, _reference, currentSignal) => { signal = currentSignal; return "merged"; }); + let reserved: Awaited> | undefined = await db.$client.reserve(); + const [{ pid }] = await reserved<{ pid: number }[]>`select pg_backend_pid() as pid`; + if (phase === "issue lock") { + await reserved.unsafe("begin"); + await reserved`select id from issues where id=${issueId} for update`; + } else { + await db.$client.unsafe(`create function maintenance_hold_merge_acceptance() returns trigger language plpgsql as $$begin + if new.id = '${card.id}'::uuid and new.status = 'accepted' then perform pg_advisory_xact_lock(930582); end if; return new; end$$`); + await db.$client.unsafe(`create trigger maintenance_hold_merge_acceptance before update on issue_thread_interactions for each row execute function maintenance_hold_merge_acceptance()`); + await reserved`select pg_advisory_lock(930582)`; + } + let stopping: Promise | undefined; + try { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = registerMergeConfirmationWork(coordinator, service(), () => true); + let blocked: { query: string } | undefined; + for (let attempt = 0; attempt < 200; attempt++) { + [blocked] = await db.$client<{ query: string }[]>`select query from pg_stat_activity + where datname=current_database() and wait_event_type='Lock' and ${pid} = any(pg_blocking_pids(pid))`; + if (blocked) break; + await delay(10); + } + expect(blocked?.query).toContain(phase === "issue lock" ? '"issues"' : '"issue_thread_interactions"'); + let stopped = false; + stopping = coordinator.stop().then(() => { stopped = true; }); + await delay(20); + expect(signal?.aborted).toBe(true); + expect(stopped).toBe(false); + expect((await stored(card.id)).status).toBe("pending"); + const releasedAt = performance.now(); + if (phase === "issue lock") await reserved.unsafe("commit"); + else await reserved`select pg_advisory_unlock(930582)`; + reserved.release(); reserved = undefined; + await stopping; await worker.ready; coordinator = undefined; + expect(performance.now() - releasedAt).toBeLessThan(1000); + expect((await stored(card.id)).status).toBe("pending"); + expect(wakeup).not.toHaveBeenCalled(); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(0); + expect(await db.select().from(chatActions)).toHaveLength(0); + await start(); + expect((await stored(card.id)).status).toBe("accepted"); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + } finally { + if (reserved) { + if (phase === "issue lock") await reserved.unsafe("rollback"); + else await reserved`select pg_advisory_unlock(930582)`; + reserved.release(); + } + await stopping; + await coordinator?.stop(); coordinator = undefined; + if (phase === "acceptance write") { + await db.$client.unsafe("drop trigger if exists maintenance_hold_merge_acceptance on issue_thread_interactions"); + await db.$client.unsafe("drop function if exists maintenance_hold_merge_acceptance()"); + } + } + }); + it("awaits an in-flight observation at shutdown and reconstructs its unaccepted card on restart", async () => { let release!: () => void, observed!: () => void; const barrier = new Promise(resolve => { release = resolve; }); diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index e69b38d45b9..194cddbfee9 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -2278,10 +2278,12 @@ export function issueThreadInteractionService( input: AcceptIssueThreadInteraction; actor: InteractionActor; mutationOptions?: InteractionResolutionMutationOptions; + signal?: AbortSignal; }): Promise<{ interaction: IssueThreadInteraction; continuationIssue: IssueWakeTarget | null; }> { + args.signal?.throwIfAborted(); const expired = await expireStaleRequestConfirmationTarget(db, { row: args.current, actor: args.actor, @@ -2324,6 +2326,7 @@ export function issueThreadInteractionService( .for("update") .then((rows: IssueResolutionContext[]) => rows[0] ?? null); + args.signal?.throwIfAborted(); if (!issueContext || issueContext.companyId !== args.issue.companyId) { throw notFound("Issue not found"); } @@ -2376,6 +2379,7 @@ export function issueThreadInteractionService( : undefined; if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]); + args.signal?.throwIfAborted(); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2547,6 +2551,7 @@ export function issueThreadInteractionService( }); } + args.signal?.throwIfAborted(); return { interaction: acceptedInteraction, continuationIssue, @@ -3103,6 +3108,7 @@ export function issueThreadInteractionService( resolved = await acceptRequestConfirmation({ issue: candidate.issue, current: candidate.interaction, + signal, input: {}, actor: { systemId: "system:pr-merged",