From a8d3627072df579a077164f674c509b907ad7bbd Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 19 Feb 2026 19:42:55 +0000 Subject: [PATCH] Add link-layer heartbeat and liveness timeout for dead peer detection Sends a 1-byte encrypted heartbeat (0x51) to each peer every 10s. If no frame is received from a peer within 30s, the peer is removed via remove_active_peer(), triggering tree reconvergence, coord cache flush, and bloom filter recomputation. This fixes the critical bug where UDP peers that silently died (e.g., container stopped) were never detected or removed, leaving the spanning tree permanently stale. Both intervals are configurable via node.heartbeat_interval_secs and node.link_dead_timeout_secs. --- src/config/node.rs | 13 +++++++ src/mmp/receiver.rs | 4 +++ src/node/handlers/dispatch.rs | 6 +++- src/node/handlers/mmp.rs | 66 +++++++++++++++++++++++++++++++++-- src/node/handlers/rx_loop.rs | 1 + src/peer/active.rs | 18 ++++++++++ src/protocol/link.rs | 6 ++++ 7 files changed, 110 insertions(+), 4 deletions(-) diff --git a/src/config/node.rs b/src/config/node.rs index d9036a9..62e8add 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -367,6 +367,15 @@ pub struct NodeConfig { #[serde(default = "NodeConfig::default_base_rtt_ms")] pub base_rtt_ms: u64, + /// Link heartbeat send interval in seconds (`node.heartbeat_interval_secs`). + #[serde(default = "NodeConfig::default_heartbeat_interval_secs")] + pub heartbeat_interval_secs: u64, + + /// Link dead timeout in seconds (`node.link_dead_timeout_secs`). + /// Peers silent for this duration are removed. + #[serde(default = "NodeConfig::default_link_dead_timeout_secs")] + pub link_dead_timeout_secs: u64, + /// Resource limits (`node.limits.*`). #[serde(default)] pub limits: LimitsConfig, @@ -419,6 +428,8 @@ impl Default for NodeConfig { leaf_only: false, tick_interval_secs: 1, base_rtt_ms: 100, + heartbeat_interval_secs: 10, + link_dead_timeout_secs: 30, limits: LimitsConfig::default(), rate_limit: RateLimitConfig::default(), retry: RetryConfig::default(), @@ -437,4 +448,6 @@ impl Default for NodeConfig { impl NodeConfig { fn default_tick_interval_secs() -> u64 { 1 } fn default_base_rtt_ms() -> u64 { 100 } + fn default_heartbeat_interval_secs() -> u64 { 10 } + fn default_link_dead_timeout_secs() -> u64 { 30 } } diff --git a/src/mmp/receiver.rs b/src/mmp/receiver.rs index e0873f0..230d83d 100644 --- a/src/mmp/receiver.rs +++ b/src/mmp/receiver.rs @@ -356,6 +356,10 @@ impl ReceiverState { pub fn report_interval(&self) -> Duration { self.report_interval } + + pub fn last_recv_time(&self) -> Option { + self.last_recv_time + } } impl Default for ReceiverState { diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 47b80ca..7342e12 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -2,7 +2,7 @@ use crate::node::Node; use crate::NodeAddr; -use tracing::{debug, info}; +use tracing::{debug, info, trace}; impl Node { /// Dispatch a decrypted link message to the appropriate handler. @@ -49,6 +49,10 @@ impl Node { // Disconnect self.handle_disconnect(from, payload); } + 0x51 => { + // Heartbeat — no-op, last_recv_time already updated by record_recv() + trace!(peer = %self.peer_display_name(from), "Received heartbeat"); + } _ => { debug!(msg_type = msg_type, "Unknown link message type"); } diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 8076ff3..171582f 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -9,11 +9,12 @@ use crate::mmp::MmpSessionState; use crate::mmp::report::{ReceiverReport, SenderReport}; use crate::node::Node; use crate::protocol::{ - PathMtuNotification, SessionMessageType, SessionReceiverReport, SessionSenderReport, + LinkMessageType, PathMtuNotification, SessionMessageType, SessionReceiverReport, + SessionSenderReport, }; use crate::NodeAddr; -use std::time::Instant; -use tracing::{debug, info, trace}; +use std::time::{Duration, Instant}; +use tracing::{debug, info, trace, warn}; /// Format bytes/sec as human-readable throughput. fn format_throughput(bps: f64) -> String { @@ -381,4 +382,63 @@ impl Node { "MMP session teardown" ); } + + /// Send heartbeats and remove dead peers. + /// + /// Called from the tick handler. Sends a 1-byte heartbeat to each peer + /// whose heartbeat interval has elapsed, and removes any peer that + /// hasn't sent us a frame within the link dead timeout. + pub(in crate::node) async fn check_link_heartbeats(&mut self) { + let now = Instant::now(); + let heartbeat_interval = Duration::from_secs(self.config.node.heartbeat_interval_secs); + let dead_timeout = Duration::from_secs(self.config.node.link_dead_timeout_secs); + let heartbeat_msg = [LinkMessageType::Heartbeat.to_byte()]; + + // Collect heartbeats to send and dead peers to remove + let mut heartbeats: Vec = Vec::new(); + let mut dead_peers: Vec = Vec::new(); + + for (node_addr, peer) in self.peers.iter() { + // Check liveness via MMP receiver last_recv_time + if let Some(mmp) = peer.mmp() + && let Some(last_recv) = mmp.receiver.last_recv_time() + && now.duration_since(last_recv) >= dead_timeout + { + dead_peers.push(*node_addr); + continue; + } + + // Check if heartbeat is due + let needs_heartbeat = match peer.last_heartbeat_sent() { + None => true, + Some(last) => now.duration_since(last) >= heartbeat_interval, + }; + if needs_heartbeat { + heartbeats.push(*node_addr); + } + } + + // Remove dead peers + for addr in &dead_peers { + warn!( + peer = %self.peer_display_name(addr), + timeout_secs = self.config.node.link_dead_timeout_secs, + "Removing peer: link dead timeout" + ); + self.remove_active_peer(addr); + } + + // Send heartbeats (skip peers we just removed) + for addr in heartbeats { + if dead_peers.contains(&addr) { + continue; + } + if let Some(peer) = self.peers.get_mut(&addr) { + peer.mark_heartbeat_sent(now); + } + if let Err(e) = self.send_encrypted_link_message(&addr, &heartbeat_msg).await { + trace!(peer = %self.peer_display_name(&addr), error = %e, "Failed to send heartbeat"); + } + } + } } diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index a88de14..432a780 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -87,6 +87,7 @@ impl Node { self.check_bloom_state().await; self.check_mmp_reports().await; self.check_session_mmp_reports().await; + self.check_link_heartbeats().await; self.purge_stale_lookups(now_ms); } } diff --git a/src/peer/active.rs b/src/peer/active.rs index f73637c..6bf73d4 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -131,6 +131,10 @@ pub struct ActivePeer { /// Per-peer MMP state (None for legacy peers without Noise sessions). mmp: Option, + // === Heartbeat === + /// When we last sent a heartbeat to this peer. + last_heartbeat_sent: Option, + // === Handshake Resend === /// Wire-format msg2 for resend on duplicate msg1 (responder only). /// Cleared after the handshake timeout window. @@ -166,6 +170,7 @@ impl ActivePeer { authenticated_at, last_seen: authenticated_at, mmp: None, + last_heartbeat_sent: None, handshake_msg2: None, } } @@ -226,6 +231,7 @@ impl ActivePeer { authenticated_at, last_seen: authenticated_at, mmp: Some(MmpPeerState::new(mmp_config, is_initiator)), + last_heartbeat_sent: None, handshake_msg2: None, } } @@ -485,6 +491,18 @@ impl ActivePeer { self.session_start.elapsed().as_millis() as u32 } + // === Heartbeat === + + /// When we last sent a heartbeat to this peer. + pub fn last_heartbeat_sent(&self) -> Option { + self.last_heartbeat_sent + } + + /// Record that we sent a heartbeat. + pub fn mark_heartbeat_sent(&mut self, now: Instant) { + self.last_heartbeat_sent = Some(now); + } + // === State Updates === /// Update last seen timestamp. diff --git a/src/protocol/link.rs b/src/protocol/link.rs index 49fd09c..dc77126 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -96,6 +96,9 @@ pub enum LinkMessageType { // Link Control (0x50-0x5F) /// Orderly disconnect notification before link closure. Disconnect = 0x50, + /// Periodic heartbeat for link liveness detection. + /// No payload — the msg_type byte alone is sufficient. + Heartbeat = 0x51, } impl LinkMessageType { @@ -110,6 +113,7 @@ impl LinkMessageType { 0x30 => Some(LinkMessageType::LookupRequest), 0x31 => Some(LinkMessageType::LookupResponse), 0x50 => Some(LinkMessageType::Disconnect), + 0x51 => Some(LinkMessageType::Heartbeat), _ => None, } } @@ -131,6 +135,7 @@ impl fmt::Display for LinkMessageType { LinkMessageType::LookupRequest => "LookupRequest", LinkMessageType::LookupResponse => "LookupResponse", LinkMessageType::Disconnect => "Disconnect", + LinkMessageType::Heartbeat => "Heartbeat", }; write!(f, "{}", name) } @@ -423,6 +428,7 @@ mod tests { LinkMessageType::LookupResponse, LinkMessageType::SessionDatagram, LinkMessageType::Disconnect, + LinkMessageType::Heartbeat, ]; for ty in types {