From ae93abd3f6d25460dfc868fab0492bbf5f6f868e Mon Sep 17 00:00:00 2001 From: KKRainbow <5665404+KKRainbow@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:15:27 +0800 Subject: [PATCH] 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 | 23 ++-- easytier-core/src/peers/conn/peer_map.rs | 145 ++++++++++++++++++++++- easytier-core/src/peers/peer_manager.rs | 60 ++++++---- 3 files changed, 194 insertions(+), 34 deletions(-) diff --git a/easytier-core/src/peers/conn/peer.rs b/easytier-core/src/peers/conn/peer.rs index a27c5ae3..6ae92ff6 100644 --- a/easytier-core/src/peers/conn/peer.rs +++ b/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, 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> { diff --git a/easytier-core/src/peers/conn/peer_map.rs b/easytier-core/src/peers/conn/peer_map.rs index 8dfff5d1..d9b5e6cc 100644 --- a/easytier-core/src/peers/conn/peer_map.rs +++ b/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,144 @@ fn min_measured_latency_us(latencies_us: impl IntoIterator) -> 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 { + Ok(0) + } + + async fn close(&self) {} + + async fn get_next_hop(&self, peer_id: PeerId) -> Option { + (peer_id == 2).then_some(3) + } + + async fn list_routes(&self) -> Vec { + Vec::new() + } + + async fn list_proxy_cidrs(&self) -> BTreeSet { + BTreeSet::new() + } + + async fn list_proxy_cidrs_v6(&self) -> BTreeSet { + BTreeSet::new() + } + + async fn get_peer_info(&self, _peer_id: PeerId) -> Option { + None + } + + async fn get_peer_info_last_update_time(&self) -> quanta::Instant { + quanta::Instant::now() + } + + fn get_peer_groups(&self, _peer_id: PeerId) -> Arc> { + 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, + }), + ] { + 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; + send_msg_internal(&peers, &foreign, &relay, None, packet, 2) + .await + .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() { diff --git a/easytier-core/src/peers/peer_manager.rs b/easytier-core/src/peers/peer_manager.rs index 354f1107..dc878eed 100644 --- a/easytier-core/src/peers/peer_manager.rs +++ b/easytier-core/src/peers/peer_manager.rs @@ -3534,34 +3534,44 @@ 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(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