use super::budget::{
ENDPOINT_DRAIN_BUDGET, LATENCY_PACKET_DRAIN_BUDGET, PACKET_DRAIN_BUDGET,
RX_LOOP_FAST_MAINTENANCE_TIMEOUT, TUN_DRAIN_BUDGET, endpoint_drain_budget,
mixed_dataplane_crypto_budget, tun_drain_budget,
};
use super::drain::{
RxLoopDataDrainStats, RxLoopMaintenancePlan, RxLoopMaintenanceState, SingleLaneDrainCursor,
};
use super::rx_loop_fast_maintenance_within_budget;
use crate::control::protocol::Request;
use std::time::{Duration, Instant};
#[tokio::test(start_paused = true)]
async fn fast_maintenance_loses_no_more_than_its_total_budget() {
let started = tokio::time::Instant::now();
let completed = rx_loop_fast_maintenance_within_budget(std::future::pending::<()>()).await;
assert!(!completed);
assert_eq!(started.elapsed(), RX_LOOP_FAST_MAINTENANCE_TIMEOUT);
}
#[test]
fn endpoint_drain_budget_caps_large_packet_turns() {
assert_eq!(endpoint_drain_budget(0), 0);
assert_eq!(endpoint_drain_budget(8), 8);
assert_eq!(
endpoint_drain_budget(PACKET_DRAIN_BUDGET),
ENDPOINT_DRAIN_BUDGET
);
}
#[test]
fn tun_outbound_gets_dataplane_sized_turns() {
assert_eq!(
endpoint_drain_budget(PACKET_DRAIN_BUDGET),
ENDPOINT_DRAIN_BUDGET
);
assert_eq!(tun_drain_budget(PACKET_DRAIN_BUDGET), TUN_DRAIN_BUDGET);
assert_eq!(TUN_DRAIN_BUDGET, LATENCY_PACKET_DRAIN_BUDGET);
const {
assert!(
TUN_DRAIN_BUDGET > ENDPOINT_DRAIN_BUDGET,
"canonical TUN packet ingress must not inherit the endpoint/control slice"
);
}
}
#[test]
fn mixed_packet_and_tun_turn_crypto_budget_covers_admitted_sources() {
let crypto_budget = mixed_dataplane_crypto_budget(
LATENCY_PACKET_DRAIN_BUDGET,
ENDPOINT_DRAIN_BUDGET,
TUN_DRAIN_BUDGET,
);
assert_eq!(
crypto_budget,
LATENCY_PACKET_DRAIN_BUDGET + ENDPOINT_DRAIN_BUDGET + TUN_DRAIN_BUDGET
);
}
#[test]
fn admission_pressure_relief_runs_only_for_bulk_capacity_pressure() {
for (admission_dropped, runnable_work, turns, elapsed, priority_ready, expected) in [
(true, true, 1, Duration::from_micros(100), 0, true),
(false, true, 1, Duration::from_micros(100), 0, false),
(true, false, 1, Duration::from_micros(100), 0, false),
(
true,
true,
super::budget::RX_LOOP_BULK_SERVICE_MAX_TURNS,
Duration::from_micros(100),
0,
false,
),
(
true,
true,
1,
super::budget::RX_LOOP_BULK_SERVICE_MAX_ELAPSED,
0,
false,
),
(true, true, 1, Duration::from_micros(100), 1, false),
] {
assert_eq!(
super::bulk_admission_pressure_relief_due(
admission_dropped,
runnable_work,
turns,
elapsed,
priority_ready,
),
expected
);
}
}
#[test]
fn rx_loop_data_drain_stats_owns_counts_total_and_pressure() {
let empty = RxLoopDataDrainStats::default();
assert_eq!(empty.total(), 0);
assert_eq!(empty.data_total(), 0);
assert!(!empty.has_drained());
assert!(!empty.has_data_drained());
assert!(!empty.data_pressure(false));
assert!(empty.data_pressure(true));
let drained = RxLoopDataDrainStats::new(2, 3, 5, 0);
assert_eq!(drained.data_total(), 10);
assert_eq!(drained.total(), 10);
assert!(drained.has_drained());
assert!(drained.has_data_drained());
assert!(drained.data_pressure(false));
assert!(drained.data_pressure(true));
let control_only = RxLoopDataDrainStats::new(0, 0, 0, 2);
assert_eq!(control_only.data_total(), 0);
assert_eq!(control_only.total(), 2);
assert!(control_only.has_drained());
assert!(!control_only.has_data_drained());
assert!(
!control_only.data_pressure(false),
"read-only control progress must not look like dataplane pressure"
);
}
#[tokio::test]
async fn drain_control_queries_answers_show_requests() {
let mut node =
crate::node::Node::new(crate::config::Config::new()).expect("node should construct");
let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(2);
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
control_tx
.send((
Request {
command: "show_stats_list".to_string(),
params: None,
},
response_tx,
))
.await
.unwrap();
let drained = node.drain_control_queries(&mut control_rx, None, 1).await;
assert_eq!(drained, 1);
let response = response_rx.await.expect("query response");
assert_eq!(response.status, "ok");
assert!(response.data.is_some());
assert!(control_rx.try_recv().is_err());
}
#[tokio::test]
async fn dataplane_turn_uses_rx_loop_owned_channels() {
let mut node =
crate::node::Node::new(crate::config::Config::new()).expect("node should construct");
let (_packet_tx, mut packet_rx) = crate::transport::packet_channel(1);
let (_endpoint_tx, mut endpoint_rx) = crate::node::endpoint_data_batch_channel(1);
let (_tun_outbound_tx, mut tun_outbound_rx) = crate::upper::tun::tun_outbound_channel(1);
let (_fast_tx, mut fast_ingress_rx) = tokio::sync::mpsc::channel(1);
let (_tun_tx, tun_rx) = crate::upper::tun::write_channel();
let mut endpoint_io = node
.attach_endpoint_data_io(1)
.expect("endpoint io should attach before start");
let turn = {
let mut dataplane_io = super::rx_loop_dataplane_io(
&mut packet_rx,
&mut fast_ingress_rx,
&mut endpoint_rx,
&mut tun_outbound_rx,
&endpoint_io.event_tx,
);
node.drain_dataplane_turn_with_firsts(
&mut dataplane_io,
crate::dataplane::DataplaneLiveTurnFirsts::default(),
super::RxLoopDataplaneTurnLimits::new(4, 4, 4, 4),
)
.await
};
assert_eq!(
turn.summary(),
crate::dataplane::DataplaneRuntimeSummary::default()
);
assert!(!turn.has_activity());
assert!(!turn.has_failures());
assert!(turn.raw_ingress_drops().is_empty());
assert!(turn.output_drops().is_empty());
assert!(turn.drops().is_empty());
assert!(turn.endpoint_data_drops().is_empty());
assert!(turn.tun_outbound_drops().is_empty());
assert!(tun_rx.try_recv_packet().is_err());
assert!(endpoint_io.event_rx.try_recv().is_err());
}
#[tokio::test]
async fn dataplane_turn_reports_raw_ingress_failures() {
let mut node =
crate::node::Node::new(crate::config::Config::new()).expect("node should construct");
let (packet_tx, mut packet_rx) = crate::transport::packet_channel(1);
let (_endpoint_tx, mut endpoint_rx) = crate::node::endpoint_data_batch_channel(1);
let (_tun_outbound_tx, mut tun_outbound_rx) = crate::upper::tun::tun_outbound_channel(1);
let (_fast_tx, mut fast_ingress_rx) = tokio::sync::mpsc::channel(1);
let (_tun_tx, tun_rx) = crate::upper::tun::write_channel();
let mut endpoint_io = node
.attach_endpoint_data_io(1)
.expect("endpoint io should attach before start");
packet_tx
.send(crate::transport::ReceivedPacket::with_timestamp(
crate::transport::TransportId::new(7),
crate::transport::TransportAddr::from_string("198.51.100.7:9000"),
crate::transport::PacketBuffer::new(vec![0]),
123_456,
))
.expect("malformed packet queued");
let turn = {
let mut dataplane_io = super::rx_loop_dataplane_io(
&mut packet_rx,
&mut fast_ingress_rx,
&mut endpoint_rx,
&mut tun_outbound_rx,
&endpoint_io.event_tx,
);
node.drain_dataplane_turn_with_firsts(
&mut dataplane_io,
crate::dataplane::DataplaneLiveTurnFirsts::default(),
super::RxLoopDataplaneTurnLimits::new(4, 4, 4, 4),
)
.await
};
assert!(turn.has_activity());
assert!(turn.has_failures());
assert_eq!(turn.summary().raw_ingress_dropped(), 1);
assert_eq!(turn.raw_ingress_drops().len(), 1);
assert_eq!(
turn.raw_ingress_drops()[0].reason(),
crate::dataplane::DataplaneRawIngressDropReason::Wire(
crate::dataplane::WirePreflightError::TooShort
)
);
assert_eq!(
turn.raw_ingress_drops()[0].transport_id(),
crate::transport::TransportId::new(7)
);
assert!(turn.output_drops().is_empty());
assert!(turn.drops().is_empty());
assert!(turn.endpoint_data_drops().is_empty());
assert!(turn.tun_outbound_drops().is_empty());
assert!(packet_rx.try_recv().is_err());
assert!(tun_rx.try_recv_packet().is_err());
assert!(endpoint_io.event_rx.try_recv().is_err());
}
#[test]
fn rx_loop_maintenance_state_owns_activity_window_and_timeout_skip() {
let start = Instant::now();
let window = Duration::from_secs(2);
let empty = RxLoopDataDrainStats::default();
let drained = RxLoopDataDrainStats::new(1, 0, 0, 0);
let mut state = RxLoopMaintenanceState::default();
assert!(!state.data_pressure(empty, start, window));
assert!(!state.skip_slow_maintenance(false));
assert!(
!state.skip_slow_maintenance(true),
"queued dataplane work should timebox slow maintenance instead of starving it"
);
state.record_data_activity(start);
assert!(state.data_pressure(empty, start + Duration::from_secs(1), window));
assert!(!state.data_pressure(empty, start + Duration::from_secs(3), window));
assert!(state.data_pressure(drained, start + Duration::from_secs(3), window));
state.record_maintenance_result(true, true);
assert!(state.skip_slow_maintenance(true));
assert!(!state.skip_slow_maintenance(false));
state.record_maintenance_result(true, false);
assert!(
!state.skip_slow_maintenance(true),
"one skipped or successful busy tick should clear the timeout latch"
);
state.record_maintenance_result(false, true);
assert!(!state.skip_slow_maintenance(true));
}
#[test]
fn rx_loop_maintenance_plan_owns_pressure_skip_and_timeout_budget() {
let start = Instant::now();
let window = Duration::from_secs(2);
let idle_timeout = Duration::from_millis(100);
let busy_timeout = Duration::from_millis(10);
let empty = RxLoopDataDrainStats::default();
let drained = RxLoopDataDrainStats::new(1, 0, 0, 0);
let mut state = RxLoopMaintenanceState::default();
let idle = state.plan_maintenance(empty, start, window, idle_timeout, busy_timeout);
assert_eq!(
idle,
RxLoopMaintenancePlan::new(false, false, idle_timeout, busy_timeout)
);
assert_eq!(
RxLoopMaintenancePlan::new(false, true, idle_timeout, busy_timeout).slow_timeout(),
Some(idle_timeout)
);
assert!(!idle.data_pressure());
assert_eq!(idle.slow_timeout(), Some(idle_timeout));
state.record_data_activity(start);
let recent_busy = state.plan_maintenance(
empty,
start + Duration::from_secs(1),
window,
idle_timeout,
busy_timeout,
);
assert!(recent_busy.data_pressure());
assert_eq!(recent_busy.slow_timeout(), Some(busy_timeout));
state.record_maintenance_result(true, true);
let skipped_busy_after_timeout = state.plan_maintenance(
empty,
start + Duration::from_secs(1),
window,
idle_timeout,
busy_timeout,
);
assert!(skipped_busy_after_timeout.data_pressure());
assert_eq!(skipped_busy_after_timeout.slow_timeout(), None);
state.record_maintenance_result(true, false);
let retried_busy_after_skip = state.plan_maintenance(
empty,
start + Duration::from_secs(1),
window,
idle_timeout,
busy_timeout,
);
assert!(retried_busy_after_skip.data_pressure());
assert_eq!(
retried_busy_after_skip.slow_timeout(),
Some(busy_timeout),
"slow maintenance should retry under sustained data pressure after one skip"
);
let busy_with_queued_data = RxLoopMaintenanceState::default().plan_maintenance(
drained,
start + Duration::from_secs(1),
window,
idle_timeout,
busy_timeout,
);
assert!(busy_with_queued_data.data_pressure());
assert_eq!(busy_with_queued_data.slow_timeout(), Some(busy_timeout));
let expired_idle = state.plan_maintenance(
empty,
start + Duration::from_secs(3),
window,
idle_timeout,
busy_timeout,
);
assert!(!expired_idle.data_pressure());
assert_eq!(expired_idle.slow_timeout(), Some(idle_timeout));
}
#[tokio::test]
async fn single_lane_drain_leaves_other_lanes_for_later_turns() {
let (selected_tx, mut selected_rx) = tokio::sync::mpsc::channel(4);
let (other_tx, mut other_rx) = tokio::sync::mpsc::channel(4);
selected_tx.send("queued-selected").await.unwrap();
other_tx.send("queued-other").await.unwrap();
let mut drain = SingleLaneDrainCursor::new(Some("selected-first"), 4);
assert_eq!(drain.next(&mut selected_rx), Some("selected-first"));
assert_eq!(drain.next(&mut selected_rx), Some("queued-selected"));
assert_eq!(drain.next(&mut selected_rx), None);
assert_eq!(other_rx.try_recv().ok(), Some("queued-other"));
assert_eq!(drain.drained(), 2);
}
#[tokio::test]
async fn single_lane_drain_cursor_owns_first_item_and_budget() {
let (tun_tx, mut tun_rx) = tokio::sync::mpsc::channel(4);
tun_tx.send("queued-1").await.unwrap();
tun_tx.send("queued-2").await.unwrap();
tun_tx.send("queued-3").await.unwrap();
let mut drain = SingleLaneDrainCursor::new(Some("selected"), 3);
assert_eq!(drain.next(&mut tun_rx), Some("selected"));
assert_eq!(drain.next(&mut tun_rx), Some("queued-1"));
assert_eq!(drain.next(&mut tun_rx), Some("queued-2"));
assert_eq!(drain.next(&mut tun_rx), None);
assert_eq!(tun_rx.try_recv().ok(), Some("queued-3"));
assert_eq!(drain.drained(), 3);
}