fix(storage): retry small uploads and resume broken downloads (#20313)

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

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<void>(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<void>(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 = `<?xml version="1.0" encoding="UTF-8"?>
+<Error><Code>InternalError</Code><Message>We encountered an internal error.</Message><RequestId>req-500</RequestId></Error>`;
+
+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(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>AccessDenied</Code><Message>denied</Message></Error>`);
+    };
+
+    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<void>(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<void>(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(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>PreconditionFailed</Code><Message>changed</Message></Error>`);
+    };
+
+    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"']);
+  });
+});

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<T>(signal: AbortSignal | undefined, send: () => Promise<T>): Promise<T> {
+  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");

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<Readable> {
   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<Readable>, 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,