mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): settle a remote workspace save whose process died mid-save (#20326)
## 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: [<dead invocation>]` 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
0eacad0d0c
commit
251c6d5c69
8 files changed
+207
-22
No files matched your search
@@ -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 |
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<string, unknown>, 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<string, unknown> | 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) }) });
|
||||
|
||||
@@ -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})
|
||||
|
||||
@@ -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<typeof workspaceOperations.$inferSelect, "id" | "companyId" | "heartbeatRunId" | "issueId"
|
||||
| "metadata" | "updatedAt">;
|
||||
|
||||
@@ -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" })),
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<void>(resolve => { release = resolve; });
|
||||
let entered!: () => void;
|
||||
const started = new Promise<void>(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<string, { bootId: string; pid: number }> };
|
||||
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<typeof db.transaction>) => {
|
||||
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<string, { bootId: string; pid: number }> };
|
||||
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!,
|
||||
|
||||
@@ -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<string, ControllerProcess> };
|
||||
const liveSaveInvocationIds = new Set<string>();
|
||||
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<string, unknown> | 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<ReturnType<typeof captureRemoteWorkspaceArchive>> | 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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user