forge lake reads what is new in events.jsonl, metrics.jsonl, every repo's review logs and refs/notes/ci, and adds it to a DuckLake (DuckDB 1.5.6 through @duckdb/node-api) with its catalog in .forge/lake/catalog.sqlite and its data in .forge/lake/data, or on S3 or R2 with FORGE_LAKE_DATA. The rows and where each source got to go in one transaction, so a run that dies adds nothing and a second run adds nothing new. Small inserts stay inlined in the catalog, and a checkpoint about once a day flushes them to Parquet and merges files. events.jsonl is cut to its last 1000 lines by cmd_build, into a new file. Each source remembers its inode, first line and the hashes of the last lines read; when they no longer match, the file is read whole and taken up after the last line read, or, if the cut went past that, whole, saying the lines in between are gone. One writer: cmd_lake holds lake.lock and a second run waits for it (or gives up after FORGE_LAKE_WAIT seconds). Builds start it in the background the way they start the mirror, debounced by FORGE_LAKE_DELAY. forge lake query runs reads only: the lake attached read-only, only SELECT, EXPLAIN and PRAGMA statements, no file access beyond the lake's data and settings locked. The extensions (ducklake, sqlite, httpfs) are installed into the image at build time and checked against the versions this was tested with, so the container runs offline. Events now take FORGE_NOW like review logs, so the demo's and tests' events have their clock. test/lib.sh's exit trap no longer fails when the processes it stops are gone.
- commit
- dc590460b92a9f89fe5b7a25eb51219ed8e87653
- parent
- 49243f1e974926e5980facf27975ce8859ca890b
- author
- cat <admin@dev.site>
- kind
- feat risk score 1.9 of 3, Jev's estimate
| M | .dockerignore | 1 | + |
| M | .gitignore | 1 | + |
| M | Dockerfile | 7 | +- |
| M | bin/ | 57 | +++- |
| A | lib/ | 379 | +++++++++++++++++++++++ |
| A | lib/ | 501 | ++++++++++++++++++++++++++++++ |
| M | package-lock.json | 142 | +++++++++ |
| M | package.json | 1 | + |
| M | server/ | 7 | + |
| M | test/ | 10 | +- |
| M | test/ | 4 | + |
| A | test/ | 29 | ++ |
| A | test/ | 28 | ++ |
| A | test/ | 32 | ++ |
| A | test/ | 41 | ++ |
| A | test/ | 39 | ++ |
| A | test/ | 59 | ++++ |
| A | test/ | 47 | +++ |
@@ -27,8 +27,13 @@ COPY server/container/sshd.conf /etc/ssh/sshd_config.d/forge.conf
2727 2828 WORKDIR /opt/forge2929 COPY package.json package-lock.json ./30-RUN npm ci --omit=dev --no-audit --no-fund30+# npm also installs DuckDB's musl build, which glibc doesn't need (65M).31+RUN npm ci --omit=dev --no-audit --no-fund && rm -rf node_modules/@duckdb/node-bindings-linux-*-musl3132 COPY . .33+# forge lake's DuckDB extensions (DuckLake, SQLite for its catalog, httpfs34+# for data on R2), installed now so the container never downloads one; the35+# install checks they are the builds lib/lake.ts was tested with.36+RUN FORGE_DUCKDB_EXTENSIONS=/opt/forge/.duckdb/extensions node lib/lake.ts install3237 RUN chmod -R a+rX /opt/forge && ln -s /opt/forge/server/container/forge-admin /usr/local/bin/forge-admin3338 3439 EXPOSE 2222 8080@@ -27,6 +27,8 @@
2727 # forge build [owner/repo]2828 # forge jev [owner/repo] ask Jev about new commits and reviews (needs TYPESAFE_API_KEY)2929 # forge metrics [n] [owner/repo] recent builds and where their time went30+# forge lake [--dry-run] add new events, builds, reviews and CI results to the lake31+# forge lake query "<sql>" [--json|--csv] read-only SQL over the lake3032 # forge gc [--dry-run] [--age minutes] remove unused site and release files, hostnames; repack, commit-graph3133 # forge backup [--dry-run] [--force] snapshot the forge with restic, then check it and test-restore it3234 # forge backups [--status] the last backup's health, then the snapshots@@ -1448,7 +1450,7 @@ ev_private() { [ -f "$1/git-daemon-export-ok" ] || printf ',"private":true'; }
14481450 # After a comment or approval: tell open pages, and rebuild the repo the way14491451 # post-receive does after a push.14501452 review_changed() {1451- printf '{"t":%s000,"repo":"%s","type":"%s","user":"%s","review":%s%s}\n' "$(date +%s)" "$1" "$2" "$user" "$RID" "$(ev_private "$RDIR")" >>"$FORGE_STATE/events.jsonl"1453+ printf '{"t":%s000,"repo":"%s","type":"%s","user":"%s","review":%s%s}\n' "${FORGE_NOW:-$(date +%s)}" "$1" "$2" "$user" "$RID" "$(ev_private "$RDIR")" >>"$FORGE_STATE/events.jsonl"14521454 rebuild "$1" "$2"14531455 }14541456 @@ -2200,7 +2202,7 @@ noreview
22002202 if [ "$rel" = 1 ] && [ -f "$GIT_DIR/git-daemon-export-ok" ]; then release_latest "$FORGE_WWW"; fi22012203 git update-server-info22022204 refs=$(printf '%s' "$PUSHED" | tr -d '"\\')2203- printf '{"t":%s000,"repo":"%s","type":"push","user":"%s","refs":"%s"%s}\n' "$(date +%s)" "$OWNER/$NAME" "${USER_:-admin}" "$refs" "$(ev_private "$GIT_DIR")" >>"$FORGE_STATE/events.jsonl"2205+ printf '{"t":%s000,"repo":"%s","type":"push","user":"%s","refs":"%s"%s}\n' "${FORGE_NOW:-$(date +%s)}" "$OWNER/$NAME" "${USER_:-admin}" "$refs" "$(ev_private "$GIT_DIR")" >>"$FORGE_STATE/events.jsonl"22042206 rebuild "$OWNER/$NAME" push22052207 }22062208 @@ -2310,6 +2312,7 @@ cmd_build() {
23102312 if [ -f "$FORGE_STATE/events.jsonl" ] && [ "$(wc -c <"$FORGE_STATE/events.jsonl")" -gt 1048576 ]; then23112313 tail -n 1000 "$FORGE_STATE/events.jsonl" >"$FORGE_STATE/events.jsonl.tmp" && mv "$FORGE_STATE/events.jsonl.tmp" "$FORGE_STATE/events.jsonl"23122314 fi2315+ lake_kick23132316 mirror_kick23142317 say "built ${only:-all repos} in $(($(date +%s) - t0))s -> $FORGE_WWW"23152318 }@@ -2319,6 +2322,55 @@ cmd_metrics() {
23192322 node "$FORGE_HOME/lib/metrics-report.ts" "$@"23202323 }23212324 2325+# ---------------------------------------------------------------- the lake2326+#2327+# forge lake adds the forge's facts (events, builds, review logs, CI notes)2328+# to a DuckLake in $FORGE_STATE/lake (lib/lake.ts). One writer at a time,2329+# under lake.lock; a second run waits for it. Builds start it the way they2330+# start the mirror: each stamps lake.want and waits FORGE_LAKE_DELAY seconds2331+# (10), and only the newest stamp runs, so a burst of pushes makes one run.2332+# FORGE_LAKE=0 turns that off.2333+2334+lake_kick() {2335+ [ "${FORGE_LAKE:-1}" != 0 ] || return 02336+ t=$$.$(date +%s)2337+ printf '%s\n' "$t" >"$FORGE_STATE/lake.want"2338+ (2339+ # Not the build's EXIT trap: that frees the build lock.2340+ trap - EXIT2341+ cd /2342+ sleep "${FORGE_LAKE_DELAY:-10}"2343+ [ "$(cat "$FORGE_STATE/lake.want" 2>/dev/null)" = "$t" ] || exit 02344+ exec "$FORGE_BIN" lake2345+ ) </dev/null >>"$FORGE_STATE/lake.log" 2>&1 &2346+}2347+2348+# The lake's data on S3 or R2 (FORGE_LAKE_DATA=s3://<bucket>/<path>, in2349+# backup.env or the config) takes FORGE_LAKE_* keys, or the backup's.2350+lake_env() {2351+ backup_env2352+ case ${FORGE_LAKE_DATA:-} in2353+ *://*) export FORGE_LAKE_DATA FORGE_LAKE_KEY_ID="${FORGE_LAKE_KEY_ID:-${FORGE_BACKUP_KEY_ID:-}}" \2354+ FORGE_LAKE_SECRET="${FORGE_LAKE_SECRET:-${FORGE_BACKUP_SECRET:-}}" FORGE_LAKE_ENDPOINT="${FORGE_LAKE_ENDPOINT:-}" FORGE_LAKE_REGION="${FORGE_LAKE_REGION:-auto}" ;;2355+ esac2356+}2357+2358+cmd_lake() {2359+ load_config2360+ lake_env2361+ case ${1:-} in2362+ query) shift; exec node "$FORGE_HOME/lib/lake.ts" query "$@" ;;2363+ '') dry= ;;2364+ -n | --dry-run) dry=--dry-run ;;2365+ *) die 'usage: forge lake [--dry-run] | forge lake query "<sql>" [--json|--csv]' ;;2366+ esac2367+ wait=${FORGE_LAKE_WAIT:-300}2368+ take_lock "$FORGE_STATE/lake.lock" "$wait" || die "lake: another forge lake has held $FORGE_STATE/lake.lock for ${wait}s"2369+ trap 'rm -rf "$FORGE_STATE/lake.lock"' EXIT2370+ trim_log "$FORGE_STATE/lake.log"2371+ node "$FORGE_HOME/lib/lake.ts" ingest $dry2372+}2373+23222374 # Was branch $2 ever a review in repo $1? Then its previews are leftovers.23232375 was_review() { grep -qxF "branch $2" "$1"/forge/reviews/* 2>/dev/null; }23242376 @@ -2987,6 +3039,7 @@ repo) cmd_repo "$@" ;;
29873039 build) cmd_build "$@" ;;29883040 jev) cmd_jev "$@" ;;29893041 metrics) cmd_metrics "$@" ;;3042+lake) cmd_lake "$@" ;;29903043 gc) cmd_gc "$@" ;;29913044 backup) cmd_backup "$@" ;;29923045 backups) cmd_backups "$@" ;;@@ -0,0 +1,379 @@
1+// What `forge lake` reads (lib/lake.ts): the forge's append-only facts, as2+// rows, and where it got to in each source so the next run starts there.3+//4+// events.jsonl and metrics.jsonl are read from a byte offset. cmd_build cuts5+// events.jsonl to its last 1000 lines once it passes 1 MB, which makes a new6+// file (tail into a temporary file, mv over it), so an offset alone would7+// skip or repeat lines. Each source also remembers the file's inode, a hash8+// of its first line, and hashes of the last lines it read. If the inode, the9+// first line or the line before the offset changed, or the file is shorter,10+// it was cut or replaced: the run reads it whole and starts after the last11+// line it had read, which is still there unless the cut went past it. Then12+// the lines between are gone, and so is everything before them.13+import fs from 'node:fs';14+import path from 'node:path';15+import { createHash } from 'node:crypto';16+import { catFileBatch, isAncestor, tryRun } from './git.ts';17+import { parseNote, type Verifier } from './attest.ts';18+import { readReviewLogs } from './reviews.ts';19+import type { MetricsLine } from './metrics.ts';20+import type { ReviewLog } from './types.ts';21+22+// Where a jsonl source was read up to.23+export interface Offset {24+ ino: number;25+ pos: number;26+ head: string;27+ // Hashes of the last lines read, oldest first (at most TAIL).28+ tail: string[];29+}30+31+export interface Read {32+ lines: string[];33+ next: Offset;34+ // continue: appended to; found: cut or replaced, and the run found where it35+ // had got to; lost: cut past that, so lines in between are gone; new: first read.36+ how: 'continue' | 'found' | 'lost' | 'new';37+}38+39+const TAIL = 16;40+const NL = 0x0a;41+export const lineHash = (b: Buffer | string): string => createHash('sha1').update(b).digest('hex').slice(0, 16);42+43+function readAt(fd: number, pos: number, len: number): Buffer {44+ const buf = Buffer.alloc(len);45+ let got = 0;46+ while (got < len) {47+ const n = fs.readSync(fd, buf, got, len - got, pos + got);48+ if (!n) break;49+ got += n;50+ }51+ return buf.subarray(0, got);52+}53+54+// The complete lines of buf (each ends in a newline), with the byte offset55+// just past each.56+function split(buf: Buffer, base = 0): { line: Buffer; end: number }[] {57+ const out: { line: Buffer; end: number }[] = [];58+ let at = 0;59+ for (let nl = buf.indexOf(NL); nl >= 0; nl = buf.indexOf(NL, at)) {60+ out.push({ line: buf.subarray(at, nl), end: base + nl + 1 });61+ at = nl + 1;62+ }63+ return out;64+}65+66+function firstLine(fd: number, size: number): string {67+ for (let len = 4096; ; len *= 4) {68+ const buf = readAt(fd, 0, Math.min(len, size));69+ const nl = buf.indexOf(NL);70+ if (nl >= 0) return lineHash(buf.subarray(0, nl));71+ if (buf.length >= size) return '';72+ }73+}74+75+// Does the line that ends at pos hash to want? (Reads back at most 1 MB; a76+// longer line counts as changed, and the file is read whole.)77+function endsWith(fd: number, pos: number, want: string | undefined): boolean {78+ if (!want) return pos === 0;79+ if (pos === 0) return false;80+ const start = Math.max(0, pos - (1 << 20));81+ const buf = readAt(fd, start, pos - start);82+ if (buf.length !== pos - start || buf[buf.length - 1] !== NL) return false;83+ const prev = buf.lastIndexOf(NL, buf.length - 2);84+ if (prev < 0 && start > 0) return false;85+ return lineHash(buf.subarray(prev + 1, buf.length - 1)) === want;86+}87+88+// Where to start in a file that was cut or replaced: after the last place89+// whose preceding lines are the tail that was read. A match may run back to90+// the start of the file, when the cut fell inside the tail.91+function resume(hashes: string[], tail: string[]): number | null {92+ for (let p = hashes.length; p >= 1; p--) {93+ const m = Math.min(tail.length, p);94+ if (m < tail.length && p !== m) continue;95+ let ok = m > 0;96+ for (let i = 1; ok && i <= m; i++) ok = hashes[p - i] === tail[tail.length - i];97+ if (ok) return p;98+ }99+ return null;100+}101+102+// The lines appended to file since prev, and where the next read starts.103+// Only complete lines count: one being written is read next time.104+export function readNew(file: string, prev: Offset | null): Read | null {105+ let fd: number;106+ try {107+ fd = fs.openSync(file, 'r');108+ } catch {109+ return null;110+ }111+ try {112+ const st = fs.fstatSync(fd);113+ const head = firstLine(fd, st.size);114+ if (prev && prev.ino === st.ino && st.size >= prev.pos && prev.head === head && endsWith(fd, prev.pos, prev.tail.at(-1))) {115+ const got = split(readAt(fd, prev.pos, st.size - prev.pos), prev.pos);116+ const tail = [...prev.tail, ...got.map((l) => lineHash(l.line))].slice(-TAIL);117+ return { lines: got.map((l) => l.line.toString('utf8')), next: { ino: st.ino, pos: got.at(-1)?.end ?? prev.pos, head, tail }, how: 'continue' };118+ }119+ const all = split(readAt(fd, 0, st.size));120+ const hashes = all.map((l) => lineHash(l.line));121+ const at = prev ? resume(hashes, prev.tail) : null;122+ const from = at ?? 0;123+ return {124+ lines: all.slice(from).map((l) => l.line.toString('utf8')),125+ next: { ino: st.ino, pos: all.at(-1)?.end ?? 0, head, tail: hashes.slice(-TAIL) },126+ how: !prev ? 'new' : at == null ? 'lost' : 'found',127+ };128+ } finally {129+ fs.closeSync(fd);130+ }131+}132+133+// ---------------------------------------------------------------- rows134+135+// Times are UTC strings DuckDB reads as TIMESTAMP.136+export const ts = (ms: number | null | undefined): string | null =>137+ ms == null || !Number.isFinite(ms) ? null : new Date(ms).toISOString().replace('T', ' ').replace('Z', '');138+139+const int = (v: unknown): number | null => (typeof v === 'number' && Number.isFinite(v) ? Math.round(v) : null);140+const num = (v: unknown): number | null => (typeof v === 'number' && Number.isFinite(v) ? v : null);141+const str = (v: unknown): string | null => (typeof v === 'string' ? v : null);142+143+export interface EventRow {144+ t: string | null;145+ repo: string | null;146+ type: string | null;147+ user: string | null;148+ review: number | null;149+ refs: string | null;150+ ms: number | null;151+ rendered: number | null;152+ private: boolean;153+}154+155+// One line of events.jsonl (hooks and builds append them; lib/events-server.ts).156+export function eventRow(line: string): EventRow | null {157+ let e: Record<string, unknown>;158+ try {159+ e = JSON.parse(line);160+ } catch {161+ return null;162+ }163+ if (!e || typeof e !== 'object') return null;164+ return {165+ t: ts(num(e.t)), repo: str(e.repo), type: str(e.type), user: str(e.user), review: int(e.review),166+ refs: str(e.refs)?.trim() || null, ms: num(e.ms), rendered: int(e.rendered), private: e.private === true,167+ };168+}169+170+export interface BuildRow {171+ id: string;172+ at: string | null;173+ code: string | null;174+ reason: string | null;175+ scope: string;176+ ms_total: number | null;177+ ms_data: number | null;178+ ms_render: number | null;179+ rss_mb: number | null;180+ written: number | null;181+ linked: number | null;182+ line: string;183+}184+185+export interface BuildRepoRow {186+ build: string;187+ at: string | null;188+ repo: string;189+ ms_load: number | null;190+ pages: number | null;191+ rendered: number | null;192+ skipped: number | null;193+ stages: string;194+}195+196+// One line of metrics.jsonl (lib/metrics.ts): the build, and a row for each197+// repo it loaded. The id is a hash of the line, which forge metrics uses to198+// merge the lake's history with lines it hasn't read yet.199+export function buildRows(line: string): { build: BuildRow; repos: BuildRepoRow[] } | null {200+ let m: MetricsLine;201+ try {202+ m = JSON.parse(line);203+ } catch {204+ return null;205+ }206+ if (!m || typeof m !== 'object' || typeof m.at !== 'string') return null;207+ const id = lineHash(line);208+ const at = ts(Date.parse(m.at));209+ return {210+ build: {211+ id, at, code: str(m.code), reason: str(m.reason), scope: m.only || 'all',212+ ms_total: num(m.ms?.total), ms_data: num(m.ms?.data), ms_render: num(m.ms?.render), rss_mb: int(m.rss_mb),213+ written: int(m.written), linked: int(m.linked), line,214+ },215+ repos: (Array.isArray(m.repos) ? m.repos : []).map((r) => ({216+ build: id, at, repo: r.slug, ms_load: num(r.ms?.load), pages: int(r.counts?.pages), rendered: int(r.counts?.rendered),217+ skipped: int(r.counts?.skipped), stages: JSON.stringify(r.ms ?? {}),218+ })),219+ };220+}221+222+export interface RepoRef {223+ slug: string;224+ dir: string;225+ public: boolean;226+}227+228+// Every repo under the root, as bin/forge sees them: <owner>/<name>.git with229+// a HEAD; public while it has git-daemon-export-ok.230+export function listRepos(root: string): RepoRef[] {231+ const out: RepoRef[] = [];232+ let owners: string[];233+ try {234+ owners = fs.readdirSync(root).filter((o) => !o.startsWith('.')).sort();235+ } catch {236+ return out;237+ }238+ for (const o of owners) {239+ let names: string[];240+ try {241+ names = fs.readdirSync(path.join(root, o)).filter((d) => d.endsWith('.git')).sort();242+ } catch {243+ continue;244+ }245+ for (const d of names) {246+ const dir = path.join(root, o, d);247+ if (!fs.existsSync(path.join(dir, 'HEAD'))) continue;248+ out.push({ slug: `${o}/${d.slice(0, -4)}`, dir, public: fs.existsSync(path.join(dir, 'git-daemon-export-ok')) });249+ }250+ }251+ return out;252+}253+254+export interface ReviewRow {255+ repo: string;256+ review: number;257+ kind: 'created' | 'rev' | 'approve' | 'deleted';258+ ts: string | null;259+ user: string | null;260+ rev: number | null;261+ sha: string | null;262+ base: string | null;263+ tree: string | null;264+ branch: string | null;265+ title: string | null;266+}267+268+export interface ReviewEnd {269+ repo: string;270+ review: number;271+ kind: 'landed' | 'closed';272+ ts: string | null;273+ user: string | null;274+ sha: string | null;275+ // The time is when the lake first saw it landed, not a logged one: the276+ // branch is still there, so no deletion was logged.277+ observed: boolean;278+}279+280+export interface ReviewLogFile {281+ log: ReviewLog;282+ hash: string;283+}284+285+// A repo's review logs, each with a hash of its file, so a run only redoes286+// the reviews whose logs grew.287+export function reviewLogs(dir: string): ReviewLogFile[] {288+ const logs = readReviewLogs(dir);289+ return logs.map((log) => {290+ let hash = '';291+ try {292+ hash = lineHash(fs.readFileSync(path.join(dir, 'forge', 'reviews', String(log.id))));293+ } catch {}294+ return { log, hash };295+ });296+}297+298+// Every line of one review log, as rows: created, each revision, each299+// approval (with the revision whose tip it approved), deleted.300+export function reviewRows(slug: string, log: ReviewLog): ReviewRow[] {301+ const row = (kind: ReviewRow['kind'], t: number | null, user: string | null, extra: Partial<ReviewRow> = {}): ReviewRow =>302+ ({ repo: slug, review: log.id, kind, ts: t ? ts(t * 1000) : null, user, rev: null, sha: null, base: null, tree: null, branch: null, title: null, ...extra });303+ const rows: ReviewRow[] = [];304+ if (log.created) rows.push(row('created', log.created.ts, log.created.user, { branch: log.branch, title: log.opts.title || null }));305+ for (const r of log.revs) rows.push(row('rev', r.ts, r.user, { rev: r.n, sha: r.sha, base: r.base }));306+ for (const a of log.approvals) {307+ const rev = [...log.revs].reverse().find((r) => r.sha === a.sha)?.n ?? null;308+ rows.push(row('approve', a.ts, a.user, { rev, sha: a.sha, tree: a.tree }));309+ }310+ if (log.deleted) rows.push(row('deleted', log.deleted.ts, log.deleted.user));311+ return rows;312+}313+314+// Has this review ended, and how? Deleted with its tip in the default branch315+// is landed (forge land and ssh land delete what they land), deleted316+// otherwise is closed, and a branch still there whose tip is in the default317+// branch landed without the deletion (observed at `now`). A tip that is its318+// own merge base had no commits of its own, so it isn't counted as landed.319+export function reviewEnd(slug: string, dir: string, log: ReviewLog, now: number): ReviewEnd | null {320+ const last = log.revs.at(-1);321+ if (!last) return null;322+ const def = tryRun(dir, ['symbolic-ref', '--short', 'HEAD'])?.trim() || 'main';323+ const defTip = tryRun(dir, ['rev-parse', '-q', '--verify', `refs/heads/${def}^{commit}`])?.trim();324+ const live = !log.deleted ? tryRun(dir, ['rev-parse', '-q', '--verify', `refs/heads/${log.branch}^{commit}`])?.trim() : null;325+ const tip = live || last.sha;326+ const own = !(tip === last.sha && tip === last.base);327+ const landed = !!defTip && own && isAncestor(dir, tip, defTip);328+ const end = (kind: ReviewEnd['kind'], t: number, user: string | null, observed: boolean): ReviewEnd =>329+ ({ repo: slug, review: log.id, kind, ts: ts(t * 1000), user, sha: tip, observed });330+ if (log.deleted) return end(landed ? 'landed' : 'closed', log.deleted.ts, log.deleted.user, false);331+ if (live && landed) return end('landed', now, null, true);332+ return null;333+}334+335+export interface CiRow {336+ id: string;337+ repo: string;338+ commit: string | null;339+ tree: string | null;340+ result: string | null;341+ exit: number | null;342+ started: string | null;343+ duration_s: number | null;344+ runner: string | null;345+ system: string | null;346+ env: string | null;347+ signer: string | null;348+ verified: boolean;349+}350+351+// The tip of a repo's CI notes, so a run skips repos whose notes didn't move.352+export const notesTip = (dir: string): string | null => tryRun(dir, ['rev-parse', '-q', '--verify', 'refs/notes/ci^{commit}'])?.trim() || null;353+354+// The attestations in a repo's refs/notes/ci that aren't in `known` (ids are355+// a hash of the signed statement and its signature, so a note that gained a356+// block only adds that block). Signatures are checked as they are read.357+export function ciRows(slug: string, dir: string, known: Set<string>, verifier: Verifier): CiRow[] {358+ const list = tryRun(dir, ['ls-tree', '-r', 'refs/notes/ci']);359+ if (!list) return [];360+ const blobs = list.split('\n').filter(Boolean).map((l) => l.split(/\s+/)).filter((f) => f[1] === 'blob').map((f) => f[2]);361+ const out: CiRow[] = [];362+ const seen = new Set(known);363+ for (const b of catFileBatch(dir, blobs).values()) {364+ for (const a of parseNote(b.data.toString('utf8'))) {365+ const id = createHash('sha256').update(a.payload + a.signature).digest('hex').slice(0, 32);366+ if (seen.has(id)) continue;367+ seen.add(id);368+ const v = verifier.verify(a);369+ const dur = /^(\d+)s$/.exec(a.duration ?? '');370+ out.push({371+ id, repo: slug, commit: a.commit ?? null, tree: a.tree ?? null, result: a.result ?? null,372+ exit: a.exit != null && /^-?\d+$/.test(a.exit) ? Number(a.exit) : null,373+ started: a.started ? ts(Date.parse(a.started)) : null, duration_s: dur ? Number(dur[1]) : null,374+ runner: a.runner ?? null, system: a.system ?? null, env: a.env ?? null, signer: v.signer, verified: v.ok,375+ });376+ }377+ }378+ return out;379+}@@ -0,0 +1,501 @@
1+// forge lake: the forge's facts in a DuckLake, so their history outlives the2+// files they come from. Events, builds, review lifecycles and CI3+// attestations are append-only facts spread over files that get cut4+// (events.jsonl), aren't backed up (metrics.jsonl) or live in each repo5+// (review logs, refs/notes/ci). Each run reads what is new since the last6+// one (lib/lake-sources.ts) and adds it in one transaction, together with7+// where it got to, so a run that dies adds nothing and the next one starts8+// from the same place.9+//10+// One writer: bin/forge runs this under $FORGE_STATE/lake.lock. DuckDB11+// can't share a database file between processes, and DuckLake over SQLite12+// takes writes from several but turns many away with "database is locked"13+// under contention. Readers (forge lake query, forge metrics) attach it14+// read-only and can run alongside.15+//16+// node lib/lake.ts ingest [--dry-run] add what is new17+// node lib/lake.ts query <sql> [--json|--csv] read-only, no file access18+// node lib/lake.ts install install the DuckDB extensions (the Dockerfile runs it)19+import fs from 'node:fs';20+import os from 'node:os';21+import path from 'node:path';22+import { fileURLToPath } from 'node:url';23+import { DuckDBInstance, StatementType, type DuckDBConnection } from '@duckdb/node-api';24+import { Verifier } from './attest.ts';25+import { tryRun } from './git.ts';26+import {27+ buildRows, ciRows, eventRow, listRepos, notesTip, readNew, reviewEnd, reviewLogs, reviewRows,28+ type BuildRepoRow, type BuildRow, type CiRow, type EventRow, type Offset, type Read, type ReviewEnd, type ReviewRow,29+} from './lake-sources.ts';30+31+const home = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');32+33+// The extensions DuckDB 1.5.6 installs from its core repository, and the34+// builds of them this code was tested with. DuckDB checks their signatures;35+// `install` also checks these versions, so an image never ships different36+// ones unnoticed.37+export const EXTENSIONS: Record<string, string> = { ducklake: 'ac7595b0', sqlite_scanner: '5274128', httpfs: '4bc690d' };38+39+export interface LakeConfig {40+ root: string;41+ state: string;42+ www: string;43+ dir: string;44+ catalog: string;45+ // A directory, or an s3:// URL (FORGE_LAKE_DATA).46+ data: string;47+ local: boolean;48+ exportDir: string;49+ extensions: string;50+}51+52+export function lakeConfig(env: NodeJS.ProcessEnv = process.env): LakeConfig {53+ const root = path.resolve(env.FORGE_ROOT || 'demo/root');54+ const state = path.join(root, '.forge');55+ const dir = path.join(state, 'lake');56+ const data = env.FORGE_LAKE_DATA || path.join(dir, 'data');57+ const local = !/^[a-z0-9]+:\/\//i.test(data);58+ // Extensions: the image's (installed at build time, so the container never59+ // downloads one), else a cache outside FORGE_ROOT's backups.60+ const bundled = path.join(home, '.duckdb', 'extensions');61+ const extensions = env.FORGE_DUCKDB_EXTENSIONS ||62+ (fs.existsSync(bundled) ? bundled : path.join(env.XDG_CACHE_HOME || path.join(os.homedir(), '.cache'), 'duckdb', 'extensions'));63+ return {64+ root, state, www: env.FORGE_WWW || path.join(state, 'www'), dir, catalog: path.join(dir, 'catalog.sqlite'),65+ data: local ? path.resolve(data) + '/' : data.replace(/\/?$/, '/'), local, exportDir: path.join(dir, 'export'), extensions,66+ };67+}68+69+export const lit = (s: string): string => `'${s.replaceAll("'", "''")}'`;70+export const ident = (s: string): string => `"${s.replaceAll('"', '""')}"`;71+72+// Tables and their columns. Times are UTC.73+const TABLES: Record<string, [string, string][]> = {74+ // events.jsonl: pushes, builds, comments, approvals, as appended.75+ events: [['t', 'TIMESTAMP'], ['repo', 'VARCHAR'], ['type', 'VARCHAR'], ['user', 'VARCHAR'], ['review', 'INTEGER'], ['refs', 'VARCHAR'], ['ms', 'DOUBLE'], ['rendered', 'INTEGER'], ['private', 'BOOLEAN']],76+ // metrics.jsonl: one row a build (scope is the repo, "-" or "all"), the77+ // line itself for forge metrics, and a row for each repo it loaded.78+ builds: [['id', 'VARCHAR'], ['at', 'TIMESTAMP'], ['code', 'VARCHAR'], ['reason', 'VARCHAR'], ['scope', 'VARCHAR'], ['ms_total', 'DOUBLE'], ['ms_data', 'DOUBLE'], ['ms_render', 'DOUBLE'], ['rss_mb', 'INTEGER'], ['written', 'INTEGER'], ['linked', 'INTEGER'], ['line', 'VARCHAR']],79+ build_repos: [['build', 'VARCHAR'], ['at', 'TIMESTAMP'], ['repo', 'VARCHAR'], ['ms_load', 'DOUBLE'], ['pages', 'INTEGER'], ['rendered', 'INTEGER'], ['skipped', 'INTEGER'], ['stages', 'VARCHAR']],80+ // Review logs: created, rev, approve, deleted; and how each review ended.81+ reviews: [['repo', 'VARCHAR'], ['review', 'INTEGER'], ['kind', 'VARCHAR'], ['ts', 'TIMESTAMP'], ['user', 'VARCHAR'], ['rev', 'INTEGER'], ['sha', 'VARCHAR'], ['base', 'VARCHAR'], ['tree', 'VARCHAR'], ['branch', 'VARCHAR'], ['title', 'VARCHAR']],82+ review_ends: [['repo', 'VARCHAR'], ['review', 'INTEGER'], ['kind', 'VARCHAR'], ['ts', 'TIMESTAMP'], ['user', 'VARCHAR'], ['sha', 'VARCHAR'], ['observed', 'BOOLEAN']],83+ // refs/notes/ci: one row an attestation.84+ ci: [['id', 'VARCHAR'], ['repo', 'VARCHAR'], ['commit', 'VARCHAR'], ['tree', 'VARCHAR'], ['result', 'VARCHAR'], ['exit', 'INTEGER'], ['started', 'TIMESTAMP'], ['duration_s', 'INTEGER'], ['runner', 'VARCHAR'], ['system', 'VARCHAR'], ['env', 'VARCHAR'], ['signer', 'VARCHAR'], ['verified', 'BOOLEAN']],85+ // Where each source was read up to (lib/lake-sources.ts), written in the86+ // same transaction as its rows.87+ sources: [['source', 'VARCHAR'], ['ino', 'BIGINT'], ['pos', 'BIGINT'], ['head', 'VARCHAR'], ['tail', 'VARCHAR'], ['at', 'TIMESTAMP']],88+};89+90+export class Lake {91+ cfg: LakeConfig;92+ conn: DuckDBConnection;93+ private db: DuckDBInstance;94+95+ private constructor(cfg: LakeConfig, db: DuckDBInstance, conn: DuckDBConnection) {96+ this.cfg = cfg;97+ this.db = db;98+ this.conn = conn;99+ }100+101+ // Rows as arrays of JSON values (BIGINTs as strings).102+ async rows(sql: string): Promise<unknown[][]> {103+ return (await this.conn.runAndReadAll(sql)).getRowsJson() as unknown[][];104+ }105+106+ async run(sql: string): Promise<void> {107+ await this.conn.run(sql);108+ }109+110+ close(): void {111+ this.conn.closeSync();112+ this.db.closeSync();113+ }114+115+ // Attach the lake: to write (creating it the first time), or read-only.116+ static async open(cfg: LakeConfig, { write }: { write: boolean }): Promise<Lake> {117+ if (!write && !fs.existsSync(cfg.catalog)) throw new LakeError(`no lake yet (${cfg.catalog}); forge lake makes it`);118+ const db = await DuckDBInstance.create(':memory:', {119+ extension_directory: cfg.extensions, autoinstall_known_extensions: 'false', autoload_known_extensions: 'false',120+ });121+ const conn = await db.connect();122+ const lake = new Lake(cfg, db, conn);123+ try {124+ await lake.load(cfg.local ? ['ducklake', 'sqlite'] : ['ducklake', 'sqlite', 'httpfs']);125+ if (!cfg.local) await lake.secret();126+ const fresh = !fs.existsSync(cfg.catalog);127+ if (write) {128+ fs.mkdirSync(cfg.dir, { recursive: true });129+ if (cfg.local) fs.mkdirSync(cfg.data, { recursive: true });130+ }131+ // The data path is stored in the catalog; this host's wins, so a lake132+ // restored under another root still finds its files.133+ await lake.run(`ATTACH ${lit('ducklake:sqlite:' + cfg.catalog)} AS lake (DATA_PATH ${lit(cfg.data)}, OVERRIDE_DATA_PATH true${write ? '' : ', READ_ONLY'})`);134+ if (write) {135+ await lake.run('BEGIN TRANSACTION');136+ if (fresh) {137+ // Snapshots older than 40 days are dropped at a checkpoint, and the138+ // files only they used are deleted a week later, so a backup of the139+ // catalog from the last five weeks still finds its files.140+ for (const [k, v] of [['expire_older_than', '40 days'], ['delete_older_than', '7 days'], ['parquet_compression', 'zstd']]) {141+ await lake.run(`CALL lake.set_option(${lit(k)}, ${lit(v)})`);142+ }143+ }144+ for (const [t, cols] of Object.entries(TABLES)) {145+ await lake.run(`CREATE TABLE IF NOT EXISTS lake.${t} (${cols.map(([c, ty]) => `${ident(c)} ${ty}`).join(', ')})`);146+ }147+ await lake.run('COMMIT');148+ }149+ } catch (e) {150+ lake.close();151+ throw e;152+ }153+ return lake;154+ }155+156+ // Load extensions, installing missing ones (only outside the image, which157+ // has them; DuckDB downloads them from its repository).158+ private async load(names: string[]): Promise<void> {159+ const have = new Set((await this.rows(`SELECT extension_name FROM duckdb_extensions() WHERE installed`)).map((r) => String(r[0])));160+ for (const n of names) {161+ const key = n === 'sqlite' ? 'sqlite_scanner' : n;162+ if (!have.has(key)) {163+ try {164+ await this.run(`INSTALL ${n}`);165+ } catch (e) {166+ throw new LakeError(`the DuckDB extension ${n} isn't installed in ${this.cfg.extensions}, and installing it failed: ${firstLine(e)}`);167+ }168+ }169+ await this.run(`LOAD ${n}`);170+ }171+ }172+173+ // Credentials for an s3:// data path (bin/forge's lake_env passes them).174+ private async secret(): Promise<void> {175+ const env = process.env;176+ const key = env.FORGE_LAKE_KEY_ID || '';177+ const secret = env.FORGE_LAKE_SECRET || '';178+ const endpoint = (env.FORGE_LAKE_ENDPOINT || '').replace(/^https?:\/\//, '').replace(/\/$/, '');179+ const parts = [`TYPE s3`, `KEY_ID ${lit(key)}`, `SECRET ${lit(secret)}`, `REGION ${lit(env.FORGE_LAKE_REGION || 'auto')}`];180+ if (endpoint) parts.push(`ENDPOINT ${lit(endpoint)}`, `URL_STYLE 'path'`);181+ await this.run(`CREATE SECRET lake_data (${parts.join(', ')})`);182+ }183+184+ // NDJSON rows into a table, through a scratch file DuckDB reads typed.185+ async insert(table: string, rows: object[], scratch: string): Promise<void> {186+ if (!rows.length) return;187+ const cols = TABLES[table];188+ const file = path.join(scratch, `${table}.jsonl`);189+ fs.writeFileSync(file, rows.map((r) => JSON.stringify(r)).join('\n') + '\n');190+ const spec = cols.map(([c, ty]) => `${lit(c)}: ${lit(ty)}`).join(', ');191+ await this.run(`INSERT INTO lake.${table} SELECT ${cols.map(([c]) => ident(c)).join(', ')} FROM read_json(${lit(file)}, format = 'newline_delimited', columns = {${spec}})`);192+ fs.rmSync(file, { force: true });193+ }194+}195+196+export class LakeError extends Error {}197+198+const firstLine = (e: unknown): string => String((e as Error)?.message ?? e).split('\n')[0];199+200+// ---------------------------------------------------------------- ingest201+202+export interface Summary {203+ events: number;204+ builds: number;205+ reviews: number;206+ ended: number;207+ ci: number;208+ bad: number;209+ notes: string[];210+ // Repos whose facts changed, for the exports.211+ touched: Set<string>;212+}213+214+interface SourceState extends Offset {215+ source: string;216+}217+218+const offsetOf = (s: SourceState | undefined): Offset | null => (s ? { ino: s.ino, pos: s.pos, head: s.head, tail: s.tail } : null);219+220+// Add what is new in every source. With dry, the same reads and inserts run221+// and are rolled back, so it says what a run would add.222+export async function ingest(lake: Lake, { dry = false, now = Date.now() }: { dry?: boolean; now?: number } = {}): Promise<Summary> {223+ const cfg = lake.cfg;224+ const sources = new Map<string, SourceState>();225+ for (const [source, ino, pos, head, tail] of await lake.rows(`SELECT source, ino, pos, head, tail FROM lake.sources`)) {226+ sources.set(String(source), { source: String(source), ino: Number(ino), pos: Number(pos), head: String(head ?? ''), tail: JSON.parse(String(tail ?? '[]')) });227+ }228+ const sum: Summary = { events: 0, builds: 0, reviews: 0, ended: 0, ci: 0, bad: 0, notes: [], touched: new Set() };229+ const next = new Map<string, Offset>();230+ const events: EventRow[] = [];231+ const builds: BuildRow[] = [];232+ const buildRepos: BuildRepoRow[] = [];233+ const reviews: ReviewRow[] = [];234+ const redo: [string, number][] = [];235+ const ends: ReviewEnd[] = [];236+ const ci: CiRow[] = [];237+238+ const jsonl = <T>(name: string, file: string, parse: (line: string) => T | null, take: (row: T) => void): void => {239+ const r: Read | null = readNew(file, offsetOf(sources.get(name)));240+ if (!r) return;241+ if (r.how === 'found') sum.notes.push(`${path.basename(file)} was cut or replaced; picked up where the last run left off`);242+ if (r.how === 'lost') sum.notes.push(`${path.basename(file)} was cut past where the last run left off; what was in between is gone`);243+ for (const line of r.lines) {244+ if (!line.trim()) continue;245+ const row = parse(line);246+ if (row) take(row);247+ else sum.bad++;248+ }249+ const was = sources.get(name);250+ if (!was || was.ino !== r.next.ino || was.pos !== r.next.pos || was.head !== r.next.head) next.set(name, r.next);251+ };252+ jsonl('events', path.join(cfg.state, 'events.jsonl'), eventRow, (e) => {253+ events.push(e);254+ if (e.repo) sum.touched.add(e.repo);255+ });256+ jsonl('metrics', path.join(cfg.state, 'metrics.jsonl'), buildRows, (b) => {257+ builds.push(b.build);258+ buildRepos.push(...b.repos);259+ sum.touched.add(b.build.scope);260+ });261+262+ // Review logs: a log that changed is read again whole (they are small).263+ // Whether a review ended is checked when its log or the default branch moved.264+ const ended = new Set((await lake.rows(`SELECT repo, review FROM lake.review_ends`)).map(([r, n]) => `${r} ${n}`));265+ let known: Map<string, Set<string>> | null = null;266+ // FORGE_NOW pins the clock in the demo and tests, as for review logs.267+ const nowS = Number(process.env.FORGE_NOW) || Math.floor(now / 1000);268+ const verifier = new Verifier({ allowedSigners: path.join(cfg.state, 'allowed_signers'), cacheFile: null });269+ for (const repo of listRepos(cfg.root)) {270+ const main = `main ${repo.slug}`;271+ const tip = tryRun(repo.dir, ['rev-parse', '-q', '--verify', 'HEAD^{commit}'])?.trim() ?? '';272+ const moved = sources.get(main)?.head !== tip;273+ if (moved) next.set(main, { ino: 0, pos: 0, head: tip, tail: [] });274+ for (const { log, hash } of reviewLogs(repo.dir)) {275+ const key = `review ${repo.slug} ${log.id}`;276+ const changed = sources.get(key)?.head !== hash;277+ if (changed) {278+ const rows = reviewRows(repo.slug, log);279+ reviews.push(...rows);280+ redo.push([repo.slug, log.id]);281+ next.set(key, { ino: 0, pos: 0, head: hash, tail: [] });282+ sum.reviews += rows.length;283+ sum.touched.add(repo.slug);284+ }285+ if ((changed || moved) && !ended.has(`${repo.slug} ${log.id}`)) {286+ const end = reviewEnd(repo.slug, repo.dir, log, nowS);287+ if (end) {288+ ends.push(end);289+ sum.touched.add(repo.slug);290+ }291+ }292+ }293+ const notes = notesTip(repo.dir);294+ const nkey = `notes ${repo.slug}`;295+ if (notes && sources.get(nkey)?.head !== notes) {296+ if (!known) {297+ known = new Map();298+ for (const [r, id] of await lake.rows(`SELECT repo, id FROM lake.ci`)) {299+ if (!known.has(String(r))) known.set(String(r), new Set());300+ known.get(String(r))!.add(String(id));301+ }302+ }303+ const rows = ciRows(repo.slug, repo.dir, known.get(repo.slug) ?? new Set(), verifier);304+ ci.push(...rows);305+ if (rows.length) sum.touched.add(repo.slug);306+ next.set(nkey, { ino: 0, pos: 0, head: notes, tail: [] });307+ }308+ }309+ sum.events = events.length;310+ sum.builds = builds.length;311+ sum.ended = ends.length;312+ sum.ci = ci.length;313+ if (!next.size) return sum;314+315+ const scratch = fs.mkdtempSync(path.join(cfg.dir, 'tmp.'));316+ const at = new Date(now).toISOString().replace('T', ' ').replace('Z', '');317+ try {318+ await lake.run('BEGIN TRANSACTION');319+ for (let i = 0; i < redo.length; i += 500) {320+ const keys = redo.slice(i, i + 500).map(([r, n]) => `(${lit(r)}, ${n})`).join(', ');321+ await lake.run(`DELETE FROM lake.reviews WHERE (repo, review) IN (${keys})`);322+ }323+ await lake.insert('events', events, scratch);324+ await lake.insert('builds', builds, scratch);325+ await lake.insert('build_repos', buildRepos, scratch);326+ await lake.insert('reviews', reviews, scratch);327+ await lake.insert('review_ends', ends, scratch);328+ await lake.insert('ci', ci, scratch);329+ const names = [...next.keys()];330+ for (let i = 0; i < names.length; i += 500) {331+ await lake.run(`DELETE FROM lake.sources WHERE source IN (${names.slice(i, i + 500).map(lit).join(', ')})`);332+ }333+ await lake.insert('sources', [...next].map(([source, o]) => ({ source, ino: o.ino, pos: o.pos, head: o.head, tail: JSON.stringify(o.tail), at })), scratch);334+ await lake.run(dry ? 'ROLLBACK' : 'COMMIT');335+ } catch (e) {336+ await lake.run('ROLLBACK').catch(() => {});337+ throw e;338+ } finally {339+ fs.rmSync(scratch, { recursive: true, force: true });340+ }341+ return sum;342+}343+344+// Inlined rows (DuckLake keeps inserts under 10 rows in the catalog) go to345+// Parquet, small files are merged and old snapshots dropped, about once a346+// day (FORGE_LAKE_CHECKPOINT hours).347+async function checkpoint(lake: Lake, now: number): Promise<boolean> {348+ const hours = Number(process.env.FORGE_LAKE_CHECKPOINT ?? 24);349+ const last = (await lake.rows(`SELECT pos FROM lake.sources WHERE source = 'checkpoint'`))[0]?.[0];350+ if (last != null && now / 1000 - Number(last) < hours * 3600) return false;351+ // A new lake starts the clock; its first rows are new files already.352+ if (last != null) await lake.run('CHECKPOINT lake');353+ await lake.run('BEGIN TRANSACTION');354+ await lake.run(`DELETE FROM lake.sources WHERE source = 'checkpoint'`);355+ await lake.run(`INSERT INTO lake.sources VALUES ('checkpoint', 0, ${Math.floor(now / 1000)}, '', '[]', now()::TIMESTAMP)`);356+ await lake.run('COMMIT');357+ return last != null;358+}359+360+// ---------------------------------------------------------------- query361+362+// A table, as the DuckDB CLI's plain mode does it.363+function table(cols: string[], rows: unknown[][]): string {364+ const cell = (v: unknown): string => (v == null ? 'NULL' : typeof v === 'object' ? JSON.stringify(v) : String(v)).replace(/\s+/g, ' ');365+ const text = [cols, ...rows.map((r) => r.map(cell))];366+ const w = cols.map((_, i) => Math.min(60, Math.max(...text.map((r) => r[i].length))));367+ const fmt = (r: string[]): string => r.map((c, i) => (c.length > w[i] ? c.slice(0, w[i] - 1) + '…' : c.padEnd(w[i]))).join(' ').trimEnd();368+ return [fmt(cols), w.map((n) => '-'.repeat(n)).join(' '), ...text.slice(1).map(fmt)].join('\n') + `\n(${rows.length} ${rows.length === 1 ? 'row' : 'rows'})\n`;369+}370+371+// Statements a query may run: reads.372+const READS = new Set([StatementType.SELECT, StatementType.EXPLAIN, StatementType.PRAGMA]);373+374+// Read-only, three ways: the lake is attached READ_ONLY; only SELECT,375+// EXPLAIN and PRAGMA statements run, so no COPY or EXPORT writes a file;376+// and file access stops at the lake's data, with settings locked, so a377+// query can't read other files, attach a database or undo either.378+export async function query(cfg: LakeConfig, sql: string, format: 'table' | 'json' | 'csv' = 'table'): Promise<string> {379+ const lake = await Lake.open(cfg, { write: false });380+ try {381+ await lake.run('USE lake');382+ if (cfg.local) await lake.run(`SET allowed_directories = [${lit(cfg.data)}]`);383+ await lake.run('SET enable_external_access = false');384+ await lake.run('SET lock_configuration = true');385+ const stmts = await lake.conn.extractStatements(sql);386+ if (!stmts.count) throw new LakeError('no SQL to run');387+ let res = null;388+ for (let i = 0; i < stmts.count; i++) {389+ const p = await stmts.prepare(i);390+ if (!READS.has(p.statementType)) throw new LakeError(`forge lake query only reads: ${StatementType[p.statementType]} statements don't run`);391+ res = await p.runAndReadAll();392+ }393+ if (!res) throw new LakeError('no SQL to run');394+ const cols = res.columnNames();395+ const rows = res.getRowsJson() as unknown[][];396+ if (format === 'json') return rows.map((r) => JSON.stringify(Object.fromEntries(cols.map((c, i) => [c, r[i]])))).join('\n') + (rows.length ? '\n' : '');397+ if (format === 'csv') {398+ const q = (v: unknown): string => {399+ const s = v == null ? '' : typeof v === 'object' ? JSON.stringify(v) : String(v);400+ return /[",\n]/.test(s) ? `"${s.replaceAll('"', '""')}"` : s;401+ };402+ return [cols, ...rows].map((r) => r.map(q).join(',')).join('\n') + '\n';403+ }404+ return table(cols, rows);405+ } finally {406+ lake.close();407+ }408+}409+410+// Every metrics.jsonl line the lake has, oldest first, for forge metrics;411+// null without a lake.412+export async function buildLines(cfg: LakeConfig): Promise<{ id: string; line: string }[] | null> {413+ if (!fs.existsSync(cfg.catalog)) return null;414+ const lake = await Lake.open(cfg, { write: false });415+ try {416+ return (await lake.rows(`SELECT id, line FROM lake.builds ORDER BY at, id`)).map(([id, line]) => ({ id: String(id), line: String(line) }));417+ } finally {418+ lake.close();419+ }420+}421+422+// ---------------------------------------------------------------- install423+424+// Install the extensions into the extension directory and check their425+// versions (the Dockerfile runs this, so the container never downloads one).426+async function install(cfg: LakeConfig): Promise<void> {427+ fs.mkdirSync(cfg.extensions, { recursive: true });428+ const db = await DuckDBInstance.create(':memory:', { extension_directory: cfg.extensions, autoinstall_known_extensions: 'false', autoload_known_extensions: 'false' });429+ const conn = await db.connect();430+ for (const n of ['ducklake', 'sqlite', 'httpfs']) await conn.run(`INSTALL ${n}`);431+ const got = (await conn.runAndReadAll(`SELECT extension_name, extension_version, install_path FROM duckdb_extensions() WHERE installed AND install_path NOT LIKE '(%'`)).getRowsJson() as string[][];432+ let bad = 0;433+ for (const [name, version, file] of got) {434+ if (!(name in EXTENSIONS)) continue;435+ const ok = EXTENSIONS[name] === version;436+ if (!ok) bad++;437+ console.log(`${name} ${version} ${(fs.statSync(file).size / (1 << 20)).toFixed(1)}M${ok ? '' : ` (expected ${EXTENSIONS[name]}; check it, then update EXTENSIONS in lib/lake.ts)`}`);438+ }439+ conn.closeSync();440+ db.closeSync();441+ if (bad) process.exit(1);442+}443+444+// ---------------------------------------------------------------- main445+446+const plural = (n: number, w: string): string => `${n} ${w}${n === 1 ? '' : 's'}`;447+448+async function main(argv: string[]): Promise<void> {449+ const cfg = lakeConfig();450+ const [cmd, ...rest] = argv;451+ if (cmd === 'ingest') {452+ const dry = rest.includes('--dry-run') || rest.includes('-n');453+ const t0 = performance.now();454+ const now = Date.now();455+ // A dry run on a forge without a lake yet reads into a scratch one.456+ const scratch = dry && !fs.existsSync(cfg.catalog) ? fs.mkdtempSync(path.join(os.tmpdir(), 'forge-lake.')) : null;457+ const lake = await Lake.open(scratch ? { ...cfg, dir: scratch, catalog: path.join(scratch, 'catalog.sqlite'), data: scratch + '/data/', local: true } : cfg, { write: true });458+ try {459+ const s = await ingest(lake, { dry, now });460+ const what = [plural(s.events, 'event'), plural(s.builds, 'build'), plural(s.reviews, 'review log line'), `${s.ended} ended`, plural(s.ci, 'attestation')].join(', ');461+ for (const n of s.notes) console.log(`lake: ${n}`);462+ if (s.bad) console.log(`lake: skipped ${plural(s.bad, 'line')} that aren't JSON`);463+ if (dry) {464+ console.log(`lake: would add ${what} (dry run: nothing written)`);465+ return;466+ }467+ const ck = await checkpoint(lake, now);468+ const size = du(cfg.dir);469+ console.log(`lake: added ${what}${ck ? '; checkpointed' : ''}; ${((performance.now() - t0) / 1000).toFixed(2)}s, ${(size / 1024).toFixed(0)}K on disk`);470+ } finally {471+ lake.close();472+ if (scratch) fs.rmSync(scratch, { recursive: true, force: true });473+ }474+ } else if (cmd === 'query') {475+ const fmt = rest.includes('--json') ? 'json' : rest.includes('--csv') ? 'csv' : 'table';476+ const sql = rest.filter((a) => a !== '--json' && a !== '--csv').join(' ');477+ if (!sql.trim()) throw new LakeError('usage: forge lake query "<sql>" [--json|--csv]');478+ process.stdout.write(await query(cfg, sql, fmt));479+ } else if (cmd === 'install') {480+ await install(cfg);481+ } else {482+ throw new LakeError('usage: forge lake [--dry-run] | forge lake query "<sql>" [--json|--csv]');483+ }484+}485+486+function du(p: string): number {487+ let n = 0;488+ try {489+ const st = fs.lstatSync(p);490+ if (!st.isDirectory()) return st.size;491+ for (const e of fs.readdirSync(p)) n += du(path.join(p, e));492+ } catch {}493+ return n;494+}495+496+if (process.argv[1] && fs.realpathSync(process.argv[1]) === fileURLToPath(import.meta.url)) {497+ main(process.argv.slice(2)).catch((e) => {498+ console.error(`forge: lake: ${e instanceof LakeError ? e.message : firstLine(e)}`);499+ process.exit(1);500+ });501+}@@ -7,6 +7,7 @@
77 "name": "forge",88 "dependencies": {99 "@11ty/eleventy": "^3.1.7",10+ "@duckdb/node-api": "^1.5.6-r.1",1011 "@sqlite.org/sqlite-wasm": "^3.53.4-build2",1112 "@typesafe-ai/sdk": "^0.7.0",1213 "highlight.js": "^11.12.0",@@ -270,6 +271,138 @@
270271 "integrity": "sha512-SzKYY68rseOrb9KILVzPsl6SjkQtdWw01SRGB5aCVI5kFcHltQsOYV3JB0Ffst72DgkvHCyDQ0NLwxtmCCf97Q==",271272 "license": "Apache-2.0"272273 },274+ "node_modules/@duckdb/node-api": {275+ "version": "1.5.6-r.1",276+ "resolved": "https://registry.npmjs.org/@duckdb/node-api/-/node-api-1.5.6-r.1.tgz",277+ "integrity": "sha512-yIPol6P4EQmsAUsb27donJf7iADrM31Ei/WkvhkZpSscjhEErYsuF19lsJ/qz5mUWxsKFQpLdjvrlhPQnMTBqw==",278+ "license": "MIT",279+ "dependencies": {280+ "@duckdb/node-bindings": "1.5.6-r.1"281+ }282+ },283+ "node_modules/@duckdb/node-bindings": {284+ "version": "1.5.6-r.1",285+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings/-/node-bindings-1.5.6-r.1.tgz",286+ "integrity": "sha512-/pWlTyvKC1d16iSMJCBAPwlmzoeXRs5Y9cqG+TpKd/YAW6Ke05FJg099v/Im2bkS9BgznNyNcAK1NxA1eQhSUw==",287+ "license": "MIT",288+ "dependencies": {289+ "detect-libc": "^2.1.2"290+ },291+ "optionalDependencies": {292+ "@duckdb/node-bindings-darwin-arm64": "1.5.6-r.1",293+ "@duckdb/node-bindings-darwin-x64": "1.5.6-r.1",294+ "@duckdb/node-bindings-linux-arm64": "1.5.6-r.1",295+ "@duckdb/node-bindings-linux-arm64-musl": "1.5.6-r.1",296+ "@duckdb/node-bindings-linux-x64": "1.5.6-r.1",297+ "@duckdb/node-bindings-linux-x64-musl": "1.5.6-r.1",298+ "@duckdb/node-bindings-win32-arm64": "1.5.6-r.1",299+ "@duckdb/node-bindings-win32-x64": "1.5.6-r.1"300+ }301+ },302+ "node_modules/@duckdb/node-bindings-darwin-arm64": {303+ "version": "1.5.6-r.1",304+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-darwin-arm64/-/node-bindings-darwin-arm64-1.5.6-r.1.tgz",305+ "integrity": "sha512-KFT/J/pyouuAD1jftAIfcWVRdorg+c/yVR/BJL8LJKrST+mYC4d7QSuQ+iXSYbFpHENUAeZUagXrMPtZT8Xzsg==",306+ "cpu": [307+ "arm64"308+ ],309+ "license": "MIT",310+ "optional": true,311+ "os": [312+ "darwin"313+ ]314+ },315+ "node_modules/@duckdb/node-bindings-darwin-x64": {316+ "version": "1.5.6-r.1",317+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-darwin-x64/-/node-bindings-darwin-x64-1.5.6-r.1.tgz",318+ "integrity": "sha512-tVwUZNvdQoaPJhrh3q2e6jxm1MhahNHJQK/pvCyco46NX0mEuGzlw9i0n2MArgsyzccdCwScOK9ctbyr+3oMbQ==",319+ "cpu": [320+ "x64"321+ ],322+ "license": "MIT",323+ "optional": true,324+ "os": [325+ "darwin"326+ ]327+ },328+ "node_modules/@duckdb/node-bindings-linux-arm64": {329+ "version": "1.5.6-r.1",330+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-linux-arm64/-/node-bindings-linux-arm64-1.5.6-r.1.tgz",331+ "integrity": "sha512-3jnUaDil1BrF/6vAZwVaInJ1xoBaWGHB4PDiUnzJWKCo8a8TLx9j7/gPyPl/UGRBftrYIuTihRZSfF4mWMgtWA==",332+ "cpu": [333+ "arm64"334+ ],335+ "license": "MIT",336+ "optional": true,337+ "os": [338+ "linux"339+ ]340+ },341+ "node_modules/@duckdb/node-bindings-linux-arm64-musl": {342+ "version": "1.5.6-r.1",343+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-linux-arm64-musl/-/node-bindings-linux-arm64-musl-1.5.6-r.1.tgz",344+ "integrity": "sha512-j6dcNLWnFvlQyKMtl36HyeDsOaetXFoj0dH21Q9MbLjABNuJs5ekDV6zMSEda/g5+z1Tm6r2KFOS8j3tGUCRiw==",345+ "cpu": [346+ "arm64"347+ ],348+ "license": "MIT",349+ "optional": true,350+ "os": [351+ "linux"352+ ]353+ },354+ "node_modules/@duckdb/node-bindings-linux-x64": {355+ "version": "1.5.6-r.1",356+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-linux-x64/-/node-bindings-linux-x64-1.5.6-r.1.tgz",357+ "integrity": "sha512-IXcrZJE6kEqM0Z4wwdVnGBHX/frOS33VPNAwDXQ/Q+Fs84b058Qe94chrCCIcNJlG8Y3UHIUU4eBeCQxW5mN1Q==",358+ "cpu": [359+ "x64"360+ ],361+ "license": "MIT",362+ "optional": true,363+ "os": [364+ "linux"365+ ]366+ },367+ "node_modules/@duckdb/node-bindings-linux-x64-musl": {368+ "version": "1.5.6-r.1",369+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-linux-x64-musl/-/node-bindings-linux-x64-musl-1.5.6-r.1.tgz",370+ "integrity": "sha512-PVzIEZqp7QN07WBgjmIH0cK2GKXR735KmL6wnghTzRnhHWrePxx3MjZB9zZHI8tB73wdF8luarBm4EgqSjGyVA==",371+ "cpu": [372+ "x64"373+ ],374+ "license": "MIT",375+ "optional": true,376+ "os": [377+ "linux"378+ ]379+ },380+ "node_modules/@duckdb/node-bindings-win32-arm64": {381+ "version": "1.5.6-r.1",382+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-win32-arm64/-/node-bindings-win32-arm64-1.5.6-r.1.tgz",383+ "integrity": "sha512-nzNTZx9rus5QpuX+KH6UOPNeZp8yF25TxV6DiUly4RKgR5mu6xo8c7a7lS80AqxTJfJ9BwUIiqlCuYKWztRM5A==",384+ "cpu": [385+ "arm64"386+ ],387+ "license": "MIT",388+ "optional": true,389+ "os": [390+ "win32"391+ ]392+ },393+ "node_modules/@duckdb/node-bindings-win32-x64": {394+ "version": "1.5.6-r.1",395+ "resolved": "https://registry.npmjs.org/@duckdb/node-bindings-win32-x64/-/node-bindings-win32-x64-1.5.6-r.1.tgz",396+ "integrity": "sha512-5Rpn5WXEG5gY/u7Keh+c4kuwG/LiE31pCP9CoCohv+u0aFensk/yRIQVW/CXuIoITky9xFhdRzaWUtAnGTYxJQ==",397+ "cpu": [398+ "x64"399+ ],400+ "license": "MIT",401+ "optional": true,402+ "os": [403+ "win32"404+ ]405+ },273406 "node_modules/@iconify/types": {274407 "version": "2.0.0",275408 "resolved": "https://registry.npmjs.org/@iconify/types/-/types-2.0.0.tgz",@@ -1829,6 +1962,15 @@
18291962 "node": ">=4"18301963 }18311964 },1965+ "node_modules/detect-libc": {1966+ "version": "2.1.2",1967+ "resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz",1968+ "integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==",1969+ "license": "Apache-2.0",1970+ "engines": {1971+ "node": ">=8"1972+ }1973+ },18321974 "node_modules/dom-serializer": {18331975 "version": "1.4.1",18341976 "resolved": "https://registry.npmjs.org/dom-serializer/-/dom-serializer-1.4.1.tgz",@@ -10,6 +10,8 @@
1010 # TYPESAFE_API_KEY for hooks and builds (written to .forge/secrets)1111 # FORGE_BACKUP_*, FORGE_MIRROR_* backups and the pages mirror (written to1212 # .forge/backup.env; README.md, "Backups and moving hosts")13+# FORGE_LAKE_* the lake's data on R2 instead of the volume (also14+# .forge/backup.env; README.md, "The lake and insights")1315 set -eu1416 root=/srv/git1517 forge=/opt/forge/bin/forge@@ -73,6 +75,11 @@ FORGE_MIRROR_SECRET=$(q "${FORGE_MIRROR_SECRET:-}")
7375 FORGE_MIRROR_PROVIDER=$(q "${FORGE_MIRROR_PROVIDER:-}")7476 FORGE_MIRROR_REGION=$(q "${FORGE_MIRROR_REGION:-}")7577 FORGE_MIRROR_DELAY=$(q "${FORGE_MIRROR_DELAY:-}")78+FORGE_LAKE_DATA=$(q "${FORGE_LAKE_DATA:-}")79+FORGE_LAKE_KEY_ID=$(q "${FORGE_LAKE_KEY_ID:-}")80+FORGE_LAKE_SECRET=$(q "${FORGE_LAKE_SECRET:-}")81+FORGE_LAKE_ENDPOINT=$(q "${FORGE_LAKE_ENDPOINT:-}")82+FORGE_LAKE_REGION=$(q "${FORGE_LAKE_REGION:-}")7683 EOF7784 7885 if [ -n "${FORGE_OWNER:-}" ] && [ -n "${FORGE_OWNER_KEY:-}" ] && ! grep -q "^$FORGE_OWNER " $root/.forge/keys; then@@ -18,6 +18,7 @@
1818 # release_commit w v1 dir [key] a release commit of dir's files, to push to refs/releases/v11919 # aside dami/secret ... back dami/secret move a repo out of the forge and back2020 # sync_builds hooks and ssh commands build pages before returning21+# lq "<sql>" forge lake query: the rows as CSV, no header2122 #2223 # Assertions print what failed and stop the scenario:2324 # ok cmd... cmd succeeds fails cmd... cmd fails@@ -32,13 +33,15 @@ T=$(mktemp -d "${TMPDIR:-/tmp}/forge-scenario.XXXXXX")
3233 T=$(CDPATH= cd -- "$T" && pwd -P)3334 # Processes a scenario starts in the background go in T_PIDS, to be stopped.3435 T_PIDS=35-trap 'st=$?; [ -z "$T_PIDS" ] || kill $T_PIDS 2>/dev/null; [ -n "${FORGE_TEST_KEEP:-}" ] && echo "kept $T" >&2 || rm -rf "$T"; exit $st' EXIT36+trap 'st=$?; [ -z "$T_PIDS" ] || kill $T_PIDS 2>/dev/null || true; [ -n "${FORGE_TEST_KEEP:-}" ] && echo "kept $T" >&2 || rm -rf "$T"; exit $st' EXIT3637 3738 export FORGE_ROOT=$T/root3839 export FORGE_WWW=$FORGE_ROOT/.forge/www3940 export FORGE_SSH=git@forge.test4041 export FORGE_NO_BUILD=14142 export FORGE_JEV=043+# Scenarios run forge lake themselves, not in the background after builds.44+export FORGE_LAKE=04245 export GIT_CONFIG_GLOBAL=$T/gitconfig4346 export GIT_CONFIG_NOSYSTEM=14447 export FORGE_CACHE=$T/cache@@ -216,3 +219,8 @@ back() { mv "$T/aside.git" "$FORGE_ROOT/$1.git"; }
216219 217220 # From here on, what hooks and ssh commands rebuild is built before they return.218221 sync_builds() { export FORGE_NO_BUILD=0 FORGE_SYNC_BUILD=1; }222+223+lq() {224+ "$FORGE" lake query "$1" --csv >"$T/lq" 2>"$T_LOG" || fail "forge lake query failed: $1"225+ sed 1d "$T/lq"226+}@@ -26,6 +26,10 @@ for s in "$here"/scenarios/*.sh; do
2626 done2727 [ -n "$list" ] || { echo "no scenarios match: $*" >&2; exit 1; }2828 29+# The lake's DuckDB extensions go into a cache shared by every scenario;30+# fetch them once here rather than in several scenarios at the same time.31+node "$here/../lib/lake.ts" install >/dev/null 2>&1 || true32+2933 # One scenario: its result line, and its output kept for a failure.3034 export out3135 printf '%s' "$list" | xargs -P "$jobs" -I{} sh -c '@@ -0,0 +1,29 @@
1+# CI attestations land in ci, one row each: commit and tree, result,2+# duration, runner, and the signer once the signature checks out. Notes3+# that didn't move are skipped; a new attestation adds one row.4+. "$(dirname "$0")/../lib.sh"5+user dami6+repo dami/greet7+clone dami dami/greet "$T/w/d"8+mkdir -p "$T/w/d/.forge"9+printf '#!/bin/sh\ntest -f ok\n' >"$T/w/d/.forge/ci"10+chmod +x "$T/w/d/.forge/ci"11+at "$T/w/d" commit "ci script, no ok file yet"12+fails at "$T/w/d" "$FORGE" attest13+at "$T/w/d" commit "add ok" ok14+ok at "$T/w/d" "$FORGE" attest15+ok push_as dami "$T/w/d" origin main refs/notes/ci16+ok forge lake17+has "$T_LOG" ', 2 attestations;'18+pass=$(git -C "$T/w/d" rev-parse HEAD)19+eq "$(lq "SELECT string_agg(result, ' ' ORDER BY result) FROM ci")" "fail pass"20+eq "$(lq "SELECT \"commit\" || ' ' || tree FROM ci WHERE result = 'pass'")" "$pass $(git -C "$T/w/d" rev-parse HEAD^{tree})"21+eq "$(lq "SELECT count(*) FROM ci WHERE verified AND signer = 'dami' AND runner LIKE '%@%' AND duration_s >= 0 AND exit = (result = 'fail')::INT AND started IS NOT NULL")" 222+ok forge lake23+has "$T_LOG" ', 0 attestations;'24+at "$T/w/d" commit "more"25+ok at "$T/w/d" "$FORGE" attest26+ok push_as dami "$T/w/d" origin main refs/notes/ci27+ok forge lake28+has "$T_LOG" ', 1 attestation;'29+eq "$(lq "SELECT count(*) || ' ' || count(DISTINCT id) FROM ci")" "3 3"@@ -0,0 +1,28 @@
1+# A second forge lake adds nothing, and a later one adds only what is new2+# since: every source remembers where it got to.3+. "$(dirname "$0")/../lib.sh"4+user dami5+repo dami/greet6+clone dami dami/greet "$T/w/d"7+at "$T/w/d" commit "first"8+push_as dami "$T/w/d" origin main9+at "$T/w/d" commit "idea"10+push_as dami "$T/w/d" origin HEAD:dami/idea11+build dami/greet12+ok forge lake13+counts() { lq "SELECT (SELECT count(*) FROM events) || ' ' || (SELECT count(*) FROM builds) || ' ' || (SELECT count(*) FROM reviews)"; }14+eq "$(counts)" "3 1 2"15+ok forge lake16+has "$T_LOG" '^lake: added 0 events, 0 builds, 0 review log lines, 0 ended, 0 attestations;'17+eq "$(counts)" "3 1 2"18+# A revision: one push event and the review's log read again, whole (its19+# old rows are replaced, not repeated); one more build.20+at "$T/w/d" git commit -q --amend -m "idea, again"21+push_as dami "$T/w/d" -f origin HEAD:dami/idea22+build dami/greet23+ok forge lake24+has "$T_LOG" '^lake: added 2 events, 1 build, 3 review log lines,'25+eq "$(counts)" "5 2 3"26+eq "$(lq "SELECT string_agg(rev::VARCHAR, ' ' ORDER BY rev) FROM reviews WHERE kind = 'rev'")" "1 2"27+ok forge lake28+eq "$(counts)" "5 2 3"@@ -0,0 +1,32 @@
1+# forge lake adds the forge's facts to the lake: events (pushes, comments,2+# builds), build metrics with a row per repo, and review log lines.3+. "$(dirname "$0")/../lib.sh"4+user dami5+user ana6+repo dami/greet7+clone dami dami/greet "$T/w/d"8+at "$T/w/d" commit "first"9+push_as dami "$T/w/d" origin main10+clone ana dami/greet "$T/w/a"11+at "$T/w/a" commit "idea"12+push_as ana "$T/w/a" origin HEAD:ana/idea13+echo "looks fine" | ssh_as dami comment dami/greet 1 >"$T_LOG" 2>&1 || fail "comment failed"14+build dami/greet15+ok forge lake16+has "$T_LOG" '^lake: added 4 events, 1 build, 2 review log lines, 0 ended, 0 attestations;'17+exists "$FORGE_ROOT/.forge/lake/catalog.sqlite"18+exists "$FORGE_ROOT/.forge/lake/data"19+eq "$(lq "SELECT string_agg(type, ' ' ORDER BY type) FROM events")" "built comment push push"20+eq "$(lq "SELECT repo || ' ' || \"user\" FROM events WHERE type = 'comment'")" "dami/greet dami"21+eq "$(lq "SELECT scope || ' ' || reason FROM builds")" "dami/greet manual"22+eq "$(lq "SELECT repo FROM build_repos")" "dami/greet"23+eq "$(lq "SELECT string_agg(kind, ' ' ORDER BY ts, kind) FROM reviews WHERE review = 1")" "created rev"24+eq "$(lq "SELECT \"user\" || ' ' || branch FROM reviews WHERE kind = 'created'")" "ana ana/idea"25+# Event times are the clock the hooks stamp (FORGE_NOW here).26+eq "$(lq "SELECT epoch(min(t))::BIGINT FROM events WHERE type = 'push'")" "$T_NOW"27+# A dry run says what it would add and adds nothing.28+at "$T/w/d" commit "second"29+push_as dami "$T/w/d" origin main30+ok forge lake --dry-run31+has "$T_LOG" 'would add 1 event, 0 builds'32+eq "$(lq "SELECT count(*) FROM events")" 4@@ -0,0 +1,41 @@
1+# One writer at a time, under lake.lock: a second forge lake waits for the2+# first, or gives up after FORGE_LAKE_WAIT seconds and says so. A lock whose3+# process is gone is taken over. Two runs at once add each fact once.4+. "$(dirname "$0")/../lib.sh"5+user dami6+repo dami/greet7+clone dami dami/greet "$T/w/d"8+at "$T/w/d" commit "first"9+push_as dami "$T/w/d" origin main10+lock=$FORGE_ROOT/.forge/lake.lock11+hold() { sleep "$1" & holder=$!; T_PIDS="$T_PIDS $holder"; mkdir "$lock"; echo "$holder" >"$lock/pid"; }12+hold 30013+fails_with "another forge lake has held $lock for 1s" env FORGE_LAKE_WAIT=1 "$FORGE" lake14+missing "$FORGE_ROOT/.forge/lake/catalog.sqlite"15+# Its holder died: the next run takes the lock over.16+kill "$holder"17+wait "$holder" 2>/dev/null || true18+ok forge lake19+has "$T_LOG" '^lake: added 1 event,'20+missing "$lock"21+# A run that finds the lock taken waits for it.22+hold 30023+(sleep 2 && rm -rf "$lock") &24+t0=$(date +%s)25+ok env FORGE_LAKE_WAIT=30 "$FORGE" lake26+[ $(($(date +%s) - t0)) -ge 2 ] || fail "forge lake didn't wait for the lock"27+kill "$holder"28+# Two at once: one adds the new events, the other finds nothing left.29+ev=$FORGE_ROOT/.forge/events.jsonl30+i=031+while [ "$i" -lt 50 ]; do32+ i=$((i + 1))33+ printf '{"t":%s000,"repo":"dami/greet","type":"push","user":"dami","refs":"refs/heads/n%s"}\n' "$((T_NOW + i))" "$i" >>"$ev"34+done35+"$FORGE" lake >"$T/a" 2>&1 & a=$!36+"$FORGE" lake >"$T/b" 2>&1 & b=$!37+wait "$a" || fail "the first of two forge lakes failed: $(cat "$T/a")"38+wait "$b" || fail "the second of two forge lakes failed: $(cat "$T/b")"39+eq "$(cat "$T/a" "$T/b" | grep -c '^lake: added 50 events,')" 140+eq "$(cat "$T/a" "$T/b" | grep -c '^lake: added 0 events,')" 141+eq "$(lq "SELECT count(*) || ' ' || count(DISTINCT refs) FROM events WHERE refs LIKE 'refs/heads/n%'")" "50 50"@@ -0,0 +1,39 @@
1+# forge lake query runs SQL over the lake and only reads: writes, COPY,2+# ATTACH, settings and files outside the lake are refused, and the lake is3+# the same afterwards.4+. "$(dirname "$0")/../lib.sh"5+user dami6+repo dami/greet7+fails_with 'no lake yet' forge lake query "SELECT 1"8+clone dami dami/greet "$T/w/d"9+at "$T/w/d" commit "first"10+push_as dami "$T/w/d" origin main11+at "$T/w/d" commit "second"12+push_as dami "$T/w/d" origin main13+ok forge lake14+ok forge lake query "SELECT repo, count(*) AS pushes FROM events GROUP BY repo"15+has "$T_LOG" '^repo +pushes$'16+has "$T_LOG" '^dami/greet +2$'17+has "$T_LOG" '^\(1 row\)$'18+ok forge lake query "SELECT type FROM events LIMIT 1" --json19+has "$T_LOG" '^\{"type":"push"\}$'20+# Tables of the lake are there to read, and DuckLake's own functions.21+ok forge lake query "SHOW TABLES"22+has "$T_LOG" '^review_ends$'23+ok forge lake query "SELECT count(*) > 0 AS ok FROM lake.snapshots()"24+has "$T_LOG" '^true$'25+sum=$(cksum <"$FORGE_ROOT/.forge/lake/catalog.sqlite")26+fails_with 'only reads: INSERT' forge lake query "INSERT INTO events (type) VALUES ('x')"27+fails_with 'only reads: DELETE' forge lake query "DELETE FROM events"28+fails_with 'only reads: CREATE' forge lake query "CREATE TABLE x (a INT)"29+fails_with 'only reads: COPY|Permission Error' forge lake query "COPY (SELECT 1) TO '$T/out.csv'"30+fails_with 'only reads: ATTACH|Permission Error' forge lake query "ATTACH '$T/other.db' AS other"31+fails_with 'only reads: SET' forge lake query "SET enable_external_access = true"32+fails_with 'only reads: DELETE' forge lake query "SELECT 1; DELETE FROM events"33+fails_with 'Permission Error' forge lake query "SELECT * FROM read_text('$FORGE_ROOT/.forge/keys')"34+fails_with 'Permission Error' forge lake query "SELECT * FROM read_csv('$FORGE_ROOT/.forge/events.jsonl')"35+fails_with 'only reads: SET' forge lake query "PRAGMA memory_limit = '1GB'"36+missing "$T/out.csv"37+missing "$T/other.db"38+eq "$(cksum <"$FORGE_ROOT/.forge/lake/catalog.sqlite")" "$sum"39+eq "$(lq "SELECT count(*) FROM events")" 2@@ -0,0 +1,59 @@
1+# Review lifecycles land in reviews (created, each revision, approvals,2+# deleted) and review_ends: landed by ssh land at its logged deletion,3+# closed when deleted unlanded, landed when main took its tip while the4+# branch stayed (at the time the lake saw it); an open one hasn't ended.5+. "$(dirname "$0")/../lib.sh"6+user dami7+user ana8+repo dami/greet9+clone dami dami/greet "$T/w/dami"10+at "$T/w/dami" commit "first"11+push_as dami "$T/w/dami" origin main12+clone ana dami/greet "$T/w/ana"13+t0=$T_NOW14+# #1: two revisions, an approval, landed by ssh land two hours in.15+at "$T/w/ana" git switch -q -c ana/feature16+at "$T/w/ana" commit "feature"17+push_as ana "$T/w/ana" origin ana/feature18+tick 180019+at "$T/w/ana" git commit -q --amend -m "feature, again"20+push_as ana "$T/w/ana" -f origin ana/feature21+tick 180022+ok ssh_as dami approve dami/greet 123+tick 360024+ok ssh_as dami land dami/greet 125+# #2: pushed, then deleted.26+at "$T/w/ana" git fetch -q origin27+at "$T/w/ana" git switch -q -c ana/abandon origin/main28+at "$T/w/ana" commit "abandon"29+push_as ana "$T/w/ana" origin ana/abandon30+tick 60031+push_as ana "$T/w/ana" origin :ana/abandon32+# #3: main is pushed to its tip and the branch stays.33+at "$T/w/dami" git pull -q origin main34+at "$T/w/dami" git switch -q -c dami/kept35+at "$T/w/dami" commit "kept"36+push_as dami "$T/w/dami" origin dami/kept37+tick 90038+push_as dami "$T/w/dami" origin dami/kept:main39+# #4: still open.40+at "$T/w/dami" commit "open"41+push_as dami "$T/w/dami" origin HEAD:dami/open42+tick 6043+ok forge lake44+has "$T_LOG" ', 3 ended,'45+eq "$(lq "SELECT string_agg(review || ':' || kind, ' ' ORDER BY review, ts, kind) FROM reviews")" \46+ "1:created 1:rev 1:rev 1:approve 1:deleted 2:created 2:rev 2:deleted 3:created 3:rev 4:created 4:rev"47+eq "$(lq "SELECT string_agg(rev::VARCHAR, ' ' ORDER BY rev) FROM reviews WHERE review = 1 AND kind IN ('rev', 'approve')")" "1 2 2"48+eq "$(lq "SELECT \"user\" FROM reviews WHERE review = 1 AND kind = 'approve'")" "dami"49+eq "$(lq "SELECT string_agg(review || ':' || kind || ':' || observed, ' ' ORDER BY review) FROM review_ends")" "1:landed:false 2:closed:false 3:landed:true"50+# Time to land: from created to the deletion ssh land logged.51+eq "$(lq "SELECT (epoch(e.ts) - epoch(r.ts))::INT FROM review_ends e JOIN reviews r USING (repo, review) WHERE review = 1 AND r.kind = 'created'")" 720052+eq "$(lq "SELECT epoch(ts)::BIGINT FROM review_ends WHERE review = 1")" "$((t0 + 7200))"53+eq "$(lq "SELECT epoch(ts)::BIGINT FROM review_ends WHERE review = 3")" "$T_NOW"54+# An ended review stays ended as it was; the open one ends when it lands.55+tick 60056+ok ssh_as dami land dami/greet 457+ok forge lake58+eq "$(lq "SELECT string_agg(review || ':' || kind, ' ' ORDER BY review) FROM review_ends")" "1:landed 2:closed 3:landed 4:landed"59+eq "$(lq "SELECT epoch(ts)::BIGINT FROM review_ends WHERE review = 3")" "$((T_NOW - 600))"@@ -0,0 +1,47 @@
1+# cmd_build cuts events.jsonl to its last 1000 lines once it passes 1 MB,2+# into a new file. The lake picks up after the last line it read: what it3+# had is kept, nothing is added twice, and lines appended before the cut4+# that it hadn't read yet still arrive. A cut past where it had got to says5+# so; only the lines that were gone by then are missing.6+. "$(dirname "$0")/../lib.sh"7+user dami8+repo dami/greet9+clone dami dami/greet "$T/w/d"10+at "$T/w/d" commit "first"11+push_as dami "$T/w/d" origin main12+ev=$FORGE_ROOT/.forge/events.jsonl13+pad=$(printf '%0360d' 0)14+# events from to: synthetic pushes, about 400 bytes each, numbered.15+events() {16+ awk -v a="$1" -v b="$2" -v p="$pad" 'BEGIN { for (i = a; i <= b; i++) printf "{\"t\":%d000,\"repo\":\"dami/greet\",\"type\":\"push\",\"user\":\"dami\",\"refs\":\"refs/heads/n%d %s\"}\n", 1788220800 + i, i, p }' >>"$ev"17+}18+synthetic() { lq "SELECT count(*) || ' ' || count(DISTINCT refs) FROM events WHERE refs LIKE 'refs/heads/n%'"; }19+events 1 300020+ok forge lake21+has "$T_LOG" '^lake: added 3001 events,'22+# 200 more, not read yet, then a build cuts the file.23+events 3001 320024+ino=$(ls -i "$ev" | awk '{ print $1 }')25+build dami/greet26+[ "$(ls -i "$ev" | awk '{ print $1 }')" != "$ino" ] || fail "the build didn't cut events.jsonl"27+eq "$(wc -l <"$ev" | tr -d ' ')" 100028+ok forge lake29+has "$T_LOG" 'events.jsonl was cut or replaced; picked up where the last run left off'30+has "$T_LOG" '^lake: added 201 events, 1 build,'31+eq "$(synthetic)" "3200 3200"32+eq "$(lq "SELECT count(*) FROM events WHERE type = 'built'")" 133+# Appended to after the cut, the new file is read on from where it was.34+events 3201 321035+ok forge lake36+has "$T_LOG" '^lake: added 10 events,'37+eq "$(synthetic)" "3210 3210"38+# Now a cut past the last line read: 2000 lines arrive and the build keeps39+# the last 1000 of them (with its own event), so the first 1001 are gone.40+events 3211 521041+build dami/greet42+ok forge lake43+has "$T_LOG" 'events.jsonl was cut past where the last run left off; what was in between is gone'44+has "$T_LOG" '^lake: added 1000 events,'45+eq "$(synthetic)" "4209 4209"46+n="CAST(regexp_extract(refs, 'n([0-9]+)', 1) AS INTEGER)"47+eq "$(lq "SELECT count(*) FILTER ($n BETWEEN 3211 AND 4211) || ' ' || count(*) FILTER ($n BETWEEN 4212 AND 5210) FROM events WHERE refs LIKE 'refs/heads/n%'")" "0 999"