ab/tsunagi
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Report peer tunnel protocols accurately and eliminate overlay shutdown timeout

Browse Source 
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.
This commit is contained in:
Aleksandr Bogomiakov committed 2026-09-30 17:31:17 +01:00
1 parent 41990c9934
commit a4e341cd66
4 files changed
+202 -69
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/tsunagi-cli/src/main.rs
+133 -58
View File
@@ -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)]
crates/tsunagi/src/dataplane/tunnel.rs
+62 -7
View File
@@ -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<C: TunnelCodec> {
codec: Arc<C>,
context: Mutex<Option<PluginContext>>,
links: RwLock<HashMap<(NetworkId, EndpointId), SharedLink>>,
links: RwLock<HashMap<(NetworkId, EndpointId), Arc<TunnelPeerState>>>,
tasks: Mutex<HashMap<(NetworkId, EndpointId), JoinHandle<()>>>,
}
@@ -103,7 +131,7 @@ impl<C: TunnelCodec> GenericTunnelPlugin<C> {
}
/// 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<TunnelPeerReport> {
let guard = match self.links.read() {
Ok(g) => g,
Err(p) => p.into_inner(),
@@ -111,7 +139,15 @@ impl<C: TunnelCodec> GenericTunnelPlugin<C> {
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<C: TunnelCodec> IpPlugin for GenericTunnelPlugin<C> {
}
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<C: TunnelCodec> IpPlugin for GenericTunnelPlugin<C> {
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<C: TunnelCodec> IpPlugin for GenericTunnelPlugin<C> {
}
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<C: TunnelCodec> IpPlugin for GenericTunnelPlugin<C> {
};
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");
}
}
crates/tsunagi/src/ipc/mod.rs
+3 -1
View File
@@ -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<String>,
crates/tsunagi/src/overlay/interface.rs
+4 -3
View File
@@ -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.