nordisk/nornir

crates/mimir-ingest/tests/ingest/compact_beside.rs

//! A compaction running beside the consumer never cuts a run short: every
//! commit the run set out to consume is consumed, whatever the lake lane
//! commits meanwhile. (The release loaded its table handles before taking
//! the log's commit lock, so a replace landing in between failed its
//! catalog commit with a conflict — an "infrastructure" error that stopped
//! the run after fewer commits than it had. nornir main 60cbd9af.)

use std::path::Path;

use mimir_ingest::{CompactOptions, Consumer, RepoSpec};

use crate::common::{FakeDumper, Scratch, open_log};

const REPO: &str = "r";
const COMMITS: usize = 12;

fn spec(tmp: &Path, src: &Scratch) -> RepoSpec {
    RepoSpec {
        name: REPO.into(),
        source: src.root.to_string_lossy().into_owned(),
        branch: "main".into(),
        worktree: tmp.join("clone"),
        initial_offset: None,
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_compaction_beside_the_consumer_never_cuts_a_run_short() {
    let tmp = tempfile::tempdir().unwrap();
    let src = Scratch::new(tmp.path().join("src"));
    for i in 0..COMMITS {
        src.write("crates/a/src/lib.rs", &format!("fn f (x: i32) | v{i}\nfn g () | two\ncall g -> f\n"));
        src.commit(&format!("c{i}"));
    }
    let log = open_log(&tmp.path().join("lake")).await;
    let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();

    // the lake lane: compaction after compaction while the run goes
    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let lane = {
        let (log, stop) = (log.clone(), stop.clone());
        tokio::spawn(async move {
            let opts = CompactOptions { grace: std::time::Duration::ZERO, ..Default::default() };
            let mut runs = 0usize;
            while !stop.load(std::sync::atomic::Ordering::Acquire) {
                log.compact(&opts).await.unwrap();
                runs += 1;
            }
            runs
        })
    };
    let outcome = c.run(&spec(tmp.path(), &src)).await;
    stop.store(true, std::sync::atomic::Ordering::Release);
    let compactions = lane.await.unwrap();
    let reports = outcome.unwrap_or_else(|e| panic!("the run stopped ({compactions} compactions meanwhile): {e:#}"));
    assert_eq!(reports.len(), COMMITS, "every commit consumed ({compactions} compactions meanwhile)");
    assert!(reports.iter().all(|r| r.status == "ok"), "{:?}", reports.iter().map(|r| (r.seq, r.status.clone())).collect::<Vec<_>>());
    let (log, _) = c.into_parts();
    assert_eq!(log.offset(REPO).await.unwrap().map(|o| o.seq), Some(COMMITS as i64));
    assert!(log.runs(REPO).await.unwrap().iter().all(|r| r.status == "ok"), "no error row");
    assert!(compactions >= 1, "the lane ran");
}