Track path validity for all paths

Author: matheus23

iroh/src/magicsock/node_map.rs

@@ -26,6 +26,7 @@ use crate::{
 mod best_addr;
 mod node_state;
 mod path_state;
+mod path_validity;
 mod udp_paths;
 
 pub use node_state::{ConnectionType, ControlMsg, DirectAddrInfo, RemoteInfo};

iroh/src/magicsock/node_map/node_state.rs

@@ -25,7 +25,10 @@ use super::{
 use crate::endpoint::PathSelection;
 use crate::{
     disco::{self, SendAddr},
-    magicsock::{ActorMessage, MagicsockMetrics, NodeIdMappedAddr, HEARTBEAT_INTERVAL},
+    magicsock::{
+        node_map::path_validity::PathValidity, ActorMessage, MagicsockMetrics, NodeIdMappedAddr,
+        HEARTBEAT_INTERVAL,
+    },
     watchable::{Watchable, Watcher},
 };
 
@@ -242,7 +245,7 @@ impl NodeState {
             .iter()
             .map(|(addr, path_state)| DirectAddrInfo {
                 addr: SocketAddr::from(*addr),
-                latency: path_state.recent_pong.as_ref().map(|pong| pong.latency),
+                latency: path_state.validity.get_pong().map(|pong| pong.latency),
                 last_control: path_state.last_control_msg(now),
                 last_payload: path_state
                     .last_payload_msg
@@ -445,7 +448,7 @@ impl NodeState {
                             // which we should have received the pong, clear best addr and
                             // pong.  Both are used to select this path again, but we know
                             // it's not a usable path now.
-                            path_state.recent_pong = None;
+                            path_state.validity = PathValidity::empty();
                             self.udp_paths.best_addr.clear_if_equals(
                                 addr,
                                 ClearReason::PongTimeout,
@@ -894,7 +897,7 @@ impl NodeState {
         event!(
             target: "iroh::_events::pong::recv",
             Level::DEBUG,
-            remote_node = self.node_id.fmt_short(),
+            remote_node = %self.node_id.fmt_short(),
             ?src,
             txn = ?m.tx_id,
         );
@@ -1034,9 +1037,9 @@ impl NodeState {
             if !call_me_maybe_ipps.contains(ipp) {
                 // TODO: This seems like a weird way to signal that the endpoint no longer
                 // thinks it has this IpPort as an available path.
-                if st.recent_pong.is_some() {
+                if !st.validity.is_empty() {
                     debug!(path=?ipp ,"clearing recent pong");
-                    st.recent_pong = None;
+                    st.validity = PathValidity::empty();
                 }
             }
         }
@@ -1067,7 +1070,9 @@ impl NodeState {
         };
         state.receive_payload(now);
         self.last_used = Some(now);
-        if let Some((latency, confirmed_at)) = state.valid_best_addr_candidate(now) {
+        if state.validity.is_valid(now) {
+            let confirmed_at = state.validity.confirmed_at().unwrap();
+            let latency = state.validity.get_pong().unwrap().latency;
             self.udp_paths
                 .best_addr
                 .insert_if_soon_outdated_or_reconfirm(

iroh/src/magicsock/node_map/path_state.rs

@@ -16,7 +16,10 @@ use super::{
 };
 use crate::{
     disco::SendAddr,
-    magicsock::{node_map::best_addr::TRUST_UDP_ADDR_DURATION, HEARTBEAT_INTERVAL},
+    magicsock::{
+        node_map::path_validity::{self, PathValidity},
+        HEARTBEAT_INTERVAL,
+    },
 };
 
 /// The minimum time between pings to an endpoint.
@@ -30,7 +33,7 @@ const DISCO_PING_INTERVAL: Duration = Duration::from_secs(5);
 /// This state is used for both the relay path and any direct UDP paths.
 ///
 /// [`NodeState`]: super::node_state::NodeState
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone)]
 pub(super) struct PathState {
     /// The node for which this path exists.
     node_id: NodeId,
@@ -47,18 +50,19 @@ pub(super) struct PathState {
     /// The time this endpoint was last advertised via a call-me-maybe DISCO message.
     pub(super) call_me_maybe_time: Option<Instant>,
 
-    /// The most recent [`PongReply`].
+    // /// The most recent [`PongReply`].
+    // ///
+    // /// Previous replies are cleared when they are no longer relevant to determine whether
+    // /// this path can still be used to reach the remote node.
+    // pub(super) recent_pong: Option<PongReply>,
+    /// Tracks whether this path is valid.
     ///
-    /// Previous replies are cleared when they are no longer relevant to determine whether
-    /// this path can still be used to reach the remote node.
-    pub(super) recent_pong: Option<PongReply>,
+    /// See [`PathValidity`] docs.
+    pub(super) validity: PathValidity,
     /// When the last payload data was **received** via this path.
     ///
     /// This excludes DISCO messages.
     pub(super) last_payload_msg: Option<Instant>,
-    /// Whether the last payload msg was within [`TRUST_UDP_ADDR_DURATION`] of either
-    /// the last trusted payload message or a recent pong.
-    pub(super) last_payload_trusted: bool,
     /// Sources is a map of [`Source`]s to [`Instant`]s, keeping track of all the ways we have
     /// learned about this path
     ///
@@ -77,9 +81,8 @@ impl PathState {
             last_ping: None,
             last_got_ping: None,
             call_me_maybe_time: None,
-            recent_pong: None,
+            validity: PathValidity::empty(),
             last_payload_msg: None,
-            last_payload_trusted: false,
             sources,
         }
     }
@@ -105,9 +108,8 @@ impl PathState {
             last_ping: None,
             last_got_ping: None,
             call_me_maybe_time: None,
-            recent_pong: None,
+            validity: PathValidity::empty(),
             last_payload_msg: Some(now),
-            last_payload_trusted: false,
             sources,
         }
     }
@@ -126,7 +128,7 @@ impl PathState {
 
     pub(super) fn add_pong_reply(&mut self, r: PongReply) {
         if let SendAddr::Udp(ref path) = self.path {
-            if self.recent_pong.is_none() {
+            if self.validity.is_empty() {
                 event!(
                     target: "iroh::_events::holepunched",
                     Level::DEBUG,
@@ -136,27 +138,16 @@ impl PathState {
                 );
             }
         }
-        self.recent_pong = Some(r);
 
-        if let Some(last_payload_msg) = self.last_payload_msg {
-            self.last_payload_trusted = self.within_trusted_pong_duration(last_payload_msg);
-        }
-    }
-
-    fn within_trusted_pong_duration(&self, instant: Instant) -> bool {
-        if let Some(pong) = &self.recent_pong {
-            return pong.pong_at <= instant && instant < pong.pong_at + TRUST_UDP_ADDR_DURATION;
-        }
-
-        false
+        self.validity = PathValidity::new(r);
     }
 
     pub(super) fn receive_payload(&mut self, now: Instant) {
         self.last_payload_msg = Some(now);
-
-        if self.within_trusted_pong_duration(now) {
-            self.last_payload_trusted = true;
-        }
+        // TODO(matheus23): It's not necessarily UDP. Kinda weird to have this thing
+        // Also, it all results in the same 6.5s timeout anyways, maybe we just remove it?
+        self.validity
+            .receive_payload(now, path_validity::Source::Udp);
     }
 
     #[cfg(test)]
@@ -167,9 +158,8 @@ impl PathState {
             last_ping: None,
             last_got_ping: None,
             call_me_maybe_time: None,
-            recent_pong: Some(r),
+            validity: PathValidity::new(r),
             last_payload_msg: None,
-            last_payload_trusted: false,
             sources: HashMap::new(),
         }
     }
@@ -205,8 +195,8 @@ impl PathState {
     /// - When the last payload transmission occurred.
     /// - when the last ping from them was received.
     pub(super) fn last_alive(&self) -> Option<Instant> {
-        self.recent_pong
-            .as_ref()
+        self.validity
+            .get_pong()
             .map(|pong| &pong.pong_at)
             .into_iter()
             .chain(self.last_payload_msg.as_ref())
@@ -216,21 +206,6 @@ impl PathState {
             .copied()
     }
 
-    fn last_confirmed_at(&self) -> Option<Instant> {
-        let last_trusted_payload_msg = if self.last_payload_trusted {
-            self.last_payload_msg
-        } else {
-            None
-        };
-
-        self.recent_pong
-            .as_ref()
-            .map(|pong| pong.pong_at)
-            .into_iter()
-            .chain(last_trusted_payload_msg)
-            .max()
-    }
-
     /// The last control or DISCO message **about** this path.
     ///
     /// This is the most recent instant among:
@@ -242,8 +217,8 @@ impl PathState {
     pub(super) fn last_control_msg(&self, now: Instant) -> Option<(Duration, ControlMsg)> {
         // get every control message and assign it its kind
         let last_pong = self
-            .recent_pong
-            .as_ref()
+            .validity
+            .get_pong()
             .map(|pong| (pong.pong_at, ControlMsg::Pong));
         let last_call_me_maybe = self
             .call_me_maybe_time
@@ -263,19 +238,7 @@ impl PathState {
 
     /// Returns the latency from the most recent pong, if available.
     pub(super) fn latency(&self) -> Option<Duration> {
-        self.recent_pong.as_ref().map(|p| p.latency)
-    }
-
-    /// Returns the latency and confirmed_at time if this path would make for a valid best addr `now`
-    pub(crate) fn valid_best_addr_candidate(&self, now: Instant) -> Option<(Duration, Instant)> {
-        let latency = self.recent_pong.as_ref()?.latency;
-        let confirmed_at = self.last_confirmed_at()?;
-
-        if confirmed_at <= now && now < confirmed_at + TRUST_UDP_ADDR_DURATION {
-            return Some((latency, confirmed_at));
-        }
-
-        None
+        self.validity.get_pong().map(|p| p.latency)
     }
 
     pub(super) fn needs_ping(&self, now: &Instant) -> bool {
@@ -336,15 +299,15 @@ impl PathState {
         self.last_ping = None;
         self.last_got_ping = None;
         self.call_me_maybe_time = None;
-        self.recent_pong = None;
+        self.validity = PathValidity::empty();
     }
 
     fn summary(&self, mut w: impl std::fmt::Write) -> std::fmt::Result {
         write!(w, "{{ ")?;
         if self.is_active() {
             write!(w, "active ")?;
         }
-        if let Some(ref pong) = self.recent_pong {
+        if let Some(pong) = self.validity.get_pong() {
             write!(w, "pong-received({:?} ago) ", pong.pong_at.elapsed())?;
         }
         if let Some(when) = self.last_incoming_ping() {

iroh/src/magicsock/node_map/path_validity.rs

@@ -0,0 +1,92 @@
+use n0_future::time::{Duration, Instant};
+
+use crate::magicsock::node_map::node_state::PongReply;
+
+/// How long we trust a UDP address as the exclusive path (without using relay) without having heard a Pong reply.
+const TRUST_UDP_ADDR_DURATION: Duration = Duration::from_millis(6500);
+
+/// Tracks a path's validity.
+///
+/// A path is valid:
+/// - For [`Source::trust_duration`] after a successful [`PongReply`].
+/// - For [`Source::trust_duration`] longer starting at the most recent
+///   received application payload *while the path was valid*.
+#[derive(Debug, Clone, Default)]
+pub(super) struct PathValidity(Option<Inner>);
+
+#[derive(Debug, Clone)]
+struct Inner {
+    recent_pong: PongReply,
+    confirmed_at: Instant,
+    trust_until: Instant,
+}
+
+#[derive(Debug)]
+pub(super) enum Source {
+    ReceivedPong,
+    // BestCandidate,
+    Udp,
+}
+
+impl Source {
+    fn trust_duration(&self) -> Duration {
+        match self {
+            Source::ReceivedPong => TRUST_UDP_ADDR_DURATION,
+            // // TODO: Fix time
+            // Source::BestCandidate => Duration::from_secs(60 * 60),
+            Source::Udp => TRUST_UDP_ADDR_DURATION,
+        }
+    }
+}
+
+impl PathValidity {
+    pub(super) fn new(recent_pong: PongReply) -> Self {
+        Self(Some(Inner {
+            confirmed_at: recent_pong.pong_at,
+            trust_until: recent_pong.pong_at + Source::ReceivedPong.trust_duration(),
+            recent_pong,
+        }))
+    }
+
+    pub(super) fn empty() -> Self {
+        Self(None)
+    }
+
+    pub(super) fn is_empty(&self) -> bool {
+        self.0.is_none()
+    }
+
+    pub(super) fn is_valid(&self, now: Instant) -> bool {
+        let Some(state) = self.0.as_ref() else {
+            return false;
+        };
+
+        state.is_valid(now)
+    }
+
+    /// Reconfirms path validity, if a payload was received while the
+    /// path was valid.
+    pub(super) fn receive_payload(&mut self, now: Instant, source: Source) {
+        let Some(state) = self.0.as_mut() else {
+            return;
+        };
+
+        if state.is_valid(now) {
+            state.trust_until = now + source.trust_duration();
+        }
+    }
+
+    pub(super) fn get_pong(&self) -> Option<&PongReply> {
+        self.0.as_ref().map(|inner| &inner.recent_pong)
+    }
+
+    pub(super) fn confirmed_at(&self) -> Option<Instant> {
+        self.0.as_ref().map(|inner| inner.confirmed_at)
+    }
+}
+
+impl Inner {
+    fn is_valid(&self, now: Instant) -> bool {
+        self.confirmed_at <= now && now < self.trust_until
+    }
+}

iroh/src/magicsock/node_map/udp_paths.rs

@@ -151,10 +151,10 @@ impl NodeUdpPaths {
             let best_latency = best_pong
                 .map(|p: &PongReply| p.latency)
                 .unwrap_or(MAX_LATENCY);
-            match state.recent_pong {
+            match state.validity.get_pong() {
                 // This pong is better if it has a lower latency, or if it has the same
                 // latency but on an IPv6 path.
-                Some(ref pong)
+                Some(pong)
                     if pong.latency < best_latency
                         || (pong.latency == best_latency && ipp.ip().is_ipv6()) =>
                 {