<!-- nornir:class: design -->
# Deep-scan pipeline
Reference + design for **deep-scan** — the sweep that knowledge-scans a
workspace's *entire transitive dependency closure* into the warehouse, so the
viz 🕸 Call Graph and the `knowledge_*` MCP tools answer over every crate the
workspace pulls in (deps of deps, the whole tree), not just the workspace
members.
**Scope — this is opt-in and separate from the member scan.** deep-scan runs
**only for a workspace with `deep_scan = true`** in its descriptor (e.g. holger).
It is the heavy path: holger's closure is **~1749 crates / ~50k files**, scanned
in **≈9s** (was ≈33s before this design). A workspace *without* `deep_scan`
(the default) republishes only its **own member repos'** knowledge — unchanged
members ⇒ ≈0.1–1s, a fresh full member scan ⇒ proportional to the members'
*own* source (not their deps). The member scan (`scan_repo`) is a distinct path
from this dependency-closure path; it is **no longer serial** — its `syn` parse
fans across all cores (`parse_files_parallel`, a heaviest-first/LPT, batched,
no-barrier worker pool — the in-repo mirror of zoomies
`gatling_for_each_balanced`; ≈3.3× under load, identical output). deep-scan below
is the dependency-closure path.
Source code: [`src/deepscan.rs`](../src/deepscan.rs) (the two-stage gatling
pipeline — **the authority**), [`src/knowledge/symbols.rs`](../src/knowledge/symbols.rs)
(`scan_file` / `scan_content`, the per-file parse unit), and
[`src/warehouse/iceberg.rs`](../src/warehouse/iceberg.rs)
(`ingest_symbol_scans_async` + `build_groups_parallel`, the all-core flush into skade).
The whole pipeline is **pure Rust, zero C FFI** and runs under the **mimalloc**
global allocator (per-thread heaps — set in the server/CLI bins so 32-worker
columnization doesn't serialize on the glibc arena lock).
## 1. What it is
`deep_scan(member_dirs, wh)` does four things, in order:
```
resolve closure → Stage 1 DECOMPRESS → flat Vec<CrateFile> → Stage 2 PARSE → parallel FLUSH → warehouse
Cargo.lock 32 gatling workers crate · tar_path 32 gatling workers build_groups_parallel symbol_facts
in-process, lgz, per-crate · content (.rs) per-FILE syn parse → skade::ingest_parallel call_edges
no shell (~0s) (all cores build+encode) feature_gate_facts
```
Rows land under `repo = "deps"` (`DEPS_REPO`), tagged with **one** snapshot id
for the whole run, so the closure shows up as a single repo `deps` with one
crate entry per dependency.
## 2. The two-stage shape — parse is FILE-granular
Deep-scan splits **decompress** from **parse** and makes parse fan out one file
at a time:
- **Stage 1 (decompress)** is crate-granular — a cargo `.crate` is a single gzip
stream, so one crate decompresses on one worker. The huge single-stream
`windows-*` gzips therefore decompress on *separate* workers **concurrently**.
- **Stage 2 (parse)** is **one file per work unit**. A monster crate's thousands
of generated `.rs` files spread across all 32 cores, so there is no serial
tail — every core stays busy until the last file is parsed. (A crate-granular
parse would pin one core per crate and let the `windows-*` giants stall the
sweep while the rest idle.)
That is why the pipeline materializes a **flat `Vec<CrateFile>`** between the
stages: it is the unit boundary that lets parse fan out file-by-file regardless
of which crate a file came from. `scan_file` rows self-tag with `crate_name`, so
the flat list needs *no* per-crate reassembly downstream.
## 3. The gatling no-barrier engine
Both stages run on **znippy's gatling engine** (`znippy_zoomies::gatling::run_typed`),
the no-barrier worker-pool streamer. The shape is always **1 reader → N workers
→ 1 sink**:
- the **reader** streams the input as newline-delimited text in small chunks
(`chunk_size: 8 KiB`, `ring_slots: 8`, `SlotFill::Incremental`) — there is *no*
whole-input single-chunk barrier;
- a shared `line_split` codec cuts **one segment per complete line** (a partial
trailing line becomes carry and waits for the next chunk);
- each **worker** runs `transform` on one segment, entirely independent of its
siblings;
- the **sink** sees outputs **in stream order** on the single collector thread
(single writer, no lock war).
**No barrier** means a worker grabs the next segment the *instant* it finishes
its current one — it never waits for siblings to reach a checkpoint. A fast file
and a slow file simply pull different numbers of units; no core idles waiting on
a phase boundary.
`workers = std::thread::available_parallelism()` (32 on the box). **No `rayon`
anywhere in the scan path** — gatling owns the parallelism.
### Stage 1 — DECOMPRESS (`DecompressCodec` + `CollectSink`)
```
input : one "label\tdir\n" line per crate
split : line_split → one segment per crate line
transform(seg): # decompress_crate(label, dir)
tarball = registry/src/.../<name-ver>/ → registry/cache/.../<name-ver>.crate
if tarball exists:
bytes = read(tarball) # the .crate gzip
entries = lgz::decompress_tar_gz_filter(bytes, ".rs") # zero-copy, ".rs" only
return [ CrateFile{ crate_name=label, tar_path=path, content } for (path, bytes) in entries ]
else: # path/git dep, no .crate
walk <dir>/{src,tests,benches,examples} for *.rs
return [ CrateFile{ label, tar_path="<label>/<sub>/<rel>", content } ... ]
sink : CollectSink — files.extend(out) # flat Vec<CrateFile> across ALL crates
```
`lgz` is the home-made parallel zero-copy gunzip. A cargo `.crate` is a *single*
gzip stream, so lgz falls back to single-threaded *per crate* — but it is fast
(≈824 MB/s; the 123 MB `windows` crate decompresses in ≈0.15s), and crates run
on different workers concurrently.
### Stage 2 — PARSE (`ParseCodec<'a>` + `WarehouseSink`)
```
input : one file INDEX per line — "0\n1\n2\n…" for 0..files.len()
codec : ParseCodec holds &files, the run snapshot id, the run ts
split : line_split → one segment per index
transform(seg):
idx = parse(seg) as usize
f = files[idx] # ONE file
return symbols::scan_file(f.crate_name, f.tar_path, f.content, "deps", snap, ts)
# scan_file: strip "<name-ver>/" → rel; section must be src|tests|benches|examples;
# syn-parse content → SymbolScan{ symbols, calls, features }
sink : WarehouseSink.process(Ok(scan)):
push scan into `pending` IF it has any symbols|calls|features (skip empties)
WarehouseSink.process(Err(e)):
report.errors.push(e) # one bad file never stops the sweep
# NO flush here — deep_scan flushes explicitly so parse and flush time separately
```
`scan_file` is the per-file parse unit; it returns a `SymbolScan` with three row
vectors:
| field | row type | warehouse table |
|------------|-----------------|----------------------|
| `symbols` | `SymbolRow` | `symbol_facts` |
| `calls` | `CallEdgeRow` | `call_edges` |
| `features` | `FeatureGateRow`| `feature_gate_facts` |
## 4. The all-core flush
`WarehouseSink::flush()` calls `IcebergWarehouse::ingest_symbol_scans` (sync
wrapper over `ingest_symbol_scans_async`). The parse workers never block on a
commit — they only buffer `SymbolScan`s into `pending`; the flush runs once per
crate-batch (and once at the end) and is itself **all-core**, in two parts:
```
ingest_symbol_scans_async(scans) -> IngestStats: # scans = the buffered per-FILE SymbolScans
n_groups = available_parallelism() # ~N_CORES
chunk = ceil(scans.len() / n_groups) # balanced bucket size
tables = []
for each table in [symbol_facts, call_edges, feature_gate_facts, test_inventory]:
# (1) COLUMNIZE IN PARALLEL — build_groups_parallel:
# std::thread::scope, N workers, each grabs the next bucket index
# from a shared AtomicUsize (gatling-style, no rayon) and columnizes
# its bucket's rows into ONE Arrow RecordBatch. Returns batches in
# bucket order. This is the step that used to be the single-core floor.
groups = build_groups_parallel(scans, chunk, n_groups, |bucket| columnize(bucket))
tables.push((table, groups)) # gather — DON'T commit yet
# (2) ENCODE + ONE CROSS-TABLE COMMIT — skade encodes every table's parquet
# across ALL cores, stages each table's fast_append, and flips ALL of them
# in a SINGLE redb transaction = ONE fsync (was 4 separate commits).
stats = skade::ingest_multi(catalog, tables, WriteProps::new(warehouse zstd))
# stats.commits == 1 for a non-empty flush.
```
So per flush there is **one parquet file per bucket per table**, and the whole
four-table flush is **one commit / one fsync** (not one commit per table) —
via skade's cross-table `ingest_multi` (the multi-table populate fast lane
that stages every table's `fast_append` and flips them in one all-or-nothing
redb txn). The data files skade writes are **byte-identical** to the old
per-table `ingest_parallel_with` path's; only the commit collapses **4 → 1**.
Two distinct all-core stages: **`build_groups_parallel`** columnizes (the
String→Arrow copy spread over every core), then **`skade::ingest_multi`**
encodes on every core + does the single staged commit.
> **fsync law (RED-when-broken):** `deepscan_multi_table_flush_is_one_fsync_and_lands_every_table`
> (in `warehouse::iceberg` tests) ingests one scan that populates all four
> tables and asserts `IngestStats.commits == 1` **and** exactly one row landed
> per table — reverting to the per-table commit loop makes `commits == 4` and
> the test goes red.
Both halves allocate heavily (millions of small Arrow strings), so the
**mimalloc** global allocator is what lets them actually scale — under the stock
glibc allocator the 32 columnize workers serialize on the arena lock (~2× instead
of ~32×). Measured on holger's closure: **arrow-build ≈1.2s | skade-ingest ≈1s**
(the flush was ≈9s single-core before the parallel build + mimalloc).
`A/B probe`: `NORNIR_DEEPSCAN_NOSAVE=1` skips the write entirely to isolate
parse cost from flush cost.
## 5. The driver loop (`deep_scan`)
```
deep_scan(member_dirs, wh):
closure = resolve_dep_closure(member_dirs) # in-process Cargo.lock, no shell; ~1749 crates ~0.0s
present = [ (label, dir) for (label, dir) in closure if dir.is_dir() ]
skipped = closure.len() - present.len() # .crate not on disk / git dep
workers = available_parallelism() # 32
snap = new uuid # ONE snapshot id for the whole run
ts = now
for batch in present.chunks(BATCH): # BATCH = 1_000_000 → effectively one batch
# Stage 1
crate_input = "".join(f"{label}\t{dir}\n" for (label, dir) in batch)
collect = CollectSink{}
gatling::run_typed(crate_input, DecompressCodec, collect, workers, scan_cfg())
files = collect.files # flat Vec<CrateFile>
if files empty: continue
# Stage 2
idx_input = "".join(f"{i}\n" for i in 0..files.len())
gatling::run_typed(idx_input, ParseCodec{files, snap, ts}, sink, workers, scan_cfg())
# FLUSH this batch
sink.flush()
sink.flush() # final
report.crates = present.len()
return report
```
One batch is intentional: the box has ~500 GB RAM and a closure's decompressed
source is only a few GB, so decompressing all crates together is what lets the
big `windows-*` gzips overlap on separate workers, and the whole scan does a
single bulk commit. `BATCH` exists only to bound peak memory if that ever
changes.
Timing is reported split — the driver prints
`⏱ resolve_dep_closure … {N} crate(s) in {t}s` and then
`⏱ decompress {a}s | parse {b}s | flush {c}s | total {d}s ({files} files, {crates} crates)`
so each stage is measured independently.
## 6. Laws this pipeline upholds
- **No shelling.** The closure is resolved by reading each member's `Cargo.lock`
in-process — no `cargo fetch`/`cargo metadata` subprocess. The pipeline is pure
Rust end-to-end (lgz=`flate2`/`miniz_oxide`, `syn`, arrow/parquet `UNCOMPRESSED`
so no zstd C, redb); no C FFI on the hot path.
- **No rayon in the scan path.** Both gatling stages and the parallel columnize
(`build_groups_parallel`, `std::thread::scope`) use plain worker pools; the only
`rayon` is *inside* skade during `ingest_parallel`, behind the warehouse boundary.
- **Zero-copy / zero-allocations is the goal; mimalloc is the floor-lifter.**
`lgz` decompresses `.crate` gzips zero-copy; `syn::parse_file` borrows the source
(zero-copy input); the flush hands Arrow to skade which recasts to parquet
without re-buffering. Where allocation is unavoidable (columnizing rows into
Arrow), the **mimalloc** global allocator keeps 32-worker allocation from
serializing on the system allocator.
- **No-barrier gatling.** A worker grabs the next unit the instant it finishes —
it never waits for siblings. File-level parse means even a monster crate's
files keep all 32 cores busy to the very end.
- **Persist to the warehouse, not /tmp.** Results land in the Iceberg tables
(`symbol_facts`, `call_edges`, `feature_gate_facts`); the viz and MCP read from
there. (After deep-scan the caller republishes snapshots.)
## 7. Tests
`src/deepscan.rs` + `src/warehouse/iceberg.rs` carry the matrix entries that
exercise the real pipeline (inject real input, assert real output):
- `resolve_dep_closure_reads_cargo_lock_in_process_excludes_path_members` — a
synthetic `Cargo.lock` + fake `CARGO_HOME` registry: registry deps (a `source`)
are included and mapped to their src dir; path/workspace members (no `source`)
are excluded. Proves the in-process, no-shell resolve.
- `line_split_*` — the shared splitter cuts one segment per complete line and
carries a partial tail.
- `decompress_transform_isolates_bad_lines` — a malformed crate line yields no
files, not a panic.
- `decompress_then_scan_file_finds_symbols` — `decompress_crate` (dir fallback) →
`scan_file` actually finds the injected `fn`.
- `deep_scan_two_stage_end_to_end_into_warehouse` — two synthetic dep crates
through the *real* two-stage gatling pipeline (decompress dir-fallback → parse
files → ingest) into a temp warehouse, asserting both files decompress and the
scan has no errors.
- `build_groups_parallel_preserves_order_skips_empty_loses_nothing`
(`iceberg.rs`) — the parallel columnize keeps bucket order, skips empty
buckets, and loses nothing across 1/4/8/32 workers.
Test scratch lands under `/home/rickard/workX` (the repo's `.cargo/config.toml`
pins `TMPDIR`), never the T9 mount the checkout lives on.