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}`))