mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 23:43:16 +01:00
fix: cancel merge confirmation acceptance under lock
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
420206ce0a
commit
4d3e046e88
3 files changed
+99
-1
No files matched your search
@@ -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.
|
||||
|
||||
@@ -22,7 +22,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<ReturnType<typeof db.$client.reserve>> | 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<void> | 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<void>(resolve => { release = resolve; });
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user