From a4e341cd66f1613e1e06710b9c2e2e4a048ca891 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Wed, 30 Sep 2026 17:31:17 +0100 Subject: [PATCH] Report peer tunnel protocols accurately and eliminate overlay shutdown timeout The status output previously reported all overlay peers under WireGuard diagnostics: when a non-WireGuard protocol (like tcp-tls) was actively carrying traffic, the report still complained that there was 'no WireGuard handshake yet; the tunnel cannot carry traffic' and marked peer health as degraded (warn). Furthermore, during agent shutdown or interface removal, the overlay packet loop waited on device.recv() for the full 5-second TASK_GRACE period, producing a spurious warning: 'task did not wind down in time; aborting it task="overlay packet loop"'. Key improvements: - Add 'protocol' field to OverlayPeerReport in tsunagi::ipc. - Track per-peer packet metrics (tx, rx, dropped) and uptime in GenericTunnelPlugin via TunnelPeerReport. - Update build_report to accurately inspect and populate active tunnel protocols (tcp-tls, wg-quic) and their real traffic counters. - Format member rows in CLI status with the actual protocol name, uptime, and real traffic metrics; mark health as Good when an active tunnel exists. - In Interface::remove(), abort the overlay packet loop task immediately before destroying the device to remove the 5s hang and spurious timeout warning. Validation on Linux: - cargo fmt --all -- --check passed. - cargo clippy --locked --workspace --all-targets -- -D warnings passed. - cargo test --locked --workspace --all-targets passed (348 tests passed). - cargo build --release passed. --- crates/tsunagi-cli/src/main.rs | 191 +++++++++++++++++------- crates/tsunagi/src/dataplane/tunnel.rs | 69 ++++++++- crates/tsunagi/src/ipc/mod.rs | 4 +- crates/tsunagi/src/overlay/interface.rs | 7 +- 4 files changed, 202 insertions(+), 69 deletions(-) diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index e3c59d6..6bcd9a3 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -2824,9 +2824,19 @@ fn member_row(row: &MemberRow<'_>) -> report::Row { let mut out = Row::new(health, row.label(), detail); if let Some(tunnel) = row.tunnel { + let proto = if tunnel.protocol.is_empty() { + "wg-quic" + } else { + &tunnel.protocol + }; out = out.with_note(match tunnel.handshake_secs_ago { Some(secs) => format!( - "tunnel up, handshake {secs}s ago, tx {} rx {}{} · {}", + "tunnel up ({proto}), {} tx {} rx {}{} · {}", + if proto == "tcp-tls" { + format!("uptime {secs}s,") + } else { + format!("handshake {secs}s ago,") + }, tunnel.tx_packets, tunnel.rx_packets, if tunnel.dropped > 0 { @@ -2836,7 +2846,13 @@ fn member_row(row: &MemberRow<'_>) -> report::Row { }, tunnel.path ), - None => "no WireGuard handshake yet; the tunnel cannot carry traffic".to_string(), + None => { + if proto == "wg-quic" { + "no WireGuard handshake yet; the tunnel cannot carry traffic".to_string() + } else { + format!("no {proto} handshake yet; the tunnel cannot carry traffic") + } + } }); } out @@ -3710,56 +3726,81 @@ async fn build_report( .iter() .map(|network| { let wg_view = wireguard.and_then(|plugin| plugin.overview(network.network_id)); - let tcp_tls_peers = tcp_tls + let tcp_tls_reports = tcp_tls .map(|p| p.active_peers(network.network_id)) .unwrap_or_default(); let overlay = if let Some(view) = wg_view { let mut peers = Vec::new(); for peer in &view.peers { - let active_tcp_tls_path = tcp_tls_peers - .iter() - .find(|(id, _)| *id == peer.endpoint_id) - .map(|(_, path)| path.clone()); + let tcp_tls_match = tcp_tls_reports.iter().find(|r| r.peer == peer.endpoint_id); - let (path, handshake) = if let Some(path) = active_tcp_tls_path { - ( - path, - peer.tunnel - .as_ref() - .and_then(|t| t.health.since_handshake) - .map(|s| s.as_secs()) - .or(Some(0)), - ) + if let Some(tcp_tls) = tcp_tls_match { + peers.push(OverlayPeerReport { + endpoint_id: peer.endpoint_id.to_string(), + protocol: tcp_tls.protocol.clone(), + public_key: peer.public_key.to_string(), + address: peer.overlay_address_v4.map(|addr| addr.to_string()), + handshake_secs_ago: Some(tcp_tls.uptime_secs), + tx_packets: tcp_tls.tx_packets, + rx_packets: tcp_tls.rx_packets, + dropped: tcp_tls.dropped, + protocol_errors: 0, + path: tcp_tls.path.clone(), + }); + } else if let Some(tunnel) = &peer.tunnel { + peers.push(OverlayPeerReport { + endpoint_id: peer.endpoint_id.to_string(), + protocol: "wg-quic".into(), + public_key: peer.public_key.to_string(), + address: peer.overlay_address_v4.map(|addr| addr.to_string()), + handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()), + tx_packets: tunnel.stats.tx_packets, + rx_packets: tunnel.stats.rx_packets, + dropped: tunnel.stats.dropped_wrong_source + + tunnel.stats.dropped_oversize, + protocol_errors: tunnel.stats.protocol_errors, + path: tunnel.path.clone(), + }); } else { - ( - peer.tunnel - .as_ref() - .map(|t| t.path.clone()) - .unwrap_or_else(|| "no data link".into()), - peer.tunnel - .as_ref() - .and_then(|t| t.health.since_handshake) - .map(|s| s.as_secs()), - ) - }; - - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: handshake, - tx_packets: peer.tunnel.as_ref().map_or(0, |t| t.stats.tx_packets), - rx_packets: peer.tunnel.as_ref().map_or(0, |t| t.stats.rx_packets), - dropped: peer.tunnel.as_ref().map_or(0, |t| { - t.stats.dropped_wrong_source + t.stats.dropped_oversize - }), - protocol_errors: peer - .tunnel - .as_ref() - .map_or(0, |t| t.stats.protocol_errors), - path, - }); + peers.push(OverlayPeerReport { + endpoint_id: peer.endpoint_id.to_string(), + protocol: "wg-quic".into(), + public_key: peer.public_key.to_string(), + address: peer.overlay_address_v4.map(|addr| addr.to_string()), + handshake_secs_ago: None, + tx_packets: 0, + rx_packets: 0, + dropped: 0, + protocol_errors: 0, + path: "no data link".into(), + }); + } + } + for report in &tcp_tls_reports { + if !peers + .iter() + .any(|p| p.endpoint_id == report.peer.to_string()) + { + let address = network + .members + .iter() + .find(|m| m.endpoint_id == report.peer) + .and_then(|m| m.overlay_address_v4) + .map(|a| a.to_string()); + peers.push(OverlayPeerReport { + endpoint_id: report.peer.to_string(), + protocol: report.protocol.clone(), + public_key: String::new(), + address, + handshake_secs_ago: Some(report.uptime_secs), + tx_packets: report.tx_packets, + rx_packets: report.rx_packets, + dropped: report.dropped, + protocol_errors: 0, + path: report.path.clone(), + }); + } } Some(OverlayReport { interface: overlay @@ -3776,19 +3817,28 @@ async fn build_report( .as_ref() .and_then(|o| o.counters.unroutable_sample.map(|a| a.to_string())), }) - } else if overlay.is_some() && !tcp_tls_peers.is_empty() { - let peers = tcp_tls_peers + } else if overlay.is_some() && !tcp_tls_reports.is_empty() { + let peers = tcp_tls_reports .into_iter() - .map(|(id, path)| OverlayPeerReport { - endpoint_id: id.to_string(), - public_key: String::new(), - address: None, - handshake_secs_ago: Some(0), - tx_packets: 0, - rx_packets: 0, - dropped: 0, - protocol_errors: 0, - path, + .map(|report| { + let address = network + .members + .iter() + .find(|m| m.endpoint_id == report.peer) + .and_then(|m| m.overlay_address_v4) + .map(|a| a.to_string()); + OverlayPeerReport { + endpoint_id: report.peer.to_string(), + protocol: report.protocol, + public_key: String::new(), + address, + handshake_secs_ago: Some(report.uptime_secs), + tx_packets: report.tx_packets, + rx_packets: report.rx_packets, + dropped: report.dropped, + protocol_errors: 0, + path: report.path, + } }) .collect(); Some(OverlayReport { @@ -3797,8 +3847,13 @@ async fn build_report( .map_or_else(String::new, |o| o.interface.clone()), on_host: overlay.as_ref().is_some_and(|o| o.on_host), mtu: overlay.as_ref().map_or(0, |o| o.mtu), - address: None, - prefix_len: 0, + address: network + .members + .iter() + .find(|m| m.endpoint_id == status.endpoint_id) + .and_then(|m| m.overlay_address_v4) + .map(|a| a.to_string()), + prefix_len: network.range.map_or(0, |r| r.prefix_len), peers, unroutable_packets: overlay.as_ref().map_or(0, |o| o.counters.unroutable), multicast_packets: overlay.as_ref().map_or(0, |o| o.counters.multicast), @@ -4399,6 +4454,26 @@ mod status_tests { assert_eq!(out.worst(), Health::Degraded, "{text}"); assert!(text.contains("same secret"), "{text}"); } + + #[test] + fn a_tcp_tls_tunnel_shows_healthy_state_and_metrics() { + let mut network = network_after_a_peer_returned(); + let mut t = tunnel(ONLINE, Some(42)); + t.protocol = "tcp-tls".into(); + t.path = "tcp-tls via 192.0.2.1:443".into(); + t.tx_packets = 100; + t.rx_packets = 200; + network.overlay = Some(overlay(vec![t])); + + let mut out = report::Report::new(); + out.push(network_section(&network, OWN, false)); + let text = out.render(false); + assert_eq!(out.worst(), Health::Good, "{text}"); + assert!(text.contains("tunnel up (tcp-tls)"), "{text}"); + assert!(text.contains("uptime 42s"), "{text}"); + assert!(text.contains("tx 100 rx 200"), "{text}"); + assert!(text.contains("tcp-tls via 192.0.2.1:443"), "{text}"); + } } #[cfg(test)] diff --git a/crates/tsunagi/src/dataplane/tunnel.rs b/crates/tsunagi/src/dataplane/tunnel.rs index 0c85451..6da28ad 100644 --- a/crates/tsunagi/src/dataplane/tunnel.rs +++ b/crates/tsunagi/src/dataplane/tunnel.rs @@ -8,6 +8,7 @@ //! [`GenericTunnelPlugin`] adapts any [`TunnelCodec`] into a full [`super::IpPlugin`]. use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use bytes::Bytes; @@ -78,11 +79,38 @@ pub trait TunnelCodec: Send + Sync + 'static { } } +struct TunnelPeerState { + link: SharedLink, + tx_packets: AtomicU64, + rx_packets: AtomicU64, + dropped: AtomicU64, + established_at: std::time::Instant, +} + +/// Statistics and status for one active peer in [`GenericTunnelPlugin`]. +#[derive(Debug, Clone)] +pub struct TunnelPeerReport { + /// Peer endpoint identity. + pub peer: EndpointId, + /// Protocol identifier. + pub protocol: String, + /// Human-readable path description. + pub path: String, + /// Packets successfully encrypted and sent. + pub tx_packets: u64, + /// Packets successfully received and decrypted. + pub rx_packets: u64, + /// Packets dropped due to encryption/decryption/send failures. + pub dropped: u64, + /// Uptime in seconds since the data link was established. + pub uptime_secs: u64, +} + /// Generic wrapper that turns a [`TunnelCodec`] into a full [`IpPlugin`]. pub struct GenericTunnelPlugin { codec: Arc, context: Mutex>, - links: RwLock>, + links: RwLock>>, tasks: Mutex>>, } @@ -103,7 +131,7 @@ impl GenericTunnelPlugin { } /// Returns the active peers with established data links for a network. - pub fn active_peers(&self, network: NetworkId) -> Vec<(EndpointId, String)> { + pub fn active_peers(&self, network: NetworkId) -> Vec { let guard = match self.links.read() { Ok(g) => g, Err(p) => p.into_inner(), @@ -111,7 +139,15 @@ impl GenericTunnelPlugin { guard .iter() .filter(|((net, _), _)| *net == network) - .map(|((_, peer), link)| (*peer, link.path_description())) + .map(|((_, peer), state)| TunnelPeerReport { + peer: *peer, + protocol: self.codec.protocol_id().to_string(), + path: state.link.path_description(), + tx_packets: state.tx_packets.load(Ordering::Relaxed), + rx_packets: state.rx_packets.load(Ordering::Relaxed), + dropped: state.dropped.load(Ordering::Relaxed), + uptime_secs: state.established_at.elapsed().as_secs(), + }) .collect() } } @@ -171,7 +207,7 @@ impl IpPlugin for GenericTunnelPlugin { } fn carry(&self, network: NetworkId, peer: EndpointId, packet: bytes::Bytes) -> bool { - let link = { + let state = { let guard = match self.links.read() { Ok(g) => g, Err(p) => p.into_inner(), @@ -179,13 +215,22 @@ impl IpPlugin for GenericTunnelPlugin { guard.get(&(network, peer)).cloned() }; - let Some(link) = link else { + let Some(state) = state else { return false; }; match self.codec.encrypt(network, peer, &packet) { - Ok(ciphertext) => link.send(ciphertext).is_ok(), + Ok(ciphertext) => { + if state.link.send(ciphertext).is_ok() { + state.tx_packets.fetch_add(1, Ordering::Relaxed); + true + } else { + state.dropped.fetch_add(1, Ordering::Relaxed); + false + } + } Err(err) => { + state.dropped.fetch_add(1, Ordering::Relaxed); tracing::debug!(%peer, %err, "tunnel encrypt failed"); false } @@ -193,12 +238,19 @@ impl IpPlugin for GenericTunnelPlugin { } fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) { + let state = Arc::new(TunnelPeerState { + link: Arc::clone(&link), + tx_packets: AtomicU64::new(0), + rx_packets: AtomicU64::new(0), + dropped: AtomicU64::new(0), + established_at: std::time::Instant::now(), + }); { let mut guard = match self.links.write() { Ok(g) => g, Err(p) => p.into_inner(), }; - guard.insert((network, peer), Arc::clone(&link)); + guard.insert((network, peer), Arc::clone(&state)); } self.codec.on_peer_link_up(network, peer, &link); @@ -210,15 +262,18 @@ impl IpPlugin for GenericTunnelPlugin { }; let rx_link = Arc::clone(&link); + let rx_state = Arc::clone(&state); let handle = tokio::spawn(async move { while let Some(payload) = rx_link.recv().await { match codec.decrypt(network, peer, &payload) { Ok(packet) => { + rx_state.rx_packets.fetch_add(1, Ordering::Relaxed); if let Some(ref sink) = sink { sink.deliver(network, peer, packet).await; } } Err(err) => { + rx_state.dropped.fetch_add(1, Ordering::Relaxed); tracing::debug!(%peer, %err, "tunnel decrypt failed"); } } diff --git a/crates/tsunagi/src/ipc/mod.rs b/crates/tsunagi/src/ipc/mod.rs index 9706e7d..4689927 100644 --- a/crates/tsunagi/src/ipc/mod.rs +++ b/crates/tsunagi/src/ipc/mod.rs @@ -394,7 +394,9 @@ pub struct OverlayPeerReport { /// The peer's control plane identity, so a tunnel can be matched to the /// session and the member it belongs to. pub endpoint_id: String, - /// The peer's WireGuard public key. + /// Protocol carrying overlay traffic for this peer (e.g. "tcp-tls", "wg-quic"). + pub protocol: String, + /// The peer's WireGuard public key (if applicable). pub public_key: String, /// The overlay address the network agreed it holds. pub address: Option, diff --git a/crates/tsunagi/src/overlay/interface.rs b/crates/tsunagi/src/overlay/interface.rs index 921868a..5b8ce02 100644 --- a/crates/tsunagi/src/overlay/interface.rs +++ b/crates/tsunagi/src/overlay/interface.rs @@ -288,15 +288,16 @@ impl Interface { pub async fn remove(&self) { // The packet loop holds the device open, and it ends only when the // device reports end of stream — which a real interface never does - // while it exists. So it is stopped here, not left to notice. + // while it exists. Stop it directly before destroying the interface. let task = match self.task.lock() { Ok(mut guard) => guard.take(), Err(poisoned) => poisoned.into_inner().take(), }; - self.factory.destroy(self.device.name()).await; if let Some(task) = task { - crate::task::wind_down(task, crate::task::TASK_GRACE, "overlay packet loop").await; + task.abort(); + let _ = task.await; } + self.factory.destroy(self.device.name()).await; } /// The counters as they stand.