keep a skipped repo's store generation in state and index unchanged repos in knot.json
Signed-off-by: dawn <90008@klbr.net>
17 unmodified lines181920212223242526272829303132333435362337383940414243443132454647484950515253545556575859606162636465666768697071727339 unmodified lines1131141151161171181191201215 unmodified lines17 unmodified lines } return chain;};
const MANIFEST = /^at:\/\/(did:[^/]+)\/org\.tangled\.repo\.manifest\/self$/;
const repoOfParams = (params) => { if (!params) return null; if (params.repo?.startsWith("did:")) return params.repo; if (params.repoDid?.startsWith("did:")) return params.repoDid; if (typeof params.subject === "string") { const m = MANIFEST.exec(params.subject); if (m) return m[1]; if (params.subject.startsWith("did:")) return params.subject; } return null;};
const parse = async (files) => { const diffs = { entries: 0, exact: new Set(), defaults: new Set(), any: new Map(), resets: new Set() }; const diffs = { entries: 0, exact: new Set(), defaults: new Set(), any: new Map(), resets: new Set(), repos: new Set(), global: false }; for (const file of files) { let text; try { text = await readFile(file, "utf8"); } catch { return null; } for (const line of text.split("\n")) { if (!line) continue; for (let start = 0, end = 0; start < text.length; start = end + 1) { end = text.indexOf("\n", start); if (end === -1) end = text.length; if (end === start) continue; const line = text.slice(start, end); diffs.entries++; const entry = JSON.parse(line); if (entry.op === "reset") { diffs.resets.add(entry.params.repo); diffs.repos.add(entry.params.repo); } else if (entry.default) { diffs.defaults.add(entry.nsid); diffs.global = true; } else if (entry.any) { const repo = repoOfParams(entry.params); if (repo) diffs.repos.add(repo); else diffs.global = true; const params = { ...entry.params }; for (const name of entry.any) delete params[name]; if (!diffs.any.has(entry.nsid)) diffs.any.set(entry.nsid, { names: entry.any, keys: new Set() }); diffs.any.get(entry.nsid).keys.add(keyOf(entry.nsid, params)); } else { const repo = repoOfParams(entry.params); if (repo) diffs.repos.add(repo); else diffs.global = true; // an added key is one a page may have read while the map default answered it diffs.exact.add(keyOf(entry.nsid, entry.params)); } }39 unmodified lines// the routes whose reads changed: pages to render again, and pages the budget or a failed// read left out, which may stand nowexport const dirtyPages = (prior, diffs) => { if (diffs.resets.has(prior.did)) return "all"; if (!diffs.global && diffs.repos && !diffs.repos.has(prior.did)) return new Set(); const dirty = new Set(); for (const [route, page] of Object.entries(prior.pages)) { if (page.deps.some((key) => touched(diffs, key))) dirty.add(route); }5 unmodified lines80 unmodified lines81828384858687888990919293949596979899100101102103104105106107891081099111011111294951131141151161171181191201211222 unmodified lines12512612712810612913013113213311 unmodified lines1451461471481491501511521535 unmodified lines15916016116213916316416516616716842 unmodified lines80 unmodified lines !knotState || knotState.knot_format !== hook.format || knotState.renderer_format !== RENDERER_FORMAT || knotState.build !== build; const knownRepos = (!full && knotState?.repos) || {}; const nextRepos = {}; const plans = []; let diffEntries = 0; for (const repo of hook.repos) { const prefix = prefixOf(repo); const summary = knownRepos[repo.did]; if (!full && summary && repo.change === "unchanged" && summary.prefix === prefix) { if (summary.generation === hook.generation) { nextRepos[repo.did] = summary; continue; } const diffs = await readDiff(hookPath, hook, summary.generation); if (diffs !== null) { diffEntries = Math.max(diffEntries, diffs.entries); if (!diffs.global && !diffs.repos.has(repo.did)) { nextRepos[repo.did] = { generation: hook.generation, prefix, pages: summary.pages }; status.repos[repo.did] = { skipped: true }; continue; } } } const prior = loadState(repo.did); const fresh = full || !prior || repo.change === "added" || prior.prefix !== prefixOf(repo); const fresh = full || !prior || repo.change === "added" || prior.prefix !== prefix; if (fresh) { plans.push({ repo, work: "all", weight: prior ? Object.keys(prior.pages).length : Infinity }); plans.push({ repo, work: "all", weight: prior ? Object.keys(prior.pages).length : (summary?.pages ?? Infinity) }); continue; } if (prior.generation === hook.generation) continue; const diffs = await readDiff(hookPath, hook, prior.generation); const since = summary?.generation ?? prior.generation; if (since === hook.generation) { nextRepos[repo.did] = { generation: hook.generation, prefix, pages: Object.keys(prior.pages).length }; continue; } const diffs = await readDiff(hookPath, hook, since); if (diffs === null) { plans.push({ repo, work: "all", weight: Object.keys(prior.pages).length }); continue; }2 unmodified lines if (dirty === "all" || dirty.size > 0 || repo.change !== "unchanged") { plans.push({ repo, work: dirty === "all" ? "all" : [...dirty], weight: dirty === "all" ? Object.keys(prior.pages).length : dirty.size }); } else { // nothing it read changed: the pages stand for this generation too await atomicJson(path.join(stateDir, tokenOf(repo.did) + ".json"), { ...prior, generation: hook.generation }); nextRepos[repo.did] = { generation: hook.generation, prefix, pages: Object.keys(prior.pages).length }; status.repos[repo.did] = { skipped: true }; } } // biggest first, so the last worker to finish isn't the one holding ghostty11 unmodified lines if (result.error) { failures.push(plan.repo.did); status.repos[plan.repo.did] = { error: result.error }; } else { nextRepos[plan.repo.did] = { generation: hook.generation, prefix: prefixOf(plan.repo), pages: result.pages }; status.repos[plan.repo.did] = { ...result, wall_ms: Math.round(performance.now() - t) }; for (const key of Object.keys(totals)) totals[key] += result[key] ?? 0; } await writeStatus();5 unmodified lines await atomicJson(path.join(stateDir, "knot.json"), { knot_format: hook.format, renderer_format: RENDERER_FORMAT, build, generation: hook.generation generation: hook.generation, repos: nextRepos }); } const summary = { state: failures.length === 0 ? "done" : "failed",42 unmodified lines12345456789198 unmodified linesExpand all208209210211212213214215216217218219220221222223224225226227228214215229230231232217218219220221233234235236237238239240241242243244245246247248249128 unmodified linesExpand all378379380381358382383384385386103 unmodified linesExpand all// renders one repo's pages in its own process, so repos render in parallel across cores.// the coordinator sends {repo, work, budget} and gets the repo's stats backimport { AsyncLocalStorage } from "node:async_hooks";import { readFileSync } from "node:fs";import { link, mkdir, readdir, rename, rm, symlink, writeFile } from "node:fs/promises";import { linkSync, mkdirSync, readFileSync } from "node:fs";import { link, mkdir, readdir, realpath, rename, rm, symlink, writeFile } from "node:fs/promises";import path from "node:path";import { pathToFileURL } from "node:url";import { parseArgs } from "node:util";import { gzipSync } from "node:zlib";198 unmodified linesExpand all // serves the .gz next to it const empty = path.join(tmpDir, ".empty"); await writeFile(empty, "");
const madeDirs = new Set([tmpDir]); const dirPromises = new Map(); const ensureDir = (dir) => { if (madeDirs.has(dir)) return; let p = dirPromises.get(dir); if (!p) { p = mkdir(dir, { recursive: true }).then(() => { madeDirs.add(dir); dirPromises.delete(dir); }); dirPromises.set(dir, p); } return p; };
const stats = { rendered: 0, kept: 0, dropped: 0, over_budget: 0, render_ms: 0 }; if (prior) { const priorDir = path.join(storeOf(repo.did), prior.generation); await pool(Object.keys(pages).filter((route) => !dirty.has(route)), async (route) => { const priorDir = await realpath(path.join(out, "site", repo.did)).catch(() => path.join(storeOf(repo.did), prior.generation)); for (const route of Object.keys(pages)) { if (dirty.has(route)) continue; const to = fileFor(tmpDir, prefix, route); await mkdir(path.dirname(to), { recursive: true }); await link(empty, to); // a page whose file went missing renders again await link(fileFor(priorDir, prefix, route) + ".gz", to + ".gz").catch(() => dirty.add(route)); }); const dir = path.dirname(to); if (!madeDirs.has(dir)) { mkdirSync(dir, { recursive: true }); madeDirs.add(dir); } try { linkSync(empty, to); linkSync(fileFor(priorDir, prefix, route) + ".gz", to + ".gz"); } catch { // a page whose file went missing renders again dirty.add(route); } } }
let bytes = Object.values(pages).reduce((sum, page) => sum + page.bytes, 0); const minViewBytes = new Map();128 unmodified linesExpand all const file = fileFor(tmpDir, prefix, item.route); await rm(file, { force: true }); await rm(file + ".gz", { force: true }); if (!item.gz) continue; await mkdir(path.dirname(file), { recursive: true }); await ensureDir(path.dirname(file)); await link(empty, file); await writeFile(file + ".gz", item.gz); item.gz = null; }103 unmodified linesExpand all