diff --git a/easytier-core/src/peers/conn/peer_conn_ping.rs b/easytier-core/src/peers/conn/peer_conn_ping.rs index 05b1f533..1826f3ff 100644 --- a/easytier-core/src/peers/conn/peer_conn_ping.rs +++ b/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" + ); + } +} diff --git a/easytier-core/src/peers/peer_center/server.rs b/easytier-core/src/peers/peer_center/server.rs index 1fdac7a1..e91f933d 100644 --- a/easytier-core/src/peers/peer_center/server.rs +++ b/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, + digest: Digest, +} + #[derive(Debug, Default)] struct PeerCenterServerData { - global_peer_map: DashMap, - peer_report_time: DashMap, - digest: AtomicCell, + state: RwLock, } #[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) -> Digest { let mut hasher = std::collections::hash_map::DefaultHasher::new(); - data.global_peer_map - .iter() - .map(|v| v.key().clone()) - .collect::>() - .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 { 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)); + } } diff --git a/easytier-core/src/peers/peer_manager.rs b/easytier-core/src/peers/peer_manager.rs index 396f34ac..061e9fb8 100644 --- a/easytier-core/src/peers/peer_manager.rs +++ b/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)); diff --git a/easytier-core/src/peers/relay_peer_map.rs b/easytier-core/src/peers/relay_peer_map.rs index 763141cf..f3249cab 100644 --- a/easytier-core/src/peers/relay_peer_map.rs +++ b/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 { + Some(self.secure_mode.clone()) + } + + fn flags(&self) -> FlagsInConfig { + self.flags.clone() + } + } + + struct TestRelayTransport { + remote: StdMutex>>, + remote_pubkey: Vec, + ack_policies: StdMutex>, + } + + impl TestRelayTransport { + fn new(remote_pubkey: Vec) -> Self { + Self { + remote: StdMutex::new(None), + remote_pubkey, + ack_policies: StdMutex::new(Vec::new()), + } + } + + fn set_remote(&self, remote: &Arc) { + *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 { + 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) { + 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::*;