From 8812c1941d82d0f8960a2db272086cff57655cf0 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 06:58:10 -0600 Subject: [PATCH] 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 --- .../__tests__/storage-s3-multipart.test.ts | 122 ++++++++++++++++-- server/src/storage/s3-multipart.ts | 64 ++++++--- 2 files changed, 160 insertions(+), 26 deletions(-) diff --git a/server/src/__tests__/storage-s3-multipart.test.ts b/server/src/__tests__/storage-s3-multipart.test.ts index 418344700d7..0074dd88f22 100644 --- a/server/src/__tests__/storage-s3-multipart.test.ts +++ b/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(), received = new Map(); + let release!: () => void, active = 0, peak = 0, completed = false; + const reading = new Promise(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(); const responses = new Map(); @@ -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(); 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(resolve => { release = resolve; }); + const started = new Map(); + 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(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 | 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(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); diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index a8744eb8588..d6803f0a06b 100644 --- a/server/src/storage/s3-multipart.ts +++ b/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) => { + 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");