refactor(transport): eliminate latency channel — update registry directly

The Pong handler in the control loop now calls
registry.update_latency() directly instead of forwarding RTT
measurements through a dedicated mpsc channel. This removes 4 channel
pair creations (one per transport impl), the latency_rx field from
ConnectResult/AcceptResult, and the latency_rx parameter from
spawn_heartbeat.

The peer name for registry lookups is extracted from the Handshake
message — on the server side at construction, on the client side when
the first Handshake arrives through the control stream. QuicClient and
WsClient gain a peer_registry field that daemon modes set before
calling connect().

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: maxholman

crates/core/src/client/client.rs

@@ -44,8 +44,6 @@ pub struct ConnectResult<T: wallhack_transport::Transport + ?Sized> {
     control_tx: mpsc::Sender<ControlMessage>,
     /// Receiver for the server's `Handshake` (delivered via the control loop).
     peer_handshake_rx: Option<oneshot::Receiver<Handshake>>,
-    /// Pong-derived latency measurements from the control loop (milliseconds).
-    latency_rx: Option<mpsc::Receiver<f64>>,
 }
 
 impl<T: wallhack_transport::Transport + ?Sized> ConnectResult<T> {
@@ -57,7 +55,6 @@ impl<T: wallhack_transport::Transport + ?Sized> ConnectResult<T> {
         tasks: ConnectionTasks,
         control_tx: mpsc::Sender<ControlMessage>,
         peer_handshake_rx: Option<oneshot::Receiver<Handshake>>,
-        latency_rx: Option<mpsc::Receiver<f64>>,
     ) -> Self {
         Self {
             channels,
@@ -66,7 +63,6 @@ impl<T: wallhack_transport::Transport + ?Sized> ConnectResult<T> {
             transport,
             control_tx,
             peer_handshake_rx,
-            latency_rx,
         }
     }
 
@@ -110,8 +106,6 @@ pub struct ErasedConnectResult {
     pub control_tx: mpsc::Sender<ControlMessage>,
     pub peer_handshake_rx: Option<oneshot::Receiver<Handshake>>,
     pub peer_addr: String,
-    /// Pong-derived latency measurements from the control loop (milliseconds).
-    pub latency_rx: Option<mpsc::Receiver<f64>>,
 }
 
 impl<T> ConnectResult<T>
@@ -135,7 +129,6 @@ where
             channels: self.channels,
             tasks: self.tasks,
             control_tx: self.control_tx,
-            latency_rx: self.latency_rx,
         }
     }
 }

crates/core/src/client/quic/mod.rs

@@ -64,6 +64,9 @@ pub struct QuicClient {
     name: Option<String>,
     psk: Option<zeroize::Zeroizing<String>>,
     local_handshake: Option<Handshake>,
+    /// Peer registry for direct latency updates in the control loop.
+    /// Set by the daemon mode before calling `connect()`.
+    pub peer_registry: Option<std::sync::Arc<crate::control::peers::Registry>>,
 }
 
 impl Client for QuicClient {
@@ -101,6 +104,7 @@ impl Client for QuicClient {
             name: config.name,
             psk: config.psk,
             local_handshake: config.local_handshake,
+            peer_registry: None,
         })
     }
 
