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

@@ -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 }
 }

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<Instant> {
+        self.last_recv_time
+    }
 }
 
 impl Default for ReceiverState {

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");
             }

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<NodeAddr> = Vec::new();
+        let mut dead_peers: Vec<NodeAddr> = 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");
+            }
+        }
+    }
 }

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);
                 }
             }

src/peer/active.rs

@@ -131,6 +131,10 @@ pub struct ActivePeer {
     /// Per-peer MMP state (None for legacy peers without Noise sessions).
     mmp: Option<MmpPeerState>,
 
+    // === Heartbeat ===
+    /// When we last sent a heartbeat to this peer.
+    last_heartbeat_sent: Option<Instant>,
+
     // === 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<Instant> {
+        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.

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 {