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 {