diff --git a/.changeset/turn-tree-pooled-copy.md b/.changeset/turn-tree-pooled-copy.md new file mode 100644 index 0000000000..2142b8e362 --- /dev/null +++ b/.changeset/turn-tree-pooled-copy.md @@ -0,0 +1,5 @@ +--- +"@schlessera/brain-ui-server": patch +--- + +A strict, voice or autonomous turn's data tree is built faster: admission copies its files and the index seed sixteen at a time off the event loop, requesting a reflink, instead of one file after another, and an aborted admission stops the copy and removes it. On the native fixture this halves the copy (2,000 notes 100 ms to 51 ms, 10,000 notes 471 ms to 245 ms). diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index 6248f3e329..c2c5635912 100644 --- a/docs/decisions/governed-writable-workers.md +++ b/docs/decisions/governed-writable-workers.md @@ -852,6 +852,24 @@ configuration still loads from the frozen view at the brain path. approved index copy. Applications write only non-hidden `.md` outside `context/policies`, and the image-mask operation only its PNG and its scratch prune's removals, so every other byte in the tree equals the view. +- **The copy runs as a pool (#2891).** Admission plans the tree (directories + and links) in one walk, then copies the files and the index seed sixteen at + a time off the event loop, requesting a reflink, and closes each directory's + mode and times last. Measured with the real admission code on a native + Btrfs checkpoint into the server's temporary directory, warm, as UID 1000 + and as root (receipts on #2891, 2026-10-25): 2,000 notes of 16 KiB with a + 115 MiB index 100 ms → 51 ms, 10,000 notes with a 574 MiB index + 471 ms → 245 ms; a cold first copy 283 ms → 59 ms and 1,234 ms → 247 ms. + Two cheaper constructions were measured and not adopted. A reflink + placement on the managed volume takes 43 ms and 202 ms with one file after + another, because Bun's copy reflinks there already, but the server owns no + directory on that volume outside the live brain (the manager requires a + root-owned store, and the live subvolume is what governed peers write and + checkpoints qualify), so it needs a manager-provisioned location and a + capacity ruling first. Hardlinks from a shared base are refused by the + readers that require one link (`checkGovernancePolicy`, the embedding + policy, storage-attachment facts) and cannot cross subvolumes. The tree + still holds one copy per concurrent turn on the temporary directory. - **Committed bytes enter the tree inside the commit boundary.** Each application hands the tree its committed effects with the live base each one replaced, partial staged failures included; an archive writes the diff --git a/docs/hosting/agent-workers.md b/docs/hosting/agent-workers.md index 3cad844eb0..91a16a0254 100644 --- a/docs/hosting/agent-workers.md +++ b/docs/hosting/agent-workers.md @@ -401,8 +401,9 @@ write verified scratch/workspace copies through the aliases below. commands that name the brain path see the turn-start bytes; a Claude `Read` of a path the turn changed is refused and points to `mcp__brain__brain_read`, and Claude's file tools do not read the tree itself. - Admission copies the startup view's documents into it (O(brain): 0.1-0.3 s - for 2,000 notes) and seeds its index, on the server's temporary directory. + Admission copies the startup view's documents into it (O(brain): about + 0.05 s for 2,000 notes and 0.25 s for 10,000 on the native fixture) and + seeds its index, on the server's temporary directory. Under a provider rule the tree leaves out the masked folders, and only changes that pass the provenance rule enter it; the turn's private index is read-only to the agent. diff --git a/packages/ui-server/src/brain/governed-turn.ts b/packages/ui-server/src/brain/governed-turn.ts index ca1c11900b..5358d6dc0c 100644 --- a/packages/ui-server/src/brain/governed-turn.ts +++ b/packages/ui-server/src/brain/governed-turn.ts @@ -287,7 +287,7 @@ export function createGovernedTurns(options: { // this turn's own applications (#2848); a masked one keeps its private // index current from it (#2846). Built from this final checkpoint. if (workerProfile(resolved) !== "governed" || privateIndex) { - dataTree = createTurnDataTree({ brainPath: options.brainPath, view: startupView!, maskPaths: options.maskPaths, mount: workerProfile(resolved) !== "governed", + dataTree = await createTurnDataTree({ brainPath: options.brainPath, view: startupView!, maskPaths: options.maskPaths, mount: workerProfile(resolved) !== "governed", signal, ...(privateIndex ? { index: privateIndex, masks: { folders: maskedFolders, buildFolders: privateIndexBuildFolders(maskedFolders, startupView!.root), types: providerDeniedTypes(policy.policy, recipient) } } : {}), // A recipient allowed every folder still never sees the generated-text caches. diff --git a/packages/ui-server/src/brain/turn-data-tree.ts b/packages/ui-server/src/brain/turn-data-tree.ts index fd1b19d35d..dc46e136dc 100644 --- a/packages/ui-server/src/brain/turn-data-tree.ts +++ b/packages/ui-server/src/brain/turn-data-tree.ts @@ -24,8 +24,9 @@ * the next update drops it from the index too. A governed masked turn reads * the live brain, so its tree feeds only its index updates. */ -import { chmodSync, chownSync, closeSync, constants, copyFileSync, fstatSync, lstatSync, mkdirSync, mkdtempSync, openSync, readdirSync, +import { chmodSync, chownSync, closeSync, constants, fstatSync, lstatSync, mkdirSync, mkdtempSync, openSync, readdirSync, readFileSync, readlinkSync, renameSync, rmSync, symlinkSync, unlinkSync, utimesSync, writeFileSync } from "node:fs"; +import { chmod, copyFile, utimes } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join, normalize, relative } from "node:path"; import { Database, constants as sqlite } from "bun:sqlite"; @@ -90,15 +91,43 @@ const COPY_ENTRIES = 200_000, COPY_DEPTH = 256; const errno = (error: unknown) => (error as NodeJS.ErrnoException)?.code; /** - * Copy the view's documents into `target`: links stay links, timestamps and - * modes are kept, nothing else (sockets, devices) is copied. `skip` names - * repo-relative paths left out. Across UIDs, a file or directory the server - * cannot read (a worker's own `0600` file) is left out too: the worker that - * could read it sees it at the brain path. + * Files copied at once. A copy onto the server's temporary directory crosses + * filesystems and waits on I/O, so a pool takes about half the time of one + * file after another (#2891); a reflink, where the filesystem offers one, is + * requested and taken. */ -function copyView(view: string, target: string, skip: (rel: string) => boolean): void { +const COPY_WIDTH = 16; +const COPY_FLAGS = constants.COPYFILE_EXCL | constants.COPYFILE_FICLONE; + +/** + * Run `jobs` at most `COPY_WIDTH` at a time. The first failure, or an abort, + * stops new jobs; it is thrown once every started job has settled, so nothing + * still writes into a tree its caller is about to remove. + */ +async function pooled(jobs: readonly (() => Promise)[], signal?: AbortSignal): Promise { + let next = 0, failed = false, failure: unknown; + await Promise.all(Array.from({ length: Math.min(COPY_WIDTH, jobs.length) }, async () => { + while (!failed && next < jobs.length) { + const job = jobs[next++]!; + try { signal?.throwIfAborted(); await job(); } catch (error) { if (!failed) { failed = true; failure = error; } } + } + })); + if (failed) throw failure; +} + +/** + * Plan the copy of the view's documents into `target`: links stay links, + * timestamps and modes are kept, nothing else (sockets, devices) is copied. + * `skip` names repo-relative paths left out. Across UIDs, a file or directory + * the server cannot read (a worker's own `0600` file) is left out too: the + * worker that could read it sees it at the brain path. Directories and links + * are made here; the returned jobs copy the files, and `finish` then gives + * each directory its mode and times, deepest first. + */ +function planCopy(view: string, target: string, skip: (rel: string) => boolean): { files: (() => Promise)[]; finish(): void } { let entries = 0; const unreadable = (error: unknown) => errno(error) === "EACCES"; + const files: (() => Promise)[] = [], directories: { to: string; info: import("node:fs").Stats }[] = []; const walk = (rel: string, depth: number) => { if (depth > COPY_DEPTH) throw new Error("The startup view exceeds the data tree's directory depth bound."); let listed: import("node:fs").Dirent[]; @@ -109,28 +138,28 @@ function copyView(view: string, target: string, skip: (rel: string) => boolean): if (++entries > COPY_ENTRIES) throw new Error(`The startup view exceeds the data tree's ${COPY_ENTRIES} entry bound.`); const from = join(view, path), to = join(target, path), info = lstatSync(from); if (info.isSymbolicLink()) symlinkSync(readlinkSync(from), to); - else if (info.isDirectory()) { - mkdirSync(to, { mode: 0o700 }); walk(path, depth + 1); - chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime); - } else if (info.isFile()) { - try { copyFileSync(from, to, constants.COPYFILE_EXCL); } catch (error) { if (unreadable(error)) continue; throw error; } - chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime); - } + else if (info.isDirectory()) { mkdirSync(to, { mode: 0o700 }); directories.push({ to, info }); walk(path, depth + 1); } + else if (info.isFile()) files.push(async () => { + try { await copyFile(from, to, COPY_FLAGS); } catch (error) { if (unreadable(error)) return; throw error; } + await chmod(to, info.mode & 0o7777); await utimes(to, info.atime, info.mtime); + }); } }; walk("", 0); + // A directory is planned before everything in it, so the reverse order + // closes each one after its contents. + return { files, finish() { for (const { to, info } of directories.reverse()) { chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime); } } }; } -/** Seed `index` with the view's index files. The updater runs as the worker identity, which shares the server's group. */ -function seedIndex(view: string, index: string, separateIdentity: boolean): void { +/** Jobs that seed `index` with the view's index files. The updater runs as the worker identity, which shares the server's group. */ +function seedIndex(view: string, index: string, separateIdentity: boolean): (() => Promise)[] { const source = join(view, INDEX_DIR); const names = (() => { try { return readdirSync(source, { withFileTypes: true }); } catch (error) { if (errno(error) === "ENOENT") return []; throw error; } })(); - for (const entry of names) { - if (!entry.isFile()) continue; + return names.filter(entry => entry.isFile()).map(entry => async () => { const to = join(index, entry.name); - copyFileSync(join(source, entry.name), to, constants.COPYFILE_EXCL); - chmodSync(to, separateIdentity ? 0o660 : 0o600); - } + await copyFile(join(source, entry.name), to, COPY_FLAGS); + await chmod(to, separateIdentity ? 0o660 : 0o600); + }); } /** @@ -278,9 +307,11 @@ async function runOnTree(options: { * Copy the view into a new tree, before the worker launches. `index` is the * masked turn's private index, already built; an unmasked turn's is seeded * from the view. `pending` names paths the view's index may lag its documents - * on (`IndexBatch.watermark`): the first index read updates them. + * on (`IndexBatch.watermark`): the first index read updates them. The files + * and the index seed are copied together, off the event loop; an abort stops + * the copy and removes what it made. */ -export function createTurnDataTree(options: { +export async function createTurnDataTree(options: { brainPath: string; view: WorkerStartupView; maskPaths: readonly string[]; /** Mount it for the worker: false for a governed turn, which reads the live brain. */ mount: boolean; @@ -295,7 +326,9 @@ export function createTurnDataTree(options: { update?: (tree: WorkerDataTree, paths: readonly string[]) => Promise; /** Default: the maintenance program's capture planning, on the tree. */ plan?: (tree: WorkerDataTree, input: Record, signal: AbortSignal) => Promise; -}): TurnDataTree { + /** The admission's: stops the copy. */ + signal?: AbortSignal; +}): Promise { const separateIdentity = Boolean(workerIdentityHelper()); const root = mkdtempSync(join(tmpdir(), "brain-turn-data-")); let index: string | undefined; @@ -307,14 +340,16 @@ export function createTurnDataTree(options: { // The `.env*` masks cover link targets at the brain path only, so the tree // leaves those targets out as well. const hidden = [...(masks?.buildFolders ?? []), ...envMaskedPaths(options.view.root, options.brainPath)]; - copyView(options.view.root, root, rel => { + const copy = planCopy(options.view.root, root, rel => { const top = !rel.includes("/"); if (top && (OMITTED.has(rel) || isEnvName(rel) || (caches && PROVIDER_CACHE_FILES.includes(rel)) || (masks && rel === PROVIDER_CACHE_DIR))) return true; return hidden.some(folder => rel === folder || rel.startsWith(`${folder}/`)); }); mkdirSync(join(root, INDEX_DIR), { mode: 0o755 }); index = options.index ?? createPrivateIndexDirectory(); - if (!options.index) seedIndex(options.view.root, index, separateIdentity); + // The index files are the largest; they start first. + await pooled([...options.index ? [] : seedIndex(options.view.root, index, separateIdentity), ...copy.files], options.signal); + copy.finish(); settleIndex(index, separateIdentity); } catch (error) { try { removeCopy(root); } finally { if (index && !options.index) removePrivateIndexDirectory(index); } diff --git a/packages/ui-server/src/inbox/autonomous-turn.ts b/packages/ui-server/src/inbox/autonomous-turn.ts index 5d64adb9a5..967713a19a 100644 --- a/packages/ui-server/src/inbox/autonomous-turn.ts +++ b/packages/ui-server/src/inbox/autonomous-turn.ts @@ -394,7 +394,7 @@ async function executeAutonomousTurn( controller.signal.throwIfAborted(); workerProviderMasks = Object.freeze({ folders: Object.freeze([...maskedFolders]), index: privateIndex }); } - dataTree = createTurnDataTree({ brainPath, view: startupView, maskPaths: deps.privateIndex?.maskPaths ?? [], mount: true, + dataTree = await createTurnDataTree({ brainPath, view: startupView, maskPaths: deps.privateIndex?.maskPaths ?? [], mount: true, signal: controller.signal, ...(privateIndex ? { index: privateIndex, masks: { folders: maskedFolders, buildFolders: privateIndexBuildFolders(maskedFolders, startupView.root), types: providerDeniedTypes(policy.policy, undefined) } } : {}), pending: watermark?.paths() ?? [] }); diff --git a/packages/ui-server/tests/private-index-stdio-release.test.ts b/packages/ui-server/tests/private-index-stdio-release.test.ts index 2a38003424..b69fd0396c 100644 --- a/packages/ui-server/tests/private-index-stdio-release.test.ts +++ b/packages/ui-server/tests/private-index-stdio-release.test.ts @@ -105,7 +105,7 @@ for (const mode of ["success", "failure"] as const) { const f = fixture(); const answer = mode === "success" ? { ok: true, plan: "Ithaca" } : { ok: false, error: "Penelope" }; const seen = observeLaunch({ pool: "input", script: `IFS= read -r request; sleep 0.3; printf '%s\\n' '${JSON.stringify(answer)}'` }); - const tree = createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }); + const tree = await createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }); cleanup.push(() => tree.close()); const result = await outcome(tree.planAdd({ title: "Ithaca" }, new AbortController().signal)); const open = seen.open(); @@ -124,7 +124,7 @@ for (const path of ["masked CLI", "turn tree"] as const) for (const abort of [fa let child: Launched | undefined, settled = false; observeLaunch({ script: path === "masked CLI" ? "printf Ithaca" : 'IFS= read -r request; printf \'{"ok":true,"plan":"Ithaca"}\\n\'', withheld: { promise: release.promise, called: calling.resolve }, launched: launched => { child = launched; } }); - const tree = path === "turn tree" ? createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }) : undefined; + const tree = path === "turn tree" ? await createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }) : undefined; if (tree) cleanup.push(() => tree.close()); const work = outcome(tree ? tree.planAdd({}, controller.signal) : runMaskedBrainCli({ ...f, signal: controller.signal, command: [] })).then(result => { settled = true; return result; }); diff --git a/packages/ui-server/tests/turn-data-tree.test.ts b/packages/ui-server/tests/turn-data-tree.test.ts index 9eaf00a675..83422716b6 100644 --- a/packages/ui-server/tests/turn-data-tree.test.ts +++ b/packages/ui-server/tests/turn-data-tree.test.ts @@ -5,7 +5,7 @@ * backend is a script that reads through the worker the turn admitted. */ import { afterEach, expect, spyOn, test, type Mock } from "bun:test"; -import { chmodSync, cpSync, existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; +import { chmodSync, cpSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; import { createHash } from "node:crypto"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; @@ -480,7 +480,7 @@ test("the tree leaves out .env link targets and, under any provider rule, the le try { // A recipient allowed every provider folder: no folder is masked, and // neither is `.brain-cache/`; a legacy root-level cache still is (#2725). - const tree = createTurnDataTree({ brainPath: f.brain, view, maskPaths: [], mount: true, providerRules: true, update: async () => {} }); + const tree = await createTurnDataTree({ brainPath: f.brain, view, maskPaths: [], mount: true, providerRules: true, update: async () => {} }); const root = tree.mount!.root; expect(["settings/ithaca-credentials.txt", ".env", ".context-cache.jsonl"].filter(rel => existsSync(join(root, rel))), "TREE_OMITS_ENV_TARGETS_AND_LEGACY_CACHES").toEqual([]); expect(existsSync(join(root, ".brain-cache/contexts.jsonl")), "ALLOWED_TREE_KEEPS_THE_CACHE_DIRECTORY").toBe(true); @@ -490,3 +490,65 @@ test("the tree leaves out .env link targets and, under any provider rule, the le expect([existsSync(root), existsSync(tree.index)], "TREE_AND_INDEX_REMOVED_DESPITE_READ_ONLY_DIRECTORIES").toEqual([false, false]); } finally { await view.cleanup(); } }); + +/** Every entry under `root` but the index: kind, mode, mtime, and bytes or link target. */ +function inventory(root: string, rel = ""): Record { + const listed: Record = {}; + for (const name of readdirSync(join(root, rel)).sort()) { + const path = rel ? `${rel}/${name}` : name; + if (path === ".brain-index") continue; + const info = lstatSync(join(root, path)); + const kind = info.isSymbolicLink() ? "link" : info.isDirectory() ? "dir" : "file"; + // A link is made anew: its target is what the copy keeps. + listed[path] = { kind, ...(kind === "link" ? { target: readlinkSync(join(root, path)) } : { mode: (info.mode & 0o7777).toString(8), mtime: Math.floor(info.mtimeMs) }), + ...(kind === "link" ? {} : kind === "file" ? { sha: createHash("sha256").update(readFileSync(join(root, path))).digest("hex") } : {}) }; + if (kind === "dir") Object.assign(listed, inventory(root, path)); + } + return listed; +} + +test("the pooled copy is the view, entry for entry: bytes, links, modes and times, and the seeded index", async () => { + const f = fixture(false); + // More files than the pool is wide, nested, with read-only directories and modes to keep. + for (let i = 0; i < 70; i++) { + const dir = join(f.brain, `voyages/leg-${i % 7}`); + mkdirSync(dir, { recursive: true }); + writeFileSync(join(dir, `log-${i}.md`), NOTE(`Log ${i}`, `Odysseus logs leg ${i}.`)); + utimesSync(join(dir, `log-${i}.md`), new Date(Date.UTC(2026, 6, 12, 0, i)), new Date(Date.UTC(2026, 6, 12, 1, i))); + } + chmodSync(join(f.brain, "voyages/leg-3/log-3.md"), 0o600); + chmodSync(join(f.brain, "voyages/leg-4/log-4.md"), 0o755); + // Bits a creating process's umask would strip. + chmodSync(join(f.brain, "voyages/leg-5/log-5.md"), 0o664); + symlinkSync("../leg-1/log-1.md", join(f.brain, "voyages/leg-2/sister.md")); + utimesSync(join(f.brain, "voyages/leg-5"), new Date(Date.UTC(2026, 6, 1)), new Date(Date.UTC(2026, 6, 2))); + chmodSync(join(f.brain, "voyages/leg-6"), 0o555); + cleanup.push(() => chmodSync(join(f.brain, "voyages/leg-6"), 0o755)); + const view = await startup.prepareNativeWorkerStartupView(f.brain, undefined as never); + try { + const tree = await createTurnDataTree({ brainPath: f.brain, view, maskPaths: [], mount: true, update: async () => {} }); + try { + const omitted = new Set([".git", "node_modules"]); + const expected = Object.fromEntries(Object.entries(inventory(view.root)).filter(([path]) => !omitted.has(path.split("/")[0]!))); + expect(inventory(tree.mount!.root), "THE_TREE_IS_THE_VIEW_ENTRY_FOR_ENTRY").toEqual(expected); + expect(existsSync(join(tree.index, "brain.db")), "THE_SEED_COPIED_THE_INDEX").toBe(true); + expect(snapshot(tree.index), "THE_SEEDED_INDEX_IS_THE_VIEWS").toEqual(snapshot(join(view.root, ".brain-index"))); + } finally { await tree.close(); } + } finally { await view.cleanup(); } +}); + +test("an aborted admission stops the copy and leaves no tree or index behind", async () => { + const f = fixture(false); + for (let i = 0; i < 400; i++) writeFileSync(join(f.brain, "notes", `oar-${i}.md`), NOTE(`Oar ${i}`, "The crew rows on.")); + const view = await startup.prepareNativeWorkerStartupView(f.brain, undefined as never); + const left = () => readdirSync(tmpdir()).filter(name => name.startsWith("brain-turn-data-") || name.startsWith("brain-private-index-")); + try { + const before = left(); + const controller = new AbortController(); + const pending = createTurnDataTree({ brainPath: f.brain, view, maskPaths: [], mount: true, update: async () => {}, signal: controller.signal }); + controller.abort(new Error("Odysseus cancels the voyage.")); + const outcome = await Promise.resolve(pending).then(tree => { void tree.close(); return "created"; }, (error: Error) => error.message); + expect(outcome, "AN_ABORTED_COPY_REFUSES").toBe("Odysseus cancels the voyage."); + expect(left(), "AN_ABORTED_COPY_LEAVES_NOTHING").toEqual(before); + } finally { await view.cleanup(); } +}); diff --git a/scripts/native-proof/hosted-app-child.ts b/scripts/native-proof/hosted-app-child.ts index e5b93355ad..220ac383d9 100644 --- a/scripts/native-proof/hosted-app-child.ts +++ b/scripts/native-proof/hosted-app-child.ts @@ -37,8 +37,8 @@ if (APPLY_TIMING) { // #2848: how long admission takes to copy the turn's data tree and seed its index. const trees = await import(join(source!, "packages/ui-server/src/brain/turn-data-tree.ts")); const createTurnDataTree = trees.createTurnDataTree; - spyOn(trees, "createTurnDataTree").mockImplementation((...args: Parameters) => { - const start = performance.now(); const tree = createTurnDataTree(...args); + spyOn(trees, "createTurnDataTree").mockImplementation(async (...args: Parameters) => { + const start = performance.now(); const tree = await createTurnDataTree(...args); event("turn_tree_seed", { ms: performance.now() - start, masked: Boolean((args[0] as { masks?: unknown }).masks), mounted: Boolean(tree.mount) }); return tree; });