Introduce QUIC connection lifecycle tracing and dead connection pruning (#50)

* Introduce verbose QUIC connection lifecycle tracing

Log close_reason when detecting dead connections on client side
and when server-side stream accept loop terminates. Log when
server replaces a validator connection during re-handshake.

* Introduce dead connection pruning and synapse-level query tracing

Investigate systematic QUIC synapse query failures by adding
diagnostic tracing to send_synapse_packet and server dispatch
paths. Resolve stale connections persisting in the registry by
pruning dead connections during update_miner_registry, and
report actual connection health status in get_connection_stats.

* Resolve all clippy warnings and replace fixed sleep with bounded poll

Eliminate field_reassign_with_default in client config tests using struct
update syntax. Replace unnecessary_get_then_check in registry with
contains_key. Reorder signing.rs to place feature-gated items before test
module. Replace unconditional 5s sleep in integration test with bounded
poll loop that retries at 250ms intervals with a 10s deadline.

Author: HudsonGraeme

crates/btlightning/src/client.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
 use tokio::time::Instant;
-use tracing::{error, info, instrument, warn};
+use tracing::{debug, error, info, instrument, warn};
 
 #[cfg(feature = "subtensor")]
 use crate::metagraph::{Metagraph, MetagraphMonitorConfig};
@@ -556,9 +556,26 @@ impl LightningClient {
         let max_fp = self.config.max_frame_payload_bytes;
         match connection {
             Some(conn) if conn.close_reason().is_none() => {
+                debug!(
+                    addr = %addr_key,
+                    stable_id = conn.stable_id(),
+                    "query_axon: connection alive, sending synapse"
+                );
                 send_synapse_packet(&conn, request, max_fp).await
             }
-            _ => {
+            Some(conn) => {
+                let reason = conn.close_reason();
+                warn!(
+                    addr = %addr_key,
+                    stable_id = conn.stable_id(),
+                    close_reason = ?reason,
+                    "QUIC connection closed, triggering reconnect"
+                );
+                self.try_reconnect_and_query(&addr_key, &axon_info, request)
+                    .await
+            }
+            None => {
+                debug!(addr = %addr_key, "query_axon: no connection in registry");
                 self.try_reconnect_and_query(&addr_key, &axon_info, request)
                     .await
             }
@@ -605,7 +622,17 @@ impl LightningClient {
                 )
                 .await
             }
-            _ => {
+            Some(conn) => {
+                let reason = conn.close_reason();
+                warn!(
+                    addr = %addr_key,
+                    close_reason = ?reason,
+                    "QUIC connection closed, triggering reconnect (stream)"
+                );
+                self.try_reconnect_and_stream(&addr_key, &axon_info, request)
+                    .await
+            }
+            None => {
                 self.try_reconnect_and_stream(&addr_key, &axon_info, request)
                     .await
             }
@@ -843,7 +870,17 @@ impl LightningClient {
         );
 
         for addr_key in state.registry.connection_addrs() {
-            stats.insert(format!("connection_{}", addr_key), "active".to_string());
+            let status = match state.registry.get_connection(addr_key) {
+                Some(conn) => {
+                    if let Some(reason) = conn.close_reason() {
+                        format!("closed({:?})", reason)
+                    } else {
+                        "active".to_string()
+                    }
+                }
+                None => "missing".to_string(),
+            };
+            stats.insert(format!("connection_{}", addr_key), status);
         }
 
         Ok(stats)
@@ -1078,6 +1115,30 @@ async fn update_miner_registry_inner(
             }
         }
 
+        let dead_addrs: Vec<PeerAddr> = active_addrs
+            .iter()
+            .filter(|addr| {
+                st.registry
+                    .get_connection(addr)
+                    .is_some_and(|c| c.close_reason().is_some())
+            })
+            .cloned()
+            .collect();
+        for addr_key in &dead_addrs {
+            if let Some(conn) = st.registry.remove_connection(addr_key) {
+                let hotkeys = st.registry.hotkeys_at_addr(addr_key);
+                info!(
+                    addr = %addr_key,
+                    close_reason = ?conn.close_reason(),
+                    hotkeys = ?hotkeys,
+                    "Pruning dead connection and deregistering miners"
+                );
+                for hk in &hotkeys {
+                    st.registry.deregister(hk);
+                }
+            }
+        }
+
         for new_miner in new_by_hotkey.values() {
             if let Some(old_miner) = st.registry.active_miner(&new_miner.hotkey) {
                 let old_addr = old_miner.addr_key();
@@ -1500,16 +1561,24 @@ async fn send_synapse_packet(
     request: QuicRequest,
     max_frame_payload: usize,
 ) -> Result<QuicResponse> {
+    let stable_id = connection.stable_id();
+    debug!(stable_id, "send_synapse_packet: opening bi stream");
     let (mut send, mut recv) = connection
         .open_bi()
         .await
         .map_err(|e| LightningError::Connection(format!("Failed to open stream: {}", e)))?;
+    debug!(stable_id, "send_synapse_packet: bi stream opened");
 
     let start = Instant::now();
 
     send_synapse_frame(&mut send, request).await?;
+    debug!(
+        stable_id,
+        "send_synapse_packet: frame sent, awaiting response"
+    );
 
     let (msg_type, payload) = read_frame(&mut recv, max_frame_payload).await?;
+    debug!(stable_id, msg_type = ?msg_type, elapsed_ms = start.elapsed().as_millis() as u64, "send_synapse_packet: response received");
 
     match msg_type {
         MessageType::SynapseResponse => {
@@ -1685,8 +1754,10 @@ mod tests {
 
     #[test]
     fn with_config_rejects_frame_payload_below_minimum() {
-        let mut cfg = LightningClientConfig::default();
-        cfg.max_frame_payload_bytes = 512;
+        let cfg = LightningClientConfig {
+            max_frame_payload_bytes: 512,
+            ..LightningClientConfig::default()
+        };
         assert!(LightningClient::with_config("hk".into(), cfg).is_err());
     }
 
@@ -1697,23 +1768,30 @@ mod tests {
             Ok(v) => v,
             Err(_) => return,
         };
-        let mut cfg = LightningClientConfig::default();
-        cfg.max_frame_payload_bytes = val;
-        cfg.max_stream_payload_bytes = val;
+        let cfg = LightningClientConfig {
+            max_frame_payload_bytes: val,
+            max_stream_payload_bytes: val,
+            ..LightningClientConfig::default()
+        };
         assert!(LightningClient::with_config("hk".into(), cfg).is_err());
     }
 
     #[test]
     fn with_config_rejects_stream_below_frame() {
-        let mut cfg = LightningClientConfig::default();
-        cfg.max_stream_payload_bytes = cfg.max_frame_payload_bytes - 1;
+        let base = LightningClientConfig::default();
+        let cfg = LightningClientConfig {
+            max_stream_payload_bytes: base.max_frame_payload_bytes - 1,
+            ..base
+        };
         assert!(LightningClient::with_config("hk".into(), cfg).is_err());
     }
 
     #[test]
     fn with_config_rejects_zero_stream_chunk_timeout() {
-        let mut cfg = LightningClientConfig::default();
-        cfg.stream_chunk_timeout = Some(Duration::ZERO);
+        let cfg = LightningClientConfig {
+            stream_chunk_timeout: Some(Duration::ZERO),
+            ..LightningClientConfig::default()
+        };
         assert!(LightningClient::with_config("hk".into(), cfg).is_err());
     }
 

crates/btlightning/src/registry.rs

@@ -329,10 +329,10 @@ mod tests {
         reg.register(QuicAxonInfo::new("hk1".into(), "1.2.3.4".into(), 8080, 4));
         let rs = reg.reconnect_state_or_insert(old_addr.clone());
         rs.attempts = 3;
-        assert!(reg.reconnect_states.get(&old_addr).is_some());
+        assert!(reg.reconnect_states.contains_key(&old_addr));
 
         reg.register(QuicAxonInfo::new("hk1".into(), "5.6.7.8".into(), 9090, 4));
-        assert!(reg.reconnect_states.get(&old_addr).is_none());
+        assert!(!reg.reconnect_states.contains_key(&old_addr));
         reg.assert_invariants();
     }
 

crates/btlightning/src/server/dispatch.rs

@@ -8,14 +8,18 @@ use crate::util::unix_timestamp_secs;
 use quinn::{Connection, RecvStream, SendStream};
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info, warn};
 
 pub(super) async fn handle_connection(connection: Connection, ctx: ServerContext) {
     let connection = Arc::new(connection);
+    let stable_id = connection.stable_id();
+    let remote = connection.remote_address();
+    debug!(stable_id, %remote, "handle_connection: entering accept_bi loop");
 
     loop {
         match connection.accept_bi().await {
             Ok((send, recv)) => {
+                debug!(stable_id, "handle_connection: accepted bi stream");
                 let conn = connection.clone();
                 let ctx = ctx.clone();
 
@@ -24,7 +28,13 @@ pub(super) async fn handle_connection(connection: Connection, ctx: ServerContext
                 });
             }
             Err(e) => {
-                debug!("Connection ended: {}", e);
+                let close_reason = connection.close_reason();
+                info!(
+                    remote = %connection.remote_address(),
+                    error = %e,
+                    close_reason = ?close_reason,
+                    "QUIC connection stream loop ended"
+                );
                 break;
             }
         }
@@ -74,6 +84,7 @@ async fn handle_stream(
         }
     };
 
+    debug!(msg_type = ?frame.0, payload_len = frame.1.len(), "handle_stream: frame received");
     match frame {
         (MessageType::SynapsePacket, payload) => {
             let packet: SynapsePacket = match rmp_serde::from_slice(&payload) {

crates/btlightning/src/server/handshake.rs

@@ -111,6 +111,12 @@ pub(super) async fn process_handshake(
         connections_guard.insert(request.validator_hotkey.clone(), validator_conn)
     {
         if !Arc::ptr_eq(&prev_conn.connection, &connection) {
+            warn!(
+                validator = %request.validator_hotkey,
+                prev_addr = %prev_conn.connection.remote_address(),
+                new_addr = %remote_addr,
+                "closing previous connection for validator (replaced by new handshake)"
+            );
             prev_conn.connection.close(0u32.into(), b"replaced");
             let prev_addr = prev_conn.connection.remote_address();
             if prev_addr != remote_addr {

crates/btlightning/src/signing.rs

@@ -86,6 +86,47 @@ impl<F: Fn(&[u8]) -> Result<Vec<u8>> + Send + Sync> Signer for CallbackSigner<F>
     }
 }
 
+/// [`Signer`] backed by a `btwallet` keypair loaded from the Bittensor wallet directory.
+///
+/// Requires the `btwallet` feature.
+#[cfg(feature = "btwallet")]
+pub struct BtWalletSigner {
+    keypair: bittensor_wallet::Keypair,
+}
+
+#[cfg(feature = "btwallet")]
+impl BtWalletSigner {
+    /// Wraps an existing `btwallet::Keypair`.
+    pub fn new(keypair: bittensor_wallet::Keypair) -> Self {
+        Self { keypair }
+    }
+
+    /// Loads the hotkey keypair from `~/<path>/<name>/hotkeys/<hotkey_name>`.
+    pub fn from_wallet(name: &str, path: &str, hotkey_name: &str) -> Result<Self> {
+        let wallet = bittensor_wallet::Wallet::new(
+            Some(name.to_string()),
+            Some(path.to_string()),
+            Some(hotkey_name.to_string()),
+            None,
+        );
+        let keypair = wallet
+            .get_hotkey(Some(hotkey_name.to_string()))
+            .map_err(|e| {
+                LightningError::Config(format!("failed to load hotkey from wallet: {}", e))
+            })?;
+        Ok(Self { keypair })
+    }
+}
+
+#[cfg(feature = "btwallet")]
+impl Signer for BtWalletSigner {
+    fn sign(&self, message: &[u8]) -> Result<Vec<u8>> {
+        self.keypair
+            .sign(message.to_vec())
+            .map_err(LightningError::Signing)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -157,44 +198,3 @@ mod tests {
         assert!(err.to_string().contains("network down"));
     }
 }
-
-/// [`Signer`] backed by a `btwallet` keypair loaded from the Bittensor wallet directory.
-///
-/// Requires the `btwallet` feature.
-#[cfg(feature = "btwallet")]
-pub struct BtWalletSigner {
-    keypair: bittensor_wallet::Keypair,
-}
-
-#[cfg(feature = "btwallet")]
-impl BtWalletSigner {
-    /// Wraps an existing `btwallet::Keypair`.
-    pub fn new(keypair: bittensor_wallet::Keypair) -> Self {
-        Self { keypair }
-    }
-
-    /// Loads the hotkey keypair from `~/<path>/<name>/hotkeys/<hotkey_name>`.
-    pub fn from_wallet(name: &str, path: &str, hotkey_name: &str) -> Result<Self> {
-        let wallet = bittensor_wallet::Wallet::new(
-            Some(name.to_string()),
-            Some(path.to_string()),
-            Some(hotkey_name.to_string()),
-            None,
-        );
-        let keypair = wallet
-            .get_hotkey(Some(hotkey_name.to_string()))
-            .map_err(|e| {
-                LightningError::Config(format!("failed to load hotkey from wallet: {}", e))
-            })?;
-        Ok(Self { keypair })
-    }
-}
-
-#[cfg(feature = "btwallet")]
-impl Signer for BtWalletSigner {
-    fn sign(&self, message: &[u8]) -> Result<Vec<u8>> {
-        self.keypair
-            .sign(message.to_vec())
-            .map_err(LightningError::Signing)
-    }
-}