Dependency Graphs & Incremental Builds - Task DAGs, Topological Scheduling, Content Hashing & Affected Detection

"""Mini build engine: topological scheduling plus three ways to decide "is it dirty?".

Rebuilders compared (names from "Build Systems a la Carte", Mokhov, Mitchell, Peyton Jones 2018):
  - mtime     : Make-style dirty bit; rebuild if any input is newer than the output
  - hash      : content hashes for change detection, but any re-run dependency marks
                dependents dirty (dirtiness propagates, no early cutoff)
  - hash+cut  : verifying trace; rebuild only if an input's hash differs from the last
                build, so a rebuilt output with an identical hash stops propagation (early cutoff)
Time is simulated with a counter so the output is deterministic.
"""
import hashlib
from collections import deque

def sha(s: str) -> str:
    return hashlib.sha256(s.encode()).hexdigest()[:8]

# key -> (inputs, function of input values). Sources have no entry.
rules = {
    "util.o": (["util.c", "util.h"], lambda v: "O[" + strip_comments(v[0]) + "]"),
    "main.o": (["main.c", "util.h"], lambda v: "O[" + strip_comments(v[0]) + "]"),
    "app":    (["util.o", "main.o"], lambda v: "EXE[" + v[0] + v[1] + "]"),
}

def strip_comments(code: str) -> str:
    """Our 'compiler' ignores // comments, so a comment edit yields identical object code."""
    return " ".join(line.split("//")[0].strip() for line in code.splitlines()).strip()

def topo_levels(rules):
    """Kahn's algorithm, grouped into levels: everything in one level can run in parallel."""
    nodes = set(rules) | {i for ins, _ in rules.values() for i in ins}
    indeg = {n: 0 for n in nodes}
    users = {n: [] for n in nodes}
    for out, (ins, _) in rules.items():
        for i in ins:
            indeg[out] += 1; users[i].append(out)
    level = sorted(n for n in nodes if indeg[n] == 0)
    levels, seen = [], 0
    while level:
        levels.append(level); seen += len(level)
        nxt = []
        for n in level:
            for u in users[n]:
                indeg[u] -= 1
                if indeg[u] == 0: nxt.append(u)
        level = sorted(nxt)
    if seen != len(nodes):
        raise ValueError("cycle detected: a build graph must be a DAG")
    return levels

class Engine:
    def __init__(self, mode):
        self.mode = mode; self.clock = 0
        self.files = {}    # name -> (content, mtime)
        self.trace = {}    # out -> list of input hashes seen at last build
    def write(self, name, content, touch_only=False):
        self.clock += 1
        old = self.files.get(name, (content, 0))[0]
        self.files[name] = (old if touch_only else content, self.clock)
    def build(self):
        ran = []
        for level in topo_levels(rules):
            for out in level:
                if out not in rules: continue
                ins, fn = rules[out]
                if self.dirty(out, ins, ran):
                    ran.append(out)
                    self.write(out, fn([self.files[i][0] for i in ins]))
                    self.trace[out] = [sha(self.files[i][0]) for i in ins]
        return ran
    def dirty(self, out, ins, ran):
        if out not in self.files: return True
        if self.mode == "mtime":
            return any(self.files[i][1] > self.files[out][1] for i in ins)
        changed = self.trace.get(out) != [sha(self.files[i][0]) for i in ins]
        if self.mode == "hash":
            return changed or any(i in ran for i in ins)
        return changed   # hash+cut: identical input hashes mean nothing to do

print("parallel levels:", topo_levels(rules))
for mode in ["mtime", "hash", "hash+cut"]:
    e = Engine(mode)
    e.write("util.h", "int add(int,int);")
    e.write("util.c", "int add(int a,int b){return a+b;}")
    e.write("main.c", "int main(){return add(1,2);}")
    e.build()
    a = e.build()
    e.write("util.h", "", touch_only=True)                      # `touch util.h`
    b = e.build()
    e.write("main.c", "int main(){return add(1,2);} // tidy")   # comment-only edit
    c = e.build()
    print(f"{mode:9} no-op={a} touch={b} comment-edit={c}")

rules["util.h"] = (["app"], lambda v: v[0])   # introduce a cycle
try:
    topo_levels(rules)
except ValueError as err:
    print("error:", err)

parallel levels: [['main.c', 'util.c', 'util.h'], ['main.o', 'util.o'], ['app']]
mtime     no-op=[] touch=['main.o', 'util.o', 'app'] comment-edit=['main.o', 'app']
hash      no-op=[] touch=[] comment-edit=['main.o', 'app']
hash+cut  no-op=[] touch=[] comment-edit=['main.o']
error: cycle detected: a build graph must be a DAG

