fix(peers): preserve p2p-only rejection during relay failover

The direct-peer existence check can pass just before the last connection
closes. Returning an unsent packet to relay routing must preserve the
caller's p2p-only policy at the actual connection selection result.

Pass the existing policy snapshot from local outbound callers and keep
PeerNoConnectionError when no connection can be selected. Transit
traffic retains its original forwarding behavior. Normal successful
direct sends do not take an additional policy branch.

Extend the same ring-tunnel regression to cover p2p-only direct delivery
and rejection before ordinary traffic falls back to the relay, both with
and without flow bindings.

Validation: 879 core tests, all 9 existing native p2p-only integration
cases, cargo fmt, full-feature Clippy with -D warnings, and WASI check.

Domain context impact: none; existing p2p-only semantics are preserved.

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> {

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