start removing ping logic from the node_map

Author: dignifiedquire

iroh/src/magicsock.rs

@@ -29,7 +29,6 @@ use std::{
 };
 
 use bytes::Bytes;
-use data_encoding::HEXLOWER;
 use iroh_base::{NodeAddr, NodeId, PublicKey, RelayUrl, SecretKey};
 use iroh_relay::RelayMap;
 use n0_future::{
@@ -59,7 +58,7 @@ use url::Url;
 use self::transports::IpTransport;
 use self::{
     metrics::Metrics as MagicsockMetrics,
-    node_map::{NodeMap, PingAction, PingRole, SendPing},
+    node_map::{NodeMap, PingAction},
     transports::{RelayActorConfig, RelayTransport, Transports, UdpSender},
 };
 #[cfg(not(wasm_browser))]
@@ -880,13 +879,8 @@ impl MagicSock {
         let _guard = span.enter();
         trace!("receive disco message");
         match dm {
-            disco::Message::Ping(ping) => {
-                self.metrics.magicsock.recv_disco_ping.inc();
-                self.handle_ping(ping, sender, src);
-            }
-            disco::Message::Pong(pong) => {
-                self.metrics.magicsock.recv_disco_pong.inc();
-                self.node_map.handle_pong(sender, src, pong);
+            disco::Message::Ping(..) | disco::Message::Pong(..) => {
+                unreachable!("not used anymore");
             }
             disco::Message::CallMeMaybe(cm) => {
                 self.metrics.magicsock.recv_disco_call_me_maybe.inc();
@@ -913,99 +907,19 @@ impl MagicSock {
                     direct_addresses: cm.my_numbers.iter().copied().collect(),
                 });
 
-                let ping_actions =
-                    self.node_map
-                        .handle_call_me_maybe(sender, cm, &self.metrics.magicsock);
-
-                for action in ping_actions {
-                    match action {
-                        PingAction::SendCallMeMaybe { .. } => {
-                            warn!("Unexpected CallMeMaybe as response of handling a CallMeMaybe");
-                        }
-                        PingAction::SendPing(ping) => {
-                            self.send_ping_queued(ping);
-                        }
-                    }
-                }
+                self.node_map
+                    .handle_call_me_maybe(sender, cm, &self.metrics.magicsock);
             }
         }
         trace!("disco message handled");
     }
 
-    /// Handle a ping message.
-    fn handle_ping(&self, dm: disco::Ping, sender: NodeId, src: &transports::Addr) {
-        // Insert the ping into the node map, and return whether a ping with this tx_id was already
-        // received.
-        let addr: SendAddr = src.clone().into();
-        let handled = self.node_map.handle_ping(sender, addr.clone(), dm.tx_id);
-        match handled.role {
-            PingRole::Duplicate => {
-                debug!(?src, tx = %HEXLOWER.encode(&dm.tx_id), "received ping: path already confirmed, skip");
-                return;
-            }
-            PingRole::LikelyHeartbeat => {}
-            PingRole::NewPath => {
-                debug!(?src, tx = %HEXLOWER.encode(&dm.tx_id), "received ping: new path");
-            }
-            PingRole::Activate => {
-                debug!(?src, tx = %HEXLOWER.encode(&dm.tx_id), "received ping: path active");
-            }
-        }
-
-        // Send a pong.
-        debug!(tx = %HEXLOWER.encode(&dm.tx_id), %addr, dstkey = %sender.fmt_short(),
-               "sending pong");
-        let pong = disco::Message::Pong(disco::Pong {
-            tx_id: dm.tx_id,
-            ping_observed_addr: addr.clone(),
-        });
-        event!(
-            target: "iroh::_events::pong::sent",
-            Level::DEBUG,
-            remote_node = %sender.fmt_short(),
-            dst = ?addr,
-            txn = ?dm.tx_id,
-        );
-
-        if !self.disco.try_send(addr.clone(), sender, pong) {
-            warn!(%addr, "failed to queue pong");
-        }
-
-        if let Some(ping) = handled.needs_ping_back {
-            debug!(
-                %addr,
-                dstkey = %sender.fmt_short(),
-                "sending direct ping back",
-            );
-            self.send_ping_queued(ping);
-        }
-    }
-
-    fn send_ping_queued(&self, ping: SendPing) {
-        let SendPing {
-            id,
-            dst,
-            dst_node,
-            tx_id,
-            purpose,
-        } = ping;
-        let msg = disco::Message::Ping(disco::Ping {
-            tx_id,
-            node_key: self.public_key,
-        });
-        let sent = self.disco.try_send(dst.clone(), dst_node, msg);
-        if sent {
-            let msg_sender = self.actor_sender.clone();
-            trace!(%dst, tx = %HEXLOWER.encode(&tx_id), ?purpose, "ping sent (queued)");
-            self.node_map
-                .notify_ping_sent(id, dst, tx_id, purpose, msg_sender);
-        } else {
-            warn!(dst = ?dst, tx = %HEXLOWER.encode(&tx_id), ?purpose, "failed to send ping: queues full");
-        }
-    }
-
     /// Send the given ping actions out.
-    async fn send_ping_actions(&self, sender: &UdpSender, msgs: Vec<PingAction>) -> io::Result<()> {
+    async fn send_ping_actions(
+        &self,
+        _sender: &UdpSender,
+        msgs: Vec<PingAction>,
+    ) -> io::Result<()> {
         for msg in msgs {
             // Abort sending as soon as we know we are shutting down.
             if self.is_closing() || self.is_closed() {
@@ -1050,25 +964,6 @@ impl MagicSock {
                         }
                     }
                 }
-                PingAction::SendPing(SendPing {
-                    id,
-                    dst,
-                    dst_node,
-                    tx_id,
-                    purpose,
-                }) => {
-                    let msg = disco::Message::Ping(disco::Ping {
-                        tx_id,
-                        node_key: self.public_key,
-                    });
-
-                    self.send_disco_message(sender, dst.clone(), dst_node, msg)
-                        .await?;
-                    debug!(%dst, tx = %HEXLOWER.encode(&tx_id), ?purpose, "ping sent");
-                    let msg_sender = self.actor_sender.clone();
-                    self.node_map
-                        .notify_ping_sent(id, dst, tx_id, purpose, msg_sender);
-                }
             }
         }
         Ok(())
@@ -1740,7 +1635,6 @@ impl AsyncUdpSocket for MagicUdpSocket {
 #[derive(Debug)]
 enum ActorMessage {
     PingActions(Vec<PingAction>),
-    EndpointPingExpired(usize, stun_rs::TransactionId),
     NetworkChange,
     ScheduleDirectAddrUpdate(UpdateReason, Option<(NodeId, RelayUrl)>),
     #[cfg(test)]
@@ -2035,9 +1929,6 @@ impl Actor {
     /// Returns `true` if it was a shutdown.
     async fn handle_actor_message(&mut self, msg: ActorMessage, sender: &UdpSender) {
         match msg {
-            ActorMessage::EndpointPingExpired(id, txid) => {
-                self.msock.node_map.notify_ping_timeout(id, txid);
-            }
             ActorMessage::NetworkChange => {
                 self.network_monitor.network_change().await.ok();
             }

iroh/src/magicsock/node_map.rs

@@ -8,15 +8,14 @@ use std::{
 use iroh_base::{NodeAddr, NodeId, PublicKey, RelayUrl};
 use n0_future::time::Instant;
 use serde::{Deserialize, Serialize};
-use stun_rs::TransactionId;
 use tracing::{debug, info, instrument, trace, warn};
 
 use self::{
     best_addr::ClearReason,
-    node_state::{NodeState, Options, PingHandled},
+    node_state::{NodeState, Options},
 };
-use super::{metrics::Metrics, transports, ActorMessage, NodeIdMappedAddr};
-use crate::disco::{CallMeMaybe, Pong, SendAddr};
+use super::{metrics::Metrics, NodeIdMappedAddr};
+use crate::disco::CallMeMaybe;
 #[cfg(any(test, feature = "test-utils"))]
 use crate::endpoint::PathSelection;
 
@@ -25,8 +24,8 @@ mod node_state;
 mod path_state;
 mod udp_paths;
 
+pub(super) use node_state::PingAction;
 pub use node_state::{ConnectionType, ControlMsg, DirectAddrInfo, RemoteInfo};
-pub(super) use node_state::{DiscoPingPurpose, PingAction, PingRole, SendPing};
 
 /// Number of nodes that are inactive for which we keep info about. This limit is enforced
 /// periodically via [`NodeMap::prune_inactive`].
@@ -71,7 +70,6 @@ pub(super) struct NodeMapInner {
 /// have for the node.  These are all the keys the [`NodeMap`] can use.
 #[derive(Debug, Clone)]
 enum NodeStateKey {
-    Idx(usize),
     NodeId(NodeId),
     NodeIdMappedAddr(NodeIdMappedAddr),
     IpPort(IpPort),
@@ -171,35 +169,6 @@ impl NodeMap {
             .receive_relay(relay_url, src)
     }
 
-    pub(super) fn notify_ping_sent(
-        &self,
-        id: usize,
-        dst: SendAddr,
-        tx_id: stun_rs::TransactionId,
-        purpose: DiscoPingPurpose,
-        msg_sender: tokio::sync::mpsc::Sender<ActorMessage>,
-    ) {
-        if let Some(ep) = self
-            .inner
-            .lock()
-            .expect("poisoned")
-            .get_mut(NodeStateKey::Idx(id))
-        {
-            ep.ping_sent(dst, tx_id, purpose, msg_sender);
-        }
-    }
-
-    pub(super) fn notify_ping_timeout(&self, id: usize, tx_id: stun_rs::TransactionId) {
-        if let Some(ep) = self
-            .inner
-            .lock()
-            .expect("poisoned")
-            .get_mut(NodeStateKey::Idx(id))
-        {
-            ep.ping_timeout(tx_id);
-        }
-    }
-
     pub(super) fn get_quic_mapped_addr_for_node_key(
         &self,
         node_key: NodeId,
@@ -211,38 +180,16 @@ impl NodeMap {
             .map(|ep| *ep.quic_mapped_addr())
     }
 
-    /// Insert a received ping into the node map, and return whether a ping with this tx_id was already
-    /// received.
-    pub(super) fn handle_ping(
-        &self,
-        sender: PublicKey,
-        src: SendAddr,
-        tx_id: TransactionId,
-    ) -> PingHandled {
-        self.inner
-            .lock()
-            .expect("poisoned")
-            .handle_ping(sender, src, tx_id)
-    }
-
-    pub(super) fn handle_pong(&self, sender: PublicKey, src: &transports::Addr, pong: Pong) {
-        self.inner
-            .lock()
-            .expect("poisoned")
-            .handle_pong(sender, src, pong)
-    }
-
-    #[must_use = "actions must be handled"]
     pub(super) fn handle_call_me_maybe(
         &self,
         sender: PublicKey,
         cm: CallMeMaybe,
         metrics: &Metrics,
-    ) -> Vec<PingAction> {
+    ) {
         self.inner
             .lock()
             .expect("poisoned")
-            .handle_call_me_maybe(sender, cm, metrics)
+            .handle_call_me_maybe(sender, cm, metrics);
     }
 
     #[allow(clippy::type_complexity)]
@@ -404,7 +351,6 @@ impl NodeMapInner {
 
     fn get_id(&self, id: NodeStateKey) -> Option<usize> {
         match id {
-            NodeStateKey::Idx(id) => Some(id),
             NodeStateKey::NodeId(node_key) => self.by_node_key.get(&node_key).copied(),
             NodeStateKey::NodeIdMappedAddr(addr) => self.by_quic_mapped_addr.get(&addr).copied(),
             NodeStateKey::IpPort(ipp) => self.by_ip_port.get(&ipp).copied(),
@@ -500,25 +446,7 @@ impl NodeMapInner {
             .map(|ep| ep.conn_type())
     }
 
-    fn handle_pong(&mut self, sender: NodeId, src: &transports::Addr, pong: Pong) {
-        if let Some(ns) = self.get_mut(NodeStateKey::NodeId(sender)).as_mut() {
-            let insert = ns.handle_pong(&pong, src.clone().into());
-            if let Some((src, key)) = insert {
-                self.set_node_key_for_ip_port(src, &key);
-            }
-            trace!(?insert, "received pong")
-        } else {
-            warn!("received pong: node unknown, ignore")
-        }
-    }
-
-    #[must_use = "actions must be handled"]
-    fn handle_call_me_maybe(
-        &mut self,
-        sender: NodeId,
-        cm: CallMeMaybe,
-        metrics: &Metrics,
-    ) -> Vec<PingAction> {
+    fn handle_call_me_maybe(&mut self, sender: NodeId, cm: CallMeMaybe, metrics: &Metrics) {
         let ns_id = NodeStateKey::NodeId(sender);
         if let Some(id) = self.get_id(ns_id.clone()) {
             for number in &cm.my_numbers {
@@ -530,43 +458,13 @@ impl NodeMapInner {
             None => {
                 debug!("received call-me-maybe: ignore, node is unknown");
                 metrics.recv_disco_call_me_maybe_bad_disco.inc();
-                vec![]
             }
             Some(ns) => {
                 debug!(endpoints = ?cm.my_numbers, "received call-me-maybe");
 
-                ns.handle_call_me_maybe(cm)
-            }
-        }
-    }
-
-    fn handle_ping(&mut self, sender: NodeId, src: SendAddr, tx_id: TransactionId) -> PingHandled {
-        #[cfg(any(test, feature = "test-utils"))]
-        let path_selection = self.path_selection;
-        let node_state = self.get_or_insert_with(NodeStateKey::NodeId(sender), || {
-            debug!("received ping: node unknown, add to node map");
-            let source = if src.is_relay() {
-                Source::Relay
-            } else {
-                Source::Udp
-            };
-            Options {
-                node_id: sender,
-                relay_url: src.relay_url(),
-                active: true,
-                source,
-                #[cfg(any(test, feature = "test-utils"))]
-                path_selection,
-            }
-        });
-
-        let handled = node_state.handle_ping(src.clone(), tx_id);
-        if let SendAddr::Udp(ref addr) = src {
-            if matches!(handled.role, PingRole::NewPath) {
-                self.set_node_key_for_ip_port(*addr, &sender);
+                ns.handle_call_me_maybe(cm);
             }
         }
-        handled
     }
 
     /// Inserts a new node into the [`NodeMap`].
@@ -704,6 +602,7 @@ mod tests {
     use tracing_test::traced_test;
 
     use super::{node_state::MAX_INACTIVE_DIRECT_ADDRESSES, *};
+    use crate::disco::SendAddr;
 
     impl NodeMap {
         #[track_caller]
@@ -836,7 +735,7 @@ mod tests {
             let txid = stun_rs::TransactionId::from([i as u8; 12]);
             // Note that this already invokes .prune_direct_addresses() because these are
             // new UDP paths.
-            endpoint.handle_ping(addr, txid);
+            // endpoint.handle_ping(addr, txid);
         }
 
         info!("Pruning addresses");