mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(workspaces): give run preparation its own source lock scope (#19734)
<!-- Write all pull request text in Simplified Technical English (ASD-STE100): short sentences, one instruction per sentence, simple approved vocabulary, and the active voice. --> ## 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:<company>:<issue>`. 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
839d41aabb
commit
c97c0dd248
5 files changed
+106
-14
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<void>(resolve => { fire = resolve; });
|
||||
const started = new Promise<void>(resolve => { entered = resolve; });
|
||||
const pending = new Promise<void>(resolve => { release = resolve; });
|
||||
let detached!: Promise<unknown>;
|
||||
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<void>(resolve => { entered = resolve; });
|
||||
const pending = new Promise<void>(resolve => { release = resolve; });
|
||||
let preparation!: Promise<unknown>;
|
||||
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" });
|
||||
});
|
||||
});
|
||||
@@ -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<Parameters<Db["transaction"]>[0]>[0];
|
||||
type SourceLockScope = { db: Db; companyId: string; keys: Set<string>; transaction: SourceTransaction };
|
||||
type SourceLockScope = { db: Db; companyId: string; keys: Set<string>; transaction: SourceTransaction; active: boolean };
|
||||
const sourceLocks = new AsyncLocalStorage<SourceLockScope>();
|
||||
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<T>(db: Db,
|
||||
input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] },
|
||||
action: (transaction: SourceTransaction) => Promise<T>, tryOnly = false): Promise<T> {
|
||||
return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly));
|
||||
}
|
||||
|
||||
export function withIsolatedRepositorySourceLock<T>(db: Db, companyId: string, issueId: string,
|
||||
action: () => Promise<T>, tryOnly = false): Promise<T> {
|
||||
return withWorkspaceSourceLocks(db, { companyId, issueIds: [issueId] }, action, tryOnly);
|
||||
@@ -25,7 +43,7 @@ export async function withWorkspaceSourceLocks<T>(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<T>(db: Db,
|
||||
return action(held.transaction);
|
||||
}
|
||||
return withDedicatedDbConnection(db, connection => connection.transaction(async transaction => {
|
||||
const scope = { db, companyId: input.companyId, transaction, keys: new Set<string>() };
|
||||
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<string>(), 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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}`))
|
||||
|
||||
Reference in new issue
Block a user