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 {