From 1d7e52ca291fa6bfad2e8b022f4a14e1af6d227a Mon Sep 17 00:00:00 2001 From: Christopher Monsanto Date: Mon, 17 Aug 2026 19:06:03 -0400 Subject: [PATCH] Make the build core's file I/O async hash and cas move to node:fs/promises: source reconciliation runs a small worker pool over the stat cache, the executor's clean check gathers CAS stats concurrently, and inserts/sweeps await their way through. The sqlite store stays sync; that is all node:sqlite offers. Co-Authored-By: Claude Fable 5 --- tools/build/cas.ts | 57 ++++++++++++++++------------- tools/build/driver.ts | 4 +-- tools/build/executor.ts | 29 ++++++++------- tools/build/hash.ts | 70 ++++++++++++++++++++---------------- tools/build/test/cas.test.ts | 44 +++++++++++------------ 5 files changed, 114 insertions(+), 90 deletions(-) diff --git a/tools/build/cas.ts b/tools/build/cas.ts index 3829306b..669ddc1a 100644 --- a/tools/build/cas.ts +++ b/tools/build/cas.ts @@ -1,8 +1,8 @@ -import * as fs from 'node:fs'; +import * as fs from 'node:fs/promises'; import * as pathlib from 'node:path'; -import {hashFileSync} from './hash.ts'; +import {hashFile} from './hash.ts'; // Content-addressed store for build outputs. Objects live at // //. (hh = first two hex chars), carry their @@ -15,16 +15,16 @@ export function casPath(casDir: string, digest: string, ext: string): string { return pathlib.join(casDir, digest.slice(0, 2), `${digest}.${ext}`); } -export function casExists(casDir: string, digest: string, ext: string): boolean { - return casStat(casDir, digest, ext) !== null; +export async function casExists(casDir: string, digest: string, ext: string): Promise { + return await casStat(casDir, digest, ext) !== null; } // Size of an object, or null if absent. Callers verify it against the // recorded size: a crash between rename and data flush can leave a // truncated object, which must read as dirty, not clean. -export function casStat(casDir: string, digest: string, ext: string): bigint | null { +export async function casStat(casDir: string, digest: string, ext: string): Promise { try { - let st = fs.statSync(casPath(casDir, digest, ext), {bigint: true}); + let st = await fs.stat(casPath(casDir, digest, ext), {bigint: true}); return st.isFile() ? st.size : null; } catch { return null; @@ -32,42 +32,51 @@ export function casStat(casDir: string, digest: string, ext: string): bigint | n } export type CasObject = { - digest: string; // sha256 hex of the bytes + digest: string, // sha256 hex of the bytes size: bigint, }; +async function exists(path: string): Promise { + try { + await fs.access(path); + return true; + } catch { + return false; + } +} + // Move tmpPath into the store, returning the content digest and size. An // existing object is trusted only if its bytes actually hash to the digest; // otherwise (crash-truncated object) the fresh bytes replace it. -export function casInsert(casDir: string, tmpPath: string, ext: string): CasObject { - let digest = hashFileSync(tmpPath).toString('hex'); - let size = fs.statSync(tmpPath, {bigint: true}).size; +export async function casInsert(casDir: string, tmpPath: string, ext: string): Promise { + let digest = (await hashFile(tmpPath)).toString('hex'); + let size = (await fs.stat(tmpPath, {bigint: true})).size; let target = casPath(casDir, digest, ext); - if (fs.existsSync(target) && hashFileSync(target).toString('hex') === digest) { - fs.unlinkSync(tmpPath); + if (await exists(target) && (await hashFile(target)).toString('hex') === digest) { + await fs.unlink(tmpPath); return {digest, size}; } - fs.mkdirSync(pathlib.dirname(target), {recursive: true}); - fs.chmodSync(tmpPath, 0o444); + await fs.mkdir(pathlib.dirname(target), {recursive: true}); + await fs.chmod(tmpPath, 0o444); // Flush the bytes before the rename becomes visible, so a power loss // cannot journal the rename while dropping the data pages. - let fd = fs.openSync(tmpPath, 'r'); + let fh = await fs.open(tmpPath, 'r'); try { - fs.fsyncSync(fd); + await fh.sync(); } finally { - fs.closeSync(fd); + await fh.close(); } - fs.renameSync(tmpPath, target); + await fs.rename(tmpPath, target); return {digest, size}; } // Remove every object not in `live` (keys are ".", the object // basename) and prune emptied fanout directories. Returns the removal count. -export function casSweep(casDir: string, live: Set): number { +export async function casSweep(casDir: string, live: Set): Promise { let removed = 0; - let fanout: fs.Dirent[]; + let fanout; try { - fanout = fs.readdirSync(casDir, {withFileTypes: true}); + fanout = await fs.readdir(casDir, {withFileTypes: true}); } catch (err) { if ((err as {code?: string}).code === 'ENOENT') { return 0; @@ -79,14 +88,14 @@ export function casSweep(casDir: string, live: Set): number { continue; } let dirPath = pathlib.join(casDir, dir.name); - for (let name of fs.readdirSync(dirPath)) { + for (let name of await fs.readdir(dirPath)) { if (!live.has(name)) { - fs.unlinkSync(pathlib.join(dirPath, name)); + await fs.unlink(pathlib.join(dirPath, name)); removed++; } } try { - fs.rmdirSync(dirPath); + await fs.rmdir(dirPath); } catch { // not empty } diff --git a/tools/build/driver.ts b/tools/build/driver.ts index 0f346ace..b2429c7f 100644 --- a/tools/build/driver.ts +++ b/tools/build/driver.ts @@ -52,7 +52,7 @@ export async function build(decls: readonly RuleDecl[], opts: BuildOpts): Promis } }; decls.forEach(add); - let {hashes, updated, missing} = reconcileHashes(sources, store.loadFileCache()); + let {hashes, updated, missing} = await reconcileHashes(sources, store.loadFileCache()); if (missing.length > 0) { throw new BuildError(`Missing input files:\n ${missing.slice(0, 20).join('\n ')}` + (missing.length > 20 ? `\n ... and ${missing.length - 20} more` : '')); @@ -112,7 +112,7 @@ export async function build(decls: readonly RuleDecl[], opts: BuildOpts): Promis if (opts.gc && !opts.dryRun && !interrupted && result.ok && result.keys.size === decls.length) { let removedRules = store.deleteKeysNotIn(new Set(result.keys.values())); - let removedObjects = casSweep(opts.casDir, store.liveObjects()); + let removedObjects = await casSweep(opts.casDir, store.liveObjects()); store.pruneFileCache(sources); if (removedRules > 0 || removedObjects > 0) { log(`gc: removed ${removedRules} rules, ${removedObjects} objects`); diff --git a/tools/build/executor.ts b/tools/build/executor.ts index 8caca175..9d0e7a22 100644 --- a/tools/build/executor.ts +++ b/tools/build/executor.ts @@ -1,5 +1,5 @@ -import * as fs from 'node:fs'; +import * as fs from 'node:fs/promises'; import * as pathlib from 'node:path'; import * as artifact from './artifact.ts'; @@ -212,10 +212,12 @@ export class Executor { let stored = store.lookupRule(key); if (stored !== null && stored.length === decl.outputs.length - && stored.every((o, n) => o.ext === paired(decl.outputs, n).ext) - && stored.every(o => cas.casStat(casDir, o.digest, o.ext) === o.size)) { - this.#outcomes.set(decl, {status: 'clean'}); - return stored.map(o => o.digest); + && stored.every((o, n) => o.ext === paired(decl.outputs, n).ext)) { + let sizes = await Promise.all(stored.map(o => cas.casStat(casDir, o.digest, o.ext))); + if (stored.every((o, n) => sizes[n] === o.size)) { + this.#outcomes.set(decl, {status: 'clean'}); + return stored.map(o => o.digest); + } } let reason: DirtyReason = stored === null ? 'new' : 'cas-missing'; @@ -239,7 +241,7 @@ export class Executor { async #execute(decl: artifact.RuleDecl, key: string, reason: DirtyReason): Promise { let {store, casDir, tmpDir, root} = this.#opts; let ruleTmp = pathlib.join(tmpDir, String(this.#tmpSeq++)); - fs.mkdirSync(ruleTmp, {recursive: true}); + await fs.mkdir(ruleTmp, {recursive: true}); let tempOutputs = decl.outputs.map(o => pathlib.join(ruleTmp, o.filename)); let concreteInputs = decl.inputs.map( i => typeof i === 'string' ? i : cas.casPath(casDir, i.hash, i.ext)); @@ -251,12 +253,15 @@ export class Executor { throw new Aborted(); // killed by the abort, not a real failure; stays dirty } if (result.code === 0) { - let missing = tempOutputs.filter(p => !fs.existsSync(p)); + let present = await Promise.all(tempOutputs.map( + p => fs.access(p).then(() => true, () => false))); + let missing = tempOutputs.filter((_, n) => !present[n]); if (missing.length === 0) { - let outputs = decl.outputs.map((o, n) => { - let object = cas.casInsert(casDir, paired(tempOutputs, n), o.ext); - return {digest: object.digest, ext: o.ext, size: object.size}; - }); + let outputs = []; + for (let [n, o] of decl.outputs.entries()) { + let object = await cas.casInsert(casDir, paired(tempOutputs, n), o.ext); + outputs.push({digest: object.digest, ext: o.ext, size: object.size}); + } store.recordRule(key, decl.cmds, outputs); this.#outcomes.set(decl, {status: 'ran', reason}); this.#log(`[${++this.#counter}] ${label(decl)}` @@ -271,7 +276,7 @@ export class Executor { } return this.#fail(decl, command, result.output, `exit status ${result.code ?? result.signal}`); } finally { - fs.rmSync(ruleTmp, {recursive: true, force: true}); + await fs.rm(ruleTmp, {recursive: true, force: true}); } } diff --git a/tools/build/hash.ts b/tools/build/hash.ts index b9a4a56c..71f3bd56 100644 --- a/tools/build/hash.ts +++ b/tools/build/hash.ts @@ -1,6 +1,6 @@ -import * as fs from 'node:fs'; import {createHash} from 'node:crypto'; +import * as fs from 'node:fs/promises'; export type FileStat = { size: bigint, @@ -8,44 +8,54 @@ export type FileStat = { hash: Buffer, }; -export function hashFileSync(path: string): Buffer { - return createHash('sha256').update(fs.readFileSync(path)).digest(); +export async function hashFile(path: string): Promise { + return createHash('sha256').update(await fs.readFile(path)).digest(); } export type ReconcileResult = { - hashes: Map; // current content hash for every extant path - updated: Map; // cache entries that changed (to persist) - missing: string[]; // paths that don't exist or aren't files + hashes: Map, // current content hash for every extant path + updated: Map, // cache entries that changed (to persist) + missing: string[], // paths that don't exist or aren't files }; // Content hashes with a stat cache: only files whose (size, mtime_ns) changed // since the recorded cache entry are rehashed. mtime_ns exceeds 2^53, hence -// bigint stats throughout. -export function reconcileHashes(paths: Iterable, cache: Map): ReconcileResult { +// bigint stats throughout. A small worker pool keeps the fs thread pool fed +// without holding thousands of files open at once. +export async function reconcileHashes(paths: Iterable, + cache: Map): Promise { + let queue = [...paths]; let hashes = new Map(); let updated = new Map(); - let missing = []; - for (let path of paths) { - let st; - try { - st = fs.statSync(path, {bigint: true}); - } catch { - missing.push(path); - continue; + let missing: string[] = []; + let work = async () => { + for (;;) { + let path = queue.pop(); + if (path === undefined) { + return; + } + let st; + try { + st = await fs.stat(path, {bigint: true}); + } catch { + missing.push(path); + continue; + } + if (!st.isFile()) { + missing.push(path); + continue; + } + let cached = cache.get(path); + let hash; + if (cached !== undefined && cached.size === st.size && cached.mtimeNs === st.mtimeNs) { + hash = cached.hash; + } else { + hash = await hashFile(path); + updated.set(path, {size: st.size, mtimeNs: st.mtimeNs, hash}); + } + hashes.set(path, hash); } - if (!st.isFile()) { - missing.push(path); - continue; - } - let cached = cache.get(path); - let hash; - if (cached !== undefined && cached.size === st.size && cached.mtimeNs === st.mtimeNs) { - hash = cached.hash; - } else { - hash = hashFileSync(path); - updated.set(path, {size: st.size, mtimeNs: st.mtimeNs, hash}); - } - hashes.set(path, hash); - } + }; + await Promise.all(Array.from({length: 16}, work)); return {hashes, updated, missing}; } diff --git a/tools/build/test/cas.test.ts b/tools/build/test/cas.test.ts index eb294570..07e55aad 100644 --- a/tools/build/test/cas.test.ts +++ b/tools/build/test/cas.test.ts @@ -18,66 +18,66 @@ function stage(root: string, data: string): string { return p; } -test('casInsert stores by content digest, read-only', () => { +test('casInsert stores by content digest, read-only', async () => { let root = makeTmpRoot(); let cas = pathlib.join(root, 'cas'); - let {digest, size} = casInsert(cas, stage(root, 'hello'), 'png'); + let {digest, size} = await casInsert(cas, stage(root, 'hello'), 'png'); assert.equal(digest, createHash('sha256').update('hello').digest('hex')); assert.equal(size, 5n); let obj = casPath(cas, digest, 'png'); assert.equal(fs.readFileSync(obj, 'utf8'), 'hello'); assert.equal(fs.statSync(obj).mode & 0o777, 0o444); - assert.ok(casExists(cas, digest, 'png')); - assert.equal(casStat(cas, digest, 'png'), 5n); - assert.equal(casStat(cas, digest, 'gif'), null); + assert.ok(await casExists(cas, digest, 'png')); + assert.equal(await casStat(cas, digest, 'png'), 5n); + assert.equal(await casStat(cas, digest, 'gif'), null); fs.rmSync(root, {recursive: true, force: true}); }); -test('casInsert dedupes an existing object and discards the temp', () => { +test('casInsert dedupes an existing object and discards the temp', async () => { let root = makeTmpRoot(); let cas = pathlib.join(root, 'cas'); - let d1 = casInsert(cas, stage(root, 'same'), 'png').digest; + let d1 = (await casInsert(cas, stage(root, 'same'), 'png')).digest; let tmp2 = stage(root, 'same'); - let d2 = casInsert(cas, tmp2, 'png').digest; + let d2 = (await casInsert(cas, tmp2, 'png')).digest; assert.equal(d1, d2); assert.ok(!fs.existsSync(tmp2)); // Same bytes under a different extension is a distinct object. - let d3 = casInsert(cas, stage(root, 'same'), 'gif').digest; + let d3 = (await casInsert(cas, stage(root, 'same'), 'gif')).digest; assert.equal(d1, d3); - assert.ok(casExists(cas, d1, 'png')); - assert.ok(casExists(cas, d1, 'gif')); + assert.ok(await casExists(cas, d1, 'png')); + assert.ok(await casExists(cas, d1, 'gif')); fs.rmSync(root, {recursive: true, force: true}); }); -test('casInsert replaces a corrupt object instead of trusting it', () => { +test('casInsert replaces a corrupt object instead of trusting it', async () => { let root = makeTmpRoot(); let cas = pathlib.join(root, 'cas'); - let {digest} = casInsert(cas, stage(root, 'good bytes'), 'png'); + let {digest} = await casInsert(cas, stage(root, 'good bytes'), 'png'); let obj = casPath(cas, digest, 'png'); // Simulate a crash-truncated object under the same digest name. fs.chmodSync(obj, 0o644); fs.truncateSync(obj); - let again = casInsert(cas, stage(root, 'good bytes'), 'png'); + let again = await casInsert(cas, stage(root, 'good bytes'), 'png'); assert.equal(again.digest, digest); assert.equal(fs.readFileSync(obj, 'utf8'), 'good bytes'); assert.equal(fs.statSync(obj).mode & 0o777, 0o444); fs.rmSync(root, {recursive: true, force: true}); }); -test('casSweep removes non-live objects and prunes empty fanout dirs', () => { +test('casSweep removes non-live objects and prunes empty fanout dirs', async () => { let root = makeTmpRoot(); let cas = pathlib.join(root, 'cas'); - let keep = casInsert(cas, stage(root, 'keep'), 'png').digest; - let drop = casInsert(cas, stage(root, 'drop'), 'png').digest; - let removed = casSweep(cas, new Set([`${keep}.png`])); + let keep = (await casInsert(cas, stage(root, 'keep'), 'png')).digest; + let drop = (await casInsert(cas, stage(root, 'drop'), 'png')).digest; + let removed = await casSweep(cas, new Set([`${keep}.png`])); assert.equal(removed, 1); - assert.ok(casExists(cas, keep, 'png')); - assert.ok(!casExists(cas, drop, 'png')); + assert.ok(await casExists(cas, keep, 'png')); + assert.ok(!await casExists(cas, drop, 'png')); assert.ok(!fs.existsSync(pathlib.join(cas, drop.slice(0, 2)))); assert.ok(fs.existsSync(pathlib.join(cas, keep.slice(0, 2)))); fs.rmSync(root, {recursive: true, force: true}); }); -test('casSweep on a missing store is a no-op', () => { - assert.equal(casSweep(pathlib.join(makeTmpRoot(), 'nope'), new Set()), 0); +test('casSweep on a missing store is a no-op', async () => { + assert.equal(await casSweep(pathlib.join(makeTmpRoot(), 'nope'), new Set()), 0); });