mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-14 00:26:28 +01:00
fix(workspaces): retry a workspace_finalize record write that loses a deadlock (#20971)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - When a run ends, Paperclip records a `workspace_finalize` workspace operation. Dependent work reads that record to know if the workspace is safe to use. > - The record insert has foreign keys to `execution_workspaces`, `companies`, `heartbeat_runs` and `issues`. Postgres takes a key-share lock on each parent row, in trigger order: workspace, company, run, issue. > - The native finalizer and other writers lock the issue row and then the run row (the order that #20819 documents). When such a transaction holds the issue and then asks for the run, while the finalize insert holds the run and waits for the issue, Postgres detects a deadlock and aborts one side. > - In restart-chaos wave `w-20261111T205259-s6301` (task t01, run `e7129ec7`) the insert lost. The log says "failed to record workspace_finalize=failed operation; dependents may remain gated". The failed-finalize record was lost. > - The insert's lock order comes from the foreign keys, so the code cannot reorder it without a new explicit lock order that other writers (for example the remote-workspace release path, workspace then run) could invert again. A deadlock victim is safe to run again, because its transaction rolled back completely. > - This pull request retries the finalize record write on lock timeout, deadlock and serialization failure (55P03, 40P01, 40001), with the same bounded policy that #20850 added for the provider-pack admission write. > - The benefit is that a short lock conflict can no longer delete the record that releases dependent work. ## Linked Issues or Issue Description **What happened?** Run `e7129ec7` failed its native session recovery (`runner_harness_state_mismatch: backup_without_reusable_lease`). The failure path called `recordWorkspaceFinalize("failed")`. The insert into `workspace_operations` failed with `PostgresError: deadlock detected` (40P01). The server logged that dependents may remain gated. The run then settled as failed with `native_session_interrupted`; the next `finalizeNativeRun` call logged `native_finalization_missing` because that run had no proposed result. **Expected behavior** A transient deadlock does not lose the finalize record. The write runs again and the record is saved. **Steps to reproduce** 1. Transaction A: lock the issue row `FOR UPDATE`. 2. Start a `workspace_finalize` record for a run on that issue. Its insert takes key-share on the workspace, company and run rows, then waits for the issue. 3. Transaction A: lock the run row `FOR UPDATE`. 4. Postgres reports a deadlock. Before this change the insert is the victim and the record is lost. **Additional context** The new test reproduces this with two real transactions on embedded Postgres. It gives transaction A a longer `deadlock_timeout`, so the insert is always the victim, as on the chaos run. Refs #20850 Refs #20819 ## What Changed - New `server/src/services/transient-lock-contention.ts`: `postgresErrorCode`, `isTransientLockContention` and `retryTransientLockContention` (5 attempts, 100 ms × attempt backoff, only 55P03, 40P01 and 40001). - `workspaceOperationService.createRecorder().recordOperation` wraps its `workspace_finalize` state writes (the running-row insert and the terminal update) in that retry. Each attempt is a new transaction, or a new savepoint when the service runs inside a caller's transaction. - `native-provider-pack-admission.ts` uses the shared helper. Its behavior and its exported `nativeProviderPackAdmissionFailureCode` name do not change. - New test in `server/src/__tests__/workspace-operations-reconciliation.test.ts`: a real two-transaction deadlock between an issue-then-run locker and the finalize insert. ## Verification - Red on master: `cd server && npx vitest run src/__tests__/workspace-operations-reconciliation.test.ts -t "loses a deadlock"` failed with `deadlock detected`, code `40P01`, on the same `insert into "workspace_operations"` query as the chaos log. - Green: the same test passes; I ran it 3 times. The full file and the provider-pack admission tests pass (49 tests). `native-result-commit-work.test.ts` passes. - `npx tsc --noEmit -p server` passes. - I stopped `pnpm verify` after it selected 2533 test files; required CI supplies broad coverage. - Chaos: no new cell. The existing W2 × K1 cell (Codex, Daytona warm) reached this path when the restart fault broke native session recovery. I did not rerun a wave; both wave slots were busy. ## Risks - Low risk. Only the `workspace_finalize` state writes retry, and only on the three contention codes. Other errors throw as before. - A deadlock victim's transaction is fully rolled back, so the retried insert (same operation id) and the retried update are idempotent. - The worst added delay is about 1 s of backoff across 5 attempts, on the failure path only. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local test execution). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1880526603
commit
87fe4d9793
4 files changed
+83
-15
No files matched your search
@@ -5,13 +5,16 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
executionWorkspaces,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
projects,
|
||||
workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { subscribeDatabaseWork } from "@paperclipai/db/work-signals";
|
||||
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
||||
import { getWorkspaceOperationLogStore } from "../services/workspace-operation-log-store.js";
|
||||
@@ -152,6 +155,53 @@ describeEmbeddedPostgres("workspace operation reconciliation", () => {
|
||||
} finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
it("keeps the finalization record when its write loses a deadlock to an issue-then-run locker", async () => {
|
||||
const scope = await seedFinalizationWorkspace();
|
||||
const agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
await db.insert(agents).values({ id: agentId, companyId: scope.companyId, name: "Finalizer", adapterType: "paperclip_runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId: scope.companyId, title: "Finalize", status: "in_progress",
|
||||
assigneeAgentId: agentId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId: scope.companyId, agentId, status: "running",
|
||||
contextSnapshot: { issueId } });
|
||||
const locker = createDb(externalUrl ?? tempDb!.connectionString);
|
||||
let lockedIssue!: () => void, releaseLocker!: () => void;
|
||||
const issueHeld = new Promise<void>(resolve => { lockedIssue = resolve; });
|
||||
const lockerGate = new Promise<void>(resolve => { releaseLocker = resolve; });
|
||||
const lockerDone = locker.transaction(async tx => {
|
||||
await tx.execute(sql`set local deadlock_timeout = '20s'`);
|
||||
await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, issueId)).for("update");
|
||||
lockedIssue();
|
||||
await lockerGate;
|
||||
await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
|
||||
});
|
||||
try {
|
||||
await Promise.race([issueHeld, lockerDone.then(() => { throw new Error("locker ended before it held the issue"); })]);
|
||||
const recording = workspaceOperationService(db)
|
||||
.createRecorder({ ...scope, heartbeatRunId: runId, issueId })
|
||||
.recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "failed" }) })
|
||||
.then(value => ({ value }), error => ({ error }));
|
||||
await vi.waitFor(async () => {
|
||||
const [waiting] = await locker.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity
|
||||
where wait_event_type = 'Lock' and query like 'insert into "workspace_operations"%'`);
|
||||
expect(waiting?.count).toBe(1);
|
||||
}, { timeout: 10_000, interval: 50 });
|
||||
releaseLocker();
|
||||
await lockerDone;
|
||||
const outcome = await recording;
|
||||
expect(outcome).not.toHaveProperty("error");
|
||||
expect((await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, runId)))
|
||||
.map(row => [row.phase, row.status])).toEqual([["workspace_finalize", "failed"]]);
|
||||
} finally {
|
||||
releaseLocker();
|
||||
await lockerDone.catch(() => undefined);
|
||||
await locker.$client.end({ timeout: 1 });
|
||||
await db.delete(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, runId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
await db.delete(issues).where(eq(issues.id, issueId));
|
||||
await db.delete(agents).where(eq(agents.id, agentId));
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("does not publish a finalization receipt rolled back by the outer transaction", async () => {
|
||||
const scope = await seedFinalizationWorkspace();
|
||||
await expect(db.transaction(async tx => {
|
||||
|
||||
@@ -4,6 +4,7 @@ import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"
|
||||
import type { NativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { canonicalProviderPackJson, type ProviderPackIdentity } from "./remote-provider-pack-verification.js";
|
||||
import { assertNativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js";
|
||||
import { isTransientLockContention, postgresErrorCode } from "../transient-lock-contention.js";
|
||||
|
||||
type Selection = {
|
||||
admittedRunId: string;
|
||||
@@ -131,19 +132,10 @@ type AdmissionInput = {
|
||||
waitForClaim?: boolean;
|
||||
};
|
||||
|
||||
/** Lock timeouts, deadlocks and serialization aborts that clear on their own. */
|
||||
const CONTENDED_ADMISSION_CODES = new Set(["55P03", "40P01", "40001"]);
|
||||
/** Attempts of one admission write while its claim is contended. */
|
||||
export const NATIVE_PROVIDER_PACK_ADMISSION_ATTEMPTS = 5;
|
||||
|
||||
export function nativeProviderPackAdmissionFailureCode(error: unknown): string | null {
|
||||
for (let value = error, depth = 0; value && typeof value === "object" && depth < 4; depth += 1) {
|
||||
const code = (value as { code?: unknown }).code;
|
||||
if (typeof code === "string" && /^[0-9A-Z]{5}$/.test(code)) return code;
|
||||
value = (value as { cause?: unknown }).cause;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
export const nativeProviderPackAdmissionFailureCode = postgresErrorCode;
|
||||
|
||||
/**
|
||||
* Saves the session's verified provider-pack admission on its run row. A run
|
||||
@@ -159,9 +151,8 @@ export async function recordNativeProviderPackAdmission(db: Db, input: Admission
|
||||
try {
|
||||
return await recordNativeProviderPackAdmissionOnce(db, input);
|
||||
} catch (error) {
|
||||
const code = nativeProviderPackAdmissionFailureCode(error);
|
||||
if (!input.waitForClaim || attempt >= NATIVE_PROVIDER_PACK_ADMISSION_ATTEMPTS
|
||||
|| !code || !CONTENDED_ADMISSION_CODES.has(code)) throw error;
|
||||
|| !isTransientLockContention(error)) throw error;
|
||||
await new Promise(resolve => setTimeout(resolve, 100 * attempt));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
export const TRANSIENT_LOCK_CONTENTION_CODES: ReadonlySet<string> = new Set(["55P03", "40P01", "40001"]);
|
||||
|
||||
export function postgresErrorCode(error: unknown): string | null {
|
||||
for (let value = error, depth = 0; value && typeof value === "object" && depth < 4; depth += 1) {
|
||||
const code = (value as { code?: unknown }).code;
|
||||
if (typeof code === "string" && /^[0-9A-Z]{5}$/.test(code)) return code;
|
||||
value = (value as { cause?: unknown }).cause;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function isTransientLockContention(error: unknown): boolean {
|
||||
const code = postgresErrorCode(error);
|
||||
return code !== null && TRANSIENT_LOCK_CONTENTION_CODES.has(code);
|
||||
}
|
||||
|
||||
export async function retryTransientLockContention<T>(write: () => Promise<T>, attempts = 5): Promise<T> {
|
||||
for (let attempt = 1; ; attempt += 1) {
|
||||
try {
|
||||
return await write();
|
||||
} catch (error) {
|
||||
if (attempt >= attempts || !isTransientLockContention(error)) throw error;
|
||||
await new Promise(resolve => setTimeout(resolve, 100 * attempt));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,7 @@ import { getWorkspaceOperationLogStore } from "./workspace-operation-log-store.j
|
||||
import { persistActivity, publishActivity, type ActivityPublication } from "./activity-log.js";
|
||||
import { crashPoint } from "@paperclipai/shared/crash-points";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { retryTransientLockContention } from "./transient-lock-contention.js";
|
||||
|
||||
type WorkspaceOperationRow = typeof workspaceOperations.$inferSelect;
|
||||
|
||||
@@ -583,7 +584,7 @@ export function workspaceOperationService(db: Db) {
|
||||
async recordOperation(recordInput) {
|
||||
async function writeState<T>(write: (database: Db) => Promise<T>): Promise<T> {
|
||||
if (recordInput.phase !== "workspace_finalize" || (!input.heartbeatRunId && !executionWorkspaceId && !input.issueId)) return write(db);
|
||||
return db.transaction(async tx => {
|
||||
return retryTransientLockContention(() => db.transaction(async tx => {
|
||||
if (executionWorkspaceId) await notifyTerminalWorkspaceWork(tx, `workspace:${executionWorkspaceId}`);
|
||||
else if (input.issueId) await notifyTerminalWorkspaceWork(tx, `issue:${input.issueId}`);
|
||||
if (!input.heartbeatRunId) {
|
||||
@@ -599,7 +600,7 @@ export function workspaceOperationService(db: Db) {
|
||||
await notifyNativeStatusReassessment(tx, { companyId: input.companyId,
|
||||
runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined });
|
||||
return write(tx as unknown as Db);
|
||||
});
|
||||
}));
|
||||
}
|
||||
const currentUserRedactionOptions = {
|
||||
enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs,
|
||||
|
||||
Reference in new issue
Block a user