// Mini monorepo task engine: package graph -> task graph -> topological waves,
// content-hash based skipping, and "affected" selection from a list of changed files.
// Tiny non-cryptographic hash (FNV-1a, 32-bit) so the demo needs no Node type packages.
// Real tools use SHA-256 (Bazel, Nix) or fast hashes such as xxHash; the idea is identical.
function sha(s: string): string {
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193) >>> 0; }
  return h.toString(16).padStart(8, "0");
}

type Pkg = { deps: string[]; files: Record<string, string> };
const repo: Record<string, Pkg> = {
  utils: { deps: [], files: { "utils/src/index.ts": "export const x = 1" } },
  ui: { deps: ["utils"], files: { "ui/src/button.tsx": "button v1" } },
  api: { deps: ["utils"], files: { "api/src/client.ts": "client v1" } },
  web: { deps: ["ui", "api"], files: { "web/src/page.tsx": "page v1" } },
  docs: { deps: ["ui"], files: { "docs/src/index.md": "docs v1" } },
};

// Task graph: "build" depends on the build of every dependency package (Turborepo "^build").
// "test" depends on the same package's build.
type Task = string; // "pkg:task"
const deps = (t: Task): Task[] => {
  const [p, name] = t.split(":");
  return name === "build" ? repo[p].deps.map((d) => `${d}:build`) : [`${p}:build`];
};

// Topological waves (Kahn). Tasks in the same wave can run in parallel.
function waves(targets: Task[]): Task[][] {
  const all = new Set<Task>();
  const visit = (t: Task) => { if (!all.has(t)) { all.add(t); deps(t).forEach(visit); } };
  targets.forEach(visit);
  const indeg = new Map<Task, number>([...all].map((t) => [t, deps(t).length]));
  const out: Task[][] = [];
  let ready = [...all].filter((t) => indeg.get(t) === 0).sort();
  while (ready.length) {
    out.push(ready);
    const next: Task[] = [];
    for (const t of all) {
      if (indeg.get(t)! > 0 && deps(t).every((d) => out.flat().includes(d)) && !next.includes(t)) next.push(t);
    }
    next.forEach((t) => indeg.set(t, 0));
    ready = next.sort();
  }
  return out;
}

// Task hash = own files + task name + hashes of the tasks it depends on (so changes propagate).
const memo = new Map<Task, string>();
function taskHash(t: Task): string {
  if (memo.has(t)) return memo.get(t)!;
  const p = t.split(":")[0];
  const own = Object.entries(repo[p].files).map(([f, c]) => f + sha(c)).join("|");
  const h = sha(t + own + deps(t).map(taskHash).join(","));
  memo.set(t, h);
  return h;
}

const cache = new Set<string>();
function run(targets: Task[], label: string) {
  memo.clear();
  const ran: Task[] = [];
  const hit: Task[] = [];
  for (const wave of waves(targets)) {
    for (const t of wave) {
      const k = taskHash(t);
      if (cache.has(k)) hit.push(t); else { cache.add(k); ran.push(t); }
    }
  }
  console.log(`${label}: ran ${ran.length} [${ran.join(" ")}] cache hits ${hit.length}`);
}

// "Affected": owning packages of changed files, plus everything that depends on them.
function affected(changedFiles: string[]): string[] {
  const set = new Set(Object.keys(repo).filter((p) => changedFiles.some((f) => f.startsWith(p + "/"))));
  let grew = true;
  while (grew) {
    grew = false;
    for (const [p, pkg] of Object.entries(repo)) {
      if (!set.has(p) && pkg.deps.some((d) => set.has(d))) { set.add(p); grew = true; }
    }
  }
  return [...set].sort();
}

const everything = Object.keys(repo).flatMap((p) => [`${p}:build`, `${p}:test`]);
console.log("waves:", waves(everything).map((w) => `[${w.join(" ")}]`).join(" -> "));
run(everything, "cold build");
run(everything, "second build, nothing changed");

repo.api.files["api/src/client.ts"] = "client v2";
const changed = ["api/src/client.ts"];
const aff = affected(changed);
console.log("affected packages:", aff.join(", "));
run(aff.flatMap((p) => [`${p}:build`, `${p}:test`]), "affected-only run after editing api");

waves: [utils:build] -> [api:build ui:build utils:test] -> [api:test docs:build ui:test web:build] -> [docs:test web:test]
cold build: ran 10 [utils:build api:build ui:build utils:test api:test docs:build ui:test web:build docs:test web:test] cache hits 0
second build, nothing changed: ran 0 [] cache hits 10
affected packages: api, web
affected-only run after editing api: ran 4 [api:build api:test web:build web:test] cache hits 2