fix(peer): recover from asymmetric direct connections (#2476)

Require matching pong responses before resetting consecutive liveness
failures so half-open direct connections leave the peer map.

Carry latency-first policy on relay handshakes and route replies around
stale direct peers, including handshakes started during decryption.

Store peer-center reports as atomic per-peer snapshots and include
topology costs in the digest so removals and latency updates propagate.

Drop data packets at a saturated host egress boundary instead of
blocking the shared peer packet router and shutdown path.

Add regressions for asymmetric traffic, relay ACK routing, peer-center
invalidation, and bounded host egress.

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

@@ -302,14 +302,12 @@ impl PeerConnPinger {
             )),
         );
 
-        let throughput = self.throughput_stats.clone();
-        let mut last_rx_packets = throughput.rx_packets();
-
         while let Some(ret) = ping_res_receiver.recv().await {
             if let Ok(lat) = ret {
                 latency_stats.record_latency(lat as u32);
 
                 loss_rate_stats_1.record_latency(0);
+                loss_counter.store(0, Ordering::Relaxed);
             } else {
                 loss_rate_stats_1.record_latency(1);
                 loss_counter.fetch_add(1, Ordering::Relaxed);
@@ -324,19 +322,10 @@ impl PeerConnPinger {
                 "pingpong task recv pingpong_once result"
             );
 
-            let current_rx_packets = throughput.rx_packets();
-            if last_rx_packets != current_rx_packets {
-                // if we receive some packet from peers, reset the counter to avoid conn close.
-                // conn will close only if we have 5 continous round pingpong loss after no packet received.
-                loss_counter.store(0, Ordering::Relaxed);
-            }
-
             tracing::debug!(
-                "loss_counter: {:?}, loss_rate_1: {}, cur_rx_packets: {}, last_rx: {}, node_id: {}",
+                "loss_counter: {:?}, loss_rate_1: {}, node_id: {}",
                 loss_counter,
                 loss_rate_1,
-                current_rx_packets,
-                last_rx_packets,
                 my_node_id
             );
 
@@ -346,14 +335,11 @@ impl PeerConnPinger {
                     ?self,
                     ?loss_rate_1,
                     ?loss_counter,
-                    ?last_rx_packets,
-                    ?current_rx_packets,
-                    "pingpong loss too much pingpong packet and no other ingress packets, closing the connection",
+                    "too many consecutive pingpong failures, closing the connection",
                 );
                 break;
             }
 
-            last_rx_packets = throughput.rx_packets();
             self.loss_rate_stats
                 .store((loss_rate_1 * 100.0) as u32, Ordering::Relaxed);
         }
@@ -362,3 +348,46 @@ impl PeerConnPinger {
         ping_res_receiver.close();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        peers::test_support::NoopPeerContext,
+        tunnel::{mpsc::MpscTunnel, ring::create_ring_tunnel_pair},
+    };
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn ingress_traffic_does_not_mask_failed_round_trips() {
+        let (local_tunnel, _remote_tunnel) = create_ring_tunnel_pair();
+        let tunnel = MpscTunnel::new(local_tunnel, None);
+        let (ctrl_sender, _) = broadcast::channel(16);
+        let throughput = Arc::new(Throughput::new());
+        let mut pinger = PeerConnPinger::new(
+            1,
+            2,
+            tunnel.get_sink(),
+            ctrl_sender,
+            Arc::new(WindowLatency::new(15)),
+            Arc::new(AtomicU32::new(0)),
+            throughput.clone(),
+            Arc::new(NoopPeerContext::default()),
+            "test".to_owned(),
+        );
+
+        let ingress = tokio::spawn(async move {
+            loop {
+                crate::foundation::time::sleep(Duration::from_millis(100)).await;
+                throughput.record_rx_bytes(1);
+            }
+        });
+
+        let result = timeout(Duration::from_secs(12), pinger.pingpong()).await;
+        ingress.abort();
+
+        assert!(
+            result.is_ok(),
+            "unrelated ingress traffic kept a failed round-trip alive"
+        );
+    }
+}