@@ -170,18 +174,17 @@ impl Client for QuicClient {
 
         // Create oneshot for receiving server's Handshake via the control loop.
         let (handshake_tx, handshake_rx) = tokio::sync::oneshot::channel::<Handshake>();
-        let (latency_tx, latency_rx) = tokio::sync::mpsc::channel::<f64>(4);
-
         // Spawn control stream task
         let control_handle = {
             let transport = Arc::clone(&transport);
+            let peer_registry = self.peer_registry.clone();
             tokio::spawn(async move {
                 let mut channels = protocol::ControlChannels {
                     outgoing_rx: control_rx,
                     handshake_tx: Some(handshake_tx),
-                    latency_tx: Some(latency_tx),
                     control_response_tx: None,
-                    peer_registry: None,
+                    peer_registry,
+                    peer_name: None,
                 };
                 match protocol::run_control_stream_initiator(
                     &*transport,
@@ -235,7 +238,6 @@ impl Client for QuicClient {
             tasks,
             control_tx,
             Some(handshake_rx),
-            Some(latency_rx),
         ))
     }
 

crates/core/src/client/ws/mod.rs

@@ -196,6 +196,9 @@ impl Default for WsClientConfig {
 pub struct WsClient {
     config: WsClientConfig,
     tls_connector: Option<TlsConnector>,
+    /// Peer registry for direct latency updates in the control loop.
+    /// Set by the daemon mode before calling `connect()`.
+    pub peer_registry: Option<std::sync::Arc<crate::control::peers::Registry>>,
 }
 
 impl WsClient {
@@ -223,6 +226,7 @@ impl WsClient {
         Ok(Self {
             config,
             tls_connector,
+            peer_registry: None,
         })
     }
 
@@ -387,18 +391,17 @@ impl WsClient {
 
         // Create oneshot for receiving server's Handshake via the control loop.
         let (handshake_tx, handshake_rx) = tokio::sync::oneshot::channel::<Handshake>();
-        let (latency_tx, latency_rx) = tokio::sync::mpsc::channel::<f64>(4);
-
         // Spawn control stream task
         let control_handle = {
             let transport = Arc::clone(&transport);
+            let peer_registry = self.peer_registry.clone();
             tokio::spawn(async move {
                 let mut channels = protocol::ControlChannels {
                     outgoing_rx: control_rx,
                     handshake_tx: Some(handshake_tx), // receive server's Handshake
-                    latency_tx: Some(latency_tx),
                     control_response_tx: None,
-                    peer_registry: None,
+                    peer_registry,
+                    peer_name: None,
                 };
                 match protocol::run_control_stream_initiator(
                     &*transport,
@@ -452,7 +455,6 @@ impl WsClient {
             tasks,
             control_tx,
             Some(handshake_rx),
-            Some(latency_rx),
         ))
     }
 }

crates/core/src/server/quic/mod.rs

@@ -210,9 +210,7 @@ impl Server for QuicServer {
             .clone()
             .unwrap_or_else(RouteTable::shared);
 
-        // Create latency channel so pong-derived RTT measurements are available
-        // to the caller (e.g. for registry updates and one-shot ping responses).
-        let (latency_tx, latency_rx) = tokio::sync::mpsc::channel::<f64>(4);
+        let peer_name = peer_handshake.as_ref().map(|hs| hs.name.clone());
         let route_updates = self.options.route_updates.clone().unwrap_or_else(|| {
             let (tx, _) = tokio::sync::broadcast::channel(16);
             tx
@@ -225,10 +223,10 @@ impl Server for QuicServer {
                 let handler = Handler::new(handler_config, metrics, peers, routes, route_updates);
                 let mut channels = protocol::ControlChannels {
                     outgoing_rx: control_rx,
-                    handshake_tx: None, // Handshake already read above
-                    latency_tx: Some(latency_tx),
+                    handshake_tx: None,        // Handshake already read above
                     control_response_tx: None, // server doesn't issue ControlRequests
                     peer_registry: Some(peer_registry),
+                    peer_name,
                 };
                 let mut control_stream =
                     wallhack_transport::erased::BoxBiStream::new(control_stream);
@@ -247,7 +245,6 @@ impl Server for QuicServer {
             metrics,
             peer_handshake,
             control_tx,
-            latency_rx,
             channel_binding,
         )))
     }

crates/core/src/server/server.rs

@@ -65,9 +65,6 @@ pub struct AcceptResult<T: Transport> {
     transport: Arc<T>,
     /// Channel for injecting messages into the control stream.
     control_tx: mpsc::Sender<ControlMessage>,
-    /// Receiver for pong-derived latency measurements (milliseconds) from the
-    /// control loop. Used by one-shot ping callers.
-    latency_rx: Option<mpsc::Receiver<f64>>,
     /// TLS channel binding bytes for PSK proof verification.
     channel_binding: Option<[u8; crate::psk::CHANNEL_BINDING_LEN]>,
 }
@@ -80,7 +77,6 @@ pub struct ErasedAcceptResult {
     pub peer_handshake: Option<Handshake>,
     pub transport: Arc<dyn ErasedTransport>,
     pub control_tx: mpsc::Sender<ControlMessage>,
-    pub latency_rx: Option<mpsc::Receiver<f64>>,
     pub channel_binding: Option<[u8; crate::psk::CHANNEL_BINDING_LEN]>,
 }
 
@@ -100,7 +96,6 @@ where
             peer_handshake: self.peer_handshake.take(),
             transport: self.transport as Arc<dyn ErasedTransport>,
             control_tx: self.control_tx,
-            latency_rx: self.latency_rx.take(),
             channel_binding: self.channel_binding,
         }
     }
@@ -110,15 +105,13 @@ impl<T: Transport> AcceptResult<T> {
     /// Creates a new accept result with an already-received peer `Handshake`
     /// and a latency receiver for pong-derived RTT measurements.
     #[must_use]
-    #[allow(clippy::too_many_arguments)] // accept result construction; will be simplified when builder pattern is adopted
     pub fn with_handshake(
         transport: Arc<T>,
         channels: DataChannels,
         peer_addr: String,
         metrics: SharedMetrics,
         peer_handshake: Option<Handshake>,
         control_tx: mpsc::Sender<ControlMessage>,
-        latency_rx: mpsc::Receiver<f64>,
         channel_binding: Option<[u8; crate::psk::CHANNEL_BINDING_LEN]>,
     ) -> Self {
         Self {
@@ -128,7 +121,6 @@ impl<T: Transport> AcceptResult<T> {
             peer_handshake,
             transport,
             control_tx,
-            latency_rx: Some(latency_rx),
             channel_binding,
         }
     }
@@ -178,11 +170,6 @@ impl<T: Transport> AcceptResult<T> {
         &self.control_tx
     }
 
-    /// Takes the latency receiver for pong-derived RTT measurements.
-    pub fn take_latency_rx(&mut self) -> Option<mpsc::Receiver<f64>> {
-        self.latency_rx.take()
-    }
-
     /// Returns the TLS channel binding bytes for this connection.
     #[must_use]
     pub fn channel_binding(&self) -> Option<&[u8; crate::psk::CHANNEL_BINDING_LEN]> {

crates/core/src/server/ws/mod.rs

@@ -291,9 +291,7 @@ impl Server for WebSocketServer {
             .clone()
             .unwrap_or_else(RouteTable::shared);
 
-        // Create latency channel so pong-derived RTT measurements are available
-        // to the caller (e.g. for registry updates and one-shot ping responses).
-        let (latency_tx, latency_rx) = tokio::sync::mpsc::channel::<f64>(4);
+        let peer_name = peer_handshake.as_ref().map(|hs| hs.name.clone());
         let route_updates = self.options.route_updates.clone().unwrap_or_else(|| {
             let (tx, _) = tokio::sync::broadcast::channel(16);
             tx
@@ -306,10 +304,10 @@ impl Server for WebSocketServer {
                 let handler = Handler::new(handler_config, metrics, peers, routes, route_updates);
                 let mut channels = protocol::ControlChannels {
                     outgoing_rx: control_rx,
-                    handshake_tx: None, // Handshake already read above
-                    latency_tx: Some(latency_tx),
+                    handshake_tx: None,        // Handshake already read above
                     control_response_tx: None, // server doesn't issue ControlRequests
                     peer_registry: Some(peer_registry),
+                    peer_name,
                 };
                 let mut control_stream =
                     wallhack_transport::erased::BoxBiStream::new(control_stream);
@@ -328,7 +326,6 @@ impl Server for WebSocketServer {
             metrics,
             peer_handshake,
             control_tx,
-            latency_rx,
             channel_binding,
         )))
     }