From de327e45276c9f0b42d69f3ea83fa6daee80b158 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 5 Jun 2026 03:32:57 +0000 Subject: [PATCH 1/2] test: run node-level mesh tests over an in-process loopback transport Add a Loopback variant to TransportHandle backed by an unbounded in-process channel and a shared address-to-receiver registry, so node-level multi-node tests deliver packets directly between nodes instead of over real localhost UDP sockets. This removes the kernel UDP receive-buffer overflow that dropped handshake packets when many tests ran in parallel under CPU contention, and lets the large-network convergence tests run reliably in the default suite again (their parallel-load ignore markers are removed). The new transport and its enum variant are cfg(test)-gated, so the daemon build is unaffected. --- src/node/tests/ble.rs | 4 +- src/node/tests/bloom.rs | 1 - src/node/tests/discovery.rs | 1 - src/node/tests/ethernet.rs | 2 +- src/node/tests/mod.rs | 2 +- src/node/tests/routing.rs | 2 - src/node/tests/session.rs | 2 - src/node/tests/spanning_tree.rs | 101 ++++++++++++------ src/node/tests/tcp.rs | 2 +- src/transport/loopback.rs | 174 ++++++++++++++++++++++++++++++++ src/transport/mod.rs | 53 ++++++++++ 11 files changed, 303 insertions(+), 41 deletions(-) create mode 100644 src/transport/loopback.rs diff --git a/src/node/tests/ble.rs b/src/node/tests/ble.rs index a88f997..696ed7d 100644 --- a/src/node/tests/ble.rs +++ b/src/node/tests/ble.rs @@ -63,7 +63,7 @@ async fn make_test_node_ble(node_num: u8) -> TestNode { TestNode { node, transport_id, - packet_rx, + packet_rx: spanning_tree::bridge_to_unbounded(packet_rx), addr: ta, } } @@ -303,7 +303,7 @@ async fn test_ble_discovery() { let mut nodes = vec![TestNode { node, transport_id, - packet_rx, + packet_rx: spanning_tree::bridge_to_unbounded(packet_rx), addr: ta, }]; cleanup_nodes(&mut nodes).await; diff --git a/src/node/tests/bloom.rs b/src/node/tests/bloom.rs index b03e073..ccac45b 100644 --- a/src/node/tests/bloom.rs +++ b/src/node/tests/bloom.rs @@ -535,7 +535,6 @@ fn compute_mesh_size_skips_parent_under_stale_peer_declaration() { /// 100-node random graph: bloom filter exchange at scale. #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_bloom_filter_convergence_100_nodes() { let _guard = lock_large_network_test().await; diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index 458201d..5af82c7 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -774,7 +774,6 @@ async fn test_apply_outgoing_link_mtu_to_response_unknown_peer_noop() { } #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_response_path_mtu_three_node_chain() { // Topology: node0 — node1 — node2 // Node0 initiates lookup for node2. The response travels node2→node1→node0. diff --git a/src/node/tests/ethernet.rs b/src/node/tests/ethernet.rs index 6ef316c..227ef3a 100644 --- a/src/node/tests/ethernet.rs +++ b/src/node/tests/ethernet.rs @@ -104,7 +104,7 @@ async fn make_test_node_ethernet(interface: &str) -> TestNode { TestNode { node, transport_id, - packet_rx, + packet_rx: spanning_tree::bridge_to_unbounded(packet_rx), addr, } } diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 78aa4af..e8df952 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -1,6 +1,6 @@ use super::*; use crate::PeerIdentity; -use crate::transport::{LinkDirection, TransportAddr, packet_channel}; +use crate::transport::{LinkDirection, ReceivedPacket, TransportAddr, packet_channel}; use crate::utils::index::SessionIndex; use std::time::Duration; diff --git a/src/node/tests/routing.rs b/src/node/tests/routing.rs index ee7609e..0fb4449 100644 --- a/src/node/tests/routing.rs +++ b/src/node/tests/routing.rs @@ -670,7 +670,6 @@ fn simulate_forwarding( /// forwarding between every pair of nodes. Every packet must be delivered /// without loops. #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_routing_reachability_100_nodes() { let _guard = lock_large_network_test().await; @@ -991,7 +990,6 @@ async fn test_routing_bloom_only_transit() { /// routing needs dest_coords at each hop for loop-free forwarding through /// non-adjacent nodes. Direct peer adjacency handles the last hop. #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_routing_source_only_coords_100_nodes() { let _guard = lock_large_network_test().await; diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 014a7d2..40907aa 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -571,7 +571,6 @@ async fn drain_to_quiescence(nodes: &mut [TestNode]) { } #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_session_100_nodes() { let _guard = lock_large_network_test().await; @@ -1252,7 +1251,6 @@ async fn test_tun_outbound_3node_forwarded() { } #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_tun_outbound_pending_queue_flush() { // Send multiple packets before session exists — all should be delivered let edges = vec![(0, 1)]; diff --git a/src/node/tests/spanning_tree.rs b/src/node/tests/spanning_tree.rs index a17b100..70f31aa 100644 --- a/src/node/tests/spanning_tree.rs +++ b/src/node/tests/spanning_tree.rs @@ -6,11 +6,51 @@ use super::*; use crate::protocol::TreeAnnounce; +use crate::transport::loopback::{LoopbackRegistry, LoopbackTransport, new_registry}; use crate::tree::{CoordEntry, ParentDeclaration, TreeCoordinate}; static LARGE_NETWORK_TEST_LOCK: std::sync::LazyLock> = std::sync::LazyLock::new(|| tokio::sync::Mutex::new(())); +/// Process-wide shared loopback registry for node-level mesh tests. +/// +/// All loopback test nodes register here so they can locate each other by +/// address. Each node gets a unique synthetic address (`loopback:{n}`) from +/// `LOOPBACK_ADDR_COUNTER`, so addresses never collide across concurrently +/// running tests and stale entries from finished tests are harmless. +static LOOPBACK_REGISTRY: std::sync::LazyLock = + std::sync::LazyLock::new(new_registry); + +static LOOPBACK_ADDR_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +/// Allocate the next globally-unique loopback address. +fn next_loopback_addr() -> TransportAddr { + let n = LOOPBACK_ADDR_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + TransportAddr::from_string(&format!("loopback:{}", n)) +} + +/// Bridge a transport's bounded receive channel into the unbounded channel +/// that `TestNode` holds. +/// +/// Real transports (TCP, Ethernet, BLE) drain their kernel socket into a +/// bounded `PacketRx` via a background receive task, so a bounded channel +/// does not deadlock for them. `TestNode.packet_rx` is unbounded (required +/// by the loopback path, which has no background reader); this spawns a +/// forwarding task so non-loopback factories can still produce a `TestNode`. +pub(super) fn bridge_to_unbounded( + mut bounded_rx: PacketRx, +) -> tokio::sync::mpsc::UnboundedReceiver { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(packet) = bounded_rx.recv().await { + if tx.send(packet).is_err() { + break; + } + } + }); + rx +} + pub(super) async fn lock_large_network_test() -> tokio::sync::MutexGuard<'static, ()> { LARGE_NETWORK_TEST_LOCK.lock().await } @@ -19,52 +59,42 @@ pub(super) async fn lock_large_network_test() -> tokio::sync::MutexGuard<'static pub(super) struct TestNode { pub(super) node: Node, pub(super) transport_id: TransportId, - pub(super) packet_rx: PacketRx, + pub(super) packet_rx: tokio::sync::mpsc::UnboundedReceiver, pub(super) addr: TransportAddr, } -/// Create a test node with a live UDP transport on localhost. +/// Create a test node with an in-process loopback transport. pub(super) async fn make_test_node() -> TestNode { make_test_node_with_mtu(1280).await } /// Create a test node with a specific transport MTU. +/// +/// Uses the in-process loopback transport (not real UDP): packets are +/// delivered directly to the destination node's unbounded receive channel +/// via the shared registry. This avoids the kernel UDP receive-buffer +/// overflow that drops handshake packets when many tests run in parallel +/// under CPU contention. The `mtu` is enforced on send (MtuExceeded), +/// mirroring UDP, so heterogeneous-MTU / PMTUD tests still exercise the +/// forward-path bottleneck. pub(super) async fn make_test_node_with_mtu(mtu: u16) -> TestNode { - use crate::config::UdpConfig; - use crate::transport::udp::UdpTransport; - let mut node = make_node(); let transport_id = TransportId::new(1); - // recv_buf_size and packet_channel are sized for large-network harness - // tests (100-node burst patterns) under parallel-CPU load via - // `cargo test --lib`. The daemon's 2 MB recv default is already - // requested via UdpConfig; we ask for 8 MB so hosts with tuned - // net.core.rmem_max get the larger budget (the kernel clamps to - // rmem_max otherwise and the transport emits a warn). The - // packet_channel(8192) is the actually-effective bump on hosts with - // the typical 2 MB rmem_max — under parallel-test scheduler - // contention the in-process channel between recv loop and the test's - // packet_rx fills well before the kernel rcvbuf would. - let udp_config = UdpConfig { - bind_addr: Some("127.0.0.1:0".to_string()), - mtu: Some(mtu), - recv_buf_size: Some(8 * 1024 * 1024), - ..Default::default() - }; + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::(); + let addr = next_loopback_addr(); - let (packet_tx, packet_rx) = packet_channel(8192); - let mut transport = UdpTransport::new(transport_id, None, udp_config, packet_tx); - transport.start_async().await.unwrap(); + LOOPBACK_REGISTRY.lock().unwrap().insert(addr.clone(), tx); - let addr = TransportAddr::from_string(&transport.local_addr().unwrap().to_string()); + let loopback = + LoopbackTransport::with_mtu(transport_id, addr.clone(), mtu, LOOPBACK_REGISTRY.clone()); node.transports - .insert(transport_id, TransportHandle::Udp(transport)); + .insert(transport_id, TransportHandle::Loopback(loopback)); TestNode { node, transport_id, - packet_rx, + packet_rx: rx, addr, } } @@ -240,9 +270,21 @@ pub(super) async fn process_available_packets(nodes: &mut [TestNode]) -> usize { COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2, }; + // Snapshot the number of packets queued at every node at the start of the + // pass, before processing any node. Loopback delivery is synchronous, so a + // packet sent during this pass would otherwise land in another node's + // channel and be drained in the *same* pass. Real UDP defers such packets + // to the next pass (socket round-trip + recv task), and several tests + // depend on that one-hop-per-pass cadence. Bounding each node's drain to + // its start-of-pass count preserves it regardless of iteration order. + let queued: Vec = nodes.iter().map(|n| n.packet_rx.len()).collect(); + let mut count = 0; - for node in nodes.iter_mut() { - while let Ok(packet) = node.packet_rx.try_recv() { + for (node, &queued) in nodes.iter_mut().zip(queued.iter()) { + for _ in 0..queued { + let Ok(packet) = node.packet_rx.try_recv() else { + break; + }; if packet.data.len() < COMMON_PREFIX_SIZE { continue; } @@ -684,7 +726,6 @@ pub(super) async fn cleanup_nodes(nodes: &mut [TestNode]) { /// Integration test: 100 nodes with random connectivity converge to a /// consistent spanning tree with the correct root. #[tokio::test] -#[ignore = "parallel-load flake class — re-enable when fixed (run solo with --ignored or --test-threads=1 in the meantime)"] async fn test_spanning_tree_convergence_100_nodes() { let _guard = lock_large_network_test().await; diff --git a/src/node/tests/tcp.rs b/src/node/tests/tcp.rs index 18fe25c..fe1c39f 100644 --- a/src/node/tests/tcp.rs +++ b/src/node/tests/tcp.rs @@ -44,7 +44,7 @@ async fn make_test_node_tcp() -> TestNode { TestNode { node, transport_id, - packet_rx, + packet_rx: spanning_tree::bridge_to_unbounded(packet_rx), addr, } } diff --git a/src/transport/loopback.rs b/src/transport/loopback.rs new file mode 100644 index 0000000..d0f2d33 --- /dev/null +++ b/src/transport/loopback.rs @@ -0,0 +1,174 @@ +//! In-process loopback transport (test harness only). +//! +//! Delivers packets directly between nodes running in the same process via +//! an unbounded in-process channel and a shared address-to-receiver +//! registry, instead of going over real localhost UDP sockets. This is used +//! by node-level multi-node tests to avoid the kernel UDP receive-buffer +//! overflow that drops handshake packets when many tests run in parallel +//! under CPU contention. +//! +//! An UNBOUNDED channel is used deliberately: the test harness drains +//! packets sequentially (it fires the whole handshake burst before draining, +//! with no background reader), so a bounded awaiting send would deadlock and +//! a bounded try_send would drop. `UnboundedSender::send` is synchronous, +//! never blocks, and never drops — it only errors if the receiver is gone — +//! making delivery provably lossless and deadlock-free. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use tokio::sync::mpsc::UnboundedSender; + +use super::{ + DiscoveredPeer, ReceivedPacket, Transport, TransportAddr, TransportError, TransportId, + TransportState, TransportType, +}; + +/// Shared registry mapping each loopback address to the receiver-side +/// channel sender for the node listening on that address. +/// +/// One registry instance is shared by all loopback transports in a given +/// test run so they can locate each other by address. +pub type LoopbackRegistry = Arc>>>; + +/// Create a fresh, empty loopback registry. +pub fn new_registry() -> LoopbackRegistry { + Arc::new(Mutex::new(HashMap::new())) +} + +/// Default loopback MTU, mirroring the UDP test path. +const DEFAULT_LOOPBACK_MTU: u16 = 1280; + +/// In-process loopback transport. +pub struct LoopbackTransport { + transport_id: TransportId, + /// This transport's synthetic unique address (e.g. "loopback:7"). + my_addr: TransportAddr, + /// Transport MTU. Enforced on send to mirror UDP's MtuExceeded behavior, + /// so PMTUD/heterogeneous-MTU tests still exercise the forward-path + /// bottleneck detection. + mtu: u16, + /// Shared address-to-receiver registry. + registry: LoopbackRegistry, +} + +impl LoopbackTransport { + /// Create a new loopback transport bound to `my_addr` with the default + /// MTU, sharing `registry`. + pub fn new( + transport_id: TransportId, + my_addr: TransportAddr, + registry: LoopbackRegistry, + ) -> Self { + Self::with_mtu(transport_id, my_addr, DEFAULT_LOOPBACK_MTU, registry) + } + + /// Create a new loopback transport with an explicit MTU. + pub fn with_mtu( + transport_id: TransportId, + my_addr: TransportAddr, + mtu: u16, + registry: LoopbackRegistry, + ) -> Self { + Self { + transport_id, + my_addr, + mtu, + registry, + } + } + + /// This transport's synthetic loopback address. + pub fn my_addr(&self) -> &TransportAddr { + &self.my_addr + } + + /// Send data to a destination loopback address. + /// + /// Looks up `dest_addr` in the shared registry and, if found, delivers a + /// `ReceivedPacket` to its receiver. The packet's `remote_addr` is set to + /// the sender's own address (`my_addr`), mirroring how UDP sets + /// `remote_addr` from the datagram source, so the receiver's handlers + /// learn the peer source. + pub async fn send_async( + &self, + dest_addr: &TransportAddr, + data: &[u8], + ) -> Result { + if data.len() > self.mtu as usize { + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.mtu, + }); + } + + let dest_tx = { + let registry = self.registry.lock().map_err(|e| { + TransportError::SendFailed(format!("registry lock poisoned: {}", e)) + })?; + registry.get(dest_addr).cloned() + }; + + match dest_tx { + Some(tx) => { + let packet = + ReceivedPacket::new(self.transport_id, self.my_addr.clone(), data.to_vec()); + tx.send(packet).map_err(|_| { + TransportError::SendFailed(format!("loopback receiver gone for {}", dest_addr)) + })?; + Ok(data.len()) + } + None => Err(TransportError::SendFailed(format!( + "no loopback route to {}", + dest_addr + ))), + } + } + + /// Asynchronous start (no-op; the transport is ready on construction). + pub async fn start_async(&mut self) -> Result<(), TransportError> { + Ok(()) + } + + /// Asynchronous stop (no-op). + pub async fn stop_async(&mut self) -> Result<(), TransportError> { + Ok(()) + } +} + +impl Transport for LoopbackTransport { + fn transport_id(&self) -> TransportId { + self.transport_id + } + + fn transport_type(&self) -> &TransportType { + &TransportType::LOOPBACK + } + + fn state(&self) -> TransportState { + TransportState::Up + } + + fn mtu(&self) -> u16 { + self.mtu + } + + fn start(&mut self) -> Result<(), TransportError> { + Ok(()) + } + + fn stop(&mut self) -> Result<(), TransportError> { + Ok(()) + } + + fn send(&self, _addr: &TransportAddr, _data: &[u8]) -> Result<(), TransportError> { + // Synchronous send not supported — use send_async(). + Err(TransportError::NotSupported( + "use send_async() for loopback transport".into(), + )) + } + + fn discover(&self) -> Result, TransportError> { + Ok(Vec::new()) + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 41d0f8d..4a0e6fc 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -4,6 +4,8 @@ //! underlying communication mechanisms (UDP, Ethernet, Tor, etc.) over //! which FIPS links are established. +#[cfg(test)] +pub mod loopback; pub mod tcp; pub mod tor; pub mod udp; @@ -18,6 +20,8 @@ pub mod ble; use ble::DefaultBleTransport; #[cfg(unix)] use ethernet::EthernetTransport; +#[cfg(test)] +use loopback::LoopbackTransport; use secp256k1::XOnlyPublicKey; use std::fmt; use std::net::SocketAddr; @@ -247,6 +251,14 @@ impl TransportType { reliable: true, // L2CAP SeqPacket guarantees delivery }; + /// In-process loopback transport (test harness only). + #[cfg(test)] + pub const LOOPBACK: TransportType = TransportType { + name: "loopback", + connection_oriented: false, + reliable: true, // in-process channel delivery is lossless + }; + /// Check if the transport is connectionless. pub fn is_connectionless(&self) -> bool { !self.connection_oriented @@ -862,6 +874,9 @@ pub enum TransportHandle { /// BLE L2CAP transport. #[cfg(target_os = "linux")] Ble(DefaultBleTransport), + /// In-process loopback transport (test harness only). + #[cfg(test)] + Loopback(LoopbackTransport), } impl TransportHandle { @@ -875,6 +890,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.start_async().await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.start_async().await, + #[cfg(test)] + TransportHandle::Loopback(t) => t.start_async().await, } } @@ -888,6 +905,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.stop_async().await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.stop_async().await, + #[cfg(test)] + TransportHandle::Loopback(t) => t.stop_async().await, } } @@ -901,6 +920,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.send_async(addr, data).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.send_async(addr, data).await, + #[cfg(test)] + TransportHandle::Loopback(t) => t.send_async(addr, data).await, } } @@ -914,6 +935,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.transport_id(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.transport_id(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.transport_id(), } } @@ -927,6 +950,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.name(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.name(), + #[cfg(test)] + TransportHandle::Loopback(_) => None, } } @@ -940,6 +965,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.transport_type(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.transport_type(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.transport_type(), } } @@ -953,6 +980,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.state(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.state(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.state(), } } @@ -966,6 +995,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.mtu(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.mtu(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.mtu(), } } @@ -982,6 +1013,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.link_mtu(addr), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.link_mtu(addr), + #[cfg(test)] + TransportHandle::Loopback(t) => t.link_mtu(addr), } } @@ -995,6 +1028,8 @@ impl TransportHandle { TransportHandle::Tor(_) => None, #[cfg(target_os = "linux")] TransportHandle::Ble(_) => None, + #[cfg(test)] + TransportHandle::Loopback(_) => None, } } @@ -1008,6 +1043,8 @@ impl TransportHandle { TransportHandle::Tor(_) => None, #[cfg(target_os = "linux")] TransportHandle::Ble(_) => None, + #[cfg(test)] + TransportHandle::Loopback(_) => None, } } @@ -1045,6 +1082,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.discover(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.discover(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.discover(), } } @@ -1058,6 +1097,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.auto_connect(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.auto_connect(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.auto_connect(), } } @@ -1071,6 +1112,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.accept_connections(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.accept_connections(), + #[cfg(test)] + TransportHandle::Loopback(t) => t.accept_connections(), } } @@ -1090,6 +1133,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.connect_async(addr).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.connect_async(addr).await, + #[cfg(test)] + TransportHandle::Loopback(_) => Ok(()), // connectionless } } @@ -1107,6 +1152,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.connection_state_sync(addr), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.connection_state_sync(addr), + #[cfg(test)] + TransportHandle::Loopback(_) => ConnectionState::Connected, } } @@ -1123,6 +1170,8 @@ impl TransportHandle { TransportHandle::Tor(t) => t.close_connection_async(addr).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.close_connection_async(addr).await, + #[cfg(test)] + TransportHandle::Loopback(_) => {} // connectionless no-op } } @@ -1145,6 +1194,8 @@ impl TransportHandle { TransportHandle::Tor(_) => TransportCongestion::default(), #[cfg(target_os = "linux")] TransportHandle::Ble(_) => TransportCongestion::default(), + #[cfg(test)] + TransportHandle::Loopback(_) => TransportCongestion::default(), } } @@ -1182,6 +1233,8 @@ impl TransportHandle { TransportHandle::Ble(t) => { serde_json::to_value(t.stats().snapshot()).unwrap_or_default() } + #[cfg(test)] + TransportHandle::Loopback(_) => serde_json::json!({}), } } } From f29c2e65fa0763f65b8be4cef213d3d199a60b2e Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 5 Jun 2026 04:37:43 +0000 Subject: [PATCH 2/2] test: replace fixed convergence timeouts with progress-aware connectivity wait Add wait_until_connected to the shared convergence helpers: it polls a suite's own pairwise pings (the signal it actually asserts on), returns as soon as every pair is reachable, extends its deadline while the reachable-pair count is still climbing, and gives up only when progress stalls. Use it in the rekey, static-mesh, and sidecar suites in place of the fixed wall-clock baseline timeout and the blind sleep, which timed out under concurrent CI load while the mesh was still converging. --- testing/lib/wait-converge.sh | 56 +++++++++++++++++- testing/sidecar/scripts/test-sidecar.sh | 15 ++++- testing/static/scripts/ping-test.sh | 25 ++------- testing/static/scripts/rekey-test.sh | 75 ++++++++++++++----------- 4 files changed, 117 insertions(+), 54 deletions(-) diff --git a/testing/lib/wait-converge.sh b/testing/lib/wait-converge.sh index 3f64803..c203eca 100644 --- a/testing/lib/wait-converge.sh +++ b/testing/lib/wait-converge.sh @@ -1,12 +1,14 @@ #!/bin/bash # Shared convergence wait helpers for FIPS integration tests. # -# Source this file to get wait_for_links() and wait_for_peers(). +# Source this file to get wait_for_links(), wait_for_peers(), and +# wait_until_connected(). # # Usage: # source "$(dirname "$0")/../../lib/wait-converge.sh" # wait_for_links [timeout_secs] # wait_for_peers [timeout_secs] +# wait_until_connected [poll_secs] # Wait until a container has at least min_links active links. # Returns 0 on success, 1 on timeout. @@ -49,3 +51,55 @@ wait_for_peers() { echo " $container: TIMEOUT waiting for $min_peers peer(s) after ${timeout}s" return 1 } + +# Wait until a connectivity check reports every pair reachable, using a +# progress-aware deadline instead of a fixed one. +# +# wait_until_connected [poll_secs] +# +# is the name of a function that runs the suite's own +# connectivity check and sets two globals each call: +# PASSED number of reachable pairs this round +# FAILED number of unreachable pairs this round +# +# The convergence signal is the suite's real pings (the same signal it +# asserts on), not a structural proxy. Behaviour: +# - converged: FAILED == 0 -> return 0. +# - progressing: PASSED climbed past the best seen -> reset the stall +# clock and keep waiting (slow-but-improving is not a failure, so it +# does not false-time-out under CI load). +# - stuck: PASSED has not improved for stall_secs -> return 1 (fail +# fast rather than burn the whole budget on a genuinely wedged pair). +# - hard cap: max_secs elapsed -> return 1 (never runs unbounded). +# +# Returns 0 once fully connected, 1 on stall or timeout. +wait_until_connected() { + local ping_fn="$1" + local max_secs="$2" + local stall_secs="$3" + local poll_secs="${4:-1}" + + local start_secs=$SECONDS + local best=-1 + local last_progress=$SECONDS + + while (( SECONDS - start_secs < max_secs )); do + "$ping_fn" + if (( FAILED == 0 )); then + echo " converge: all $PASSED pair(s) reachable after $((SECONDS - start_secs))s" + return 0 + fi + if (( PASSED > best )); then + best=$PASSED + last_progress=$SECONDS + echo " converge: $PASSED reachable, $FAILED pending (progressing) after $((SECONDS - start_secs))s" + elif (( SECONDS - last_progress >= stall_secs )); then + echo " converge: STUCK at $PASSED reachable / $FAILED pending — no progress for ${stall_secs}s (after $((SECONDS - start_secs))s)" + return 1 + fi + sleep "$poll_secs" + done + + echo " converge: TIMEOUT at $PASSED reachable / $FAILED pending after ${max_secs}s" + return 1 +} diff --git a/testing/sidecar/scripts/test-sidecar.sh b/testing/sidecar/scripts/test-sidecar.sh index 2087c41..7952c2d 100755 --- a/testing/sidecar/scripts/test-sidecar.sh +++ b/testing/sidecar/scripts/test-sidecar.sh @@ -14,6 +14,8 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" SIDECAR_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" +# shellcheck source=../../lib/wait-converge.sh +source "$SCRIPT_DIR/../../lib/wait-converge.sh" # Deterministic keys derived from: derive-keys.py sidecar-test node-{a,b,c} NODE_A_NSEC="9e688d0879fa9cd025fea0487ac23495080e3de626070fdb9b78dc1f619dd453" @@ -127,8 +129,17 @@ else exit 1 fi -# Allow a few more seconds for tree convergence and coordinate propagation -sleep 3 +# Wait for end-to-end multi-hop connectivity (the same app-to-app pings +# the test asserts on) with a progress-aware deadline, instead of a +# blind fixed sleep that can fire before coordinates propagate across the +# chain. The directed pings below remain the actual assertions. +_sidecar_converged() { + PASSED=0; FAILED=0 + if docker exec sidecar-b-app-1 ping6 -c1 -W2 "${NODE_A_NPUB}.fips" >/dev/null 2>&1; then PASSED=$((PASSED+1)); else FAILED=$((FAILED+1)); fi + if docker exec sidecar-c-app-1 ping6 -c1 -W2 "${NODE_A_NPUB}.fips" >/dev/null 2>&1; then PASSED=$((PASSED+1)); else FAILED=$((FAILED+1)); fi + if docker exec sidecar-a-app-1 ping6 -c1 -W2 "${NODE_C_NPUB}.fips" >/dev/null 2>&1; then PASSED=$((PASSED+1)); else FAILED=$((FAILED+1)); fi +} +wait_until_connected _sidecar_converged "$CONVERGE_TIMEOUT" 10 || true # ── Link verification ───────────────────────────────────────────────────── diff --git a/testing/static/scripts/ping-test.sh b/testing/static/scripts/ping-test.sh index de0ede1..54690b8 100755 --- a/testing/static/scripts/ping-test.sh +++ b/testing/static/scripts/ping-test.sh @@ -71,23 +71,6 @@ ping_all_quiet() { done } -# Wait until all ping pairs succeed or timeout. -wait_for_full_connectivity() { - local timeout="${1:-30}" - local start_secs=$SECONDS - - while (( SECONDS - start_secs < timeout )); do - ping_all_quiet - if [ "$FAILED" -eq 0 ]; then - echo " All $PASSED pairs reachable after $((SECONDS - start_secs))s" - return 0 - fi - sleep 1 - done - echo " TIMEOUT: $PASSED passed, $FAILED failed after ${timeout}s" - return 1 -} - echo "=== FIPS Ping Test ($PROFILE topology) ===" echo "" @@ -108,8 +91,12 @@ elif [ "$PROFILE" = "mesh" ] || [ "$PROFILE" = "mesh-public" ]; then wait_for_peers fips-node-d 3 20 || true wait_for_peers fips-node-e 3 20 || true fi -# Wait for FSP-level connectivity (discovery + session establishment) -wait_for_full_connectivity 30 || true +# Wait for full pairwise connectivity, progress-aware: the actual pings +# are the convergence signal, the deadline extends while more pairs come +# up, and it only gives up if progress stalls — so it does not +# false-time-out under load while routing is still converging. The +# directed-pair ping assertions below remain the actual test. +wait_until_connected ping_all_quiet 45 15 || true # Reset counters for the actual test PASSED=0 diff --git a/testing/static/scripts/rekey-test.sh b/testing/static/scripts/rekey-test.sh index 9b1507f..a81b49b 100755 --- a/testing/static/scripts/rekey-test.sh +++ b/testing/static/scripts/rekey-test.sh @@ -149,8 +149,8 @@ trap 'echo ""; echo "Test interrupted"; exit 130' INT # BASELINE_CONVERGENCE_TIMEOUT must cover one full daemon # node.tree.reeval_interval_secs (default 60) plus a small margin # so any partition that only heals via the periodic TreeAnnounce -# re-broadcast lands inside the convergence window. wait_for_full_baseline -# early-exits on PASS, so successful reps are unaffected by the +# re-broadcast lands inside the convergence window. The Phase-1 baseline +# wait early-exits on PASS, so successful reps are unaffected by the # extra headroom. BASELINE_CONVERGENCE_TIMEOUT=65 REKEY_SETTLE=12 # > DRAIN_WINDOW_SECS (10) so post-rekey samples are off the old session @@ -173,8 +173,8 @@ CONVERGENCE_PING_TIMEOUT=1 # attempts drops the loss-math floor to ~(0.02)^MAX_PING_ATTEMPTS per # pair, making any residual failure attributable to a non-loss # mechanism rather than ICMP noise. Applied to Phase 1 (final strict -# ping_all after wait_for_full_baseline converges) and Phase 3 / Phase -# 5 (post-rekey strict asserts). The wait_for_full_baseline convergence +# ping_all after the Phase-1 baseline wait converges) and Phase 3 / Phase +# 5 (post-rekey strict asserts). The Phase-1 baseline wait's convergence # loop itself stays single-shot — its job is to detect when the mesh # first sees a fully clean 20-pair batch, and retries inside the loop # would conflate "transient ping loss" with "still converging." @@ -255,27 +255,11 @@ ping_all() { done } -wait_for_full_baseline() { - local timeout="${1:-30}" - local start_secs=$SECONDS - local best_passed=0 - local best_failed=20 - - while (( SECONDS - start_secs < timeout )); do - ping_all quiet "$CONVERGENCE_PING_TIMEOUT" - if [ "$PASSED" -gt "$best_passed" ]; then - best_passed="$PASSED" - best_failed="$FAILED" - fi - if [ "$FAILED" -eq 0 ]; then - return 0 - fi - sleep 1 - done - - PASSED="$best_passed" - FAILED="$best_failed" - return 1 +# Connectivity probe for the progress-aware baseline wait: one full +# all-pairs ping sweep, setting PASSED/FAILED (consumed by +# wait_until_connected). +_baseline_ping() { + ping_all quiet "$CONVERGENCE_PING_TIMEOUT" } phase_result() { @@ -370,13 +354,24 @@ echo "Config: rekey.after_secs=$REKEY_AFTER_SECS" echo "" # ── Phase 1: Pre-rekey baseline ─────────────────────────────────────── +# Wait for full pre-rekey connectivity with a progress-aware deadline: +# the all-pairs ping sweep is the convergence signal, the window extends +# while more pairs come up, and it gives up only if progress stalls — so +# it no longer false-times-out under concurrent CI load. The strict +# ping_all below is the actual assertion, run only after convergence. echo "Phase 1: Pre-rekey connectivity (waiting for convergence)" -wait_for_peers fips-node-a 2 "$BASELINE_CONVERGENCE_TIMEOUT" || true -if wait_for_full_baseline "$BASELINE_CONVERGENCE_TIMEOUT"; then +if wait_until_connected _baseline_ping "$BASELINE_CONVERGENCE_TIMEOUT" 20; then ping_all "" "$TIMEOUT" "$MAX_PING_ATTEMPTS" phase_result "Pre-rekey baseline (all 20 pairs)" + if [ "$FAILED" -ne 0 ]; then + echo "" + dump_peer_connectivity + echo "=== Results: $TOTAL_PASSED passed, $TOTAL_FAILED failed ===" + exit 1 + fi else - echo " Best observed baseline before timeout: $PASSED/$((PASSED + FAILED)) passed" + echo " Mesh did not reach a converged tree before timeout" + ping_all quiet "$CONVERGENCE_PING_TIMEOUT" phase_result "Pre-rekey baseline (all 20 pairs)" echo "" dump_peer_connectivity @@ -503,14 +498,30 @@ echo "=== Results: $TOTAL_PASSED passed, $TOTAL_FAILED failed ===" if [ "$TOTAL_FAILED" -eq 0 ]; then exit 0 else - # Dump logs on failure for diagnostics + # Dump logs on failure for diagnostics. + # + # Wider pattern + larger line cap than the original head -30 (which + # truncated before the actual ping-failure timestamp on Phase 5 + # fires, making the post-cutover convergence window invisible to + # offline triage). Two passes per node: + # 1) rekey-related events across the whole run (cap 200 lines). + # 2) Last 80 lines unfiltered, to surface forwarding decisions, + # route lookups, congestion warnings, decrypt errors, and any + # other surrounding context near the failure point. echo "" - echo "=== Node logs (rekey-related) ===" + echo "=== Node logs (rekey-related, head -200) ===" for node in $NODES; do echo "--- node-$node ---" docker logs "fips-node-$node" 2>&1 | \ - grep -E "(rekey|Rekey|cross|Cross|teardown|ERROR|PANIC|K-bit)" | \ - head -30 + grep -E "(rekey|Rekey|cross|Cross|teardown|ERROR|PANIC|K-bit|no route|next hop|TTL exhausted|MTU exceeded|Congestion|decrypt|Decrypt|AEAD|Notify|drain|Drain|promot)" | \ + head -200 + echo "" + done + + echo "=== Node logs (last 80 lines, unfiltered) ===" + for node in $NODES; do + echo "--- node-$node ---" + docker logs "fips-node-$node" 2>&1 | tail -80 echo "" done exit 1