easytier-core/src/peers/peer_center/server.rs

@@ -1,20 +1,18 @@
 use std::{
-    collections::BinaryHeap,
+    collections::BTreeMap,
     hash::{Hash, Hasher},
     sync::Arc,
 };
 
-use crossbeam::atomic::AtomicCell;
-use dashmap::DashMap;
+use parking_lot::RwLock;
 use tokio::task::JoinSet;
 
 use crate::{
     config::PeerId,
     proto::{
         peer_rpc::{
-            DirectConnectedPeerInfo, GetGlobalPeerMapRequest, GetGlobalPeerMapResponse,
-            GlobalPeerMap, PeerCenterRpc, PeerInfoForGlobalMap, ReportPeersRequest,
-            ReportPeersResponse,
+            GetGlobalPeerMapRequest, GetGlobalPeerMapResponse, PeerCenterRpc, PeerInfoForGlobalMap,
+            ReportPeersRequest, ReportPeersResponse,
         },
         rpc_types::{self, controller::BaseController},
     },
@@ -22,23 +20,21 @@ use crate::{
 
 use super::Digest;
 
-#[derive(Debug, Clone, PartialEq, PartialOrd, Ord, Eq, Hash)]
-pub(crate) struct SrcDstPeerPair {
-    src: PeerId,
-    dst: PeerId,
-}
-
 #[derive(Debug, Clone)]
-pub(crate) struct PeerCenterInfoEntry {
-    info: DirectConnectedPeerInfo,
+struct PeerCenterInfoEntry {
+    peer_info: PeerInfoForGlobalMap,
     update_time: std::time::Instant,
 }
 
+#[derive(Debug, Default)]
+struct PeerCenterServerState {
+    peer_infos: BTreeMap<PeerId, PeerCenterInfoEntry>,
+    digest: Digest,
+}
+
 #[derive(Debug, Default)]
 struct PeerCenterServerData {
-    global_peer_map: DashMap<SrcDstPeerPair, PeerCenterInfoEntry>,
-    peer_report_time: DashMap<PeerId, std::time::Instant>,
-    digest: AtomicCell<Digest>,
+    state: RwLock<PeerCenterServerState>,
 }
 
 #[derive(Clone, Debug)]
@@ -69,24 +65,27 @@ impl PeerCenterServer {
     }
 
     async fn clean_outdated_peer_data(data: &PeerCenterServerData) {
-        data.peer_report_time.retain(|_, v| {
-            std::time::Instant::now().duration_since(*v) < std::time::Duration::from_secs(180)
-        });
-        data.global_peer_map.retain(|_, v| {
-            std::time::Instant::now().duration_since(v.update_time)
-                < std::time::Duration::from_secs(180)
-        });
+        let mut state = data.state.write();
+        let previous_len = state.peer_infos.len();
+        state
+            .peer_infos
+            .retain(|_, entry| entry.update_time.elapsed() < std::time::Duration::from_secs(180));
+        if state.peer_infos.len() != previous_len {
+            state.digest = Self::calc_global_digest_data(&state.peer_infos);
+        }
     }
 
-    fn calc_global_digest_data(data: &PeerCenterServerData) -> Digest {
+    fn calc_global_digest_data(peer_infos: &BTreeMap<PeerId, PeerCenterInfoEntry>) -> Digest {
         let mut hasher = std::collections::hash_map::DefaultHasher::new();
-        data.global_peer_map
-            .iter()
-            .map(|v| v.key().clone())
-            .collect::<BinaryHeap<_>>()
-            .into_sorted_vec()
-            .into_iter()
-            .for_each(|v| v.hash(&mut hasher));
+        peer_infos.len().hash(&mut hasher);
+        for (src_peer_id, entry) in peer_infos {
+            src_peer_id.hash(&mut hasher);
+            entry.peer_info.direct_peers.len().hash(&mut hasher);
+            for (dst_peer_id, peer_info) in &entry.peer_info.direct_peers {
+                dst_peer_id.hash(&mut hasher);
+                peer_info.latency_ms.hash(&mut hasher);
+            }
+        }
         hasher.finish()
     }
 }
@@ -107,23 +106,15 @@ impl PeerCenterRpc for PeerCenterServer {
         tracing::debug!("receive report_peers");
 
         let data = &self.data;
-        data.peer_report_time
-            .insert(my_peer_id, std::time::Instant::now());
-
-        for (peer_id, peer_info) in peers.direct_peers {
-            let pair = SrcDstPeerPair {
-                src: my_peer_id,
-                dst: peer_id,
-            };
-            let entry = PeerCenterInfoEntry {
-                info: peer_info,
+        let mut state = data.state.write();
+        state.peer_infos.insert(
+            my_peer_id,
+            PeerCenterInfoEntry {
+                peer_info: peers,
                 update_time: std::time::Instant::now(),
-            };
-            data.global_peer_map.insert(pair, entry);
-        }
-
-        data.digest
-            .store(PeerCenterServer::calc_global_digest_data(data));
+            },
+        );
+        state.digest = PeerCenterServer::calc_global_digest_data(&state.peer_infos);
 
         Ok(ReportPeersResponse::default())
     }
@@ -136,27 +127,20 @@ impl PeerCenterRpc for PeerCenterServer {
     ) -> Result<GetGlobalPeerMapResponse, rpc_types::error::Error> {
         let digest = req.digest;
 
-        let data = &self.data;
-        if digest == data.digest.load() && digest != 0 {
+        let state = self.data.state.read();
+        if digest == state.digest && digest != 0 {
             return Ok(GetGlobalPeerMapResponse::default());
         }
 
-        let mut global_peer_map = GlobalPeerMap::default();
-        for item in data.global_peer_map.iter() {
-            let (pair, entry) = item.pair();
-            global_peer_map
-                .map
-                .entry(pair.src)
-                .or_insert_with(|| PeerInfoForGlobalMap {
-                    direct_peers: Default::default(),
-                })
-                .direct_peers
-                .insert(pair.dst, entry.info);
-        }
+        let global_peer_map = state
+            .peer_infos
+            .iter()
+            .map(|(peer_id, entry)| (*peer_id, entry.peer_info.clone()))
+            .collect();
 
         Ok(GetGlobalPeerMapResponse {
-            global_peer_map: global_peer_map.map,
-            digest: Some(data.digest.load()),
+            global_peer_map,
+            digest: Some(state.digest),
         })
     }
 }
@@ -164,6 +148,7 @@ impl PeerCenterRpc for PeerCenterServer {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::proto::peer_rpc::DirectConnectedPeerInfo;
 
     #[tokio::test]
     async fn server_clones_share_instance_data() {
@@ -236,4 +221,164 @@ mod tests {
             .unwrap();
         assert!(resp_b.global_peer_map.is_empty());
     }
+
+    #[tokio::test]
+    async fn peer_report_replaces_removed_neighbors() {
+        let server = PeerCenterServer::new();
+        let mut peers = PeerInfoForGlobalMap::default();
+        peers
+            .direct_peers
+            .insert(100, DirectConnectedPeerInfo { latency_ms: 3 });
+        server
+            .report_peers(
+                BaseController::default(),
+                ReportPeersRequest {
+                    my_peer_id: 99,
+                    peer_infos: Some(peers),
+                },
+            )
+            .await
+            .unwrap();
+        let initial = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest { digest: 0 },
+            )
+            .await
+            .unwrap();
+
+        server
+            .report_peers(
+                BaseController::default(),
+                ReportPeersRequest {
+                    my_peer_id: 99,
+                    peer_infos: Some(PeerInfoForGlobalMap::default()),
+                },
+            )
+            .await
+            .unwrap();
+        let updated = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest {
+                    digest: initial.digest.unwrap(),
+                },
+            )
+            .await
+            .unwrap();
+
+        assert!(
+            updated.digest.is_some(),
+            "removed peer did not change digest"
+        );
+        assert!(
+            updated
+                .global_peer_map
+                .get(&99)
+                .is_none_or(|peers| peers.direct_peers.is_empty())
+        );
+    }
+
+    #[tokio::test]
+    async fn peer_latency_change_invalidates_digest() {
+        let server = PeerCenterServer::new();
+        let mut peers = PeerInfoForGlobalMap::default();
+        peers
+            .direct_peers
+            .insert(100, DirectConnectedPeerInfo { latency_ms: 3 });
+        server
+            .report_peers(
+                BaseController::default(),
+                ReportPeersRequest {
+                    my_peer_id: 99,
+                    peer_infos: Some(peers),
+                },
+            )
+            .await
+            .unwrap();
+        let initial = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest { digest: 0 },
+            )
+            .await
+            .unwrap();
+
+        let mut peers = PeerInfoForGlobalMap::default();
+        peers
+            .direct_peers
+            .insert(100, DirectConnectedPeerInfo { latency_ms: 30 });
+        server
+            .report_peers(
+                BaseController::default(),
+                ReportPeersRequest {
+                    my_peer_id: 99,
+                    peer_infos: Some(peers),
+                },
+            )
+            .await
+            .unwrap();
+        let updated = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest {
+                    digest: initial.digest.unwrap(),
+                },
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(
+            updated.global_peer_map[&99].direct_peers[&100].latency_ms,
+            30
+        );
+    }
+
+    #[tokio::test]
+    async fn expired_peer_report_invalidates_digest() {
+        let server = PeerCenterServer::new();
+        let mut peers = PeerInfoForGlobalMap::default();
+        peers
+            .direct_peers
+            .insert(100, DirectConnectedPeerInfo { latency_ms: 3 });
+        server
+            .report_peers(
+                BaseController::default(),
+                ReportPeersRequest {
+                    my_peer_id: 99,
+                    peer_infos: Some(peers),
+                },
+            )
+            .await
+            .unwrap();
+        let initial = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest { digest: 0 },
+            )
+            .await
+            .unwrap();
+        {
+            let mut state = server.data.state.write();
+            state.peer_infos.get_mut(&99).unwrap().update_time =
+                std::time::Instant::now() - std::time::Duration::from_secs(181);
+        }
+
+        PeerCenterServer::clean_outdated_peer_data(&server.data).await;
+        let updated = server
+            .get_global_peer_map(
+                BaseController::default(),
+                GetGlobalPeerMapRequest {
+                    digest: initial.digest.unwrap(),
+                },
+            )
+            .await
+            .unwrap();
+
+        assert!(
+            updated.digest.is_some(),
+            "expired peer did not change digest"
+        );
+        assert!(!updated.global_peer_map.contains_key(&99));
+    }
 }

