tests/funnel_concurrency.rs
//! E1 — funnel **concurrency** stress (inject-and-assert, NOT no-panic).
//!
//! The funnel server is a single-writer: `IcebergWarehouse::open()` takes
//! redb's exclusive lock per `<root>/catalog.redb`, so exactly ONE writable
//! `Store` exists per root/process. This test models that faithfully with a
//! shared `Arc<Mutex<Store>>` and hammers it from N=16 worker threads while
//! reader threads take lock-tolerant `open_read_only` snapshots.
//!
//! Each worker records the ops it actually committed; after the storm we
//! **reopen from disk** (full replay) and assert:
//!
//! 1. **No lost updates** — reopened node + edge counts == initial + the sum
//! of every committed add (across all workers).
//! 2. **Final status map matches the recorded sequence** — we replay the
//! committed status events ordered by their (deterministic, strictly
//! increasing) timestamp and assert the on-disk status of every node
//! equals the last status the winning event set.
//! 3. **Frontier correctness** — after `promote_ready`, every Ready node has
//! all its deps Done, and no Pending node has all deps Done (promote is
//! complete).
//! 4. **Reader snapshots are always valid + monotonic** — every snapshot
//! replays cleanly and its event count never goes backwards.
//! 5. **No deadlock** — a wall-clock watchdog fails the test if the storm
//! doesn't finish in time.
//!
//! Runs across sizes small → very-large (up to ~1500 nodes).
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use chrono::{DateTime, TimeZone, Utc};
use nornir::funnel::{Event, IdeaId, NodeId, NodeStatus, PlanId, PlanStatus, Store};
/// Strictly-increasing event timestamps. The funnel log persists ts at
/// **microsecond** resolution and replay sorts by it, so racing threads must
/// each get a DISTINCT microsecond — a global atomic micros counter makes the
/// replay order total and deterministic (sub-µs steps would collapse on disk
/// and lose the insertion order, breaking the IdeaSubmitted→PlanCreated→…
/// dependency the validator enforces on replay).
static TS_MICROS: AtomicI64 = AtomicI64::new(1);
fn next_ts() -> DateTime<Utc> {
let micros = TS_MICROS.fetch_add(1, Ordering::SeqCst);
let secs = micros.div_euclid(1_000_000);
let nanos = (micros.rem_euclid(1_000_000) * 1_000) as u32;
Utc.timestamp_opt(secs, nanos).single().expect("valid ts")
}
/// One committed status change, kept so we can recompute the expected final
/// status of each node (the event with the largest ts wins).
#[derive(Clone)]
struct StatusOp {
node: NodeId,
status: NodeStatus,
ts: DateTime<Utc>,
}
/// Per-size run. `seed_nodes` are created up-front (a base DAG); workers then
/// add their own disjoint nodes, wire edges, flip statuses, and call topo_ready.
fn run_size(workers: usize, seed_nodes: usize, adds_per_worker: usize, budget: Duration) {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("wh");
let plan = PlanId::seq(1);
let idea = IdeaId::seq(1);
// --- seed: idea + active plan + a base chain DAG of `seed_nodes` nodes. ---
let (seed_node_ids, initial_edges) = {
let mut store = Store::open(&root).unwrap();
store
.record(Event::IdeaSubmitted {
id: idea.clone(),
source: "concurrency-test".into(),
text: "stress".into(),
refs: vec![],
item_kind: nornir::funnel::ItemKind::Idea,
ts: next_ts(),
})
.unwrap();
store
.record(Event::PlanCreated {
id: plan.clone(),
idea_id: idea.clone(),
summary: "stress plan".into(),
planner: "test".into(),
ts: next_ts(),
})
.unwrap();
store
.record(Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: next_ts(),
})
.unwrap();
let mut ids = Vec::with_capacity(seed_nodes);
for _ in 0..seed_nodes {
let nid = NodeId::seq(store.funnel.next_node.max(1));
store
.record(Event::NodeAdded {
plan_id: plan.clone(),
node_id: nid.clone(),
kind: "seed".into(),
params: serde_json::Map::new(),
targets: vec![],
prompt_excerpt: None,
ts: next_ts(),
})
.unwrap();
ids.push(nid);
}
// Chain edges seed[i-1] -> seed[i] so there's a real dep structure.
let mut edge_count = 0usize;
for w in ids.windows(2) {
store
.record(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: w[0].clone(),
to_node: w[1].clone(),
ts: next_ts(),
})
.unwrap();
edge_count += 1;
}
(ids, edge_count)
}; // writer drops → exclusive lock released, but we reopen one shared writer below
let initial_nodes = seed_node_ids.len();
// --- the single shared writable store (mirrors the server's one writer). ---
let store = Arc::new(Mutex::new(Store::open(&root).unwrap()));
// Tallies of what each worker actually committed.
let added_nodes = Arc::new(AtomicUsize::new(0));
let added_edges = Arc::new(AtomicUsize::new(0));
let status_ops: Arc<Mutex<Vec<StatusOp>>> = Arc::new(Mutex::new(Vec::new()));
let stop = Arc::new(AtomicBool::new(false));
let started = Instant::now();
let seed_ids = Arc::new(seed_node_ids);
// --- reader thread: lock-tolerant snapshots, asserting validity + monotone.
// ONE reader on a relaxed cadence: each `open_read_only` copies the whole
// catalog.redb aside (the contended path), so a tight loop would starve the
// writers — we only need enough snapshots to prove they're always valid and
// never regress, not a benchmark.
let reader_handles: Vec<_> = (0..1)
.map(|_| {
let root = root.clone();
let stop = Arc::clone(&stop);
std::thread::spawn(move || {
let mut last_count = 0usize;
let mut snapshots = 0usize;
while !stop.load(Ordering::Relaxed) {
// open_read_only never fails even while the writer holds the
// lock (copied-aside snapshot). It must always replay cleanly.
match Store::open_read_only(&root) {
Ok(mut s) => {
// Replaying succeeded ⇒ funnel is internally valid.
// promote_ready must not panic on any snapshot.
s.funnel.promote_ready();
let count = s
.funnel
.plans
.values()
.map(|p| p.nodes.len())
.sum::<usize>();
// Monotonic: node count never shrinks (events are
// append-only; no worker deletes nodes).
assert!(
count >= last_count,
"reader snapshot regressed: {count} < {last_count}"
);
last_count = count;
snapshots += 1;
}
Err(e) => panic!("open_read_only failed under contention: {e:#}"),
}
std::thread::sleep(Duration::from_millis(25));
}
snapshots
})
})
.collect();
// --- worker threads. ---
let worker_handles: Vec<_> = (0..workers)
.map(|w| {
let store = Arc::clone(&store);
let plan = plan.clone();
let added_nodes = Arc::clone(&added_nodes);
let added_edges = Arc::clone(&added_edges);
let status_ops = Arc::clone(&status_ops);
let seed_ids = Arc::clone(&seed_ids);
std::thread::spawn(move || {
// Each worker owns a disjoint set of *new* node ids it adds, so
// NodeAdded never collides. Flips/edges target seed nodes (shared
// but the status apply is last-writer-wins, validated by ts).
let mut my_new: Vec<NodeId> = Vec::with_capacity(adds_per_worker);
let statuses = [
NodeStatus::Ready,
NodeStatus::InProgress,
NodeStatus::Done,
NodeStatus::Failed,
];
for k in 0..adds_per_worker {
// 1) add a fresh node (its id comes from the shared
// high-water mark under the lock → unique).
{
let mut s = store.lock().unwrap_or_else(|e| e.into_inner());
let nid = NodeId::seq(s.funnel.next_node.max(1));
let ts = next_ts();
s.record(Event::NodeAdded {
plan_id: plan.clone(),
node_id: nid.clone(),
kind: format!("w{w}"),
params: serde_json::Map::new(),
targets: vec![],
prompt_excerpt: None,
ts,
})
.unwrap();
added_nodes.fetch_add(1, Ordering::Relaxed);
my_new.push(nid);
}
// 2) wire an edge seed[?] -> my_new[last] (dep on a seed
// node), guaranteed acyclic (new node is a fresh sink).
if !seed_ids.is_empty() {
let from = seed_ids[(w + k) % seed_ids.len()].clone();
let to = my_new.last().unwrap().clone();
let mut s = store.lock().unwrap_or_else(|e| e.into_inner());
// Skip if this exact edge already exists (another worker
// could share the same from for a different to, but
// (from,to) pairs are unique because `to` is our fresh
// node — still guard to keep the record() infallible).
let exists = s
.funnel
.plans
.get(&plan)
.map(|p| p.edges.contains(&(from.clone(), to.clone())))
.unwrap_or(false);
if !exists {
let ts = next_ts();
s.record(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: from,
to_node: to,
ts,
})
.unwrap();
added_edges.fetch_add(1, Ordering::Relaxed);
}
}
// 3) flip a status on a seed node (shared target → races).
if !seed_ids.is_empty() {
let target = seed_ids[(w * 7 + k * 3) % seed_ids.len()].clone();
let status = statuses[(w + k) % statuses.len()];
let mut s = store.lock().unwrap_or_else(|e| e.into_inner());
let ts = next_ts();
s.record(Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: target.clone(),
status,
why: None,
ts,
})
.unwrap();
status_ops
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(StatusOp {
node: target,
status,
ts,
});
}
// 4) exercise the frontier dispatcher under the lock.
{
let mut s = store.lock().unwrap_or_else(|e| e.into_inner());
let _ = nornir::funnel::topo_ready(&mut s.funnel);
}
}
})
})
.collect();
// Watchdog: a separate thread trips if the storm overruns the budget — a
// deadlock would otherwise hang the join below forever. (Each `record()`
// writes a full Iceberg snapshot, so the budget is generous; the point is to
// catch a *hang*, not to benchmark throughput.)
{
let stop = Arc::clone(&stop);
std::thread::spawn(move || {
let deadline = Instant::now() + budget;
while Instant::now() < deadline {
if stop.load(Ordering::Relaxed) {
return; // storm finished cleanly
}
std::thread::sleep(Duration::from_millis(50));
}
// Overran without `stop` being set ⇒ workers are stuck.
eprintln!("FATAL: concurrency storm exceeded budget {budget:?} — aborting (deadlock?)");
std::process::abort();
});
}
for (i, h) in worker_handles.into_iter().enumerate() {
h.join().unwrap_or_else(|_| panic!("worker {i} panicked"));
}
let storm_elapsed = started.elapsed();
stop.store(true, Ordering::Relaxed);
let mut total_snapshots = 0usize;
for h in reader_handles {
total_snapshots += h.join().unwrap();
}
assert!(total_snapshots > 0, "readers must have taken snapshots");
eprintln!(
"funnel concurrency: {workers} workers, seed {seed_nodes} nodes, +{} nodes / +{} edges in {storm_elapsed:?}, {total_snapshots} reader snapshots",
added_nodes.load(Ordering::Relaxed),
added_edges.load(Ordering::Relaxed),
);
// Drop the shared writer to release the exclusive lock before reopening.
drop(store);
// --- reopen from disk: full replay → durable truth. ---
let mut reopened = Store::open(&root).unwrap();
let p = reopened.funnel.plans.get(&plan).expect("plan persisted");
// (1) NO LOST UPDATES.
let committed_nodes = added_nodes.load(Ordering::Relaxed);
let committed_edges = added_edges.load(Ordering::Relaxed);
assert_eq!(
p.nodes.len(),
initial_nodes + committed_nodes,
"node count = initial + committed adds (no lost NodeAdded)"
);
assert_eq!(
p.edges.len(),
initial_edges + committed_edges,
"edge count = initial + committed adds (no lost EdgeAdded)"
);
// (2) FINAL STATUS MAP matches the recorded sequence (last ts wins).
let mut ops = status_ops.lock().unwrap_or_else(|e| e.into_inner()).clone();
ops.sort_by_key(|o| o.ts);
let mut expected_final: BTreeMap<NodeId, NodeStatus> = BTreeMap::new();
for op in &ops {
expected_final.insert(op.node.clone(), op.status);
}
for (nid, st) in &expected_final {
let actual = p.nodes.get(nid).expect("status-flipped node exists").status;
assert_eq!(
actual, *st,
"node {nid}: persisted status must equal the last (max-ts) recorded flip"
);
}
// (3) FRONTIER CORRECTNESS after promote_ready.
//
// Subtlety: a node can be `Ready` for TWO reasons — `promote_ready` lifted a
// Pending node whose deps are all Done, OR a worker's `NodeStatusChanged`
// event set it to Ready directly (deps irrelevant). The frontier invariant
// only constrains the *promotion* path, so we snapshot statuses BEFORE
// promote and assert only on nodes that were Pending and became Ready.
let pre: BTreeMap<NodeId, NodeStatus> = p
.nodes
.iter()
.map(|(id, n)| (id.clone(), n.status))
.collect();
reopened.funnel.promote_ready();
let p = reopened.funnel.plans.get(&plan).unwrap();
let status: BTreeMap<&NodeId, NodeStatus> =
p.nodes.iter().map(|(id, n)| (id, n.status)).collect();
let deps_done = |node: &NodeId| -> bool {
p.edges
.iter()
.filter(|(_, to)| to == node)
.all(|(from, _)| matches!(status.get(from), Some(NodeStatus::Done)))
};
for (nid, n) in &p.nodes {
let was_pending = matches!(pre.get(nid), Some(NodeStatus::Pending));
match n.status {
// Promotion path: a node Pending-before-promote that is now Ready
// must have had all deps Done. (A node *explicitly* flipped to Ready
// by an event — was_pending == false — is exempt.)
NodeStatus::Ready if was_pending => assert!(
deps_done(nid),
"promoted Ready node {nid} must have all deps Done"
),
// Completeness: a node still Pending must NOT have all deps Done
// (else promote_ready would have lifted it).
NodeStatus::Pending => assert!(
!deps_done(nid),
"Pending node {nid} with all deps Done should have been promoted"
),
_ => {}
}
}
}
// Each `record()` writes a full Iceberg snapshot (parquet + metadata), so these
// are deliberately modest in op-count — the dimension under test is *concurrent
// writers contending on the single redb lock*, not raw throughput. Budgets are
// generous: they only catch a hang, the watchdog aborts on a real deadlock.
#[test]
fn concurrency_small() {
// 16 workers × 2 adds (≈ 96 events) + 8 seed nodes.
run_size(16, 8, 2, Duration::from_secs(240));
}
#[test]
fn concurrency_medium() {
// 16 workers × 4 adds (≈ 192 events) + 32 seed nodes.
run_size(16, 32, 4, Duration::from_secs(420));
}
#[test]
#[ignore = "very-large (~1500 nodes, slow: one Iceberg snapshot per event); run with --ignored"]
fn concurrency_very_large() {
// 16 workers × ~90 adds + 64 seed ≈ 1504 nodes.
run_size(16, 64, 90, Duration::from_secs(1800));
}