diff --git a/server/src/__tests__/workspace-operations-reconciliation.test.ts b/server/src/__tests__/workspace-operations-reconciliation.test.ts index 0488294a9c7..70f28663319 100644 --- a/server/src/__tests__/workspace-operations-reconciliation.test.ts +++ b/server/src/__tests__/workspace-operations-reconciliation.test.ts @@ -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(resolve => { lockedIssue = resolve; }); + const lockerGate = new Promise(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 => { diff --git a/server/src/services/native-runtime/native-provider-pack-admission.ts b/server/src/services/native-runtime/native-provider-pack-admission.ts index c5a31b4345f..9d4ab3e40ba 100644 --- a/server/src/services/native-runtime/native-provider-pack-admission.ts +++ b/server/src/services/native-runtime/native-provider-pack-admission.ts @@ -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)); } } diff --git a/server/src/services/transient-lock-contention.ts b/server/src/services/transient-lock-contention.ts new file mode 100644 index 00000000000..b314cdfcb12 --- /dev/null +++ b/server/src/services/transient-lock-contention.ts @@ -0,0 +1,26 @@ +export const TRANSIENT_LOCK_CONTENTION_CODES: ReadonlySet = 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(write: () => Promise, attempts = 5): Promise { + 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)); + } + } +} diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index 193471e3ebb..82b0a33e9bd 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -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(write: (database: Db) => Promise): Promise { 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,