nordisk/knut

crates/knut-popviz/tests/bulk_multicore.rs

//! **Headless proof the GRAPH.BULK fill is multicore and survives scale.**
//!
//! Drives `run_load(Target::Falkor, IngestMode::Bulk)` — the exact path the
//! Tools ▸ demo-zone fill takes — against a running FalkorDB, at a scale whose
//! dominant label used to exceed FalkorDB's `proto-max-bulk-len` (512MB) as ONE
//! blob and die with `Broken pipe` while pinning a single core. With the chunked,
//! gatling-parallel encoder it now (a) splits into many <512MB blobs (no broken
//! pipe) and (b) fans the encode across cores — the `graphar-falkordb bulk:
//! encoded N blob(s) across up to M core(s)` line on stderr shows M > 1.
//!
//! Run (FalkorDB already up on :6379; bump the scale for the full 32-core run):
//! ```bash
//! KNUT_DEMO_SCALE=0.1 cargo test -p knut-popviz --test bulk_multicore \
//!   -- --ignored --nocapture
//! ```
//! `KNUT_DEMO_SCALE=1.0` is the "max people" run that previously crashed.

use std::sync::{Arc, Mutex};

use knut_popviz::{IngestMode, LoadParams, PopvizState, Target, run_load};

#[test]
#[ignore = "needs a running FalkorDB on :6379; run with --ignored --nocapture"]
fn bulk_fill_is_multicore_and_survives_scale() {
    let scale: f64 = std::env::var("KNUT_DEMO_SCALE")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(0.1);
    let url =
        std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into());

    eprintln!("── bulk fill @ scale {scale} into {url} (graph `multicore_demo`) ──");
    let params = LoadParams {
        url,
        graph: "multicore_demo".to_string(),
        scale,
        drop: true, // empty graph ⇒ the GRAPH.BULK fast lane
        target: Target::Falkor,
        ingest: IngestMode::Bulk, // pin the bulk lane (the one that crashed)
        tls: None,
    };
    let state = Arc::new(Mutex::new(PopvizState::new(scale)));

    let t0 = std::time::Instant::now();
    run_load(params, Arc::clone(&state), || {});
    let elapsed = t0.elapsed();

    let s = state.lock().unwrap();
    assert!(
        s.error.is_none(),
        "bulk fill errored (broken pipe regressed?): {:?}",
        s.error
    );
    let nodes: usize = s.labels.iter().map(|l| l.loaded).sum();
    let rels: usize = s.rels.iter().map(|r| r.loaded).sum();
    assert!(nodes > 0, "the fill loaded nodes");
    eprintln!(
        "✓ bulk fill OK — {nodes} nodes + {rels} relations at scale {scale} in {elapsed:?}, \
         NO broken pipe. (See the `encoded N blob(s) across up to M core(s)` line above for the \
         core count; bump KNUT_DEMO_SCALE for more blobs = more cores.)"
    );
}