From 4208b03fffb85534d0ec1a1c1880c7b852bdffbe Mon Sep 17 00:00:00 2001 From: KKRainbow <5665404+KKRainbow@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:38:02 +0800 Subject: [PATCH] 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 | 50 ++++++-- easytier-core/src/peers/conn/peer_map.rs | 152 ++++++++++++++++++++++- easytier-core/src/peers/peer_manager.rs | 68 ++++++---- 3 files changed, 236 insertions(+), 34 deletions(-) diff --git a/easytier-core/src/peers/conn/peer.rs b/easytier-core/src/peers/conn/peer.rs index a27c5ae3..361f432a 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> { @@ -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"); diff --git a/easytier-core/src/peers/conn/peer_map.rs b/easytier-core/src/peers/conn/peer_map.rs index 8dfff5d1..e607dab5 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,151 @@ 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, + }), + ] { + 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() { diff --git a/easytier-core/src/peers/peer_manager.rs b/easytier-core/src/peers/peer_manager.rs index 354f1107..bcc6806b 100644 --- a/easytier-core/src/peers/peer_manager.rs +++ b/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>, 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