fix(peers): allow relay failover before empty peer cleanup (#2781)

* fix(peers): route around peers without live connections

A Peer outlives its last connection until periodic cleanup. During that
window, direct sends and the least-hop shortcut still select it, even
when a usable relay route exists.

Return the original packet when connection selection finds no usable
connection, then continue through the existing relay routing path. Keep
actual send errors unchanged so packets are never retried after a send
attempt. Validate live connections in the least-hop direct shortcut.

Preserve cached connection selection and active flow bindings on normal
direct sends without an extra liveness scan or packet clone.

Add a real ring-tunnel regression covering direct and relay delivery,
with and without a flow key, while the empty Peer remains in the map.

Validation: 879 core tests, 3 native relay integration tests, cargo fmt,
and Clippy for easytier-core/easytier with full features and -D
warnings.

Domain context impact: none; peer lifecycle and protocol are unchanged.

easytier-core/src/peers/conn/peer.rs

@@ -289,30 +289,39 @@ impl Peer {
     }
 
     pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
+        match self.try_send_msg(msg).await? {
+            None => Ok(()),
+            Some(_) => Err(Error::PeerNoConnectionError(self.peer_node_id)),
+        }
+    }
+
+    /// Returns the unsent packet only when no usable connection can be selected.
+    pub(crate) async fn try_send_msg(&self, msg: ZCPacket) -> Result<Option<ZCPacket>, Error> {
         if crate::peers::traffic_metrics::data_packet_payload_len(&msg).is_some() {
             self.data_active.store(true, Ordering::Relaxed);
         }
         if let Some(key) = msg.flow_key {
-            let conn = self
-                .select_flow_conn(key)
-                .ok_or(Error::PeerNoConnectionError(self.peer_node_id))?;
-            return conn.send_msg(msg).await;
+            let Some(conn) = self.select_flow_conn(key) else {
+                return Ok(Some(msg));
+            };
+            conn.send_msg(msg).await?;
+            return Ok(None);
         }
         let default_conn = self.default_conn.load();
         if let Some(conn) = default_conn.as_ref()
             && !conn.is_closed()
         {
             conn.send_msg(msg).await?;
-            return Ok(());
+            return Ok(None);
         }
         drop(default_conn);
 
         let Some(conn) = self.select_conn() else {
-            return Err(Error::PeerNoConnectionError(self.peer_node_id));
+            return Ok(Some(msg));
         };
         conn.send_msg(msg).await?;
 
-        Ok(())
+        Ok(None)
     }
 
     pub async fn close_peer_conn(&self, conn_id: &PeerConnId) -> Result<(), Error> {
@@ -345,6 +354,14 @@ impl Peer {
     }
 
     pub fn has_live_conns(&self) -> bool {
+        if self
+            .default_conn
+            .load()
+            .as_ref()
+            .is_some_and(|conn| !conn.is_closed())
+        {
+            return true;
+        }
         self.conns.iter().any(|entry| !entry.value().is_closed())
     }
 
@@ -449,6 +466,24 @@ mod tests {
         Peer::new(2, sender, Arc::new(NoopPeerContext::default()))
     }
 
+    #[tokio::test]
+    async fn cached_live_connection_does_not_scan_connection_map() {
+        let peer = Arc::new(new_peer());
+        assert!(!peer.has_live_conns());
+        let (sender, _receiver) = create_packet_recv_chan();
+        let (conn_id, _server) = add_connection(&peer, sender, Some(1_000)).await;
+        assert!(peer.has_live_conns());
+        peer.select_conn().unwrap();
+
+        // Cached availability must not wait for the connection map's shard lock.
+        let conn_guard = peer.conns.get_mut(&conn_id).unwrap();
+        let check_peer = peer.clone();
+        let check = tokio::task::spawn_blocking(move || check_peer.has_live_conns());
+        let result = crate::foundation::time::timeout(Duration::from_secs(1), check).await;
+        drop(conn_guard);
+        assert!(result.unwrap().unwrap());
+    }
+
     #[tokio::test]
     async fn send_failure_marks_connection_closed_without_waiting_for_ping() {
         let (left, right) = create_ring_tunnel_pair();
@@ -501,6 +536,7 @@ mod tests {
         .await
         .unwrap();
         assert!(old_conn.is_closed());
+        assert!(peer.has_live_conns());
         assert_eq!(peer.get_default_conn_id(), old);
 
         let mut packet = ZCPacket::new_with_payload(b"packet after failure");

easytier-core/src/peers/conn/peer_map.rs

@@ -193,7 +193,11 @@ impl PeerMap {
             return Some(dst_peer_id);
         }
 
-        if self.has_peer(dst_peer_id) && matches!(policy, NextHopPolicy::LeastHop) {
+        if matches!(policy, NextHopPolicy::LeastHop)
+            && self
+                .get_peer_by_id(dst_peer_id)
+                .is_some_and(|peer| peer.has_live_conns())
+        {
             return Some(dst_peer_id);
         }
 
@@ -486,7 +490,151 @@ fn min_measured_latency_us(latencies_us: impl IntoIterator<Item = u64>) -> Optio
 
 #[cfg(test)]
 mod tests {
-    use super::min_measured_latency_us;
+    use super::*;
+    use crate::{
+        packet::{FlowKey, PacketType},
+        peers::{
+            conn::peer_session::PeerSessionStore,
+            create_packet_recv_chan,
+            foreign_network::client::ForeignNetworkClient,
+            peer_manager::send_msg_internal,
+            recv_packet_envelope_from_chan,
+            relay_peer_map::new_relay_peer_map,
+            route::{Route, RouteInterfaceBox},
+            test_support::NoopPeerContext,
+        },
+        tunnel::ring::create_ring_tunnel_pair,
+    };
+    use std::time::Duration;
+
+    struct TestRelayRoute;
+
+    #[async_trait::async_trait]
+    impl Route for TestRelayRoute {
+        async fn open(&self, _interface: RouteInterfaceBox) -> Result<u8, ()> {
+            Ok(0)
+        }
+
+        async fn close(&self) {}
+
+        async fn get_next_hop(&self, peer_id: PeerId) -> Option<PeerId> {
+            (peer_id == 2).then_some(3)
+        }
+
+        async fn list_routes(&self) -> Vec<CoreRoute> {
+            Vec::new()
+        }
+
+        async fn list_proxy_cidrs(&self) -> BTreeSet<cidr::Ipv4Cidr> {
+            BTreeSet::new()
+        }
+
+        async fn list_proxy_cidrs_v6(&self) -> BTreeSet<cidr::Ipv6Cidr> {
+            BTreeSet::new()
+        }
+
+        async fn get_peer_info(&self, _peer_id: PeerId) -> Option<RoutePeerInfo> {
+            None
+        }
+
+        async fn get_peer_info_last_update_time(&self) -> quanta::Instant {
+            quanta::Instant::now()
+        }
+
+        fn get_peer_groups(&self, _peer_id: PeerId) -> Arc<Vec<String>> {
+            Arc::new(Vec::new())
+        }
+    }
+
+    #[tokio::test]
+    async fn empty_peer_uses_relay_before_periodic_cleanup() {
+        let context: ArcPeerContext = Arc::new(NoopPeerContext::default());
+        let (sender, _receiver) = create_packet_recv_chan();
+        let peers = Arc::new(PeerMap::new(sender.clone(), context.clone(), 1));
+        let foreign = Arc::new(ForeignNetworkClient::new(context.clone(), sender, 1));
+        let sessions = Arc::new(PeerSessionStore::new());
+        let relay = new_relay_peer_map(
+            peers.clone(),
+            Some(foreign.clone()),
+            context.clone(),
+            1,
+            sessions.clone(),
+        );
+        peers.add_route(Arc::new(TestRelayRoute)).await;
+
+        let (sender, mut receiver) = create_packet_recv_chan();
+        let mut remote_conns = Vec::new();
+        let mut direct_conn_ids = Vec::new();
+        for peer_id in [2, 3] {
+            let (left, right) = create_ring_tunnel_pair();
+            let mut client = PeerConn::new(1, context.clone(), left, sessions.clone());
+            let mut server = PeerConn::new(peer_id, context.clone(), right, sessions.clone());
+            let (left, right) = tokio::join!(
+                client.do_handshake_as_client(),
+                server.do_handshake_as_server_ext(|_, _| Ok(())),
+            );
+            left.unwrap();
+            right.unwrap();
+            server.start_recv_loop(sender.clone()).await;
+            direct_conn_ids.push(client.get_conn_id());
+            peers.add_new_peer_conn(client).await.unwrap();
+            remote_conns.push(server);
+        }
+
+        for (direct, next_hop, remote_index) in [(true, 2, 0), (false, 3, 1)] {
+            if !direct {
+                peers.close_peer_conn(2, &direct_conn_ids[0]).await.unwrap();
+                tokio::time::timeout(Duration::from_secs(1), async {
+                    while peers.get_peer_by_id(2).unwrap().has_live_conns() {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .unwrap();
+                // Keep the empty Peer in the map: no periodic cleanup runs here.
+                assert!(peers.has_peer(2));
+            }
+            assert_eq!(
+                peers.get_gateway_peer_id(2, NextHopPolicy::LeastHop).await,
+                Some(next_hop)
+            );
+            for flow_key in [
+                None,
+                Some(FlowKey {
+                    src: ([10, 0, 0, 1], 1234).into(),
+                    dst: ([10, 0, 0, 2], 5000).into(),
+                    protocol: 6,
+                }),
+            ] {
+                for p2p_only in [true, false] {
+                    let mut packet = ZCPacket::new_with_payload(b"business data");
+                    packet.fill_peer_manager_hdr(1, 2, PacketType::Data as u8);
+                    packet.flow_key = flow_key;
+                    let result =
+                        send_msg_internal(&peers, &foreign, &relay, None, packet, 2, p2p_only)
+                            .await;
+                    if !direct && p2p_only {
+                        assert!(matches!(result, Err(Error::PeerNoConnectionError(2))));
+                        continue;
+                    }
+                    result.unwrap();
+                    let envelope = tokio::time::timeout(
+                        Duration::from_secs(1),
+                        recv_packet_envelope_from_chan(&mut receiver),
+                    )
+                    .await
+                    .unwrap()
+                    .unwrap();
+                    let (packet, ingress) = envelope.into_parts();
+                    assert_eq!(packet.payload(), b"business data");
+                    assert_eq!(
+                        ingress.peer_connection().unwrap().1,
+                        remote_conns[remote_index].get_conn_id()
+                    );
+                }
+            }
+        }
+    }
 
     #[test]
     fn unmeasured_connections_are_not_reported_as_fastest() {

easytier-core/src/peers/peer_manager.rs

@@ -2687,6 +2687,7 @@ impl PeerOutboundPacketRouter {
             Some(&self.traffic_metrics),
             msg,
             dst_peer_id,
+            packet_policy.p2p_only,
         )
         .await;
         if result.is_ok() {
@@ -2849,6 +2850,7 @@ impl PeerOutboundPacketRouter {
                 Some(&self.traffic_metrics),
                 msg,
                 cur_to_peer_id,
+                packet_policy.p2p_only,
             )
             .await;
         }
@@ -2927,6 +2929,7 @@ impl PeerOutboundPacketRouter {
                 Some(&self.traffic_metrics),
                 msg,
                 *peer_id,
+                packet_policy.p2p_only,
             )
             .await
             {
@@ -3190,6 +3193,7 @@ impl PeerPacketRouter {
                 tx_metrics,
                 ret,
                 to_peer_id,
+                false,
             )
             .await;
             if ret.is_err() {
@@ -3521,6 +3525,7 @@ pub(crate) async fn send_msg_internal(
     direct_tx_metrics: Option<&Arc<TrafficMetricRecorder>>,
     msg: ZCPacket,
     dst_peer_id: PeerId,
+    p2p_only: bool,
 ) -> Result<(), Error> {
     let policy = get_next_hop_policy(msg.peer_manager_header().unwrap().is_latency_first());
     let is_latency_first = msg.peer_manager_header().unwrap().is_latency_first();
@@ -3534,34 +3539,47 @@ pub(crate) async fn send_msg_internal(
     } else {
         None
     };
-    let send_result = if let Some(gateway) = latency_first_gateway
-        && (peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway))
-    {
-        relay_peer_map.send_msg(msg, dst_peer_id, policy).await
-    } else if let Some(peer) = peers.get_peer_by_id(dst_peer_id) {
-        peer.send_msg(msg).await
-    } else if peers.is_self(dst_peer_id) {
-        peers.send_msg_directly(msg, dst_peer_id).await
-    } else if foreign_network_client.has_next_hop(dst_peer_id) {
-        foreign_network_client.send_msg(msg, dst_peer_id).await
-    } else if let Some(gateway) = peers.get_gateway_peer_id(dst_peer_id, policy.clone()).await {
-        if peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway) {
-            relay_peer_map.send_msg(msg, dst_peer_id, policy).await
+    let send_result = async {
+        if let Some(gateway) = latency_first_gateway
+            && (peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway))
+        {
+            return relay_peer_map.send_msg(msg, dst_peer_id, policy).await;
+        }
+        let msg = if let Some(peer) = peers.get_peer_by_id(dst_peer_id) {
+            match peer.try_send_msg(msg).await? {
+                None => return Ok(()),
+                Some(_) if p2p_only => {
+                    return Err(Error::PeerNoConnectionError(dst_peer_id));
+                }
+                Some(msg) => msg,
+            }
         } else {
-            tracing::warn!(
-                ?gateway,
-                ?dst_peer_id,
-                "cannot send msg to peer through gateway"
-            );
+            msg
+        };
+        if peers.is_self(dst_peer_id) {
+            peers.send_msg_directly(msg, dst_peer_id).await
+        } else if foreign_network_client.has_next_hop(dst_peer_id) {
+            foreign_network_client.send_msg(msg, dst_peer_id).await
+        } else if let Some(gateway) = peers.get_gateway_peer_id(dst_peer_id, policy.clone()).await {
+            if peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway) {
+                relay_peer_map.send_msg(msg, dst_peer_id, policy).await
+            } else {
+                tracing::warn!(
+                    ?gateway,
+                    ?dst_peer_id,
+                    "cannot send msg to peer through gateway"
+                );
+                Err(Error::RouteError(None))
+            }
+        } else if foreign_network_client.has_next_hop(dst_peer_id) {
+            // check foreign network again. so in happy path we can avoid extra check
+            foreign_network_client.send_msg(msg, dst_peer_id).await
+        } else {
+            tracing::debug!(?dst_peer_id, "no gateway for peer");
             Err(Error::RouteError(None))
         }
-    } else if foreign_network_client.has_next_hop(dst_peer_id) {
-        // check foreign network again. so in happy path we can avoid extra check
-        foreign_network_client.send_msg(msg, dst_peer_id).await
-    } else {
-        tracing::debug!(?dst_peer_id, "no gateway for peer");
-        Err(Error::RouteError(None))
-    };
+    }
+    .await;
 
     if send_result.is_ok()
         && let Some(metrics) = direct_tx_metrics