nordisk/nornir

crates/mimir-ingest/examples/consume.rs

//! Run the consumer with the real driver on one repo and print one line per
//! commit (seq, status, crates, +, −, ms, phase split).
//!
//! ```text
//! cargo run -p mimir-ingest --example consume -- \
//!     <name> <source> <branch> <state-dir> <cargo-mimir-dump> <nightly> [initial-offset-sha]
//!     [--with-binary] [--profile dev|release|…] [--std] [--reingest-binary]
//! ```
//! `<state-dir>` holds the lake, the clone, the driver target dir and the
//! open-set cache; reruns resume from the stored offset.
//!
//! * `--with-binary`: `cargo mimir-dump --build` (ONE compile) + mimir-binary
//!   over every local bin → the binary stream, in the same atomic release.
//! * `--std`: first make sure the toolchain's std / core / alloc are in the
//!   universe (dumped once per toolchain, `<state-dir>/std`), and join them in.
//! * `--reingest-binary`: only the binary layer at the current offset (its own
//!   atomic release); no commits are consumed.
//! * the provenance split of the repo (derived) is printed at the end.

use std::path::PathBuf;

use anyhow::{Context, Result};
use mimir_ingest::{BinaryLayer, Consumer, DriverDumper, EventLog, RepoSpec, Specs, StdDumper};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
    let mut a: Vec<String> = Vec::new();
    let (mut with_binary, mut std_dump, mut reingest, mut profile) = (false, false, false, "dev".to_string());
    let mut it = std::env::args().skip(1);
    while let Some(x) = it.next() {
        match x.as_str() {
            "--with-binary" => with_binary = true,
            "--std" => std_dump = true,
            "--reingest-binary" => reingest = true,
            "--profile" => profile = it.next().context("--profile needs a value")?,
            _ => a.push(x),
        }
    }
    if a.len() < 6 {
        anyhow::bail!("usage: consume <name> <source> <branch> <state-dir> <cargo-mimir-dump> <nightly> [initial-offset] [--with-binary] [--profile P] [--std] [--reingest-binary]");
    }
    let state = PathBuf::from(&a[3]);
    std::fs::create_dir_all(&state)?;
    let wh = skade::open(state.join("lake")).await?;
    let mut log = EventLog::open(wh, Specs::mimir()?).await?;
    let mut std_dir = None;
    if std_dump {
        let sd = StdDumper { cargo_mimir_dump: PathBuf::from(&a[4]), driver: None, toolchain: a[5].clone(), root: state.join("std"), host: Default::default() };
        let o = sd.ensure(&mut log).await?;
        eprintln!("std dump: {:?} ({} crates, {} rows, release {:?}) in {} ms → {}", o.status, o.crates, o.rows, o.rustc_release, o.ms, o.dir.display());
        std_dir = Some(o.dir);
    }
    let mut dumper = DriverDumper::new(PathBuf::from(&a[4]), a[5].clone(), state.join("target"), state.join("out"), a[0].clone());
    dumper.binary = (with_binary || reingest).then(|| BinaryLayer { profile: profile.clone(), std_dump: std_dir.clone() });
    let t0 = std::time::Instant::now();
    let mut c = Consumer::boot(log, dumper).await?;
    eprintln!("boot in {} ms", t0.elapsed().as_millis());
    let spec = RepoSpec {
        name: a[0].clone(),
        source: a[1].clone(),
        branch: a[2].clone(),
        worktree: state.join("clone"),
        initial_offset: a.get(6).cloned(),
    };
    if reingest {
        for r in c.reingest_binaries(&spec).await? {
            println!("rev {:>3} seq {:>3} {} {:<40} {:<6} +{:<7} -{:<7} {:>7} ms", r.rev, r.seq, &r.sha[..10], r.artifact, r.status, r.rows_plus, r.rows_minus, r.ms);
        }
    } else {
        let result = c.run(&spec).await;
        for r in c.log().runs(&spec.name).await? {
            println!(
                "seq {:>3} {} {:<18} crates {:>4} +{:<7} -{:<7} dup {:<4} span +{:<6} -{:<6} owners# {:<3} {:>7} ms {}",
                r.seq,
                &r.sha[..10],
                r.status,
                r.crates_dumped,
                r.rows_plus,
                r.rows_minus,
                r.key_collisions,
                r.span_plus,
                r.span_minus,
                r.owner_disambiguations,
                r.ms,
                r.error.as_deref().map(|e| e.lines().next().unwrap_or_default()).unwrap_or_default()
            );
        }
        let reports = result.context("consume")?;
        for r in &reports {
            println!(
                "phases seq {:>3}: files written {} deleted {} {:?} binary rev {:?} +{} -{}",
                r.seq, r.files_written, r.files_deleted, r.phase_ms, r.bin_rev, r.bin_plus, r.bin_minus
            );
        }
        for r in c.log().bin_runs(&spec.name).await? {
            println!("bin rev {:>3} seq {:>3} {:<40} {:<6} +{:<7} -{:<7} {:>7} ms", r.rev, r.seq, r.artifact, r.status, r.rows_plus, r.rows_minus, r.ms);
        }
    }
    let t1 = std::time::Instant::now();
    let p = c.log().provenance(&spec.name, None).await?;
    eprintln!("provenance (derived, {} ms): {:?}", t1.elapsed().as_millis(), p.counts());
    Ok(())
}