crates/knut-popviz/tests/falkor_parity.rs
//! **Live byte-parity gate for knut-popviz's FalkorDB bulk fast lane.**
//!
//! `load_falkor` auto-selects the native `GRAPH.BULK` fast lane for a fresh /
//! `--drop`ped graph (see `lib.rs`). This test loads the SAME synthetic
//! population **both ways** into one live FalkorDB container — pinning
//! [`IngestMode::Bulk`] for one graph and [`IngestMode::Unwind`] for the other —
//! and asserts the two graphs are **byte-identical**: same nodes, same `_gar_id`
//! (the global, offset-based id), same per-label properties (incl. the Swedish
//! geo columns + the `ekonomiskt_bistand` flag), and the same `Person -> entity`
//! relationships. This is the proof that popviz's `_gar_id` dedup + zero-offset
//! bulk-edge reconciliation lands the identical graph the proven UNWIND path does.
//!
//! RED-when-broken: any encoder/reconciliation drift (a duplicated or 0-based
//! `_gar_id`, a mis-translated edge endpoint, a lost geo column) makes the two
//! ordered dumps diverge.
//!
//! Container-gated (`#[ignore]`, needs Podman + a FalkorDB image via the
//! `autostart` feature). Run:
//! ```bash
//! cargo test -p knut-popviz --features autostart --test falkor_parity -- --ignored --nocapture
//! ```
//!
//! # KNOWN: this test intermittently HANGS at high core counts (2026-08-01)
//!
//! Measured on oden (32 cores) against `localhost/knut-falkordb-valkey9`:
//!
//! | client `concurrency` | runs | hung |
//! |---------------------|------|------|
//! | 32 (all cores) | 4 | 3 |
//! | 16 (`taskset -c 0-15`) | 1 | 0 |
//! | 8, 4, 1 | 3 | 0 |
//!
//! A passing run finishes in **1.2–2.8 s**; a hung run sat 24 min in the original
//! report and 3 min in the reproduction, both with the test process at **0:00
//! CPU**. `load_falkor_unwind` sets `concurrency = available_parallelism()`, so
//! the core count IS the variable.
//!
//! It is not the server. While hung, FalkorDB had spent 103 ms total across 37
//! `graph.QUERY` calls and its client entry read:
//!
//! ```text
//! id=43 age=182 idle=182 qbuf=163227 qbuf-free=603 cmd=graph.QUERY lib-name=redis-rs lib-ver=0.27.6
//! ```
//!
//! — 163 KB of a **partially received** command, no new bytes for 182 s. The
//! client stopped writing mid-command while awaiting the reply to it; the server
//! cannot answer a command it has not finished reading. All 66 client threads were
//! parked (`futex_do_wait`, one on `ep_poll`).
//!
//! It always stalls in the UNWIND **vertex** phase of the second graph (2072 of
//! 6060 nodes created, 0 edges), so it never reaches the edge phase — this is a
//! DIFFERENT defect from the rows-sent-vs-created one in `graphar-falkordb`, not
//! the same root cause.
//!
//! Inferred, not proven: many cloned `redis::aio::MultiplexedConnection` handles
//! concurrently writing ~163 KB commands onto the one shared socket leave a
//! partial command on the wire. The exact redis-rs frame was NOT determined —
//! async futures are not recoverable from a gdb backtrace here. Fixing it means
//! choosing a concurrency cap or a per-window connection, which is a design call
//! and is deliberately not made here.
//!
//! # 2026-08-22 — the mechanism, and what a day of measuring did NOT show
//!
//! STINKY-19 filed R22 as *"observed, NOT diagnosed"*. Two things are now known
//! that were not, and they point in opposite directions — both are recorded,
//! because the convenient half alone would be a nicer story than the truth.
//!
//! ## Proven, from the dependency's source: the driver can starve its own reads
//!
//! `MultiplexedConnection`'s driver task is `stream.forward(PipelineSink)`
//! (`redis-0.27.6/src/aio/multiplexed_connection.rs:353`). In that file
//! `PipelineSink::poll_read` — **the only** place a reply is parsed off the
//! socket and the waiting `oneshot` completed — has exactly **one** call site:
//! line 321, the last statement of `PipelineSink::poll_flush`. And
//! `futures_util`'s `Forward::poll` reaches `poll_flush` on exactly one branch —
//! the one where the request stream returns `Pending`:
//!
//! ```text
//! if buffered_item.is_some() {
//! ready!(si.as_mut().poll_ready(cx))?; // <-- Pending returns HERE,
//! si.as_mut().start_send(...)?; // without ever calling poll_flush
//! }
//! match stream.poll_next(cx)? {
//! Pending => { ready!(si.poll_flush(cx))?; return Poll::Pending } // the only drain
//! ...
//! }
//! ```
//!
//! So while the client has an unbroken backlog of commands to send, the driver
//! never drains the read side. `buffer_unordered(concurrency)` with
//! `concurrency = available_parallelism()` produces exactly that backlog, and
//! `tokio_util`'s `Framed` carries an **8 KiB** backpressure boundary against
//! these ~163 KB rendered commands, so `poll_ready` must reach the socket on
//! essentially every send. The moment the socket stops accepting bytes,
//! `poll_ready` is `Pending`, `poll_read` is never called again, and the two
//! sides wait on each other with no deadline: `get_multiplexed_async_connection`
//! leaves `response_timeout` at `None`, so `send_recv` awaits its reply oneshot
//! forever. That is consistent with every symptom above — a partially received
//! command idle for 182 s, a server with nothing to do, and 66 parked threads.
//!
//! This is a fact about the code as written, checkable by anyone in five
//! minutes. It is **not** a claim that it is what happened here.
//!
//! ## NOT reproduced — and that narrows the cause more than a repro would have
//!
//! Driving the parallel path directly against a live FalkorDB (valkey 9.1.0,
//! graph module v4.2001) with no containers, at this test's own shape —
//! `insert_vertices_par`, `concurrency = 32`, 512-row windows, 391 windows,
//! popviz-width rows — completed **6 of 6** runs, 200 000 nodes each in
//! 15.3–24.4 s, 1.2 M rows total, zero stalls. A 2000-row-window variant
//! (60 000 rows) also completed, in 336 ms.
//!
//! So `insert_vertices_par` at `concurrency = 32` is **not sufficient** on its
//! own, which the 2026-08-01 table (32 cores → 3 of 4 hung) reads as if it were.
//! Whatever else is required lives in what this test does and that probe did
//! not: two graphs loaded back-to-back into one server, twelve labels with an
//! index phase between them, `run_load` owning its own runtime, and a
//! testcontainers-started container rather than a standing one. Note the recorded
//! stall is always in the **second** graph.
//!
//! A sampling of the client socket during those 1.2 M rows caught a 170-byte
//! reply sitting unread 5 times in ~200 two-second samples, never in consecutive
//! samples. That is far too thin to call evidence of sustained read starvation,
//! and it is written down here so nobody later mistakes it for the smoking gun.
//! It is not one.
//!
//! ## What was done about it
//!
//! Nothing that changes production behaviour, because the fix is still a design
//! call (cap the concurrency, or give each in-flight window its own connection so
//! each driver's stream goes `Pending` after every send and `poll_read` always
//! runs). What landed instead is
//! `FalkorDbLoader::with_response_timeout` — opt-in, `None` by default, so no
//! shipped path changes — plus `graphar-falkordb/tests/robot_response_timeout.rs`,
//! which drives a socket that accepts and never answers and proves both halves:
//! with a ceiling the call is refused, and **without** one the identical socket is
//! still pending after 2 s. A wedge is now something a caller can choose to be
//! told about, rather than 24 minutes of a machine that looks busy.
// NOT `#![cfg(feature = "autostart")]`. The whole file used to be gated behind that
// feature, which meant `parity_verdict`'s own red-proof below did not run in a default
// `cargo test` — it reported "0 passed" and looked exactly like a clean result. Only the
// container-driven tests need the feature, so only they are gated.
#![allow(clippy::items_after_test_module)]
#[cfg(feature = "autostart")]
use std::sync::{Arc, Mutex};
#[cfg(feature = "autostart")]
use graphar_falkordb::FalkorDbLoader;
#[cfg(feature = "autostart")]
use graphar_falkordb::falkordb_container::start_falkordb;
#[cfg(feature = "autostart")]
use knut_popviz::{IngestMode, LABELS, LoadParams, PERSON_REL, PopvizState, Target, run_load};
#[cfg(feature = "autostart")]
/// The GRAPH.QUERY reply's row block (element `[1]`) — strips the header and the
/// per-run timing stats trailer so two runs are comparable.
fn rows(reply: &redis::Value) -> redis::Value {
match reply {
redis::Value::Array(items) | redis::Value::Set(items) if items.len() >= 2 => {
items[1].clone()
}
other => other.clone(),
}
}
#[cfg(feature = "autostart")]
/// Every node property popviz can write, ordered by the global `_gar_id`. A
/// label without a given column simply returns `null` in BOTH graphs, so one
/// superset projection compares every label faithfully.
fn node_query(label: &str) -> String {
format!(
"MATCH (n:`{label}`) \
RETURN n._gar_id, n.idx, n.name, n.lon, n.lat, n.kommun, n.ekonomiskt_bistand \
ORDER BY n._gar_id"
)
}
#[cfg(feature = "autostart")]
/// The `Person -[rel]-> dst` endpoints, ordered — the relationship's shape.
fn edge_query(rel: &str, dst: &str) -> String {
format!(
"MATCH (a:Person)-[r:`{rel}`]->(b:`{dst}`) \
RETURN a._gar_id, b._gar_id ORDER BY a._gar_id, b._gar_id"
)
}
/// The edge-parity verdict: the two lanes must agree **and** must have loaded
/// something.
///
/// Split out of the test body so it is testable WITHOUT a container, because the
/// assertion it replaces could not fail on the graph shape that matters. The test
/// used to say only `assert_eq!(bulk["edges_loaded"], unwind["edges_loaded"])`,
/// and `0 == 0` satisfies that: two lanes that both wrote no edges at all
/// reported byte-identical parity. That is the live `korp_demo` shape (121 200
/// nodes, 0 edges) passing a test named for edge parity. The byte dump below it
/// was equally blind — two empty result sets compare equal.
///
/// Proven red by `zero_edges_is_not_parity` below, which is the case the old
/// assertion accepted.
fn parity_verdict(bulk_edges: u64, unwind_edges: u64) -> Result<u64, String> {
if bulk_edges != unwind_edges {
return Err(format!(
"edge count differs between lanes: bulk={bulk_edges} unwind={unwind_edges}"
));
}
if bulk_edges == 0 {
return Err(
"both lanes loaded ZERO edges — equal, but not a parity proof. The population \
always has Person->entity relationships, so 0 means the edge phase wrote \
nothing (the korp_demo shape: many nodes, no edges)"
.to_string(),
);
}
Ok(bulk_edges)
}
/// **The verdict is not blind.** `0 == 0` was accepted by the assertion this
/// replaces, so that exact input must now be rejected — and a real load must
/// still pass, so the guard is not merely always-red.
#[test]
fn zero_edges_is_not_parity() {
let err = parity_verdict(0, 0).expect_err("two empty lanes must NOT be parity");
assert!(err.contains("ZERO edges"), "names the defect: {err}");
assert_eq!(
parity_verdict(5510, 5510).expect("a real, equal load passes"),
5510,
"the measured count at scale 0.00005 on 2026-08-01"
);
let differ = parity_verdict(5510, 0).expect_err("unequal lanes must fail");
assert!(
differ.contains("bulk=5510 unwind=0"),
"reports both sides: {differ}"
);
}
#[cfg(feature = "autostart")]
/// Drive a full `run_load` (blocking — it owns its tokio runtime) into `graph`
/// with the given ingest lane and return the final introspection `state_json`.
fn drive(url: &str, graph: &str, scale: f64, ingest: IngestMode) -> serde_json::Value {
let state = Arc::new(Mutex::new(PopvizState::new(scale)));
let params = LoadParams {
url: url.to_string(),
graph: graph.to_string(),
scale,
drop: true,
target: Target::Falkor,
ingest,
tls: None,
};
run_load(params, Arc::clone(&state), || {});
let json = state.lock().unwrap().state_json();
assert_eq!(json["phase"], "done", "{graph} load reached done — {json}");
assert!(
json["error"].is_null(),
"{graph} load had no error — {json}"
);
json
}
#[cfg(feature = "autostart")]
#[test]
#[ignore = "needs podman + a FalkorDB image; run with --features autostart -- --ignored --nocapture"]
fn popviz_bulk_matches_unwind_byte_identical() {
// A dedicated runtime for the async container bring-up + read-back dumps.
// `run_load` is called OUTSIDE any `block_on` (it builds its own runtime), so
// there is no nested-runtime hazard.
let rt = tokio::runtime::Runtime::new().expect("tokio runtime");
let container = rt
.block_on(start_falkordb())
.expect("start FalkorDB container");
let url = container.url().to_string();
let scale = 0.00005_f64;
let bulk = drive(&url, "popviz_parity_bulk", scale, IngestMode::Bulk);
let unwind = drive(&url, "popviz_parity_unwind", scale, IngestMode::Unwind);
// Coarse agreement first (a fast, legible failure before the byte dump).
assert_eq!(
bulk["loaded"], unwind["loaded"],
"same vertex count both lanes"
);
assert!(bulk["loaded"].as_u64().unwrap() > 0, "non-empty population");
// Equality ALONE was the hole: `0 == 0` is agreement, and a graph with no
// edges at all is the exact live `korp_demo` shape this suite exists to catch.
// The floor is what makes the comparison mean something (see `parity_verdict`).
parity_verdict(
bulk["edges_loaded"].as_u64().unwrap(),
unwind["edges_loaded"].as_u64().unwrap(),
)
.expect("edge parity");
// The byte-parity dump: every label's nodes, every rel's edges, both graphs.
rt.block_on(async {
let mut b = FalkorDbLoader::connect(&url, "popviz_parity_bulk")
.await
.expect("connect bulk graph");
let mut u = FalkorDbLoader::connect(&url, "popviz_parity_unwind")
.await
.expect("connect unwind graph");
for (label, _) in LABELS.iter() {
let q = node_query(label);
let bn = rows(&b.run_query(&q).await.unwrap());
let un = rows(&u.run_query(&q).await.unwrap());
assert_eq!(bn, un, "label {label}: bulk nodes must byte-match UNWIND");
}
for li in 1..LABELS.len() {
let q = edge_query(PERSON_REL[li], LABELS[li].0);
let be = rows(&b.run_query(&q).await.unwrap());
let ue = rows(&u.run_query(&q).await.unwrap());
assert_eq!(
be, ue,
"rel {} (Person -> {}): bulk edges must byte-match UNWIND",
PERSON_REL[li], LABELS[li].0
);
}
eprintln!(
"\n=== knut-popviz bulk↔UNWIND parity @ scale {scale} ===\n\
nodes: {} · edges: {} · byte-identical across all 12 labels + 11 rels\n",
bulk["loaded"], bulk["edges_loaded"],
);
});
// testcontainers's async Drop needs an ambient Tokio runtime; enter `rt` for it.
let _enter = rt.enter();
drop(container);
}
/// Container-gated **bench arm**: popviz's whole-population load via the native
/// `GRAPH.BULK` fast lane vs the parallel-`UNWIND` path, at a demo-ish scale,
/// printing wall-clock + nodes/s for each (host-stamped plain numbers). Not an
/// assert-heavy test — it prints and only sanity-checks both loaded the same
/// count. Run with `--ignored --nocapture`.
#[cfg(feature = "autostart")]
#[test]
#[ignore = "needs podman; bench — run with --features autostart -- --ignored --nocapture"]
fn bench_popviz_bulk_vs_unwind() {
let rt = tokio::runtime::Runtime::new().expect("tokio runtime");
let container = rt
.block_on(start_falkordb())
.expect("start FalkorDB container");
let url = container.url().to_string();
let scale = 0.0005_f64; // ~5.5k Person + entities — a visible demo fill
let bulk = drive(&url, "popviz_bench_bulk", scale, IngestMode::Bulk);
let unwind = drive(&url, "popviz_bench_unwind", scale, IngestMode::Unwind);
assert_eq!(
bulk["loaded"], unwind["loaded"],
"both lanes loaded the same count"
);
let f = |j: &serde_json::Value| {
(
j["loaded"].as_u64().unwrap(),
j["elapsed_s"].as_f64().unwrap(),
j["rows_per_s"].as_f64().unwrap(),
)
};
let (bn, bt, brps) = f(&bulk);
let (un, ut, urps) = f(&unwind);
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(0);
eprintln!(
"\n=== knut-popviz.falkor_fill @ scale {scale} ({bn} vertices) — host {cores} cores ===\n\
parallel-UNWIND: {un} in {ut:.3}s = {urps:.0} nodes/s\n\
native GRAPH.BULK: {bn} in {bt:.3}s = {brps:.0} nodes/s\n\
speedup: {:.1}x\n",
ut / bt.max(1e-9),
);
// testcontainers's async Drop needs an ambient Tokio runtime; enter `rt` for it.
let _enter = rt.enter();
drop(container);
}