From f5db76e406dca1374c3b5f7649d209f8c138fb0c Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 00:54:45 -0400 Subject: [PATCH 01/11] test(rooms-node): capture bounded crash/rejoin evidence --- .../src/unix/rejoin_trace.rs | 184 ++++++++++++++++++ crates/vhalla-rooms-node/src/unix/tests.rs | 51 ++++- 2 files changed, 230 insertions(+), 5 deletions(-) create mode 100644 crates/vhalla-rooms-node/src/unix/rejoin_trace.rs diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs new file mode 100644 index 00000000..9006fd5d --- /dev/null +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -0,0 +1,184 @@ +//! Bounded, read-only diagnostics for the crash/rejoin test. Never read +//! WAL contents or print keys, batch bodies, or absolute home paths. + +use super::*; + +fn home(label: &str, path: &Path) { + // `recover()` mutates journal residue; NEVER call it on a live writer. + let journal = vhalla_journal::Journal::new(path.join("app/journal"), vhalla_journal::FsStore); + let indexed: Vec<_> = (1..=4) + .map(|h| match journal.at_height(h) { + Ok(Some(id)) => format!("{h}:{}", hex(&id[..8])), + Ok(None) => format!("{h}:-"), + Err(_) => format!("{h}:error"), + }) + .collect(); + let wal_bytes = std::fs::metadata(path.join("wal/consensus.wal")) + .map(|m| m.len()) + .ok(); + let wal_format_ok = std::fs::metadata(path.join("wal/FORMAT")).is_ok_and(|m| m.len() == 4) + && std::fs::read(path.join("wal/FORMAT")).is_ok_and(|bytes| bytes == WAL_FORMAT); + // 4097 signals an overflow; no unbounded listing or filename disclosure. + let count = |dir: &str| { + std::fs::read_dir(path.join(dir)) + .map(|entries| entries.take(4097).count()) + .ok() + }; + println!( + "rejoin {label}: journal_index={indexed:?} wal_bytes={wal_bytes:?} wal_format_ok={wal_format_ok} batches={:?} seen={:?}", + count("store/batches"), + count("store/seen") + ); +} + +/// Read only the four test heights. Check each indexed bundle's identity, +/// height and predecessor link; never run writer-side `recover()` on a live +/// node. Keep full identities in the assertion, not in diagnostic output. +pub(super) fn committed_chain(path: &Path, through: u64) -> Vec<([u8; 32], [u8; 32])> { + assert!((1..=4).contains(&through)); + let journal = vhalla_journal::Journal::new(path.join("app/journal"), vhalla_journal::FsStore); + let mut chain: Vec<([u8; 32], [u8; 32])> = Vec::with_capacity(through as usize); + for h in 1..=through { + let id = journal + .at_height(h) + .expect("journal height marker readable") + .unwrap_or_else(|| panic!("missing journal height marker at {h}")); + let bundle = journal + .bundle(id) + .expect("indexed journal bundle valid") + .unwrap_or_else(|| panic!("missing indexed journal bundle at {h}")); + assert_eq!(bundle.height(), h, "indexed bundle at wrong height"); + if let Some((_, previous_next)) = chain.last() { + assert!( + bundle.predecessor() == *previous_next, + "journal predecessor mismatch at height {h}" + ); + } + chain.push((id, bundle.next())); + } + chain +} + +fn node(label: &str, node: &RoomNode) { + let frontier = node.adapter.lock().unwrap().frontier(); + let acks = node.sink.lock().unwrap(); + let ack_counts: Vec<_> = (1..=4) + .map(|h| { + acks.sent + .iter() + .filter(|m| match m { + vhalla_rooms_consensus::EngineMsg::CommitAck { height } => *height == h, + vhalla_rooms_consensus::EngineMsg::NextHeightReply { height } => { + height.checked_sub(1) == Some(h) + } + }) + .count() + }) + .collect(); + println!( + "rejoin {label}: frontier={} commitment={} ack_counts={ack_counts:?} ack_total={} resupplied={} loaded={:?} app_done={} engine_done={}", + frontier.height, + hex(&frontier.commitment()[..8]), + acks.sent.len(), + *node.resupplied.lock().unwrap(), + node.loaded, + node.task.is_finished(), + node._engine.handle.is_finished() + ); + home(label, &node.home); +} + +fn trace(stage: &str, peers: &[RoomNode], first_peer: usize, restarted: Option<&RoomNode>) { + println!("rejoin stage={stage}"); + if let Some(restarted) = restarted { + node("n1", restarted); + } + for (i, peer) in peers.iter().enumerate() { + node(&format!("n{}", i + first_peer), peer); + } +} + +pub(super) fn after_crash(peers: &[RoomNode], stopped_home: &Path) { + trace("after-crash", peers, 2, None); + home("n1:stopped", stopped_home); +} + +// Same predicates/deadlines as `wait_for`, with at most four snapshots per +// wait: start, 15s, 45s, and completion or timeout. +pub(super) async fn wait_for_rejoin bool>( + what: &str, + f: F, + timeout: Duration, + peers: &[RoomNode], + first_peer: usize, + restarted: Option<&RoomNode>, +) { + let start = Instant::now(); + trace(&format!("{what}:start"), peers, first_peer, restarted); + let checkpoints = [Duration::from_secs(15), Duration::from_secs(45)]; + let mut next = 0; + loop { + // Match `wait_for`: a predicate first observed after the deadline + // cannot turn a timed-out recovery into a success. + if start.elapsed() >= timeout { + trace(&format!("{what}:timeout"), peers, first_peer, restarted); + panic!("timed out waiting for {what}"); + } + if f() { + trace(&format!("{what}:complete"), peers, first_peer, restarted); + return; + } + if next < checkpoints.len() + && checkpoints[next] < timeout + && start.elapsed() >= checkpoints[next] + { + trace( + &format!("{what}:{}s", checkpoints[next].as_secs()), + peers, + first_peer, + restarted, + ); + next += 1; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +#[test] +fn committed_chain_reads_bounded_history_without_recovery() { + use vhalla_journal::{Bundle, BundleParts, Outcome}; + + let base = fixture("rejoin-chain"); + let journal = vhalla_journal::Journal::new(base.join("app/journal"), vhalla_journal::FsStore); + let mut predecessor = vhalla_journal::GENESIS_NEXT; + let mut ids = Vec::new(); + for h in 1..=4 { + let next = [h as u8; 32]; + let bundle = Bundle::new(BundleParts { + certificate: vec![h as u8], + predecessor, + next, + batch: Vec::new(), + value: Vec::new(), + configuration: Vec::new(), + control_record: Vec::new(), + debit_marker: Vec::new(), + height: h, + }) + .unwrap(); + assert_eq!(journal.commit(&bundle).unwrap(), Outcome::Committed); + ids.push((bundle.id(), next)); + predecessor = next; + } + for h in 1..=4 { + assert_eq!(committed_chain(&base, h), ids[..h as usize]); + } + let _ = std::fs::remove_dir_all(base); +} + +#[tokio::test] +#[should_panic(expected = "timed out waiting for expired predicate")] +async fn expired_predicate_cannot_pass() { + // Even a true predicate cannot turn an expired wait into a green result. + wait_for_rejoin("expired predicate", || true, Duration::ZERO, &[], 1, None).await; +} diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index 5e6df744..b3dadabc 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -21,6 +21,9 @@ mod formal_held_reply; #[path = "formal_finalized.rs"] mod formal_finalized; +#[path = "rejoin_trace.rs"] +mod rejoin_trace; + /// The shared batch plan: `heights` room creations over 8 beneficiary /// owners and 16 eligible award sources — enough credits for two rooms /// per owner across the longest run. @@ -242,16 +245,26 @@ async fn crashed_validator_recovers_and_rejoins() { ); } - wait_for( + rejoin_trace::wait_for_rejoin( "all four nodes to commit height 2", || nodes.iter().all(|n| n.committed_height() >= 2), Duration::from_secs(60), + &nodes, + 1, + None, ) .await; + // Preserve every journal entry committed at the sampled pre-crash frontier. + // The node may already be past height 2 when the other peers catch up. + // Index and bundle reads are nonmutating even while the writer is live. + let pre_crash_height = nodes[0].committed_height().min(HEIGHTS); + let pre_crash_chain = rejoin_trace::committed_chain(&nodes[0].home, pre_crash_height); + // Crash validator 1, wait a beat, restart on the same home. let victim = nodes.remove(0); victim.crash().await; + rejoin_trace::after_crash(&nodes, &base.join("n1")); tokio::time::sleep(Duration::from_secs(2)).await; let restarted = RoomNode::start(NodeSpec { @@ -281,16 +294,22 @@ async fn crashed_validator_recovers_and_rejoins() { restarted.loaded.1 ); - wait_for( + rejoin_trace::wait_for_rejoin( "restarted validator to reach height 4", || restarted.committed_height() >= HEIGHTS, Duration::from_secs(90), + &nodes, + 2, + Some(&restarted), ) .await; - wait_for( + rejoin_trace::wait_for_rejoin( "peers to reach height 4", || nodes.iter().all(|n| n.committed_height() >= HEIGHTS), Duration::from_secs(60), + &nodes, + 2, + Some(&restarted), ) .await; @@ -301,8 +320,30 @@ async fn crashed_validator_recovers_and_rejoins() { restarted_state ); } - // Each height committed exactly once — the replay of the durable - // marker deduplicates rather than re-applying. + // Restart must not replace any pre-crash indexed bundle. At every + // height, the verified journal chain reaches the same state on all + // peers; the h=4 bundle must agree with the live frontier. Different + // nodes may retain different quorum certificates, so compare state + // commitments across peers rather than their bundle identities. + let restarted_chain = rejoin_trace::committed_chain(&restarted.home, HEIGHTS); + assert!( + restarted_chain[..pre_crash_chain.len()] == pre_crash_chain[..], + "restart replaced pre-crash journal history" + ); + assert!( + restarted_chain[3].1 == restarted_state, + "journal tip differs from the live frontier" + ); + for node in &nodes { + let peer_chain = rejoin_trace::committed_chain(&node.home, HEIGHTS); + assert!( + restarted_chain + .iter() + .zip(peer_chain.iter()) + .all(|(restarted, peer)| restarted.1 == peer.1), + "journal state differs across validators" + ); + } assert_eq!(restarted.adapter.lock().unwrap().frontier().height, HEIGHTS); for node in nodes { From 27ab9ba46169f10a36913474791dd67b42dda99a Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 04:44:23 -0400 Subject: [PATCH 02/11] test: strengthen rooms-node rejoin evidence --- .../src/unix/rejoin_trace.rs | 177 ++++++++++++++++-- crates/vhalla-rooms-node/src/unix/tests.rs | 64 +++++++ 2 files changed, 230 insertions(+), 11 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 9006fd5d..6711dbab 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -3,6 +3,96 @@ use super::*; +// Only this test's four fixed listener ports are inspected. /proc/net/tcp* +// reports TCP sockets from the network namespace, not authenticated Malachite +// peers or even necessarily sockets owned by this test. Client ports are +// ephemeral, so port direction does not establish who initiated a connection. +pub(super) const PORT_BASE: u16 = 27300; + +#[derive(Default)] +struct SocketSummary { + listeners: [bool; 4], + established_local: [bool; 4], + established_remote: [bool; 4], +} + +impl SocketSummary { + fn port_index(endpoint: &str) -> Option { + let hex_port = endpoint.rsplit_once(':')?.1; + let port = u16::from_str_radix(hex_port, 16).ok()?; + let index = usize::from(port.checked_sub(PORT_BASE)?); + (index < 4).then_some(index) + } + + fn observe(&mut self, table: &str) { + for line in table.lines().skip(1) { + let mut fields = line.split_ascii_whitespace(); + let (Some(_slot), Some(local), Some(remote), Some(state)) = + (fields.next(), fields.next(), fields.next(), fields.next()) + else { + continue; + }; + match state { + "0A" => { + if let Some(local) = Self::port_index(local) { + self.listeners[local] = true; + } + } + "01" => { + if let Some(local) = Self::port_index(local) { + self.established_local[local] = true; + } + if let Some(remote) = Self::port_index(remote) { + self.established_remote[remote] = true; + } + } + _ => {} + } + } + } + + fn report(&self, families: usize) -> String { + format!( + "tcp_families={families} listeners={:?} established_local={:?} established_remote={:?}", + self.listeners, self.established_local, self.established_remote + ) + } +} + +#[cfg(target_os = "linux")] +fn socket_trace() -> String { + use std::io::Read; + const MAX_TABLE: u64 = 1_048_576; + let mut summary = SocketSummary::default(); + let mut families = 0; + for path in ["/proc/net/tcp", "/proc/net/tcp6"] { + let file = match std::fs::File::open(path) { + Ok(file) => file, + Err(error) + if path.ends_with("tcp6") && error.kind() == std::io::ErrorKind::NotFound => + { + continue; + } + Err(_) => return "tcp=unavailable".into(), + }; + let mut table = String::new(); + if file.take(MAX_TABLE + 1).read_to_string(&mut table).is_err() { + return "tcp=unavailable".into(); + } + if table.len() as u64 > MAX_TABLE { + return "tcp=unavailable (table exceeds 1 MiB)".into(); + } + summary.observe(&table); + families += 1; + } + summary.report(families) +} + +#[cfg(not(target_os = "linux"))] +fn socket_trace() -> String { + "tcp=unavailable (not Linux)".into() +} + fn home(label: &str, path: &Path) { // `recover()` mutates journal residue; NEVER call it on a live writer. let journal = vhalla_journal::Journal::new(path.join("app/journal"), vhalla_journal::FsStore); @@ -32,11 +122,17 @@ fn home(label: &str, path: &Path) { } /// Read only the four test heights. Check each indexed bundle's identity, -/// height and predecessor link; never run writer-side `recover()` on a live -/// node. Keep full identities in the assertion, not in diagnostic output. +/// height and predecessor link, including the fixture's genesis frontier; +/// never run writer-side `recover()` on a live node. Keep full identities in +/// the assertion, not in diagnostic output. pub(super) fn committed_chain(path: &Path, through: u64) -> Vec<([u8; 32], [u8; 32])> { assert!((1..=4).contains(&through)); let journal = vhalla_journal::Journal::new(path.join("app/journal"), vhalla_journal::FsStore); + let genesis_next = batch_plan(1) + .get(&1) + .expect("rejoin fixture has a first batch") + .parent + .commitment(); let mut chain: Vec<([u8; 32], [u8; 32])> = Vec::with_capacity(through as usize); for h in 1..=through { let id = journal @@ -48,12 +144,11 @@ pub(super) fn committed_chain(path: &Path, through: u64) -> Vec<([u8; 32], [u8; .expect("indexed journal bundle valid") .unwrap_or_else(|| panic!("missing indexed journal bundle at {h}")); assert_eq!(bundle.height(), h, "indexed bundle at wrong height"); - if let Some((_, previous_next)) = chain.last() { - assert!( - bundle.predecessor() == *previous_next, - "journal predecessor mismatch at height {h}" - ); - } + let expected_predecessor = chain.last().map_or(genesis_next, |(_, next)| *next); + assert!( + bundle.predecessor() == expected_predecessor, + "journal predecessor mismatch at height {h}" + ); chain.push((id, bundle.next())); } chain @@ -89,7 +184,7 @@ fn node(label: &str, node: &RoomNode) { } fn trace(stage: &str, peers: &[RoomNode], first_peer: usize, restarted: Option<&RoomNode>) { - println!("rejoin stage={stage}"); + println!("rejoin stage={stage} {}", socket_trace()); if let Some(restarted) = restarted { node("n1", restarted); } @@ -144,13 +239,43 @@ pub(super) async fn wait_for_rejoin bool>( } } +#[test] +fn socket_table_reports_only_test_ports_and_established_links() { + let port = |n: u16| format!("{:04X}", PORT_BASE + n); + let mut summary = SocketSummary::default(); + summary.observe(&format!( + "sl local_address rem_address st\n\ + 0: 0100007F:{} 00000000:0000 0A\n\ + 1: 0100007F:{} 0100007F:CAFE 01\n\ + 2: 0100007F:CAFF 0100007F:{} 01\n\ + 3: 0100007F:{} 0100007F:{} 06\n\ + 4: 0100007F:FFFF 0100007F:FFFF 01\n", + port(0), + port(0), + port(1), + port(2), + port(3) + )); + assert_eq!(summary.listeners, [true, false, false, false]); + assert_eq!(summary.established_local, [true, false, false, false]); + assert_eq!(summary.established_remote, [false, true, false, false]); + assert!(summary + .report(1) + .contains("established_local=[true, false, false, false]")); +} + #[test] fn committed_chain_reads_bounded_history_without_recovery() { use vhalla_journal::{Bundle, BundleParts, Outcome}; let base = fixture("rejoin-chain"); - let journal = vhalla_journal::Journal::new(base.join("app/journal"), vhalla_journal::FsStore); - let mut predecessor = vhalla_journal::GENESIS_NEXT; + let genesis_next = batch_plan(1).get(&1).unwrap().parent.commitment(); + let journal = vhalla_journal::Journal::with_genesis( + base.join("app/journal"), + vhalla_journal::FsStore, + genesis_next, + ); + let mut predecessor = genesis_next; let mut ids = Vec::new(); for h in 1..=4 { let next = [h as u8; 32]; @@ -174,6 +299,36 @@ fn committed_chain_reads_bounded_history_without_recovery() { assert_eq!(committed_chain(&base, h), ids[..h as usize]); } let _ = std::fs::remove_dir_all(base); + + // A valid journal under a different genesis must not be treated as this + // scenario's history, even when its height marker and bundle id agree. + let foreign = fixture("rejoin-chain-foreign-genesis"); + let foreign_genesis = [0xF1; 32]; + assert_ne!(foreign_genesis, genesis_next); + let journal = vhalla_journal::Journal::with_genesis( + foreign.join("app/journal"), + vhalla_journal::FsStore, + foreign_genesis, + ); + let bundle = Bundle::new(BundleParts { + certificate: vec![1], + predecessor: foreign_genesis, + next: [1; 32], + batch: Vec::new(), + value: Vec::new(), + configuration: Vec::new(), + control_record: Vec::new(), + debit_marker: Vec::new(), + height: 1, + }) + .unwrap(); + assert_eq!(journal.commit(&bundle).unwrap(), Outcome::Committed); + let result = std::panic::catch_unwind(|| committed_chain(&foreign, 1)); + let _ = std::fs::remove_dir_all(foreign); + assert!( + result.is_err(), + "foreign genesis was accepted as rejoin history" + ); } #[tokio::test] diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index b3dadabc..fa4ec18b 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -344,6 +344,37 @@ async fn crashed_validator_recovers_and_rejoins() { "journal state differs across validators" ); } + // Height and journal agreement alone do not show that the application + // applied the planned creations only once. Compare the live application + // effects with the deterministic signed plan on every validator. The + // journal chain above is durable, but this assertion does not reopen + // the application stores or count in-memory apply function calls. + let expected_tip = plan.get(&HEIGHTS).unwrap(); + let scenario = fixture::scenario(8, 16); + for node in std::iter::once(&restarted).chain(nodes.iter()) { + let adapter = node.adapter.lock().unwrap(); + let app = adapter.application(); + let registry = app.registry(); + assert_eq!(registry.digest(), expected_tip.result_registry); + assert_eq!(*app.social().root().as_bytes(), expected_tip.result_social); + assert_eq!( + registry + .accounts() + .values() + .map(|account| u64::from(account.lifetime_slots)) + .sum::(), + HEIGHTS, + "unexpected cumulative room creations" + ); + for owner in scenario.owners.iter().take(HEIGHTS as usize) { + let account = registry.account(owner.id); + assert_eq!( + account.lifetime_slots, 1, + "duplicate creation for planned owner" + ); + assert_eq!(account.spent, 1, "duplicate charge for planned owner"); + } + } assert_eq!(restarted.adapter.lock().unwrap().frontier().height, HEIGHTS); for node in nodes { @@ -353,6 +384,39 @@ async fn crashed_validator_recovers_and_rejoins() { let _ = std::fs::remove_dir_all(&base); } +/// Keep the recovery assertion's expected slot/charge counts tied to +/// replaying the real signed plan, without requiring network access. +#[test] +fn rejoin_plan_creates_one_room_and_charge_per_owner() { + let plan = fixture::plan(4, 8, 16); + let mut app = vhalla_rooms_consensus::Application::genesis( + plan.genesis.archive.clone(), + plan.genesis.registry().unwrap(), + ); + for batch in plan.batches.values() { + let checked = app.validate(batch).unwrap(); + app.apply_locally(checked); + } + let scenario = fixture::scenario(8, 16); + assert_eq!( + app.registry() + .accounts() + .values() + .map(|account| u64::from(account.lifetime_slots)) + .sum::(), + 4 + ); + for owner in scenario.owners.iter().take(4) { + let account = app.registry().account(owner.id); + assert_eq!(account.lifetime_slots, 1); + assert_eq!(account.spent, 1); + } + assert_eq!( + app.registry().digest(), + plan.batches.get(&4).unwrap().result_registry + ); +} + /// Two validators of a four-validator set cannot form a quorum: neither /// side may finalize anything. This is the partition clause — a 2–2 split /// finalizes no competing allocation. From c19e008156c95f4780f5e1f589ef6ecb69be3e4a Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 05:49:44 -0400 Subject: [PATCH 03/11] test(rooms-node): release rejoin trace ack lock before I/O --- .../src/unix/rejoin_trace.rs | 38 +++++++++++-------- 1 file changed, 22 insertions(+), 16 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 6711dbab..1c9843ec 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -156,26 +156,32 @@ pub(super) fn committed_chain(path: &Path, through: u64) -> Vec<([u8; 32], [u8; fn node(label: &str, node: &RoomNode) { let frontier = node.adapter.lock().unwrap().frontier(); - let acks = node.sink.lock().unwrap(); - let ack_counts: Vec<_> = (1..=4) - .map(|h| { - acks.sent - .iter() - .filter(|m| match m { - vhalla_rooms_consensus::EngineMsg::CommitAck { height } => *height == h, - vhalla_rooms_consensus::EngineMsg::NextHeightReply { height } => { - height.checked_sub(1) == Some(h) - } - }) - .count() - }) - .collect(); + // Snapshot the ack evidence without holding the sink mutex across other + // locks, logging, or filesystem reads; the engine needs it to record acks. + let (ack_counts, ack_total) = { + let acks = node.sink.lock().unwrap(); + let counts: Vec<_> = (1..=4) + .map(|h| { + acks.sent + .iter() + .filter(|m| match m { + vhalla_rooms_consensus::EngineMsg::CommitAck { height } => *height == h, + vhalla_rooms_consensus::EngineMsg::NextHeightReply { height } => { + height.checked_sub(1) == Some(h) + } + }) + .count() + }) + .collect(); + (counts, acks.sent.len()) + }; + let resupplied = *node.resupplied.lock().unwrap(); println!( "rejoin {label}: frontier={} commitment={} ack_counts={ack_counts:?} ack_total={} resupplied={} loaded={:?} app_done={} engine_done={}", frontier.height, hex(&frontier.commitment()[..8]), - acks.sent.len(), - *node.resupplied.lock().unwrap(), + ack_total, + resupplied, node.loaded, node.task.is_finished(), node._engine.handle.is_finished() From d7b0e3d723593d85e378e82016d1a5d48a1a0747 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 07:54:40 -0400 Subject: [PATCH 04/11] test(rooms): require post-crash height-four progress --- crates/vhalla-rooms-node/src/unix/tests.rs | 29 ++++++++++++++++------ 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index fa4ec18b..1d6d02fc 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -255,16 +255,29 @@ async fn crashed_validator_recovers_and_rejoins() { ) .await; - // Preserve every journal entry committed at the sampled pre-crash frontier. - // The node may already be past height 2 when the other peers catch up. - // Index and bundle reads are nonmutating even while the writer is live. - let pre_crash_height = nodes[0].committed_height().min(HEIGHTS); - let pre_crash_chain = rejoin_trace::committed_chain(&nodes[0].home, pre_crash_height); - - // Crash validator 1, wait a beat, restart on the same home. + // Take the pre-restart chain only AFTER the victim is stopped: a live + // frontier sample can miss a height committed between sampling and crash. + // A run that finished h4 before the crash cannot prove h4 recovery. + let stopped_home = base.join("n1"); let victim = nodes.remove(0); victim.crash().await; - rejoin_trace::after_crash(&nodes, &base.join("n1")); + rejoin_trace::after_crash(&nodes, &stopped_home); + let journal = + vhalla_journal::Journal::new(stopped_home.join("app/journal"), vhalla_journal::FsStore); + let pre_crash_height = (1..=HEIGHTS) + .rev() + .find(|&h| { + journal + .at_height(h) + .expect("stopped journal marker readable") + .is_some() + }) + .expect("victim committed height 2 before crash"); + assert!( + pre_crash_height < HEIGHTS, + "victim committed h4 before crash; this run cannot prove post-restart h4 recovery" + ); + let pre_crash_chain = rejoin_trace::committed_chain(&stopped_home, pre_crash_height); tokio::time::sleep(Duration::from_secs(2)).await; let restarted = RoomNode::start(NodeSpec { From 21c3d8c72bb7090f21789d1c69b5534d1eaf37ef Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:32:07 -0400 Subject: [PATCH 05/11] test(rooms): wait for app shutdown before journal sample --- crates/vhalla-rooms-node/src/unix/tests.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index 1d6d02fc..a572ea2b 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -259,7 +259,14 @@ async fn crashed_validator_recovers_and_rejoins() { // frontier sample can miss a height committed between sampling and crash. // A run that finished h4 before the crash cannot prove h4 recovery. let stopped_home = base.join("n1"); - let victim = nodes.remove(0); + let mut victim = nodes.remove(0); + // `crash` aborts the app loop but does not join its task. A synchronous + // journal write may still finish after it returns, so wait for the task + // to stop before treating the journal markers as a pre-restart frontier. + victim.task.abort(); + let _ = tokio::time::timeout(Duration::from_secs(10), &mut victim.task) + .await + .expect("application task must stop before inspecting the journal"); victim.crash().await; rejoin_trace::after_crash(&nodes, &stopped_home); let journal = From d20d980b7a6e0c117afc3cebc28954d4d3e1f670 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:48:08 -0400 Subject: [PATCH 06/11] test(rooms): stage peer batches after victim crash --- crates/vhalla-rooms-node/src/unix/tests.rs | 46 +++++++++++++++++----- 1 file changed, 36 insertions(+), 10 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index a572ea2b..b632de1c 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -224,6 +224,14 @@ async fn late_joiner_syncs_and_commits() { async fn crashed_validator_recovers_and_rejoins() { const HEIGHTS: u64 = 4; let plan = batch_plan(HEIGHTS); + // Stage h3-h4 only after the victim stops. Otherwise the four nodes can + // complete the entire plan between the h2 observation and the crash, + // making post-restart h4 progress impossible to demonstrate. + let first_two = plan + .iter() + .filter(|(height, _)| **height <= 2) + .map(|(height, batch)| (*height, batch.clone())) + .collect::>(); let (keys, set) = validators(4); let base = fixture("crash"); let base_port = 27300usize; @@ -236,7 +244,7 @@ async fn crashed_validator_recovers_and_rejoins() { config: node_config(i + 1, 4, base_port), node_key: key.clone(), validator_sets: sched(set.clone()), - held: plan.clone(), + held: first_two.clone(), genesis: genesis(), wal_faults: None, net_gate: None, @@ -280,31 +288,49 @@ async fn crashed_validator_recovers_and_rejoins() { .is_some() }) .expect("victim committed height 2 before crash"); - assert!( - pre_crash_height < HEIGHTS, - "victim committed h4 before crash; this run cannot prove post-restart h4 recovery" + assert_eq!( + pre_crash_height, 2, + "victim must stop at h2 before any post-crash batches are released" ); let pre_crash_chain = rejoin_trace::committed_chain(&stopped_home, pre_crash_height); tokio::time::sleep(Duration::from_secs(2)).await; + // The three live validators still form a quorum. Release each remaining + // planned batch only at its actual parent frontier, then let the crashed + // validator catch up from their decided history on the same home. + for height in 3..=HEIGHTS { + for node in &nodes { + node.submit(plan.get(&height).unwrap().clone()).await; + } + rejoin_trace::wait_for_rejoin( + &format!("live peers to commit height {height} after crash"), + || nodes.iter().all(|n| n.committed_height() >= height), + Duration::from_secs(60), + &nodes, + 2, + None, + ) + .await; + } + let restarted = RoomNode::start(NodeSpec { home: base.join("n1"), config: node_config(1, 4, base_port), node_key: keys[0].clone(), validator_sets: sched(set.clone()), - held: plan.clone(), + held: first_two.clone(), genesis: genesis(), wal_faults: None, net_gate: None, }) .await; - // The application-owned store survived the crash: the four spec-held - // batch bodies plus the seen-proposal records for the heights the - // node voted on reloaded — the WAL restored votes; this restores the - // value content. + // The application-owned store survived the crash: both pre-crash + // spec-held batch bodies and the seen-proposal records reloaded. The + // restarted node does not receive h3-h4 through its spec; those values + // must arrive from the live peers as decided history. assert!( - restarted.loaded.0 >= 4, + restarted.loaded.0 >= 2, "store/batches must reload retained batches, got {}", restarted.loaded.0 ); From c9d3c217ed8bca8d45880f72fd7b20b2f99c59c0 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 09:29:35 -0400 Subject: [PATCH 07/11] test(rooms-node): verify reopened stores after rejoin --- crates/vhalla-rooms-node/src/unix/tests.rs | 44 ++++++++++++++++++++-- 1 file changed, 40 insertions(+), 4 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index b632de1c..17f36117 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -393,8 +393,8 @@ async fn crashed_validator_recovers_and_rejoins() { // Height and journal agreement alone do not show that the application // applied the planned creations only once. Compare the live application // effects with the deterministic signed plan on every validator. The - // journal chain above is durable, but this assertion does not reopen - // the application stores or count in-memory apply function calls. + // journal chain above is durable; the stopped-store reopen below checks + // recoverable persistence, not the number of apply function calls. let expected_tip = plan.get(&HEIGHTS).unwrap(); let scenario = fixture::scenario(8, 16); for node in std::iter::once(&restarted).chain(nodes.iter()) { @@ -423,10 +423,46 @@ async fn crashed_validator_recovers_and_rejoins() { } assert_eq!(restarted.adapter.lock().unwrap().frontier().height, HEIGHTS); - for node in nodes { + // Stop every application task before reading snapshots: `crash` aborts + // the task but does not join it, and an in-flight synchronous commit can + // otherwise race a supposedly post-crash store read. + for mut node in nodes.into_iter().chain(std::iter::once(restarted)) { + node.task.abort(); + let _ = tokio::time::timeout(Duration::from_secs(10), &mut node.task) + .await + .expect("application task must stop before reading snapshot stores"); node.crash().await; } - restarted.crash().await; + + // A fresh adapter must reconstruct the same application state from the + // stopped nodes' on-disk journal and snapshot stores, not their live Arcs. + // Adapter::open may reconcile retained intents, so this checks recoverable + // persistence, not whether both snapshot pins needed no repair. + let shared_genesis = genesis(); + for index in 1..=4 { + let app_home = base.join(format!("n{index}/app")); + let reopened = Adapter::open(app_home, &shared_genesis) + .expect("stopped application must reopen from durable state"); + assert_eq!(reopened.frontier().height, HEIGHTS); + let app = reopened.application(); + let registry = app.registry(); + assert_eq!(registry.digest(), expected_tip.result_registry); + assert_eq!(*app.social().root().as_bytes(), expected_tip.result_social); + assert_eq!( + registry + .accounts() + .values() + .map(|account| u64::from(account.lifetime_slots)) + .sum::(), + HEIGHTS, + "unexpected persisted room creations at n{index}" + ); + for owner in scenario.owners.iter().take(HEIGHTS as usize) { + let account = registry.account(owner.id); + assert_eq!(account.lifetime_slots, 1, "persisted slots at n{index}"); + assert_eq!(account.spent, 1, "persisted charges at n{index}"); + } + } let _ = std::fs::remove_dir_all(&base); } From a4d7702ec1ed118e370bd97d020be3b2a1600c08 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:45:52 -0400 Subject: [PATCH 08/11] test(rooms-node): distinguish local rejoin history availability --- .../src/unix/rejoin_trace.rs | 77 ++++++++++++++++++- crates/vhalla-rooms-node/src/unix/tests.rs | 26 +++++++ 2 files changed, 99 insertions(+), 4 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 1c9843ec..11004e03 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -2,6 +2,7 @@ //! WAL contents or print keys, batch bodies, or absolute home paths. use super::*; +use std::io::Read; // Only this test's four fixed listener ports are inspected. /proc/net/tcp* // reports TCP sockets from the network namespace, not authenticated Malachite @@ -61,7 +62,6 @@ impl SocketSummary { #[cfg(target_os = "linux")] fn socket_trace() -> String { - use std::io::Read; const MAX_TABLE: u64 = 1_048_576; let mut summary = SocketSummary::default(); let mut families = 0; @@ -93,6 +93,18 @@ fn socket_trace() -> String { "tcp=unavailable (not Linux)".into() } +// A marker can change between metadata and read. Cap the diagnostic read at +// one byte past the expected format so even a replaced file stays bounded. +fn wal_marker_matches(path: &Path) -> bool { + let Ok(file) = std::fs::File::open(path) else { + return false; + }; + let mut bytes = Vec::with_capacity(WAL_FORMAT.len() + 1); + file.take((WAL_FORMAT.len() + 1) as u64) + .read_to_end(&mut bytes) + .is_ok_and(|_| bytes.as_slice() == WAL_FORMAT) +} + fn home(label: &str, path: &Path) { // `recover()` mutates journal residue; NEVER call it on a live writer. let journal = vhalla_journal::Journal::new(path.join("app/journal"), vhalla_journal::FsStore); @@ -106,8 +118,7 @@ fn home(label: &str, path: &Path) { let wal_bytes = std::fs::metadata(path.join("wal/consensus.wal")) .map(|m| m.len()) .ok(); - let wal_format_ok = std::fs::metadata(path.join("wal/FORMAT")).is_ok_and(|m| m.len() == 4) - && std::fs::read(path.join("wal/FORMAT")).is_ok_and(|bytes| bytes == WAL_FORMAT); + let wal_format_ok = wal_marker_matches(&path.join("wal/FORMAT")); // 4097 signals an overflow; no unbounded listing or filename disclosure. let count = |dir: &str| { std::fs::read_dir(path.join(dir)) @@ -154,6 +165,31 @@ pub(super) fn committed_chain(path: &Path, through: u64) -> Vec<([u8; 32], [u8; chain } +/// A read-only call to the same journal-backed page builder the sync host +/// uses. This establishes local serveability, NOT that a peer requested or +/// received a page. Never print the error: it can contain a home path. +#[derive(Debug, PartialEq, Eq)] +pub(super) enum LocalHistory { + Empty, + Values(usize), + Unavailable, +} + +fn classify_history(result: Result, E>) -> LocalHistory { + match result { + Ok(values) if values.is_empty() => LocalHistory::Empty, + Ok(values) => LocalHistory::Values(values.len()), + Err(_) => LocalHistory::Unavailable, + } +} + +pub(super) fn local_history(node: &RoomNode, start: u64, end: u64) -> LocalHistory { + // read_decided_range clamps to the applied frontier, bounds the page to + // 32 values / 2 MiB, and never invokes journal recovery or changes files. + let adapter = node.adapter.lock().unwrap(); + classify_history(read_decided_range(&adapter, start, end)) +} + fn node(label: &str, node: &RoomNode) { let frontier = node.adapter.lock().unwrap().frontier(); // Snapshot the ack evidence without holding the sink mutex across other @@ -176,8 +212,11 @@ fn node(label: &str, node: &RoomNode) { (counts, acks.sent.len()) }; let resupplied = *node.resupplied.lock().unwrap(); + // The fixed h3-h4 window reports Empty and Unavailable separately. A + // positive count is source readiness, not proof of wire delivery. + let history_h3_h4 = local_history(node, 3, 4); println!( - "rejoin {label}: frontier={} commitment={} ack_counts={ack_counts:?} ack_total={} resupplied={} loaded={:?} app_done={} engine_done={}", + "rejoin {label}: frontier={} commitment={} ack_counts={ack_counts:?} ack_total={} resupplied={} loaded={:?} local_history_h3_h4={history_h3_h4:?} app_done={} engine_done={}", frontier.height, hex(&frontier.commitment()[..8]), ack_total, @@ -245,6 +284,36 @@ pub(super) async fn wait_for_rejoin bool>( } } +#[test] +fn wal_marker_diagnostic_rejects_missing_or_trailing_bytes() { + let dir = fixture("rejoin-wal-marker"); + let marker = dir.join("FORMAT"); + assert!(!wal_marker_matches(&marker)); + std::fs::write(&marker, WAL_FORMAT).unwrap(); + assert!(wal_marker_matches(&marker)); + let mut extended = WAL_FORMAT.to_vec(); + extended.push(b'X'); + std::fs::write(&marker, extended).unwrap(); + assert!(!wal_marker_matches(&marker)); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn local_history_distinguishes_empty_values_and_unavailable_without_error_details() { + assert_eq!( + classify_history(Ok::, &str>(vec![])), + LocalHistory::Empty + ); + assert_eq!( + classify_history(Ok::, &str>(vec![1, 2])), + LocalHistory::Values(2) + ); + assert_eq!( + classify_history(Err::, &str>("secret absolute home path")), + LocalHistory::Unavailable + ); +} + #[test] fn socket_table_reports_only_test_ports_and_established_links() { let port = |n: u16| format!("{:04X}", PORT_BASE + n); diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index 17f36117..719f85fd 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -293,6 +293,17 @@ async fn crashed_validator_recovers_and_rejoins() { "victim must stop at h2 before any post-crash batches are released" ); let pre_crash_chain = rejoin_trace::committed_chain(&stopped_home, pre_crash_height); + // The victim's stopped home must not already contain the h3-h4 bodies. + // Test only the two planned ids; no directory listing or WAL mutation. + for height in 3..=HEIGHTS { + let body = stopped_home + .join("store/batches") + .join(hex(&plan[&height].value_id())); + assert!( + matches!(std::fs::metadata(body), Err(error) if error.kind() == std::io::ErrorKind::NotFound), + "victim had a post-crash batch body before the peers advanced" + ); + } tokio::time::sleep(Duration::from_secs(2)).await; // The three live validators still form a quorum. Release each remaining @@ -311,6 +322,16 @@ async fn crashed_validator_recovers_and_rejoins() { None, ) .await; + // Exercise the actual bounded journal-to-sync page conversion on + // each live source. Empty history and a failed read are distinct; + // neither can count as evidence that the source can serve this page. + for node in &nodes { + assert_eq!( + rejoin_trace::local_history(node, 3, height), + rejoin_trace::LocalHistory::Values((height - 2) as usize), + "live peer cannot serve the post-crash history page" + ); + } } let restarted = RoomNode::start(NodeSpec { @@ -359,6 +380,11 @@ async fn crashed_validator_recovers_and_rejoins() { ) .await; + assert_eq!( + rejoin_trace::local_history(&restarted, 3, 4), + rejoin_trace::LocalHistory::Values(2), + "restarted victim cannot serve its recovered history page" + ); let restarted_state = restarted.adapter.lock().unwrap().frontier().commitment(); for node in &nodes { assert_eq!( From bb5abae088fcdee66f2a1366268df26468e77934 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:03:08 -0400 Subject: [PATCH 09/11] test(rooms-node): trace verified peers, WAL startup and decided sync on rejoin --- .../src/unix/rejoin_trace.rs | 565 ++++++++++++++++++ crates/vhalla-rooms-node/src/unix/tests.rs | 21 +- 2 files changed, 584 insertions(+), 2 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 11004e03..55237ece 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -412,3 +412,568 @@ async fn expired_predicate_cannot_pass() { // Even a true predicate cannot turn an expired wait into a green result. wait_for_rejoin("expired predicate", || true, Duration::ZERO, &[], 1, None).await; } + +// The real network actor's output port, not a local journal read, reports +// decoded sync requests and replies. Record only test node indices, bounded +// ranges and page lengths; no peer IDs, raw values, keys, paths or errors. +#[derive(Debug, Clone, PartialEq, Eq)] +enum WireFact { + Connected { + node: usize, + peer: usize, + }, + Disconnected { + node: usize, + peer: usize, + }, + Request { + node: usize, + peer: usize, + start: u64, + end: u64, + }, + Page { + node: usize, + peer: usize, + start: u64, + heights: Vec, + }, + Empty { + node: usize, + peer: usize, + start: u64, + }, + Unavailable { + node: usize, + peer: usize, + }, + Failed { + node: usize, + peer: usize, + }, +} + +#[derive(Default)] +struct WireEvents { + facts: Vec, + overflow: usize, +} + +struct WireCollector { + node: usize, + peers: std::sync::Arc>, + events: std::sync::Arc>, +} + +#[ractor::async_trait] +impl Actor for WireCollector { + type Msg = NetworkEvent; + type State = Self; + type Arguments = Self; + + async fn pre_start( + &self, + _myself: ActorRef, + args: Self::Arguments, + ) -> Result { + Ok(args) + } + + async fn handle( + &self, + _myself: ActorRef, + msg: Self::Msg, + state: &mut Self::State, + ) -> Result<(), ActorProcessingErr> { + use arc_malachitebft_sync::{Request, Response}; + let peer_of = |peer: PeerId| state.peers.iter().position(|p| *p == peer).map(|i| i + 1); + let fact = match msg { + NetworkEvent::PeerConnected(peer) => peer_of(peer).map(|peer| WireFact::Connected { + node: state.node, + peer, + }), + NetworkEvent::PeerDisconnected(peer) => { + peer_of(peer).map(|peer| WireFact::Disconnected { + node: state.node, + peer, + }) + } + NetworkEvent::SyncRequest(_, peer, Request::ValueRequest(request)) => peer_of(peer) + .map(|peer| WireFact::Request { + node: state.node, + peer, + start: request.range.start().as_u64(), + end: request.range.end().as_u64(), + }), + NetworkEvent::SyncResponse(_, peer, Some(Response::ValueResponse(response))) => { + peer_of(peer).map(|peer| { + let start = response.start_height.as_u64(); + if response.values.is_empty() { + WireFact::Empty { + node: state.node, + peer, + start, + } + } else { + WireFact::Page { + node: state.node, + peer, + start, + heights: response + .values + .iter() + .take(32) + .map(|v| v.height().as_u64()) + .collect(), + } + } + }) + } + NetworkEvent::SyncResponse(_, peer, None) => { + peer_of(peer).map(|peer| WireFact::Unavailable { + node: state.node, + peer, + }) + } + NetworkEvent::SyncRequestFailed(_, peer, _) => { + peer_of(peer).map(|peer| WireFact::Failed { + node: state.node, + peer, + }) + } + _ => None, + }; + if let Some(fact) = fact { + let mut events = state.events.lock().unwrap(); + if events.facts.len() < 128 { + events.facts.push(fact); + } else { + events.overflow += 1; + } + } + Ok(()) + } +} + +// Filter before the observer actor mailbox: consensus gossip is not copied. +struct SelectedWireEvents(ActorRef>); + +impl OutputPortSubscriberTrait> for SelectedWireEvents { + fn subscribe_to_port(&self, port: &OutputPort>) { + port.subscribe(self.0.clone(), |event| match event { + NetworkEvent::PeerConnected(_) + | NetworkEvent::PeerDisconnected(_) + | NetworkEvent::SyncRequest(..) + | NetworkEvent::SyncResponse(..) + | NetworkEvent::SyncRequestFailed(..) => Some(event), + _ => None, + }); + } +} + +impl Subscriber> for SelectedWireEvents { + fn send(&self, event: NetworkEvent) { + if matches!( + event, + NetworkEvent::PeerConnected(_) + | NetworkEvent::PeerDisconnected(_) + | NetworkEvent::SyncRequest(..) + | NetworkEvent::SyncResponse(..) + | NetworkEvent::SyncRequestFailed(..) + ) { + let _ = self.0.cast(event); + } + } +} + +pub(super) struct WireTrace { + events: std::sync::Arc>, + peers: std::sync::Arc>, + observers: Vec>>, +} + +impl WireTrace { + pub(super) fn new(keys: &[PrivateKey]) -> Self { + assert_eq!(keys.len(), 4); + Self { + events: Default::default(), + peers: std::sync::Arc::new( + keys.iter() + .map(|key| { + net_peer_id(&key.public_key()) + .parse() + .expect("test peer id") + }) + .collect(), + ), + observers: Vec::new(), + } + } + + pub(super) async fn attach(&mut self, node: &RoomNode, index: usize) { + let (actor, _) = Actor::spawn( + None, + WireCollector { + node: index, + peers: self.peers.clone(), + events: self.events.clone(), + }, + WireCollector { + node: index, + peers: self.peers.clone(), + events: self.events.clone(), + }, + ) + .await + .expect("wire observer"); + node.inner_net + .as_ref() + .expect("trace needs real gated network") + .cast(NetActorMsg::Subscribe(Box::new(SelectedWireEvents( + actor.clone(), + )))) + .expect("wire subscribe"); + // DumpState follows Subscribe on the same real-network mailbox, so it + // also acts as the bounded subscription barrier before releasing n1. + let _ = self.dump(node).await; + self.observers.push(actor); + } + + async fn dump(&self, node: &RoomNode) -> arc_malachitebft_engine::network::NetworkStateDump { + let (tx, rx) = tokio::sync::oneshot::channel(); + node.inner_net + .as_ref() + .expect("real network") + .cast(NetActorMsg::DumpState(tx.into())) + .expect("network dump request"); + tokio::time::timeout(Duration::from_secs(5), rx) + .await + .expect("network dump deadline") + .expect("network dump response") + .expect("running real network") + } + + pub(super) async fn assert_causal_page( + &self, + restarted: &RoomNode, + sources: &[RoomNode], + keys: &[PrivateKey], + ) { + let responders = { + let snapshot = self.events.lock().unwrap(); + println!( + "rejoin wire facts={:?} overflow={}", + snapshot.facts, snapshot.overflow + ); + assert_eq!( + snapshot.overflow, 0, + "wire trace saturated; cannot prove page delivery" + ); + let mut responders = std::collections::BTreeSet::new(); + for height in 3..=4 { + let source = (2..=4) + .find(|source| { + snapshot.facts.iter().any(|f| { + matches!(f, WireFact::Request { node, peer: 1, start, end } + if node == source && *start <= height && *end >= height) + }) && snapshot.facts.iter().any(|f| { + matches!(f, WireFact::Page { node: 1, peer, heights, .. } + if peer == source && heights.contains(&height)) + }) + }) + .unwrap_or_else(|| { + panic!("no correlated request and received decided page for h{height}") + }); + responders.insert(source); + } + responders + }; + let victim_dump = self.dump(restarted).await; + for source in responders { + let source_key = keys[source - 1].public_key(); + let source_peer = + libp2p_identity::PeerId::from_bytes(&self.peers[source - 1].to_bytes()) + .expect("test source peer id"); + let victim_key = keys[0].public_key(); + let source_dump = self.dump(&sources[source - 2]).await; + assert!( + victim_dump + .peers + .get(&source_peer) + .is_some_and(|p| p.consensus_public_key.as_deref() + == Some(source_key.as_bytes().as_slice())), + "restarted victim lacks the source's verified validator proof" + ); + assert!( + source_dump + .peers + .get( + &libp2p_identity::PeerId::from_bytes(&self.peers[0].to_bytes()) + .expect("test victim peer id") + ) + .is_some_and(|p| p.consensus_public_key.as_deref() + == Some(victim_key.as_bytes().as_slice())), + "decided-page source lacks the victim's verified validator proof" + ); + println!("rejoin verified peer lifecycle: n1<->n{source} validator proofs match"); + } + } +} + +impl Drop for WireTrace { + fn drop(&mut self) { + let snapshot = self.events.lock().unwrap(); + // Failure snapshots also include this bounded, redacted wire ledger. + println!( + "rejoin wire final={:?} overflow={}", + snapshot.facts, snapshot.overflow + ); + for observer in &self.observers { + observer.stop(None); + } + } +} + +// A minimal test-only tracing subscriber for the pinned consensus actor. It +// never formats arbitrary log fields (WAL entries can contain private data). +// It records only whitelisted startup branch names under n1's node span. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum WalFact { + EmptyFetch, + NonemptyFetch, + Deferred, + Replaying, + Superseded, + DelayElapsed, + Running, + Error, +} + +#[derive(Default)] +struct WalEvents { + facts: Vec, + overflow: usize, +} + +thread_local! { + static WAL_SPAN_STACK: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; +} + +struct SafeFields { + moniker: Option, + message: Option, + new: Option, +} + +impl tracing::field::Visit for SafeFields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if matches!(field.name(), "moniker" | "message" | "new") { + let value = format!("{value:?}"); + if value.len() <= 128 { + match field.name() { + "moniker" => self.moniker = Some(value), + "message" => self.message = Some(value), + "new" => self.new = Some(value), + _ => {} + } + } + } + } +} + +struct WalSubscriber { + next: std::sync::atomic::AtomicU64, + n1_spans: std::sync::Mutex>, + armed: std::sync::Arc, + events: std::sync::Arc>, +} + +impl tracing::Subscriber for WalSubscriber { + fn enabled(&self, meta: &tracing::Metadata<'_>) -> bool { + meta.is_span() + || (meta.target().contains("consensus") && *meta.level() <= tracing::Level::DEBUG) + } + + fn new_span(&self, attrs: &tracing::span::Attributes<'_>) -> tracing::span::Id { + let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; + let mut fields = SafeFields { + moniker: None, + message: None, + new: None, + }; + if attrs.metadata().name() == "node" { + attrs.record(&mut fields); + } + let parent = attrs + .parent() + .map(tracing::span::Id::into_u64) + .or_else(|| WAL_SPAN_STACK.with(|stack| stack.borrow().last().copied())); + if fields + .moniker + .as_deref() + .is_some_and(|v| v.trim_matches('"') == "room-node-1") + || parent.is_some_and(|parent| self.n1_spans.lock().unwrap().contains(&parent)) + { + self.n1_spans.lock().unwrap().insert(id); + } + tracing::span::Id::from_u64(id) + } + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + if !self.armed.load(Ordering::Relaxed) || !event.metadata().target().contains("consensus") { + return; + } + let parent = event + .parent() + .map(tracing::span::Id::into_u64) + .or_else(|| WAL_SPAN_STACK.with(|stack| stack.borrow().last().copied())); + if !parent.is_some_and(|parent| self.n1_spans.lock().unwrap().contains(&parent)) { + return; + } + let mut fields = SafeFields { + moniker: None, + message: None, + new: None, + }; + event.record(&mut fields); + let message = fields.message.as_deref().unwrap_or("").trim_matches('"'); + let fact = if message == "No WAL entries to replay" { + Some(WalFact::EmptyFetch) + } else if message.starts_with("Found ") && message.ends_with(" WAL entries") { + Some(WalFact::NonemptyFetch) + } else if message == "Deferring WAL replay to wait for sync" { + Some(WalFact::Deferred) + } else if message.starts_with("Replaying ") && message.ends_with(" WAL entries") { + Some(WalFact::Replaying) + } else if message == "Sync certificate applied; transitioning out of WaitingForSync" { + Some(WalFact::Superseded) + } else if message + == "WAL replay delay elapsed without consensus reaching a decision, replaying WAL" + { + Some(WalFact::DelayElapsed) + } else if message == "Phase transition" + && fields.new.as_deref().is_some_and(|v| v.contains("Running")) + { + Some(WalFact::Running) + } else if *event.metadata().level() == tracing::Level::ERROR { + Some(WalFact::Error) + } else { + None + }; + if let Some(fact) = fact { + let mut events = self.events.lock().unwrap(); + if events.facts.len() < 32 { + events.facts.push(fact); + } else { + events.overflow += 1; + } + } + } + fn enter(&self, id: &tracing::span::Id) { + WAL_SPAN_STACK.with(|stack| stack.borrow_mut().push(id.into_u64())); + } + fn exit(&self, id: &tracing::span::Id) { + WAL_SPAN_STACK.with(|stack| { + let removed = stack.borrow_mut().pop(); + debug_assert_eq!(removed, Some(id.into_u64())); + }); + } +} + +pub(super) struct WalTrace { + events: std::sync::Arc>, + armed: std::sync::Arc, +} + +impl WalTrace { + pub(super) fn install() -> Self { + let events: std::sync::Arc> = Default::default(); + let armed: std::sync::Arc = Default::default(); + tracing::subscriber::set_global_default(WalSubscriber { + next: Default::default(), + n1_spans: Default::default(), + armed: armed.clone(), + events: events.clone(), + }) + .expect("crash/rejoin test needs its own WAL path subscriber"); + Self { events, armed } + } + pub(super) fn arm(&self) { + self.armed.store(true, Ordering::SeqCst); + } + pub(super) fn assert_startup(&self) { + let trace = self.events.lock().unwrap(); + println!( + "rejoin n1 WAL startup facts={:?} overflow={}", + trace.facts, trace.overflow + ); + assert_eq!(trace.overflow, 0, "WAL startup trace saturated"); + let first_running = trace + .facts + .iter() + .position(|f| *f == WalFact::Running) + .expect("restarted n1 never reached consensus Running"); + let startup = &trace.facts[..=first_running]; + assert!( + !startup.contains(&WalFact::Error), + "restarted n1 reported consensus error" + ); + let path = if startup.contains(&WalFact::EmptyFetch) { + "valid empty WAL" + } else if startup.contains(&WalFact::NonemptyFetch) + && startup.contains(&WalFact::Deferred) + && startup.contains(&WalFact::Superseded) + { + "sync certificate superseded delayed WAL" + } else if startup.contains(&WalFact::NonemptyFetch) && startup.contains(&WalFact::Replaying) + { + "nonempty WAL replay completed" + } else { + panic!("restarted n1 WAL startup path not established: {startup:?}") + }; + println!("rejoin n1 WAL startup path={path}"); + } +} + +impl Drop for WalTrace { + fn drop(&mut self) { + let trace = self.events.lock().unwrap(); + println!( + "rejoin n1 WAL final={:?} overflow={}", + trace.facts, trace.overflow + ); + self.armed.store(false, Ordering::SeqCst); + } +} + +#[test] +fn wal_subscriber_scopes_startup_and_never_records_arbitrary_messages() { + let events: std::sync::Arc> = Default::default(); + let armed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); + let subscriber = WalSubscriber { + next: Default::default(), + n1_spans: Default::default(), + armed: armed.clone(), + events: events.clone(), + }; + let dispatch = tracing::Dispatch::new(subscriber); + tracing::dispatcher::with_default(&dispatch, || { + let other = tracing::error_span!("node", moniker = %"room-node-2"); + { + let _guard = other.enter(); + tracing::debug!(target: "consensus", "No WAL entries to replay"); + } + let n1 = tracing::error_span!("node", moniker = %"room-node-1"); + let consensus = tracing::error_span!(parent: &n1, "consensus"); + let _guard = consensus.enter(); + tracing::debug!(target: "consensus", "No WAL entries to replay"); + tracing::info!(target: "consensus", prev = ?"Recovering", new = ?"Running", "Phase transition"); + tracing::info!(target: "consensus", "private home /tmp/secret WAL entry"); + }); + assert_eq!( + events.lock().unwrap().facts, + vec![WalFact::EmptyFetch, WalFact::Running] + ); + WalTrace { events, armed }.assert_startup(); +} diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index 719f85fd..e17bfb6f 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -222,6 +222,7 @@ async fn late_joiner_syncs_and_commits() { /// journal frontier, and the node rejoins progress without double-apply. #[tokio::test(flavor = "multi_thread", worker_threads = 8)] async fn crashed_validator_recovers_and_rejoins() { + let wal_trace = rejoin_trace::WalTrace::install(); const HEIGHTS: u64 = 4; let plan = batch_plan(HEIGHTS); // Stage h3-h4 only after the victim stops. Otherwise the four nodes can @@ -247,7 +248,7 @@ async fn crashed_validator_recovers_and_rejoins() { held: first_two.clone(), genesis: genesis(), wal_faults: None, - net_gate: None, + net_gate: Some(NetGate::new()), }) .await, ); @@ -334,6 +335,16 @@ async fn crashed_validator_recovers_and_rejoins() { } } + // Attach the bounded real-wire observers to live sources before n1 can + // request history. Close only n1's test network proxy until its observer + // subscribes, so the earliest decoded reply cannot race that attachment. + let mut wire = rejoin_trace::WireTrace::new(&keys); + for (i, node) in nodes.iter().enumerate() { + wire.attach(node, i + 2).await; + } + let restart_gate = NetGate::new(); + restart_gate.close(); + wal_trace.arm(); let restarted = RoomNode::start(NodeSpec { home: base.join("n1"), config: node_config(1, 4, base_port), @@ -342,10 +353,13 @@ async fn crashed_validator_recovers_and_rejoins() { held: first_two.clone(), genesis: genesis(), wal_faults: None, - net_gate: None, + net_gate: Some(restart_gate.clone()), }) .await; + wire.attach(&restarted, 1).await; + restart_gate.reopen(); + // The application-owned store survived the crash: both pre-crash // spec-held batch bodies and the seen-proposal records reloaded. The // restarted node does not receive h3-h4 through its spec; those values @@ -380,6 +394,9 @@ async fn crashed_validator_recovers_and_rejoins() { ) .await; + wal_trace.assert_startup(); + wire.assert_causal_page(&restarted, &nodes, &keys).await; + assert_eq!( rejoin_trace::local_history(&restarted, 3, 4), rejoin_trace::LocalHistory::Values(2), From 009c292924eb9c6e57cdc444ca2cd89ed85f612d Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:18:19 -0400 Subject: [PATCH 10/11] test(rooms-node): scope WAL trace to restarted height --- .../src/unix/rejoin_trace.rs | 80 +++++++++++-------- 1 file changed, 46 insertions(+), 34 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 55237ece..47cb1bee 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -722,7 +722,10 @@ impl WireTrace { impl Drop for WireTrace { fn drop(&mut self) { - let snapshot = self.events.lock().unwrap(); + let snapshot = self + .events + .lock() + .unwrap_or_else(|poison| poison.into_inner()); // Failure snapshots also include this bounded, redacted wire ledger. println!( "rejoin wire final={:?} overflow={}", @@ -736,7 +739,8 @@ impl Drop for WireTrace { // A minimal test-only tracing subscriber for the pinned consensus actor. It // never formats arbitrary log fields (WAL entries can contain private data). -// It records only whitelisted startup branch names under n1's node span. +// It records only whitelisted startup branches at h3 after all live peers +// reached h4; the restarted n1 is the sole actor starting that height. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum WalFact { EmptyFetch, @@ -760,18 +764,18 @@ thread_local! { } struct SafeFields { - moniker: Option, + height: Option, message: Option, new: Option, } impl tracing::field::Visit for SafeFields { fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { - if matches!(field.name(), "moniker" | "message" | "new") { + if matches!(field.name(), "height" | "message" | "new") { let value = format!("{value:?}"); if value.len() <= 128 { match field.name() { - "moniker" => self.moniker = Some(value), + "height" => self.height = Some(value), "message" => self.message = Some(value), "new" => self.new = Some(value), _ => {} @@ -783,7 +787,7 @@ impl tracing::field::Visit for SafeFields { struct WalSubscriber { next: std::sync::atomic::AtomicU64, - n1_spans: std::sync::Mutex>, + height3_spans: std::sync::Mutex>, armed: std::sync::Arc, events: std::sync::Arc>, } @@ -797,24 +801,25 @@ impl tracing::Subscriber for WalSubscriber { fn new_span(&self, attrs: &tracing::span::Attributes<'_>) -> tracing::span::Id { let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; let mut fields = SafeFields { - moniker: None, + height: None, message: None, new: None, }; - if attrs.metadata().name() == "node" { - attrs.record(&mut fields); - } + attrs.record(&mut fields); let parent = attrs .parent() .map(tracing::span::Id::into_u64) .or_else(|| WAL_SPAN_STACK.with(|stack| stack.borrow().last().copied())); - if fields - .moniker + let height_is_three = fields + .height .as_deref() - .is_some_and(|v| v.trim_matches('"') == "room-node-1") - || parent.is_some_and(|parent| self.n1_spans.lock().unwrap().contains(&parent)) + .is_some_and(|v| v.trim_matches('"') == "3"); + if height_is_three + || (fields.height.is_none() + && parent + .is_some_and(|parent| self.height3_spans.lock().unwrap().contains(&parent))) { - self.n1_spans.lock().unwrap().insert(id); + self.height3_spans.lock().unwrap().insert(id); } tracing::span::Id::from_u64(id) } @@ -828,15 +833,21 @@ impl tracing::Subscriber for WalSubscriber { .parent() .map(tracing::span::Id::into_u64) .or_else(|| WAL_SPAN_STACK.with(|stack| stack.borrow().last().copied())); - if !parent.is_some_and(|parent| self.n1_spans.lock().unwrap().contains(&parent)) { - return; - } let mut fields = SafeFields { - moniker: None, + height: None, message: None, new: None, }; event.record(&mut fields); + let height_is_three = fields + .height + .as_deref() + .is_some_and(|v| v.trim_matches('"') == "3"); + if !height_is_three + && !parent.is_some_and(|parent| self.height3_spans.lock().unwrap().contains(&parent)) + { + return; + } let message = fields.message.as_deref().unwrap_or("").trim_matches('"'); let fact = if message == "No WAL entries to replay" { Some(WalFact::EmptyFetch) @@ -892,7 +903,7 @@ impl WalTrace { let armed: std::sync::Arc = Default::default(); tracing::subscriber::set_global_default(WalSubscriber { next: Default::default(), - n1_spans: Default::default(), + height3_spans: Default::default(), armed: armed.clone(), events: events.clone(), }) @@ -903,18 +914,17 @@ impl WalTrace { self.armed.store(true, Ordering::SeqCst); } pub(super) fn assert_startup(&self) { - let trace = self.events.lock().unwrap(); - println!( - "rejoin n1 WAL startup facts={:?} overflow={}", - trace.facts, trace.overflow - ); - assert_eq!(trace.overflow, 0, "WAL startup trace saturated"); - let first_running = trace - .facts + let (facts, overflow) = { + let trace = self.events.lock().unwrap(); + (trace.facts.clone(), trace.overflow) + }; + println!("rejoin n1 WAL startup facts={facts:?} overflow={overflow}"); + assert_eq!(overflow, 0, "WAL startup trace saturated"); + let first_running = facts .iter() .position(|f| *f == WalFact::Running) .expect("restarted n1 never reached consensus Running"); - let startup = &trace.facts[..=first_running]; + let startup = &facts[..=first_running]; assert!( !startup.contains(&WalFact::Error), "restarted n1 reported consensus error" @@ -938,7 +948,10 @@ impl WalTrace { impl Drop for WalTrace { fn drop(&mut self) { - let trace = self.events.lock().unwrap(); + let trace = self + .events + .lock() + .unwrap_or_else(|poison| poison.into_inner()); println!( "rejoin n1 WAL final={:?} overflow={}", trace.facts, trace.overflow @@ -953,19 +966,18 @@ fn wal_subscriber_scopes_startup_and_never_records_arbitrary_messages() { let armed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); let subscriber = WalSubscriber { next: Default::default(), - n1_spans: Default::default(), + height3_spans: Default::default(), armed: armed.clone(), events: events.clone(), }; let dispatch = tracing::Dispatch::new(subscriber); tracing::dispatcher::with_default(&dispatch, || { - let other = tracing::error_span!("node", moniker = %"room-node-2"); + let other = tracing::error_span!("consensus", height = %4); { let _guard = other.enter(); tracing::debug!(target: "consensus", "No WAL entries to replay"); } - let n1 = tracing::error_span!("node", moniker = %"room-node-1"); - let consensus = tracing::error_span!(parent: &n1, "consensus"); + let consensus = tracing::error_span!("consensus", height = %3); let _guard = consensus.enter(); tracing::debug!(target: "consensus", "No WAL entries to replay"); tracing::info!(target: "consensus", prev = ?"Recovering", new = ?"Running", "Phase transition"); From 90e361fe5ebc7e9846f723654e13c89372cbcc47 Mon Sep 17 00:00:00 2001 From: 0thernet <894119+0thernet@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:30:51 -0400 Subject: [PATCH 11/11] test(rooms-node): retain bounded restart trace in CI summary --- .../src/unix/rejoin_trace.rs | 60 ++++++++++++++++++- crates/vhalla-rooms-node/src/unix/tests.rs | 5 +- 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs index 47cb1bee..9377d25b 100644 --- a/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs +++ b/crates/vhalla-rooms-node/src/unix/rejoin_trace.rs @@ -658,7 +658,7 @@ impl WireTrace { restarted: &RoomNode, sources: &[RoomNode], keys: &[PrivateKey], - ) { + ) -> Vec { let responders = { let snapshot = self.events.lock().unwrap(); println!( @@ -689,6 +689,7 @@ impl WireTrace { responders }; let victim_dump = self.dump(restarted).await; + let mut verified = Vec::new(); for source in responders { let source_key = keys[source - 1].public_key(); let source_peer = @@ -716,7 +717,39 @@ impl WireTrace { "decided-page source lacks the victim's verified validator proof" ); println!("rejoin verified peer lifecycle: n1<->n{source} validator proofs match"); + verified.push(source); } + verified + } + + // Nextest captures passing tests' stdout. Write the same capped, redacted + // test facts to the GitHub job summary ONLY after all recovery assertions + // pass; do not report the prior intermittent timeout as explained. + pub(super) fn write_ci_receipt(&self, wal_path: &str, verified: &[usize]) { + if std::env::var("GITHUB_ACTIONS").as_deref() != Ok("true") { + return; + } + let path = std::env::var_os("GITHUB_STEP_SUMMARY") + .expect("GitHub Actions must provide its job summary path"); + let events = self.events.lock().unwrap(); + assert_eq!(events.overflow, 0, "wire summary cannot omit events"); + let facts = format!("{:?}", events.facts); + assert!(facts.len() <= 16_384, "redacted wire summary too long"); + let receipt = format!( + "\n### Rooms-node crash/rejoin diagnostic (test-only)\n\n\ + - Restarted n1 WAL startup at h3: {wal_path}.\n\ + - Source/victim pairs with matching verified validator proofs: {verified:?}.\n\ + - Wire facts (node numbers and heights only): `{facts}`.\n\ + - Same-home n1 stopped at h2 without preloaded h3–h4 bodies, then reached h4; journal prefix, peer state, planned effects, and stopped-store reopen matched.\n\ + - A passing run does not explain the earlier intermittent h4 timeout or prove application-call count.\n" + ); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open GitHub test summary"); + std::io::Write::write_all(&mut file, receipt.as_bytes()) + .expect("write bounded GitHub test summary"); } } @@ -764,6 +797,7 @@ thread_local! { } struct SafeFields { + capture_text: bool, height: Option, message: Option, new: Option, @@ -771,7 +805,9 @@ struct SafeFields { impl tracing::field::Visit for SafeFields { fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { - if matches!(field.name(), "height" | "message" | "new") { + if field.name() == "height" + || (self.capture_text && matches!(field.name(), "message" | "new")) + { let value = format!("{value:?}"); if value.len() <= 128 { match field.name() { @@ -801,6 +837,7 @@ impl tracing::Subscriber for WalSubscriber { fn new_span(&self, attrs: &tracing::span::Attributes<'_>) -> tracing::span::Id { let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; let mut fields = SafeFields { + capture_text: false, height: None, message: None, new: None, @@ -833,7 +870,23 @@ impl tracing::Subscriber for WalSubscriber { .parent() .map(tracing::span::Id::into_u64) .or_else(|| WAL_SPAN_STACK.with(|stack| stack.borrow().last().copied())); + // At this pinned Malachite revision, only these consensus.rs + // callsites contain fixed startup labels or bounded entry counts. + // Never format arbitrary consensus messages (they may contain WAL + // entry bytes or home paths), even temporarily. + let meta = event.metadata(); + let pinned_startup = meta + .file() + .is_some_and(|f| f.ends_with("/engine/src/consensus.rs")) + && meta + .line() + .is_some_and(|line| [314, 529, 820, 945, 952, 980, 1143].contains(&line)); + let synthetic_test = meta + .file() + .is_some_and(|f| f.ends_with("/unix/rejoin_trace.rs")) + && meta.target() == "consensus"; let mut fields = SafeFields { + capture_text: pinned_startup || synthetic_test, height: None, message: None, new: None, @@ -913,7 +966,7 @@ impl WalTrace { pub(super) fn arm(&self) { self.armed.store(true, Ordering::SeqCst); } - pub(super) fn assert_startup(&self) { + pub(super) fn assert_startup(&self) -> &'static str { let (facts, overflow) = { let trace = self.events.lock().unwrap(); (trace.facts.clone(), trace.overflow) @@ -943,6 +996,7 @@ impl WalTrace { panic!("restarted n1 WAL startup path not established: {startup:?}") }; println!("rejoin n1 WAL startup path={path}"); + path } } diff --git a/crates/vhalla-rooms-node/src/unix/tests.rs b/crates/vhalla-rooms-node/src/unix/tests.rs index e17bfb6f..dbf4ec6d 100644 --- a/crates/vhalla-rooms-node/src/unix/tests.rs +++ b/crates/vhalla-rooms-node/src/unix/tests.rs @@ -394,8 +394,8 @@ async fn crashed_validator_recovers_and_rejoins() { ) .await; - wal_trace.assert_startup(); - wire.assert_causal_page(&restarted, &nodes, &keys).await; + let wal_path = wal_trace.assert_startup(); + let verified = wire.assert_causal_page(&restarted, &nodes, &keys).await; assert_eq!( rejoin_trace::local_history(&restarted, 3, 4), @@ -506,6 +506,7 @@ async fn crashed_validator_recovers_and_rejoins() { assert_eq!(account.spent, 1, "persisted charges at n{index}"); } } + wire.write_ci_receipt(wal_path, &verified); let _ = std::fs::remove_dir_all(&base); }