fix(p2p): improve validator heartbeat diagnostics and stale threshold

- Add WARN-level logs for heartbeat message rejections (was DEBUG)
- Add heartbeat_rejections_total counter metric for monitoring
- Make stale threshold configurable via VALIDATOR_STALE_THRESHOLD_MS env var
- Change default stale threshold from 90s to 300s (5 minutes)
- Add unit test for configurable threshold

Root cause: Validators were marked inactive after 90s without visible logs
showing why heartbeats were rejected. Now operators can diagnose P2P issues
and configure appropriate thresholds for their network conditions.

Fixes issue where all validators became inactive simultaneously due to
network hiccups or message validation failures going unnoticed.

Author: echobt

crates/p2p-consensus/src/network.rs

@@ -725,6 +725,8 @@ pub struct P2PNetwork {
     seen_nonces: RwLock<HashMap<Hotkey, HashMap<u64, i64>>>,
     /// Message timestamps for sliding window rate limiting (hotkey -> recent message timestamps in ms)
     message_timestamps: RwLock<HashMap<Hotkey, VecDeque<i64>>>,
+    /// Counter for tracking heartbeat message rejections
+    heartbeat_rejections_total: AtomicU64,
 }
 
 impl P2PNetwork {
@@ -763,6 +765,7 @@ impl P2PNetwork {
             ),
             seen_nonces: RwLock::new(HashMap::new()),
             message_timestamps: RwLock::new(HashMap::new()),
+            heartbeat_rejections_total: AtomicU64::new(0),
         })
     }
 
@@ -797,6 +800,15 @@ impl P2PNetwork {
         self.connected_peer_count() >= min_required
     }
 
+
+    /// Get the total count of heartbeat message rejections
+    ///
+    /// This metric tracks how many heartbeat messages have been rejected
+    /// due to validation failures, rate limiting, or other errors.
+    pub fn heartbeat_rejections_total(&self) -> u64 {
+        self.heartbeat_rejections_total.load(Ordering::Relaxed)
+    }
+
     /// Create gossipsub behaviour
     fn create_gossipsub(
         &self,
@@ -1481,17 +1493,35 @@ impl P2PNetwork {
                     }
                 }
                 Err(e) => {
-                    // Try to extract hotkey from message for debugging
-                    let hotkey = bincode::deserialize::<SignedP2PMessage>(&message.data)
-                        .ok()
+                    // Try to extract hotkey from message and check type
+                    let deserialized = bincode::deserialize::<SignedP2PMessage>(&message.data).ok();
+                    let hotkey = deserialized
+                        .as_ref()
                         .map(|m| m.signer.to_ss58())
                         .unwrap_or_else(|| "unknown".to_string());
-                    debug!(
-                        source = %propagation_source,
-                        hotkey = %hotkey,
-                        error = %e,
-                        "Failed to process gossipsub message"
-                    );
+                    
+                    // Check if this is a heartbeat message for specialized warning
+                    let is_heartbeat = deserialized
+                        .as_ref()
+                        .map(|m| matches!(m.message, P2PMessage::Heartbeat(_)))
+                        .unwrap_or(false);
+                    
+                    if is_heartbeat {
+                        warn!(
+                            source = %propagation_source,
+                            hotkey = %hotkey,
+                            error = %e,
+                            "Heartbeat message rejected - validator may become inactive"
+                        );
+                        self.heartbeat_rejections_total.fetch_add(1, Ordering::Relaxed);
+                    } else {
+                        debug!(
+                            source = %propagation_source,
+                            hotkey = %hotkey,
+                            error = %e,
+                            "Failed to process gossipsub message"
+                        );
+                    }
                 }
             },
             SwarmEvent::Behaviour(CombinedEvent::Gossipsub(gossipsub::Event::Subscribed {

crates/p2p-consensus/src/validator.rs

@@ -107,11 +107,22 @@ pub struct ValidatorSet {
 impl ValidatorSet {
     /// Create a new validator set
     pub fn new(local_keypair: Keypair, min_stake: u64) -> Self {
+        // Read stale threshold from environment, default to 5 minutes
+        let stale_threshold_ms = std::env::var("VALIDATOR_STALE_THRESHOLD_MS")
+            .ok()
+            .and_then(|s| s.parse::<i64>().ok())
+            .unwrap_or(300_000); // Default: 5 minutes
+
+        info!(
+            threshold_ms = stale_threshold_ms,
+            "Validator stale threshold initialized (configurable via VALIDATOR_STALE_THRESHOLD_MS)"
+        );
+
         Self {
             validators: RwLock::new(HashMap::new()),
             min_stake,
             local_keypair,
-            stale_threshold_ms: 90_000, // 90 seconds (3x heartbeat interval)
+            stale_threshold_ms,
             verified_stakes: RwLock::new(HashMap::new()),
         }
     }
@@ -624,4 +635,26 @@ mod tests {
         assert_eq!(updated.last_sequence, 100);
         assert_eq!(updated.stake, 15_000);
     }
+
+    #[test]
+    fn test_stale_threshold_from_env() {
+        // Test env var override
+        std::env::set_var("VALIDATOR_STALE_THRESHOLD_MS", "600000");
+        let keypair = create_test_keypair();
+        let set = ValidatorSet::new(keypair, 1000);
+        assert_eq!(set.stale_threshold_ms, 600_000);
+        std::env::remove_var("VALIDATOR_STALE_THRESHOLD_MS");
+
+        // Test default
+        let keypair2 = create_test_keypair();
+        let set2 = ValidatorSet::new(keypair2, 1000);
+        assert_eq!(set2.stale_threshold_ms, 300_000);
+
+        // Test invalid env var falls back to default
+        std::env::set_var("VALIDATOR_STALE_THRESHOLD_MS", "invalid");
+        let keypair3 = create_test_keypair();
+        let set3 = ValidatorSet::new(keypair3, 1000);
+        assert_eq!(set3.stale_threshold_ms, 300_000);
+        std::env::remove_var("VALIDATOR_STALE_THRESHOLD_MS");
+    }
 }