From c97c0dd248ded798b7052482c74204546a334a44 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 23:08:57 -0600 Subject: [PATCH] fix(workspaces): give run preparation its own source lock scope (#19734) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A run prepares its task workspace under per-source advisory locks (`workspace-source-lock.ts`). The locks live in a transaction on a dedicated connection. An `AsyncLocalStorage` scope lets nested calls reuse that transaction. > - Timers and detached promises keep the async context that created them. A native recovery timer or dispatch can start a run from inside another operation's source scope. > - When that run prepares its workspace, it sees the other scope as "held". If that scope already committed, its dedicated connection is closed. The lock query then fails with `Failed query: select pg_try_advisory_xact_lock(...)` and cause `CONNECTION_CLOSED`. If that scope is still open, the run's lock is tied to the other transaction and is released early. > - In a restart chaos wave, a reattached follow-up run failed setup this way 50 ms after "Recovering the existing native runner process". The setup-failure log did not show the driver cause. > - This pull request makes only a live scope count as held, and makes run workspace preparation take its locks in an independent scope. > - The benefit is that a recovered or dispatched run gets its own source lock, and a lost-connection failure is diagnosable from the log. ## Linked Issues or Issue Description **What happened?** After a control-plane restart, a native run was reattached (`reattach_remote_runner`). Its workspace preparation failed at once with `Failed query: select pg_try_advisory_xact_lock(hashtextextended($1, 0)) as acquired params: isolated_repository_source::`. The run ended `failed`/`setup_failed`. The stack showed the "already held scope" branch of `withWorkspaceSourceLocks`, not a new acquisition. The same failure occurred earlier in the same wave on a different issue without a restart. **Steps to reproduce** 1. Call `withIsolatedRepositorySourceLock` for issue A. In its action, start a detached promise that later calls `withIsolatedRepositorySourceLock` for issue B. 2. Let the outer action return so its transaction commits. 3. Let the detached call run. It fails with `Failed query ... pg_try_advisory_xact_lock`, cause `CONNECTION_CLOSED`. **Expected behavior** Work that starts after a source scope ends, or independent work such as a run's preparation, acquires its own locks on its own connection. **Environment** Restart chaos campaign on Daytona (OpenCode, workload C1, fault K1, comment trigger), embedded Postgres. Reproduced locally on embedded Postgres by the new tests. Related: #14910 (deferred comment wake reopens a closed issue). This PR does not change wake promotion. ## What Changed - `server/src/services/workspace-source-lock.ts`: a scope has an `active` flag. The flag is cleared when its transaction ends. `withWorkspaceSourceLocks`, `retainManagedProjectSource`, `retainCurrentWorkspaceSource`, and `assertCurrentWorkspaceSourceHeld` use only a live scope. - `server/src/services/workspace-source-lock.ts`: new `withIndependentWorkspaceSourceLocks`. It acquires locks in a new scope and ignores an inherited one. - `server/src/services/heartbeat/workspace-preparation.ts`: run workspace preparation uses `withIndependentWorkspaceSourceLocks`. - `server/src/services/heartbeat/run-completion.ts`: the "heartbeat execution setup failed" log records a bounded driver `causeCode` (for example `CONNECTION_CLOSED` or a SQLSTATE). It does not log the cause text. - Tests: two embedded-Postgres tests in `isolated-remote-workspace-source-lock.test.ts`, one unit test in `run-completion.test.ts`. ## Verification - `cd server && npx vitest run src/services/isolated-remote-workspace-source-lock.test.ts` — 4 passed. - Red (old scope rule): "gives work detached from a finished scope its own source lock" failed with `Failed query: select pg_try_advisory_xact_lock(...)`, cause `CONNECTION_CLOSED`. This is the production error. - Red: "keeps an independent preparation's lock after the scope that started it commits" failed because the lock was released with the other transaction. - `cd server && npx vitest run src/services/heartbeat/run-completion.test.ts` — 95 passed. - `cd server && npx vitest run src/services/remote-workspace-online-source.test.ts src/services/heartbeat/workspace-preparation.test.ts src/services/isolated-repository-rehome.test.ts src/services/isolated-task-directory-retention.test.ts src/services/isolated-task-directory.test.ts src/__tests__/execution-workspace-reopen.test.ts` — all passed. - `cd server && npx tsc --noEmit` — clean. - No live Daytona replay was run. ## Risks - Low. A nested call inside a live scope behaves as before. - A call that inherited an ended scope now opens its own dedicated connection and lock transaction. Before, it failed. - Run preparation never joins another operation's lock scope. No caller relies on that today. If one did, it would now take a second, independent lock and could see `workspace_source_busy` instead of silently sharing. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edit, embedded Postgres tests). ## 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 --- .../services/heartbeat/run-completion.test.ts | 11 ++++ .../src/services/heartbeat/run-completion.ts | 7 ++- .../heartbeat/workspace-preparation.ts | 8 +-- ...lated-remote-workspace-source-lock.test.ts | 52 +++++++++++++++++++ server/src/services/workspace-source-lock.ts | 42 +++++++++++---- 5 files changed, 106 insertions(+), 14 deletions(-) diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index d7060488743..716fd9b68c5 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -1091,6 +1091,17 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( workspaceProviderFailure: { kind: "provider", reason: "runtime_connection_invalid", retryable: false } }); }); + it("logs the driver cause code of a wrapped setup query failure without its text", async () => { + const f = await fixture(); + const logged = vi.spyOn(logger, "error"); + onTestFinished(() => logged.mockRestore()); + const driverError = Object.assign(new Error("write CONNECTION_CLOSED 127.0.0.1:5432"), { code: "CONNECTION_CLOSED" }); + await f.completion.failRunSetup(f.setup(new Error("Failed query: select 1", { cause: driverError }))); + const setupLog = logged.mock.calls.find(call => call[1] === "heartbeat execution setup failed"); + expect(setupLog?.[0]).toMatchObject({ runId: f.run.id, errorCode: "setup_failed", causeCode: "CONNECTION_CLOSED" }); + expect(JSON.stringify(setupLog?.[0])).not.toContain("127.0.0.1:5432"); + }); + it("retries a workspace operation the stopped provider worker never received", async () => { const f = await fixture(); const refusal = new PluginWorkerUnavailableError({ pluginId: "plugin-1", workerStatus: "stopped", diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 106e64777c1..7a8f2007224 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1419,9 +1419,14 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC "setup_failed"; // Bounded kind and reason only: a workspace provider's own text never crosses into Core. const workspaceProviderFailure = findWorkspaceProviderFailure(outerErr)?.summary() ?? null; + // A wrapped query error hides its driver cause (for example + // CONNECTION_CLOSED or SQLSTATE 53300). Log the bounded code only. + const causeCode = outerErr instanceof Error ? (outerErr.cause as { code?: unknown } | undefined)?.code : undefined; + const setupFailureCauseCode = typeof causeCode === "string" && /^[A-Z0-9_]{1,64}$/.test(causeCode) ? causeCode : null; logger.error( { err: identityRedactor.redact({ message: outerErr instanceof Error ? outerErr.message : String(outerErr), stack: outerErr instanceof Error ? outerErr.stack : undefined }), runId, - errorCode: setupFailureErrorCode, ...(workspaceProviderFailure ? { workspaceProviderFailure } : {}) }, + errorCode: setupFailureErrorCode, ...(setupFailureCauseCode ? { causeCode: setupFailureCauseCode } : {}), + ...(workspaceProviderFailure ? { workspaceProviderFailure } : {}) }, "heartbeat execution setup failed", ); const setupFailureAgent = await getAgent(run.agentId).catch(() => null); diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 7af9562c9c5..d00ff0cd91d 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -47,7 +47,7 @@ import { readNativeWorkspaceSyncReference, readNativeWorkspaceSyncLocalCwd } fro import { parseObject } from "../../adapters/utils.js"; import { materializeNativeChatTaskRoot } from "../native-runtime/native-chat-workspace.js"; import { materializeIsolatedTaskDirectory } from "../isolated-task-directory.js"; -import { withWorkspaceSourceLocks } from "../workspace-source-lock.js"; +import { withIndependentWorkspaceSourceLocks } from "../workspace-source-lock.js"; import { cleanupExecutionWorkspaceArtifacts, ensurePersistedExecutionWorkspaceAvailable, @@ -171,9 +171,11 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac const sourceIssueId = input.reuse.reusableExistingExecutionWorkspace?.sourceIssueId ?? input.reuse.existingExecutionWorkspace?.sourceIssueId ?? input.task.issueId; if (input.reuse.isDotRun || resolveProviderApiPlacement(input)) return prepareHeartbeatWorkspaceLocked(db, input); - if (!sourceIssueId) return withWorkspaceSourceLocks(db, { companyId: input.agent.companyId }, + // A run can be dispatched from inside another operation's source scope, for + // example by a recovery timer. Its preparation owns its own locks. + if (!sourceIssueId) return withIndependentWorkspaceSourceLocks(db, { companyId: input.agent.companyId }, () => prepareHeartbeatWorkspaceLocked(db, input)); - return withWorkspaceSourceLocks(db, { companyId: input.agent.companyId, + return withIndependentWorkspaceSourceLocks(db, { companyId: input.agent.companyId, issueIds: [...new Set([sourceIssueId, input.task.issueId].filter((id): id is string => Boolean(id)))] }, async () => { const reuse = { ...input.reuse }; for (const key of ["existingExecutionWorkspace", "reusableExistingExecutionWorkspace"] as const) { diff --git a/server/src/services/isolated-remote-workspace-source-lock.test.ts b/server/src/services/isolated-remote-workspace-source-lock.test.ts index 36f8d460cb1..8f893b39492 100644 --- a/server/src/services/isolated-remote-workspace-source-lock.test.ts +++ b/server/src/services/isolated-remote-workspace-source-lock.test.ts @@ -5,6 +5,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createDb } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { IsolatedRepositorySourceBusy, withIsolatedRepositorySourceLock } from "./isolated-remote-workspace-source.js"; +import { withIndependentWorkspaceSourceLocks } from "./workspace-source-lock.js"; const support = await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; @@ -69,4 +70,55 @@ describePostgres("workspace source ownership during preparation", () => { await expect(withIsolatedRepositorySourceLock(db, companyId, issueId, async () => "recovered", true)) .resolves.toBe("recovered"); }); + + // RC-53: a recovery timer armed inside one source scope later dispatched a + // reattached run. Its preparation joined the committed scope and failed with + // `Failed query: select pg_try_advisory_xact_lock(...)` on a closed connection. + it("gives work detached from a finished scope its own source lock", async () => { + const companyId = randomUUID(), outerIssueId = randomUUID(), laterIssueId = randomUUID(); + let fire!: () => void, entered!: () => void, release!: () => void; + const fired = new Promise(resolve => { fire = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const pending = new Promise(resolve => { release = resolve; }); + let detached!: Promise; + await withIsolatedRepositorySourceLock(db, companyId, outerIssueId, async () => { + detached = fired.then(() => withIsolatedRepositorySourceLock(db, companyId, laterIssueId, async () => { + entered(); + await pending; + return "prepared"; + })).then(value => ({ value }), error => ({ error: String(error) })); + }); + fire(); + expect(await Promise.race([started.then(() => "entered"), detached])).toBe("entered"); + try { + await expect(withIsolatedRepositorySourceLock(db, companyId, laterIssueId, async () => "overlap", true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + } finally { + release(); + } + expect(await detached).toEqual({ value: "prepared" }); + }); + + it("keeps an independent preparation's lock after the scope that started it commits", async () => { + const companyId = randomUUID(), outerIssueId = randomUUID(), runIssueId = randomUUID(); + let entered!: () => void, release!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const pending = new Promise(resolve => { release = resolve; }); + let preparation!: Promise; + await withIsolatedRepositorySourceLock(db, companyId, outerIssueId, async () => { + preparation = withIndependentWorkspaceSourceLocks(db, { companyId, issueIds: [runIssueId] }, async () => { + entered(); + await pending; + return "prepared"; + }).then(value => ({ value }), error => ({ error: String(error) })); + await started; + }); + try { + await expect(withIsolatedRepositorySourceLock(db, companyId, runIssueId, async () => "overlap", true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + } finally { + release(); + } + expect(await preparation).toEqual({ value: "prepared" }); + }); }); diff --git a/server/src/services/workspace-source-lock.ts b/server/src/services/workspace-source-lock.ts index 6dceccd90d8..52f3aa0269b 100644 --- a/server/src/services/workspace-source-lock.ts +++ b/server/src/services/workspace-source-lock.ts @@ -8,10 +8,28 @@ import { readWorkspaceGitSourceRoots } from "./workspace-git-source-roots.js"; import { withWorkspaceLockKeepalive } from "./native-runtime/native-workspace-lock-keepalive.js"; type SourceTransaction = Parameters[0]>[0]; -type SourceLockScope = { db: Db; companyId: string; keys: Set; transaction: SourceTransaction }; +type SourceLockScope = { db: Db; companyId: string; keys: Set; transaction: SourceTransaction; active: boolean }; const sourceLocks = new AsyncLocalStorage(); export class IsolatedRepositorySourceBusy extends Error {} +// Timers and detached promises keep the async context that created them, so +// work can see a scope whose transaction has committed and whose dedicated +// connection is closed. Only a live scope holds locks. +function heldScope(): SourceLockScope | undefined { + const scope = sourceLocks.getStore(); + return scope?.active ? scope : undefined; +} + +/** Acquires source locks in a new owner scope and ignores an inherited one. + * Use it at the root of independent work, such as a run's workspace + * preparation, that can start inside another operation's source scope. + * Joining that scope would tie these locks to the other transaction. */ +export function withIndependentWorkspaceSourceLocks(db: Db, + input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] }, + action: (transaction: SourceTransaction) => Promise, tryOnly = false): Promise { + return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly)); +} + export function withIsolatedRepositorySourceLock(db: Db, companyId: string, issueId: string, action: () => Promise, tryOnly = false): Promise { return withWorkspaceSourceLocks(db, { companyId, issueIds: [issueId] }, action, tryOnly); @@ -25,7 +43,7 @@ export async function withWorkspaceSourceLocks(db: Db, ...(input.projectIds ?? []).map(id => `managed_project_source:${input.companyId}:${id}`), ])].sort(); if (keys.length > 128) throw new Error("workspace_source_scope_limit"); - const held = sourceLocks.getStore(); + const held = heldScope(); const acquire = async (scope: SourceLockScope, nonblocking: boolean) => { for (const key of keys) { if (scope.keys.has(key)) continue; @@ -45,11 +63,15 @@ export async function withWorkspaceSourceLocks(db: Db, return action(held.transaction); } return withDedicatedDbConnection(db, connection => connection.transaction(async transaction => { - const scope = { db, companyId: input.companyId, transaction, keys: new Set() }; - await acquire(scope, tryOnly); - const rows = await transaction.execute(sql`select setting::bigint as idle_timeout_ms from pg_settings where name = 'idle_in_transaction_session_timeout'`); - return withWorkspaceLockKeepalive(transaction, Number(rows[0]?.idle_timeout_ms), - () => sourceLocks.run(scope, () => action(transaction))); + const scope: SourceLockScope = { db, companyId: input.companyId, transaction, keys: new Set(), active: true }; + try { + await acquire(scope, tryOnly); + const rows = await transaction.execute(sql`select setting::bigint as idle_timeout_ms from pg_settings where name = 'idle_in_transaction_session_timeout'`); + return await withWorkspaceLockKeepalive(transaction, Number(rows[0]?.idle_timeout_ms), + () => sourceLocks.run(scope, () => action(transaction))); + } finally { + scope.active = false; + } })); } @@ -75,7 +97,7 @@ export async function isolatedTaskSourceIdentity(companyId: string, cwd: string) export async function retainManagedProjectSource(db: Db, companyId: string, cwd: string): Promise { const projectId = await managedProjectSourceIdentity(companyId, cwd); if (!projectId) return; - if (sourceLocks.getStore()?.db !== db) throw new Error("workspace_source_admission_scope_required"); + if (heldScope()?.db !== db) throw new Error("workspace_source_admission_scope_required"); await withWorkspaceSourceLocks(db, { companyId, projectIds: [projectId] }, async () => undefined, true); } @@ -96,14 +118,14 @@ export async function workspaceSourceIdentities(companyId: string, cwd: string) } export async function retainCurrentWorkspaceSource(cwd: string): Promise { - const scope = sourceLocks.getStore(); + const scope = heldScope(); if (!scope) return; const identities = await workspaceSourceIdentities(scope.companyId, cwd); await withWorkspaceSourceLocks(scope.db, { companyId: scope.companyId, ...identities }, async () => undefined, true); } export async function assertCurrentWorkspaceSourceHeld(cwd: string): Promise { - const scope = sourceLocks.getStore(); + const scope = heldScope(); if (!scope) throw new Error("workspace_source_admission_scope_required"); const identities = await workspaceSourceIdentities(scope.companyId, cwd); if (identities.projectIds.some(id => !scope.keys.has(`managed_project_source:${scope.companyId}:${id}`))