From 34c1e6469c43eea0579f1e7fdb5f6a0d56d2a92b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 12:10:57 -0600 Subject: [PATCH] fix(storage): retry small uploads and resume broken downloads (#20313) 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 > - A task's files live in a sandbox. Paperclip saves them to object storage when a run ends and restores them when the next run starts > - Object storage has short blips: a single 500, a reset connection, a download cut off halfway. Each of these used to fail the whole save or restore once, with no second try > - This pull request makes a short blip invisible. Longer outages are handled by the storage-unavailable wait (#20299 and a follow-up) > - The benefit is that tasks keep moving through ordinary storage failures without anyone clicking Retry ## Linked Issues or Issue Description Refs #19831. Storage chaos waves `w-20261109T145628-s9902` (ST1) and `w-20261109T151711-s9903` (ST2). - A single-request upload (files up to the multipart threshold, which includes most workspace archives) sent once. A 500 or connection reset failed the save. - A download whose body broke mid-stream failed the restore, even though S3 can resume from the byte offset. Related: #20299 (storage outage is shown as an infrastructure wait instead of a failure). ## What Changed - `server/src/storage/s3-multipart.ts`: `sendFileRequestWithRetry` resends a single file request and records throttles and backoff in the save timing. It uses the same throttle backoff as multipart parts for 429/503/SlowDown and up to the same part attempt count for other failures. It honours the caller's abort signal. - `server/src/storage/s3-provider.ts`: - A file-backed `putObject` under the multipart threshold uses that retry and reopens the source file for each attempt, so a resend never reuses a half-read stream. The Content-MD5 is computed once from the file and still verified by S3 on each attempt. - `getObject` resumes a body that fails with a transient socket error (reset, timeout, premature close and similar). It continues with a ranged GET from the byte where the stream stopped, at most three times. The resumed request carries `If-Match` with the first response's ETag, so a changed object fails instead of mixing bytes. Body-failure diagnostics are still recorded on every response. Destroying the returned stream closes the S3 body it is reading, even mid-read, so read timeouts and client disconnects still release the connection. - Tests: - `server/src/__tests__/storage-s3-transient-retry.test.ts` (new) runs against a local HTTP server speaking S3. ## Verification Red/green (each new test fails against master and passes with this PR): - `storage-s3-transient-retry.test.ts`: - A small upload is resent after a 500 and after a connection reset, re-reading the file. The second PUT carries the full content. - A download cut off mid-body resumes with `Range: bytes=-` and `If-Match`, and returns the complete bytes. - A download whose object changed before the resume fails with a precondition error and does not return mixed bytes. - A stalled download that its reader destroys closes its connection. Regression: the storage and workspace-finalization suites pass locally, and `tsc --noEmit` passes for `server`. ## Risks - Resending an upload is safe because each attempt writes the full object with the same Content-MD5. Resuming a download is bounded to three resumes and refuses a changed object. - No schema migration and no API change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../storage-s3-transient-retry.test.ts | 193 ++++++++++++++++++ server/src/storage/s3-multipart.ts | 37 ++++ server/src/storage/s3-provider.ts | 92 +++++++-- 3 files changed, 307 insertions(+), 15 deletions(-) create mode 100644 server/src/__tests__/storage-s3-transient-retry.test.ts diff --git a/server/src/__tests__/storage-s3-transient-retry.test.ts b/server/src/__tests__/storage-s3-transient-retry.test.ts new file mode 100644 index 00000000000..01bedcc6753 --- /dev/null +++ b/server/src/__tests__/storage-s3-transient-retry.test.ts @@ -0,0 +1,193 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; + +type Seen = { method: string; range: string | null; ifMatch: string | null; body: string }; +const seen: Seen[] = []; +let respond: (req: IncomingMessage, res: ServerResponse, body: string, index: number) => void = (_req, res) => res.end(); + +const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", chunk => chunks.push(chunk as Buffer)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + seen.push({ method: req.method ?? "", range: req.headers.range ?? null, + ifMatch: (req.headers["if-match"] as string | undefined) ?? null, body }); + respond(req, res, body, seen.length - 1); + }); +}); +let endpoint = ""; +let dir = ""; +const savedEnv = { ...process.env }; + +beforeAll(async () => { + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + endpoint = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + process.env.AWS_ACCESS_KEY_ID = "fixture-access-key"; + process.env.AWS_SECRET_ACCESS_KEY = "fixture-secret-key"; + delete process.env.AWS_SESSION_TOKEN; + delete process.env.AWS_PROFILE; + dir = await mkdtemp(join(tmpdir(), "s3-transient-")); +}); + +afterAll(async () => { + process.env = savedEnv; + await rm(dir, { recursive: true, force: true }); + await new Promise(resolve => server.close(() => resolve())); +}); + +beforeEach(() => { seen.length = 0; }); + +function provider() { + return createS3StorageProvider({ bucket: "bucket", region: "us-east-1", endpoint, forcePathStyle: true }); +} + +async function readAll(stream: NodeJS.ReadableStream) { + const chunks: Buffer[] = []; + for await (const chunk of stream) chunks.push(Buffer.from(chunk as Buffer)); + return Buffer.concat(chunks).toString("utf8"); +} + +async function sourceFile(content: string) { + const file = join(dir, `${seen.length}-${Math.random().toString(16).slice(2)}.bin`); + await writeFile(file, content); + return file; +} + +const SERVER_ERROR = ` +InternalErrorWe encountered an internal error.req-500`; + +describe("S3 transient failures on saved files", () => { + it.each([ + ["an internal error", (res: ServerResponse) => { res.writeHead(500, { "content-type": "application/xml" }); res.end(SERVER_ERROR); }], + ["a connection reset", (res: ServerResponse) => { res.socket?.destroy(); }], + ])("resends a small file upload after %s by reading the file again", async (_label, fail) => { + const content = "saved workspace archive bytes"; + respond = (_req, res, _body, index) => { + if (index === 0) { fail(res); return; } + res.writeHead(200, { etag: '"saved"' }); + res.end(); + }; + + await expect(provider().putObject({ objectKey: "company/workspace.pcws", body: Buffer.alloc(0), + contentType: "application/octet-stream", contentLength: content.length, sourceFile: await sourceFile(content) })) + .resolves.toEqual({ integrity: "server_verified_md5" }); + + expect(seen.map(entry => entry.method)).toEqual(["PUT", "PUT"]); + expect(seen[1]!.body).toBe(content); + }); + + it("does not resend a small upload the store refused for good", async () => { + respond = (_req, res) => { + res.writeHead(403, { "content-type": "application/xml" }); + res.end(`AccessDenieddenied`); + }; + + await expect(provider().putObject({ objectKey: "company/workspace.pcws", body: Buffer.alloc(0), + contentType: "application/octet-stream", contentLength: 5, sourceFile: await sourceFile("bytes") })) + .rejects.toMatchObject({ name: "AccessDenied" }); + expect(seen.map(entry => entry.method)).toEqual(["PUT"]); + }); + + it("keeps a download complete when its connection breaks after the last byte", async () => { + const content = "0123456789abcdefghij"; + respond = (_req, res) => { + res.writeHead(200, { "content-length": String(content.length + 5), etag: '"v1"' }); + res.write(content); + setTimeout(() => res.socket?.destroy(), 20); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws", range: { start: 0, end: content.length - 1 } }); + + expect(await readAll(object.stream)).toBe(content); + expect(seen.map(entry => entry.method)).toEqual(["GET"]); + }); + + it("resumes a download whose body is cut off from the byte where it stopped", async () => { + const content = "0123456789abcdefghij"; + respond = (req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": String(content.length), etag: '"v1"' }); + res.write(content.slice(0, 8)); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + const [, from, to] = /bytes=(\d+)-(\d*)/.exec(req.headers.range ?? "")!; + const end = to ? Number(to) : content.length - 1; + res.writeHead(206, { "content-length": String(end - Number(from) + 1), etag: '"v1"', + "content-range": `bytes ${from}-${end}/${content.length}` }); + res.end(content.slice(Number(from), end + 1)); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + + expect(await readAll(object.stream)).toBe(content); + expect(seen.map(entry => [entry.method, entry.range, entry.ifMatch])).toEqual([ + ["GET", null, null], ["GET", "bytes=8-19", '"v1"']]); + }); + + it("closes a stalled download's connection when its reader gives up", async () => { + let closed!: () => void; + const connectionClosed = new Promise(resolve => { closed = resolve; }); + respond = (_req, res) => { + res.writeHead(200, { "content-length": "20", etag: '"v1"' }); + res.write("01234567"); + res.on("close", () => closed()); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + const reader = object.stream[Symbol.asyncIterator](); + expect(String((await reader.next()).value)).toBe("01234567"); + object.stream.destroy(); + + await expect(Promise.race([connectionClosed.then(() => "closed"), + new Promise(resolve => setTimeout(() => resolve("still open"), 2_000))])).resolves.toBe("closed"); + }); + + it("closes a stalled resume request when its reader gives up", async () => { + let closed!: () => void; + const resumeClosed = new Promise(resolve => { closed = resolve; }); + respond = (_req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": "20", etag: '"v1"' }); + res.write("01234567"); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + res.on("close", () => closed()); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + const reader = object.stream[Symbol.asyncIterator](); + expect(String((await reader.next()).value)).toBe("01234567"); + void reader.next().catch(() => undefined); + await vi.waitFor(() => expect(seen).toHaveLength(2), { timeout: 5_000 }); + object.stream.destroy(); + + await expect(Promise.race([resumeClosed.then(() => "closed"), + new Promise(resolve => setTimeout(() => resolve("still open"), 2_000))])).resolves.toBe("closed"); + }); + + it("fails instead of mixing bytes when the object changed before the download resumed", async () => { + const content = "0123456789abcdefghij"; + respond = (_req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": String(content.length), etag: '"v1"' }); + res.write(content.slice(0, 8)); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + res.writeHead(412, { "content-type": "application/xml" }); + res.end(`PreconditionFailedchanged`); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + + await expect(readAll(object.stream)).rejects.toMatchObject({ name: "PreconditionFailed" }); + expect(seen.map(entry => entry.ifMatch)).toEqual([null, '"v1"']); + }); +}); diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index 8a18d035c59..74992e5f1a8 100644 --- a/server/src/storage/s3-multipart.ts +++ b/server/src/storage/s3-multipart.ts @@ -147,6 +147,43 @@ function isRequestRateLimited(error: unknown) { return name === "SlowDown" || $metadata?.httpStatusCode === 503 || $metadata?.httpStatusCode === 429; } +const TRANSIENT_FILE_REQUEST_STATUS = new Set([408, 429, 500, 502, 503, 504]); +const TRANSIENT_FILE_REQUEST_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "ETIMEDOUT", "EPIPE", "EAI_AGAIN", + "ENOTFOUND", "ERR_STREAM_PREMATURE_CLOSE", "UND_ERR_SOCKET"]); +const TRANSIENT_FILE_REQUEST_NAMES = new Set(["RequestTimeout", "TimeoutError", "NetworkingError", "InternalError", "ServiceUnavailable"]); + +function isTransientFileRequestFailure(error: unknown) { + if (!error || typeof error !== "object") return false; + const { name, code, $metadata } = error as { name?: unknown; code?: unknown; $metadata?: { httpStatusCode?: unknown } }; + const status = $metadata?.httpStatusCode; + if (typeof status === "number") return TRANSIENT_FILE_REQUEST_STATUS.has(status); + return (typeof code === "string" && TRANSIENT_FILE_REQUEST_CODES.has(code)) + || (typeof name === "string" && TRANSIENT_FILE_REQUEST_NAMES.has(name)); +} + +export async function sendFileRequestWithRetry(signal: AbortSignal | undefined, send: () => Promise): Promise { + for (let failures = 0, throttles = 0; ;) { + signal?.throwIfAborted(); + try { + return await send(); + } catch (error) { + if (signal?.aborted) throw error; + if (isRequestRateLimited(error)) { + recordStorageThrottle(); + if (++throttles >= THROTTLED_PART_ATTEMPTS) throw error; + const ceiling = Math.min(THROTTLE_BACKOFF_MAX_MS, THROTTLE_BACKOFF_BASE_MS * 2 ** (throttles - 1)); + const wait = Math.floor(Math.random() * ceiling); + recordStorageBackoff(wait); + await delay(wait, undefined, { signal }); + continue; + } + if (!isTransientFileRequestFailure(error) || ++failures >= FILE_PART_ATTEMPTS) throw error; + recordStorageBackoff(failures * 500); + await delay(failures * 500, undefined, { signal }); + } + } +} + /** Base64 MD5 of an inclusive byte range, as the Content-MD5 header expects. */ export async function fileRangeMd5(file: string, start: number, end: number, signal?: AbortSignal) { const digest = createHash("md5"); diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index 9e553280d7d..294afb83a6e 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -7,7 +7,7 @@ import { ListObjectsV2Command, ListObjectVersionsCommand, } from "@aws-sdk/client-s3"; -import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile } from "./s3-multipart.js"; +import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile, sendFileRequestWithRetry } from "./s3-multipart.js"; import { createReadStream } from "node:fs"; import { createHash } from "node:crypto"; import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js"; @@ -66,6 +66,59 @@ async function toReadableStream(body: unknown): Promise { throw unprocessable("Unsupported S3 body stream type"); } +function observeBodyFailures(stream: Readable, response: unknown) { + stream.on("error", function recordBodyFailure(error) { + recordS3BodyFailure(error, response); + if (stream.listenerCount("error") === 1) throw error; + }); + return stream; +} + +const BODY_RESUME_ATTEMPTS = 3; +const TRANSIENT_BODY_ERRORS = new Set(["ECONNRESET", "ETIMEDOUT", "EPIPE", "ECONNABORTED", "ERR_STREAM_PREMATURE_CLOSE", "UND_ERR_SOCKET"]); + +function isTransientBodyError(error: unknown) { + if (!error || typeof error !== "object") return false; + const { code, message } = error as { code?: unknown; message?: unknown }; + return (typeof code === "string" && TRANSIENT_BODY_ERRORS.has(code)) + || (typeof message === "string" && /socket hang up|aborted|premature close/i.test(message)); +} + +function resumableBody(first: Readable, reopen: (offset: number, signal: AbortSignal) => Promise, length: number, signal?: AbortSignal) { + let current = first, closed = false; + const reading = new AbortController(); + const resumeSignal = signal ? AbortSignal.any([signal, reading.signal]) : reading.signal; + async function* chunks() { + let offset = 0, resumes = 0; + while (true) { + try { + for await (const chunk of current) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array); + offset += bytes.length; + yield bytes; + } + return; + } catch (error) { + current.destroy(); + if (offset >= length) return; + if (closed || signal?.aborted || resumes >= BODY_RESUME_ATTEMPTS || !isTransientBodyError(error)) throw error; + resumes += 1; + current = await reopen(offset, resumeSignal); + if (closed) { current.destroy(); return; } + } + } + } + const stream = Readable.from(chunks(), { objectMode: false }); + const destroyWrapper = stream._destroy.bind(stream); + stream._destroy = (error, callback) => { + closed = true; + reading.abort(error ?? new Error("download_closed")); + current.destroy(); + destroyWrapper(error, callback); + }; + return stream; +} + function toDate(value: Date | undefined): Date | undefined { return value instanceof Date ? value : undefined; } @@ -164,13 +217,15 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid if (input.body instanceof Readable) input.body.destroy(); await assertSourceFileSize(sourceFile, input.contentLength); const contentMd5 = await fileRangeMd5(sourceFile, 0, input.contentLength - 1, input.signal); - const body = createReadStream(sourceFile); - try { - await withS3FailureDiagnostics("put_object", () => fileClient.send(new PutObjectCommand({ - Bucket: bucket, Key: key, Body: body, ContentType: input.contentType, - ContentLength: input.contentLength, ContentMD5: contentMd5, - }), { abortSignal: input.signal })); - } finally { body.destroy(); } + await sendFileRequestWithRetry(input.signal, async () => { + const body = createReadStream(sourceFile); + try { + await withS3FailureDiagnostics("put_object", () => fileClient.send(new PutObjectCommand({ + Bucket: bucket, Key: key, Body: body, ContentType: input.contentType, + ContentLength: input.contentLength, ContentMD5: contentMd5, + }), { abortSignal: input.signal })); + } finally { body.destroy(); } + }); return { integrity: "server_verified_md5" } as const; } if (input.contentLength > 16 * 1024 * 1024) { @@ -201,13 +256,20 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid { abortSignal: input.signal }, )); - const stream = await toReadableStream(output.Body); - // Observe, do not handle: a consumer without an error listener still - // sees the failure exactly as it would without this one. - stream.on("error", function recordBodyFailure(error) { - recordS3BodyFailure(error, output); - if (stream.listenerCount("error") === 1) throw error; - }); + const first = observeBodyFailures(await toReadableStream(output.Body), output); + const etag = output.ETag; + const start = input.range?.start ?? 0; + const end = input.range?.end ?? (typeof output.ContentLength === "number" ? output.ContentLength - 1 : undefined); + const reopen = async (offset: number, resumeSignal: AbortSignal) => { + const resumed = await withS3FailureDiagnostics("get_object", () => client.send(new GetObjectCommand({ + Bucket: bucket, Key: key, IfMatch: etag, + Range: `bytes=${start + offset}-${end ?? ""}`, + }), { abortSignal: resumeSignal })); + return observeBodyFailures(await toReadableStream(resumed.Body), resumed); + }; + const stream = etag && end !== undefined + ? resumableBody(first, reopen, end - start + 1, input.signal) + : first; if (input.signal) addAbortSignal(input.signal, stream); return { stream,