From 8a8abdb7b999f493d9ade9a9f2581bdba3174879 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 02:55:48 -0600 Subject: [PATCH] fix(recovery): retire the ledger row of an abandoned workspace save (#20502) 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 > - Remote workspaces save their files to object storage. Each save writes a row in the `execution_workspace_snapshots` ledger. The row stays `pending` until the upload publishes it. > - A restart that kills the server during a save leaves the save unfinished. The ownership reconciler settles the save operation as `remote_workspace_save_abandoned`. It does not touch the ledger row. > - The row then stays `pending` for good, and its `updated_at` never moves. The `remote_snapshot_ledger` ownership kind has no settle path, so every later ownership report flags it as an `I2_settled` violation. > - This pull request retires that row to a terminal state once no retry can reopen its save, and gives `remote_snapshot_ledger` a settle path. > - The benefit is a clean ownership report after a restart during a save, and a ledger that tells the truth about uploads that will never finish. ## Linked Issues or Issue Description **What happened?** A Daytona R1 × K1 restart-chaos wave (`w-20261110T071047-s7101`, `--storage s3`) killed the server during the initial workspace save of 3 runs. The save operations ended `failed` / `remote_workspace_save_abandoned`. Their `execution_workspace_snapshots` rows stayed `pending` permanently, with no object key and no upload attempt. The workspaces later committed newer snapshots (authority epoch 3, row epoch 1). Every ownership report afterwards flagged 3 `I2_settled` violations of kind `remote_snapshot_ledger`. **Expected behavior** When a save is abandoned and its run and writer epoch have ended, its pending ledger row reaches a terminal state, and the ownership report shows no `I2_settled` violation for it. **Steps to reproduce** 1. Run a Daytona R1 × K1 wave with `--storage s3` and K1 at about +65 s, so the restart lands inside the initial remote workspace save. 2. Let the runs finish or fail and the workspace commit a newer snapshot. 3. Read `execution_workspace_snapshots`: the abandoned rows are still `pending`. The server log shows `I2_settled` violations for `remote_snapshot_ledger` on each report. **Paperclip version or commit** master at 5c9641f141. ## What Changed - New `server/src/services/recovery/remote-snapshot-ledger-settlement.ts`: - `retireEndedRemoteSnapshotLedgerRow` retires one `pending` row of a `remote_workspace_snapshot` save. It requires that the save has no running invocation, the source run is terminal, and the writer epoch has ended (the workspace epoch or writer moved on). A row with no upload attempt becomes `deleted`, because no object exists. A row with upload attempts becomes `delete_requested`; no reader adopts it, and snapshot retention owns its objects. It writes `lastError` and an `abandonedUpload` provenance record. - `settleAbandonedRemoteSnapshotLedger` is the settle path of the `remote_snapshot_ledger` kind. It writes a `workspace_snapshot.reconciled` activity row. - `settleAbandonedRemoteSnapshotSave` now also retires the save's pending ledger row in the same transaction, when the conditions above already hold. - `ownership-registry.ts`: `remote_snapshot_ledger` now has `settle: settleAbandonedRemoteSnapshotLedger`. - `remote-workspace-snapshot-retention.ts`: the failed-save retry pin no longer holds a `delete_requested` row whose `abandonedUpload` names its own source operation. Retention then deletes its attempted objects after the grace deadline. All other pins stay. - No schema change: `deleted` and `delete_requested` already exist. ## Verification - Red/green: `server/src/__tests__/remote-snapshot-ledger-settlement.test.ts` (3 tests) fails on master (no settle, rows stay `pending`) and passes with this change. - Old-saved/new-resumed: a pending row in the exact shape an older build saved (failed save with `remote_workspace_save_abandoned`, run failed, workspace at epoch 3) settles on the first pass, with no `I2_settled` violation on the second pass. - Keeps an `uploaded` row, a row whose run is live, a row whose save a retry reopened, and a row whose run still holds the current writer epoch. A row with upload attempts goes to `delete_requested` and keeps its deadline. - Restart mid-upload: the same settle that abandons a running save also retires its ledger row. - Red/green: a new retention test runs retention past the deadline. It deletes the objects of a retired row and keeps a failed save's retry pin when its row was not retired. - `cd server && npx vitest run src/__tests__/remote-snapshot-ledger-settlement.test.ts src/__tests__/ownership-reconciliation-report.test.ts src/services/remote-workspace-snapshot-retention.test.ts` passes. - `cd server && npx tsc --noEmit -p tsconfig.json` passes. - Live: a Daytona R1 × K1 wave with K1 inside the initial save. Results are added in a comment below. ## Risks - Low risk. The settle only moves `pending` rows with no published object key. It never touches `uploaded`, `committed` or `superseded` rows. - A retry of the same save reuses the same ledger row. The settle refuses while the save operation is running or pending, while its save work has live or unknown invocations, while the source run is not terminal, and while that run still holds the current writer epoch. A retry is not possible after these hold. - Previous-release story: rows that older builds left `pending` carry forward. The new `remote_snapshot_ledger` settle path retires them on the next reconciler pass when the same conditions hold. - Objects of a retired row with upload attempts are deleted by snapshot retention after the row's grace deadline, and only when no other pin holds the row. Retention rechecks every pin before it deletes. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 --- .../ownership-reconciliation-report.test.ts | 2 +- .../remote-snapshot-ledger-settlement.test.ts | 143 ++++++++++++++++++ .../services/recovery/ownership-registry.ts | 3 +- .../remote-snapshot-ledger-settlement.ts | 93 ++++++++++++ .../workspace-finalize-attempt-settlement.ts | 2 + ...emote-workspace-snapshot-retention.test.ts | 17 +++ .../remote-workspace-snapshot-retention.ts | 9 +- 7 files changed, 266 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/remote-snapshot-ledger-settlement.test.ts create mode 100644 server/src/services/recovery/remote-snapshot-ledger-settlement.ts diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 40e50b67da0..1ae2fd4dbd5 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -151,7 +151,7 @@ describe("report-only ownership reconciliation", () => { expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) - .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "remote_snapshot_ledger", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts new file mode 100644 index 00000000000..6b532613118 --- /dev/null +++ b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts @@ -0,0 +1,143 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { + agents, + companies, + createDb, + executionWorkspaceSnapshots, + executionWorkspaces, + heartbeatRuns, + workspaceOperations, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { + createOwnershipReportState, + runOwnershipReconciliationPass, + type OwnershipReport, +} from "../services/recovery/ownership-invariant-report.js"; +import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); +const minutesAgo = (minutes: number) => new Date(Date.now() - minutes * 60_000); +const kindReport = (report: OwnershipReport, kind: string) => report.kinds.find(entry => entry.kind === kind)!; +const ledgerViolations = (report: OwnershipReport) => + report.violations.filter(violation => violation.kind === "remote_snapshot_ledger"); +const quiesced = { schema: "paperclip.remote-workspace-save-work.v1", activeIds: [], unknown: false, + owners: {}, quiescedAt: minutesAgo(30).toISOString() }; + +describe("remote snapshot ledger settlement", () => { + let database: Awaited> | undefined; + let db: ReturnType; + let companyId: string; + let agentId: string; + + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("remote-snapshot-ledger-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 120_000); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); + }); + + beforeEach(async () => { + await db.execute(sql`truncate execution_workspace_snapshots, activity_log, workspace_operations, execution_workspaces, + heartbeat_runs, agents, companies cascade`); + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Ledger fixture", issuePrefix: "LED" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" }); + }); + + async function run(status: string) { + const id = randomUUID(); + await db.insert(heartbeatRuns).values({ id, companyId, agentId, status, invocationSource: "manual", + ...(status === "running" ? {} : { finishedAt: minutesAgo(2), errorCode: "process_lost" }) }); + return id; + } + + async function workspace(authority: { epoch: number; writerRunId: string | null; dirty: boolean }) { + const id = randomUUID(); + await db.insert(executionWorkspaces).values({ id, companyId, mode: "isolated_workspace", strategyType: "test", name: id, + metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", snapshotId: null, + source: { kind: "git", repoUrl: "https://example.invalid/repo.git", ref: null }, location: null, ...authority } } }); + return id; + } + + async function save(input: { workspaceId: string; runId: string; epoch?: number; status?: string; + saveWork?: Record; snapshot?: Partial; lastFailure?: boolean }) { + const operationId = randomUUID(), snapshotId = randomUUID(), epoch = input.epoch ?? 1; + await db.insert(workspaceOperations).values({ id: operationId, companyId, executionWorkspaceId: input.workspaceId, + heartbeatRunId: input.runId, phase: "remote_workspace_snapshot", status: input.status ?? "failed", + metadata: { epoch, boundary: "initial", generation: 1, saveWork: input.saveWork ?? quiesced, + exportCleanup: { state: "pending" }, + ...(input.lastFailure === false ? {} : { lastFailure: { code: "remote_workspace_save_abandoned", at: minutesAgo(20).toISOString() } }) } }); + await db.insert(executionWorkspaceSnapshots).values({ id: snapshotId, companyId, executionWorkspaceId: input.workspaceId, + sourceOperationId: operationId, sourceRunId: input.runId, generation: 1, epoch, + deleteAfter: new Date(Date.now() + 7 * 86400_000), createdAt: minutesAgo(30), updatedAt: minutesAgo(30), ...input.snapshot }); + return { operationId, snapshotId }; + } + + const ledger = async (id: string) => + (await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, id)))[0]!; + + it("retires a pending row an older build left behind once its save was abandoned and its epoch ended", async () => { + const failed = await run("failed"); + const workspaceId = await workspace({ epoch: 3, writerRunId: null, dirty: false }); + const { snapshotId } = await save({ workspaceId, runId: failed }); + + const state = createOwnershipReportState(); + const first = await runOwnershipReconciliationPass(db, { state }); + const second = await runOwnershipReconciliationPass(db, { state }); + + expect(kindReport(first, "remote_snapshot_ledger")).toMatchObject({ inProgress: 1, settled: 1 }); + expect(ledgerViolations(first)).toEqual([]); + expect(ledgerViolations(second)).toEqual([]); + expect(kindReport(second, "remote_snapshot_ledger").inProgress).toBe(0); + const row = await ledger(snapshotId); + expect(row).toMatchObject({ status: "deleted", deleteAfter: null, lastError: "abandoned_remote_snapshot_upload" }); + expect(row.metadata).toMatchObject({ abandonedUpload: { schema: "paperclip.remote-snapshot-ledger-abandonment.v1", + runId: failed, epoch: 1, workspaceEpoch: 3, uploadAttempted: false, previousStatus: "pending" } }); + }); + + it("leaves objects of an attempted upload to retention, and never retires an uploaded, committed or still-owned row", async () => { + const failed = await run("failed"), live = await run("running"), writer = await run("failed"); + const ended = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const held = await workspace({ epoch: 1, writerRunId: writer, dirty: true }); + const objectKey = `${companyId}/workspace-snapshots/${ended}/x/a.pcws`; + const rows = { + attempted: await save({ workspaceId: ended, runId: failed, snapshot: { metadata: { uploads: [{ objectKey }] } } }), + uploaded: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: failed, + snapshot: { status: "uploaded", objectKey } }), + liveRun: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: live }), + reopened: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: failed, + status: "running", lastFailure: false }), + currentWriter: await save({ workspaceId: held, runId: writer }), + }; + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_ledger") }); + + expect(await ledger(rows.attempted.snapshotId)).toMatchObject({ status: "delete_requested", + metadata: { uploads: [{ objectKey }], abandonedUpload: { uploadAttempted: true } } }); + expect((await ledger(rows.attempted.snapshotId)).deleteAfter).not.toBeNull(); + expect((await ledger(rows.uploaded.snapshotId)).status).toBe("uploaded"); + expect((await ledger(rows.liveRun.snapshotId)).status).toBe("pending"); + expect((await ledger(rows.reopened.snapshotId)).status).toBe("pending"); + expect((await ledger(rows.currentWriter.snapshotId)).status).toBe("pending"); + }); + + it("retires the ledger row in the same settle that abandons a save a restart stopped mid-upload", async () => { + const failed = await run("failed"); + const workspaceId = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const { operationId, snapshotId } = await save({ workspaceId, runId: failed, status: "running", lastFailure: false }); + + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_save") }); + + expect(kindReport(report, "remote_snapshot_save")).toMatchObject({ settled: 1 }); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, operationId)); + expect(operation!.metadata).toMatchObject({ lastFailure: { code: "remote_workspace_save_abandoned" } }); + expect(await ledger(snapshotId)).toMatchObject({ status: "deleted", + metadata: { abandonedUpload: { runId: failed, operationId } } }); + }); +}); diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index d576c4f48ea..fcf8b9195c3 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -39,6 +39,7 @@ import { import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; import { settleAbandonedRemoteSnapshotSave, settleAbandonedWorkspaceCommandOperation, settleAbandonedWorkspaceFinalizeAttempt } from "./workspace-finalize-attempt-settlement.js"; import { settlePreparationOnlyRemoteWorkspaceEpoch } from "./remote-workspace-unsaved-files-settlement.js"; +import { settleAbandonedRemoteSnapshotLedger } from "./remote-snapshot-ledger-settlement.js"; import { LOCAL_WORKSPACE_COMMAND_PHASES, runtimeControlOperationOwnerStamp, settleRuntimeControlOperation } from "../workspace-operations.js"; import { copybackOwnerHoldOwner, settleCopybackOwnerHold } from "./copyback-owner-hold.js"; import { applyObservedFenceProviderProof, observedWorkspaceFenceOwner, settleObservedWorkspaceFence } from "./observed-workspace-fence.js"; @@ -946,7 +947,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ }, { id: "remote_snapshot_ledger", title: "Remote snapshot ledger", obligation: "workspace_preservation", - storage: "database", settle: null, invariantChecks: [2], + storage: "database", settle: settleAbandonedRemoteSnapshotLedger, invariantChecks: [2], listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: executionWorkspaceSnapshots.id, companyId: executionWorkspaceSnapshots.companyId, sourceRunId: executionWorkspaceSnapshots.sourceRunId, updatedAt: executionWorkspaceSnapshots.updatedAt }) diff --git a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts new file mode 100644 index 00000000000..f9e8b26bd27 --- /dev/null +++ b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts @@ -0,0 +1,93 @@ +import { and, eq, isNull, sql } from "drizzle-orm"; +import { executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import type { OwnerProof } from "../owner-proof.js"; +import { readSaveWork } from "../remote-workspaces.js"; +import { notifyRemoteWorkspaceSnapshots } from "../remote-workspace-snapshot-retention.js"; +import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; + +export const ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON = "abandoned_remote_snapshot_upload"; +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; + +type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; + +export async function retireEndedRemoteSnapshotLedgerRow(tx: Db, input: { + companyId: string; snapshotId: string; now: Date; proof: string; +}): Promise<{ retired: true; snapshot: Snapshot; status: "deleted" | "delete_requested" } | { retired: false; reason: string }> { + const [snapshot] = await tx.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.id, input.snapshotId))); + if (!snapshot || snapshot.status !== "pending") return { retired: false, reason: "ledger_no_longer_pending" }; + if (snapshot.objectKey !== null) return { retired: false, reason: "upload_published" }; + if (!snapshot.sourceRunId) return { retired: false, reason: "owner_run_missing" }; + const [operation] = await tx.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, snapshot.companyId), eq(workspaceOperations.id, snapshot.sourceOperationId))).for("update"); + if (operation) { + if (operation.phase !== "remote_workspace_snapshot") return { retired: false, reason: "not_a_workspace_save" }; + if (operation.status === "running" || operation.status === "pending") return { retired: false, reason: "save_in_progress" }; + const work = readSaveWork(operation.metadata); + if (work && (work.unknown || work.activeIds.length)) return { retired: false, reason: "save_work_unsettled" }; + } + const [run] = await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, snapshot.companyId), eq(heartbeatRuns.id, snapshot.sourceRunId))); + if (run && !TERMINAL_RUN_STATUSES.includes(run.status)) return { retired: false, reason: "owner_run_active" }; + const [workspace] = await tx.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, snapshot.companyId), eq(executionWorkspaces.id, snapshot.executionWorkspaceId))); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (authority && authority.epoch === snapshot.epoch && authority.writerRunId === snapshot.sourceRunId) { + return { retired: false, reason: "writer_epoch_current" }; + } + const uploads = snapshot.metadata?.uploads; + const attempted = Array.isArray(uploads) ? uploads.length > 0 : uploads !== undefined && uploads !== null; + const status = attempted ? "delete_requested" as const : "deleted" as const; + const provenance = { schema: "paperclip.remote-snapshot-ledger-abandonment.v1", reason: ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON, + proof: input.proof, previousStatus: snapshot.status, runId: snapshot.sourceRunId, operationId: snapshot.sourceOperationId, + epoch: snapshot.epoch, workspaceEpoch: authority?.epoch ?? null, uploadAttempted: attempted, reconciledAt: input.now.toISOString() }; + await notifyRemoteWorkspaceSnapshots(tx, `workspace:${snapshot.executionWorkspaceId}`); + const [row] = await tx.update(executionWorkspaceSnapshots).set({ status, + deleteAfter: status === "deleted" ? null : snapshot.deleteAfter, lastError: ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON, + metadata: { ...snapshot.metadata, abandonedUpload: provenance }, updatedAt: input.now }) + .where(and(eq(executionWorkspaceSnapshots.id, snapshot.id), eq(executionWorkspaceSnapshots.status, "pending"), + isNull(executionWorkspaceSnapshots.objectKey), + sql`${executionWorkspaceSnapshots.metadata} is not distinct from ${snapshot.metadata === null ? null : JSON.stringify(snapshot.metadata)}::jsonb`)) + .returning(); + if (!row) return { retired: false, reason: "ledger_no_longer_pending" }; + return { retired: true, snapshot: row, status }; +} + +export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "run" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; + const companyId = record.companyId; + let publication: ActivityPublication | undefined; + const outcome = await db.transaction(async (transaction): Promise => { + const tx = transaction as unknown as Db; + const result = await retireEndedRemoteSnapshotLedgerRow(tx, { companyId, snapshotId: record.recordId, + now: context.now, proof: context.proof.reason }); + if (!result.retired) { + return { settled: false, reason: result.reason, + ...(["upload_published", "not_a_workspace_save"].includes(result.reason) ? { watch: false } : {}) }; + } + const activity = await persistActivity(tx, { + companyId, actorType: "system", actorId: "ownership-reconciler", + action: "workspace_snapshot.reconciled", entityType: "execution_workspace", entityId: result.snapshot.executionWorkspaceId, + runId: result.snapshot.sourceRunId, details: { snapshotId: result.snapshot.id, status: result.status, + ...(result.snapshot.metadata?.abandonedUpload as Record) }, + }); + publication = activity.publication; + return { settled: true, reason: context.proof.reason }; + }); + if (outcome.settled && publication) publishActivity(publication); + return outcome; +} + +export async function retireAbandonedSaveLedgerRows(tx: Db, input: { + companyId: string; operationId: string; now: Date; proof: string; +}) { + const rows = await tx.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.sourceOperationId, input.operationId), + eq(executionWorkspaceSnapshots.status, "pending"))); + for (const row of rows) { + await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + } +} diff --git a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts index 815102ac22b..5fcb5d945fd 100644 --- a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts +++ b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts @@ -9,6 +9,7 @@ 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"; import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; +import { retireAbandonedSaveLedgerRows } from "./remote-snapshot-ledger-settlement.js"; export const ABANDONED_WORKSPACE_FINALIZE_REASON = "abandoned_workspace_finalization"; export const ABANDONED_WORKSPACE_COMMAND_REASON = "abandoned_workspace_command"; @@ -164,6 +165,7 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre }).where(and(eq(workspaceOperations.id, operation.id), eq(workspaceOperations.status, operation.status))) .returning({ id: workspaceOperations.id }); if (!row) return { settled: false, reason: "attempt_no_longer_in_progress" }; + await retireAbandonedSaveLedgerRows(tx, { companyId: operation.companyId, operationId: operation.id, now, proof: context.proof.reason }); if (operation.executionWorkspaceId) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); await tx.update(issueRecoveryActions).set({ diff --git a/server/src/services/remote-workspace-snapshot-retention.test.ts b/server/src/services/remote-workspace-snapshot-retention.test.ts index 079d5a0206f..60986fff325 100644 --- a/server/src/services/remote-workspace-snapshot-retention.test.ts +++ b/server/src/services/remote-workspace-snapshot-retention.test.ts @@ -125,6 +125,23 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () => return { row: row!, companyId, workspaceId, issueId: issue!.id, operationId, keys }; } const service = () => remoteWorkspaceSnapshotRetentionService(db, { storage, now: () => now }); + it("deletes the attempted uploads of an abandoned save once its ledger row is retired, and keeps a failed save's retry pin", async () => { + const retired = await seed("pending"), held = await seed("pending"); + for (const f of [retired, held]) { + await db.insert(workspaceOperations).values({ id: f.operationId, companyId: f.companyId, executionWorkspaceId: f.workspaceId, + phase: "remote_workspace_snapshot", status: "failed", + metadata: { epoch: 1, generation: 1, lastFailure: { code: "remote_workspace_save_abandoned" } } }); + } + await db.update(executionWorkspaceSnapshots).set({ status: "delete_requested", objectKey: null, + metadata: { ...retired.row.metadata, abandonedUpload: { operationId: retired.operationId } } }) + .where(eq(executionWorkspaceSnapshots.id, retired.row.id)); + await db.update(executionWorkspaceSnapshots).set({ objectKey: null }).where(eq(executionWorkspaceSnapshots.id, held.row.id)); + expect(await service().tick()).toMatchObject({ deleted: 1, pinned: 1, failed: 0 }); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, retired.row.id)))[0]!.status).toBe("deleted"); + for (const key of retired.keys) expect((await storage.headObject(retired.companyId, key)).exists).toBe(false); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, held.row.id)))[0]!.status).toBe("pending"); + for (const key of held.keys) expect((await storage.headObject(held.companyId, key)).exists).toBe(true); + }); it("deletes every recorded attempt and winning object after grace", async () => { const f = await seed(); const owner = service(); expect(await owner.tick()).toMatchObject({ deleted: 1, failed: 0 }); diff --git a/server/src/services/remote-workspace-snapshot-retention.ts b/server/src/services/remote-workspace-snapshot-retention.ts index d4722c07d01..2003dce614e 100644 --- a/server/src/services/remote-workspace-snapshot-retention.ts +++ b/server/src/services/remote-workspace-snapshot-retention.ts @@ -17,6 +17,12 @@ export function notifyRemoteWorkspaceSnapshots(transaction: object, hint?: strin return notifyDeliveryWork(transaction, DELIVERY_QUEUES.remoteWorkspaceSnapshots, hint); } +function abandonedUploadOperationId(row: Snapshot) { + if (row.status !== "delete_requested") return null; + const abandoned = row.metadata?.abandonedUpload as { operationId?: unknown } | undefined; + return typeof abandoned?.operationId === "string" && abandoned.operationId === row.sourceOperationId ? abandoned.operationId : null; +} + async function pinned(db: ReadDb, row: Snapshot) { const [workspace] = await db.select({ id: executionWorkspaces.id, sourceIssueId: executionWorkspaces.sourceIssueId, metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( @@ -37,7 +43,8 @@ async function pinned(db: ReadDb, row: Snapshot) { // A failed save keeps its snapshot pinned for its retry, as while it ran, // until the final save retires it. sql`(${workspaceOperations.status} = 'failed' and ${workspaceOperations.metadata} ? 'lastFailure' - and ${workspaceOperations.metadata}->'retiredSave' is null)`, + and ${workspaceOperations.metadata}->'retiredSave' is null + and ${workspaceOperations.id}::text is distinct from ${abandonedUploadOperationId(row)})`, sql`(${workspaceOperations.metadata}->'saveWork'->>'schema' = 'paperclip.remote-workspace-save-work.v1' and (${workspaceOperations.metadata}#>'{saveWork,unknown}' is distinct from 'false'::jsonb or case when jsonb_typeof(${workspaceOperations.metadata}#>'{saveWork,activeIds}') = 'array'