fix(storage): abort a snapshot upload a restart cut off once its save attempt is superseded (#20542)

## 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 <noreply@anthropic.com>

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]);
+  });
 });

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<number>((_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());
+  });
+});

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<void>(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<void>(resolve => { release = resolve; });
+    const started = new Promise<void>(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<void>(() => {});
+    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);
+  });
+});

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,

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<SettleOutcome> {
+  context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise<SettleOutcome> {
   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<SettleOutcome> => {
     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<string, unknown>) },
     });
     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;
 }

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<SettleOutcome> {
+  context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise<SettleOutcome> {
   let publication: ActivityPublication | undefined;
+  let retired: Awaited<ReturnType<typeof retireAbandonedSaveLedgerRows>> = [];
   const outcome = await db.transaction(async (transaction): Promise<SettleOutcome> => {
     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;
 }

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(`<?xml version="1.0"?><ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Bucket>bucket</Bucket><IsTruncated>false</IsTruncated></ListMultipartUploadsResult>`);
+      } else if (parsed.searchParams.has("versions")) {
         if (faults.listDenied) {
           res.writeHead(403, { "content-type": "application/xml" });
           res.end("<Error><Code>AccessDenied</Code><Message>Version listing permission is required</Message></Error>");
@@ -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 });

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<Parameters<Db["transaction"]>[0]>[0];
 type ReadDb = Pick<Db, "select"> | Pick<Tx, "select">;
@@ -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<string>, 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 });

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<string>();
+    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<never>((_resolve, reject) => { release = reject; });
+    const blocked = new Promise<void>(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");

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<ReturnType<StorageService["putFile"]>>;
+    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

server/src/shutdown.ts

@@ -14,6 +14,11 @@ export type ShutdownStepResult<T> =
   | { outcome: "timed_out"; durationMs: number }
   | { outcome: "failed"; error: unknown; durationMs: number };
 
+export async function afterStalledShutdownStep<T>(result: ShutdownStepResult<T>, onStalled: () => Promise<void>) {
+  if (result.outcome !== "completed") await onStalled();
+  return result;
+}
+
 export async function runBoundedShutdownStep<T>(input: {
   step: string;
   signal: "SIGINT" | "SIGTERM";

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<void> }>();
+
+function trackUpload(abort: (reason: unknown) => void) {
+  let settle!: () => void;
+  const entry = { abort, settled: new Promise<void>(resolve => { settle = resolve; }) };
+  inFlightUploads.add(entry);
+  return () => { inFlightUploads.delete(entry); settle(); };
+}
+
+function settledWithin(entries: Array<{ settled: Promise<void> }>, ms: number) {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  return Promise.race([Promise.all(entries.map(entry => entry.settled)).then(() => undefined),
+    new Promise<void>(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();
   }
 }

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({

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 }));
+    },
   };
 }

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<number>;
+
   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<GetObjectInput, "range" | "signal">): Promise<GetObjectResult>;
   headObject(companyId: string, objectKey: string, options?: Pick<GetObjectInput, "signal">): Promise<HeadObjectResult>;
   deleteObject(companyId: string, objectKey: string, options?: Pick<DeleteObjectInput, "signal" | "allVersions">): Promise<void>;
+  abortIncompleteUploads?(companyId: string, objectKey: string, options?: { signal?: AbortSignal }): Promise<number>;
 }

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<typeof sdk.S3Client>;
+  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);
+});