easytier-core/src/peers/peer_manager.rs

@@ -420,10 +420,15 @@ impl PeerPacketFilter for NicPacketProcessor {
                 return None;
             }
             tracing::trace!(?packet, "send packet to nic channel");
-            let _ = self
+            if let Err(error) = self
                 .nic_channel
-                .send(HostPacket::from_core_packet(packet))
-                .await;
+                .try_send(HostPacket::from_core_packet(packet))
+            {
+                tracing::trace!(
+                    ?error,
+                    "dropping packet because nic channel cannot accept it"
+                );
+            }
             None
         } else {
             Some(packet)
@@ -3284,6 +3289,28 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn full_host_packet_queue_does_not_block_peer_packet_processing() {
+        let (nic_channel, mut nic_receiver) = mpsc::channel(1);
+        nic_channel
+            .send(HostPacket::copy_from_payload(b"queued"))
+            .await
+            .unwrap();
+        let processor = NicPacketProcessor { nic_channel };
+        let mut packet = ZCPacket::new_with_payload(b"next");
+        packet.fill_peer_manager_hdr(1, 2, PacketType::Data as u8);
+
+        let result = tokio::time::timeout(
+            Duration::from_millis(100),
+            processor.try_process_packet_from_peer(packet),
+        )
+        .await;
+
+        assert!(result.is_ok(), "a full host queue blocked the peer router");
+        assert!(result.unwrap().is_none());
+        assert_eq!(nic_receiver.try_recv().unwrap().payload(), b"queued");
+    }
+
     #[test]
     fn managed_nic_pipeline_removal_preserves_in_flight_snapshot() {
         let drops = Arc::new(AtomicUsize::new(0));

easytier-core/src/peers/relay_peer_map.rs

@@ -202,6 +202,9 @@ impl RelayPeerMap {
     ) -> Result<(), Error> {
         let mut pkt = ZCPacket::new_with_payload(&payload);
         pkt.fill_peer_manager_hdr(self.my_peer_id, dst_peer_id, packet_type as u8);
+        pkt.mut_peer_manager_header()
+            .unwrap()
+            .set_latency_first(matches!(&policy, NextHopPolicy::LeastCost));
         let pkt_len = pkt.buf_len() as u64;
         self.send_via_next_hop(pkt, dst_peer_id, policy).await?;
         self.context
@@ -549,6 +552,17 @@ impl RelayPeerMap {
     }
 
     async fn handle_relay_msg1(&self, msg1: ZCPacket, remote_peer_id: PeerId) -> Result<(), Error> {
+        let header = msg1
+            .peer_manager_header()
+            .ok_or_else(|| Error::RouteError(Some("packet without header".to_string())))?;
+        let ack_policy = if header.is_latency_first() || header.forward_counter > 0 {
+            // Older peers do not mark latency-first handshakes. A forwarded
+            // request must still avoid the stale direct-peer shortcut.
+            NextHopPolicy::LeastCost
+        } else {
+            NextHopPolicy::LeastHop
+        };
+
         // Check for bidirectional handshake race condition.
         // If we are also waiting for a RelayHandshakeAck from this peer,
         // use deterministic rule: the peer with smaller peer_id becomes initiator.
@@ -657,7 +671,7 @@ impl RelayPeerMap {
             out[..out_len].to_vec(),
             PacketType::RelayHandshakeAck,
             remote_peer_id,
-            NextHopPolicy::LeastHop,
+            ack_policy,
         )
         .await?;
 
@@ -676,6 +690,11 @@ impl RelayPeerMap {
             .peer_manager_header()
             .ok_or_else(|| Error::RouteError(Some("packet without header".to_string())))?;
         let from_peer_id = hdr.from_peer_id.get();
+        let handshake_policy = if hdr.is_latency_first() || hdr.forward_counter > 0 {
+            NextHopPolicy::LeastCost
+        } else {
+            NextHopPolicy::LeastHop
+        };
         let network = self.context.network_identity();
         let key = SessionKey::new(network.network_name.clone(), from_peer_id);
         let Some(session) = self.peer_session_store.get(&key) else {
@@ -683,8 +702,7 @@ impl RelayPeerMap {
                 "relay session not found for peer {}, try handshake",
                 from_peer_id
             );
-            self.ensure_session(from_peer_id, NextHopPolicy::LeastHop)
-                .await?;
+            self.ensure_session(from_peer_id, handshake_policy).await?;
             return Ok(false);
         };
         let now = Instant::now();
@@ -737,6 +755,158 @@ impl RelayPeerMap {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use std::sync::{Mutex as StdMutex, Weak};
+
+    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
+    use x25519_dalek::{PublicKey, StaticSecret};
+
+    use super::*;
+    use crate::{
+        peers::context::{ArcPeerContext, NetworkIdentity, PeerContext},
+        proto::common::{FlagsInConfig, SecureModeConfig},
+    };
+
+    struct RelayTestContext {
+        network_identity: NetworkIdentity,
+        secure_mode: SecureModeConfig,
+        flags: FlagsInConfig,
+    }
+
+    impl PeerContext for RelayTestContext {
+        fn network_identity(&self) -> NetworkIdentity {
+            self.network_identity.clone()
+        }
+
+        fn secure_mode(&self) -> Option<SecureModeConfig> {
+            Some(self.secure_mode.clone())
+        }
+
+        fn flags(&self) -> FlagsInConfig {
+            self.flags.clone()
+        }
+    }
+
+    struct TestRelayTransport {
+        remote: StdMutex<Option<Weak<RelayPeerMap>>>,
+        remote_pubkey: Vec<u8>,
+        ack_policies: StdMutex<Vec<NextHopPolicy>>,
+    }
+
+    impl TestRelayTransport {
+        fn new(remote_pubkey: Vec<u8>) -> Self {
+            Self {
+                remote: StdMutex::new(None),
+                remote_pubkey,
+                ack_policies: StdMutex::new(Vec::new()),
+            }
+        }
+
+        fn set_remote(&self, remote: &Arc<RelayPeerMap>) {
+            *self.remote.lock().unwrap() = Some(Arc::downgrade(remote));
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl RelayRouteTransport for TestRelayTransport {
+        async fn get_route_peer_info(&self, peer_id: PeerId) -> Option<RoutePeerInfo> {
+            Some(RoutePeerInfo {
+                peer_id,
+                noise_static_pubkey: self.remote_pubkey.clone(),
+                ..Default::default()
+            })
+        }
+
+        async fn send_msg_to_next_hop(
+            &self,
+            msg: ZCPacket,
+            _dst_peer_id: PeerId,
+            policy: NextHopPolicy,
+        ) -> Result<(), Error> {
+            let packet_type = msg.peer_manager_header().unwrap().packet_type;
+            if packet_type == PacketType::RelayHandshakeAck as u8 {
+                self.ack_policies.lock().unwrap().push(policy.clone());
+                if matches!(policy, NextHopPolicy::LeastHop) {
+                    return Ok(());
+                }
+            }
+
+            let remote = self
+                .remote
+                .lock()
+                .unwrap()
+                .as_ref()
+                .and_then(Weak::upgrade)
+                .ok_or_else(|| Error::RouteError(Some("test relay is unavailable".to_owned())))?;
+            remote.handle_handshake_packet(msg).await
+        }
+    }
+
+    fn relay_test_context(seed: u8) -> (ArcPeerContext, Vec<u8>) {
+        let private = StaticSecret::from([seed; 32]);
+        let public = PublicKey::from(&private);
+        let context = RelayTestContext {
+            network_identity: NetworkIdentity {
+                network_name: "test".to_owned(),
+                network_secret: Some("secret".to_owned()),
+                network_secret_digest: None,
+            },
+            secure_mode: SecureModeConfig {
+                enabled: true,
+                local_private_key: Some(BASE64_STANDARD.encode(private.as_bytes())),
+                local_public_key: Some(BASE64_STANDARD.encode(public.as_bytes())),
+            },
+            flags: FlagsInConfig {
+                encryption_algorithm: "aes-gcm".to_owned(),
+                ..Default::default()
+            },
+        };
+        (Arc::new(context), public.as_bytes().to_vec())
+    }
+
+    #[tokio::test]
+    async fn latency_first_handshake_returns_ack_over_least_cost_route() {
+        let (context_a, pubkey_a) = relay_test_context(1);
+        let (context_b, pubkey_b) = relay_test_context(2);
+        let transport_a = Arc::new(TestRelayTransport::new(pubkey_b));
+        let transport_b = Arc::new(TestRelayTransport::new(pubkey_a));
+        let relay_a = RelayPeerMap::new(
+            transport_a.clone(),
+            context_a,
+            1,
+            Arc::new(PeerSessionStore::new()),
+        );
+        let relay_b = RelayPeerMap::new(
+            transport_b.clone(),
+            context_b,
+            2,
+            Arc::new(PeerSessionStore::new()),
+        );
+        transport_a.set_remote(&relay_b);
+        transport_b.set_remote(&relay_a);
+
+        tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            relay_a.handshake_session(2, NextHopPolicy::LeastCost, None),
+        )
+        .await
+        .expect("relay handshake ack used the blackholed least-hop route")
+        .unwrap();
+
+        assert!(relay_a.has_session_without_touch(2));
+        assert!(relay_b.has_session_without_touch(1));
+        assert!(
+            transport_b
+                .ack_policies
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|policy| matches!(policy, NextHopPolicy::LeastCost))
+        );
+    }
+}
+
 #[cfg(any(test, feature = "test-utils"))]
 mod test_utils {
     use super::*;