From ee556a266d846f6ec22ecb8b3229c487cea10e2f Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 7 Nov 2026 14:04:49 -0800 Subject: [PATCH] perf(storage): upload workspace snapshots in parallel verified parts (#19917) 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. > - With remote (sandbox) execution, every agent turn ends by saving the workspace: capture an archive, encrypt it, upload it to object storage, and verify it. > - On a busy S3-backed instance, the upload step is the slowest part of a save. A 1.5 GB workspace can hold a run open for minutes after the agent finished. > - The upload step reads the archive several times, uploads through one stream with 4 parts in flight, and then downloads the whole object again to verify it. > - Parts that use the stream path share one process-wide 32 MiB buffer window, so saves that run at the same time slow each other down. > - This pull request uploads parts in parallel straight from the encrypted file, lets the backend verify every part with Content-MD5, and replaces the full read-back with a size check for those verified uploads. > - The benefit is faster saves, a smaller run-finalization window, and no interference between concurrent saves. ## Linked Issues or Issue Description **What existing behavior does this improve?** Remote workspace snapshot upload (`uploadWorkspaceSnapshot`) to S3-compatible storage. **Subsystem affected** Server storage (`server/src/storage/*`) and remote workspace snapshot storage. **Current behavior** For one save, the upload step: 1. hashes the archive; 2. reads the archive again to encrypt it into a temporary file; 3. hashes the encrypted file; 4. uploads it as one stream with 4 parts in flight; 5. downloads the whole object to verify it. On one hosted instance over 7 days (724 saves, median 731 MB), the upload step ran at a median of 6.5 MB/s. With no other save running it reached 11–14 MB/s. With six or more saves at the same time it fell to 4.4 MB/s. From the same region and with the same part settings, a plain client reached 31 MB/s to the same kind of bucket. **Proposed behavior** Upload parts in parallel from byte ranges of the encrypted file, with Content-MD5 on each part. Confirm only the stored size after a verified upload. Digest the archive in the same pass that encrypts it. **Reason and benefit** Saves finish faster and hold their run open for less time. Concurrent saves no longer serialize on the shared buffer window. A transient part failure is retried from disk instead of failing the whole save. **Breaking changes** None. `StorageProvider.putObject` may now return a receipt. Providers that return nothing keep the current behavior, including the full read-back. Related: none found. ## What Changed - `storage/types.ts`, `storage/service.ts`: an upload can name its `sourceFile`, a private file whose bytes are exactly the body. `putObject` may return `{ integrity: "server_verified_md5" }`, and `putFile` passes it through as `PutFileResult.integrity`. - `storage/s3-multipart.ts`: new `putS3MultipartFile`. - Each part reads its own byte range of the file. Up to 8 parts upload at once, with no shared buffer window. - Each part carries the Content-MD5 of its range. - A failed part is resent from disk, up to 3 attempts. - The first part that runs out of attempts stops its siblings. The multipart upload is aborted only after every part settles. - A source file whose size differs from the declared length is refused before any request. - `storage/s3-provider.ts`: - A `sourceFile` larger than 16 MiB uses `putS3MultipartFile`. A smaller one uses one PUT with Content-MD5. - These requests use a second client with `requestChecksumCalculation: "WHEN_REQUIRED"`, so bodies are plain and the backend verifies Content-MD5 directly. The default client wraps bodies in aws-chunked framing with a CRC32 trailer. - Other uploads are unchanged. - `services/remote-workspace-snapshot-storage.ts`: - Encryption digests the plaintext in the same read. An unchanged file identity (dev, inode, size, mtime, ctime) across that read replaces the separate digest pass. A change still fails with `archive_changed` before any upload. - The encrypted file is still digested from disk, so the recorded hash covers the exact bytes that the parts read. - The upload passes `sourceFile`. When the provider reports `server_verified_md5`, a HEAD confirms the stored size instead of downloading the object. - The full read-back stays for: - other providers (local disk); - recovery of earlier attempts; - the caller that keeps a verified archive file. ## Verification - `pnpm --filter @paperclipai/server typecheck`: exit 0. - `vitest run src/__tests__/storage-s3-multipart.test.ts src/services/remote-workspace-snapshot-storage.test.ts`: 59 passed. New tests: - every part carries the Content-MD5 of its range, and 8 parts are in flight; - a failed part is resent and the upload completes; - a part that keeps failing aborts the upload only after in-flight siblings settle; - a size mismatch is refused before any request; - a small file goes as one PUT with Content-MD5, through a real HTTP server; - the read-back is skipped only for a server-verified upload, and the snapshot still restores; - the read-back stays when the caller keeps the read-back file; - a stored size mismatch is rejected; - an archive that changes during encryption is rejected before upload. - With the source change reverted, the three new snapshot tests fail. - Real S3-compatible bucket (a Railway bucket, run from a workstation): - A 3 MiB single PUT stored exactly: HEAD size, ETag and read-back SHA-256 all matched. - A 96 MiB, 13-part upload stored exactly, with the same checks. - A part with a wrong Content-MD5 was rejected with `BadDigest`. - The same backend stored a wrong `x-amz-checksum-sha256` without checking it. That is why this change uses Content-MD5. - Measured from a container in the app's region, 1 GiB with the app's settings (8 MiB parts, 4 in flight), to the same kind of bucket: 31 MB/s. With 16 in flight: 69 MB/s. - 77 storage-related suites: the only failures are the same 19 that fail on unmodified `master` in this checkout. They are in 4 files: missing bundled connector skills and an unbuilt local runner daemon. ## Risks - Integrity now depends on the backend enforcing Content-MD5 for file uploads. AWS S3 and the tested S3-compatible backend enforce it. A backend that ignores Content-MD5 would accept corrupted parts without a read-back. A later restore still fails closed: AES-GCM authentication and the plaintext SHA-256 are checked on load. - Up to 8 parallel part requests per save, instead of 4 shared ones. Memory per part is a 64 KiB stream buffer. - Archive-change detection now uses file identity instead of a second digest. The archive lives in a private temporary directory that the capture step creates. - Medium risk overall: the change is limited to file-sourced S3 uploads and the snapshot verify step. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edits, test runs, S3 API probes). AI-assisted investigation, change and tests, with human direction. ## 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 - [ ] I have updated relevant documentation to reflect my changes (no docs describe this path) - [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 Co-authored-by: Paperclip --- .../__tests__/storage-s3-multipart.test.ts | 159 +++++++++++++++++- .../remote-workspace-snapshot-storage.test.ts | 56 ++++++ .../remote-workspace-snapshot-storage.ts | 37 ++-- server/src/storage/s3-multipart.ts | 91 +++++++++- server/src/storage/s3-provider.ts | 31 +++- server/src/storage/service.ts | 7 +- server/src/storage/types.ts | 21 ++- 7 files changed, 385 insertions(+), 17 deletions(-) diff --git a/server/src/__tests__/storage-s3-multipart.test.ts b/server/src/__tests__/storage-s3-multipart.test.ts index 0074dd88f22..f7b0690f2d0 100644 --- a/server/src/__tests__/storage-s3-multipart.test.ts +++ b/server/src/__tests__/storage-s3-multipart.test.ts @@ -4,7 +4,10 @@ import { createHash } from "node:crypto"; import { createServer, type ServerResponse } from "node:http"; import type { AddressInfo } from "node:net"; import { describe, expect, it, vi } from "vitest"; -import { putS3Multipart } from "../storage/s3-multipart.js"; +import { FILE_PART_CONCURRENCY, putS3Multipart, putS3MultipartFile } from "../storage/s3-multipart.js"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { createS3StorageProvider } from "../storage/s3-provider.js"; import { createStorageService } from "../storage/service.js"; import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js"; @@ -419,3 +422,157 @@ describe("streamed S3 multipart uploads", () => { } }); }); + +describe("file-backed S3 multipart uploads", () => { + const partBytes = 8 * 1024 * 1024; + async function sourceFile(parts: number, tail = 0) { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "pc-s3-file-part-")); + const file = path.join(dir, "source.bin"); + const handle = await fs.open(file, "wx", 0o600); + try { + for (let index = 0; index < parts; index++) await handle.write(Buffer.alloc(partBytes, index + 1)); + if (tail) await handle.write(Buffer.alloc(tail, 0xee)); + } finally { await handle.close(); } + return { dir, file, size: parts * partBytes + tail }; + } + const md5 = (bytes: Buffer) => createHash("md5").update(bytes).digest("base64"); + async function drain(body: unknown) { + const chunks: Buffer[] = []; + for await (const chunk of body as Readable) chunks.push(Buffer.from(chunk)); + return Buffer.concat(chunks); + } + + it("uploads ranges in parallel and sends each part's Content-MD5", async () => { + const source = await sourceFile(10, 4096); + const client = new S3Client({ region: "us-east-1" }); + let active = 0, peak = 0, release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const seen = new Map(); + const send = vi.spyOn(client, "send").mockImplementation(async (command: any) => { + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "upload" }; + if (command instanceof UploadPartCommand) { + active++; peak = Math.max(peak, active); + if (active === FILE_PART_CONCURRENCY) release(); + await gate; + const bytes = await drain(command.input.Body); + expect(bytes.length).toBe(command.input.ContentLength); + seen.set(command.input.PartNumber!, { length: bytes.length, md5: md5(bytes), declared: command.input.ContentMD5 }); + active--; + return { ETag: `part-${command.input.PartNumber}` }; + } + expect(command).toBeInstanceOf(CompleteMultipartUploadCommand); + expect(command.input.MultipartUpload.Parts.map((part: { PartNumber: number }) => part.PartNumber)).toEqual([...Array(11).keys()].map(n => n + 1)); + return {}; + }); + const body = Readable.from([Buffer.from("unused stream body")]); + try { + const receipt = await putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: source.size, body, sourceFile: source.file }); + expect(receipt).toEqual({ integrity: "server_verified_md5" }); + expect(peak).toBe(FILE_PART_CONCURRENCY); + expect(body.destroyed).toBe(true); + expect(seen.size).toBe(11); + for (const [number, part] of seen) { + expect(part.declared).toBe(part.md5); + expect(part.length).toBe(number === 11 ? 4096 : partBytes); + } + expect(seen.get(3)!.md5).toBe(md5(Buffer.alloc(partBytes, 3))); + } finally { send.mockRestore(); client.destroy(); await fs.rm(source.dir, { recursive: true, force: true }); } + }); + + it("resends a failed part from the file instead of failing the upload", async () => { + const source = await sourceFile(3); + const client = new S3Client({ region: "us-east-1" }); + const attempts = new Map(); + 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!; + attempts.set(number, (attempts.get(number) ?? 0) + 1); + const bytes = await drain(command.input.Body); + if (number === 2 && attempts.get(number) === 1) throw new Error("An error was encountered in a non-retryable streaming request."); + expect(md5(bytes)).toBe(command.input.ContentMD5); + return { ETag: `part-${number}` }; + } + if (command instanceof AbortMultipartUploadCommand) throw new Error("must not abort a recovered upload"); + return {}; + }); + try { + await expect(putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: source.size, body: Buffer.alloc(0), sourceFile: source.file })).resolves.toEqual({ integrity: "server_verified_md5" }); + expect(attempts.get(2)).toBe(2); + expect(send.mock.calls.some(([command]) => command instanceof CompleteMultipartUploadCommand)).toBe(true); + } finally { send.mockRestore(); client.destroy(); await fs.rm(source.dir, { recursive: true, force: true }); } + }); + + it("aborts the upload only after every in-flight part settles when a part keeps failing", async () => { + const source = await sourceFile(4); + const client = new S3Client({ region: "us-east-1" }); + let inFlight = 0, abortedWhileInFlight = false; + const send = vi.spyOn(client, "send").mockImplementation(async (command: any, options?: any) => { + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "upload" }; + if (command instanceof UploadPartCommand) { + inFlight++; + try { + if (command.input.PartNumber === 1) throw new Error("storage refused part"); + await new Promise((_resolve, reject) => options?.abortSignal?.addEventListener("abort", () => reject(options.abortSignal!.reason), { once: true })); + return { ETag: "never" }; + } finally { inFlight--; } + } + if (command instanceof AbortMultipartUploadCommand) { abortedWhileInFlight = inFlight > 0; return {}; } + throw new Error("must not complete a failed upload"); + }); + try { + await expect(putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: source.size, body: Buffer.alloc(0), sourceFile: source.file })).rejects.toThrow("storage refused part"); + const partOneAttempts = send.mock.calls.filter(([command]) => command instanceof UploadPartCommand && command.input.PartNumber === 1).length; + expect(partOneAttempts).toBe(3); + expect(send.mock.calls.filter(([command]) => command instanceof AbortMultipartUploadCommand)).toHaveLength(1); + expect(abortedWhileInFlight).toBe(false); + } finally { send.mockRestore(); client.destroy(); await fs.rm(source.dir, { recursive: true, force: true }); } + }); + + it("refuses a source file whose size differs from the declared length before starting an upload", async () => { + const source = await sourceFile(3); + const client = new S3Client({ region: "us-east-1" }); + const send = vi.spyOn(client, "send"); + try { + await expect(putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: source.size + 1, body: Buffer.alloc(0), sourceFile: source.file })).rejects.toThrow("does not match its declared size"); + expect(send).not.toHaveBeenCalled(); + } finally { send.mockRestore(); client.destroy(); await fs.rm(source.dir, { recursive: true, force: true }); } + }); + + it("sends a small file source as one PUT with Content-MD5 through the real S3 HTTP client", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "pc-s3-small-file-")); + const file = path.join(dir, "small.bin"), bytes = Buffer.from("verified snapshot bytes\n".repeat(1000)); + await fs.writeFile(file, bytes, { mode: 0o600 }); + const requests: Array<{ method: string; md5: string | undefined; body: Buffer }> = []; + const server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", chunk => chunks.push(chunk)); + request.on("end", () => { + requests.push({ method: request.method!, md5: request.headers["content-md5"] as string | undefined, body: Buffer.concat(chunks) }); + response.writeHead(200, { ETag: '"small"' }).end(); + }); + }); + await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); + vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture"); vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture"); + const service = createStorageService(createS3StorageProvider({ bucket: "bucket", region: "us-east-1", forcePathStyle: true, + endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}` })); + try { + const result = await service.putFile({ companyId: "company", objectKey: "company/workspace-snapshots/small.pcws", namespace: "workspace-snapshots", + originalFilename: null, contentType: "application/octet-stream", body: Readable.from([bytes]), byteSize: bytes.length, + sha256: createHash("sha256").update(bytes).digest("hex"), sourceFile: file }); + expect(result.integrity).toBe("server_verified_md5"); + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ method: "PUT", md5: md5(bytes) }); + expect(requests[0]!.body.equals(bytes)).toBe(true); + } finally { + vi.unstubAllEnvs(); + server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + await fs.rm(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 93d9419a88a..eb5507fe761 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -19,6 +19,17 @@ import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, recoverWorkspaceSnapshotArchive, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; +// Pass-through, so one test can report a changed archive between the two +// identity reads that bracket encryption. +const lstatOverride = vi.hoisted(() => ({ current: null as null | ((file: string, real: (file: string) => Promise) => Promise) })); +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + const lstat = ((file: Parameters[0], ...rest: unknown[]) => lstatOverride.current + ? lstatOverride.current(String(file), target => actual.lstat(target)) + : (actual.lstat as (...args: unknown[]) => unknown)(file, ...rest)) as typeof actual.lstat; + return { ...actual, lstat, default: { ...actual, lstat } }; +}); + const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL; if (externalDatabaseUrl && !/^remote_workspace_[a-z_]*test$/.test(new URL(externalDatabaseUrl).pathname.slice(1))) { throw new Error("Workspace storage tests require a disposable remote_workspace_*test database"); @@ -136,6 +147,51 @@ suite("remote workspace snapshot storage", () => { expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull(); expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false); }); + it("confirms only the stored size when the backend verified every uploaded part", async () => { + const f = await fixture(randomBytes(2 * 1024 * 1024 + 91)); const put = storage.putFile.bind(storage); + let sourceFile: string | undefined; + storage.putFile = async input => { + sourceFile = "sourceFile" in input ? input.sourceFile : undefined; + return { ...await put(input), integrity: "server_verified_md5" }; + }; + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); + const ref = await uploadWorkspaceSnapshot(f.upload); + expect(sourceFile).toMatch(/archive\.encrypted$/); + expect(get).not.toHaveBeenCalled(); + expect(head).toHaveBeenCalledWith(f.companyId, ref.objectKey, expect.anything()); + const restoredArchive = path.join(root, "restored-verified.pcws"); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: restoredArchive }); + expect((await fs.readFile(restoredArchive)).equals(await fs.readFile(f.archivePath))).toBe(true); + }); + it("still reads back a server-verified upload when the caller keeps the read-back file", async () => { + const f = await fixture(); const put = storage.putFile.bind(storage); + storage.putFile = async input => ({ ...await put(input), integrity: "server_verified_md5" }); + const get = vi.spyOn(storage, "getObject"), verifiedArchivePath = path.join(root, "kept.pcws"); + await uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath }); + expect(get).toHaveBeenCalled(); + expect((await fs.readFile(verifiedArchivePath)).equals(await fs.readFile(f.archivePath))).toBe(true); + }); + it("rejects a server-verified upload whose stored size differs", async () => { + const f = await fixture(); const put = storage.putFile.bind(storage), head = storage.headObject.bind(storage); + storage.putFile = async input => ({ ...await put(input), integrity: "server_verified_md5" }); + storage.headObject = async (...args) => { const result = await head(...args); return { ...result, contentLength: (result.contentLength ?? 0) + 1 }; }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("object_size_mismatch"); + const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id)); + expect(row?.status).toBe("pending"); + }); + it("rejects an archive that changes while it is encrypted, before any upload", async () => { + const f = await fixture(); const put = vi.spyOn(storage, "putFile"); + let reads = 0; + lstatOverride.current = async (file, real) => { + const stats = await real(file); + if (file !== f.archivePath || ++reads < 2) return stats; + return Object.assign(Object.create(Object.getPrototypeOf(stats)), stats, { mtimeMs: stats.mtimeMs + 1 }); + }; + try { await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("archive_changed"); } + finally { lstatOverride.current = null; } + expect(reads).toBe(2); + expect(put).not.toHaveBeenCalled(); + }); it("reconciles an upload whose reply was lost after object persistence, without the old scratch file", async () => { const f = await fixture(); const put = storage.putFile.bind(storage); let loseReply = true; storage.putFile = async input => { const result = await put(input); if (loseReply) { loseReply = false; throw new Error("lost upload reply"); } return result; }; diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 4c520c90c8d..a9ad38d0208 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -59,6 +59,10 @@ async function digestFile(file: string, signal?: AbortSignal) { for await (const part of createReadStream(file, { signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES })) { digest.update(part); byteSize += part.length; } return { sha256: digest.digest("hex"), byteSize }; } +async function verifyStoredSize(storage: StorageService, ref: WorkspaceSnapshotRef, signal?: AbortSignal) { + const head = await storage.headObject(ref.companyId, ref.objectKey, { signal }); + if (!head.exists || head.contentLength !== ref.byteSize) failure("object_size_mismatch"); +} async function verifyObject(storage: StorageService, ref: WorkspaceSnapshotRef, signal?: AbortSignal) { const object = await storage.getObject(ref.companyId, ref.objectKey, { signal }); const digest = createHash("sha256"); let size = 0; @@ -135,7 +139,7 @@ type WorkspaceSnapshotUpload = { /** Upload attempts are journaled before remote I/O. The caller owns publication. */ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): Promise { if (input.verifiedArchivePath === undefined) - return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal)); + return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal), { acceptServerVerifiedUpload: true }); input.signal?.throwIfAborted(); const target = input.verifiedArchivePath, parent = path.dirname(target); if (!path.isAbsolute(target) || path.normalize(target) !== target) failure("readback_parent_invalid"); @@ -160,7 +164,8 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P verified.push({ ref: structuredClone(ref), file }); }; try { - const ref = await uploadVerifiedWorkspaceSnapshot(input, verify); + // This caller keeps the read-back file, so every upload is read back. + const ref = await uploadVerifiedWorkspaceSnapshot(input, verify, { acceptServerVerifiedUpload: false }); await verify(ref); const file = verified.find(saved => isDeepStrictEqual(saved.ref, ref))!.file; const wrapped = await snapshotKey(ref), handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); @@ -183,7 +188,8 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P } finally { await rm(scratch, { recursive: true, force: true }); } } -async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise): Promise { +async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise, + options: { acceptServerVerifiedUpload: boolean }): Promise { const [row] = await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)); if (!row) failure("missing"); if (input.unpublishedRecapture && row!.sourceOperationId !== input.unpublishedRecapture.sourceOperationId) failure("operation_binding_changed"); @@ -235,17 +241,23 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v try { const encrypted = path.join(scratch, "archive.encrypted"), dataKey = randomBytes(32), iv = randomBytes(12); const cipher = createCipheriv("aes-256-gcm", dataKey, iv); cipher.setAAD(aad(identity)); - const plaintext = await digestFile(input.archivePath, input.signal); - const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ schema: KEY_SCHEMA, - ...identity, dataKey: dataKey.toString("base64"), plaintextSha256: plaintext.sha256, plaintextBytes: plaintext.byteSize }) }); const header = await open(encrypted, "wx", 0o600); try { await header.writeFile(Buffer.concat([MAGIC, iv])); } finally { await header.close(); } + // One read of the archive both digests and encrypts it. An unchanged file + // identity across that read replaces the separate digest pass. + const before = await lstat(input.archivePath); const incomingDigest = createHash("sha256"); let incomingBytes = 0; const observed = new Transform({ transform(bytes, _encoding, callback) { incomingDigest.update(bytes); incomingBytes += bytes.length; callback(null, bytes); } }); await pipeline(createReadStream(input.archivePath, { signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), observed, cipher, createWriteStream(encrypted, { flags: "a", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal }); - if (incomingBytes !== plaintext.byteSize || incomingDigest.digest("hex") !== plaintext.sha256) failure("archive_changed"); + const after = await lstat(input.archivePath); + if (!before.isFile() || before.isSymbolicLink() || incomingBytes !== before.size || after.dev !== before.dev || after.ino !== before.ino + || after.size !== before.size || after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs) failure("archive_changed"); + const plaintext = { sha256: incomingDigest.digest("hex"), byteSize: incomingBytes }; const footer = await open(encrypted, "a"); try { await footer.writeFile(cipher.getAuthTag()); } finally { await footer.close(); } + const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ schema: KEY_SCHEMA, + ...identity, dataKey: dataKey.toString("base64"), plaintextSha256: plaintext.sha256, plaintextBytes: plaintext.byteSize }) }); dataKey.fill(0); + // Digest the encrypted file as stored: upload parts are read from these bytes. const content = await digestFile(encrypted, input.signal); const attempt: UploadAttempt = { objectKey: `${identity.companyId}/workspace-snapshots/${identity.workspaceId}/${identity.id}/${randomUUID()}.pcws`, ...content, key: material, manifestDigest: input.manifestDigest }; @@ -275,10 +287,15 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v } // The object key is unique per attempt, so a late losing uploader cannot // overwrite an object already used by a committed generation. - await input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, + const stored = await input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), - byteSize: attempt.byteSize, sha256: attempt.sha256, signal: input.signal }); - await verify({ schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef); + byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal }); + const uploaded = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; + // The backend checked every part of this upload against digests of these + // exact bytes, so only its size needs confirming. Any other provider, and + // a caller that needs the bytes back, still reads the object back. + if (options.acceptServerVerifiedUpload && stored.integrity === "server_verified_md5") await verifyStoredSize(input.storage, uploaded, input.signal); + else await verify(uploaded); if (recapture) { const recovered = await availableAttempt(attempts); if (recovered) return publish(recovered); diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index d6803f0a06b..0c187233eae 100644 --- a/server/src/storage/s3-multipart.ts +++ b/server/src/storage/s3-multipart.ts @@ -1,7 +1,11 @@ import { AbortMultipartUploadCommand, CompleteMultipartUploadCommand, CreateMultipartUploadCommand, UploadPartCommand, type S3Client } from "@aws-sdk/client-s3"; +import { createHash } from "node:crypto"; import { once } from "node:events"; +import { createReadStream } from "node:fs"; +import { stat } from "node:fs/promises"; import { PassThrough, Readable } from "node:stream"; -import type { PutObjectInput } from "./types.js"; +import { setTimeout as delay } from "node:timers/promises"; +import type { PutObjectInput, PutObjectReceipt } from "./types.js"; import { withS3FailureDiagnostics } from "./s3-failure.js"; import { cleanupStorageFailure } from "./failure-cleanup.js"; @@ -122,3 +126,88 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri await cleanupStorageFailure(async () => iterator.return?.()); } } + +export const FILE_PART_CONCURRENCY = 8; +const FILE_PART_ATTEMPTS = 3; +const FILE_READ_BYTES = 1024 * 1024; + +/** 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"); + for await (const chunk of createReadStream(file, { start, end, signal, highWaterMark: FILE_READ_BYTES })) digest.update(chunk); + return digest.digest("base64"); +} + +export async function assertSourceFileSize(file: string, contentLength: number) { + const stats = await stat(file); + if (!stats.isFile() || stats.size !== contentLength) throw new Error("S3 upload source file does not match its declared size"); +} + +/** A file source is read per part instead of through one stream. Parts upload + * in parallel without the shared memory window, each carries Content-MD5 so + * the backend rejects bytes it did not receive intact, and a failed part is + * resent from disk instead of failing the whole upload. */ +export async function putS3MultipartFile( + client: S3Client, bucket: string, key: string, input: PutObjectInput & { sourceFile: string }, +): Promise { + const file = input.sourceFile; + const controller = new AbortController(); + const onAbort = () => controller.abort(input.signal!.reason); + input.signal?.addEventListener("abort", onAbort, { once: true }); + if (input.signal?.aborted) onAbort(); + // This path never reads the stream body; release its file descriptor now. + if (input.body instanceof Readable) input.body.destroy(); + let uploadId: string | undefined; + try { + await assertSourceFileSize(file, input.contentLength); + controller.signal.throwIfAborted(); + const partSize = Math.max(bufferedPartSize, Math.ceil(input.contentLength / 10_000)); + const count = Math.ceil(input.contentLength / partSize); + uploadId = (await withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }))).UploadId; + if (!uploadId) throw new Error("S3 did not return a multipart upload id"); + const id = uploadId; + const parts: { ETag: string; PartNumber: number }[] = new Array(count); + const sendPart = async (index: number) => { + const start = index * partSize, length = Math.min(partSize, input.contentLength - start), end = start + length - 1; + const contentMd5 = await fileRangeMd5(file, start, end, controller.signal); + for (let attempt = 1; ; attempt++) { + controller.signal.throwIfAborted(); + const body = createReadStream(file, { start, end, highWaterMark: 64 * 1024 }); + try { + const result = await withS3FailureDiagnostics("upload_part", () => client.send(new UploadPartCommand({ + Bucket: bucket, Key: key, UploadId: id, PartNumber: index + 1, Body: body, ContentLength: length, ContentMD5: contentMd5, + }), { abortSignal: controller.signal })); + if (!result.ETag) throw new Error("S3 upload part is incomplete"); + parts[index] = { ETag: result.ETag, PartNumber: index + 1 }; + return; + } catch (error) { + // A part read from disk can be resent; the stream path cannot. + if (controller.signal.aborted || attempt >= FILE_PART_ATTEMPTS) throw error; + await delay(attempt * 500, undefined, { signal: controller.signal }); + } finally { body.destroy(); } + } + }; + let next = 0; + const worker = async () => { + while (next < count) { + const index = next++; + try { await sendPart(index); } + catch (error) { if (!controller.signal.aborted) controller.abort(error); throw error; } + } + }; + // Wait for every sibling to settle before the multipart upload is aborted. + const settled = await Promise.allSettled(Array.from({ length: Math.min(FILE_PART_CONCURRENCY, count) }, worker)); + const rejected = settled.find((entry): entry is PromiseRejectedResult => entry.status === "rejected"); + if (rejected) throw controller.signal.aborted && controller.signal.reason !== undefined ? controller.signal.reason : rejected.reason; + controller.signal.throwIfAborted(); + await withS3FailureDiagnostics("complete_multipart_upload", () => client.send(new CompleteMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: id, MultipartUpload: { Parts: parts } }), { abortSignal: controller.signal })); + uploadId = undefined; + return { integrity: "server_verified_md5" }; + } catch (error) { + const failedUpload = uploadId; + if (failedUpload) await cleanupStorageFailure(signal => client.send(new AbortMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: failedUpload }), { abortSignal: signal })); + throw error; + } finally { + input.signal?.removeEventListener("abort", onAbort); + } +} diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index 6edd3e4e152..a87d279982c 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -7,7 +7,8 @@ import { ListObjectsV2Command, ListObjectVersionsCommand, } from "@aws-sdk/client-s3"; -import { putS3Multipart } from "./s3-multipart.js"; +import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile } from "./s3-multipart.js"; +import { createReadStream } from "node:fs"; import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js"; import { addAbortSignal, Readable } from "node:stream"; import { retryS3RequestTimeouts } from "./s3-request-timeout-retry.js"; @@ -81,6 +82,16 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid forcePathStyle: Boolean(config.forcePathStyle), }); retryS3RequestTimeouts(client); + // File sources carry Content-MD5 as their integrity check. Send plain bodies + // so the backend verifies that digest directly; the default client wraps + // bodies in aws-chunked framing with its own trailing checksum. + const fileClient = new S3Client({ + region, + endpoint: config.endpoint, + forcePathStyle: Boolean(config.forcePathStyle), + requestChecksumCalculation: "WHEN_REQUIRED", + }); + retryS3RequestTimeouts(fileClient); async function listExactVersions(key: string, signal?: AbortSignal) { const versions = new Set(), cursors = new Set(); @@ -140,6 +151,24 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid async putObject(input) { const key = buildKey(prefix, input.objectKey); + const sourceFile = input.sourceFile; + if (sourceFile && input.contentLength > 16 * 1024 * 1024) { + return putS3MultipartFile(fileClient, bucket, key, { ...input, sourceFile }); + } + if (sourceFile) { + // Content-MD5 makes the backend reject any byte it did not receive intact. + 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(); } + return { integrity: "server_verified_md5" } as const; + } if (input.contentLength > 16 * 1024 * 1024) { await putS3Multipart(client, bucket, key, input); return; diff --git a/server/src/storage/service.ts b/server/src/storage/service.ts index 69023bc5bb2..b742bef8d66 100644 --- a/server/src/storage/service.ts +++ b/server/src/storage/service.ts @@ -1,7 +1,7 @@ import { createHash, randomUUID } from "node:crypto"; import path from "node:path"; import { Readable } from "node:stream"; -import type { StorageService, StorageProvider, PutFileInput, PutFileResult } from "./types.js"; +import type { StorageService, StorageProvider, PutFileInput, PutFileResult, PutObjectReceipt } from "./types.js"; import { badRequest, forbidden, unprocessable } from "../errors.js"; import { cleanupStorageFailure } from "./failure-cleanup.js"; @@ -114,13 +114,15 @@ export function createStorageService(provider: StorageProvider): StorageService 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 { - await provider.putObject({ + 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 })); @@ -136,6 +138,7 @@ export function createStorageService(provider: StorageProvider): StorageService byteSize, sha256: "sha256" in input ? input.sha256 : hashBuffer(input.body), originalFilename: input.originalFilename, + ...(receipt ? { integrity: receipt.integrity } : {}), }; }, diff --git a/server/src/storage/types.ts b/server/src/storage/types.ts index 6ab5a725664..11cb9449f41 100644 --- a/server/src/storage/types.ts +++ b/server/src/storage/types.ts @@ -9,6 +9,21 @@ export interface PutObjectInput { body: Buffer | Readable; contentType: string; contentLength: number; + /** + * Absolute path of a private file whose bytes are exactly `body`. A provider + * may read ranges of it instead of the stream, so parts can upload in + * parallel and be resent after a transient failure. + */ + sourceFile?: string; +} + +/** + * The backend verified every uploaded byte against a digest the client sent + * with it (Content-MD5 on each request), so a read-back is not needed to + * prove the stored content. + */ +export interface PutObjectReceipt { + integrity: "server_verified_md5"; } export interface GetObjectInput { @@ -46,7 +61,7 @@ export interface HeadObjectResult { export interface StorageProvider { id: StorageProviderId; - putObject(input: PutObjectInput): Promise; + putObject(input: PutObjectInput): Promise; getObject(input: GetObjectInput): Promise; headObject(input: GetObjectInput): Promise; deleteObject(input: DeleteObjectInput): Promise; @@ -70,7 +85,7 @@ export type PutFileInput = { namespace: string; originalFilename: string | null; contentType: string; -} & ({ body: Buffer } | { body: Readable; byteSize: number; sha256: string }); +} & ({ body: Buffer } | { body: Readable; byteSize: number; sha256: string; sourceFile?: string }); export interface PutFileResult { provider: StorageProviderId; @@ -79,6 +94,8 @@ export interface PutFileResult { byteSize: number; sha256: string; originalFilename: string | null; + /** Present only when the backend verified every byte on receipt. */ + integrity?: PutObjectReceipt["integrity"]; } export interface StorageService {