From 56d39f223b05839fa706e00dedf4485c24948761 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 5 Mar 2026 16:40:15 +0000 Subject: [PATCH] Add ECN congestion signaling and transport congestion detection Implement hop-by-hop ECN congestion signaling through the FMP layer, transport-level congestion detection via kernel drop counters, and chaos harness integration for end-to-end validation. FMP/session ECN plumbing: - Thread ce_flag parsed at link layer through dispatch_link_message, handle_session_datagram, handle_session_payload, and handle_encrypted_session_msg to session delivery - Replace hardcoded false in session-layer record_recv() with actual ce_flag, activating ecn_ce_count tracking in session MMP ECN congestion detection and CE relay: - Add EcnConfig (node.ecn.*) with configurable loss_threshold (5%) and etx_threshold (3.0) for transit congestion detection - Add send_encrypted_link_message_with_ce() that ORs FLAG_CE into FMP header flags; original method delegates with ce_flag=false - Compute outgoing_ce = incoming_ce || local congestion on next-hop link, enabling hop-by-hop CE relay through transit nodes IPv6 ECN-CE marking: - Mark ECN-CE (0b11) in IPv6 Traffic Class on received DataPackets before TUN delivery when FMP CE flag is set - Only marks ECN-capable packets (ECT(0)/ECT(1)); Not-ECT packets unchanged per RFC 3168 Transport congestion abstraction and UDP kernel drop detection: - Add TransportCongestion struct to transport layer for transport- agnostic local congestion indicators - Replace tokio::UdpSocket with AsyncFd using libc::recvmsg() with ancillary data parsing - Enable SO_RXQ_OVFL for kernel receive buffer drop counter on every packet, wiring up previously-stubbed UdpStats.kernel_drops - Add TransportDropState for per-transport delta tracking with 1s tick sampling via sample_transport_congestion() - Extend detect_congestion() with transport kernel drop check alongside MMP loss metrics Congestion monitoring and control: - Add CongestionStats (ce_forwarded, ce_received, congestion_detected, kernel_drop_events) to NodeStats with snapshot serialization - Wire counters into forwarding path, session handler, and transport drop sampling with rate-limited warn logging (5s interval) - Expose congestion data in show_routing control query and ecn_ce_count in show_mmp peer entries - Add congestion counters to fipstop routing tab in two-column layout Chaos harness integration: - Add query_routing(), query_transports(), snapshot_all_congestion() to chaos control module - Add congestion/kernel-drop log analysis in logs module - Add congestion-stress scenario: 10-node tree, 1 Mbps bandwidth, 5-10% netem loss, heavy iperf3 traffic - Add IngressConfig for tc ingress policing with per-peer policer filters simulating upstream bandwidth bottlenecks - Add iperf3 JSON result capture to traffic manager for throughput measurement across scenarios - Add ECN A/B test scenarios (ecn-ab-on/off.yaml) with ingress policing and comparison script - Enable TCP ECN negotiation (tcp_ecn=1 sysctl) in container entrypoint for end-to-end CE propagation Tests: - 10 ECN unit/integration tests: mark_ipv6_ecn_ce variants, CE relay chain (3-node propagation), EcnConfig serde roundtrip - 3 transport drop congestion detection unit tests Documentation: - Update fips-mesh-layer.md: replace outdated CE Echo stub with full ECN Congestion Signaling section covering detection logic, CE relay, IPv6 marking, session tracking, and monitoring counters - Update fips-configuration.md: add node.ecn.* parameter table and ecn block in complete reference YAML - Update fips-transport-layer.md: add Congestion Reporting section with TransportCongestion struct, congestion() trait method, and per-transport status; document AsyncFd/recvmsg/SO_RXQ_OVFL in UDP - Update chaos README: add congestion/ECN scenario docs, ingress traffic control, and iperf3 JSON capture sections - Update README.md: add ECN to features list and "What works today"; update transport and tooling entries --- README.md | 8 +- docs/design/fips-configuration.md | 23 ++ docs/design/fips-mesh-layer.md | 43 ++- docs/design/fips-transport-layer.md | 38 ++- src/bin/fipstop/ui/routing.rs | 28 +- src/config/node.rs | 81 +++++ src/control/queries.rs | 18 +- src/mmp/metrics.rs | 5 + src/mmp/receiver.rs | 4 + src/node/handlers/dispatch.rs | 4 +- src/node/handlers/encrypted.rs | 2 +- src/node/handlers/forwarding.rs | 78 ++++- src/node/handlers/mod.rs | 2 +- src/node/handlers/rx_loop.rs | 1 + src/node/handlers/session.rs | 42 ++- src/node/mod.rs | 41 ++- src/node/stats.rs | 51 ++++ src/node/tests/forwarding.rs | 217 ++++++++++++- src/transport/mod.rs | 29 ++ src/transport/udp/mod.rs | 99 +++--- src/transport/udp/socket.rs | 288 ++++++++++++++++++ testing/chaos/README.md | 45 +++ testing/chaos/ecn-ab-test.sh | 127 ++++++++ testing/chaos/entrypoint.sh | 6 + .../chaos/scenarios/congestion-stress.yaml | 104 +++++++ testing/chaos/scenarios/ecn-ab-off.yaml | 80 +++++ testing/chaos/scenarios/ecn-ab-on.yaml | 100 ++++++ testing/chaos/sim/control.py | 39 +++ testing/chaos/sim/logs.py | 9 + testing/chaos/sim/netem.py | 59 +++- testing/chaos/sim/runner.py | 24 +- testing/chaos/sim/scenario.py | 33 ++ testing/chaos/sim/traffic.py | 62 +++- 33 files changed, 1667 insertions(+), 123 deletions(-) create mode 100644 src/transport/udp/socket.rs create mode 100755 testing/chaos/ecn-ab-test.sh create mode 100644 testing/chaos/scenarios/congestion-stress.yaml create mode 100644 testing/chaos/scenarios/ecn-ab-off.yaml create mode 100644 testing/chaos/scenarios/ecn-ab-on.yaml diff --git a/README.md b/README.md index 03e6e70..5644dbe 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,8 @@ sessions across the mesh. unmodified IP applications - **Metrics Measurement Protocol** — per-link RTT, loss, jitter, and goodput measurement +- **ECN congestion signaling** — hop-by-hop CE flag relay with RFC 3168 IPv6 + marking, transport kernel drop detection - **Operator visibility** — `fipsctl` control socket interface for runtime inspection of peers, links, sessions, tree state, and metrics - **Zero configuration** — sensible defaults; a node can start with no config @@ -187,13 +189,15 @@ UDP/IP overlays but has not been tested beyond small meshes. - Noise IK (link layer) and Noise XK (session layer) encryption - IPv6 TUN adapter with DNS resolution of `.fips` names - Per-link metrics (RTT, loss, jitter, goodput) -- Runtime inspection via `fipsctl` +- ECN congestion signaling (hop-by-hop CE relay, IPv6 CE marking, kernel drop detection) +- UDP, TCP, and Ethernet transports +- Runtime inspection via `fipsctl` and `fipstop` - Docker-based integration and chaos testing ### Near-term priorities - Peer discovery via Nostr relays (bootstrap without static peer lists) -- Additional transports (Ethernet, Tor) +- Additional transports (Bluetooth, Tor) - Improved routing resilience under churn - Security audit of cryptographic protocols - CI pipeline and published crate diff --git a/docs/design/fips-configuration.md b/docs/design/fips-configuration.md index d31eea1..9aa7b1c 100644 --- a/docs/design/fips-configuration.md +++ b/docs/design/fips-configuration.md @@ -163,6 +163,25 @@ Controls tree construction and parent selection. Bloom filter size (1 KB), hash count (5), and size classes are protocol constants and not configurable. +### ECN Signaling (`node.ecn.*`) + +Controls hop-by-hop ECN (Explicit Congestion Notification) signaling. When +enabled, transit nodes detect congestion on outgoing links (via MMP loss/ETX +metrics or kernel buffer drops) and set the CE flag on forwarded FMP frames. +Destination nodes mark ECN-capable IPv6 packets with CE before TUN delivery +per RFC 3168, enabling end-host TCP congestion control to react. + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `node.ecn.enabled` | bool | `true` | Enable ECN congestion signaling (CE flag relay and local congestion detection) | +| `node.ecn.loss_threshold` | f64 | `0.05` | MMP loss rate threshold for CE marking (0.0–1.0). When the outgoing link's loss rate meets or exceeds this value, forwarded packets are CE-marked. | +| `node.ecn.etx_threshold` | f64 | `3.0` | MMP ETX threshold for CE marking (≥1.0). When the outgoing link's ETX meets or exceeds this value, forwarded packets are CE-marked. | + +Congestion detection triggers on any of: outgoing link loss ≥ `loss_threshold`, +outgoing link ETX ≥ `etx_threshold`, or kernel receive buffer drops detected on +any local transport. CE is relayed hop-by-hop: once set on any hop, the flag +stays set for all subsequent hops to the destination. + ### Session / Data Plane (`node.session.*`) Controls end-to-end session behavior and packet queuing. @@ -478,6 +497,10 @@ node: mode: full # full | lightweight | minimal log_interval_secs: 30 owd_window_size: 32 + ecn: + enabled: true # ECN congestion signaling (CE flag relay) + loss_threshold: 0.05 # MMP loss rate threshold for CE marking (5%) + etx_threshold: 3.0 # MMP ETX threshold for CE marking control: enabled: true socket_path: null # null = auto ($XDG_RUNTIME_DIR → /run/fips → /tmp fallback) diff --git a/docs/design/fips-mesh-layer.md b/docs/design/fips-mesh-layer.md index c68ad03..b7b13ff 100644 --- a/docs/design/fips-mesh-layer.md +++ b/docs/design/fips-mesh-layer.md @@ -465,12 +465,44 @@ MMP reports on different timers), inter-frame processing delays inflate spin bit RTT measurements unpredictably. Timestamp-echo from ReceiverReports (with dwell-time compensation) is the sole SRTT source. -### CE Echo +### ECN Congestion Signaling -The CE (Congestion Experienced) echo flag (bit 1 in the FMP flags byte) is -reserved for ECN signaling. The transport trait does not currently expose -ECN marking, so the CE echo flag is never set. One-way delay trend serves -as the sole pre-loss congestion indicator. +The CE (Congestion Experienced) flag (bit 1 in the FMP flags byte) provides +hop-by-hop congestion signaling through the mesh. Transit nodes detect +congestion on outgoing links and set CE on forwarded packets; once set, the +flag stays set for all subsequent hops to the destination. + +**Congestion detection** (`detect_congestion()`) triggers on any of: + +- Outgoing link MMP loss rate ≥ `node.ecn.loss_threshold` (default 5%) +- Outgoing link MMP ETX ≥ `node.ecn.etx_threshold` (default 3.0) +- Kernel receive buffer drops detected on any local transport (via + `SO_RXQ_OVFL` on UDP) + +**CE relay**: The forwarding path computes `outgoing_ce = incoming_ce || +local_congestion`. The `send_encrypted_link_message_with_ce()` method ORs +`FLAG_CE` into the FMP header flags when ce is true. The original +`send_encrypted_link_message()` delegates with `ce_flag=false`, leaving the +20+ existing call sites unchanged. + +**IPv6 ECN-CE marking**: When a CE-flagged DataPacket arrives at its final +destination, the IPv6 Traffic Class ECN bits are marked CE (0b11) before +TUN delivery — but only for ECN-capable packets (ECT(0) or ECT(1)). Not-ECT +packets are never marked per RFC 3168. The host TCP stack then echoes ECE in +ACKs, triggering sender cwnd reduction through standard congestion control. + +**Session-layer tracking**: The `ecn_ce_count` field in MMP ReceiverReports +tracks CE-flagged packets received per link, providing end-to-end visibility +into congestion propagation. + +**Monitoring**: `CongestionStats` tracks four counters — `ce_forwarded`, +`ce_received`, `congestion_detected`, and `kernel_drop_events` — exposed via +`fipsctl show routing` (congestion block) and `fipstop` (routing tab). +Rate-limited warn logging (5s interval) alerts on congestion detection events. + +See `node.ecn.*` in +[fips-configuration.md](fips-configuration.md#ecn-signaling-nodeecn) for +tuning parameters. ### Operator Logging @@ -540,6 +572,7 @@ an attacker sends invalid packets to elicit responses. | Inner header timestamps | **Implemented** | | Path MTU tracking (SessionDatagram) | **Implemented** | | Metrics Measurement Protocol (MMP) | **Implemented** | +| ECN congestion signaling (CE relay, IPv6 marking) | **Implemented** | | Rekey with index rotation | Planned | | Allowlist/blocklist | Planned | diff --git a/docs/design/fips-transport-layer.md b/docs/design/fips-transport-layer.md index 8faafdd..a7a6feb 100644 --- a/docs/design/fips-transport-layer.md +++ b/docs/design/fips-transport-layer.md @@ -215,8 +215,15 @@ buffer fills in ~2.5 ms; any stall in the async receive loop (decryption, routing, forwarding overhead) causes the kernel to silently drop incoming datagrams. -FIPS uses the `socket2` crate to configure socket buffers at bind time, -before the receive loop starts: +FIPS uses `socket2::Socket` wrapped in `tokio::io::unix::AsyncFd` for the +UDP receive path. This replaces `tokio::UdpSocket` and enables direct +`libc::recvmsg()` calls with ancillary data parsing — specifically the +`SO_RXQ_OVFL` socket option, which delivers a cumulative kernel receive +buffer drop counter on every received packet. The drop counter feeds into +the ECN congestion detection system (see +[fips-mesh-layer.md](fips-mesh-layer.md#ecn-congestion-signaling)). + +Socket buffers are configured at bind time via `socket2`: | Parameter | Default | Description | | ---------------- | ------- | ------------------------------------ | @@ -486,6 +493,7 @@ start() → lifecycle Bring transport up (bind socket, o stop() → lifecycle Bring transport down send(addr, data) → delivery Send datagram to transport address close_connection(addr)→ () Close a specific connection (no-op for connectionless) +congestion() → TransportCongestion Local congestion indicators (optional) discover() → Vec Report discovered FIPS endpoints (optional) auto_connect() → bool Auto-connect discovered peers (default: false) accept_connections() → bool Accept inbound handshakes (default: true) @@ -520,6 +528,30 @@ TransportType { Predefined types exist for UDP, TCP, Ethernet, WiFi, Tor, and Serial. +### Congestion Reporting + +Transports optionally report local congestion indicators via a +`TransportCongestion` struct, providing a transport-agnostic interface for +the node layer's ECN congestion detection: + +```text +TransportCongestion { + recv_drops: Option Cumulative kernel-dropped packets (monotonic) +} +``` + +The node samples each transport's congestion state on a 1-second tick via +`sample_transport_congestion()`. `TransportDropState` tracks per-transport +drop deltas: when new drops appear (rising edge), the `dropping` flag is +set, and `detect_congestion()` in the forwarding path triggers CE marking +on all forwarded datagrams. + +| Transport | Congestion Source | Mechanism | +| --------- | ----------------- | --------- | +| UDP | `SO_RXQ_OVFL` kernel drop counter | `recvmsg()` ancillary data on every packet | +| TCP | Not yet implemented | Returns `None` (TCP handles congestion internally) | +| Ethernet | Not yet implemented | Returns `None` | + ### Transport Addresses Transport addresses (`TransportAddr`) are opaque byte vectors. The transport @@ -542,7 +574,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to | Transport | Status | Notes | | --------- | ------ | ----- | -| UDP/IP | **Implemented** | Primary transport, async send/receive, configurable MTU | +| UDP/IP | **Implemented** | Primary transport, AsyncFd/recvmsg, SO_RXQ_OVFL kernel drop detection | | TCP/IP | **Implemented** | FMP header-based framing, connect-on-send, per-connection MSS MTU | | Ethernet | **Implemented** | AF_PACKET SOCK_DGRAM, EtherType 0x2121, beacon discovery, Linux only | | WiFi | Future direction | Infrastructure mode = Ethernet driver | diff --git a/src/bin/fipstop/ui/routing.rs b/src/bin/fipstop/ui/routing.rs index c09e95f..3ff9dd7 100644 --- a/src/bin/fipstop/ui/routing.rs +++ b/src/bin/fipstop/ui/routing.rs @@ -78,7 +78,14 @@ fn draw_routing_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) { let inner = block.inner(area); frame.render_widget(block, area); - let mut lines = vec![ + let cols = Layout::horizontal([ + Constraint::Percentage(50), + Constraint::Percentage(50), + ]) + .split(inner); + + // Left column: Forwarding + Discovery + let mut left = vec![ helpers::section_header("Forwarding"), fwd_line(data, "Received", "received_packets", "received_bytes"), fwd_line(data, "Delivered", "delivered_packets", "delivered_bytes"), @@ -109,17 +116,28 @@ fn draw_routing_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) { helpers::kv_line("Identity Miss", &helpers::nested_u64(data, "discovery", "resp_identity_miss")), helpers::kv_line("Proof Failed", &helpers::nested_u64(data, "discovery", "resp_proof_failed")), helpers::kv_line("Decode Error", &helpers::nested_u64(data, "discovery", "resp_decode_error")), - Line::from(""), + ]; + + // Right column: Error Signals + Congestion + let mut right = vec![ helpers::section_header("Error Signals"), helpers::kv_line("Coords Required", &helpers::nested_u64(data, "error_signals", "coords_required")), helpers::kv_line("Path Broken", &helpers::nested_u64(data, "error_signals", "path_broken")), helpers::kv_line("MTU Exceeded", &helpers::nested_u64(data, "error_signals", "mtu_exceeded")), + Line::from(""), + helpers::section_header("Congestion"), + helpers::kv_line("CE Forwarded", &helpers::nested_u64(data, "congestion", "ce_forwarded")), + helpers::kv_line("CE Received", &helpers::nested_u64(data, "congestion", "ce_received")), + helpers::kv_line("Congestion Detected", &helpers::nested_u64(data, "congestion", "congestion_detected")), + helpers::kv_line("Kernel Drops", &helpers::nested_u64(data, "congestion", "kernel_drop_events")), ]; - let max_lines = inner.height as usize; - lines.truncate(max_lines); + let max_lines = cols[0].height as usize; + left.truncate(max_lines); + right.truncate(max_lines); - frame.render_widget(Paragraph::new(lines), inner); + frame.render_widget(Paragraph::new(left), cols[0]); + frame.render_widget(Paragraph::new(right), cols[1]); } fn draw_coord_cache(frame: &mut Frame, app: &App, area: Rect) { diff --git a/src/config/node.rs b/src/config/node.rs index f108d0f..b101298 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -414,6 +414,50 @@ impl BuffersConfig { fn default_dns_channel() -> usize { 64 } } +// ============================================================================ +// ECN Congestion Signaling +// ============================================================================ + +/// ECN congestion signaling configuration (`node.ecn.*`). +/// +/// Controls the FMP CE relay chain: transit nodes detect congestion on outgoing +/// links and set the CE flag in forwarded datagrams. The destination marks +/// IPv6 ECN-CE on ECN-capable packets before TUN delivery. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EcnConfig { + /// Enable ECN congestion signaling (`node.ecn.enabled`). + #[serde(default = "EcnConfig::default_enabled")] + pub enabled: bool, + + /// Loss rate threshold for marking CE (`node.ecn.loss_threshold`). + /// When the outgoing link's loss rate meets or exceeds this value, + /// the transit node sets CE on forwarded datagrams. + #[serde(default = "EcnConfig::default_loss_threshold")] + pub loss_threshold: f64, + + /// ETX threshold for marking CE (`node.ecn.etx_threshold`). + /// When the outgoing link's ETX meets or exceeds this value, + /// the transit node sets CE on forwarded datagrams. + #[serde(default = "EcnConfig::default_etx_threshold")] + pub etx_threshold: f64, +} + +impl Default for EcnConfig { + fn default() -> Self { + Self { + enabled: true, + loss_threshold: 0.05, + etx_threshold: 3.0, + } + } +} + +impl EcnConfig { + fn default_enabled() -> bool { true } + fn default_loss_threshold() -> f64 { 0.05 } + fn default_etx_threshold() -> f64 { 3.0 } +} + // ============================================================================ // Node Configuration (Root) // ============================================================================ @@ -493,6 +537,10 @@ pub struct NodeConfig { /// Metrics Measurement Protocol — session layer (`node.session_mmp.*`). #[serde(default)] pub session_mmp: SessionMmpConfig, + + /// ECN congestion signaling (`node.ecn.*`). + #[serde(default)] + pub ecn: EcnConfig, } impl Default for NodeConfig { @@ -516,6 +564,7 @@ impl Default for NodeConfig { control: ControlConfig::default(), mmp: MmpConfig::default(), session_mmp: SessionMmpConfig::default(), + ecn: EcnConfig::default(), } } } @@ -526,3 +575,35 @@ impl NodeConfig { fn default_heartbeat_interval_secs() -> u64 { 10 } fn default_link_dead_timeout_secs() -> u64 { 30 } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_ecn_config_defaults() { + let c = EcnConfig::default(); + assert!(c.enabled); + assert!((c.loss_threshold - 0.05).abs() < 1e-9); + assert!((c.etx_threshold - 3.0).abs() < 1e-9); + } + + #[test] + fn test_ecn_config_yaml_roundtrip() { + let yaml = "loss_threshold: 0.10\netx_threshold: 2.5\nenabled: false\n"; + let c: EcnConfig = serde_yaml::from_str(yaml).unwrap(); + assert!(!c.enabled); + assert!((c.loss_threshold - 0.10).abs() < 1e-9); + assert!((c.etx_threshold - 2.5).abs() < 1e-9); + } + + #[test] + fn test_ecn_config_partial_yaml() { + // Only specify loss_threshold — others should get defaults + let yaml = "loss_threshold: 0.02\n"; + let c: EcnConfig = serde_yaml::from_str(yaml).unwrap(); + assert!(c.enabled); // default + assert!((c.loss_threshold - 0.02).abs() < 1e-9); + assert!((c.etx_threshold - 3.0).abs() < 1e-9); // default + } +} diff --git a/src/control/queries.rs b/src/control/queries.rs index aefc177..89451d2 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -119,10 +119,10 @@ pub fn show_peers(node: &Node) -> Value { if let Some(smoothed_etx) = mmp.metrics.smoothed_etx() { mmp_json["smoothed_etx"] = json!(smoothed_etx); } - if let Some(srtt) = mmp.metrics.srtt_ms() { - if let Some(setx) = mmp.metrics.smoothed_etx() { - mmp_json["lqi"] = json!(setx * (1.0 + srtt / 100.0)); - } + if let Some(srtt) = mmp.metrics.srtt_ms() + && let Some(setx) = mmp.metrics.smoothed_etx() + { + mmp_json["lqi"] = json!(setx * (1.0 + srtt / 100.0)); } peer_json["mmp"] = mmp_json; } @@ -265,10 +265,10 @@ pub fn show_sessions(node: &Node) -> Value { if let Some(smoothed_etx) = mmp.metrics.smoothed_etx() { mmp_json["smoothed_etx"] = json!(smoothed_etx); } - if let Some(srtt) = mmp.metrics.srtt_ms() { - if let Some(setx) = mmp.metrics.smoothed_etx() { - mmp_json["sqi"] = json!(setx * (1.0 + srtt / 100.0)); - } + if let Some(srtt) = mmp.metrics.srtt_ms() + && let Some(setx) = mmp.metrics.smoothed_etx() + { + mmp_json["sqi"] = json!(setx * (1.0 + srtt / 100.0)); } session_json["mmp"] = mmp_json; } @@ -362,6 +362,7 @@ pub fn show_mmp(node: &Node) -> Value { link_layer["delivery_ratio_forward"] = json!(metrics.delivery_ratio_forward); link_layer["delivery_ratio_reverse"] = json!(metrics.delivery_ratio_reverse); + link_layer["ecn_ce_count"] = json!(metrics.last_ecn_ce_count()); Some(json!({ "peer": hex::encode(addr.as_bytes()), @@ -487,6 +488,7 @@ pub fn show_routing(node: &Node) -> Value { "forwarding": serde_json::to_value(&node_stats.forwarding).unwrap_or_default(), "discovery": serde_json::to_value(&node_stats.discovery).unwrap_or_default(), "error_signals": serde_json::to_value(&node_stats.errors).unwrap_or_default(), + "congestion": serde_json::to_value(&node_stats.congestion).unwrap_or_default(), }) } diff --git a/src/mmp/metrics.rs b/src/mmp/metrics.rs index 7fa837f..b61574a 100644 --- a/src/mmp/metrics.rs +++ b/src/mmp/metrics.rs @@ -188,6 +188,11 @@ impl MmpMetrics { pub fn goodput_bps(&self) -> f64 { self.goodput_bps } + + /// Cumulative ECN CE count from the most recent ReceiverReport. + pub fn last_ecn_ce_count(&self) -> u32 { + self.prev_rr_ecn_ce + } } impl Default for MmpMetrics { diff --git a/src/mmp/receiver.rs b/src/mmp/receiver.rs index 230d83d..5794d7d 100644 --- a/src/mmp/receiver.rs +++ b/src/mmp/receiver.rs @@ -360,6 +360,10 @@ impl ReceiverState { pub fn last_recv_time(&self) -> Option { self.last_recv_time } + + pub fn ecn_ce_count(&self) -> u32 { + self.ecn_ce_count + } } impl Default for ReceiverState { diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 59d1cdf..9209710 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -8,7 +8,7 @@ impl Node { /// Dispatch a decrypted link message to the appropriate handler. /// /// Link messages are protocol messages exchanged between authenticated peers. - pub(in crate::node) async fn dispatch_link_message(&mut self, from: &NodeAddr, plaintext: &[u8]) { + pub(in crate::node) async fn dispatch_link_message(&mut self, from: &NodeAddr, plaintext: &[u8], ce_flag: bool) { if plaintext.is_empty() { return; } @@ -19,7 +19,7 @@ impl Node { match msg_type { 0x00 => { // SessionDatagram - self.handle_session_datagram(from, payload).await; + self.handle_session_datagram(from, payload, ce_flag).await; } 0x01 => { // SenderReport diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 502c5a3..b0c537c 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -146,6 +146,6 @@ impl Node { peer.touch(packet.timestamp_ms); // Dispatch to link message handler (msg_type + payload, inner header stripped) - self.dispatch_link_message(&node_addr, link_message).await; + self.dispatch_link_message(&node_addr, link_message, ce_flag).await; } } diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index fc78307..e0bb6b8 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -14,14 +14,15 @@ use crate::protocol::{ CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup, }; use crate::NodeAddr; -use tracing::debug; +use std::time::{Duration, Instant}; +use tracing::{debug, warn}; impl Node { /// Handle an incoming SessionDatagram from a peer. /// /// Called by `dispatch_link_message` for msg_type 0x00. The payload /// has already had its msg_type byte stripped by dispatch. - pub(in crate::node) async fn handle_session_datagram(&mut self, _from: &NodeAddr, payload: &[u8]) { + pub(in crate::node) async fn handle_session_datagram(&mut self, _from: &NodeAddr, payload: &[u8], incoming_ce: bool) { self.stats_mut().forwarding.record_received(payload.len()); let mut datagram = match SessionDatagram::decode(payload) { @@ -50,7 +51,7 @@ impl Node { // Local delivery: dispatch to session layer handlers if datagram.dest_addr == *self.node_addr() { self.stats_mut().forwarding.record_delivered(payload.len()); - self.handle_session_payload(&datagram.src_addr, &datagram.payload, datagram.path_mtu) + self.handle_session_payload(&datagram.src_addr, &datagram.payload, datagram.path_mtu, incoming_ce) .await; return; } @@ -77,10 +78,25 @@ impl Node { } } + // ECN CE relay: propagate incoming CE and detect local congestion + let local_congestion = self.detect_congestion(&next_hop_addr); + let outgoing_ce = incoming_ce || local_congestion; + if local_congestion { + self.stats_mut().congestion.record_congestion_detected(); + let now = Instant::now(); + let should_log = self.last_congestion_log + .map(|t| now.duration_since(t) >= Duration::from_secs(5)) + .unwrap_or(true); + if should_log { + self.last_congestion_log = Some(now); + warn!(next_hop = %next_hop_addr, "Congestion detected, CE flag set on forwarded packet"); + } + } + // Forward: re-encode (includes 0x00 type byte) and send let encoded = datagram.encode(); if let Err(e) = self - .send_encrypted_link_message(&next_hop_addr, &encoded) + .send_encrypted_link_message_with_ce(&next_hop_addr, &encoded, outgoing_ce) .await { match e { @@ -100,6 +116,9 @@ impl Node { } } else { self.stats_mut().forwarding.record_forwarded(encoded.len()); + if outgoing_ce { + self.stats_mut().congestion.record_ce_forwarded(); + } } } @@ -336,4 +355,55 @@ impl Node { ); } } + + /// Detect congestion for CE marking on forwarded datagrams. + /// + /// Checks two signal sources: + /// 1. Outgoing link MMP metrics (loss rate, ETX) against configured thresholds + /// 2. Local transport congestion (kernel drops on any transport) + /// + /// Returns `true` if any signal indicates congestion. + pub(in crate::node) fn detect_congestion(&self, next_hop: &NodeAddr) -> bool { + if !self.config.node.ecn.enabled { + return false; + } + // Outgoing link MMP metrics + if let Some(peer) = self.peers.get(next_hop) + && let Some(mmp) = peer.mmp() + { + let metrics = &mmp.metrics; + if metrics.loss_rate() >= self.config.node.ecn.loss_threshold + || metrics.etx >= self.config.node.ecn.etx_threshold + { + return true; + } + } + // Local transport congestion (kernel drops) + self.transport_drops.values().any(|s| s.dropping) + } + + /// Sample transport congestion indicators. + /// + /// Called from the tick handler (1s interval). For each transport, + /// queries the cumulative kernel drop counter and sets the `dropping` + /// flag if new drops occurred since the previous sample. + pub(in crate::node) fn sample_transport_congestion(&mut self) { + let mut new_drop_events = Vec::new(); + for (&tid, transport) in &self.transports { + let congestion = transport.congestion(); + let state = self.transport_drops.entry(tid).or_default(); + if let Some(current) = congestion.recv_drops { + let new_drops = current > state.prev_drops; + if new_drops && !state.dropping { + new_drop_events.push(tid); + } + state.dropping = new_drops; + state.prev_drops = current; + } + } + for tid in new_drop_events { + self.stats_mut().congestion.record_kernel_drop_event(); + warn!(transport_id = tid.as_u32(), "Kernel recv drops first observed on transport"); + } + } } diff --git a/src/node/handlers/mod.rs b/src/node/handlers/mod.rs index 4108845..ca16234 100644 --- a/src/node/handlers/mod.rs +++ b/src/node/handlers/mod.rs @@ -7,5 +7,5 @@ mod forwarding; mod handshake; mod mmp; mod rx_loop; -mod session; +pub(in crate::node) mod session; mod timeout; diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 225bf15..928f40a 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -118,6 +118,7 @@ impl Node { self.check_link_heartbeats().await; self.purge_stale_lookups(now_ms); self.poll_transport_discovery().await; + self.sample_transport_congestion(); } } } diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index efaaf9c..44bdd0b 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -43,6 +43,7 @@ impl Node { src_addr: &NodeAddr, payload: &[u8], path_mtu: u16, + ce_flag: bool, ) { let prefix = match FspCommonPrefix::parse(payload) { Some(p) => p, @@ -88,7 +89,7 @@ impl Node { } } FSP_PHASE_ESTABLISHED => { - self.handle_encrypted_session_msg(src_addr, payload, path_mtu).await; + self.handle_encrypted_session_msg(src_addr, payload, path_mtu, ce_flag).await; } _ => { debug!(phase = prefix.phase, "Unknown FSP phase"); @@ -105,7 +106,7 @@ impl Node { /// 4. AEAD decrypt with AAD = header_bytes /// 5. Strip FSP inner header → timestamp, msg_type, inner_flags /// 6. Dispatch by msg_type - async fn handle_encrypted_session_msg(&mut self, src_addr: &NodeAddr, payload: &[u8], path_mtu: u16) { + async fn handle_encrypted_session_msg(&mut self, src_addr: &NodeAddr, payload: &[u8], path_mtu: u16, ce_flag: bool) { // Parse the 12-byte encrypted header (includes the 4-byte prefix) let header = match FspEncryptedHeader::parse(payload) { Some(h) => h, @@ -209,7 +210,7 @@ impl Node { { let now = std::time::Instant::now(); mmp.receiver.record_recv( - header.counter, timestamp, plaintext.len(), false, now, + header.counter, timestamp, plaintext.len(), ce_flag, now, ); // Spin bit: advance state machine for correct TX reflection. // RTT samples not fed into SRTT — timestamp-echo provides @@ -233,8 +234,13 @@ impl Node { match SessionMessageType::from_byte(msg_type) { Some(SessionMessageType::DataPacket) => { // msg_type 0x10: deliver rest (IPv6 payload) to TUN + let mut packet = rest.to_vec(); + if ce_flag { + mark_ipv6_ecn_ce(&mut packet); + self.stats_mut().congestion.record_ce_received(); + } if let Some(tun_tx) = &self.tun_tx { - if let Err(e) = tun_tx.send(rest.to_vec()) { + if let Err(e) = tun_tx.send(packet) { debug!(error = %e, "Failed to deliver decrypted packet to TUN"); } } else { @@ -1472,3 +1478,31 @@ impl Node { } } } + +/// Mark ECN-CE in an IPv6 packet's Traffic Class field. +/// +/// IPv6 Traffic Class occupies bits across bytes 0 and 1: +/// byte[0] bits[3:0] = TC[7:4] +/// byte[1] bits[7:4] = TC[3:0] +/// ECN is TC[1:0]. Only marks CE (0b11) if the packet is ECN-capable +/// (ECT(0) or ECT(1)). Packets with ECN=0b00 (Not-ECT) are never marked +/// per RFC 3168. +/// +/// No checksum update needed: IPv6 has no header checksum, and the Traffic +/// Class field is not part of the TCP/UDP pseudo-header. +pub(in crate::node) fn mark_ipv6_ecn_ce(packet: &mut [u8]) { + if packet.len() < 2 { + return; + } + // Extract 8-bit Traffic Class from IPv6 header bytes 0-1 + let tc = ((packet[0] & 0x0F) << 4) | (packet[1] >> 4); + let ecn = tc & 0x03; + // Only mark CE on ECN-capable packets (ECT(0)=0b10 or ECT(1)=0b01) + if ecn == 0 { + return; + } + // Set both ECN bits to 1 (CE = 0b11) + let new_tc = tc | 0x03; + packet[0] = (packet[0] & 0xF0) | (new_tc >> 4); + packet[1] = (new_tc << 4) | (packet[1] & 0x0F); +} diff --git a/src/node/mod.rs b/src/node/mod.rs index 4947c5d..43b0d02 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -35,7 +35,7 @@ use crate::transport::ethernet::EthernetTransport; use crate::tree::TreeState; use crate::upper::icmp_rate_limit::IcmpRateLimiter; use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; -use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_SP}; +use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_SP}; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity}; use rand::Rng; use std::collections::{HashMap, VecDeque}; @@ -194,6 +194,18 @@ impl RecentRequest { /// Key for addr_to_link reverse lookup. type AddrKey = (TransportId, TransportAddr); +/// Per-transport kernel drop tracking for congestion detection. +/// +/// Sampled every tick (1s). The `dropping` flag indicates whether new +/// kernel drops were observed since the previous sample. +#[derive(Debug, Default)] +struct TransportDropState { + /// Previous `recv_drops` sample (cumulative counter). + prev_drops: u64, + /// True if drops increased since the last sample. + dropping: bool, +} + /// A running FIPS node instance. /// /// This is the top-level container holding all node state. @@ -248,6 +260,8 @@ pub struct Node { // === Transports & Links === /// Active transports (owned by Node). transports: HashMap, + /// Per-transport kernel drop tracking for congestion detection. + transport_drops: HashMap, /// Active links. links: HashMap, /// Reverse lookup: (transport_id, remote_addr) -> link_id. @@ -358,6 +372,10 @@ pub struct Node { /// Timestamp of last periodic parent re-evaluation (for pacing). last_parent_reeval: Option, + // === Congestion Logging === + /// Timestamp of last congestion detection log (rate-limited to 5s). + last_congestion_log: Option, + // === Display Names === /// Human-readable names for configured peers (alias or short npub). /// Populated at startup from peer config. @@ -427,6 +445,7 @@ impl Node { coord_cache, recent_requests: HashMap::new(), transports: HashMap::new(), + transport_drops: HashMap::new(), links: HashMap::new(), addr_to_link: HashMap::new(), packet_tx: None, @@ -462,6 +481,7 @@ impl Node { ), retry_pending: HashMap::new(), last_parent_reeval: None, + last_congestion_log: None, peer_aliases: HashMap::new(), }) } @@ -522,6 +542,7 @@ impl Node { coord_cache, recent_requests: HashMap::new(), transports: HashMap::new(), + transport_drops: HashMap::new(), links: HashMap::new(), addr_to_link: HashMap::new(), packet_tx: None, @@ -557,6 +578,7 @@ impl Node { ), retry_pending: HashMap::new(), last_parent_reeval: None, + last_congestion_log: None, peer_aliases: HashMap::new(), } } @@ -1288,6 +1310,18 @@ impl Node { &mut self, node_addr: &NodeAddr, plaintext: &[u8], + ) -> Result<(), NodeError> { + self.send_encrypted_link_message_with_ce(node_addr, plaintext, false).await + } + + /// Like `send_encrypted_link_message` but allows setting the FMP CE flag. + /// + /// Used by the forwarding path to relay congestion signals hop-by-hop. + pub(super) async fn send_encrypted_link_message_with_ce( + &mut self, + node_addr: &NodeAddr, + plaintext: &[u8], + ce_flag: bool, ) -> Result<(), NodeError> { let peer = self.peers.get_mut(node_addr) .ok_or(NodeError::PeerNotFound(*node_addr))?; @@ -1312,7 +1346,10 @@ impl Node { let sp_flag = peer.mmp() .map(|mmp| mmp.spin_bit.tx_bit()) .unwrap_or(false); - let flags = if sp_flag { FLAG_SP } else { 0 }; + let mut flags = if sp_flag { FLAG_SP } else { 0 }; + if ce_flag { + flags |= FLAG_CE; + } let session = peer.noise_session_mut().ok_or_else(|| NodeError::SendFailed { node_addr: *node_addr, diff --git a/src/node/stats.rs b/src/node/stats.rs index 8f1f657..b24aa5c 100644 --- a/src/node/stats.rs +++ b/src/node/stats.rs @@ -245,6 +245,46 @@ impl ErrorSignalStats { } } +/// Congestion event statistics — ECN CE tracking and detection triggers. +#[derive(Default)] +pub struct CongestionStats { + /// Packets forwarded with the CE flag set (incoming CE or locally detected). + pub ce_forwarded: u64, + /// CE-flagged packets received at this node as final destination. + pub ce_received: u64, + /// Number of times detect_congestion() returned true. + pub congestion_detected: u64, + /// Rising-edge transport kernel drop events (not-dropping → dropping). + pub kernel_drop_events: u64, +} + +impl CongestionStats { + pub fn record_ce_forwarded(&mut self) { + self.ce_forwarded += 1; + } + + pub fn record_ce_received(&mut self) { + self.ce_received += 1; + } + + pub fn record_congestion_detected(&mut self) { + self.congestion_detected += 1; + } + + pub fn record_kernel_drop_event(&mut self) { + self.kernel_drop_events += 1; + } + + pub fn snapshot(&self) -> CongestionStatsSnapshot { + CongestionStatsSnapshot { + ce_forwarded: self.ce_forwarded, + ce_received: self.ce_received, + congestion_detected: self.congestion_detected, + kernel_drop_events: self.kernel_drop_events, + } + } +} + /// Aggregate node statistics. #[derive(Default)] pub struct NodeStats { @@ -253,6 +293,7 @@ pub struct NodeStats { pub tree: TreeStats, pub bloom: BloomStats, pub errors: ErrorSignalStats, + pub congestion: CongestionStats, } impl NodeStats { @@ -267,6 +308,7 @@ impl NodeStats { tree: self.tree.snapshot(), bloom: self.bloom.snapshot(), errors: self.errors.snapshot(), + congestion: self.congestion.snapshot(), } } } @@ -356,6 +398,14 @@ pub struct ErrorSignalStatsSnapshot { pub mtu_exceeded: u64, } +#[derive(Clone, Debug, Default, Serialize)] +pub struct CongestionStatsSnapshot { + pub ce_forwarded: u64, + pub ce_received: u64, + pub congestion_detected: u64, + pub kernel_drop_events: u64, +} + #[derive(Clone, Debug, Default, Serialize)] pub struct NodeStatsSnapshot { pub forwarding: ForwardingStatsSnapshot, @@ -363,4 +413,5 @@ pub struct NodeStatsSnapshot { pub tree: TreeStatsSnapshot, pub bloom: BloomStatsSnapshot, pub errors: ErrorSignalStatsSnapshot, + pub congestion: CongestionStatsSnapshot, } diff --git a/src/node/tests/forwarding.rs b/src/node/tests/forwarding.rs index 1070e17..beffa4d 100644 --- a/src/node/tests/forwarding.rs +++ b/src/node/tests/forwarding.rs @@ -24,7 +24,7 @@ async fn test_forwarding_decode_error() { let mut node = make_node(); let from = make_node_addr(0xAA); // Too-short payload: should log error and return without panic - node.handle_session_datagram(&from, &[0x00; 5]).await; + node.handle_session_datagram(&from, &[0x00; 5], false).await; } // --- TTL --- @@ -39,7 +39,7 @@ async fn test_forwarding_hop_limit_exhausted() { .with_ttl(0); let encoded = dg.encode(); // Dispatch with payload after msg_type byte - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; // No panic, no send (node has no peers) } @@ -56,7 +56,7 @@ async fn test_forwarding_hop_limit_one_drops_at_transit() { .with_ttl(1); let encoded = dg.encode(); // Should succeed — ttl=1 decrements to 0 but packet is still processed - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; } // --- Local delivery --- @@ -69,7 +69,7 @@ async fn test_forwarding_local_delivery() { let dg = SessionDatagram::new(from, my_addr, vec![0x10, 0x00, 0x00, 0x00]); let encoded = dg.encode(); // Should detect local delivery and return without forwarding - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; } // --- Direct peer forwarding --- @@ -92,7 +92,7 @@ async fn test_forwarding_direct_peer() { // Handle on node 0: should forward to node 1 (direct peer) nodes[0] .node - .handle_session_datagram(&node0_addr, &encoded[1..]) + .handle_session_datagram(&node0_addr, &encoded[1..], false) .await; // Process packets — node 1 should receive the forwarded datagram @@ -135,7 +135,7 @@ async fn test_coord_cache_warming_session_setup() { // Handle the datagram (will be local delivery or no-route, but cache warming // happens before routing decision) - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; // After: both src and dest coords should be cached let cached_src = node.coord_cache().get(&src_addr, now_ms); @@ -175,7 +175,7 @@ async fn test_coord_cache_warming_session_ack() { assert!(node.coord_cache().get(&src_addr, now_ms).is_none()); assert!(node.coord_cache().get(&dest_addr, now_ms).is_none()); - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; // SessionAck caches both src_coords and dest_coords let cached_src = node.coord_cache().get(&src_addr, now_ms); @@ -217,7 +217,7 @@ async fn test_coord_cache_warming_encrypted_msg_with_coords() { assert!(node.coord_cache().get(&src_addr, now_ms).is_none()); assert!(node.coord_cache().get(&dest_addr, now_ms).is_none()); - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; assert!( node.coord_cache().get(&src_addr, now_ms).is_some(), @@ -250,7 +250,7 @@ async fn test_coord_cache_warming_encrypted_msg_no_coords() { .unwrap() .as_millis() as u64; - node.handle_session_datagram(&from, &encoded[1..]).await; + node.handle_session_datagram(&from, &encoded[1..], false).await; assert!( node.coord_cache().get(&src_addr, now_ms).is_none(), @@ -567,3 +567,202 @@ async fn test_forwarding_with_cache_warming_enables_routing() { cleanup_nodes(&mut nodes).await; } + +// ============================================================================ +// ECN Tests +// ============================================================================ + +use crate::node::handlers::session::mark_ipv6_ecn_ce; +use crate::node::TransportDropState; +use crate::transport::TransportId; + +/// Build a minimal IPv6 header (40 bytes) with specified ECN bits. +fn make_ipv6_packet_with_ecn(ecn: u8) -> Vec { + let mut pkt = vec![0u8; 40]; + let tc = ecn; // DSCP=0, ECN=ecn + pkt[0] = 0x60 | (tc >> 4); + pkt[1] = tc << 4; + pkt +} + +/// Extract ECN bits from an IPv6 packet. +fn read_ecn(pkt: &[u8]) -> u8 { + let tc = ((pkt[0] & 0x0F) << 4) | (pkt[1] >> 4); + tc & 0x03 +} + +#[test] +fn test_mark_ecn_ce_on_ect0() { + let mut pkt = make_ipv6_packet_with_ecn(0b10); + assert_eq!(read_ecn(&pkt), 0b10); + mark_ipv6_ecn_ce(&mut pkt); + assert_eq!(read_ecn(&pkt), 0b11); +} + +#[test] +fn test_mark_ecn_ce_on_ect1() { + let mut pkt = make_ipv6_packet_with_ecn(0b01); + assert_eq!(read_ecn(&pkt), 0b01); + mark_ipv6_ecn_ce(&mut pkt); + assert_eq!(read_ecn(&pkt), 0b11); +} + +#[test] +fn test_mark_ecn_ce_on_not_ect() { + let mut pkt = make_ipv6_packet_with_ecn(0b00); + mark_ipv6_ecn_ce(&mut pkt); + assert_eq!(read_ecn(&pkt), 0b00); +} + +#[test] +fn test_mark_ecn_ce_already_ce() { + let mut pkt = make_ipv6_packet_with_ecn(0b11); + mark_ipv6_ecn_ce(&mut pkt); + assert_eq!(read_ecn(&pkt), 0b11); +} + +#[test] +fn test_mark_ecn_ce_preserves_dscp_and_flow_label() { + let mut pkt = vec![0u8; 40]; + // DSCP=0b101100 (46=EF), ECN=ECT(0)=0b10 → TC=0xB2 + let tc: u8 = 0xB2; + pkt[0] = 0x60 | (tc >> 4); // 0x6B + pkt[1] = (tc << 4) | 0x0A; // 0x2A (flow label high nibble = 0xA) + pkt[2] = 0xBC; + pkt[3] = 0xDE; + + mark_ipv6_ecn_ce(&mut pkt); + + let new_tc = ((pkt[0] & 0x0F) << 4) | (pkt[1] >> 4); + assert_eq!(new_tc, 0xB3, "TC should be 0xB3 (DSCP preserved, ECN=CE)"); + assert_eq!(pkt[0] >> 4, 6, "Version nibble preserved"); + assert_eq!(pkt[1] & 0x0F, 0x0A, "Flow label high nibble preserved"); + assert_eq!(pkt[2], 0xBC, "Flow label byte 2 preserved"); + assert_eq!(pkt[3], 0xDE, "Flow label byte 3 preserved"); +} + +#[test] +fn test_mark_ecn_ce_short_packet() { + let mut pkt = vec![0x60]; + mark_ipv6_ecn_ce(&mut pkt); + assert_eq!(pkt, vec![0x60]); + + let mut empty: Vec = vec![]; + mark_ipv6_ecn_ce(&mut empty); + assert!(empty.is_empty()); +} + +#[tokio::test] +async fn test_ce_relay_through_forwarding() { + // 3-node chain: 0 -- 1 -- 2 + // Send a datagram with CE set from node 0 to node 1. + // Node 1 should relay CE to node 2. + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test(3, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node2_addr = *nodes[2].node.node_addr(); + + // Record ecn_ce_count at node 2 before + let ce_before = nodes[2] + .node + .get_peer(&node1_addr) + .and_then(|p| p.mmp()) + .map(|m| m.receiver.ecn_ce_count()) + .unwrap_or(0); + + // Build a SessionDatagram from node 0 to node 2 + let dg = SessionDatagram::new( + node0_addr, + node2_addr, + vec![0x10, 0x00, 0x04, 0x00, 1, 2, 3, 4], + ); + let encoded = dg.encode(); + + // Send from node 0 to node 1 with CE flag set + nodes[0] + .node + .send_encrypted_link_message_with_ce(&node1_addr, &encoded, true) + .await + .unwrap(); + + // Process: node 1 receives (CE set), forwards to node 2 (CE relayed) + for _ in 0..3 { + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + } + + // Node 2's link-layer MMP should have received a CE-flagged frame from node 1 + let ce_after = nodes[2] + .node + .get_peer(&node1_addr) + .and_then(|p| p.mmp()) + .map(|m| m.receiver.ecn_ce_count()) + .unwrap_or(0); + + assert!( + ce_after > ce_before, + "Node 2 should see CE flag relayed from node 1 (before={ce_before}, after={ce_after})" + ); + + cleanup_nodes(&mut nodes).await; +} + +#[test] +fn test_detect_congestion_with_transport_drops() { + let mut node = make_node(); + + // No drops — detect_congestion should return false for any address + let fake_addr = NodeAddr::from_bytes([1; 16]); + assert!(!node.detect_congestion(&fake_addr)); + + // Simulate transport kernel drops + let tid = TransportId::new(1); + node.transport_drops.insert(tid, TransportDropState { + prev_drops: 100, + dropping: true, + }); + + // Now detect_congestion should return true (local transport congestion) + assert!(node.detect_congestion(&fake_addr)); + + // Clear the dropping flag — should return false again + node.transport_drops.get_mut(&tid).unwrap().dropping = false; + assert!(!node.detect_congestion(&fake_addr)); +} + +#[test] +fn test_detect_congestion_disabled_ecn() { + let mut node = make_node(); + node.config.node.ecn.enabled = false; + + // Even with transport drops, disabled ECN should return false + let tid = TransportId::new(1); + node.transport_drops.insert(tid, TransportDropState { + prev_drops: 50, + dropping: true, + }); + + let fake_addr = NodeAddr::from_bytes([1; 16]); + assert!(!node.detect_congestion(&fake_addr)); +} + +#[test] +fn test_sample_transport_congestion() { + let mut node = make_node(); + + // Insert a transport drop state with a baseline + let tid = TransportId::new(1); + node.transport_drops.insert(tid, TransportDropState { + prev_drops: 0, + dropping: false, + }); + + // No transports registered — sample_transport_congestion is a no-op + // (transport_drops entry stays unchanged) + node.sample_transport_congestion(); + assert!(!node.transport_drops[&tid].dropping); +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index b5a98ac..0d223a8 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -791,6 +791,21 @@ pub trait Transport { } } +// ============================================================================ +// Transport Congestion +// ============================================================================ + +/// Transport-local congestion indicators. +/// +/// All fields are optional — transports report what they can. +/// Consumers compute deltas from cumulative counters. +#[derive(Clone, Debug, Default)] +pub struct TransportCongestion { + /// Cumulative packets dropped by kernel/OS before reaching the application. + /// Monotonically increasing since transport start. + pub recv_drops: Option, +} + // ============================================================================ // Transport Handle // ============================================================================ @@ -971,6 +986,20 @@ impl TransportHandle { self.state().is_operational() } + /// Query transport-local congestion indicators. + /// + /// Returns a snapshot of congestion signals that the transport can + /// observe locally (e.g., kernel receive buffer drops). Fields are + /// `None` when the transport doesn't support that signal. + pub fn congestion(&self) -> TransportCongestion { + match self { + TransportHandle::Udp(t) => t.congestion(), + #[cfg(target_os = "linux")] + TransportHandle::Ethernet(_) => TransportCongestion::default(), + TransportHandle::Tcp(_) => TransportCongestion::default(), + } + } + /// Get transport-specific stats as a JSON value. /// /// Returns a snapshot of counters for the specific transport type. diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index 61252dd..1b9e1d8 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -6,13 +6,13 @@ use super::{ DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, TransportId, TransportState, TransportType, }; +mod socket; mod stats; +use socket::{AsyncUdpSocket, UdpRawSocket}; use stats::UdpStats; use crate::config::UdpConfig; -use socket2::{Domain, Protocol, Socket, Type}; use std::net::SocketAddr; use std::sync::Arc; -use tokio::net::UdpSocket; use tokio::task::JoinHandle; use tracing::{debug, info, trace, warn}; @@ -31,7 +31,7 @@ pub struct UdpTransport { /// Current state. state: TransportState, /// Bound socket (None until started). - socket: Option>, + socket: Option, /// Channel for delivering received packets to Node. packet_tx: PacketTx, /// Receive loop task handle. @@ -73,16 +73,18 @@ impl UdpTransport { self.local_addr } - /// Get a reference to the socket (only valid after start). - pub fn socket(&self) -> Option<&Arc> { - self.socket.as_ref() - } - /// Get the transport statistics. pub fn stats(&self) -> &Arc { &self.stats } + /// Query transport-local congestion indicators. + pub fn congestion(&self) -> super::TransportCongestion { + super::TransportCongestion { + recv_drops: Some(self.stats.kernel_drops.load(std::sync::atomic::Ordering::Relaxed)), + } + } + /// Start the transport asynchronously. /// /// Binds the UDP socket and spawns the receive loop. @@ -100,56 +102,20 @@ impl UdpTransport { .parse() .map_err(|e| TransportError::StartFailed(format!("invalid bind address: {}", e)))?; - // Create socket via socket2 for buffer size control - let domain = if bind_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 }; - let sock2 = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) - .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; - sock2.set_nonblocking(true) - .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; - sock2.bind(&bind_addr.into()) - .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + // Create, bind, and configure UDP socket + let raw_socket = UdpRawSocket::open( + bind_addr, + self.config.recv_buf_size(), + self.config.send_buf_size(), + )?; - // Set socket buffer sizes - let recv_buf = self.config.recv_buf_size(); - let send_buf = self.config.send_buf_size(); - sock2.set_recv_buffer_size(recv_buf) - .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; - sock2.set_send_buffer_size(send_buf) - .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + let actual_recv = raw_socket.recv_buffer_size()?; + let actual_send = raw_socket.send_buffer_size()?; + self.local_addr = Some(raw_socket.local_addr()); - let actual_recv = sock2.recv_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; - let actual_send = sock2.send_buffer_size() - .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; - - if actual_recv < recv_buf { - warn!( - requested = recv_buf, - actual = actual_recv, - "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" - ); - } - if actual_send < send_buf { - warn!( - requested = send_buf, - actual = actual_send, - "UDP send buffer clamped by kernel (increase net.core.wmem_max)" - ); - } - - // Convert to tokio UdpSocket - let std_socket: std::net::UdpSocket = sock2.into(); - let socket = UdpSocket::from_std(std_socket) - .map_err(|e| TransportError::StartFailed(format!("tokio socket failed: {}", e)))?; - - self.local_addr = Some( - socket - .local_addr() - .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?, - ); - - let socket = Arc::new(socket); - self.socket = Some(socket.clone()); + // Wrap in AsyncFd for tokio integration + let async_socket = raw_socket.into_async()?; + self.socket = Some(async_socket.clone()); // Spawn receive loop let transport_id = self.transport_id; @@ -158,7 +124,7 @@ impl UdpTransport { let stats = self.stats.clone(); let recv_task = tokio::spawn(async move { - udp_receive_loop(socket, transport_id, packet_tx, mtu, stats).await; + udp_receive_loop(async_socket, transport_id, packet_tx, mtu, stats).await; }); self.recv_task = Some(recv_task); @@ -231,7 +197,7 @@ impl UdpTransport { let socket_addr = parse_socket_addr(addr)?; let socket = self.socket.as_ref().ok_or(TransportError::NotStarted)?; - match socket.send_to(data, socket_addr).await { + match socket.send_to(data, &socket_addr).await { Ok(bytes_sent) => { self.stats.record_send(bytes_sent); trace!( @@ -244,7 +210,7 @@ impl UdpTransport { } Err(e) => { self.stats.record_send_error(); - Err(TransportError::SendFailed(format!("{}", e))) + Err(e) } } } @@ -305,7 +271,7 @@ fn parse_socket_addr(addr: &TransportAddr) -> Result /// UDP receive loop - runs as a spawned task. async fn udp_receive_loop( - socket: Arc, + socket: AsyncUdpSocket, transport_id: TransportId, packet_tx: PacketTx, mtu: u16, @@ -318,8 +284,9 @@ async fn udp_receive_loop( loop { match socket.recv_from(&mut buf).await { - Ok((len, remote_addr)) => { + Ok((len, remote_addr, kernel_drops)) => { stats.record_recv(len); + stats.set_kernel_drops(kernel_drops as u64); let data = buf[..len].to_vec(); let addr = TransportAddr::from_string(&remote_addr.to_string()); @@ -573,4 +540,14 @@ mod tests { let binary = TransportAddr::new(vec![0xff, 0x80]); assert!(parse_socket_addr(&binary).is_err()); } + + #[tokio::test] + async fn test_congestion_reports_kernel_drops() { + let (tx, _rx) = packet_channel(100); + let transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); + + // Before start, congestion should still report (from stats) + let cong = transport.congestion(); + assert_eq!(cong.recv_drops, Some(0)); + } } diff --git a/src/transport/udp/socket.rs b/src/transport/udp/socket.rs new file mode 100644 index 0000000..e33f0a0 --- /dev/null +++ b/src/transport/udp/socket.rs @@ -0,0 +1,288 @@ +//! UDP socket wrapper with `SO_RXQ_OVFL` kernel drop counter support. +//! +//! Provides sync send/recv operations over a `socket2::Socket` with +//! `recvmsg()` ancillary data parsing for the kernel receive buffer +//! drop counter. The async wrapper uses `tokio::io::unix::AsyncFd` +//! for integration with the tokio runtime. +//! +//! Follows the pattern established by `transport/ethernet/socket.rs`. + +use crate::transport::TransportError; +use socket2::{Domain, Protocol, Socket, Type}; +use std::net::SocketAddr; +use std::os::unix::io::{AsRawFd, RawFd}; +use std::sync::Arc; +use tokio::io::unix::AsyncFd; +use tracing::warn; + +/// Wrapper around a `socket2::Socket` providing sync send/recv with +/// `SO_RXQ_OVFL` ancillary data parsing. +pub struct UdpRawSocket { + inner: Socket, + local_addr: SocketAddr, +} + +impl UdpRawSocket { + /// Create, bind, and configure a UDP socket. + /// + /// Enables `SO_RXQ_OVFL` for kernel drop counting (non-fatal if + /// unsupported). Sets non-blocking mode for async integration. + pub fn open( + bind_addr: SocketAddr, + recv_buf_size: usize, + send_buf_size: usize, + ) -> Result { + let domain = if bind_addr.is_ipv4() { + Domain::IPV4 + } else { + Domain::IPV6 + }; + let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)) + .map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?; + + sock.set_nonblocking(true) + .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + + sock.bind(&bind_addr.into()) + .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + + // Set socket buffer sizes + sock.set_recv_buffer_size(recv_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; + sock.set_send_buffer_size(send_buf_size) + .map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?; + + let actual_recv = sock + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?; + let actual_send = sock + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?; + + if actual_recv < recv_buf_size { + warn!( + requested = recv_buf_size, + actual = actual_recv, + "UDP recv buffer clamped by kernel (increase net.core.rmem_max)" + ); + } + if actual_send < send_buf_size { + warn!( + requested = send_buf_size, + actual = actual_send, + "UDP send buffer clamped by kernel (increase net.core.wmem_max)" + ); + } + + // Enable SO_RXQ_OVFL for kernel drop counter in recvmsg ancillary data. + // Non-fatal: older kernels or non-Linux platforms may not support it. + #[cfg(target_os = "linux")] + { + let enable: libc::c_int = 1; + let ret = unsafe { + libc::setsockopt( + sock.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RXQ_OVFL, + &enable as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + warn!( + "setsockopt(SO_RXQ_OVFL) failed: {}", + std::io::Error::last_os_error() + ); + } + } + + let local_addr = sock + .local_addr() + .map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))? + .as_socket() + .ok_or_else(|| { + TransportError::StartFailed("local address is not an IP socket".into()) + })?; + + Ok(Self { + inner: sock, + local_addr, + }) + } + + /// Get the local bound address. + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + /// Get the actual receive buffer size granted by the kernel. + pub fn recv_buffer_size(&self) -> Result { + self.inner + .recv_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e))) + } + + /// Get the actual send buffer size granted by the kernel. + pub fn send_buffer_size(&self) -> Result { + self.inner + .send_buffer_size() + .map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e))) + } + + /// Synchronous send to a destination address. + /// + /// Returns the number of bytes sent, or an `io::Error`. + pub fn send_to(&self, data: &[u8], dest: &SocketAddr) -> std::io::Result { + let dest: socket2::SockAddr = (*dest).into(); + self.inner.send_to(data, &dest) + } + + /// Synchronous receive with `SO_RXQ_OVFL` ancillary data parsing. + /// + /// Returns `(bytes_read, source_addr, kernel_drops)`. The `kernel_drops` + /// value is a cumulative counter since socket creation; it is 0 if + /// `SO_RXQ_OVFL` is not supported. + pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> { + let fd = self.inner.as_raw_fd(); + + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }; + + // Control message buffer sized for SO_RXQ_OVFL (u32). + // CMSG_SPACE computes the aligned size including header. + const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize; + let mut cmsg_buf = [0u8; CMSG_BUF_SIZE]; + + let mut src_addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_name = &mut src_addr as *mut _ as *mut libc::c_void; + msg.msg_namelen = std::mem::size_of::() as libc::socklen_t; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = cmsg_buf.len() as _; + + let n = unsafe { libc::recvmsg(fd, &mut msg, 0) }; + if n < 0 { + return Err(std::io::Error::last_os_error()); + } + + // Parse source address from sockaddr_storage + let addr = sockaddr_to_socket_addr(&src_addr)?; + + // Walk cmsg chain for SO_RXQ_OVFL drop counter + let mut drops: u32 = 0; + #[cfg(target_os = "linux")] + unsafe { + let mut cmsg = libc::CMSG_FIRSTHDR(&msg); + while !cmsg.is_null() { + if (*cmsg).cmsg_level == libc::SOL_SOCKET + && (*cmsg).cmsg_type == libc::SO_RXQ_OVFL + { + let data = libc::CMSG_DATA(cmsg); + drops = std::ptr::read_unaligned(data as *const u32); + } + cmsg = libc::CMSG_NXTHDR(&msg, cmsg); + } + } + + Ok((n as usize, addr, drops)) + } + + /// Wrap this socket in a tokio `AsyncFd` for async I/O. + pub fn into_async(self) -> Result { + let async_fd = AsyncFd::new(self) + .map_err(|e| TransportError::StartFailed(format!("AsyncFd::new failed: {}", e)))?; + Ok(AsyncUdpSocket { + inner: Arc::new(async_fd), + }) + } +} + +impl AsRawFd for UdpRawSocket { + fn as_raw_fd(&self) -> RawFd { + self.inner.as_raw_fd() + } +} + +/// Async wrapper around `UdpRawSocket` using tokio's `AsyncFd`. +/// +/// `Arc`-shareable between send and receive tasks. `AsyncFd` is +/// `Sync` when `T: Send`, which `socket2::Socket` satisfies. +#[derive(Clone)] +pub struct AsyncUdpSocket { + inner: Arc>, +} + +impl AsyncUdpSocket { + /// Send a payload to a destination address. + pub async fn send_to( + &self, + data: &[u8], + dest: &SocketAddr, + ) -> Result { + loop { + let mut guard = self + .inner + .writable() + .await + .map_err(|e| TransportError::SendFailed(format!("writable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().send_to(data, dest)) { + Ok(Ok(n)) => return Ok(n), + Ok(Err(e)) => return Err(TransportError::SendFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } + + /// Receive a payload, source address, and kernel drop counter. + /// + /// Returns `(bytes_read, source_addr, kernel_drops)`. + pub async fn recv_from( + &self, + buf: &mut [u8], + ) -> Result<(usize, SocketAddr, u32), TransportError> { + loop { + let mut guard = self + .inner + .readable() + .await + .map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?; + + match guard.try_io(|inner| inner.get_ref().recv_from(buf)) { + Ok(Ok(result)) => return Ok(result), + Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))), + Err(_would_block) => continue, + } + } + } +} + +/// Convert a `libc::sockaddr_storage` to `std::net::SocketAddr`. +fn sockaddr_to_socket_addr( + storage: &libc::sockaddr_storage, +) -> std::io::Result { + match storage.ss_family as libc::c_int { + libc::AF_INET => { + let addr: &libc::sockaddr_in = + unsafe { &*(storage as *const _ as *const libc::sockaddr_in) }; + let ip = std::net::Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); + let port = u16::from_be(addr.sin_port); + Ok(SocketAddr::from((ip, port))) + } + libc::AF_INET6 => { + let addr: &libc::sockaddr_in6 = + unsafe { &*(storage as *const _ as *const libc::sockaddr_in6) }; + let ip = std::net::Ipv6Addr::from(addr.sin6_addr.s6_addr); + let port = u16::from_be(addr.sin6_port); + Ok(SocketAddr::from((ip, port))) + } + family => Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("unsupported address family: {}", family), + )), + } +} diff --git a/testing/chaos/README.md b/testing/chaos/README.md index 3a65669..1168dac 100644 --- a/testing/chaos/README.md +++ b/testing/chaos/README.md @@ -105,6 +105,51 @@ Explicit topologies exercising non-UDP transports. static peer config. Netem mutation (30% fraction, every 20-40s) and link flaps (1 link max, 10-20s down). +### Congestion and ECN + +Scenarios testing ECN congestion signaling and transport-level congestion +detection. + +| Scenario | Nodes | Topology | Duration | What it tests | +| ------------------ | ----- | -------- | -------- | ---------------------------------------------------------- | +| congestion-stress | 10 | Tree | 120s | CE marking under kernel drops and MMP loss detection | +| ecn-ab-on / ecn-ab-off | 6 | Tree | 120s | A/B throughput comparison: ECN enabled vs disabled | + +- **congestion-stress**: 10-node tree with 1 Mbps egress bandwidth caps, + 5-10% netem loss, and heavy iperf3 traffic. Ingress policing (1000 kbps) + and small `recv_buf_size` (4 KB) trigger both MMP loss detection and + `SO_RXQ_OVFL` kernel socket drops. Validates end-to-end CE propagation: + transit nodes detect congestion, set CE flag, destinations receive + CE-marked packets, `ecn_ce_count` reported in MMP. +- **ecn-ab-on / ecn-ab-off**: Paired scenarios with identical conditions + (6-node tree, 10 Mbps egress, 1000 kbps ingress policing, 10ms link + delay, 8 KB recv buffer) differing only in `ecn.enabled`. + `ecn-ab-test.sh` runs both and compares throughput and congestion + counters. Initial results: +10.2% recv throughput with ECN enabled. + +### Ingress Traffic Control + +Scenarios can include `ingress` configuration to simulate upstream bandwidth +bottlenecks using tc ingress policing: + +```yaml +ingress: + enabled: true + tiers_kbps: [1000] # per-peer rate limit in kbps + burst_bytes: 10000 # policer burst allowance +``` + +Per-peer u32 filters on the ingress qdisc (`parent ffff:`) rate-limit +inbound packets. Combined with small `recv_buf_size`, this reliably triggers +`SO_RXQ_OVFL` kernel socket drops for congestion detection testing. + +### iperf3 JSON Capture + +Traffic sessions capture iperf3 results using `--json` output. Results are +collected per-session from containers and saved as `iperf3-results.json` in +the scenario output directory, enabling automated throughput analysis across +scenario runs. + ## CLI Options | Option | Description | diff --git a/testing/chaos/ecn-ab-test.sh b/testing/chaos/ecn-ab-test.sh new file mode 100755 index 0000000..1aa47e5 --- /dev/null +++ b/testing/chaos/ecn-ab-test.sh @@ -0,0 +1,127 @@ +#!/usr/bin/env bash +# ECN A/B Throughput Test +# +# Runs two identical chaos scenarios — one with ECN enabled, one disabled — +# and compares iperf3 throughput and congestion counter results. +# +# Usage: ./ecn-ab-test.sh [--seed N] [--duration N] + +set -euo pipefail +cd "$(dirname "$0")" + +EXTRA_ARGS=() +while [[ $# -gt 0 ]]; do + case "$1" in + --seed|--duration) + EXTRA_ARGS+=("$1" "$2"); shift 2 ;; + *) + echo "Unknown arg: $1"; exit 1 ;; + esac +done + +echo "=== ECN A/B Throughput Test ===" +echo "" + +# --- Run A: ECN ON --- +echo "--- Phase A: ECN ENABLED ---" +sudo python3 -m sim scenarios/ecn-ab-on.yaml "${EXTRA_ARGS[@]}" || true +echo "" + +# --- Run B: ECN OFF --- +echo "--- Phase B: ECN DISABLED ---" +sudo python3 -m sim scenarios/ecn-ab-off.yaml "${EXTRA_ARGS[@]}" || true +echo "" + +# --- Compare results --- +echo "=== Results ===" +echo "" + +python3 - <<'PYEOF' +import json +import os +import sys + +def load_results(path): + if not os.path.exists(path): + return [] + with open(path) as f: + return json.load(f) + +def extract_throughput(results): + """Extract per-session throughput in Mbps from iperf3 JSON.""" + sessions = [] + for r in results: + meta = r.get("_meta", {}) + end = r.get("end", {}) + # sum_sent / sum_received contain aggregate stats + sent = end.get("sum_sent", {}) + recv = end.get("sum_received", {}) + sent_mbps = sent.get("bits_per_second", 0) / 1e6 + recv_mbps = recv.get("bits_per_second", 0) / 1e6 + sessions.append({ + "client": meta.get("client", "?"), + "server": meta.get("server", "?"), + "sent_mbps": sent_mbps, + "recv_mbps": recv_mbps, + }) + return sessions + +def load_congestion(path): + if not os.path.exists(path): + return {} + with open(path) as f: + return json.load(f) + +def print_sessions(label, sessions): + # Filter out incomplete sessions (killed at teardown, no valid data) + valid = [s for s in sessions if s["recv_mbps"] > 0] + incomplete = len(sessions) - len(valid) + if not valid: + print(f" {label}: no completed iperf3 sessions ({incomplete} incomplete)") + return 0 + print(f" {label}:") + total_sent = 0 + total_recv = 0 + for s in valid: + print(f" {s['client']:>4} -> {s['server']:<4} " + f"sent={s['sent_mbps']:7.2f} Mbps recv={s['recv_mbps']:7.2f} Mbps") + total_sent += s["sent_mbps"] + total_recv += s["recv_mbps"] + n = len(valid) + print(f" {'':>14} avg sent={total_sent/n:7.2f} Mbps avg recv={total_recv/n:7.2f} Mbps") + print(f" {'':>14} completed={n} incomplete={incomplete}") + return total_recv / n + +on_results = load_results("sim-results/ecn-ab-on/iperf3-results.json") +off_results = load_results("sim-results/ecn-ab-off/iperf3-results.json") + +on_sessions = extract_throughput(on_results) +off_sessions = extract_throughput(off_results) + +print("Throughput:") +avg_on = print_sessions("ECN ON", on_sessions) or 0 +print() +avg_off = print_sessions("ECN OFF", off_sessions) or 0 +print() + +if avg_on and avg_off: + delta_pct = ((avg_on - avg_off) / avg_off) * 100 + print(f" Delta: ECN ON vs OFF = {delta_pct:+.1f}% avg recv throughput") + print() + +# Congestion counters +print("Congestion Counters (final snapshot):") +for label, path in [("ECN ON", "sim-results/ecn-ab-on/congestion-snapshot-final.json"), + ("ECN OFF", "sim-results/ecn-ab-off/congestion-snapshot-final.json")]: + snap = load_congestion(path) + if not snap: + print(f" {label}: no snapshot") + continue + totals = {"ce_forwarded": 0, "ce_received": 0, "congestion_detected": 0, "kernel_drop_events": 0} + for node_id, data in sorted(snap.items()): + cong = data.get("congestion", {}) + for k in totals: + totals[k] += cong.get(k, 0) + print(f" {label}: ce_fwd={totals['ce_forwarded']} ce_recv={totals['ce_received']} " + f"cong_detect={totals['congestion_detected']} kern_drops={totals['kernel_drop_events']}") +PYEOF diff --git a/testing/chaos/entrypoint.sh b/testing/chaos/entrypoint.sh index 98c776c..c0e13c0 100755 --- a/testing/chaos/entrypoint.sh +++ b/testing/chaos/entrypoint.sh @@ -8,6 +8,12 @@ set -e +# Enable TCP ECN negotiation for both IPv4 and IPv6 connections. +# Despite the "ipv4" name, this sysctl controls ECN for all TCP. +# Without this, IPv6 packets traversing the FIPS mesh carry ECN=0b00 +# (Not-ECT), and mark_ipv6_ecn_ce() is a no-op per RFC 3168. +sysctl -w net.ipv4.tcp_ecn=1 >/dev/null 2>&1 || true + # Start background services dnsmasq /usr/sbin/sshd diff --git a/testing/chaos/scenarios/congestion-stress.yaml b/testing/chaos/scenarios/congestion-stress.yaml new file mode 100644 index 0000000..8b68c72 --- /dev/null +++ b/testing/chaos/scenarios/congestion-stress.yaml @@ -0,0 +1,104 @@ +# Congestion Stress: exercise both MMP-based and kernel-drop congestion detection +# +# Topology: 10-node tree with all links bandwidth-limited to 1 Mbps +# (egress HTB + ingress policing). Heavy iperf3 traffic (8 concurrent +# sessions, 8 parallel streams) combined with 5-10% netem loss. +# +# Congestion detection signals exercised: +# 1. MMP loss detection: netem loss exceeds the 5% loss_threshold, +# triggering detect_congestion() via MMP metrics on transit nodes. +# 2. Kernel socket drops: small recv_buf_size (8KB) combined with +# traffic saturation causes SO_RXQ_OVFL on the UDP socket, +# triggering the transport drop detection path. +# 3. Ingress policing: tc policer on the receive side drops excess +# inbound packets, creating bursty arrival patterns. +# +# ECN is explicitly enabled via fips_overrides. +# +# Success criteria (verified via post-run congestion snapshot): +# - congestion_detected > 0 on at least one forwarding node +# - kernel_drop_events > 0 on at least one node +# - ce_forwarded > 0 on transit nodes +# - ce_received > 0 on destination nodes +# +# Topology: +# +# n01 (root) +# / \ +# n02 n03 +# / \ / \ +# n04 n05 n06 n07 +# | | +# n08 n09 +# | +# n10 + +scenario: + name: "congestion-stress" + seed: 7 + duration_secs: 120 + +topology: + algorithm: explicit + num_nodes: 10 + params: + adjacency: + - [n01, n02] + - [n01, n03] + - [n02, n04] + - [n02, n05] + - [n03, n06] + - [n03, n07] + - [n04, n08] + - [n05, n09] + - [n08, n10] + subnet: "172.20.0.0/24" + ip_start: 10 + +bandwidth: + enabled: true + tiers_mbps: [1] + +ingress: + enabled: true + tiers_kbps: [1000] + burst_bytes: 16000 + +netem: + enabled: true + default_policy: + delay_ms: [2, 5] + jitter_ms: [0, 1] + loss_pct: [5, 10] + mutation: + interval_secs: {min: 9999, max: 9999} + fraction: 0.0 + policies: + baseline: + delay_ms: [2, 5] + loss_pct: [5, 10] + +link_flaps: + enabled: false + +traffic: + enabled: true + max_concurrent: 8 + interval_secs: {min: 2, max: 5} + duration_secs: {min: 30, max: 60} + parallel_streams: 8 + +node_churn: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results" + +fips_overrides: + node: + ecn: + enabled: true + transports: + udp: + recv_buf_size: 4096 diff --git a/testing/chaos/scenarios/ecn-ab-off.yaml b/testing/chaos/scenarios/ecn-ab-off.yaml new file mode 100644 index 0000000..8cb05e0 --- /dev/null +++ b/testing/chaos/scenarios/ecn-ab-off.yaml @@ -0,0 +1,80 @@ +# ECN A/B Test: ECN DISABLED +# +# Identical to ecn-ab-on.yaml except ecn.enabled = false. +# Run both scenarios and compare iperf3 throughput results. +# +# Same conditions: 10 Mbps egress, 1000 kbps ingress policing, 10ms +# link delay, 8KB recv buffer. Without ECN, the only congestion signal +# TCP gets is packet loss (timeouts/triple dup-ack). +# +# n01 (root) +# / \ +# n02 n03 +# / \ \ +# n04 n05 n06 + +scenario: + name: "ecn-ab-off" + seed: 42 + duration_secs: 120 + +topology: + algorithm: explicit + num_nodes: 6 + params: + adjacency: + - [n01, n02] + - [n01, n03] + - [n02, n04] + - [n02, n05] + - [n03, n06] + subnet: "172.20.0.0/24" + ip_start: 10 + +bandwidth: + enabled: true + tiers_mbps: [10] + +ingress: + enabled: true + tiers_kbps: [1000] + burst_bytes: 16000 + +netem: + enabled: true + default_policy: + delay_ms: [10, 10] + jitter_ms: [0, 0] + loss_pct: [0, 0] + mutation: + interval_secs: {min: 9999, max: 9999} + fraction: 0.0 + policies: + baseline: + delay_ms: [10, 10] + loss_pct: [0, 0] + +link_flaps: + enabled: false + +traffic: + enabled: true + max_concurrent: 4 + interval_secs: {min: 3, max: 5} + duration_secs: {min: 20, max: 30} + parallel_streams: 4 + +node_churn: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results/ecn-ab-off" + +fips_overrides: + node: + ecn: + enabled: false + transports: + udp: + recv_buf_size: 8192 diff --git a/testing/chaos/scenarios/ecn-ab-on.yaml b/testing/chaos/scenarios/ecn-ab-on.yaml new file mode 100644 index 0000000..a7002e0 --- /dev/null +++ b/testing/chaos/scenarios/ecn-ab-on.yaml @@ -0,0 +1,100 @@ +# ECN A/B Test: ECN ENABLED +# +# Identical to ecn-ab-off.yaml except ecn.enabled = true. +# Run both scenarios and compare iperf3 throughput results. +# +# Design rationale: +# - Ingress policing at 1000 kbps: unlike HTB (which paces smoothly), +# the ingress policer *drops* excess inbound packets at the kernel +# level. This creates real packet loss visible to the FIPS UDP +# socket, triggering detect_congestion() via transport_drops. +# - 10 Mbps egress bandwidth: lets traffic flow freely outbound so +# that the ingress policer is the bottleneck. Multiple flows +# competing through bottleneck nodes generate excess traffic that +# the policer drops. +# - 10ms link delay: creates realistic RTT and causes TCP to batch +# packets into bursts, making ingress drops more likely. +# - 8KB recv_buf_size: small buffer amplifies the effect of bursty +# arrivals — dropped packets at ingress + buffer overflow trigger +# detect_congestion() via both MMP loss and transport_drops paths. +# - tcp_ecn=1 in entrypoint.sh: TCP negotiates ECN so CE marks +# reach TCP congestion control, enabling proactive backoff. +# - 4 concurrent sessions with 4 parallel streams: moderate load +# that creates real contention without overwhelming everything. +# +# With ECN enabled, transit nodes detect congestion early and set CE +# flags, causing TCP to reduce its window proactively. Without ECN, +# TCP only learns about congestion from packet drops (timeouts/triple +# dup-ack), which is slower and causes more throughput loss. +# +# n01 (root) +# / \ +# n02 n03 +# / \ \ +# n04 n05 n06 + +scenario: + name: "ecn-ab-on" + seed: 42 + duration_secs: 120 + +topology: + algorithm: explicit + num_nodes: 6 + params: + adjacency: + - [n01, n02] + - [n01, n03] + - [n02, n04] + - [n02, n05] + - [n03, n06] + subnet: "172.20.0.0/24" + ip_start: 10 + +bandwidth: + enabled: true + tiers_mbps: [10] + +ingress: + enabled: true + tiers_kbps: [1000] + burst_bytes: 16000 + +netem: + enabled: true + default_policy: + delay_ms: [10, 10] + jitter_ms: [0, 0] + loss_pct: [0, 0] + mutation: + interval_secs: {min: 9999, max: 9999} + fraction: 0.0 + policies: + baseline: + delay_ms: [10, 10] + loss_pct: [0, 0] + +link_flaps: + enabled: false + +traffic: + enabled: true + max_concurrent: 4 + interval_secs: {min: 3, max: 5} + duration_secs: {min: 20, max: 30} + parallel_streams: 4 + +node_churn: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results/ecn-ab-on" + +fips_overrides: + node: + ecn: + enabled: true + transports: + udp: + recv_buf_size: 8192 diff --git a/testing/chaos/sim/control.py b/testing/chaos/sim/control.py index 41cf6ed..9f9af40 100644 --- a/testing/chaos/sim/control.py +++ b/testing/chaos/sim/control.py @@ -101,3 +101,42 @@ def snapshot_all_mmp(topology: SimTopology) -> dict[str, dict]: else: log.warning("No MMP data from %s", node_id) return result + + +def query_routing(container: str) -> dict | None: + """Query a node's routing stats (includes congestion counters).""" + return query_node(container, "show_routing") + + +def query_transports(container: str) -> dict | None: + """Query a node's transport state (includes kernel drop counters).""" + return query_node(container, "show_transports") + + +def snapshot_all_congestion(topology: SimTopology) -> dict[str, dict]: + """Query show_routing on all nodes to capture congestion counters. + + Returns {node_id: {"congestion": {...}, "kernel_drops": [...]}}. + Nodes that fail to respond are omitted from the result. + """ + result = {} + for node_id in sorted(topology.nodes): + container = topology.container_name(node_id) + routing = query_routing(container) + transports = query_transports(container) + if routing is not None: + entry = {"congestion": routing.get("congestion", {})} + if transports is not None: + drops = [] + for t in transports.get("transports", []): + stats = t.get("stats", {}) + drops.append({ + "transport_id": t.get("transport_id"), + "name": t.get("name"), + "kernel_drops": stats.get("kernel_drops"), + }) + entry["kernel_drops"] = drops + result[node_id] = entry + else: + log.warning("No routing data from %s", node_id) + return result diff --git a/testing/chaos/sim/logs.py b/testing/chaos/sim/logs.py index 30cec94..5aff9e6 100644 --- a/testing/chaos/sim/logs.py +++ b/testing/chaos/sim/logs.py @@ -26,6 +26,8 @@ class AnalysisResult: mmp_session_metrics: list[tuple[str, str]] = field(default_factory=list) handshake_timeouts: list[tuple[str, str]] = field(default_factory=list) panics: list[tuple[str, str]] = field(default_factory=list) + congestion_detected: list[tuple[str, str]] = field(default_factory=list) + kernel_drop_events: list[tuple[str, str]] = field(default_factory=list) def summary(self) -> str: lines = [ @@ -41,6 +43,8 @@ class AnalysisResult: f"Handshake timeouts: {len(self.handshake_timeouts)}", f"MMP link samples: {len(self.mmp_link_metrics)}", f"MMP session samples: {len(self.mmp_session_metrics)}", + f"Congestion events: {len(self.congestion_detected)}", + f"Kernel drop events: {len(self.kernel_drop_events)}", ] if self.panics: @@ -134,6 +138,11 @@ def analyze_logs(logs: dict[str, str]) -> AnalysisResult: result.mmp_link_metrics.append((container, line)) if "MMP session metrics" in line: result.mmp_session_metrics.append((container, line)) + # Congestion events + if "Congestion detected" in line: + result.congestion_detected.append((container, line)) + if "Kernel recv drops first observed" in line: + result.kernel_drop_events.append((container, line)) return result diff --git a/testing/chaos/sim/netem.py b/testing/chaos/sim/netem.py index d2de330..45720a8 100644 --- a/testing/chaos/sim/netem.py +++ b/testing/chaos/sim/netem.py @@ -8,6 +8,13 @@ u32 filters match destination IP to direct traffic to the right class. ├── class 1:1 → netem 11: (peer 1) ← u32 filter: dst= ├── class 1:2 → netem 12: (peer 2) ← u32 filter: dst= └── class 1:99 → pfifo (default, no impairment) + +Optional ingress policing adds rate-based packet dropping on the +receive side via tc ingress qdisc + policer filters: + + eth0 ingress (ffff:) + ├── u32 filter: src= → police rate kbit burst drop + └── u32 filter: src= → police rate kbit burst drop """ from __future__ import annotations @@ -17,7 +24,7 @@ import random from dataclasses import dataclass, field from .docker_exec import docker_exec_quiet, is_container_running -from .scenario import BandwidthConfig, LinkPolicyOverride, NetemConfig, NetemPolicy +from .scenario import BandwidthConfig, IngressConfig, LinkPolicyOverride, NetemConfig, NetemPolicy from .topology import SimTopology, veth_interface_name log = logging.getLogger(__name__) @@ -65,6 +72,9 @@ class LinkNetemState: netem_handle: str # e.g., "11:" params: NetemParams = field(default_factory=NetemParams) rate_mbit: int = 0 # 0 = unlimited (1gbit default) + ingress_rate_kbps: int = 0 # 0 = no ingress policing + ingress_burst_bytes: int = 32000 + ingress_filter_prio: int = 0 # u32 filter priority for this peer @dataclass @@ -85,6 +95,7 @@ class NetemManager: config: NetemConfig, rng: random.Random, bandwidth: BandwidthConfig | None = None, + ingress: IngressConfig | None = None, ): self.topology = topology self.config = config @@ -102,6 +113,14 @@ class NetemManager: rate = rng.choice(bandwidth.tiers_mbps) self._edge_rates[(a, b)] = rate self._edge_rates[(b, a)] = rate + # Per-edge ingress policing: (node_a, node_b) -> rate in kbps + self._ingress_config = ingress + self._ingress_rates: dict[tuple[str, str], int] = {} + if ingress and ingress.enabled: + for a, b in topology.edges: + rate = rng.choice(ingress.tiers_kbps) + self._ingress_rates[(a, b)] = rate + self._ingress_rates[(b, a)] = rate # Per-edge policy overrides: canonical "nXX-nYY" -> NetemPolicy # Build a set of canonical edge strings for validation topo_edge_strs = {"-".join(sorted([a, b])) for a, b in topology.edges} @@ -194,6 +213,9 @@ class NetemManager: f"prio {idx} u32 match ip dst {dest_ip}/32 flowid {class_id}" ) + ingress_rate = self._ingress_rates.get((node_id, peer_id), 0) + ingress_burst = self._ingress_config.burst_bytes if self._ingress_config else 32000 + state = LinkNetemState( container=container, dest_ip=dest_ip, @@ -201,16 +223,33 @@ class NetemManager: netem_handle=netem_handle, params=params, rate_mbit=rate_mbit, + ingress_rate_kbps=ingress_rate, + ingress_burst_bytes=ingress_burst, + ingress_filter_prio=idx, ) container_states[dest_ip] = state + # Ingress policing: add ingress qdisc + per-peer policer filters + if self._ingress_rates: + cmds.append(f"tc qdisc add dev {IFACE} ingress 2>/dev/null || true") + for idx, (peer_id, dest_ip) in enumerate(ip_peers.items(), start=1): + ingress_rate = self._ingress_rates.get((node_id, peer_id), 0) + if ingress_rate > 0: + ingress_burst = self._ingress_config.burst_bytes if self._ingress_config else 32000 + cmds.append( + f"tc filter add dev {IFACE} parent ffff: protocol ip " + f"prio {idx} u32 match ip src {dest_ip}/32 " + f"police rate {ingress_rate}kbit burst {ingress_burst} drop" + ) + full_cmd = " && ".join(cmds) result = docker_exec_quiet(container, full_cmd, timeout=30) if result is not None: log.info( - "Configured per-link netem on %s (%d IP peers)", + "Configured per-link netem on %s (%d IP peers%s)", container, len(ip_peers), + ", ingress policing" if self._ingress_rates else "", ) else: log.warning("Failed to configure netem on %s", container) @@ -282,13 +321,27 @@ class NetemManager: f"prio {prio} u32 match ip dst {dest_ip}/32 flowid {state.class_id}" ) + # Re-apply ingress policing + has_ingress = any(s.ingress_rate_kbps > 0 for s in container_states.values()) + if has_ingress: + cmds.append(f"tc qdisc add dev {IFACE} ingress 2>/dev/null || true") + for dest_ip, state in container_states.items(): + if state.ingress_rate_kbps > 0: + cmds.append( + f"tc filter add dev {IFACE} parent ffff: protocol ip " + f"prio {state.ingress_filter_prio} u32 match ip src {dest_ip}/32 " + f"police rate {state.ingress_rate_kbps}kbit " + f"burst {state.ingress_burst_bytes} drop" + ) + full_cmd = " && ".join(cmds) result = docker_exec_quiet(container, full_cmd, timeout=30) if result is not None: log.info( - "Re-applied IP netem on %s (%d peers)", + "Re-applied IP netem on %s (%d peers%s)", container, len(container_states), + ", ingress policing" if has_ingress else "", ) else: log.warning("Failed to re-apply IP netem on %s", container) diff --git a/testing/chaos/sim/runner.py b/testing/chaos/sim/runner.py index 6c94dec..f1280d1 100644 --- a/testing/chaos/sim/runner.py +++ b/testing/chaos/sim/runner.py @@ -12,7 +12,7 @@ import time from .compose import generate_compose from .config_gen import write_configs -from .control import snapshot_all_mmp, snapshot_all_trees +from .control import snapshot_all_congestion, snapshot_all_mmp, snapshot_all_trees from .docker_exec import docker_compose from .links import LinkManager from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata @@ -140,7 +140,8 @@ class SimRunner: # 7. Initialize managers if s.netem.enabled: bw = s.bandwidth if s.bandwidth.enabled else None - self.netem_mgr = NetemManager(self.topology, s.netem, self.rng, bandwidth=bw) + ig = s.ingress if s.ingress.enabled else None + self.netem_mgr = NetemManager(self.topology, s.netem, self.rng, bandwidth=bw, ingress=ig) self.netem_mgr.down_nodes = self._down_nodes log.info("Applying initial per-link netem...") self.netem_mgr.setup_initial() @@ -254,6 +255,15 @@ class SimRunner: log.info("Restoring stopped nodes...") self.node_mgr.restore_all() + # Collect iperf3 throughput results before containers stop + if self.traffic_mgr: + iperf_results = self.traffic_mgr.collect_results() + if iperf_results: + iperf_path = os.path.join(self.output_dir, "iperf3-results.json") + with open(iperf_path, "w") as f: + json.dump(iperf_results, f, indent=2) + log.info("Saved %d iperf3 results to %s", len(iperf_results), iperf_path) + # Take final tree snapshot while nodes are still running self._take_snapshot("final") @@ -298,27 +308,33 @@ class SimRunner: return result def _take_snapshot(self, label: str): - """Query all nodes via control socket and save tree/MMP snapshots.""" + """Query all nodes via control socket and save tree/MMP/congestion snapshots.""" if not self.topology: return log.info("Taking %s snapshot...", label) tree_snap = snapshot_all_trees(self.topology) mmp_snap = snapshot_all_mmp(self.topology) + congestion_snap = snapshot_all_congestion(self.topology) tree_path = os.path.join(self.output_dir, f"tree-snapshot-{label}.json") mmp_path = os.path.join(self.output_dir, f"mmp-snapshot-{label}.json") + congestion_path = os.path.join(self.output_dir, f"congestion-snapshot-{label}.json") os.makedirs(self.output_dir, exist_ok=True) with open(tree_path, "w") as f: json.dump(tree_snap, f, indent=2) with open(mmp_path, "w") as f: json.dump(mmp_snap, f, indent=2) + with open(congestion_path, "w") as f: + json.dump(congestion_snap, f, indent=2) log.info( - "Snapshot %s: %d/%d tree, %d/%d mmp responses", + "Snapshot %s: %d/%d tree, %d/%d mmp, %d/%d congestion responses", label, len(tree_snap), len(self.topology.nodes), len(mmp_snap), len(self.topology.nodes), + len(congestion_snap), + len(self.topology.nodes), ) def _schedule_next(self, now: float, interval) -> float: diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index 6416ffc..f1c5efe 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -120,6 +120,22 @@ class BandwidthConfig: tiers_mbps: list[int] = field(default_factory=lambda: [1, 10, 100, 1000]) +@dataclass +class IngressConfig: + """Per-link ingress policing via tc ingress qdisc + policer. + + When enabled, each link gets an ingress policer that drops inbound + packets exceeding the configured rate. Unlike egress HTB (which + paces packets smoothly), the policer drops excess packets at the + kernel level, creating bursty arrival patterns that can overflow + small socket buffers and trigger SO_RXQ_OVFL kernel drops. + """ + + enabled: bool = False + tiers_kbps: list[int] = field(default_factory=lambda: [1000]) + burst_bytes: int = 32000 + + @dataclass class LoggingConfig: rust_log: str = "info" @@ -137,6 +153,7 @@ class Scenario: traffic: TrafficConfig = field(default_factory=TrafficConfig) node_churn: NodeChurnConfig = field(default_factory=NodeChurnConfig) bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig) + ingress: IngressConfig = field(default_factory=IngressConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Raw YAML dict appended to each generated FIPS node config. # Allows scenarios to override any FIPS config parameter @@ -271,6 +288,16 @@ def load_scenario(path: str) -> Scenario: raise ValueError("bandwidth.tiers_mbps must be a non-empty list") s.bandwidth.tiers_mbps = [int(t) for t in tiers] + # Ingress section + ig = raw.get("ingress", {}) + s.ingress.enabled = ig.get("enabled", False) + if "tiers_kbps" in ig: + tiers = ig["tiers_kbps"] + if not isinstance(tiers, list) or not tiers: + raise ValueError("ingress.tiers_kbps must be a non-empty list") + s.ingress.tiers_kbps = [int(t) for t in tiers] + s.ingress.burst_bytes = int(ig.get("burst_bytes", 32000)) + # Logging section lg = raw.get("logging", {}) s.logging.rust_log = lg.get("rust_log", "info") @@ -376,3 +403,9 @@ def _validate(s: Scenario): for tier in s.bandwidth.tiers_mbps: if tier <= 0: raise ValueError(f"bandwidth.tiers_mbps: all values must be > 0, got {tier}") + if s.ingress.enabled: + for tier in s.ingress.tiers_kbps: + if tier <= 0: + raise ValueError(f"ingress.tiers_kbps: all values must be > 0, got {tier}") + if s.ingress.burst_bytes <= 0: + raise ValueError(f"ingress.burst_bytes must be > 0, got {s.ingress.burst_bytes}") diff --git a/testing/chaos/sim/traffic.py b/testing/chaos/sim/traffic.py index 12ea717..76061be 100644 --- a/testing/chaos/sim/traffic.py +++ b/testing/chaos/sim/traffic.py @@ -7,6 +7,7 @@ entrypoint). from __future__ import annotations +import json import logging import random import time @@ -26,6 +27,7 @@ class TrafficSession: started_at: float duration_secs: int container: str + result_file: str = "" class TrafficManager: @@ -43,6 +45,7 @@ class TrafficManager: self.rng = rng self.down_nodes = down_nodes or set() self.active_sessions: list[TrafficSession] = [] + self.completed_results: list[dict] = [] @property def active_count(self) -> int: @@ -74,10 +77,14 @@ class TrafficManager: ) streams = self.config.parallel_streams - # Start iperf3 in background (nohup, stdout to /dev/null) + # Result file inside the container for JSON capture + ts = int(time.time()) + result_file = f"/tmp/iperf3-{client}-{server}-{ts}.json" + + # Start iperf3 in background with JSON output cmd = ( f"nohup iperf3 -c {server_npub}.fips -t {duration} " - f"-P {streams} > /dev/null 2>&1 &" + f"-P {streams} --json > {result_file} 2>&1 &" ) result = docker_exec_quiet(container, cmd) if result is not None: @@ -87,6 +94,7 @@ class TrafficManager: started_at=time.time(), duration_secs=duration, container=container, + result_file=result_file, ) self.active_sessions.append(session) log.info( @@ -103,16 +111,49 @@ class TrafficManager: """Remove sessions that have completed (based on time).""" now = time.time() grace = 5 # seconds after expected completion - before = len(self.active_sessions) - self.active_sessions = [ - s - for s in self.active_sessions - if now - s.started_at < s.duration_secs + grace - ] - removed = before - len(self.active_sessions) + still_active = [] + for s in self.active_sessions: + if now - s.started_at >= s.duration_secs + grace: + self._collect_result(s) + else: + still_active.append(s) + removed = len(self.active_sessions) - len(still_active) + self.active_sessions = still_active if removed > 0: log.debug("Cleaned up %d expired traffic sessions", removed) + def _collect_result(self, session: TrafficSession): + """Retrieve iperf3 JSON result from container.""" + if not session.result_file: + return + if session.client_node in self.down_nodes: + return + stdout = docker_exec_quiet( + session.container, + f"cat {session.result_file} 2>/dev/null; rm -f {session.result_file}", + ) + if stdout is None: + return + try: + data = json.loads(stdout.strip()) + except (json.JSONDecodeError, ValueError): + log.debug("Could not parse iperf3 result for %s -> %s", + session.client_node, session.server_node) + return + data["_meta"] = { + "client": session.client_node, + "server": session.server_node, + "duration_secs": session.duration_secs, + } + self.completed_results.append(data) + + def collect_results(self) -> list[dict]: + """Return all completed iperf3 JSON results.""" + # Collect any remaining active sessions that may have finished + for s in self.active_sessions: + self._collect_result(s) + return self.completed_results + def stop_all(self): """Kill all iperf3 client processes in running containers.""" seen = set() @@ -124,4 +165,7 @@ class TrafficManager: "killall iperf3 2>/dev/null; true", ) seen.add(session.container) + # Collect results before clearing (iperf3 writes partial JSON on kill) + for s in self.active_sessions: + self._collect_result(s) self.active_sessions.clear()