From 2f47d0fadbeb20d06a278d8bf0276a0ea73a360c Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:42:11 -0600 Subject: [PATCH] perf(workspace): run portable archives in bounded CPU workers (#20404) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip preserves agent work when its control plane changes. > - Workspace imports capture and verify portable filesystem archives. > - The archive function uses asynchronous I/O but hashes and handles manifests on the caller's JavaScript thread. > - Concurrent imports therefore share one CPU execution context for that work. > - This pull request runs the same archive function in bounded native Node workers. > - The parent keeps all database, storage, source locks and publication decisions. ## Linked Issues or Issue Description **What happened?** Concurrent administrative workspace imports can spend many minutes capturing archives while the owning Node process uses about one CPU core. Promise concurrency does not move hashing and manifest work off that process's JavaScript thread. **Expected behavior** Independent archive calls use separate CPU execution contexts. At most eight archive workers run in one process. Cancellation must wait for file cleanup and actual worker exit before the caller releases its locks. **Steps to reproduce** Run several workspace captures in one process. The existing wrappers call the same archive function on the parent thread. The regression holds eight real worker threads at a CPU barrier while the parent remains responsive, then cancels a queued ninth call without spawning it. **Deployment mode** All server callers of the portable capture and restore wrappers. Related import work: https://github.com/paperclipai/paperclip/pull/20311 and https://github.com/paperclipai/paperclip/pull/20394. ## What Changed - Run the existing self-contained archive function in native workers. Its algorithm and format stay byte-identical. - Limit active workers to eight per process, with a 1 GiB old-generation heap limit each. Workers stop when their owned call finishes. - Forward cancellation to the worker's AbortController. Wait for actual exit before returning, throwing or releasing an active slot. Cancel queued calls before they spawn. - Own capture scratch in a private parent-created worker temporary root. Remove it after actual exit, including abnormal exit. - Keep direct archive algorithm fault tests on the canonical implementation. Add real-worker byte, manifest, restore, CPU barrier and cancellation tests. ## Verification - Four real-worker regressions passed: forced-exit capture scratch cleanup; exact capture bytes/manifest and restored files/modes/links; eight CPU barriers plus queued abort; abort cleanup joined through actual worker exit. - Canonical workspaceArchive declaration/body hash is unchanged from the base: `d6cffeab112b817fab8f65a3c0d8ecdc8d23589d37bf44ae30fe9e53213d54f4`. - The combined archive run passed 114 cases. Three Unix-socket cases failed because the sandbox denied listen with EPERM; a focused retry with socket permission passed all three. The initial separate worker run passed all three cases; the new head passed all four, including forced-exit cleanup. All server layered typechecks and module boundary checks passed. - The initial ac4bc5760d source passed a synthetic smoke on staging’s Linux Node v24.22.2 through the installed tsx loader: direct/worker byte and digest parity, worker restore of three files plus a symlink, and private fixture cleanup. It made no database, object-storage or provider calls. - Greptile reviewed final head `3a51fe67ff17a56b2ddd7877c89734ba60afd18e` at 5/5 with no actionable issues. All applicable final-head CI passed, including aggregate verify, typecheck, build, workspace consumer suites and all server shards. The second review remains queued. No measured migration speedup claim is made yet. ## Risks Each active worker adds a bounded V8 heap and native overhead. Eight workers can use more memory and I/O bandwidth than the former single JavaScript thread. The existing 64 MiB manifest limit and streaming file buffers stay in place. Worker failures refuse the call and cannot publish authority. The canonical restore staging path stays unchanged; this PR does not claim cleanup after every arbitrary process crash. Cooperative cancellation retains the existing cleanup path; the parent waits for actual worker exit. No database, key, object-storage or provider behavior changes. ## Model Used OpenAI Codex, based on GPT-6, with tool use and code execution. The exact deployment identifier and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs: #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../remote-workspace-archive-worker.test.ts | 121 ++++++++++++++++++ .../remote-workspace-snapshot-archive.test.ts | 5 +- .../remote-workspace-snapshot-archive.ts | 81 +++++++++++- 3 files changed, 203 insertions(+), 4 deletions(-) create mode 100644 server/src/services/remote-workspace-archive-worker.test.ts diff --git a/server/src/services/remote-workspace-archive-worker.test.ts b/server/src/services/remote-workspace-archive-worker.test.ts new file mode 100644 index 00000000000..a1302befdb3 --- /dev/null +++ b/server/src/services/remote-workspace-archive-worker.test.ts @@ -0,0 +1,121 @@ +import { createHash } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const state = vi.hoisted(() => ({ control: undefined as SharedArrayBuffer | undefined, + mode: "" as "" | "cpu" | "abort" | "crash", spawned: 0, exited: 0, roots: [] as string[] })); +vi.mock("node:worker_threads", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, Worker: class extends actual.Worker { + constructor(source: string | URL, options?: import("node:worker_threads").WorkerOptions) { + if (state.control && typeof source === "string") { + const setup = `const testControl=new Int32Array(workerData.testControl);`; + source = source.replace("const archive=", `${setup}const archive=`); + if (state.mode === "cpu") source = source.replace("const before = await walk(root, true);", ` + Atomics.add(testControl,0,1); + while(!Atomics.load(testControl,1)) Atomics.add(testControl,2,1); + const before = await walk(root, true);`); + if (state.mode === "abort") { + source = source.replace("const before = await walk(root, true);", ` + Atomics.store(testControl,0,1); + await new Promise(resolve=>{input.signal.addEventListener('abort',resolve,{once:true});if(input.signal.aborted)resolve();}); + check(); + const before = await walk(root, true);`); + source = source.replace(/await fs\.rm\(scratch,\s*\{\s*recursive:\s*true,\s*force:\s*true\s*\}\);?/, ` + Atomics.store(testControl,3,1); + while(!Atomics.load(testControl,4)) await new Promise(resolve=>setTimeout(resolve,5)); + await fs.rm(scratch, { recursive: true, force: true });`); + } + if (state.mode === "crash") source = source.replace("const before = await walk(root, true);", ` + await fs.mkdir(path.join(scratch,'copied-repository')); + await fs.writeFile(path.join(scratch,'copied-repository','retained'),'synthetic repository copy'); + process.exit(73); + const before = await walk(root, true);`); + options = { ...options, workerData: { ...options?.workerData, testControl: state.control } }; + } + super(source, options); + if (options?.env && typeof options.env === "object" && typeof options.env.TMPDIR === "string") state.roots.push(options.env.TMPDIR); + state.spawned++; + this.once("exit", () => { state.exited++; }); + } + } }; +}); +import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive, workspaceArchive } from "./remote-workspace-snapshot-archive.js"; + +describe("workspace archive worker boundary", () => { + let root: string, cwd: string; + beforeEach(async () => { + state.control = undefined; state.mode = ""; state.spawned = 0; state.exited = 0; state.roots = []; + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-archive-worker-test-")); + cwd = path.join(root, "source"); await fs.mkdir(cwd); + await fs.writeFile(path.join(cwd, "saved.txt"), "retained work\n"); + await fs.mkdir(path.join(cwd, "nested")); + await fs.writeFile(path.join(cwd, "nested", "bytes"), Buffer.alloc(1024 * 1024, 42)); + await fs.chmod(path.join(cwd, "saved.txt"), 0o751); + await fs.symlink("saved.txt", path.join(cwd, "link")); + }); + afterEach(async () => { + if (state.control) { const control = new Int32Array(state.control); Atomics.store(control,1,1); Atomics.store(control,4,1); } + await fs.rm(root, { recursive: true, force: true }); + }); + it("preserves exact archive bytes, manifest and restored files across real workers", async () => { + const directPath = path.join(root, "direct.pcws"), archivePath = path.join(root, "worker.pcws"); + const direct = await workspaceArchive({ kind: "capture", cwd, archivePath: directPath }); + const captured = await captureRemoteWorkspaceDirectory({ cwd, archivePath }); + expect(captured).toEqual(direct); + expect(await fs.readFile(archivePath)).toEqual(await fs.readFile(directPath)); + const targetDir = path.join(root, "restored"); + await expect(restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: direct.manifestDigest })).resolves.toEqual(direct); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("retained work\n"); + expect((await fs.stat(path.join(targetDir, "saved.txt"))).mode & 0o777).toBe(0o751); + expect(await fs.readlink(path.join(targetDir, "link"))).toBe("saved.txt"); + expect(createHash("sha256").update(await fs.readFile(path.join(targetDir, "nested", "bytes"))).digest("hex")) + .toBe(createHash("sha256").update(await fs.readFile(path.join(cwd, "nested", "bytes"))).digest("hex")); + expect(state.spawned).toBe(2); expect(state.exited).toBe(2); + }); + it("runs CPU barriers on eight real workers while an aborted queued capture never spawns", async () => { + state.control = new SharedArrayBuffer(5 * 4); state.mode = "cpu"; + const control = new Int32Array(state.control), controller = new AbortController(), reason = new Error("queued capture cancelled"); + const captures = Array.from({ length: 8 }, (_, i) => captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, `${i}.pcws`) })); + const queued = captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "queued.pcws"), signal: controller.signal }).catch(error => error); + try { + await expect.poll(() => Atomics.load(control,0), { timeout: 10_000 }).toBe(8); + expect(state.spawned).toBe(8); expect(state.exited).toBe(0); + expect(Atomics.load(control,2)).toBeGreaterThan(0); + controller.abort(reason); + expect(await queued).toBe(reason); expect(state.spawned).toBe(8); + Atomics.store(control,1,1); + await Promise.all(captures); + expect(state.exited).toBe(8); + await expect(fs.access(path.join(root, "queued.pcws"))).rejects.toMatchObject({ code: "ENOENT" }); + } finally { Atomics.store(control,1,1); await Promise.allSettled(captures); } + }); + it("removes parent-owned capture scratch after an abnormal worker exit", async () => { + state.control = new SharedArrayBuffer(5 * 4); state.mode = "crash"; + await expect(captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "crashed.pcws") })) + .rejects.toThrow("remote_workspace_archive_worker_failed"); + expect(state.exited).toBe(1); expect(state.roots).toHaveLength(1); + await expect(fs.access(state.roots[0]!)).rejects.toMatchObject({ code: "ENOENT" }); + expect(await fs.readFile(path.join(cwd,"saved.txt"),"utf8")).toBe("retained work\n"); + }); + it("waits for abort cleanup and actual worker exit before rejecting or releasing its slot", async () => { + state.control = new SharedArrayBuffer(5 * 4); state.mode = "abort"; + const control = new Int32Array(state.control), controller = new AbortController(), reason = new Error("owned capture cancelled"); + let settled = false; + const archivePath = path.join(root, "cancelled.pcws"); + const capture = captureRemoteWorkspaceDirectory({ cwd, archivePath, signal: controller.signal }).then( + () => { settled = true; return null; }, error => { settled = true; return error; }); + try { + await expect.poll(() => Atomics.load(control,0), { timeout: 10_000 }).toBe(1); + controller.abort(reason); + await expect.poll(() => Atomics.load(control,3), { timeout: 10_000 }).toBe(1); + expect(settled).toBe(false); expect(state.exited).toBe(0); + Atomics.store(control,4,1); + expect(await capture).toBe(reason); + expect(state.exited).toBe(1); + await expect(fs.access(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { Atomics.store(control,4,1); controller.abort(reason); await capture; } + }); +}); diff --git a/server/src/services/remote-workspace-snapshot-archive.test.ts b/server/src/services/remote-workspace-snapshot-archive.test.ts index 8b49be22069..3401d7e1aa7 100644 --- a/server/src/services/remote-workspace-snapshot-archive.test.ts +++ b/server/src/services/remote-workspace-snapshot-archive.test.ts @@ -6,7 +6,7 @@ import os from "node:os"; import path from "node:path"; import { promisify } from "node:util"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { captureRemoteWorkspaceDirectory, readRemoteWorkspaceArchiveDirectories, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import { workspaceArchive, readRemoteWorkspaceArchiveDirectories, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage } from "./remote-workspace-snapshot-archive.js"; const archiveIo = vi.hoisted(() => ({ afterChmod: undefined as ((target: string, mode: number) => void) | undefined, @@ -35,6 +35,9 @@ vi.mock("node:fs/promises", async importOriginal => { } }; }); +const captureRemoteWorkspaceDirectory = (input: Omit[0], "kind">) => workspaceArchive({ ...input, kind: "capture" }); +const restoreRemoteWorkspaceArchive = (input: Omit[0], "kind">) => workspaceArchive({ ...input, kind: "restore" }); + const execute = promisify(execFile); const env = { ...process.env, GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: os.devNull, GIT_AUTHOR_NAME: "Snapshot", GIT_COMMITTER_NAME: "Snapshot", GIT_AUTHOR_EMAIL: "snapshot@example.test", GIT_COMMITTER_EMAIL: "snapshot@example.test" }; diff --git a/server/src/services/remote-workspace-snapshot-archive.ts b/server/src/services/remote-workspace-snapshot-archive.ts index 3bed12712b1..3e729aed92a 100644 --- a/server/src/services/remote-workspace-snapshot-archive.ts +++ b/server/src/services/remote-workspace-snapshot-archive.ts @@ -1,8 +1,10 @@ +import { Worker } from "node:worker_threads"; + /** * A portable, streaming workspace archive. This function is intentionally * self-contained: the same implementation runs locally and in a sandbox. */ -async function workspaceArchive(input: { +export async function workspaceArchive(input: { kind: "capture" | "restore"; cwd?: string; archivePath: string; @@ -662,8 +664,81 @@ async function workspaceArchive(input: { } } +type ArchiveInput = Parameters[0]; +let activeArchiveWorkers = 0; +const archiveWaiters: Array<() => void> = []; + +async function archiveWorkerSlot(signal?: AbortSignal): Promise<() => void> { + signal?.throwIfAborted(); + await new Promise((resolve, reject) => { + const start = () => { + signal?.removeEventListener("abort", abort); + activeArchiveWorkers++; + resolve(); + }; + const abort = () => { + const index = archiveWaiters.indexOf(start); + if (index >= 0) archiveWaiters.splice(index, 1); + reject(signal!.reason); + }; + if (activeArchiveWorkers < 8) start(); + else { + archiveWaiters.push(start); + signal?.addEventListener("abort", abort, { once: true }); + } + }); + return () => { activeArchiveWorkers--; archiveWaiters.shift()?.(); }; +} + +async function runArchiveWorker(input: ArchiveInput): ReturnType { + const release = await archiveWorkerSlot(input.signal); + let workerScratch: string | undefined; + const fs = await import("node:fs/promises"); + try { + input.signal?.throwIfAborted(); + const path = await import("node:path"), os = await import("node:os"); + workerScratch = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-archive-worker-")); + await fs.chmod(workerScratch, 0o700); + const { signal, ...parameters } = input; + const source = workspaceArchive.toString().replaceAll("__vite_ssr_dynamic_import__", "import"); + const worker = new Worker(`const {parentPort,workerData}=require("node:worker_threads"); +const __name=value=>value; +const controller=new AbortController(); +parentPort.on("message",()=>controller.abort()); +const archive=${source}; +archive({...workerData,signal:controller.signal}).then(result=>parentPort.postMessage({result}),error=>parentPort.postMessage({error:{...error,name:error.name,message:error.message}})).finally(()=>parentPort.close());`, { + eval: true, workerData: parameters, execArgv: [], stdout: true, stderr: true, + env: { ...process.env, TMPDIR: workerScratch, TMP: workerScratch, TEMP: workerScratch }, + resourceLimits: { maxOldGenerationSizeMb: 1024, maxYoungGenerationSizeMb: 64, stackSizeMb: 4 }, + }); + worker.stdout.resume(); worker.stderr.resume(); + const abort = () => worker.postMessage("abort"); + signal?.addEventListener("abort", abort, { once: true }); + if (signal?.aborted) abort(); + try { + return await new Promise>>((resolve, reject) => { + let result: Awaited> | undefined, failure: Error | undefined; + worker.once("message", (message: { result?: typeof result; error?: Record }) => { + result = message.result; + if (message.error) failure = Object.assign(new Error(String(message.error.message)), message.error); + }); + worker.once("error", error => { failure = error; }); + worker.once("exit", code => { + if (signal?.aborted) reject(signal.reason); + else if (failure) reject(failure); + else if (code === 0 && result) resolve(result); + else reject(new Error("remote_workspace_archive_worker_failed")); + }); + }); + } finally { signal?.removeEventListener("abort", abort); } + } finally { + try { if (workerScratch) await fs.rm(workerScratch, { recursive: true, force: true }); } + finally { release(); } + } +} + export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePath: string; manifestPath?: string; exclude?: string[]; requireNoGit?: boolean; gitSourceRoots?: string[]; signal?: AbortSignal }) { - return workspaceArchive({ ...input, kind: "capture" }); + return runArchiveWorker({ ...input, kind: "capture" }); } export async function readRemoteWorkspaceArchiveDirectories(archivePath: string, expectedManifestDigest: string) { @@ -691,7 +766,7 @@ export async function readRemoteWorkspaceArchiveDirectories(archivePath: string, } export function restoreRemoteWorkspaceArchive(input: { archivePath: string; targetDir: string; expectedManifestDigest: string; signal?: AbortSignal; maxEntries?: number; maxManifestBytes?: number }) { - return workspaceArchive({ ...input, kind: "restore" }); + return runArchiveWorker({ ...input, kind: "restore" }); } export function remoteWorkspaceCaptureFailureMessage(error: unknown): string | null {