From d0c30e69a9cea912db7c5290c5d585fdd63ec228 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 10:56:52 -0600 Subject: [PATCH] fix(storage): abort a snapshot upload a restart cut off once its save attempt is superseded (#20542) 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 as encrypted snapshots in object storage. Snapshots over 16 MiB use S3 multipart uploads. > - A server restart can cut off a snapshot upload in the middle of an `UploadPart` request. > - The server aborts a multipart upload only in-process, on an error or an abort signal. A SIGKILL can never do that. On SIGTERM, the drain step timed out while the store held the part, and the process exited without `AbortMultipartUpload`. > - The incomplete upload then stays in the bucket. Nothing records its upload id and nothing finds it again, so incomplete uploads pile up and the operator pays for their parts. > - This pull request aborts the incomplete uploads of each save attempt when that attempt can no longer win, with no polling and no schema change. > - The benefit is that no incomplete upload outlives its save attempt. ## Linked Issues or Issue Description **What happened?** A restart-chaos wave (Daytona, R1 workload, storage fault ST4, `--storage-trigger snapshot-part`, faults K1 and K2, master `fea4530a30`) passed all 3 tasks. All 3 committed snapshots read back intact. But the storage audit found 3 incomplete multipart uploads. Each one was the exact upload that a restart cut off in the middle of `UploadPart`. The retried save then published a different attempt of the same snapshot. **Expected behavior** When a save attempt can no longer become the snapshot's object, its incomplete multipart uploads are aborted. The storage audit shows 0 incomplete uploads after the wave. **Steps to reproduce** 1. Run a snapshot save larger than 16 MiB against S3. 2. Kill the server (SIGKILL), or send SIGTERM while the store holds an `UploadPart` request. 3. Let the save retry and publish. 4. List the bucket's multipart uploads. The cut-off attempt's upload is still there. **Paperclip version or commit** master `fea4530a30` and later. **Relevant logs or output** `shutdown step timed out; continuing shutdown without it {"step":"delivery_work_stop","durationMs":20001}`, then `Paperclip did not stop within 30000ms; sending SIGKILL`. ## What Changed - The S3 storage provider and the storage service get `abortIncompleteUploads(objectKey)`. It lists the multipart uploads under that exact key (`ListMultipartUploads` with the key as the prefix), skips other keys that only share the prefix, and aborts each one. `NoSuchUpload` counts as already done. The local-disk provider has no multipart uploads and does nothing. - Save publication: the save journals every attempt's object key in `metadata.uploads` before any remote I/O. When one attempt moves the snapshot from `pending` to `uploaded`, no other attempt can win any more. The winning process now aborts the incomplete uploads of every other journaled attempt key. Each abort is best effort and bounded to 5 s. A snapshot with one attempt sends no extra request. - A losing uploader that is still live: if its upload fails after another attempt has published the snapshot, it now returns the published snapshot (after it verifies it) instead of the error. So an abort from the winner never fails a save that is still running in another process. - Snapshot retention: before it deletes a retired snapshot's objects (this includes saves that #20502 retires as abandoned), retention aborts the incomplete uploads of each attempt key. A store that refuses the listing (for example a missing `s3:ListBucketMultipartUploads` permission) logs a warning and does not block the deletion. - Abandoned saves: when the ownership reconciler retires an abandoned save (`remote_snapshot_save` and `remote_snapshot_ledger` settlement, from #20502), it now aborts the incomplete uploads of every journaled attempt key after its transaction commits. The owner run is terminal and the save work is settled, so no live uploader can own those uploads. This is the durable path for SIGKILL: the reconciler pass after boot settles the dead save and cleans its uploads. - SIGTERM drain: the multipart upload code keeps an in-process set of running uploads. Shutdown aborts them right away when `delivery_work_stop` times out. It also aborts them after the heartbeat run drain, before the finalizer and workspace-recovery waits, after a 5 s grace for them to finish. It waits at most 10 s for the `AbortMultipartUpload` calls. Shutdown does nothing extra when no upload runs. - No polling and no periodic sweep. All triggers are events that already exist: publication, abandonment settlement, retention of a retired snapshot, and shutdown. Previous-release story (`rules/upgrade-safe-run-state.mdc`): this adds no persisted state and changes no saved shape. A snapshot that the previous release saved with cut-off attempts in `metadata.uploads` carries forward. Its incomplete uploads are aborted when the new release publishes a retry of that snapshot, or when retention retires the snapshot. ## Verification - Red before, green after (all fail on master, pass with this change): - `server/src/services/remote-workspace-snapshot-storage.test.ts`: "aborts the incomplete upload of an attempt a restart cut off once a retried save publishes" and "adopts the published save when another uploader aborts this attempt's upload". - `server/src/services/remote-workspace-snapshot-retention.test.ts`: "aborts each attempt's incomplete multipart upload before deleting a retired snapshot's objects". A second test proves that a refused listing does not block retention. - `server/src/__tests__/storage-s3-multipart-shutdown.test.ts`: file and streamed uploads whose part the store holds are aborted at shutdown, and the provider aborts only the exact key across listing pages. - `server/src/__tests__/remote-snapshot-ledger-settlement.test.ts`: "aborts the incomplete multipart uploads of every attempt when the reconciler retires an abandoned save" (both settlement kinds; a key outside the snapshot's prefix is skipped). - `tests/restart-chaos/storage-orphan-uploads.test.ts` (MinIO): the real provider aborts both cut-off uploads of one attempt key, and keeps a sibling attempt and a longer key that shares the prefix. - `cd server && npx vitest run src/__tests__/storage-s3 src/storage src/__tests__/remote-snapshot-ledger-settlement.test.ts src/services/remote-workspace-snapshot src/services/remote-workspace-transport.test.ts src/__tests__/remote-workspace-file-resources.test.ts`: 19 files, 517 tests pass; with `src/services/recovery`, `src/__tests__/ownership*` and `heartbeat-workspace-busy.test.ts`, 32 files and 580 tests pass. - `cd server && npx tsc --noEmit` and `node scripts/typecheck-test-projects.mjs tests/restart-chaos` pass. - Live proof, Daytona R1, ST4, `--storage-trigger snapshot-part`, K1/K2, seed 1033, the evidence wave's flags, image `paperclip-daytona-runner@sha256:05a4a092…` (controllerMatch: match): - Before (master `fea4530a30`): `w-20261110T130541-s1033`: 3/3 pass, **3 incomplete multipart uploads**. - `w-20261110T134623-s1033` on an earlier revision of this PR (publish + retention hooks only): 2 incomplete uploads remained. Those saves were abandoned, not retried, so neither hook ran. This led to the settlement hook. - After (head `8bc8fc66d0`): `w-20261110T141144-s1033`: 3/3 tasks pass, exit 0, every fired restart proved it cut the held `UploadPart`, the reconciler settled 3 abandoned saves, shutdown logged one in-flight abort, and the storage audit shows **0 incomplete multipart uploads** and 3 committed snapshots that read back intact. ## Risks - A live uploader in another process can lose its incomplete upload when a different attempt publishes first. That uploader now adopts the published snapshot, so its save still succeeds. - Narrow gap: an uploader that recorded its attempt but had not yet sent `CreateMultipartUpload` when the winner published, and that then crashes during its upload, leaves an incomplete upload until retention retires the snapshot. An S3 lifecycle rule for incomplete uploads stays a good backstop. - Stores without `ListMultipartUploads` permission log a warning and keep the previous behavior. - Shutdown can take up to 10 s longer, and only when an upload is still running after the drain. ## 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 - [x] All Paperclip CI gates are green - [x] 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 --- .../remote-snapshot-ledger-settlement.test.ts | 30 +++- .../snapshot-attempt-upload-abort.test.ts | 33 ++++ .../storage-s3-multipart-shutdown.test.ts | 151 ++++++++++++++++++ server/src/index.ts | 15 +- .../remote-snapshot-ledger-settlement.ts | 22 ++- .../workspace-finalize-attempt-settlement.ts | 9 +- ...emote-workspace-snapshot-retention.test.ts | 18 ++- .../remote-workspace-snapshot-retention.ts | 36 +++++ .../remote-workspace-snapshot-storage.test.ts | 35 ++++ .../remote-workspace-snapshot-storage.ts | 17 +- server/src/shutdown.ts | 5 + server/src/storage/s3-multipart.ts | 29 +++- server/src/storage/s3-provider.ts | 29 ++++ server/src/storage/service.ts | 6 + server/src/storage/types.ts | 3 + .../storage-orphan-uploads.test.ts | 54 +++++++ 16 files changed, 476 insertions(+), 16 deletions(-) create mode 100644 server/src/__tests__/snapshot-attempt-upload-abort.test.ts create mode 100644 server/src/__tests__/storage-s3-multipart-shutdown.test.ts create mode 100644 tests/restart-chaos/storage-orphan-uploads.test.ts diff --git a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts index 6b532613118..e4df204ce36 100644 --- a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts +++ b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; import { agents, @@ -19,6 +19,11 @@ import { } from "../services/recovery/ownership-invariant-report.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +const aborted = vi.hoisted(() => ({ keys: [] as string[] })); +vi.mock("../storage/index.js", () => ({ + getStorageService: () => ({ abortIncompleteUploads: async (_companyId: string, key: string) => { aborted.keys.push(key); return 1; } }), +})); + 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)!; @@ -44,7 +49,7 @@ describe("remote snapshot ledger settlement", () => { 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(); + companyId = randomUUID(); agentId = randomUUID(); aborted.keys = []; 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" }); }); @@ -140,4 +145,25 @@ describe("remote snapshot ledger settlement", () => { expect(await ledger(snapshotId)).toMatchObject({ status: "deleted", metadata: { abandonedUpload: { runId: failed, operationId } } }); }); + + it("aborts the incomplete multipart uploads of every attempt when the reconciler retires an abandoned save", async () => { + const failed = await run("failed"); + const ledgerWorkspace = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const ledgerId = randomUUID(), ledgerKeys = ["a", "b"].map(name => `${companyId}/workspace-snapshots/${ledgerWorkspace}/${ledgerId}/${name}.pcws`); + await save({ workspaceId: ledgerWorkspace, runId: failed, + snapshot: { id: ledgerId, metadata: { uploads: [...ledgerKeys.map(objectKey => ({ objectKey })), { objectKey: `${companyId}/elsewhere/c.pcws` }] } } }); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_ledger") }); + expect(aborted.keys).toEqual(ledgerKeys); + + aborted.keys = []; + const saveWorkspace = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const saveId = randomUUID(), saveKey = `${companyId}/workspace-snapshots/${saveWorkspace}/${saveId}/cut.pcws`; + await save({ workspaceId: saveWorkspace, runId: failed, status: "running", lastFailure: false, + snapshot: { id: saveId, metadata: { uploads: [{ objectKey: saveKey }] } } }); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_save") }); + expect(await ledger(saveId)).toMatchObject({ status: "delete_requested" }); + expect(aborted.keys).toEqual([saveKey]); + }); }); diff --git a/server/src/__tests__/snapshot-attempt-upload-abort.test.ts b/server/src/__tests__/snapshot-attempt-upload-abort.test.ts new file mode 100644 index 00000000000..775b8873297 --- /dev/null +++ b/server/src/__tests__/snapshot-attempt-upload-abort.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from "vitest"; +import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js"; +import type { StorageService } from "../storage/types.js"; +import { abortSnapshotAttemptUploads } from "../services/remote-workspace-snapshot-retention.js"; + +describe("aborting a snapshot's attempt uploads", () => { + it("shares one deadline across every attempt key of a hanging store", async () => { + const started: string[] = []; + const storage = { + abortIncompleteUploads: (_companyId: string, key: string, options?: { signal?: AbortSignal }) => { + started.push(key); + return new Promise((_resolve, reject) => { + options?.signal?.addEventListener("abort", () => reject(options.signal!.reason), { once: true }); + }); + }, + } as unknown as StorageService; + const keys = Array.from({ length: 20 }, (_, index) => `company/workspace-snapshots/ws/s/${index}.pcws`); + const begun = Date.now(); + expect(await abortSnapshotAttemptUploads(storage, "company", keys)).toBe(0); + const elapsed = Date.now() - begun; + expect(elapsed).toBeGreaterThanOrEqual(STORAGE_FAILURE_CLEANUP_MS - 100); + expect(elapsed).toBeLessThan(2 * STORAGE_FAILURE_CLEANUP_MS); + expect(started).toHaveLength(8); + }, 3 * STORAGE_FAILURE_CLEANUP_MS); + + it("aborts every key of a responsive store and counts each abort", async () => { + const seen: string[] = []; + const storage = { abortIncompleteUploads: async (_companyId: string, key: string) => { seen.push(key); return 1; } } as unknown as StorageService; + const keys = Array.from({ length: 20 }, (_, index) => `k${index}`); + expect(await abortSnapshotAttemptUploads(storage, "company", keys)).toBe(20); + expect(seen.sort()).toEqual([...keys].sort()); + }); +}); diff --git a/server/src/__tests__/storage-s3-multipart-shutdown.test.ts b/server/src/__tests__/storage-s3-multipart-shutdown.test.ts new file mode 100644 index 00000000000..54f9065828e --- /dev/null +++ b/server/src/__tests__/storage-s3-multipart-shutdown.test.ts @@ -0,0 +1,151 @@ +import { AbortMultipartUploadCommand, CompleteMultipartUploadCommand, CreateMultipartUploadCommand, ListMultipartUploadsCommand, S3Client, UploadPartCommand } from "@aws-sdk/client-s3"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { Readable } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { abortInFlightMultipartUploads, putS3Multipart, putS3MultipartFile } from "../storage/s3-multipart.js"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; +import { afterStalledShutdownStep, runBoundedShutdownStep } from "../shutdown.js"; + +const MIB = 1024 * 1024; + +function heldPartClient() { + const client = new S3Client({ region: "us-east-1" }), aborted: string[] = []; + let partStarted!: () => void; + const started = new Promise(resolve => { partStarted = resolve; }); + vi.spyOn(client, "send").mockImplementation(async (command: any, options?: any) => { + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "held-upload" }; + if (command instanceof UploadPartCommand) { + partStarted(); + const signal = options?.abortSignal as AbortSignal; + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + } + if (command instanceof AbortMultipartUploadCommand) { aborted.push(command.input.UploadId!); return {}; } + throw new Error(`unexpected ${command.constructor.name}`); + }); + return { client, aborted, started }; +} + +describe("multipart uploads in flight at shutdown", () => { + let root: string | null = null; + afterEach(async () => { vi.restoreAllMocks(); if (root) await fs.rm(root, { recursive: true, force: true }); root = null; }); + + it("has nothing to abort when no upload is running", async () => { + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(0); + }); + + it("aborts a file upload whose part the store is holding", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const { client, aborted, started } = heldPartClient(); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }).catch(error => error); + await started; + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(1); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(0); + }); + + it("lets an upload that finishes inside the grace complete without aborting it", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const client = new S3Client({ region: "us-east-1" }), sent: string[] = []; + let release!: () => void, partStarted!: () => void; + const released = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { partStarted = resolve; }); + vi.spyOn(client, "send").mockImplementation(async (command: any) => { + sent.push(command.constructor.name); + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "graceful-upload" }; + if (command instanceof UploadPartCommand) { partStarted(); await released; return { ETag: `part-${command.input.PartNumber}` }; } + if (command instanceof CompleteMultipartUploadCommand) return {}; + throw new Error(`unexpected ${command.constructor.name}`); + }); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }); + await started; + const draining = abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs: 5_000 }); + setTimeout(release, 50); + expect(await draining).toBe(0); + expect(await saving).toEqual({ integrity: "server_verified_md5" }); + expect(sent).toContain("CompleteMultipartUploadCommand"); + expect(sent).not.toContain("AbortMultipartUploadCommand"); + }); + + it("aborts an upload still held after the grace and counts it once", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const { client, aborted, started } = heldPartClient(); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }).catch(error => error); + await started; + const begun = Date.now(); + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs: 100 })).toBe(1); + expect(Date.now() - begun).toBeGreaterThanOrEqual(90); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + }); + + it("aborts a streamed upload whose part the store is holding", async () => { + const { client, aborted, started } = heldPartClient(); + const body = Readable.from((async function* () { for (let index = 0; index < 272; index++) yield Buffer.alloc(64 * 1024, 3); })()); + const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body }).catch(error => error); + await started; + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(1); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + }); +}); + +describe("aborting a key's incomplete multipart uploads", () => { + afterEach(() => { vi.restoreAllMocks(); vi.unstubAllEnvs(); }); + + it("aborts every upload of the exact key across listing pages and leaves neighboring keys alone", async () => { + vi.stubEnv("AWS_ACCESS_KEY_ID", "synthetic-access-key"); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", "synthetic-secret-key"); + const aborted: string[] = [], listed: Array<{ Prefix?: string; KeyMarker?: string; UploadIdMarker?: string }> = []; + const send = vi.spyOn(S3Client.prototype, "send").mockImplementation(async (command: any) => { + if (command instanceof ListMultipartUploadsCommand) { + listed.push(command.input); + if (!command.input.KeyMarker) return { IsTruncated: true, NextKeyMarker: "p/c/a.pcws", NextUploadIdMarker: "u2", + Uploads: [{ Key: "p/c/a.pcws", UploadId: "u1" }, { Key: "p/c/a.pcws", UploadId: "u2" }] }; + return { IsTruncated: false, Uploads: [{ Key: "p/c/a.pcws.next", UploadId: "u3" }, { Key: "p/c/a.pcws", UploadId: "u4" }] }; + } + if (command instanceof AbortMultipartUploadCommand) { + if (command.input.UploadId === "u4") throw Object.assign(new Error("gone"), { name: "NoSuchUpload" }); + aborted.push(`${command.input.Key}#${command.input.UploadId}`); return {}; + } + throw new Error(`unexpected ${command.constructor.name}`); + }); + const provider = createS3StorageProvider({ bucket: "bucket", region: "us-east-1", endpoint: "http://127.0.0.1:9", prefix: "p" }); + expect(await provider.abortIncompleteUploads!({ objectKey: "c/a.pcws" })).toBe(2); + expect(listed.map(input => [input.Prefix, input.KeyMarker, input.UploadIdMarker])) + .toEqual([["p/c/a.pcws", undefined, undefined], ["p/c/a.pcws", "p/c/a.pcws", "u2"]]); + expect(aborted).toEqual(["p/c/a.pcws#u1", "p/c/a.pcws#u2"]); + send.mockRestore(); + }); +}); + +describe("shutdown steps that stall on a held upload", () => { + const log = { info: () => {}, error: () => {} }; + + it("aborts in-flight uploads after a step that timed out or failed, and not after one that completed", async () => { + const onStalled = vi.fn(async () => {}); + const held = new Promise(() => {}); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 5, run: () => held, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(1); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 5, + run: async () => { throw new Error("stop failed"); }, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(2); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 1000, run: async () => {}, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 7f13f2f2dd1..8e785370055 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -195,6 +195,7 @@ import { buildRuntimeApiCandidateUrls, choosePrimaryRuntimeApiUrl, resolveLocalR import { isLoopbackHost, rewriteLoopbackUrlPort } from "./url-utils.js"; import { createPluginWorkerManager } from "./services/plugin-worker-manager.js"; import { createStorageServiceFromConfig } from "./storage/index.js"; +import { abortInFlightMultipartUploads } from "./storage/s3-multipart.js"; import { printStartupBanner } from "./startup-banner.js"; import { getBoardClaimWarningUrl, initializeBoardClaimChallenge } from "./board-claim.js"; import { maybePersistWorktreeRuntimePorts } from "./worktree-config.js"; @@ -203,6 +204,7 @@ import { conflict } from "./errors.js"; import { ensureDecisionSigningSecret } from "./services/decision-signing.js"; import { createDecisionRetentionNotifyOriginAgent, createDecisionWakeOriginAgent } from "./services/decision-wakeup.js"; import { + afterStalledShutdownStep, closeHttpListenerForShutdown, coordinateHeartbeatSchedulerShutdown, createPreDrainShutdownBudget, @@ -2044,6 +2046,11 @@ async function startServerWithDatabaseTeardown( ); } + const abortStuckMultipartUploads = async (signal: "SIGINT" | "SIGTERM", graceMs = 0) => { + const abortedUploads = await abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs }); + if (abortedUploads) logger.info({ signal, abortedUploads }, "aborted in-flight multipart uploads before exit"); + }; + const performShutdown = async ( signal: "SIGINT" | "SIGTERM", exitProcess: boolean, @@ -2071,10 +2078,11 @@ async function startServerWithDatabaseTeardown( unsubscribeRetainedNativeSessionCleanup?.(); // B can accept Stop/answers while A still owns an execution. Preserve the // owner command workers and their transport until the handoff below ends. - await preDrain.run("delivery_work_stop", () => deliveryWork.stop(deploymentHints ? new Set([ + const deliveryWorkStop = await preDrain.run("delivery_work_stop", () => deliveryWork.stop(deploymentHints ? new Set([ DELIVERY_QUEUES.nativeCancellation, DELIVERY_QUEUES.nativeRuntimeResponses, DELIVERY_QUEUES.question, DELIVERY_QUEUES.deploymentNotifications, DELIVERY_QUEUES.liveInvalidations, ]) : undefined)); + await afterStalledShutdownStep(deliveryWorkStop, () => abortStuckMultipartUploads(signal)); await preDrain.run("run_evidence_migration_stop", async () => { await runEvidenceMigrationWork?.stop(); }); @@ -2083,7 +2091,8 @@ async function startServerWithDatabaseTeardown( await skillVersionRetentionWorker?.dispose(); }); const workspaceRecoveryStop = workspaceRecoveryWork.stop(); - await preDrain.run("workspace_recovery_stop", () => workspaceRecoveryStop); + await afterStalledShutdownStep(await preDrain.run("workspace_recovery_stop", () => workspaceRecoveryStop), + () => abortStuckMultipartUploads(signal)); unregisterTerminalWorkspaceProbe?.(); unsubscribeNativeWorkspaceSyncRetention?.(); @@ -2134,6 +2143,8 @@ async function startServerWithDatabaseTeardown( } } + await abortStuckMultipartUploads(signal, 5_000); + if (!skipHeartbeatDrain) { await drainRunExecutionFinalizersForShutdown({ signal, diff --git a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts index f9e8b26bd27..9a5dbe2bf94 100644 --- a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts +++ b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts @@ -4,7 +4,9 @@ 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 { abortSnapshotAttemptUploads, notifyRemoteWorkspaceSnapshots, snapshotAttemptKeys } from "../remote-workspace-snapshot-retention.js"; +import { getStorageService } from "../../storage/index.js"; +import type { StorageService } from "../../storage/types.js"; import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; export const ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON = "abandoned_remote_snapshot_upload"; @@ -55,11 +57,20 @@ export async function retireEndedRemoteSnapshotLedgerRow(tx: Db, input: { return { retired: true, snapshot: row, status }; } +export async function abortRetiredSnapshotUploads(snapshots: Snapshot[], storage?: StorageService) { + const attempted = snapshots.filter(snapshot => snapshotAttemptKeys(snapshot).length); + if (!attempted.length) return; + let service: StorageService; + try { service = storage ?? getStorageService(); } catch { return; } + for (const snapshot of attempted) await abortSnapshotAttemptUploads(service, snapshot.companyId, snapshotAttemptKeys(snapshot)); +} + export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProgressRecord, - context: { now: Date; proof: OwnerProof }): Promise { + context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise { if (record.owner.kind !== "run" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; const companyId = record.companyId; let publication: ActivityPublication | undefined; + let retired: Snapshot | 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, @@ -75,9 +86,11 @@ export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProg ...(result.snapshot.metadata?.abandonedUpload as Record) }, }); publication = activity.publication; + retired = result.snapshot; return { settled: true, reason: context.proof.reason }; }); if (outcome.settled && publication) publishActivity(publication); + if (outcome.settled && retired) await abortRetiredSnapshotUploads([retired], options.storage); return outcome; } @@ -87,7 +100,10 @@ export async function retireAbandonedSaveLedgerRows(tx: Db, input: { 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"))); + const retired: Snapshot[] = []; for (const row of rows) { - await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + const result = await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + if (result.retired) retired.push(result.snapshot); } + return retired; } diff --git a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts index 5fcb5d945fd..1fbe10195e2 100644 --- a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts +++ b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts @@ -9,7 +9,8 @@ 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"; +import type { StorageService } from "../../storage/types.js"; +import { abortRetiredSnapshotUploads, 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"; @@ -112,8 +113,9 @@ function abandonedWorkspaceOperationSettle(spec: { phases: readonly string[]; st export const ABANDONED_REMOTE_SNAPSHOT_SAVE_REASON = "abandoned_remote_snapshot_save"; export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgressRecord, - context: { now: Date; proof: OwnerProof }): Promise { + context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise { let publication: ActivityPublication | undefined; + let retired: Awaited> = []; const outcome = await db.transaction(async (transaction): Promise => { const tx = transaction as unknown as Db; const [operation] = await tx.select().from(workspaceOperations) @@ -165,7 +167,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 }); + retired = 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({ @@ -185,5 +187,6 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre return { settled: true, reason: context.proof.reason }; }); if (outcome.settled && publication) publishActivity(publication); + if (outcome.settled) await abortRetiredSnapshotUploads(retired, options.storage); return outcome; } diff --git a/server/src/services/remote-workspace-snapshot-retention.test.ts b/server/src/services/remote-workspace-snapshot-retention.test.ts index 60986fff325..88612a5583a 100644 --- a/server/src/services/remote-workspace-snapshot-retention.test.ts +++ b/server/src/services/remote-workspace-snapshot-retention.test.ts @@ -44,7 +44,10 @@ async function versionedS3Fixture() { const versionId = parsed.searchParams.get("versionId"); const keyMarker = parsed.searchParams.get("key-marker"), versionMarker = parsed.searchParams.get("version-id-marker"); requests.push({ method: req.method!, key, versionId, keyMarker, versionMarker }); - if (parsed.searchParams.has("versions")) { + if (parsed.searchParams.has("uploads")) { + res.setHeader("content-type", "application/xml"); + res.end(`bucketfalse`); + } else if (parsed.searchParams.has("versions")) { if (faults.listDenied) { res.writeHead(403, { "content-type": "application/xml" }); res.end("AccessDeniedVersion listing permission is required"); @@ -142,6 +145,19 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () => 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("aborts each attempt's incomplete multipart upload before deleting a retired snapshot's objects", async () => { + const f = await seed(), order: string[] = [], deleteObject = storage.deleteObject.bind(storage); + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => { order.push(`abort:${key}`); return 1; }); + storage.deleteObject = async (companyId, key, options) => { order.push(`delete:${key}`); return deleteObject(companyId, key, options); }; + expect(await service().tick()).toMatchObject({ deleted: 1, failed: 0 }); + expect(order).toEqual([...f.keys.map(key => `abort:${key}`), ...f.keys.map(key => `delete:${key}`)]); + }); + it("still retires a snapshot when the store refuses to list its multipart uploads", async () => { + const f = await seed(); + storage.abortIncompleteUploads = vi.fn(async () => { throw Object.assign(new Error("Access Denied"), { name: "AccessDenied" }); }); + expect(await service().tick()).toMatchObject({ deleted: 1, failed: 0 }); + for (const key of f.keys) expect((await storage.headObject(f.companyId, key)).exists).toBe(false); + }); 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 2003dce614e..904558671c2 100644 --- a/server/src/services/remote-workspace-snapshot-retention.ts +++ b/server/src/services/remote-workspace-snapshot-retention.ts @@ -5,9 +5,12 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import type { StorageService } from "../storage/types.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js"; +import { logger } from "../middleware/logger.js"; export const REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS = 7 * 86400_000; export const REMOTE_WORKSPACE_SNAPSHOT_RETRY_MS = 15 * 60_000; +const SNAPSHOT_ABORT_CONCURRENCY = 8; type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; type Tx = Parameters[0]>[0]; type ReadDb = Pick | Pick; @@ -17,6 +20,38 @@ export function notifyRemoteWorkspaceSnapshots(transaction: object, hint?: strin return notifyDeliveryWork(transaction, DELIVERY_QUEUES.remoteWorkspaceSnapshots, hint); } +export function snapshotAttemptKeys(row: Snapshot) { + const uploads = Array.isArray(row.metadata?.uploads) ? row.metadata.uploads as Array<{ objectKey?: unknown }> : []; + const prefix = `${row.companyId}/workspace-snapshots/${row.executionWorkspaceId}/${row.id}/`; + return [...new Set(uploads.flatMap(upload => typeof upload?.objectKey === "string" && upload.objectKey !== row.objectKey + && upload.objectKey.startsWith(prefix) && !upload.objectKey.includes("..") ? [upload.objectKey] : []))]; +} + +export async function abortSnapshotAttemptUploads(storage: StorageService, companyId: string, keys: Iterable, signal?: AbortSignal) { + const abort = storage.abortIncompleteUploads?.bind(storage); + const pending = [...keys]; + if (!abort || !pending.length) return 0; + signal?.throwIfAborted(); + const bounded = AbortSignal.timeout(STORAGE_FAILURE_CLEANUP_MS); + const deadline = signal ? AbortSignal.any([signal, bounded]) : bounded; + let aborted = 0; + const worker = async () => { + for (let key = pending.shift(); key !== undefined && !deadline.aborted; key = pending.shift()) { + try { + const count = await abort(companyId, key, { signal: deadline }); + aborted += count; + } catch (error) { + if (signal?.aborted) throw error; + logger.warn({ err: error, companyId, objectKey: key }, "workspace snapshot multipart upload abort failed"); + } + } + }; + await Promise.all(Array.from({ length: Math.min(SNAPSHOT_ABORT_CONCURRENCY, pending.length) }, worker)); + signal?.throwIfAborted(); + if (pending.length) logger.warn({ companyId, remaining: pending.length }, "workspace snapshot multipart upload abort ran out of time"); + return aborted; +} + function abandonedUploadOperationId(row: Snapshot) { if (row.status !== "delete_requested") return null; const abandoned = row.metadata?.abandonedUpload as { operationId?: unknown } | undefined; @@ -148,6 +183,7 @@ export function remoteWorkspaceSnapshotRetentionService(db: Db, options: { for (const key of keys) { if (!key.startsWith(prefix) || key.includes("..") || key.includes("\\")) throw new Error("snapshot_upload_scope_invalid"); } + await abortSnapshotAttemptUploads(options.storage, claimed.companyId, keys, signal); for (const key of keys) { signal?.throwIfAborted(); await options.storage.deleteObject(claimed.companyId, key, { signal, allVersions: true }); diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 688b7a25431..d3eab02d39a 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -244,6 +244,41 @@ suite("remote workspace snapshot storage", () => { const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref.id)); expect((row?.metadata?.uploads as unknown[]).length).toBe(1); }); + it("aborts the incomplete upload of an attempt a restart cut off once a retried save publishes", async () => { + const f = await fixture(), put = storage.putFile.bind(storage), incomplete = new Set(); + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => Number(incomplete.delete(key))); + storage.putFile = async input => { incomplete.add(input.objectKey!); throw new Error("process killed mid UploadPart"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("process killed mid UploadPart"); + const [cut] = [...incomplete]; + storage.putFile = put; + const ref = await uploadWorkspaceSnapshot(f.upload); + expect(ref.objectKey).not.toBe(cut); + expect(storage.abortIncompleteUploads).toHaveBeenCalledTimes(1); + expect(storage.abortIncompleteUploads).toHaveBeenCalledWith(f.companyId, cut, expect.anything()); + expect(incomplete.size).toBe(0); + await uploadWorkspaceSnapshot(f.upload); + expect(storage.abortIncompleteUploads).toHaveBeenCalledTimes(1); + }); + it("adopts the published save when another uploader aborts this attempt's upload", async () => { + const f = await fixture(), put = storage.putFile.bind(storage); + let release!: (error: Error) => void, started!: () => void; + const gate = new Promise((_resolve, reject) => { release = reject; }); + const blocked = new Promise(resolve => { started = resolve; }); + let first = true, aborted: string | null = null; + storage.putFile = async input => { + if (!first) return put(input); + first = false; aborted = input.objectKey!; started(); return gate; + }; + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => { + if (key === aborted) release(Object.assign(new Error("NoSuchUpload"), { name: "NoSuchUpload" })); + return 1; + }); + const loser = uploadWorkspaceSnapshot(f.upload); + await blocked; + const winner = await uploadWorkspaceSnapshot(f.upload); + expect(storage.abortIncompleteUploads).toHaveBeenCalledWith(f.companyId, aborted, expect.anything()); + expect(await loser).toEqual(winner); + }); it("retains an unsuccessful operation and rejects a content-changing retry", async () => { const f = await fixture(); storage.putFile = async () => { throw new Error("storage unavailable"); }; await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("storage unavailable"); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 9bdaf0d6f4f..e20ce211512 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -14,7 +14,7 @@ import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import type { StoredSecretVersionMaterial } from "../secrets/types.js"; import type { StorageService } from "../storage/types.js"; -import { notifyRemoteWorkspaceSnapshots, REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS } from "./remote-workspace-snapshot-retention.js"; +import { abortSnapshotAttemptUploads, notifyRemoteWorkspaceSnapshots, snapshotAttemptKeys, REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS } from "./remote-workspace-snapshot-retention.js"; import { WORKSPACE_ARCHIVE_IO_BUFFER_BYTES } from "./remote-workspace-archive-io.js"; const SCHEMA = "paperclip.workspace-snapshot.v1"; @@ -296,6 +296,7 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v const winning = updated ?? (await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)))[0]; if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed"); const saved = reference(winning!); + if (updated) await abortSnapshotAttemptUploads(input.storage, saved.companyId, snapshotAttemptKeys(updated)); if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); if (saved.objectKey !== attempt.objectKey) await verifyTimed(saved); return saved; @@ -376,9 +377,17 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v } // The object key is unique per attempt, so a late losing uploader cannot // overwrite an object already used by a committed generation. - const stored = await timedStep(input.timing, "putMs", "storage.snapshot.put", () => input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, - originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), - byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal })); + let stored: Awaited>; + try { + stored = await timedStep(input.timing, "putMs", "storage.snapshot.put", () => input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, + originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), + byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal })); + } catch (error) { + if (input.signal?.aborted) throw error; + const saved = await getWorkspaceSnapshot(input.db, input); + if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) throw error; + await verifyTimed(saved); return saved; + } const uploaded = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; // The backend checked every part of this upload against digests of these // exact bytes, so only its size needs confirming. Any other provider, and diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 475df7c095b..894a05f2e35 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -14,6 +14,11 @@ export type ShutdownStepResult = | { outcome: "timed_out"; durationMs: number } | { outcome: "failed"; error: unknown; durationMs: number }; +export async function afterStalledShutdownStep(result: ShutdownStepResult, onStalled: () => Promise) { + if (result.outcome !== "completed") await onStalled(); + return result; +} + export async function runBoundedShutdownStep(input: { step: string; signal: "SIGINT" | "SIGTERM"; diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index 2f64817bc09..515faa56d45 100644 --- a/server/src/storage/s3-multipart.ts +++ b/server/src/storage/s3-multipart.ts @@ -9,11 +9,34 @@ import { PassThrough, Readable } from "node:stream"; import { setTimeout as delay } from "node:timers/promises"; import type { PutObjectInput, PutObjectReceipt } from "./types.js"; import { withS3FailureDiagnostics } from "./s3-failure.js"; -import { cleanupStorageFailure } from "./failure-cleanup.js"; +import { cleanupStorageFailure, STORAGE_FAILURE_CLEANUP_MS } from "./failure-cleanup.js"; const bufferedPartSize = 8 * 1024 * 1024; const bufferedPayloadLimit = 4 * bufferedPartSize; let bufferedPayloadBytes = 0; +const inFlightUploads = new Set<{ abort: (reason: unknown) => void; settled: Promise }>(); + +function trackUpload(abort: (reason: unknown) => void) { + let settle!: () => void; + const entry = { abort, settled: new Promise(resolve => { settle = resolve; }) }; + inFlightUploads.add(entry); + return () => { inFlightUploads.delete(entry); settle(); }; +} + +function settledWithin(entries: Array<{ settled: Promise }>, ms: number) { + let timer: ReturnType | undefined; + return Promise.race([Promise.all(entries.map(entry => entry.settled)).then(() => undefined), + new Promise(resolve => { timer = setTimeout(resolve, ms); })]).finally(() => clearTimeout(timer)); +} + +export async function abortInFlightMultipartUploads(reason: unknown, options: { graceMs?: number; timeoutMs?: number } = {}) { + if (options.graceMs && inFlightUploads.size) await settledWithin([...inFlightUploads], options.graceMs); + const entries = [...inFlightUploads]; + if (!entries.length) return 0; + for (const entry of entries) entry.abort(reason); + await settledWithin(entries, options.timeoutMs ?? 2 * STORAGE_FAILURE_CLEANUP_MS); + return entries.length; +} /** Each part is streamed too: multipart must not become another whole-file * buffer or a fixed part-count ceiling below the storage provider's limits. */ @@ -37,6 +60,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri const onAbort = () => stop(input.signal!.reason); input.signal?.addEventListener("abort", onAbort, { once: true }); if (input.signal?.aborted) onAbort(); + const untrack = trackUpload(stop); const readNext = async () => { controller.signal.throwIfAborted(); let abort!: () => void; @@ -126,6 +150,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri input.signal?.removeEventListener("abort", onAbort); source.destroy(); await cleanupStorageFailure(async () => iterator.return?.()); + untrack(); } } @@ -211,6 +236,7 @@ export async function putS3MultipartFile( const onAbort = () => controller.abort(input.signal!.reason); input.signal?.addEventListener("abort", onAbort, { once: true }); if (input.signal?.aborted) onAbort(); + const untrack = trackUpload(reason => { if (!controller.signal.aborted) controller.abort(reason); }); // This path never reads the stream body; release its file descriptor now. if (input.body instanceof Readable) input.body.destroy(); let uploadId: string | undefined; @@ -283,5 +309,6 @@ export async function putS3MultipartFile( throw error; } finally { input.signal?.removeEventListener("abort", onAbort); + untrack(); } } diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index 7ff5012f4b4..3d0cdd8e531 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -6,6 +6,8 @@ import { PutObjectCommand, ListObjectsV2Command, ListObjectVersionsCommand, + ListMultipartUploadsCommand, + AbortMultipartUploadCommand, } from "@aws-sdk/client-s3"; import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile, sendFileRequestWithRetry } from "./s3-multipart.js"; import { createReadStream } from "node:fs"; @@ -198,6 +200,33 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid return { etag: output.ETag }; }, + async abortIncompleteUploads(input) { + const key = buildKey(prefix, input.objectKey); + const uploads: string[] = []; + let keyMarker: string | undefined, uploadIdMarker: string | undefined; + for (let pages = 0; ; pages++) { + input.signal?.throwIfAborted(); + if (pages >= 100) throw new Error("s3_multipart_listing_cursor_invalid"); + const output = await client.send(new ListMultipartUploadsCommand({ Bucket: bucket, Prefix: key, + KeyMarker: keyMarker, UploadIdMarker: uploadIdMarker, MaxUploads: 1000 }), { abortSignal: input.signal }); + for (const upload of output.Uploads ?? []) if (upload.Key === key && upload.UploadId) uploads.push(upload.UploadId); + if (!output.IsTruncated) break; + if (!output.NextKeyMarker && !output.NextUploadIdMarker) throw new Error("s3_multipart_listing_cursor_invalid"); + keyMarker = output.NextKeyMarker; uploadIdMarker = output.NextUploadIdMarker; + } + let aborted = 0; + for (const uploadId of uploads) { + input.signal?.throwIfAborted(); + try { + await client.send(new AbortMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: uploadId }), { abortSignal: input.signal }); + aborted++; + } catch (error) { + if ((error as { name?: string }).name !== "NoSuchUpload") throw error; + } + } + return aborted; + }, + async listObjects(input) { const keyPrefix = buildKey(prefix, input.prefix); const output = await client.send(new ListObjectsV2Command({ diff --git a/server/src/storage/service.ts b/server/src/storage/service.ts index 9203903d0d5..e55afb81b33 100644 --- a/server/src/storage/service.ts +++ b/server/src/storage/service.ts @@ -158,5 +158,11 @@ export function createStorageService(provider: StorageProvider): StorageService await cancellable(options?.signal, () => provider.deleteObject({ objectKey, signal: options?.signal, ...(options?.allVersions ? { allVersions: true } : {}) })); }, + + async abortIncompleteUploads(companyId: string, objectKey: string, options) { + ensureCompanyPrefix(companyId, objectKey); + if (!provider.abortIncompleteUploads) return 0; + return cancellable(options?.signal, () => provider.abortIncompleteUploads!({ objectKey, signal: options?.signal })); + }, }; } diff --git a/server/src/storage/types.ts b/server/src/storage/types.ts index cec0dcdb473..c45f36b560b 100644 --- a/server/src/storage/types.ts +++ b/server/src/storage/types.ts @@ -72,6 +72,8 @@ export interface StorageProvider { | { ifMatch: string; ifNoneMatch?: never } )): Promise<{ etag: string }>; + abortIncompleteUploads?(input: { objectKey: string; signal?: AbortSignal }): Promise; + listObjects?(input: { prefix: string; continuationToken?: string; signal?: AbortSignal }): Promise<{ keys: string[]; continuationToken?: string; @@ -106,4 +108,5 @@ export interface StorageService { getObject(companyId: string, objectKey: string, options?: Pick): Promise; headObject(companyId: string, objectKey: string, options?: Pick): Promise; deleteObject(companyId: string, objectKey: string, options?: Pick): Promise; + abortIncompleteUploads?(companyId: string, objectKey: string, options?: { signal?: AbortSignal }): Promise; } diff --git a/tests/restart-chaos/storage-orphan-uploads.test.ts b/tests/restart-chaos/storage-orphan-uploads.test.ts new file mode 100644 index 00000000000..03428e9a699 --- /dev/null +++ b/tests/restart-chaos/storage-orphan-uploads.test.ts @@ -0,0 +1,54 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { createRequire } from "node:module"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { createS3StorageProvider } from "../../server/src/storage/s3-provider.js"; +import { minioBinary, startMinio, type ObjectStore } from "./object-store.js"; +import { listIncompleteUploads } from "./storage-audit.js"; + +const binary = minioBinary(); +const sdk = createRequire(new URL("../../server/package.json", import.meta.url))("@aws-sdk/client-s3"); +const MIB = 1024 * 1024; + +describe.skipIf(!binary)("incomplete snapshot uploads against MinIO", () => { + let root: string; + let store: ObjectStore; + let client: InstanceType; + const prefix = "chaos/w-orphan"; + const snapshot = "company/workspace-snapshots/ws/snapshot"; + + beforeAll(async () => { + root = await mkdtemp(path.join(os.tmpdir(), "restart-chaos-orphan-")); + store = await startMinio({ binary: binary!, root }); + client = new sdk.S3Client({ endpoint: store.endpoint, region: store.region, forcePathStyle: true, maxAttempts: 1, + credentials: { accessKeyId: store.accessKeyId, secretAccessKey: store.secretAccessKey } }); + }, 60_000); + afterEach(() => { vi.unstubAllEnvs(); }); + afterAll(async () => { + client?.destroy(); + await store?.stop(); + await rm(root, { recursive: true, force: true }); + }); + + async function cutUpload(key: string) { + const { UploadId } = await client.send(new sdk.CreateMultipartUploadCommand({ Bucket: store.bucket, Key: `${prefix}/${key}` })); + await client.send(new sdk.UploadPartCommand({ Bucket: store.bucket, Key: `${prefix}/${key}`, UploadId, PartNumber: 1, Body: Buffer.alloc(5 * MIB, 1) })); + return UploadId as string; + } + + it("aborts only the cut-off attempt's uploads and leaves a sibling attempt and a longer key alone", async () => { + vi.stubEnv("AWS_ACCESS_KEY_ID", store.accessKeyId); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", store.secretAccessKey); + vi.stubEnv("AWS_SESSION_TOKEN", ""); + const cut = `${snapshot}/cut.pcws`, sibling = `${snapshot}/sibling.pcws`, longer = `${snapshot}/cut.pcws.longer`; + await cutUpload(cut); await cutUpload(cut); + const siblingId = await cutUpload(sibling), longerId = await cutUpload(longer); + expect(await listIncompleteUploads(store, `${prefix}/`)).toHaveLength(4); + const provider = createS3StorageProvider({ bucket: store.bucket, region: store.region, endpoint: store.endpoint, prefix, forcePathStyle: true }); + expect(await provider.abortIncompleteUploads!({ objectKey: cut })).toBe(2); + expect((await listIncompleteUploads(store, `${prefix}/`)).map(upload => [upload.key, upload.uploadId]).sort()) + .toEqual([[`${prefix}/${longer}`, longerId], [`${prefix}/${sibling}`, siblingId]].sort()); + expect(await provider.abortIncompleteUploads!({ objectKey: cut })).toBe(0); + }, 60_000); +});