From 251c6d5c6961f0507a398202864971130f04b9eb Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:27:44 -0600 Subject: [PATCH] fix(recovery): settle a remote workspace save whose process died mid-save (#20326) 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 > - Native runs in remote sandboxes save the workspace as snapshot operations (`workspace_operations`, phase `remote_workspace_snapshot`). Each save invocation is listed in `saveWork.activeIds` while it runs > - When a controller process is killed during a save, its invocation stays listed and the operation stays `running` > - Nothing recorded which process ran the invocation, so the shared ownership reconciler could not prove the owner gone and never settled the save > - The task's bounded retry then waited behind that save forever > - This pull request records the owning process for each save invocation, and lets the reconciler settle a save whose owner is proven gone > - The benefit is that a restart during a save no longer wedges the task ## Linked Issues or Issue Description **What happened?** In a restart-chaos wave on Daytona with a real cloned repository, two tasks each started their first remote snapshot save (`boundary: initial`, epoch 1). A SIGKILL hit the controller during the save, before any upload. Each run ended `process_lost` and got a bounded retry. The retries stayed `queued` forever, the snapshot operations stayed `running` with `saveWork.activeIds: []` and `quiescedAt: null`, and the sandbox leases stayed active. The ownership reconciler reported the saves as owner-gone but had no way to settle them. **Expected behavior** After a restart, recovery must settle a remote snapshot save whose owning process is gone. It is marked failed with provenance, so the task's retry can continue. A save that is still running must stay untouched. **Steps to reproduce** 1. Run a native task with a remote workspace (for example Daytona with a `git_repo` project source). 2. Kill the controller (SIGKILL) while the first workspace save is in flight, before its upload. 3. After the restart, the run ends `process_lost` and its retry is promoted to `queued`, but the save operation stays `running` and the retry never starts. **Paperclip version or commit** `master` at 1b7fe33432. Refs #20291 (the quiesced case: a save with no active invocation) and #20233 (the same ownership pattern for local workspace commands). ## What Changed - `saveRemoteWorkspace` records the controller process stamp of each invocation in `saveWork.owners` when it registers the invocation. It removes that entry when the invocation settles. A module-level set marks invocations that are still running in this process (`isRemoteWorkspaceSaveInvocationLive`). - The ownership registry owns a `remote_snapshot_save` with active invocations by that stamp. A save whose active invocations have no stamps was written before this change and is owned by its run, as with finalize attempts. A mix of stamps, or more than one process, is unprovable. - `settleAbandonedRemoteSnapshotSave` settles a save with active invocations once its owner is proven gone: - Process-owned: the proven stamp must still match every active invocation, and none may be live in this process. - Legacy run-owned: the run must be terminal. - Either way it clears the active invocations, records them as `abandonedInvocationIds`, and marks the save `failed` with `lastFailure: remote_workspace_save_abandoned`. The #20291 re-arm of blocked fallbacks still applies. - The controller instance lease counts running saves stamped with this boot as held work, so a long save keeps its owner provably alive. - The ownership inventory row for this kind describes the stamp. ## Verification - Red before the fix, green after: `settles a remote snapshot save whose process died mid-save, and keeps live, foreign or unprovable ones` (`ownership-reconciliation-report.test.ts`). On master it settles 0 rows. With the fix it settles exactly two: - a save stamped by a dead process (proof `pid_dead`); - the wave's shape: an unstamped active invocation on a `process_lost` run. It keeps five kinds of save running: a save stamped by this live process, a stamp from another PID space, a legacy save on a live run, a save with mixed stamps, and (in the older #20291 test) a stamped active invocation. - `stamps each save invocation with the process that runs it and clears the stamp when it settles` (`remote-workspace-save-concurrency.test.ts`) pauses a real save mid-flight. It checks the owner stamp and the in-process live mark, then checks that both are cleared after the save finishes. - The reconciler, save-concurrency, remote workspace lifecycle, snapshot retention, retained-source fallback, preparation gates, and controller instance lease suites pass (153 tests, 1 environment-gated file skipped). - `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` in `server` passed. - Not yet verified: the full Daytona wave replay that shows the queued retry starts after the save settles. I asked the chaos harness owner to replay it on this build. ## Risks - A legacy save (written before this change) with an active invocation is settled when its run is terminal. The finalize-attempt kind uses the same rule. During a rolling deploy, an old-release process could still be running such a save for a terminal run. If that save later succeeds, its publication still marks the operation `succeeded`. - If the retry in the wave is also held by the dead writer's dirty workspace authority, settling the save may not be enough to start it. The replay will show whether a second change is needed. - No schema or API change. `saveWork.owners` is new optional metadata. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edits). ## 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 --- doc/plans/2026-11-07-ownership-inventory.md | 2 +- .../controller-instance-lease.test.ts | 27 +++++++- .../ownership-reconciliation-report.test.ts | 50 ++++++++++++++- .../src/services/controller-instance-lease.ts | 3 + .../services/recovery/ownership-registry.ts | 20 +++++- .../workspace-finalize-attempt-settlement.ts | 30 +++++++-- .../remote-workspace-save-concurrency.test.ts | 62 ++++++++++++++++++- server/src/services/remote-workspaces.ts | 35 ++++++++--- 8 files changed, 207 insertions(+), 22 deletions(-) diff --git a/doc/plans/2026-11-07-ownership-inventory.md b/doc/plans/2026-11-07-ownership-inventory.md index bb2ea28dfc7..1c24479d7f4 100644 --- a/doc/plans/2026-11-07-ownership-inventory.md +++ b/doc/plans/2026-11-07-ownership-inventory.md @@ -346,7 +346,7 @@ Scope: workspace operations (#19807 intents), runtime leases, runtime services, | Managed runtime-control operation (start/stop/restart/repair/run) | `workspace_operations.status` (+ `metadata.runtimeControlOwner`) | workspace-operations.ts:559-578 (stamp :539-550, 10 s heartbeat :586-612) | `running` | `{ownerId: random UUID per process, pid, heartbeatAt}`. No host, boot id, or process start time (:25-29, :69-74) | Owner settle :671-690, owner failure :701-728, own deadline 30 min / 4 h (:46, :53, :656-668). Sweep `sweepRuntimeControlOperations` :297-415, run at startup (index.ts:1087) and before each control (:448-454, routes/execution-workspaces.ts:484, :570) | Same ownerId and not in the in-process set → abandoned (:334-335). Otherwise `process.kill(pid,0)` (:126-135) or **TTL/age**: 60 s without a heartbeat (:38, :346). CAS on `updated_at` (:142-147, :379-383) | ownership | 1 (no host/boot), 2 | A, C | | Native workspace finalize attempt (copyback export) | `workspace_operations.status`, phase `workspace_finalize`, `metadata.owningService='native_workspace_finalizer'` | native-runtime/native-workspace-finalizer.ts:170 → workspace-operations.ts:559 | `running` (`pending` is also read as unresolved) | none on the row. Run id only. Physical owner is the `nativeWorkspaceFinalizationOwner` receipt (row below) | (a) Normal settle workspace-operations.ts:671-728. (b) #19807 supersession native-workspace-finalization-supersession.ts:21-91, called from native-workspace-finalizer.ts:94. (c) Offline maintenance offline-workspace-operation-reconciliation.ts:30-118, called from maintenance/ephemeral-migration-data.ts:52. (d) Online variant :152-157 (**no production caller**) | (b) Caller holds finalization ownership, plus the in-process registry `isNativeWorkspaceFinalizationOperationActive` (workspace-operations.ts:114-119) and a later terminal attempt. (c) Advisory lock :41, committed coordinator with no lease :52-57, finalized sync hash :58-60, accepted result :65-70, later succeeded receipt :83-92. Never age | workspace-preservation | 1, 2 | B, C | | Remote provision command intent | `workspace_operations.status` + `metadata.uncertain`, phase `remote_workspace_provision` | remote-workspaces.ts:973-975 (`running`, `uncertain:true`) | `running` / `uncertain=true` | `heartbeatRunId` + `metadata.epoch` (fenced by authority epoch/writerRunId :964) | Success only: remote-workspaces.ts:989-990. On a non-zero exit, a timeout (:980-982), or a crash, **no write settles the row** (grep finds no other writer) | none. Reservation for any other run is refused with `remote_workspace_provision_unverified` (:220-225). Offline migration counts it as a blocker (remote-workspace-migration-tree.ts:160, remote-workspace-migration.ts:237) | workspace-preservation | 1, 2, 3, 7 | B, C | -| Remote snapshot save (initial/checkpoint/final) | `workspace_operations.status`, phase `remote_workspace_snapshot` | remote-workspaces.ts:729 | `running`, `pending` | `heartbeatRunId` + `metadata.epoch`. Fenced by `assertOwned` and the authority writerRunId/epoch (:776-779) | Owner settle in `saveRemoteWorkspace`. Earlier saves are retired only after a final save commits: `retireQuiescedSaves` :592-630 | Save-work quiescence (`readSaveWork` activeIds/quiescedAt :616-618) plus a committed final snapshot (:597-608). A quiesced save whose run is terminal is settled `failed` by the ownership reconciler (`settleAbandonedRemoteSnapshotSave`) | workspace-preservation | 1, 2 | B, C | +| Remote snapshot save (initial/checkpoint/final) | `workspace_operations.status`, phase `remote_workspace_snapshot` | remote-workspaces.ts:729 | `running`, `pending` | `heartbeatRunId` + `metadata.epoch`. Fenced by `assertOwned` and the authority writerRunId/epoch (:776-779) | Owner settle in `saveRemoteWorkspace`. Earlier saves are retired only after a final save commits: `retireQuiescedSaves` :592-630 | Save-work quiescence (`readSaveWork` activeIds/quiescedAt :616-618) plus a committed final snapshot (:597-608). Each save invocation records its controller process in `saveWork.owners`. The ownership reconciler (`settleAbandonedRemoteSnapshotSave`) settles a save `failed` when its active invocations' process is proven gone, or when it is quiesced (or saved before owner stamps) and its run is terminal | workspace-preservation | 1, 2 | B, C | | Remote workspace migration op | `workspace_operations.status`, phase `remote_workspace_migration` | remote-workspace-migration.ts:375 | `running` | none (workspace id) | Explicit publication or repair by migration (unverified line). The runtime-control sweep skips it on purpose (workspace-operations.ts:319-323) | none | workspace-preservation | 1, 2 | B, C | | Execution workspace runtime lease | `execution_workspace_runtime_leases` (row existence) | workspace-runtime-leases.ts:194-209 (create), :235-247 (renew), :255-273 (reclaim). Called only from routes/execution-workspaces.ts:567 | row present | `owner_key` `issue:`/`run:`/`agent:` (+ FK columns). No process identity | `release` :291-313. In production it is called only with `force` on archive (routes/execution-workspaces.ts:1270). Otherwise the lease is reclaimed lazily at the next claim (:250-283) | `evaluateStaleOwner` :85-119: issue hidden or terminal, run terminal, or **TTL/age** 30 min (:39, :90). An agent-scoped owner has TTL only (:116-118) | ownership | 1, 2 | A, C | | Workspace runtime service (local process) | `workspace_runtime_services.status` | workspace-runtime.ts (insert line unverified) | `provisioning`, `starting`, `running` | `startedByRunId`, `ownerAgentId`. The pid lives in a local service registry file, not on the row | Startup `reconcilePersistedRuntimeServicesOnStartup` workspace-runtime.ts:8636, which adopts or stops each service (index.ts:1101, not awaited) | Registry record match (workspace-runtime.ts:8781-8797) plus `isPidAlive` local-service-supervisor.ts:218, :245-250. Same host only | ownership | 1, 2 | A, C | diff --git a/server/src/__tests__/controller-instance-lease.test.ts b/server/src/__tests__/controller-instance-lease.test.ts index 53a1eecdac5..9552e66e666 100644 --- a/server/src/__tests__/controller-instance-lease.test.ts +++ b/server/src/__tests__/controller-instance-lease.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, companies, controllerInstances, createDb, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { agents, companies, controllerInstances, createDb, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { controllerInstanceLeaseEnabled, proveControllerInstance, registerControllerInstanceDb, registerControllerInstanceLeaseWork, @@ -181,6 +181,31 @@ suite("controller instance lease: a replaced container's records are provable", } finally { vi.useRealTimers(); } }); + it("keeps renewing while this boot runs a remote workspace save", async () => { + let now = 0; + vi.useFakeTimers({ toFake: ["Date"], now: new Date() }); + try { + await startHere({ now: () => now }, false); + currentControllerProcessStamp(); + await tickControllerInstanceLease(); + now = 120_000; + vi.setSystemTime(Date.now() + 120_000); + const invocation = randomUUID(); + const owner = { bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null }; + await db.insert(workspaceOperations).values({ companyId, phase: "remote_workspace_snapshot", status: "running", + metadata: { boundary: "final", epoch: 1, saveWork: { schema: "paperclip.remote-workspace-save-work.v1", activeIds: [invocation], + unknown: false, quiescedAt: null, owners: { [invocation]: owner } } } }); + beginIdleTrackedWork()(); + await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` }); + await tickControllerInstanceLease(); + expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live"); + await db.delete(workspaceOperations); + await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` }); + await tickControllerInstanceLease(); + expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("expired"); + } finally { vi.useRealTimers(); } + }); + it("keeps renewing while this boot runs a legacy run", async () => { vi.useFakeTimers({ toFake: ["Date"], now: new Date() }); try { diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 58bf048e852..1067046bb5b 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -33,6 +33,7 @@ import { type OwnershipReport, } from "../services/recovery/ownership-invariant-report.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +import { trackRemoteWorkspaceSaveInvocationForTests } from "../services/remote-workspaces.js"; import { workspaceOperationService } from "../services/workspace-operations.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); @@ -375,6 +376,7 @@ describe("report-only ownership reconciliation", () => { const runs = { failed: randomUUID(), live: randomUUID() }; await db.insert(heartbeatRuns).values([run(runs.failed, "failed"), run(runs.live, "running")]); const quiesced = { schema: "paperclip.remote-workspace-save-work.v1", activeIds: [], unknown: false, quiescedAt: minutesAgo(30).toISOString() }; + const liveInvocation = randomUUID(); const ops = { abandoned: randomUUID(), pendingAbandoned: randomUUID(), liveRun: randomUUID(), activeInvocation: randomUUID(), unknownWork: randomUUID(), notQuiesced: randomUUID(), invalidWork: randomUUID() }; const op = (id: string, runId: string, saveWork: Record, status = "running") => ({ id, companyId, heartbeatRunId: runId, @@ -384,7 +386,8 @@ describe("report-only ownership reconciliation", () => { op(ops.abandoned, runs.failed, quiesced), op(ops.pendingAbandoned, runs.failed, quiesced, "pending"), op(ops.liveRun, runs.live, quiesced), - op(ops.activeInvocation, runs.failed, { ...quiesced, activeIds: [randomUUID()], quiescedAt: null }), + op(ops.activeInvocation, runs.failed, { ...quiesced, activeIds: [liveInvocation], quiescedAt: null, + owners: { [liveInvocation]: { bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null } } }), op(ops.unknownWork, runs.failed, { ...quiesced, unknown: true }), op(ops.notQuiesced, runs.failed, { ...quiesced, quiescedAt: null }), op(ops.invalidWork, runs.failed, { ...quiesced, quiescedAt: "" }), @@ -393,7 +396,8 @@ describe("report-only ownership reconciliation", () => { .from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_snapshot"))) .map(row => [Object.keys(ops).find(name => ops[name as keyof typeof ops] === row.id), row.status])); - const first = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + const untrack = trackRemoteWorkspaceSaveInvocationForTests(liveInvocation); + const first = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }).finally(untrack); expect(kindReport(first, "remote_snapshot_save")).toMatchObject({ inProgress: 7, settled: 2 }); expect(await statuses()).toEqual({ abandoned: "failed", pendingAbandoned: "failed", liveRun: "running", activeInvocation: "running", unknownWork: "running", notQuiesced: "running", invalidWork: "running" }); @@ -406,6 +410,48 @@ describe("report-only ownership reconciliation", () => { expect(failedRun.status).toBe("failed"); }); + it("settles a remote snapshot save whose process died mid-save, and keeps live, foreign or unprovable ones", async () => { + const run = (id: string, status: string) => ({ id, companyId, agentId, status, invocationSource: "manual", + ...(status === "running" ? {} : { finishedAt: minutesAgo(2), errorCode: "process_lost" }) }); + const runs = { lost: randomUUID(), live: randomUUID() }; + await db.insert(heartbeatRuns).values([run(runs.lost, "failed"), run(runs.live, "running")]); + const dead = { bootId: "previous-boot", pid: 4_194_303, hostHash: currentProcessSpaceHash(), + processStartedAt: "2026-11-07T09:00:00.000Z" }; + const self = { bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null }; + const ops = { deadOwner: randomUUID(), selfOwner: randomUUID(), foreignOwner: randomUUID(), legacyLost: randomUUID(), + legacyLive: randomUUID(), mixed: randomUUID() }; + const op = (id: string, runId: string, owners: Record | undefined, activeIds = [randomUUID()]) => ({ + id, companyId, heartbeatRunId: runId, phase: "remote_workspace_snapshot", status: "running", + metadata: { epoch: 1, boundary: "initial", generation: 1, exportCleanup: { state: "pending" }, + saveWork: { schema: "paperclip.remote-workspace-save-work.v1", activeIds, unknown: false, quiescedAt: null, + ...(owners ? { owners: Object.fromEntries(activeIds.map((invocation, index) => [invocation, Object.values(owners)[index]]) + .filter(([, stamp]) => stamp !== undefined)) } : {}) } } }); + const twoIds = [randomUUID(), randomUUID()], selfInvocation = randomUUID(); + await db.insert(workspaceOperations).values([ + op(ops.deadOwner, runs.live, { a: dead }), + op(ops.selfOwner, runs.lost, { a: self }, [selfInvocation]), + op(ops.foreignOwner, runs.lost, { a: { ...dead, hostHash: "another-pid-space" } }), + op(ops.legacyLost, runs.lost, undefined), + op(ops.legacyLive, runs.live, undefined), + op(ops.mixed, runs.lost, { a: dead }, twoIds), + ]); + const statuses = async () => Object.fromEntries((await db.select({ id: workspaceOperations.id, status: workspaceOperations.status }) + .from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_snapshot"))) + .map(row => [Object.keys(ops).find(name => ops[name as keyof typeof ops] === row.id), row.status])); + + const untrack = trackRemoteWorkspaceSaveInvocationForTests(selfInvocation); + const first = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }).finally(untrack); + expect(kindReport(first, "remote_snapshot_save")).toMatchObject({ inProgress: 6, settled: 2 }); + expect(await statuses()).toEqual({ deadOwner: "failed", selfOwner: "running", foreignOwner: "running", + legacyLost: "failed", legacyLive: "running", mixed: "running" }); + const [settled] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, ops.deadOwner)); + expect(settled.metadata).toMatchObject({ saveWork: { activeIds: [], owners: {}, unknown: false, quiescedAt: expect.any(String) }, + abandonedSave: { reason: "abandoned_remote_snapshot_save", proof: "pid_dead", abandonedInvocationIds: [expect.any(String)] }, + lastFailure: { code: "remote_workspace_save_abandoned" } }); + const [liveRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runs.live)); + expect(liveRun.status).toBe("running"); + }); + it("settles a workspace command whose process died mid-command, and keeps live or unprovable ones", async () => { const run = (id: string, status: string) => ({ id, companyId, agentId, status, invocationSource: "manual", ...(status === "running" ? {} : { finishedAt: minutesAgo(2) }) }); diff --git a/server/src/services/controller-instance-lease.ts b/server/src/services/controller-instance-lease.ts index ac711e7104e..08d6de7a790 100644 --- a/server/src/services/controller-instance-lease.ts +++ b/server/src/services/controller-instance-lease.ts @@ -73,6 +73,9 @@ function holdsStampedWorkSql(bootId: string) { or exists (select 1 from sandbox_creation_operations where state = 'pending' and metadata->'creationController'->>'bootId' = ${bootId}) or exists (select 1 from workspace_operations where status in ('running', 'pending') and metadata->'finalizeOwner'->>'bootId' = ${bootId}) or exists (select 1 from workspace_operations where status = 'running' and metadata->'commandOwner'->>'bootId' = ${bootId}) + or exists (select 1 from workspace_operations, jsonb_each(case when jsonb_typeof(metadata#>'{saveWork,owners}') = 'object' + then metadata#>'{saveWork,owners}' else '{}'::jsonb end) as save_owner(invocation_id, stamp) + where status in ('running', 'pending') and phase = 'remote_workspace_snapshot' and save_owner.stamp->>'bootId' = ${bootId}) or exists (select 1 from agents where lifecycle_operation->'leaseHolder'->>'bootId' = ${bootId}) or exists (select 1 from native_run_finalizations where lease_owner is not null and lease_holder->>'token' = lease_owner and lease_holder->>'bootId' = ${bootId}) diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index bb4954921c8..a4865834760 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -21,7 +21,8 @@ import { type Db, } from "@paperclipai/db"; import { EXECUTION_RECONCILIATION_CAUSES } from "@paperclipai/shared"; -import { readControllerProcess } from "../controller-process-identity.js"; +import { currentControllerBootId, readControllerProcess } from "../controller-process-identity.js"; +import { isRemoteWorkspaceSaveInvocationLive } from "../remote-workspaces.js"; import { idleAccountingSpoolPath, idleOrphanSpoolPath } from "../idle-local-work.js"; import { NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT, @@ -520,6 +521,21 @@ function controllerStamp(value: unknown, scope: string): RecordOwner { // ---------------------------------------------------------------------------- // workspace_operations +function remoteSnapshotSaveOwner(row: OperationRow): RecordOwner { + const work = json(json(row.metadata).saveWork); + const active = Array.isArray(work.activeIds) ? work.activeIds.filter((id): id is string => typeof id === "string") : []; + if (!active.length) return runOwner(row.heartbeatRunId); + const owners = json(work.owners); + const stamps = active.map(id => readControllerProcess(owners[id])); + if (stamps.every(stamp => stamp === null) && !active.some(id => owners[id] !== undefined)) return runOwner(row.heartbeatRunId); + const [first] = stamps; + if (!first || stamps.some(stamp => !stamp || stamp.bootId !== first.bootId || stamp.pid !== first.pid)) return { kind: "none" }; + if (first.bootId === currentControllerBootId() && first.pid === process.pid && !active.some(isRemoteWorkspaceSaveInvocationLive)) { + return runOwner(row.heartbeatRunId); + } + return controllerStamp(owners[active[0]!], "remote_snapshot_save"); +} + type OperationRow = Pick; @@ -876,7 +892,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ row => runOwner(row.heartbeatRunId)), { ...operationKind("remote_snapshot_save", "Remote snapshot save", "workspace_preservation", and(eq(workspaceOperations.phase, "remote_workspace_snapshot"), inArray(workspaceOperations.status, ["running", "pending"])), - row => runOwner(row.heartbeatRunId)), settle: settleAbandonedRemoteSnapshotSave }, + remoteSnapshotSaveOwner), settle: settleAbandonedRemoteSnapshotSave }, operationKind("remote_workspace_migration", "Remote workspace migration op", "workspace_preservation", and(eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "running")), () => ({ kind: "none" })), diff --git a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts index f6f70c99440..815102ac22b 100644 --- a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts +++ b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts @@ -2,8 +2,9 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import { heartbeatRuns, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; import type { OwnerProof } from "../owner-proof.js"; +import { currentControllerBootId } from "../controller-process-identity.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; -import { readSaveWork } from "../remote-workspaces.js"; +import { isRemoteWorkspaceSaveInvocationLive, readSaveWork } from "../remote-workspaces.js"; import { notifyTerminalWorkspaceWork } from "../terminal-workspace-work.js"; import { RETAINED_SOURCE_FALLBACK_CHECK_KEY } from "../workspace-recovery-fallback-discovery.js"; import { LOCAL_WORKSPACE_COMMAND_PHASES, isOwnerStampedWorkspaceOperationLive } from "../workspace-operations.js"; @@ -121,10 +122,25 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre if (!operation) return { settled: false, reason: "attempt_no_longer_in_progress" }; const metadata = operation.metadata ?? {}; const work = readSaveWork(metadata); - if (!work || work.unknown || work.activeIds.length > 0 || !work.quiescedAt) return { settled: false, reason: "save_work_unsettled" }; - const [run] = operation.heartbeatRunId ? await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns) - .where(and(eq(heartbeatRuns.id, operation.heartbeatRunId), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES))) : []; - if (!run) return { settled: false, reason: "owner_run_active" }; + if (!work || work.unknown || (!work.activeIds.length && !work.quiescedAt)) return { settled: false, reason: "save_work_unsettled" }; + const owners = work.owners ?? {}; + if (work.activeIds.some(isRemoteWorkspaceSaveInvocationLive)) return { settled: false, reason: "attempt_live_in_this_process" }; + if (record.owner.kind === "process") { + const { bootId, pid } = record.owner.stamp; + if (!work.activeIds.length || work.activeIds.some(id => owners[id]?.bootId !== bootId || owners[id]?.pid !== pid)) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + } else if (record.owner.kind === "run") { + if (work.activeIds.some(id => owners[id] !== undefined + && (owners[id]!.bootId !== currentControllerBootId() || owners[id]!.pid !== process.pid))) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + const [run] = operation.heartbeatRunId ? await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, operation.heartbeatRunId), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES))) : []; + if (!run) return { settled: false, reason: "owner_run_active" }; + } else { + return { settled: false, reason: "owner_not_provable" }; + } const now = context.now; const provenance = { schema: "paperclip.remote-snapshot-save-abandonment.v1", @@ -133,6 +149,7 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre previousStatus: operation.status, runId: operation.heartbeatRunId, quiescedAt: work.quiescedAt, + abandonedInvocationIds: work.activeIds, reconciledAt: now.toISOString(), }; const workKey = operation.executionWorkspaceId ? `workspace:${operation.executionWorkspaceId}` @@ -140,8 +157,9 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre if (workKey) await notifyTerminalWorkspaceWork(tx, workKey); const [row] = await tx.update(workspaceOperations).set({ status: "failed", exitCode: 1, finishedAt: now, updatedAt: now, - stderrExcerpt: `${operation.stderrExcerpt ?? ""}The workspace save stopped without finishing, and its run has ended.\n`.slice(-4096), + stderrExcerpt: `${operation.stderrExcerpt ?? ""}The workspace save stopped without finishing; the process that ran it is gone.\n`.slice(-4096), metadata: { ...metadata, abandonedSave: provenance, + saveWork: { ...work, activeIds: [], owners: {}, quiescedAt: work.quiescedAt ?? now.toISOString() }, lastFailure: { code: "remote_workspace_save_abandoned", at: now.toISOString() } }, }).where(and(eq(workspaceOperations.id, operation.id), eq(workspaceOperations.status, operation.status))) .returning({ id: workspaceOperations.id }); diff --git a/server/src/services/remote-workspace-save-concurrency.test.ts b/server/src/services/remote-workspace-save-concurrency.test.ts index 16e7293375d..29eae98db7d 100644 --- a/server/src/services/remote-workspace-save-concurrency.test.ts +++ b/server/src/services/remote-workspace-save-concurrency.test.ts @@ -18,8 +18,10 @@ import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.j import { preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js"; import { createSavedWorkspaceLoader, prepareRemoteWorkspace, readRemoteWorkspaceAuthority, - provisionRemoteWorkspace, reserveRemoteWorkspace, saveRemoteWorkspace, + isRemoteWorkspaceSaveInvocationLive, provisionRemoteWorkspace, reserveRemoteWorkspace, saveRemoteWorkspace, } from "./remote-workspaces.js"; +import { currentControllerBootId } from "./controller-process-identity.js"; +import { createOwnershipReportState, runOwnershipReconciliationPass } from "./recovery/ownership-invariant-report.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; @@ -487,6 +489,64 @@ describe.skipIf(!support.supported)("remote workspace save parent lock order", ( expect(await readFile(path.join(authority.location!.remoteCwd, "saved.txt"), "utf8")).toBe("one preserved result\n"); }, 30_000); + it("stamps each save invocation with the process that runs it and clears the stamp when it settles", async () => { + const f = await preparedFailure(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const paused: AdapterSandboxExecutionTarget = { ...f.target, runner: { ...f.target.runner!, + execute: async input => { entered(); await gate; return f.target.runner!.execute(input); } } }; + const saving = saveRemoteWorkspace({ db, storage: f.storage, workspace: f.workspace, runId: f.runId, lease: f.lease, + target: paused, assertOwned: async () => undefined }); + const snapshotOperation = async () => (await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), + sql`${workspaceOperations.metadata}->>'boundary' = 'final'`)))[0]!; + let work: { activeIds: string[]; owners: Record }; + try { + await started; + work = (await snapshotOperation()).metadata!.saveWork as typeof work; + expect(work.activeIds).toHaveLength(1); + expect(work.owners[work.activeIds[0]!]).toMatchObject({ bootId: currentControllerBootId(), pid: process.pid }); + expect(isRemoteWorkspaceSaveInvocationLive(work.activeIds[0]!)).toBe(true); + } finally { + release(); + await saving; + } + expect((await snapshotOperation()).metadata).toMatchObject({ saveWork: { activeIds: [], owners: {} } }); + expect(isRemoteWorkspaceSaveInvocationLive(work.activeIds[0]!)).toBe(false); + }, 30_000); + + it("lets the reconciler settle a save whose settlement write failed once its run has ended", async () => { + const f = await preparedFailure(); + let failNextTransaction = false; + const flaky = new Proxy(db, { get(target, property, receiver) { + if (property === "transaction") return (...args: Parameters) => { + if (failNextTransaction) { failNextTransaction = false; return Promise.reject(new Error("connection terminated")); } + return target.transaction(...args); + }; + const value = Reflect.get(target, property, receiver); + return typeof value === "function" ? value.bind(target) : value; + } }) as typeof db; + const refused: AdapterSandboxExecutionTarget = { ...f.target, runner: { ...f.target.runner!, + execute: async () => { failNextTransaction = true; throw new Error("Daytona sandbox lease SYNTHETIC-private-id is no longer active."); } } }; + await expect(saveRemoteWorkspace({ db: flaky, storage: f.storage, workspace: f.workspace, runId: f.runId, lease: f.lease, + target: refused, assertOwned: async () => undefined })).rejects.toThrow("no longer active"); + const snapshotOperation = async () => (await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), + sql`${workspaceOperations.metadata}->>'boundary' = 'final'`)))[0]!; + const pending = await snapshotOperation(); + const work = pending.metadata!.saveWork as { activeIds: string[]; owners: Record }; + expect(pending.status).toBe("running"); + expect(work.owners[work.activeIds[0]!]).toMatchObject({ bootId: currentControllerBootId(), pid: process.pid }); + expect(isRemoteWorkspaceSaveInvocationLive(work.activeIds[0]!)).toBe(false); + + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(await snapshotOperation()).toMatchObject({ status: "failed", + metadata: { saveWork: { activeIds: [], owners: {} }, abandonedSave: { reason: "abandoned_remote_snapshot_save" } } }); + }, 30_000); + it("ends a failed save operation with its redacted cause and reopens it for the retry", async () => { const f = await preparedFailure(); const refused: AdapterSandboxExecutionTarget = { ...f.target, runner: { ...f.target.runner!, diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index ea990eb8fc0..ea511c813df 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -1,4 +1,5 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { currentControllerProcessStamp, type ControllerProcess } from "./controller-process-identity.js"; import { workspaceAllocationFiles } from "./remote-workspace-allocation.js"; import { readNativeWorkspaceMigration } from "./native-runtime/native-workspace-migration.js"; import { createHash, randomUUID } from "node:crypto"; @@ -68,7 +69,16 @@ type RemoteWorkspaceSaveTiming = RemoteWorkspaceSavePhaseTiming & { storage?: WorkspaceSnapshotStorageTiming; }; const RETIRED_SAVE_SCHEMA = "paperclip.remote-workspace-retired-save.v1"; -type SaveWork = { schema: typeof SAVE_WORK_SCHEMA; activeIds: string[]; unknown: boolean; quiescedAt: string | null }; +type SaveWork = { schema: typeof SAVE_WORK_SCHEMA; activeIds: string[]; unknown: boolean; quiescedAt: string | null; + owners?: Record }; +const liveSaveInvocationIds = new Set(); +export function isRemoteWorkspaceSaveInvocationLive(invocationId: string) { + return liveSaveInvocationIds.has(invocationId); +} +export function trackRemoteWorkspaceSaveInvocationForTests(invocationId: string) { + liveSaveInvocationIds.add(invocationId); + return () => { liveSaveInvocationIds.delete(invocationId); }; +} export function readSaveWork(metadata: Record | null | undefined): SaveWork | null { const work = object(metadata?.saveWork); return work.schema === SAVE_WORK_SCHEMA && Array.isArray(work.activeIds) @@ -705,6 +715,7 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId const work = readSaveWork(operation?.metadata); if (!operation || !work || !work.activeIds.includes(input.invocationId)) return false; const activeIds = work.activeIds.filter(id => id !== input.invocationId); + const owners = Object.fromEntries(Object.entries(work.owners ?? {}).filter(([id]) => id !== input.invocationId)); if (!activeIds.length && !work.unknown) await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); const now = new Date(); // The last settled invocation of a failed save ends the operation. A later @@ -712,7 +723,7 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId const failure = input.failure; const failed = Boolean(failure) && operation.status === "running" && !activeIds.length && !work.unknown; await tx.update(workspaceOperations).set({ metadata: { ...operation.metadata, - saveWork: { ...work, activeIds, quiescedAt: activeIds.length ? null : now.toISOString() }, + saveWork: { ...work, activeIds, owners, quiescedAt: activeIds.length ? null : now.toISOString() }, ...(input.timing ? { saveTiming: saveTimingRecord(input.timing, input.failure ? "failed" : "succeeded", now) } : {}), ...(failed && failure ? { lastFailure: { ...failure, at: now.toISOString() } } : {}) }, ...(failed && failure ? { status: "failed", exitCode: 1, finishedAt: now, @@ -786,6 +797,8 @@ export async function saveRemoteWorkspace(input: { if (!expected?.location || expected.location.providerLeaseId !== input.lease.providerLeaseId || expected.location.leaseId !== input.lease.id) throw new Error("remote_workspace_save_binding_mismatch"); await input.assertOwned(); const invocationId = randomUUID(); + const owner = { [invocationId]: currentControllerProcessStamp() }; + liveSaveInvocationIds.add(invocationId); const reserved = await input.db.transaction(async tx => { const issueId = await lockWorkspaceRunParents(tx, input.workspace.companyId, input.runId); const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id); @@ -820,7 +833,8 @@ export async function saveRemoteWorkspace(input: { const reopened = previous.status === "failed" ? { status: "running", finishedAt: null, exitCode: null } : {}; const [registered] = await tx.update(workspaceOperations).set({ ...reopened, metadata: { ...previous.metadata, saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [...(work?.activeIds ?? []), invocationId], - unknown: work?.unknown ?? true, quiescedAt: null } }, updatedAt: new Date() }).where(eq(workspaceOperations.id, previous.id)).returning(); + unknown: work?.unknown ?? true, quiescedAt: null, owners: { ...(work?.owners ?? {}), ...owner } } }, updatedAt: new Date() }) + .where(eq(workspaceOperations.id, previous.id)).returning(); return { kind: "registered" as const, operation: registered! }; } // The workspace lock serializes generation reservations. Failed/deleted @@ -842,12 +856,12 @@ export async function saveRemoteWorkspace(input: { const [created] = await tx.insert(workspaceOperations).values({ companyId: workspace.companyId, executionWorkspaceId: workspace.id, heartbeatRunId: input.runId, issueId, phase: "remote_workspace_snapshot", status: "running", metadata: { boundary, epoch: current.epoch, parentSnapshotId: current.snapshotId, generation, - saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [invocationId], unknown: false, quiescedAt: null }, + saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [invocationId], unknown: false, quiescedAt: null, owners: owner }, exportCleanup: { ...current.location, state: "pending" } }, }).returning(); return { kind: "registered" as const, operation: created! }; - }); - if (reserved.kind === "unchanged") return reserved.saved; + }).catch(error => { liveSaveInvocationIds.delete(invocationId); throw error; }); + if (reserved.kind === "unchanged") { liveSaveInvocationIds.delete(invocationId); return reserved.saved; } const { operation } = reserved; let captured: Awaited> | null = null; let publicationScratch: string | undefined; @@ -939,14 +953,17 @@ export async function saveRemoteWorkspace(input: { if (publicationScratch) await rm(publicationScratch, { recursive: true, force: true }).catch(() => undefined); try { await captured?.cleanup(); } catch { logger.warn({ runId: input.runId, operationId: operation.id }, "Remote workspace local archive cleanup failed"); } + const location = expected.location; + const settle = () => settleSaveInvocation({ db: input.db, workspace: input.workspace, runId: input.runId, epoch: expected.epoch, + location, operationId: operation.id, invocationId, failure, + timing: replayedCommittedSave ? undefined : timing }); try { - if (await settleSaveInvocation({ db: input.db, workspace: input.workspace, runId: input.runId, epoch: expected.epoch, - location: expected.location, operationId: operation.id, invocationId, failure, - timing: replayedCommittedSave ? undefined : timing })) { + if (await settle()) { await cleanupSavedExports({ db: input.db, companyId: input.workspace.companyId, workspaceId: input.workspace.id, location: expected.location, target, assertOwned: input.assertOwned }); } } catch { logger.warn({ runId: input.runId, operationId: operation.id }, "Remote workspace save settlement or export cleanup remains pending"); } + liveSaveInvocationIds.delete(invocationId); } }