From 9abc54f48cc8da030da4c08be87f898dfd560a88 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 10 Oct 2026 09:31:25 -0500 Subject: [PATCH] fix(runtime): upload large runner binaries in bounded parallel chunks --- .../native-session-executor.test.ts | 38 ++++++ .../native-runtime/native-session-executor.ts | 124 +++++++++++------- 2 files changed, 112 insertions(+), 50 deletions(-) diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 4ff4f34492b..cdc18095718 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -2483,6 +2483,44 @@ describe("remote provider checkpoint restores", () => { } finally { await rm(root, { recursive: true, force: true }); } }); + it.each([false, true])("uploads a binary in bounded chunks and preserves the installed binary on corruption=%s", async (corrupt) => { + const root = await mkdtemp(join(tmpdir(), "paperclip-runner-large-file-")); + const sourcePath = join(root, "source"); + const targetPath = join(root, "remote runner's binary"); + const bytes = randomBytes(9 * 1024 * 1024 + 17); + await writeFile(sourcePath, bytes); + await writeFile(targetPath, "previous verified binary"); + let inFlight = 0; + let maxInFlight = 0; + const execute = vi.fn(async (request: { command: string; args: string[]; stdin?: string }) => { + expect(request.stdin?.length ?? 0).toBeLessThanOrEqual(Math.ceil(4 * 1024 * 1024 / 3) * 4); + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + try { + await new Promise((resolve) => setTimeout(resolve, 5)); + const stdin = corrupt && request.stdin ? Buffer.from("corrupt").toString("base64") : request.stdin; + execFileSync(request.command, request.args, { input: stdin }); + return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; + } finally { inFlight--; } + }); + try { + const stage = stageRemoteRunnerFile({ + target: { kind: "remote", transport: "computer" } as never, + runner: { execute } as never, sourcePath, targetPath, mode: 0o700, + }); + if (corrupt) { + await expect(stage).rejects.toThrow(); + expect(await readFile(targetPath, "utf8")).toBe("previous verified binary"); + } else { + await stage; + expect(await readFile(targetPath)).toEqual(bytes); + expect((await lstat(targetPath)).mode & 0o777).toBe(0o700); + } + expect(maxInFlight).toBe(3); + expect((await readdir(root)).some((name) => name.includes(".upload-"))).toBe(false); + } finally { await rm(root, { recursive: true, force: true }); } + }); + it.each([{ sizeMiB: 0, corrupt: false }, { sizeMiB: 65, corrupt: false }, { sizeMiB: 0, corrupt: true }])("stages a $sizeMiB MiB provider pack through a command-only computer runner (corrupt=$corrupt)", async ({ sizeMiB, corrupt }) => { const root = await mkdtemp(join(tmpdir(), "paperclip-computer-pack-")); const sourcePath = join(root, "source"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index d4e04a90be1..91eb8c7b970 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10273,21 +10273,76 @@ export async function stageRemoteRunnerFile(input: { ]); return; } - const bytes = readFileSync(input.sourcePath); - const directory = posix.dirname(input.targetPath); - const script = - `umask 077; mkdir -p '${directory.replaceAll("'", "'\\''")}' && ` + - `base64 -d > '${input.targetPath.replaceAll("'", "'\\''")}' && ` + - `chmod ${input.mode.toString(8)} '${input.targetPath.replaceAll("'", "'\\''")}'`; - const result = await runner.execute({ - command: "sh", - args: ["-c", script], - stdin: bytes.toString("base64"), - bypassSession: true, - timeoutMs: 180_000, - }); - if (result.exitCode !== 0 || result.timedOut) { - throw new Error(`runner_remote_staging_failed: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`); + const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; + const run = async (script: string, stdin?: string) => { + const result = await runner.execute({ + command: "sh", args: ["-c", script], stdin, + bypassSession: true, timeoutMs: 180_000, + }); + if (result.exitCode !== 0 || result.timedOut) { + throw new Error(`runner_remote_staging_failed: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`); + } + }; + if (lstatSync(input.sourcePath).size <= 4 * 1024 * 1024) { + await run(`umask 077; mkdir -p ${quote(posix.dirname(input.targetPath))} && ` + + `base64 -d > ${quote(input.targetPath)} && chmod ${input.mode.toString(8)} ${quote(input.targetPath)}`, + readFileSync(input.sourcePath).toString("base64")); + return; + } + const remoteUpload = `${input.targetPath}.upload-${randomUUID()}`; + try { + await uploadRemoteRunnerFileChunks({ sourcePath: input.sourcePath, remotePath: remoteUpload, run }); + await run(`chmod ${input.mode.toString(8)} ${quote(remoteUpload)} && mv -f ${quote(remoteUpload)} ${quote(input.targetPath)}`); + } finally { + await run(`rm -f ${quote(remoteUpload)}`).catch(() => undefined); + } +} + +async function uploadRemoteRunnerFileChunks(input: { + sourcePath: string; + remotePath: string; + run: (script: string, stdin?: string) => Promise; + onProgress?: (completedBytes: number, totalBytes: number) => Promise; +}): Promise { + const { run, remotePath } = input; + const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; + const pending: Promise[] = []; + let uploadError: unknown; + try { + await run(`umask 077; mkdir -p ${quote(posix.dirname(remotePath))} && : > ${quote(remotePath)}`); + const digest = createHash("sha256"); + const totalBytes = lstatSync(input.sourcePath).size; + let completedBytes = 0; + let progressAt = Date.now(); + await input.onProgress?.(0, totalBytes); + let chunkIndex = 0; + const stream = createReadStream(input.sourcePath, { highWaterMark: 4 * 1024 * 1024 }); + for await (const bytes of stream) { + const chunk = bytes as Buffer; + digest.update(chunk); + completedBytes += chunk.length; + // Disjoint fixed offsets permit bounded parallel upload and idempotent + // writes. The final digest rejects missing, duplicated, or partial bytes. + pending.push(run(`base64 -d | dd of=${quote(remotePath)} bs=4194304 seek=${chunkIndex++} conv=notrunc 2>/dev/null`, + chunk.toString("base64")).catch((error) => { uploadError ??= error; })); + if (pending.length === 4) { + await Promise.all(pending); + pending.length = 0; + if (uploadError) throw uploadError; + if (Date.now() - progressAt >= 15_000) { + await input.onProgress?.(completedBytes, totalBytes); + progressAt = Date.now(); + } + } + } + await Promise.all(pending); + if (uploadError) throw uploadError; + await input.onProgress?.(totalBytes, totalBytes); + const expectedDigest = digest.digest("hex"); + await run(`test "$(if command -v sha256sum >/dev/null 2>&1; then sha256sum ${quote(remotePath)}; else shasum -a 256 ${quote(remotePath)}; fi | cut -d ' ' -f 1)" = ${quote(expectedDigest)}`); + } finally { + // Do not remove or rename the upload while another offset write is active. + await Promise.all(pending); } } @@ -10369,8 +10424,6 @@ export async function stageRemoteRunnerDirectory(input: { throw new Error(`runner_remote_directory_staging_failed: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`); } }; - const pending: Promise[] = []; - let uploadError: unknown; try { // Provider packs can exceed a gigabyte. Keep compression off the event loop // and transfer from disk in bounded chunks instead of buffering the archive. @@ -10379,41 +10432,12 @@ export async function stageRemoteRunnerDirectory(input: { { maxBuffer: 1024 * 1024, env: { ...process.env, COPYFILE_DISABLE: "1" } }, (error) => error ? reject(error) : resolve()); }); - await run(`umask 077; mkdir -p ${quote(posix.dirname(remoteArchive))} && : > ${quote(remoteArchive)}`); - const digest = createHash("sha256"); - const totalBytes = lstatSync(archivePath).size; - let completedBytes = 0; - let progressAt = Date.now(); - await input.onProgress?.(0, totalBytes); - let chunkIndex = 0; - const stream = createReadStream(archivePath, { highWaterMark: 4 * 1024 * 1024 }); - for await (const bytes of stream) { - const chunk = bytes as Buffer; - digest.update(chunk); - completedBytes += chunk.length; - // Disjoint fixed offsets permit bounded parallel upload and idempotent - // writes. The final digest rejects missing, duplicated, or partial bytes. - pending.push(run(`base64 -d | dd of=${quote(remoteArchive)} bs=4194304 seek=${chunkIndex++} conv=notrunc 2>/dev/null`, - chunk.toString("base64")).catch((error) => { uploadError ??= error; })); - if (pending.length === 4) { - await Promise.all(pending); - pending.length = 0; - if (uploadError) throw uploadError; - if (Date.now() - progressAt >= 15_000) { - await input.onProgress?.(completedBytes, totalBytes); - progressAt = Date.now(); - } - } - } - await Promise.all(pending); - if (uploadError) throw uploadError; - await input.onProgress?.(totalBytes, totalBytes); - const expectedDigest = digest.digest("hex"); - await run(`test "$(if command -v sha256sum >/dev/null 2>&1; then sha256sum ${quote(remoteArchive)}; else shasum -a 256 ${quote(remoteArchive)}; fi | cut -d ' ' -f 1)" = ${quote(expectedDigest)} && ` + - `umask 077 && mkdir -p ${quote(input.targetPath)} && ` + + await uploadRemoteRunnerFileChunks({ + sourcePath: archivePath, remotePath: remoteArchive, run, onProgress: input.onProgress, + }); + await run(`umask 077 && mkdir -p ${quote(input.targetPath)} && ` + `tar -xzf ${quote(remoteArchive)} -C ${quote(input.targetPath)} && chmod ${input.mode.toString(8)} ${quote(input.targetPath)}`); } finally { - await Promise.all(pending); rmSync(stagingRoot, { recursive: true, force: true }); await run(`rm -f ${quote(remoteArchive)}`).catch(() => undefined); }