nordisk/nornir

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 (the two-stage gatling pipeline — the authority), src/knowledge/symbols.rs (scan_file / scan_content, the per-file parse unit), and 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:

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 SymbolScans 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.