From 6ba5c04b5e2f0dc03ea431a77a28df402ec70b47 Mon Sep 17 00:00:00 2001 From: Christopher Monsanto Date: Mon, 17 Aug 2026 00:23:28 -0400 Subject: [PATCH] Add the demand-driven executor and build driver Keys resolve after producers finish, so there is no upfront plan; identical keys share one execution, failures block dependents, and GC (union builds only) drops undeclared keys then sweeps the CAS. Co-Authored-By: Claude Fable 5 --- tools/build/artifact.ts | 3 + tools/build/driver.ts | 113 ++++++++++++ tools/build/executor.ts | 275 ++++++++++++++++++++++++++++++ tools/build/test/executor.test.ts | 240 ++++++++++++++++++++++++++ 4 files changed, 631 insertions(+) create mode 100644 tools/build/driver.ts create mode 100644 tools/build/executor.ts create mode 100644 tools/build/test/executor.test.ts diff --git a/tools/build/artifact.ts b/tools/build/artifact.ts index abe3cb10..628afb76 100644 --- a/tools/build/artifact.ts +++ b/tools/build/artifact.ts @@ -166,6 +166,9 @@ function makeDecl(inputs : Input[], deps : Input[], spec : CmdSpec, outputs : st if (out.includes('/') || out.includes('%')) { throw new Error(`Rule outputs are nominal filenames, no paths or substitutions: ${out}`); } + if (outputs.indexOf(out) !== index) { + throw new Error(`Duplicate rule output: ${out}`); + } const ext = pathlib.extname(out); if (ext === '' || ext === '.') { throw new Error(`Rule output needs an extension: ${out}`); diff --git a/tools/build/driver.ts b/tools/build/driver.ts new file mode 100644 index 00000000..c386de40 --- /dev/null +++ b/tools/build/driver.ts @@ -0,0 +1,113 @@ + +import {type RuleDecl} from './artifact.ts'; +import {casSweep} from './cas.ts'; +import {BuildError} from './errors.ts'; +import {type BuildResult, Executor, label} from './executor.ts'; +import {reconcileHashes} from './hash.ts'; +import {type Store} from './store.ts'; + +export interface BuildOpts { + root : string; + store : Store; + casDir : string; // relative to root; substituted into commands + tmpDir : string; + jobs : number; + dryRun : boolean; + failFast : boolean; + verbose : boolean; + // GC after a fully successful build: drop rules whose key is no longer + // declared, sweep unreferenced CAS objects, prune the file cache. Only + // safe when `decls` is the full rule universe (a partial build would GC + // the other deploys' state), so the CLI sets it for union builds only. + gc : boolean; + signal : AbortSignal; + log? : (line : string) => void; + logError? : (line : string) => void; +} + +export interface DriveResult extends BuildResult { + interrupted : boolean; +} + +export async function build(decls : readonly RuleDecl[], opts : BuildOpts) : Promise { + const log = opts.log ?? console.log; + const logError = opts.logError ?? console.error; + const {store} = opts; + + const sources = new Set(); + for (const decl of decls) { + for (const input of [...decl.inputs, ...decl.deps]) { + if (typeof input === 'string') { + sources.add(input); + } + } + } + const {hashes, updated, missing} = 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` : '')); + } + if (!opts.dryRun && updated.size > 0) { + store.saveFileCache(updated); + } + + const executor = new Executor({ + root: opts.root, + store, + casDir: opts.casDir, + tmpDir: opts.tmpDir, + jobs: opts.jobs, + dryRun: opts.dryRun, + failFast: opts.failFast, + verbose: opts.verbose, + sourceHashes: hashes, + signal: opts.signal, + log, + logError, + }); + const result = await executor.build(decls); + const interrupted = opts.signal.aborted; + + const counts = {clean: 0, ran: 0, 'would-run': 0, failed: 0, blocked: 0}; + for (const decl of decls) { + const outcome = result.outcomes.get(decl); + if (outcome !== undefined) { + counts[outcome.status]++; + } + } + if (opts.dryRun) { + log(`would run ${counts['would-run']} (${counts.clean} up to date)`); + } else { + const parts = [`${counts.clean} up to date`]; + if (counts.ran > 0) { + parts.push(`${counts.ran} ran`); + } + if (counts.failed > 0) { + parts.push(`${counts.failed} FAILED`); + } + if (counts.blocked > 0) { + parts.push(`${counts.blocked} blocked`); + } + log(parts.join(', ') + '.'); + if (counts.failed > 0) { + logError('Failed rules:'); + for (const decl of decls) { + if (result.outcomes.get(decl)?.status === 'failed') { + logError(` ${label(decl)}`); + } + } + } + } + + if (opts.gc && !opts.dryRun && !interrupted + && result.ok && result.keys.size === decls.length) { + const removedRules = store.deleteKeysNotIn(new Set(result.keys.values())); + const removedObjects = casSweep(opts.casDir, store.liveObjects()); + store.pruneFileCache(sources); + if (removedRules > 0 || removedObjects > 0) { + log(`gc: removed ${removedRules} rules, ${removedObjects} objects`); + } + } + + return {...result, interrupted}; +} diff --git a/tools/build/executor.ts b/tools/build/executor.ts new file mode 100644 index 00000000..386c9e77 --- /dev/null +++ b/tools/build/executor.ts @@ -0,0 +1,275 @@ + +import fs from 'fs'; +import pathlib from 'path'; + +import {type Input, type RuleDecl, computeKey} from './artifact.ts'; +import {casExists, casInsert, casPath} from './cas.ts'; +import {BuildError} from './errors.ts'; +import {runShell} from './exec.ts'; +import {type Store} from './store.ts'; +import {substitute} from './subst.ts'; + +export type DirtyReason = 'new' | 'cas-missing'; + +export type RuleOutcome = + | {status : 'clean'} // key hit, CAS objects present + | {status : 'ran', reason : DirtyReason} + | {status : 'would-run', reason : DirtyReason | 'blocked'} // dry run + | {status : 'failed', message : string} + | {status : 'blocked'}; // a producer failed + +export interface BuildResult { + outcomes : Map; // no entry = not attempted (aborted) + keys : Map; // only decls whose key resolved + ok : boolean; // every decl clean or ran +} + +export interface ExecutorOpts { + root : string; // cwd for commands + store : Store; + casDir : string; // relative to root (substituted into commands) + tmpDir : string; + jobs : number; + dryRun : boolean; + failFast : boolean; + verbose? : boolean; // annotate run lines with the dirty reason + sourceHashes : Map; // every source input/dep, pre-reconciled + signal : AbortSignal; + log? : (line : string) => void; + logError? : (line : string) => void; +} + +export function label(decl : RuleDecl) : string { + return decl.display ?? decl.cmds[0]!; +} + +function indent(text : string) : string { + return text.replace(/\n$/, '').split('\n').map(l => ' ' + l).join('\n'); +} + +// Sentinels thrown through the demand graph. They carry no message; the +// outcome map is the report. +class RuleFailed extends Error {} +class DryDirty extends Error {} +class Aborted extends Error {} + +class Semaphore { + private available : number; + private waiters : (() => void)[] = []; + + constructor(n : number) { + this.available = n; + } + + async acquire() : Promise { + if (this.available > 0) { + this.available--; + return; + } + await new Promise(resolve => this.waiters.push(resolve)); + } + + release() : void { + const waiter = this.waiters.shift(); + if (waiter !== undefined) { + waiter(); + } else { + this.available++; + } + } +} + +// Demand-driven memoized executor. Each rule's identity key is computable +// only once its artifact inputs have digests, so there is no upfront plan: +// demanding a rule awaits its producers, computes the key, and skips or runs. +// The artifact graph is a DAG by construction (a rule can only reference +// artifacts that already exist as values), so there is no cycle check. +export class Executor { + private opts : ExecutorOpts; + private memo = new Map>(); + private inflightByKey = new Map>(); + private outcomes = new Map(); + private keys = new Map(); + private semaphore : Semaphore; + private failAc = new AbortController(); + private runSignal : AbortSignal; + private counter = 0; + private tmpSeq = 0; + private log : (line : string) => void; + private logError : (line : string) => void; + + constructor(opts : ExecutorOpts) { + this.opts = opts; + this.semaphore = new Semaphore(opts.jobs); + this.runSignal = AbortSignal.any([opts.signal, this.failAc.signal]); + this.log = opts.log ?? console.log; + this.logError = opts.logError ?? console.error; + } + + async build(decls : readonly RuleDecl[]) : Promise { + await Promise.allSettled(decls.map(d => this.demand(d))); + const ok = decls.every(d => { + const status = this.outcomes.get(d)?.status; + return status === 'clean' || status === 'ran'; + }); + return {outcomes: this.outcomes, keys: this.keys, ok}; + } + + private demand(decl : RuleDecl) : Promise { + let p = this.memo.get(decl); + if (p === undefined) { + p = this.demandInner(decl); + this.memo.set(decl, p); + } + return p; + } + + private async digestOf(i : Input) : Promise { + if (typeof i !== 'string') { + await this.demand(i.decl); + return i.hash; + } + const hash = this.opts.sourceHashes.get(i); + if (hash === undefined) { + throw new BuildError(`No hash for source ${i}`); + } + return hash.toString('hex'); + } + + private async demandInner(decl : RuleDecl) : Promise { + let digests : Map; + try { + const inputs = [...decl.inputs, ...decl.deps]; + const resolved = await Promise.all(inputs.map(i => this.digestOf(i))); + digests = new Map(inputs.map((i, n) => [i, resolved[n]!])); + } catch (err) { + if (err instanceof RuleFailed) { + this.outcomes.set(decl, {status: 'blocked'}); + } else if (err instanceof DryDirty) { + this.outcomes.set(decl, {status: 'would-run', reason: 'blocked'}); + this.log(`would run (blocked by dirty producer): ${label(decl)}`); + } + throw err; + } + + const key = computeKey(decl, i => digests.get(i)!); + this.keys.set(decl, key); + + // Byte-identical duplicate declarations share one execution. + const existing = this.inflightByKey.get(key); + if (existing !== undefined) { + try { + const shared = await existing; + decl.outputs.forEach((o, n) => o.resolve(shared[n]!)); + this.outcomes.set(decl, {status: 'clean'}); + return shared; + } catch (err) { + if (err instanceof RuleFailed) { + this.outcomes.set(decl, {status: 'blocked'}); + } else if (err instanceof DryDirty) { + this.outcomes.set(decl, {status: 'would-run', reason: 'blocked'}); + } + throw err; + } + } + const work = this.perform(decl, key); + this.inflightByKey.set(key, work); + const result = await work; + decl.outputs.forEach((o, n) => o.resolve(result[n]!)); + return result; + } + + private async perform(decl : RuleDecl, key : string) : Promise { + const {store, casDir} = this.opts; + const stored = store.lookupRule(key); + if (stored !== null + && stored.length === decl.outputs.length + && stored.every((o, n) => o.ext === decl.outputs[n]!.ext) + && stored.every(o => casExists(casDir, o.digest, o.ext))) { + this.outcomes.set(decl, {status: 'clean'}); + return stored.map(o => o.digest); + } + const reason : DirtyReason = stored === null ? 'new' : 'cas-missing'; + + if (this.opts.dryRun) { + this.outcomes.set(decl, {status: 'would-run', reason}); + this.log(`would run (${reason}): ${label(decl)}`); + throw new DryDirty(); + } + + await this.semaphore.acquire(); + try { + if (this.runSignal.aborted) { + throw new Aborted(); + } + return await this.execute(decl, key, reason); + } catch (err) { + if (err instanceof Aborted || err instanceof RuleFailed) { + throw err; + } + // Unexpected (infrastructure) error: count the rule failed and + // stop scheduling; something systemic is wrong. + this.outcomes.set(decl, {status: 'failed', message: 'internal error'}); + this.logError(`FAILED (internal error): ${label(decl)}`); + this.logError(indent(err instanceof Error ? err.stack ?? err.message : String(err))); + this.failAc.abort(); + throw new RuleFailed(); + } finally { + this.semaphore.release(); + } + } + + private async execute(decl : RuleDecl, key : string, reason : DirtyReason) : Promise { + const {store, casDir, tmpDir, root} = this.opts; + const ruleTmp = pathlib.join(tmpDir, String(this.tmpSeq++)); + fs.mkdirSync(ruleTmp, {recursive: true}); + const tempOutputs = decl.outputs.map(o => pathlib.join(ruleTmp, o.filename)); + const concreteInputs = decl.inputs.map( + i => typeof i === 'string' ? i : casPath(casDir, i.hash, i.ext)); + const command = decl.cmds.map(c => substitute(c, concreteInputs, tempOutputs)).join(' && '); + + try { + const result = await runShell(command, {cwd: root, signal: this.runSignal}); + if (this.runSignal.aborted && result.code !== 0) { + throw new Aborted(); // killed by the abort, not a real failure; stays dirty + } + if (result.code === 0) { + const missing = tempOutputs.filter(p => !fs.existsSync(p)); + if (missing.length === 0) { + const outputs = decl.outputs.map((o, n) => ({ + digest: casInsert(casDir, tempOutputs[n]!, o.ext), + ext: o.ext, + })); + store.recordRule(key, decl.cmds, outputs); + this.outcomes.set(decl, {status: 'ran', reason}); + this.log(`[${++this.counter}] ${label(decl)}` + + (this.opts.verbose ? ` (${reason})` : '')); + if (result.output !== '') { + this.log(indent(result.output)); + } + return outputs.map(o => o.digest); + } + return this.fail(decl, command, result.output, + `command succeeded but did not produce: ${missing.map(p => pathlib.basename(p)).join(' ')}`); + } + return this.fail(decl, command, result.output, `exit status ${result.code ?? result.signal}`); + } finally { + fs.rmSync(ruleTmp, {recursive: true, force: true}); + } + } + + // Nothing is recorded for a failure: failed and never-ran are the same + // state, so the rule stays dirty. + private fail(decl : RuleDecl, command : string, output : string, message : string) : never { + this.outcomes.set(decl, {status: 'failed', message}); + this.logError(`FAILED: ${label(decl)} (${message})`); + this.logError(` command: ${command}`); + if (output !== '') { + this.logError(indent(output)); + } + if (this.opts.failFast) { + this.failAc.abort(); + } + throw new RuleFailed(); + } +} diff --git a/tools/build/test/executor.test.ts b/tools/build/test/executor.test.ts new file mode 100644 index 00000000..63294fe2 --- /dev/null +++ b/tools/build/test/executor.test.ts @@ -0,0 +1,240 @@ + +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import pathlib from 'node:path'; +import {beforeEach, test} from 'node:test'; + +import {type Artifact, type CmdSpec, getDecls, resetDecls, rule} from '../artifact.ts'; +import {casPath} from '../cas.ts'; +import {type DriveResult, build} from '../driver.ts'; +import {type RuleOutcome} from '../executor.ts'; +import {Store} from '../store.ts'; + +beforeEach(resetDecls); + +interface Env { + root : string; + dbPath : string; + casDir : string; + tmpDir : string; + execLog : string; +} + +function setup() : Env { + const root = fs.mkdtempSync(pathlib.join(os.tmpdir(), 'executor-test-')); + fs.mkdirSync(pathlib.join(root, 'src')); + return { + root, + dbPath: pathlib.join(root, '.build/db.sqlite'), + casDir: pathlib.join(root, '.build/cas'), + tmpDir: pathlib.join(root, '.build/tmp'), + execLog: pathlib.join(root, 'exec.log'), + }; +} + +function src(env : Env, name : string, content : string) : string { + const p = pathlib.join(env.root, 'src', name); + fs.writeFileSync(p, content); + return p; +} + +// A copy rule that also counts its executions in env.execLog. +function copyRule(env : Env, input : string | Artifact, out : string, + extra : Partial = {}) : Artifact { + return rule(input, {cmds: [`cat %f > %o && echo x >> ${env.execLog}`], ...extra}, [out])[0]!; +} + +function execCount(env : Env) : number { + try { + return fs.readFileSync(env.execLog, 'utf8').split('\n').filter(l => l !== '').length; + } catch { + return 0; + } +} + +async function runBuild(env : Env, opts : {dryRun? : boolean, gc? : boolean} = {}) : Promise { + const store = new Store(env.dbPath); + try { + return await build(getDecls(), { + root: env.root, + store, + casDir: env.casDir, + tmpDir: env.tmpDir, + jobs: 2, + dryRun: opts.dryRun ?? false, + failFast: false, + verbose: false, + gc: opts.gc ?? false, + signal: new AbortController().signal, + log: () => {}, + logError: () => {}, + }); + } finally { + store.close(); + } +} + +function statuses(result : DriveResult) : RuleOutcome['status'][] { + return getDecls().map(d => result.outcomes.get(d)!.status); +} + +test('second build is clean and executes nothing', async () => { + const env = setup(); + const out = copyRule(env, src(env, 'a.txt', 'hello'), 'out.txt'); + const first = await runBuild(env); + assert.ok(first.ok); + assert.deepEqual(statuses(first), ['ran']); + assert.equal(fs.readFileSync(casPath(env.casDir, out.hash, 'txt'), 'utf8'), 'hello'); + + resetDecls(); + copyRule(env, pathlib.join(env.root, 'src/a.txt'), 'out.txt'); + const second = await runBuild(env); + assert.deepEqual(statuses(second), ['clean']); + assert.equal(execCount(env), 1); +}); + +test('input content change reruns; gc removes the old rule and object', async () => { + const env = setup(); + const a = src(env, 'a.txt', 'v1'); + const out1 = copyRule(env, a, 'out.txt'); + await runBuild(env, {gc: true}); + const oldObject = casPath(env.casDir, out1.hash, 'txt'); + + resetDecls(); + fs.writeFileSync(a, 'v2'); + copyRule(env, a, 'out.txt'); + const result = await runBuild(env, {gc: true}); + assert.deepEqual(statuses(result), ['ran']); + assert.equal(execCount(env), 2); + assert.ok(!fs.existsSync(oldObject)); + + const store = new Store(env.dbPath); + assert.deepEqual(store.liveObjects().size, 1); + store.close(); +}); + +test('same-bytes source rename executes nothing; nameSensitive reruns', async () => { + const env = setup(); + copyRule(env, src(env, 'a.txt', 'stable'), 'out.txt'); + copyRule(env, src(env, 'ns.txt', 'stable'), 'ns-out.txt', {nameSensitive: true}); + await runBuild(env); + assert.equal(execCount(env), 2); + + resetDecls(); + fs.renameSync(pathlib.join(env.root, 'src/a.txt'), pathlib.join(env.root, 'src/moved.txt')); + fs.renameSync(pathlib.join(env.root, 'src/ns.txt'), pathlib.join(env.root, 'src/ns2.txt')); + copyRule(env, pathlib.join(env.root, 'src/moved.txt'), 'out.txt'); + copyRule(env, pathlib.join(env.root, 'src/ns2.txt'), 'ns-out.txt', {nameSensitive: true}); + const result = await runBuild(env); + assert.deepEqual(statuses(result), ['clean', 'ran']); + assert.equal(execCount(env), 3); +}); + +test('byte-identical inputs share one execution across two declarations', async () => { + const env = setup(); + const one = copyRule(env, src(env, 'one.txt', 'same-bytes'), 'one.txt'); + const two = copyRule(env, src(env, 'two.txt', 'same-bytes'), 'two.txt'); + const result = await runBuild(env); + assert.ok(result.ok); + assert.equal(execCount(env), 1); + assert.deepEqual(statuses(result).sort(), ['clean', 'ran']); + assert.equal(one.hash, two.hash); + assert.equal(one.filename, 'one.txt'); + assert.equal(two.filename, 'two.txt'); +}); + +// The consumer must be a *different* computation: an identical command over +// identical bytes would share the producer's key (by design). +function upcaseRule(env : Env, input : Artifact, out : string) : Artifact { + return rule(input, [`tr a-z A-Z < %f > %o && echo x >> ${env.execLog}`], [out])[0]!; +} + +test('chained rules: consumer follows producer, cas-missing reruns alone', async () => { + const env = setup(); + const mid = copyRule(env, src(env, 'a.txt', 'chain'), 'mid.txt'); + upcaseRule(env, mid, 'final.txt'); + const first = await runBuild(env); + assert.ok(first.ok); + assert.equal(execCount(env), 2); + + resetDecls(); + const mid2 = copyRule(env, pathlib.join(env.root, 'src/a.txt'), 'mid.txt'); + const final2 = upcaseRule(env, mid2, 'final.txt'); + const clean = await runBuild(env); + assert.deepEqual(statuses(clean), ['clean', 'clean']); + assert.equal(fs.readFileSync(casPath(env.casDir, final2.hash, 'txt'), 'utf8'), 'CHAIN'); + + resetDecls(); + fs.rmSync(casPath(env.casDir, final2.hash, 'txt')); + const mid3 = copyRule(env, pathlib.join(env.root, 'src/a.txt'), 'mid.txt'); + upcaseRule(env, mid3, 'final.txt'); + const rerun = await runBuild(env); + assert.deepEqual(statuses(rerun), ['clean', 'ran']); + const outcome = rerun.outcomes.get(getDecls()[1]!)!; + assert.deepEqual(outcome, {status: 'ran', reason: 'cas-missing'}); +}); + +test('failed producer blocks the consumer; nothing is recorded', async () => { + const env = setup(); + const bad = rule(src(env, 'a.txt', 'x'), ['false'], ['mid.txt'])[0]!; + copyRule(env, bad, 'final.txt'); + const result = await runBuild(env); + assert.ok(!result.ok); + assert.deepEqual(statuses(result), ['failed', 'blocked']); + const store = new Store(env.dbPath); + assert.equal(store.liveObjects().size, 0); + store.close(); +}); + +test('multi-output rules route %oN and skip all-or-nothing', async () => { + const env = setup(); + const declare = () => rule(src(env, 'a.txt', 'multi'), + [`printf one > %o1 && printf two > %o2 && echo x >> ${env.execLog}`], + ['x.txt', 'y.css']); + const [x1, y1] = declare(); + await runBuild(env); + assert.equal(fs.readFileSync(casPath(env.casDir, x1!.hash, 'txt'), 'utf8'), 'one'); + assert.equal(fs.readFileSync(casPath(env.casDir, y1!.hash, 'css'), 'utf8'), 'two'); + + resetDecls(); + fs.rmSync(casPath(env.casDir, y1!.hash, 'css')); + declare(); + const rerun = await runBuild(env); + assert.deepEqual(statuses(rerun), ['ran']); + assert.equal(execCount(env), 2); +}); + +test('a missing declared output fails the rule', async () => { + const env = setup(); + rule(src(env, 'a.txt', 'x'), ['printf one > %o1'], ['x.txt', 'missing.css']); + const result = await runBuild(env); + assert.ok(!result.ok); + const outcome = result.outcomes.get(getDecls()[0]!)!; + assert.equal(outcome.status, 'failed'); + assert.match((outcome as {message : string}).message, /did not produce: missing.css/); + const store = new Store(env.dbPath); + assert.equal(store.liveObjects().size, 0); + store.close(); +}); + +test('dry run reports without writing state', async () => { + const env = setup(); + const mid = copyRule(env, src(env, 'a.txt', 'dry'), 'mid.txt'); + copyRule(env, mid, 'final.txt'); + const result = await runBuild(env, {dryRun: true}); + assert.ok(!result.ok); + assert.deepEqual(result.outcomes.get(getDecls()[0]!), {status: 'would-run', reason: 'new'}); + assert.deepEqual(result.outcomes.get(getDecls()[1]!), {status: 'would-run', reason: 'blocked'}); + assert.equal(execCount(env), 0); + assert.ok(!fs.existsSync(env.casDir)); + const store = new Store(env.dbPath); + assert.equal(store.liveObjects().size, 0); + store.close(); +}); + +test('missing sources fail upfront', async () => { + const env = setup(); + copyRule(env, pathlib.join(env.root, 'src/nope.txt'), 'out.txt'); + await assert.rejects(() => runBuild(env), /Missing input files/); +});