From 57ae2fc3093cbb91fbe63c751b6a72d62f5fe0a0 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:24:42 -0600 Subject: [PATCH] perf(workspaces): reduce snapshot stream scheduling overhead (#19900) Use bounded 1 MiB filesystem buffers during workspace archive hashing, encryption, and restore. Preserve integrity, ownership, cancellation, and snapshot format guarantees. Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 17 ++++ scripts/benchmark-workspace-archive-io.mjs | 91 +++++++++++++++++++ .../services/remote-workspace-archive-io.ts | 1 + .../remote-workspace-snapshot-storage.test.ts | 27 ++++-- .../remote-workspace-snapshot-storage.ts | 17 ++-- .../remote-workspace-transport.test.ts | 13 ++- .../services/remote-workspace-transport.ts | 3 +- 7 files changed, 152 insertions(+), 17 deletions(-) create mode 100644 scripts/benchmark-workspace-archive-io.mjs create mode 100644 server/src/services/remote-workspace-archive-io.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index f927e8224cc..fb9b1717564 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2179,6 +2179,23 @@ S3 multipart uploads stream up to four parts concurrently. One producer reads the source in order, and bounded stream buffers preserve backpressure. Failed uploads cancel and settle their active requests before multipart cleanup. Snapshot publication still requires full object hash and size verification. +Local snapshot hashing, encryption, and decryption use bounded 1 MiB filesystem +stream buffers to reduce event-loop handoffs under background load. Each stream +retains its existing integrity and cancellation checks; archive contents and +saved snapshot formats are unchanged. Reproduce the scheduling effect with: + +```sh +node --expose-gc scripts/benchmark-workspace-archive-io.mjs +``` + +This bounded benchmark alternates three baseline/treatment pairs over the same +32 MiB archive with 5 ms of competing main-thread work per turn. It runs real +hashing and encryption file passes, compares plaintext and ciphertext digests, +and reports callback counts and diagnostic peak memory growth. Larger buffers +trade additional bounded stream memory for fewer scheduling handoffs. The result +is synthetic; it does not identify the competing workload or establish hosted +finalization latency, and excludes sandbox, database, and object-storage time. + Preparation creates a saved baseline, and native finalization saves the stopped writer's files before releasing its persistence obligation. Replacing a webserver reconnects to the recorded allocation without copying files back to diff --git a/scripts/benchmark-workspace-archive-io.mjs b/scripts/benchmark-workspace-archive-io.mjs new file mode 100644 index 00000000000..f5103c9dfcb --- /dev/null +++ b/scripts/benchmark-workspace-archive-io.mjs @@ -0,0 +1,91 @@ +#!/usr/bin/env node +import { spawnSync } from "node:child_process"; +import { createCipheriv, createHash, randomBytes } from "node:crypto"; +import { createReadStream, createWriteStream } from "node:fs"; +import { mkdtemp, open, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { Transform } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import { fileURLToPath } from "node:url"; +import { WORKSPACE_ARCHIVE_IO_BUFFER_BYTES } from "../server/src/services/remote-workspace-archive-io.ts"; + +const baselineBytes = 64 * 1024; +const archiveBytes = 32 * 1024 * 1024 + 137; +const contentionMs = 5; + +async function worker(file, highWaterMark) { + const scratch = await mkdtemp(path.join(os.tmpdir(), "pc-archive-io-worker-")); + const encrypted = path.join(scratch, "encrypted"); + global.gc?.(); + const initial = process.memoryUsage(); + let peakRss = initial.rss, peakExternal = initial.external, chunks = 0, loadTurns = 0; + const observe = () => { + const memory = process.memoryUsage(); + peakRss = Math.max(peakRss, memory.rss); peakExternal = Math.max(peakExternal, memory.external); + }; + const digest = async name => { + const hash = createHash("sha256"); let bytes = 0; + for await (const chunk of createReadStream(name, { highWaterMark })) { + hash.update(chunk); bytes += chunk.length; chunks++; observe(); + } + return { sha256: hash.digest("hex"), bytes }; + }; + let active = true, pending; + const load = () => { + if (!active) return; + const until = performance.now() + contentionMs; + while (performance.now() < until) {} + loadTurns++; pending = setImmediate(load); + }; + pending = setImmediate(load); + const start = performance.now(); + try { + const verified = await digest(file), plaintext = await digest(file); + const cipher = createCipheriv("aes-256-gcm", Buffer.alloc(32, 1), Buffer.alloc(12, 2)); + cipher.setAAD(Buffer.from("synthetic-workspace-archive-io-benchmark")); + const header = await open(encrypted, "wx", 0o600); + await header.writeFile(Buffer.concat([Buffer.from("PCWSE1\n"), Buffer.alloc(12, 2)])); await header.close(); + const incoming = createHash("sha256"); let incomingBytes = 0; + const observed = new Transform({ transform(chunk, _encoding, callback) { + incoming.update(chunk); incomingBytes += chunk.length; chunks++; observe(); callback(null, chunk); + } }); + await pipeline(createReadStream(file, { highWaterMark }), observed, cipher, + createWriteStream(encrypted, { flags: "a", mode: 0o600, highWaterMark })); + const footer = await open(encrypted, "a"); await footer.writeFile(cipher.getAuthTag()); await footer.close(); + const ciphertext = await digest(encrypted); + if (verified.sha256 !== plaintext.sha256 || incoming.digest("hex") !== plaintext.sha256 + || verified.bytes !== archiveBytes || incomingBytes !== archiveBytes) throw new Error("Integrity comparison failed"); + return { highWaterMark, bytes: archiveBytes, chunks, loadTurns, elapsedMs: Math.round(performance.now() - start), + sha256: plaintext.sha256, encryptedSha256: ciphertext.sha256, + peakRssGrowthBytes: Math.max(0, peakRss - initial.rss), peakExternalGrowthBytes: Math.max(0, peakExternal - initial.external), + configuredReadWriteThresholdBytes: 2 * highWaterMark }; + } finally { active = false; clearImmediate(pending); await rm(scratch, { recursive: true, force: true }); } +} + +if (process.argv[2] === "--worker") { + console.log(JSON.stringify(await worker(process.argv[3], Number(process.argv[4])))); +} else { + const root = await mkdtemp(path.join(os.tmpdir(), "pc-archive-io-benchmark-")), file = path.join(root, "archive"); + try { + const handle = await open(file, "wx", 0o600), block = randomBytes(1024 * 1024); + try { for (let offset = 0; offset < archiveBytes; offset += block.length) + await handle.writeFile(block.subarray(0, Math.min(block.length, archiveBytes - offset))); } + finally { await handle.close(); } + const samples = []; + for (let iteration = 0; iteration < 3; iteration++) { + const order = iteration % 2 ? [WORKSPACE_ARCHIVE_IO_BUFFER_BYTES, baselineBytes] : [baselineBytes, WORKSPACE_ARCHIVE_IO_BUFFER_BYTES]; + for (const size of order) { + const result = spawnSync(process.execPath, ["--expose-gc", fileURLToPath(import.meta.url), "--worker", file, String(size)], + { encoding: "utf8", timeout: 60_000 }); + if (result.status !== 0) throw new Error(result.stderr || String(result.error)); + samples.push({ iteration, ...JSON.parse(result.stdout) }); + } + } + if (new Set(samples.map(sample => sample.sha256)).size !== 1 + || new Set(samples.map(sample => sample.encryptedSha256)).size !== 1) throw new Error("Baseline/treatment digests differ"); + console.log(JSON.stringify({ schema: "paperclip.workspace-archive-io-benchmark.v1", node: process.version, + platform: `${process.platform}/${process.arch}`, contentionMsPerTurn: contentionMs, + samples, identicalDigests: true, limits: "Synthetic main-thread contention; excludes sandbox, database, and object storage. RSS/GC samples are diagnostic. Stream thresholds bound queues, not total process memory." }, null, 2)); + } finally { await rm(root, { recursive: true, force: true }); } +} diff --git a/server/src/services/remote-workspace-archive-io.ts b/server/src/services/remote-workspace-archive-io.ts new file mode 100644 index 00000000000..af338322d85 --- /dev/null +++ b/server/src/services/remote-workspace-archive-io.ts @@ -0,0 +1 @@ +export const WORKSPACE_ARCHIVE_IO_BUFFER_BYTES = 1024 * 1024; diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 1451c747793..93d9419a88a 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { randomBytes, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -40,10 +40,10 @@ suite("remote workspace snapshot storage", () => { storage = createStorageService(createLocalDiskStorageProvider(path.join(root, "objects"))); }); afterEach(async () => { vi.unstubAllEnvs(); await db.delete(executionWorkspaceSnapshots); await fs.rm(root, { recursive: true, force: true }); }); - async function fixture() { + async function fixture(content: string | Buffer = "acknowledged workspace content\n") { const companyId = randomUUID(), workspaceId = randomUUID(), operationId = randomUUID(); const cwd = path.join(root, "source"), archivePath = path.join(root, "archive.pcws"); await fs.mkdir(cwd); - await fs.writeFile(path.join(cwd, "saved.txt"), "acknowledged workspace content\n"); + await fs.writeFile(path.join(cwd, "saved.txt"), content); const captured = await captureRemoteWorkspaceDirectory({ cwd, archivePath }); const row = await beginWorkspaceSnapshot(db, { companyId, workspaceId, operationId, generation: 1, epoch: 3 }); const upload = { db, storage, snapshotId: row.id, companyId, workspaceId, archivePath, manifestDigest: captured.manifestDigest }; @@ -63,6 +63,18 @@ suite("remote workspace snapshot storage", () => { const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref.id)); expect(row?.status).toBe("uploaded"); // The owning service still controls atomic publication. }); + it("preserves multi-chunk archive bytes through encryption, authenticated readback, and cold restore", async () => { + const content = randomBytes(3 * 1024 * 1024 + 137), f = await fixture(content); + const verifiedArchivePath = path.join(root, "verified-large.pcws"); + const ref = await uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath }); + expect((await fs.readFile(verifiedArchivePath)).equals(await fs.readFile(f.archivePath))).toBe(true); + await fs.rm(f.cwd, { recursive: true }); await fs.rm(f.archivePath); + const restoredArchive = path.join(root, "cold-large.pcws"), targetDir = path.join(root, "cold-large"); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: restoredArchive }); + expect((await fs.readFile(restoredArchive)).equals(await fs.readFile(verifiedArchivePath))).toBe(true); + await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect((await fs.readFile(path.join(targetDir, "saved.txt"))).equals(content)).toBe(true); + }); it.each(["fresh", "uploaded", "lost-reply"])("reuses one authenticated %s readback for a complete cold restore", async state => { const f = await fixture(), put = storage.putFile.bind(storage); if (state === "uploaded") await uploadWorkspaceSnapshot(f.upload); @@ -332,15 +344,16 @@ suite("remote workspace snapshot storage", () => { await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); }); it("cancels an active object body and removes private decryption scratch before rejecting", async () => { - const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload); + const f = await fixture(randomBytes(3 * 1024 * 1024 + 137)), ref = await uploadWorkspaceSnapshot(f.upload); const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey)); const controller = new AbortController(); - let produced = 0, closed = false; + let produced = 0, reads = 0, closed = false; const body = new Readable({ highWaterMark: 1, read() { - if (produced === 2) { controller.abort(new Error("client disconnected")); return; } - this.push(ciphertext.subarray(produced, ++produced)); + if (reads++ === 2) { controller.abort(new Error("client disconnected")); return; } + const start = produced; produced += 1024 * 1024 + 17; + this.push(ciphertext.subarray(start, produced)); }, destroy(error, callback) { closed = true; callback(error); }, }); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 697ce5a0821..4c520c90c8d 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -12,6 +12,7 @@ import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import type { StoredSecretVersionMaterial } from "../secrets/types.js"; import type { StorageService } from "../storage/types.js"; import { notifyRemoteWorkspaceSnapshots, REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS } from "./remote-workspace-snapshot-retention.js"; +import { WORKSPACE_ARCHIVE_IO_BUFFER_BYTES } from "./remote-workspace-archive-io.js"; const SCHEMA = "paperclip.workspace-snapshot.v1"; const KEY_SCHEMA = "paperclip.workspace-snapshot-key.v1"; @@ -55,7 +56,7 @@ function validateReference(ref: WorkspaceSnapshotRef) { } async function digestFile(file: string, signal?: AbortSignal) { const digest = createHash("sha256"); let byteSize = 0; - for await (const part of createReadStream(file, { signal })) { digest.update(part); byteSize += part.length; } + 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 verifyObject(storage: StorageService, ref: WorkspaceSnapshotRef, signal?: AbortSignal) { @@ -167,7 +168,7 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P const before = await handle.stat(), digest = createHash("sha256"); let bytes = 0; if (!before.isFile() || (before.mode & 0o777) !== 0o600 || before.nlink !== 1 || typeof process.getuid === "function" && before.uid !== process.getuid()) failure("readback_file_changed"); - for await (const chunk of createReadStream(file, { fd: handle.fd, autoClose: false, signal: input.signal })) { + for await (const chunk of createReadStream(file, { fd: handle.fd, autoClose: false, signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES })) { digest.update(chunk); bytes += chunk.length; } const after = await handle.stat(), named = await lstat(file); @@ -241,7 +242,7 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v try { await header.writeFile(Buffer.concat([MAGIC, iv])); } finally { await header.close(); } 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 }), observed, cipher, createWriteStream(encrypted, { flags: "a", mode: 0o600 }), { signal: input.signal }); + 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 footer = await open(encrypted, "a"); try { await footer.writeFile(cipher.getAuthTag()); } finally { await footer.close(); } dataKey.fill(0); @@ -275,7 +276,7 @@ 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, - originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted), + 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); if (recapture) { @@ -318,7 +319,7 @@ export async function loadWorkspaceSnapshotArchive(input: { received += bytes.length; digest.update(bytes); callback(received > ref.byteSize ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes); } }); - try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600 }), { signal }); } + try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal }); } finally { object.stream.destroy(); } } catch (error) { if (signal.aborted && signal.reason === stalled) throw stalled; @@ -332,8 +333,8 @@ export async function loadWorkspaceSnapshotArchive(input: { const decipher = createDecipheriv("aes-256-gcm", key, header.subarray(MAGIC.length)); decipher.setAAD(aad(ref)); decipher.setAuthTag(tag); try { - await pipeline(createReadStream(ciphertext, { start: header.length, end: ref.byteSize - tag.length - 1, signal: input.signal }), - decipher, createWriteStream(plaintext, { flags: "wx", mode: 0o600 }), { signal: input.signal }); + await pipeline(createReadStream(ciphertext, { start: header.length, end: ref.byteSize - tag.length - 1, signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), + decipher, createWriteStream(plaintext, { flags: "wx", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal }); } finally { key.fill(0); } const actual = await digestFile(plaintext, input.signal); if (actual.sha256 !== wrapped.plaintextSha256 || actual.byteSize !== wrapped.plaintextBytes) failure("plaintext_hash_mismatch"); @@ -341,7 +342,7 @@ export async function loadWorkspaceSnapshotArchive(input: { // failed/cancelled relay cannot leave a plaintext cache that looks complete. const staged = path.join(path.dirname(input.targetPath), `.workspace-snapshot-${randomUUID()}.tmp`); try { - await pipeline(createReadStream(plaintext, { signal: input.signal }), createWriteStream(staged, { flags: "wx", mode: 0o600 }), { signal: input.signal }); + await pipeline(createReadStream(plaintext, { signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), createWriteStream(staged, { flags: "wx", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal }); input.signal?.throwIfAborted(); await link(staged, input.targetPath); } finally { await rm(staged, { force: true }); } } finally { await rm(scratch, { recursive: true, force: true }); } diff --git a/server/src/services/remote-workspace-transport.test.ts b/server/src/services/remote-workspace-transport.test.ts index 65c7e406dfc..cfaa5eb7581 100644 --- a/server/src/services/remote-workspace-transport.test.ts +++ b/server/src/services/remote-workspace-transport.test.ts @@ -1,5 +1,5 @@ import { execFile } from "node:child_process"; -import { createHash, randomUUID } from "node:crypto"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -162,6 +162,17 @@ describe("remote workspace transport safety with real sandbox commands", () => { expect(await fs.lstat(root).catch(() => null)).toBeNull(); }); + it("verifies a multi-chunk downloaded archive and restores every byte", async () => { + const content = randomBytes(3 * 1024 * 1024 + 137); + await fs.writeFile(path.join(workspace, "large.bin"), content); + const captured = await captureRemoteWorkspaceArchive({ target: target().remote, remoteCwd: workspace, operationId: randomUUID() }); + try { + const targetDir = path.join(scratch, "large-restored"); + await restoreRemoteWorkspaceArchive({ archivePath: captured.archivePath, targetDir, expectedManifestDigest: captured.manifestDigest }); + expect((await fs.readFile(path.join(targetDir, "large.bin"))).equals(content)).toBe(true); + } finally { await captured.cleanup(); } + }); + it("adopts an atomic restored folder after a lost reply without overwriting later edits", async () => { await fs.writeFile(path.join(workspace, "saved.txt"), "saved synthetic files\n"); const { remote, execute } = target(); diff --git a/server/src/services/remote-workspace-transport.ts b/server/src/services/remote-workspace-transport.ts index 04ef461726a..f0e85f6bfd3 100644 --- a/server/src/services/remote-workspace-transport.ts +++ b/server/src/services/remote-workspace-transport.ts @@ -11,6 +11,7 @@ import { removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; import type { GitRemoteAuthProvider } from "./git-credentials.js"; import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge"; import { MAX_SANDBOX_SYNC_TIMEOUT_MS, type SandboxSyncOperation } from "@paperclipai/adapter-utils/sandbox-managed-runtime"; +import { WORKSPACE_ARCHIVE_IO_BUFFER_BYTES } from "./remote-workspace-archive-io.js"; export type RemoteWorkspaceTarget = Extract; @@ -580,7 +581,7 @@ async function captureSandboxWorkspaceArchive(input: { timing.transferMs = elapsedSince(transferStarted); const verifyStarted = performance.now(); const hash = createHash("sha256"); let bytes = 0; - for await (const chunk of createReadStream(archivePath)) { hash.update(chunk); bytes += chunk.length; } + for await (const chunk of createReadStream(archivePath, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES })) { hash.update(chunk); bytes += chunk.length; } timing.verifyMs = elapsedSince(verifyStarted); if (hash.digest("hex") !== receipt.sha256 || bytes !== receipt.byteSize) throw new Error("remote_workspace_capture_transfer_mismatch"); return { ...receipt, archivePath, cleanup: () => rm(root, { recursive: true, force: true }) };