diff --git a/doc/sandbox-work-folders.md b/doc/sandbox-work-folders.md index 95f94619ff..e7a524ddd0 100644 --- a/doc/sandbox-work-folders.md +++ b/doc/sandbox-work-folders.md @@ -237,8 +237,9 @@ saving outgoing changes. Imported bytes therefore cannot overwrite a newer shared version by being mistaken for an agent edit; actual subsequent edits still synchronize normally. Failed explicit refreshes retain the same intent. Other providers keep the small-argument transport. Incoming storage responses -are prefetched four at a time and closed if transfer fails. Repository restores -use the same path, then recreate confined repository links. +are prefetched sixteen at a time without consuming queued response bodies, and +closed if transfer fails. The transport keeps its separate bounded batch buffer. +Repository restores use the same path, then recreate confined repository links. Read-only sandbox commands retry transient connection failures and HTTP 502/503/504 responses up to three attempts within one 120-second deadline. diff --git a/server/src/__tests__/work-folder-transfer.test.ts b/server/src/__tests__/work-folder-transfer.test.ts index 8975b1d191..69033c9670 100644 --- a/server/src/__tests__/work-folder-transfer.test.ts +++ b/server/src/__tests__/work-folder-transfer.test.ts @@ -4,7 +4,7 @@ import { prefetchWorkFiles } from "../services/work-folder-transfer.js"; const entry = { path: "file", kind: "file" as const, byteSize: 0, sha256: null, executable: false }; describe("bounded work file prefetch", () => { - it("preserves order and keeps at most four streams open", async () => { + it("preserves order and keeps at most sixteen streams open", async () => { let opened = 0, peak = 0; const seen: string[] = []; for await (const transfer of prefetchWorkFiles(Array.from({ length: 30 }, (_, i) => i), async (i) => { @@ -13,9 +13,28 @@ describe("bounded work file prefetch", () => { return { entry: { ...entry, path: String(i) }, body }; })) seen.push(transfer.entry.path); expect(seen).toEqual(Array.from({ length: 30 }, (_, i) => String(i))); - expect(peak).toBe(4); + expect(peak).toBe(16); expect(opened).toBe(0); }); + it("opens a bounded response window without reading queued bodies", async () => { + const bodies: Readable[] = []; + let reads = 0; + const transfers = prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), async (i) => { + const body = new Readable({ read() { reads++; this.push(Buffer.alloc(1024)); } }); + bodies.push(body); + return { entry: { ...entry, path: String(i), byteSize: 1024 }, body }; + }); + try { + const first = await transfers.next(); + expect(first.value?.entry.path).toBe("0"); + expect(bodies).toHaveLength(16); + expect(reads).toBe(0); + } finally { + await transfers.return(); + } + expect(bodies.every((body) => body.destroyed)).toBe(true); + expect(reads).toBe(0); + }); it("handles a prefetched stream failure before the consumer reaches it", async () => { const source = new Readable({ read() {} }); await expect((async () => { @@ -39,12 +58,12 @@ describe("bounded work file prefetch", () => { const body = Readable.from([]); streams.push(body); return { entry, body }; }; - for await (const _transfer of prefetchWorkFiles([0, 1, 2, 3, 4], open)) break; - expect(streams).toHaveLength(3); + for await (const _transfer of prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), open)) break; + expect(streams).toHaveLength(15); expect(streams.every((stream) => stream.destroyed)).toBe(true); streams.length = 0; await expect((async () => { - for await (const _transfer of prefetchWorkFiles([0, 1, 2, 3, 4], open)) { /* consume */ } + for await (const _transfer of prefetchWorkFiles(Array.from({ length: 40 }, (_, i) => i), open)) { /* consume */ } })()).rejects.toThrow("storage unavailable"); expect(streams.every((stream) => stream.destroyed)).toBe(true); }); diff --git a/server/src/services/sandbox-work-folders.ts b/server/src/services/sandbox-work-folders.ts index 1a1fb0522f..8c312b3e0d 100644 --- a/server/src/services/sandbox-work-folders.ts +++ b/server/src/services/sandbox-work-folders.ts @@ -1,5 +1,5 @@ import { measureSandboxOperation, measureSandboxStream, captureSandboxPerformanceContext } from "./sandbox-performance.js"; -import { prefetchWorkFiles } from "./work-folder-transfer.js"; +import { prefetchWorkFiles, WORK_FOLDER_PREFETCH_CONCURRENCY } from "./work-folder-transfer.js"; import { createHash, randomUUID } from "node:crypto"; import path from "node:path"; import { managedAgentFiles } from "./work-folder-agent-import.js"; @@ -242,7 +242,7 @@ export async function prepareSandboxWorkFolders(input: { } const changed = saved.sort((a, b) => a.path.length - b.path.length) .filter((entry) => signature(current.get(entry.path)) !== signature(entry)); - await measureSandboxOperation("work_folder.scope.hydrate", { scope, files: changed.length, parallelism: 4 }, async () => (transport.writeMany(paths[scope]!, staging, prefetchWorkFiles(changed, async (entry, fileIndex) => { + await measureSandboxOperation("work_folder.scope.hydrate", { scope, files: changed.length, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => (transport.writeMany(paths[scope]!, staging, prefetchWorkFiles(changed, async (entry, fileIndex) => { if (entry.kind === "directory") return { entry }; const result = await svc.content(folder, entry.path, fileIndex); // A shared file can change after listing. Validate and baseline the diff --git a/server/src/services/work-folder-transfer.ts b/server/src/services/work-folder-transfer.ts index e9ce4a2693..ee2cf4a624 100644 --- a/server/src/services/work-folder-transfer.ts +++ b/server/src/services/work-folder-transfer.ts @@ -1,7 +1,11 @@ import { measureSandboxOperation } from "./sandbox-performance.js"; import type { WorkFileTransfer } from "./work-folder-transport.js"; -// Open a few storage responses ahead without buffering their bodies. Drain all +// Bound open responses independently of file size. Bodies stay paused until +// consumed, and the transport retains its separate batch byte limit. +export const WORK_FOLDER_PREFETCH_CONCURRENCY = 16; + +// Open storage responses ahead without buffering their bodies. Drain all // pending opens on failure so abandoned HTTP response streams are closed too. export async function* prefetchWorkFiles( entries: Iterable, @@ -15,7 +19,7 @@ export async function* prefetchWorkFiles( const next = iterator.next(); if (!next.done) { const fileIndex = opened++; - pending.push(Promise.resolve().then(() => measureSandboxOperation("work_folder.prefetch.open", { fileIndex, parallelism: 4 }, async () => open(next.value, fileIndex))) + pending.push(Promise.resolve().then(() => measureSandboxOperation("work_folder.prefetch.open", { fileIndex, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => open(next.value, fileIndex))) .then((value): Result => { // A response can fail while queued, before its async iterator exists. // Keep that error handled; consuming the stream still throws it. @@ -25,16 +29,16 @@ export async function* prefetchWorkFiles( } } try { - for (let i = 0; i < 4; i++) enqueue(); + for (let i = 0; i < WORK_FOLDER_PREFETCH_CONCURRENCY; i++) enqueue(); while (pending.length) { const next = pending.shift()!; - const result = await measureSandboxOperation("work_folder.prefetch.wait", { fileIndex: consumed++, parallelism: 4 }, async () => next); + const result = await measureSandboxOperation("work_folder.prefetch.wait", { fileIndex: consumed++, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => next); if ("error" in result) throw result.error; try { yield result.value; } finally { result.value.body?.destroy(); } enqueue(); } } finally { - for (const result of await measureSandboxOperation("work_folder.prefetch.drain", { files: pending.length, parallelism: 4 }, async () => Promise.all(pending))) { + for (const result of await measureSandboxOperation("work_folder.prefetch.drain", { files: pending.length, parallelism: WORK_FOLDER_PREFETCH_CONCURRENCY }, async () => Promise.all(pending))) { if ("value" in result) result.value.body?.destroy(); } iterator.return?.();