perf(storage): overlap bounded S3 multipart payloads (#19817)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The storage service uploads large workspace snapshots through S3
multipart uploads.
> - The current uploader permits four requests but fills each streamed
body before it starts the next one.
> - Slow payload delivery therefore remains serial even when request
acknowledgements overlap.
> - This pull request fills bounded part buffers and sends those
payloads concurrently through the existing SDK streams.
> - This lets the four-request window carry payload data while it
retains backpressure and failure cleanup.

## Linked Issues or Issue Description

Related implementation: #18659. This fixes payload serialization within
that existing multipart owner. A search found no duplicate open storage
multipart fix.

**What happened?**

A controlled 24 MiB upload started part 2 only after part 1 had
delivered 99.22% of its bytes. The same sequence occurred for part 3.
The four-request cap overlapped acknowledgement waits but did not
provide useful overlap during payload delivery.

**Expected behavior**

Up to four part payloads can travel at the same time. Memory stays
bounded. Exact lengths, bytes, ordered completion, cancellation and
abort cleanup still apply.

**Steps to reproduce**

Run the new slow-consumer test against the previous uploader. Hold all
part consumers until four requests have started. The previous
implementation starts one request and the assertion fails. The revised
implementation starts four requests and then verifies every payload hash
and byte count.

**Paperclip version or commit**

Reproduced on the source from `585c885168`. The final branch is rebased
onto `ae1f781cac`.

**Deployment mode**

The S3 storage provider. Local tests use fake credentials and an owned
loopback HTTP server. No hosted writes were made.

## What Changed

- Share a 32 MiB part-buffer allowance across uploads. Stream up to four
buffered payloads concurrently through the existing SDK requests. When
the allowance is full, use the existing streaming path without an upload
queue.
- Keep the original streaming path for dynamic parts larger than 8 MiB.
This retains the provider part-count limit without larger buffers.
- Extend the existing tests for concurrent slow consumers, the shared
memory window and cancellation in both paths. Verify allowance release
after source and request failures. Check part identity independently of
acknowledgement order.

## Verification

- Baseline RED: the new controlled test failed with one request started
instead of four. Its cleanup released and awaited the upload.
- Final review-repair head: 26/26 tests passed across
`storage-s3-multipart.test.ts`, `storage-s3-failure-diagnostics.test.ts`
and `storage/config.test.ts`.
- These tests include the real S3 HTTP client, exact payload hashes and
ordered parts, excess and truncated input, early acknowledgements,
sibling failure, abort cleanup, stalled input and the bounded four-part
window.
- Strict server TypeScript, module boundary checks and `git diff
--check` passed on the final review-repair head. CI then caught an
optional SDK `Key` type in the new test fixture. A focused test
typecheck reproduced both TS2345 errors. The fixture now binds its
required controlled key; that typecheck and the named shared-window
regression pass. The uploader source is unchanged by this repair.
- Controlled 24 MiB comparison: 640.58 ms before and 393.88 ms after.
The new requests started within 7.37 ms. Both runs used 1 ms per
consumed chunk and 250 ms acknowledgement delays. This measures the
controlled fixture only. It does not predict hosted S3 throughput or
total migration time.
- Review RED: two blocked uploads consumed 64 MiB of source chunks under
the original per-upload buffer scheme. The new shared allowance kept the
source reads within 32 MiB plus stream lookahead. Three new controls
passed for concurrent uploads and allowance release after source/request
failures.
- The first treatment run passed 12/13 cases. Its remaining test assumed
completion order matched part order. The test now binds lengths to part
numbers and still checks the ordered completion request. An initial
typecheck setup attempt lacked the built plugin SDK; the canonical
dependency build and final typecheck passed.

- Final head `fbadf4f1cf427b896bbc255ffe2e7ad480ce06d3`: [full
CI](https://github.com/paperclipai/paperclip/actions/runs/42373571408)
passed, including build, test shards, Typecheck + Release Registry and
`ci / verify`. All 61 reported contexts are terminal: 56 success, four
skipped and one neutral review. Greptile and Paperclip both gave a fresh
5/5 for this head. No review threads remain open.

## Risks

- Multipart uploads share at most 32 MiB of live part buffers per
process, plus existing stream and source-chunk overhead. The previous
path used smaller streaming buffers. With no buffer allowance left, an
upload uses the original streaming path. It does not wait for a shared
queue.
- Objects whose dynamic part size exceeds 8 MiB retain the current
serial payload behavior. No new object-size ceiling is introduced.
- Request bodies remain non-replayable streams. The SDK retry policy,
multipart protocol, cleanup policy, object format and public storage API
remain unchanged.
- No disk spool or cache is added. All current-head CI gates and both
reviews passed.

## Model Used

OpenAI `gpt-6.1-sol`, with reasoning, tool use and code execution. The
context window size is not exposed by this session.

## 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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/storage-s3-multipart.test.ts

@@ -10,6 +10,40 @@ import { createStorageService } from "../storage/service.js";
 import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js";
 
 describe("streamed S3 multipart uploads", () => {
+  it("delivers four slow part payloads concurrently before any part is drained", async () => {
+    const client = new S3Client({ region: "us-east-1" }), started = new Set<number>(), received = new Map<number, number>();
+    let release!: () => void, active = 0, peak = 0, completed = false;
+    const reading = new Promise<void>(resolve => { release = resolve; });
+    const send = vi.spyOn(client, "send").mockImplementation(async (command: any) => {
+      if (command instanceof CreateMultipartUploadCommand) return { UploadId: "upload" };
+      if (command instanceof UploadPartCommand) {
+        const number = command.input.PartNumber!; started.add(number);
+        await reading; active++; peak = Math.max(peak, active);
+        let bytes = 0; const digest = createHash("sha256");
+        for await (const chunk of command.input.Body as Readable) {
+          digest.update(chunk); bytes += chunk.length;
+          await new Promise(resolve => setTimeout(resolve, 1));
+        }
+        expect(bytes).toBe(8 * 1024 * 1024);
+        expect(digest.digest("hex")).toBe(createHash("sha256").update(Buffer.alloc(bytes, number)).digest("hex"));
+        received.set(number, bytes); active--;
+        return { ETag: `part-${number}` };
+      }
+      expect(command).toBeInstanceOf(CompleteMultipartUploadCommand);
+      expect(command.input.MultipartUpload.Parts).toEqual([1, 2, 3, 4].map(PartNumber => ({ PartNumber, ETag: `part-${PartNumber}` })));
+      completed = true; return {};
+    });
+    const body = Readable.from((async function* () {
+      for (let index = 0; index < 512; index++) yield Buffer.alloc(64 * 1024, Math.floor(index / 128) + 1);
+    })());
+    const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", contentLength: 32 * 1024 * 1024, body }).catch(error => error);
+    try {
+      await expect.poll(() => started.size, { timeout: 1500 }).toBe(4);
+      expect(received.size).toBe(0); release();
+      expect(await saving).toBeUndefined();
+      expect(peak).toBe(4); expect(received.size).toBe(4); expect(completed).toBe(true);
+    } finally { release(); await saving; send.mockRestore(); client.destroy(); }
+  });
   it("sends concurrent exact part bodies through the real S3 HTTP client", async () => {
     const received = new Map<number, string>();
     const responses = new Map<number, ServerResponse>();
@@ -100,7 +134,7 @@ describe("streamed S3 multipart uploads", () => {
 
   it("streams exact ordered parts without buffering the full response", async () => {
     const client = new S3Client({ region: "us-east-1" });
-    const lengths: number[] = [];
+    const lengths = new Map<number, number>();
     const send = vi.spyOn(client, "send").mockImplementation(async (command: any) => {
       if (command instanceof CreateMultipartUploadCommand) return { UploadId: "upload" };
       if (command instanceof UploadPartCommand) {
@@ -111,7 +145,7 @@ describe("streamed S3 multipart uploads", () => {
           length += chunk.length;
         }
         expect(length).toBe(command.input.ContentLength);
-        lengths.push(length);
+        lengths.set(command.input.PartNumber!, length);
         return { ETag: `part-${command.input.PartNumber}` };
       }
       expect(command).toBeInstanceOf(CompleteMultipartUploadCommand);
@@ -121,7 +155,7 @@ describe("streamed S3 multipart uploads", () => {
     const body = Readable.from((async function* () { for (let i = 0; i < 272; i++) yield Buffer.alloc(64 * 1024); })());
     try {
       await putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "text/plain", contentLength: 17 * 1024 * 1024, body });
-      expect(lengths).toEqual([8, 8, 1].map(mb => mb * 1024 * 1024));
+      expect([...lengths].sort(([a], [b]) => a - b)).toEqual([[1, 8], [2, 8], [3, 1]].map(([part, mb]) => [part, mb * 1024 * 1024]));
     } finally { send.mockRestore(); client.destroy(); }
   });
   it("keeps source backpressure and at most four part requests while acknowledgements stall", async () => {
@@ -149,8 +183,8 @@ describe("streamed S3 multipart uploads", () => {
     })());
     const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", contentLength: 100 * 1024 * 1024, body });
     try {
-      await expect.poll(() => active).toBe(1);
-      expect(yielded).toBeLessThanOrEqual(3);
+      await expect.poll(() => active).toBe(4);
+      expect(yielded).toBeLessThanOrEqual(4 * 128 + 3);
       releaseReading();
       await expect.poll(() => received).toBe(4);
       expect(yielded).toBeLessThanOrEqual(4 * 128 + 3);
@@ -163,6 +197,76 @@ describe("streamed S3 multipart uploads", () => {
     } finally { releaseReading(); releaseAcknowledgements(); await saving; send.mockRestore(); client.destroy(); }
   });
 
+  it("shares the buffered payload window across simultaneous uploads", async () => {
+    const client = new S3Client({ region: "us-east-1" });
+    let release!: () => void, yielded = 0;
+    const reading = new Promise<void>(resolve => { release = resolve; });
+    const started = new Map<string, number>();
+    const send = vi.spyOn(client, "send").mockImplementation(async (command: any) => {
+      if (command instanceof CreateMultipartUploadCommand) return { UploadId: command.input.Key };
+      if (command instanceof UploadPartCommand) {
+        const key = command.input.Key!;
+        started.set(key, (started.get(key) ?? 0) + 1);
+        await reading;
+        let bytes = 0;
+        for await (const chunk of command.input.Body as Readable) bytes += chunk.length;
+        expect(bytes).toBe(command.input.ContentLength);
+        return { ETag: `part-${command.input.PartNumber}` };
+      }
+      expect(command).toBeInstanceOf(CompleteMultipartUploadCommand);
+      return {};
+    });
+    const saving = Promise.all(["a", "b"].map(key => putS3Multipart(client, "bucket", key, {
+      objectKey: key, contentType: "application/octet-stream", contentLength: 32 * 1024 * 1024,
+      body: Readable.from((async function* () {
+        for (let index = 0; index < 512; index++) { yielded++; yield Buffer.alloc(64 * 1024); }
+      })()),
+    })));
+    try {
+      await expect.poll(() => ["a", "b"].every(key => (started.get(key) ?? 0) >= 3)).toBe(true);
+      expect(yielded).toBeLessThanOrEqual(4 * 128 + 8);
+      release(); await saving;
+      expect(yielded).toBe(1024);
+      expect([...started.values()]).toEqual([4, 4]);
+    } finally { release(); await saving; send.mockRestore(); client.destroy(); }
+  });
+
+  it.each(["source", "request"])("releases the shared payload allowance after a %s failure", async failure => {
+    const client = new S3Client({ region: "us-east-1" });
+    let release!: () => void, yielded = 0, started = 0;
+    const reading = new Promise<void>(resolve => { release = resolve; });
+    const send = vi.spyOn(client, "send").mockImplementation(async (command: any) => {
+      if (command instanceof CreateMultipartUploadCommand) return { UploadId: command.input.Key };
+      if (command instanceof UploadPartCommand) {
+        if (command.input.Key === "failed") throw new Error("request failed");
+        started++; await reading;
+        for await (const _chunk of command.input.Body as Readable) {}
+        return { ETag: `part-${command.input.PartNumber}` };
+      }
+      expect(command instanceof CompleteMultipartUploadCommand || command instanceof AbortMultipartUploadCommand).toBe(true);
+      return {};
+    });
+    const first = Readable.from((async function* () {
+      if (failure === "source") throw new Error("source failed");
+      yield Buffer.alloc(8 * 1024 * 1024);
+    })());
+    let saving: Promise<void> | undefined;
+    try {
+      await expect(putS3Multipart(client, "bucket", "failed", {
+        objectKey: "failed", contentType: "application/octet-stream", contentLength: 8 * 1024 * 1024, body: first,
+      })).rejects.toThrow(`${failure} failed`);
+      saving = putS3Multipart(client, "bucket", "next", {
+        objectKey: "next", contentType: "application/octet-stream", contentLength: 32 * 1024 * 1024,
+        body: Readable.from((async function* () {
+          for (let index = 0; index < 512; index++) { yielded++; yield Buffer.alloc(64 * 1024); }
+        })()),
+      });
+      await expect.poll(() => started).toBe(4);
+      expect(yielded).toBe(512);
+      release(); await saving;
+    } finally { release(); await saving; send.mockRestore(); client.destroy(); }
+  });
+
   it("settles cancelled siblings before aborting a failed multipart upload", async () => {
     const client = new S3Client({ region: "us-east-1" });
     const cancellations: AbortSignal[] = [];
@@ -274,11 +378,11 @@ describe("streamed S3 multipart uploads", () => {
     } finally { send.mockRestore(); vi.useRealTimers(); }
   });
 
-  it("bounds cancellation while an asynchronous source read is stalled", async () => {
+  it.each([128 * 1024, 8 * 1024 * 1024 * 10_000 + 1])("bounds cancellation while an asynchronous source read is stalled (declared bytes: %s)", async contentLength => {
     vi.useFakeTimers();
     const client = new S3Client({ region: "us-east-1" });
     const cancellation = new AbortController();
-    let release!: () => void, waiting = false, settled = false;
+    let release!: () => void, waiting = false, settled = false, partLength: number | undefined;
     const sourceGate = new Promise<void>(resolve => { release = resolve; });
     const body = Readable.from((async function* () {
       yield Buffer.alloc(64 * 1024);
@@ -289,17 +393,19 @@ describe("streamed S3 multipart uploads", () => {
     const send = vi.spyOn(client, "send").mockImplementation(async (command: any, options: any) => {
       if (command instanceof CreateMultipartUploadCommand) return { UploadId: "upload" };
       if (command instanceof UploadPartCommand) {
+        partLength = command.input.ContentLength;
         (command.input.Body as Readable).resume();
         return new Promise((_resolve, reject) => options.abortSignal.addEventListener("abort", () => reject(options.abortSignal.reason), { once: true }));
       }
       expect(command).toBeInstanceOf(AbortMultipartUploadCommand);
       return {};
     });
-    const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "text/plain", contentLength: 128 * 1024, body, signal: cancellation.signal })
+    const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "text/plain", contentLength, body, signal: cancellation.signal })
       .catch(error => error).finally(() => { settled = true; });
     try {
       await vi.advanceTimersByTimeAsync(0);
       expect(waiting).toBe(true);
+      if (contentLength > 8 * 1024 * 1024 * 10_000) expect(partLength).toBe(8 * 1024 * 1024 + 1);
       cancellation.abort(new Error("cancelled during source read"));
       await vi.advanceTimersByTimeAsync(STORAGE_FAILURE_CLEANUP_MS + 1);
       expect(settled).toBe(true);

server/src/storage/s3-multipart.ts

@@ -5,6 +5,10 @@ import type { PutObjectInput } from "./types.js";
 import { withS3FailureDiagnostics } from "./s3-failure.js";
 import { cleanupStorageFailure } from "./failure-cleanup.js";
 
+const bufferedPartSize = 8 * 1024 * 1024;
+const bufferedPayloadLimit = 4 * bufferedPartSize;
+let bufferedPayloadBytes = 0;
+
 /** 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. */
 export async function putS3Multipart(client: S3Client, bucket: string, key: string, input: PutObjectInput) {
@@ -38,9 +42,24 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri
     finally { controller.signal.removeEventListener("abort", abort); }
   };
   let pending = Buffer.alloc(0);
+  const readPart = async (length: number, write: (chunk: Buffer) => Promise<void>) => {
+    while (length > 0) {
+      check();
+      if (!pending.length) {
+        const next = await readNext();
+        check();
+        if (next.done) throw new Error("S3 upload source was truncated");
+        pending = Buffer.isBuffer(next.value) ? next.value : Buffer.from(next.value);
+        if (!pending.length) continue;
+      }
+      const count = Math.min(length, pending.length, 64 * 1024), chunk = pending.subarray(0, count);
+      pending = pending.subarray(count); length -= count;
+      await write(chunk);
+    }
+  };
   let uploadId: string | undefined;
   // Scale part sizes to the S3 part-count limit while retaining stream backpressure.
-  const partSize = Math.max(8 * 1024 * 1024, Math.ceil(input.contentLength / 10_000));
+  const partSize = Math.max(bufferedPartSize, Math.ceil(input.contentLength / 10_000));
   try {
     check();
     uploadId = (await withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }))).UploadId;
@@ -52,31 +71,40 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri
       check();
       const length = Math.min(partSize, input.contentLength - offset);
       const number = ++partNumber;
+      let buffered: Buffer | null = null;
+      if (partSize <= bufferedPartSize && bufferedPayloadBytes + length <= bufferedPayloadLimit) {
+        bufferedPayloadBytes += length;
+        try {
+          const payload = Buffer.allocUnsafe(length);
+          let position = 0;
+          await readPart(length, async chunk => { chunk.copy(payload, position); position += chunk.length; });
+          buffered = payload;
+        } catch (error) { bufferedPayloadBytes -= length; throw error; }
+      }
       let remaining = length;
       const body = new PassThrough({ highWaterMark: 64 * 1024 });
       bodies.add(body);
-      const uploading = (async () => {
+      const write = async (chunk: Buffer) => {
+        check(); remaining -= chunk.length;
+        if (!body.write(chunk)) await once(body, "drain", { signal: controller.signal });
+      };
+      const sending = (async () => {
         const result = await withS3FailureDiagnostics("upload_part", () => client.send(new UploadPartCommand({ Bucket: bucket, Key: key, UploadId: uploadId, PartNumber: number, Body: body, ContentLength: length }), { abortSignal: controller.signal }));
         if (remaining || !body.readableEnded || !result.ETag) throw new Error("S3 upload part is incomplete");
         parts[number - 1] = { ETag: result.ETag, PartNumber: number };
-      })().catch(stop).finally(() => { active.delete(uploading); bodies.delete(body); body.destroy(); });
+      })().catch(stop);
+      const producing = buffered ? (async () => {
+        for (let position = 0; position < buffered.length; position += 64 * 1024)
+          await write(buffered.subarray(position, Math.min(position + 64 * 1024, buffered.length)));
+        body.end();
+      })().catch(stop) : Promise.resolve();
+      const uploading = Promise.all([sending, producing]).then(() => undefined)
+        .finally(() => { if (buffered) bufferedPayloadBytes -= length; active.delete(uploading); bodies.delete(body); body.destroy(); });
       active.add(uploading);
-      while (remaining > 0) {
-        check();
-        if (!pending.length) {
-          const next = await readNext();
-          check();
-          if (next.done) throw new Error("S3 upload source was truncated");
-          pending = Buffer.isBuffer(next.value) ? next.value : Buffer.from(next.value);
-          if (!pending.length) continue;
-        }
-        const count = Math.min(remaining, pending.length, 64 * 1024);
-        const chunk = pending.subarray(0, count);
-        pending = pending.subarray(count);
-        remaining -= count;
-        if (!body.write(chunk)) await once(body, "drain", { signal: controller.signal });
+      if (!buffered) {
+        await readPart(length, write);
+        body.end();
       }
-      body.end();
       offset += length;
     }
     if (pending.length || !(await readNext()).done) throw new Error("S3 upload source exceeds its declared size");