use super::*;
pub(super) const SESSION_100_NODE_NETWORK: &str = "fips-session-100-nodes";
async fn drain_synthetic_dataplane_activity(nodes: &mut [TestNode]) -> usize {
let mut activity = process_available_packets(nodes).await;
for node in nodes {
activity =
activity.saturating_add(node.node.drain_deferred_dataplane_control_turns().await);
}
activity
}
async fn quiesce_synthetic_dataplanes(nodes: &mut [TestNode]) -> usize {
use futures::StreamExt;
const MAX_TURNS: usize = 64;
const COMPLETION_QUIET: Duration = Duration::from_secs(1);
if nodes.is_empty() {
return 0;
}
let mut total = 0usize;
for _ in 0..MAX_TURNS {
let activity = drain_synthetic_dataplane_activity(nodes).await;
total = total.saturating_add(activity);
if activity > 0
|| nodes
.iter()
.any(|node| node.node.dataplane.has_runnable_work())
{
continue;
}
let mut completions = futures::stream::FuturesUnordered::new();
for node in nodes.iter() {
let notify = node.node.dataplane.readiness_notify();
completions.push(async move { notify.notified().await });
}
if tokio::time::timeout(COMPLETION_QUIET, completions.next())
.await
.is_ok()
{
continue;
}
let final_activity = drain_synthetic_dataplane_activity(nodes).await;
total = total.saturating_add(final_activity);
if final_activity == 0
&& !nodes
.iter()
.any(|node| node.node.dataplane.has_runnable_work())
{
return total;
}
}
let runnable = nodes
.iter()
.enumerate()
.filter_map(|(index, node)| node.node.dataplane.has_runnable_work().then_some(index))
.collect::<Vec<_>>();
panic!("synthetic dataplanes did not quiesce; runnable nodes: {runnable:?}");
}
pub(super) async fn replace_session_100_node_carriers(
nodes: &mut [TestNode],
edges: &[(usize, usize)],
) -> crate::SimNetwork {
quiesce_synthetic_dataplanes(nodes).await;
let sim_addresses = (0..nodes.len())
.map(|index| TransportAddr::from_string(&format!("session-node-{index}")))
.collect::<Vec<_>>();
let address_by_node = nodes
.iter()
.zip(&sim_addresses)
.map(|(node, addr)| (*node.node.node_addr(), addr.clone()))
.collect::<std::collections::HashMap<_, _>>();
let network = crate::SimNetwork::new(42);
network.set_default_link(crate::SimLink {
up: false,
..Default::default()
});
for &(left, right) in edges {
network.set_link(
sim_addresses[left].as_str().expect("sim address"),
sim_addresses[right].as_str().expect("sim address"),
crate::SimLink::default(),
);
}
crate::register_sim_network(SESSION_100_NODE_NETWORK, network.clone());
for (index, node) in nodes.iter_mut().enumerate() {
let mut old_transport = node
.node
.transports
.remove(&node.transport_id)
.expect("UDP test transport should exist");
old_transport
.stop()
.await
.expect("UDP test transport should stop");
let peer_addrs = node.node.peers.keys().copied().collect::<Vec<_>>();
for peer_addr in peer_addrs {
let remote_addr = address_by_node
.get(&peer_addr)
.expect("topology peer has a Sim address")
.clone();
let link_id = node
.node
.get_peer(&peer_addr)
.expect("topology peer retained")
.link_id();
let old_link = node
.node
.links
.remove(&link_id)
.expect("topology peer link retained");
let mut new_link = Link::new_with_timestamp(
link_id,
node.transport_id,
remote_addr.clone(),
old_link.direction(),
old_link.base_rtt(),
old_link.created_at(),
);
*new_link.stats_mut() = old_link.stats().clone();
match old_link.state() {
crate::transport::LinkState::Connected => new_link.set_connected(),
crate::transport::LinkState::Disconnected => new_link.set_disconnected(),
crate::transport::LinkState::Failed => new_link.set_failed(),
crate::transport::LinkState::Connecting => {}
}
node.node.links.insert(link_id, new_link);
let peer = node.node.get_peer_mut(&peer_addr).expect("topology peer");
peer.set_current_addr(node.transport_id, &remote_addr);
peer.clear_preferred_send_addr();
}
node.addr = sim_addresses[index].clone();
let (packet_tx, packet_rx) = crate::packet_channel(256);
let config = crate::SimTransportConfig {
network: Some(SESSION_100_NODE_NETWORK.to_string()),
addr: Some(node.addr.as_str().expect("sim address").to_string()),
mtu: Some(1280),
..Default::default()
};
let mut transport = crate::SimTransport::new(node.transport_id, None, config, packet_tx);
transport.start_async().await.expect("sim transport start");
node.packet_rx = packet_rx;
node.node
.transports
.insert(node.transport_id, crate::TransportHandle::Sim(transport));
for peer_addr in node.node.peers.keys().copied().collect::<Vec<_>>() {
assert!(
node.node.sync_dataplane_fmp_owner(&peer_addr),
"retargeted Sim peer owner should install"
);
}
}
network
}
#[tokio::test]
async fn carrier_boundary_drains_leftover_dataplane_work() {
const PACKETS: usize = 300;
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let peer = *nodes[1].node.node_addr();
let heartbeat = [crate::protocol::LinkMessageType::Heartbeat.to_byte()];
let outbound = (0..PACKETS)
.map(|_| {
nodes[0]
.node
.prepare_dataplane_fmp_link_outbound(
peer,
crate::transport::PacketBuffer::new(heartbeat.to_vec()),
false,
crate::dataplane::ActivityTick::new(Node::now_ms()),
)
.expect("prepare synthetic leftover")
.0
})
.collect();
let first = nodes[0]
.node
.pump_dataplane_pending_outbound_firsts(
crate::dataplane::DataplaneLiveOutboundFirsts {
initial_outbound_batch: outbound,
..Default::default()
},
0,
0,
1,
)
.await;
assert_eq!(first.summary().outbound_admitted(), PACKETS);
assert!(nodes[0].node.dataplane.has_runnable_work());
let drained = quiesce_synthetic_dataplanes(&mut nodes).await;
assert!(drained > 0);
assert!(
!nodes
.iter()
.any(|node| node.node.dataplane.has_runnable_work())
);
assert_eq!(process_available_packets(&mut nodes).await, 0);
cleanup_nodes(&mut nodes).await;
}