Files
PaperClipAI/server/src/storage/service.ts
T
DottaandClaude Opus 5.5 d0c30e69a9 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>
2026-11-10 10:56:52 -06:00

169 lines
6.4 KiB
TypeScript

import { createHash, randomUUID } from "node:crypto";
import path from "node:path";
import { Readable } from "node:stream";
import type { StorageService, StorageProvider, PutFileInput, PutFileResult, PutObjectReceipt } from "./types.js";
import { badRequest, forbidden, unprocessable } from "../errors.js";
import { cleanupStorageFailure } from "./failure-cleanup.js";
const MAX_SEGMENT_LENGTH = 120;
async function cancellable<T>(signal: AbortSignal | undefined, action: () => Promise<T>): Promise<T> {
signal?.throwIfAborted();
if (!signal) return action();
let abort: () => void = () => {};
const cancelled = new Promise<never>((_resolve, reject) => {
abort = () => reject(signal.reason);
signal.addEventListener("abort", abort, { once: true });
});
try { return await Promise.race([action(), cancelled]); }
finally { signal.removeEventListener("abort", abort); }
}
function sanitizeSegment(value: string): string {
const cleaned = value
.trim()
.replace(/[^a-zA-Z0-9._-]+/g, "_")
.replace(/_{2,}/g, "_")
.replace(/^_+|_+$/g, "");
if (!cleaned) return "file";
return cleaned.slice(0, MAX_SEGMENT_LENGTH);
}
function normalizeNamespace(namespace: string): string {
const normalized = namespace
.split("/")
.map((entry) => entry.trim())
.filter((entry) => entry.length > 0)
.map((entry) => sanitizeSegment(entry));
if (normalized.length === 0) return "misc";
return normalized.join("/");
}
function splitFilename(filename: string | null): { stem: string; ext: string } {
if (!filename) return { stem: "file", ext: "" };
const base = path.basename(filename).trim();
if (!base) return { stem: "file", ext: "" };
const extRaw = path.extname(base);
const stemRaw = extRaw ? base.slice(0, base.length - extRaw.length) : base;
const stem = sanitizeSegment(stemRaw);
const ext = extRaw
.toLowerCase()
.replace(/[^a-z0-9.]/g, "")
.slice(0, 16);
return {
stem,
ext,
};
}
function ensureCompanyPrefix(companyId: string, objectKey: string): void {
const expectedPrefix = `${companyId}/`;
if (!objectKey.startsWith(expectedPrefix)) {
throw forbidden("Object does not belong to company");
}
if (objectKey.includes("..")) {
throw badRequest("Invalid object key");
}
}
function hashBuffer(input: Buffer): string {
return createHash("sha256").update(input).digest("hex");
}
function buildObjectKey(companyId: string, namespace: string, originalFilename: string | null): string {
const ns = normalizeNamespace(namespace);
const now = new Date();
const year = String(now.getUTCFullYear());
const month = String(now.getUTCMonth() + 1).padStart(2, "0");
const day = String(now.getUTCDate()).padStart(2, "0");
const { stem, ext } = splitFilename(originalFilename);
const suffix = randomUUID();
const filename = `${suffix}-${stem}${ext}`;
return `${companyId}/${ns}/${year}/${month}/${day}/${filename}`;
}
function assertPutFileInput(input: PutFileInput): void {
if (!input.companyId || input.companyId.trim().length === 0) {
throw unprocessable("companyId is required");
}
if (!input.namespace || input.namespace.trim().length === 0) {
throw unprocessable("namespace is required");
}
if (!input.contentType || input.contentType.trim().length === 0) {
throw unprocessable("contentType is required");
}
if (!(input.body instanceof Buffer) && !(input.body instanceof Readable)) {
throw unprocessable("body must be a Buffer or Readable");
}
if (!Buffer.isBuffer(input.body) && (!("byteSize" in input) || !Number.isSafeInteger(input.byteSize) || input.byteSize < 0 || !/^[0-9a-f]{64}$/.test(input.sha256))) {
throw unprocessable("Streamed files require an exact byte size and SHA-256");
}
if (("byteSize" in input ? input.byteSize : input.body.length) <= 0) {
throw unprocessable("File is empty");
}
}
export function createStorageService(provider: StorageProvider): StorageService {
return {
provider: provider.id,
...(provider.immutableObjectNamespace ? { immutableObjectNamespace: provider.immutableObjectNamespace } : {}),
async putFile(input: PutFileInput): Promise<PutFileResult> {
assertPutFileInput(input);
const objectKey = input.objectKey ?? buildObjectKey(input.companyId, input.namespace, input.originalFilename);
ensureCompanyPrefix(input.companyId, objectKey);
const byteSize = "byteSize" in input ? input.byteSize : input.body.length;
const contentType = input.contentType.trim().toLowerCase();
let receipt: void | PutObjectReceipt;
try {
receipt = await provider.putObject({
objectKey,
body: input.body,
contentType,
contentLength: byteSize,
signal: input.signal,
...("sourceFile" in input && input.sourceFile ? { sourceFile: input.sourceFile } : {}),
});
} catch (error) {
await cleanupStorageFailure(signal => provider.deleteObject({ objectKey, signal }));
throw error;
} finally {
if (input.body instanceof Readable) input.body.destroy();
}
return {
provider: provider.id,
objectKey,
contentType,
byteSize,
sha256: "sha256" in input ? input.sha256 : hashBuffer(input.body),
originalFilename: input.originalFilename,
...(receipt ? { integrity: receipt.integrity } : {}),
};
},
async getObject(companyId: string, objectKey: string, options) {
ensureCompanyPrefix(companyId, objectKey);
return provider.getObject({ objectKey, range: options?.range, signal: options?.signal });
},
async headObject(companyId: string, objectKey: string, options) {
ensureCompanyPrefix(companyId, objectKey);
return cancellable(options?.signal, () => provider.headObject({ objectKey, signal: options?.signal }));
},
async deleteObject(companyId: string, objectKey: string, options) {
ensureCompanyPrefix(companyId, objectKey);
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 }));
},
};
}