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.
This commit is contained in:
Johnathan Corgan
2026-06-05 03:32:57 +00:00
parent 4af3730be6
commit de327e4527
11 changed files with 303 additions and 41 deletions
+2 -2
View File
@@ -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;
-1
View File
@@ -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;
-1
View File
@@ -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.
+1 -1
View File
@@ -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,
}
}
+1 -1
View File
@@ -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;
-2
View File
@@ -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;
-2
View File
@@ -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)];
+71 -30
View File
@@ -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<tokio::sync::Mutex<()>> =
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<LoopbackRegistry> =
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<ReceivedPacket> {
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<ReceivedPacket>,
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::<ReceivedPacket>();
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<usize> = 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;
+1 -1
View File
@@ -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,
}
}
+174
View File
@@ -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<Mutex<HashMap<TransportAddr, UnboundedSender<ReceivedPacket>>>>;
/// 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<usize, TransportError> {
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<Vec<DiscoveredPeer>, TransportError> {
Ok(Vec::new())
}
}
+53
View File
@@ -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!({}),
}
}
}