fix(peers): keep TCP hole-punched connections alive with 1s pings (#2632)

Disable ping interval backoff for TCP hole-punched connections so idle
connections continue to send keepalive traffic every second. Preserve
the existing backoff and loss handling for other connections.

Keep randomized backoff above zero and cover the one-second schedule
and the existing backoff and loss retry behavior with tests.

* refactor(peers): model connection origins at admission

Record manual, direct, listener, TCP/UDP hole-punch, and attached
origins when constructing peer connections. Derive hole-punch state
from this origin instead of maintaining separate mutable flags.

Choose the one-second TCP hole-punch ping limit in PeerConn and pass
only a maximum interval to the pinger. Keep other origins on the
existing backoff schedule without inspecting tunnel type strings.

Keep origin selection internal and preserve the dedicated attached
admission paths. Update public admission callers and cover origin
propagation, relay restrictions, and ping interval limits.
This commit is contained in:
KKRainbow authored and GitHub committed 2026-10-05 16:25:08 +08:00
1 parent 6abc3731ae
commit d3a5ae684a
17 files changed
+256 -117

No files matched your search

+8 -3
View File
@@ -28,8 +28,9 @@ use crate::{
foundation::task::{PeerTaskLauncher, PeerTaskManager}, foundation::task::{PeerTaskLauncher, PeerTaskManager},
host::dns::DnsResolver, host::dns::DnsResolver,
peers::{ peers::{
conn::peer_conn::PeerConnId, foreign_network::ForeignNetworkRpcRegistrar, PeerConnectionOrigin, conn::peer_conn::PeerConnId,
peer_manager::PeerManagerCore, peer_rpc::PeerRpcManager, foreign_network::ForeignNetworkRpcRegistrar, peer_manager::PeerManagerCore,
peer_rpc::PeerRpcManager,
}, },
process_runtime::ProtectedTcpPortRegistry, process_runtime::ProtectedTcpPortRegistry,
proto::{ proto::{
@@ -870,7 +871,11 @@ where
dst_peer_id: PeerId, dst_peer_id: PeerId,
) -> anyhow::Result<(PeerId, PeerConnId)> { ) -> anyhow::Result<(PeerId, PeerConnId)> {
self.peer_manager self.peer_manager
.add_client_tunnel_with_peer_id_hint(tunnel, true, Some(dst_peer_id)) .add_client_tunnel_with_peer_id_hint(
tunnel,
PeerConnectionOrigin::Direct,
Some(dst_peer_id),
)
.await .await
.map_err(Into::into) .map_err(Into::into)
} }
@@ -1,5 +1,6 @@
use async_trait::async_trait; use async_trait::async_trait;
use crate::peers::PeerConnectionOrigin;
use crate::proto::rpc_types::{controller::BaseController, handler::Handler}; use crate::proto::rpc_types::{controller::BaseController, handler::Handler};
use crate::tunnel::Tunnel; use crate::tunnel::Tunnel;
@@ -27,7 +28,15 @@ pub(crate) trait HolePunchRpcRegistry: Send + Sync + 'static {
#[async_trait] #[async_trait]
pub(crate) trait HolePunchTunnelSink: Send + Sync + 'static { pub(crate) trait HolePunchTunnelSink: Send + Sync + 'static {
async fn add_client_tunnel(&self, tunnel: Box<dyn Tunnel>) -> anyhow::Result<()>; async fn add_client_tunnel(
&self,
tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()>;
async fn add_server_tunnel(&self, tunnel: Box<dyn Tunnel>) -> anyhow::Result<()>; async fn add_server_tunnel(
&self,
tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()>;
} }
@@ -9,7 +9,7 @@ use quanta::Instant;
use crate::{ use crate::{
config::{P2pPolicyFlags, PeerId}, config::{P2pPolicyFlags, PeerId},
foundation::task::ExternalTaskSignal, foundation::task::ExternalTaskSignal,
peers::peer_manager::PeerManagerCore, peers::{PeerConnectionOrigin, peer_manager::PeerManagerCore},
proto::{ proto::{
common::NatType, common::NatType,
peer_rpc::{ peer_rpc::{
@@ -97,16 +97,25 @@ impl UdpHolePunchRpcSource for PeerManagerCore {
#[async_trait] #[async_trait]
impl HolePunchTunnelSink for PeerManagerCore { impl HolePunchTunnelSink for PeerManagerCore {
async fn add_client_tunnel(&self, tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_client_tunnel(
PeerManagerCore::add_client_tunnel(self, tunnel, false) &self,
tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
self.add_client_tunnel_with_peer_id_hint(tunnel, origin, None)
.await .await
.map(|_| ()) .map(|_| ())
.map_err(anyhow::Error::from) .map_err(anyhow::Error::from)
} }
async fn add_server_tunnel(&self, tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_server_tunnel(
PeerManagerCore::add_tunnel_as_server(self, tunnel, false) &self,
tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
self.add_tunnel_as_server_with_origin(tunnel, origin)
.await .await
.map(|_| ())
.map_err(anyhow::Error::from) .map_err(anyhow::Error::from)
} }
} }
@@ -31,6 +31,7 @@ use crate::{
foundation::task::{ foundation::task::{
ExternalTaskSignal, PeerTaskLauncher, PeerTaskManager, reap_joinset_background, ExternalTaskSignal, PeerTaskLauncher, PeerTaskManager, reap_joinset_background,
}, },
peers::PeerConnectionOrigin,
proto::{ proto::{
common::{NatType, PeerFeatureFlag}, common::{NatType, PeerFeatureFlag},
peer_rpc::{ peer_rpc::{
@@ -159,8 +160,16 @@ where
.await .await
.map_err(TcpHolePunchTransportError::Upgrade)?; .map_err(TcpHolePunchTransportError::Upgrade)?;
match admission { match admission {
TcpHolePunchAdmission::Client => self.tunnel_sink.add_client_tunnel(tunnel).await, TcpHolePunchAdmission::Client => {
TcpHolePunchAdmission::Server => self.tunnel_sink.add_server_tunnel(tunnel).await, self.tunnel_sink
.add_client_tunnel(tunnel, PeerConnectionOrigin::TcpHolePunch)
.await
}
TcpHolePunchAdmission::Server => {
self.tunnel_sink
.add_server_tunnel(tunnel, PeerConnectionOrigin::TcpHolePunch)
.await
}
} }
.map_err(TcpHolePunchTransportError::Admission) .map_err(TcpHolePunchTransportError::Admission)
} }
@@ -181,7 +190,7 @@ where
))); )));
}; };
self.tunnel_sink self.tunnel_sink
.add_server_tunnel(tunnel) .add_server_tunnel(tunnel, PeerConnectionOrigin::TcpHolePunch)
.await .await
.map_err(TcpHolePunchTransportError::Admission) .map_err(TcpHolePunchTransportError::Admission)
} }
@@ -979,7 +988,12 @@ mod tests {
#[async_trait] #[async_trait]
impl HolePunchTunnelSink for MockTunnelSink { impl HolePunchTunnelSink for MockTunnelSink {
async fn add_client_tunnel(&self, _tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_client_tunnel(
&self,
_tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
assert_eq!(origin, PeerConnectionOrigin::TcpHolePunch);
if self.fail_client_admission.load(Ordering::Relaxed) { if self.fail_client_admission.load(Ordering::Relaxed) {
anyhow::bail!("mock client admission failure"); anyhow::bail!("mock client admission failure");
} }
@@ -987,7 +1001,12 @@ mod tests {
Ok(()) Ok(())
} }
async fn add_server_tunnel(&self, _tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_server_tunnel(
&self,
_tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
assert_eq!(origin, PeerConnectionOrigin::TcpHolePunch);
self.servers.fetch_add(1, Ordering::Relaxed); self.servers.fetch_add(1, Ordering::Relaxed);
Ok(()) Ok(())
} }
@@ -14,6 +14,7 @@ use crate::{
transport::{ConnectedTransport, ConnectedUdpSession}, transport::{ConnectedTransport, ConnectedUdpSession},
}, },
foundation::task::ExternalTaskSignal, foundation::task::ExternalTaskSignal,
peers::PeerConnectionOrigin,
socket::{ socket::{
ListenerConnectionCounter, SocketContext, ListenerConnectionCounter, SocketContext,
udp::{UdpBindOptions, UdpSession, VirtualUdpSocket, VirtualUdpSocketFactory}, udp::{UdpBindOptions, UdpSession, VirtualUdpSocket, VirtualUdpSocketFactory},
@@ -271,7 +272,9 @@ where
requested_url: url::Url, requested_url: url::Url,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
let tunnel = self.upgrade(connected, requested_url).await?; let tunnel = self.upgrade(connected, requested_url).await?;
self.tunnel_sink.add_client_tunnel(tunnel).await self.tunnel_sink
.add_client_tunnel(tunnel, PeerConnectionOrigin::UdpHolePunch)
.await
} }
async fn add_server_transport( async fn add_server_transport(
@@ -280,7 +283,9 @@ where
requested_url: url::Url, requested_url: url::Url,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
let tunnel = self.upgrade(connected, requested_url).await?; let tunnel = self.upgrade(connected, requested_url).await?;
self.tunnel_sink.add_server_tunnel(tunnel).await self.tunnel_sink
.add_server_tunnel(tunnel, PeerConnectionOrigin::UdpHolePunch)
.await
} }
} }
@@ -428,12 +433,22 @@ mod tests {
#[async_trait] #[async_trait]
impl HolePunchTunnelSink for MockTunnelSink { impl HolePunchTunnelSink for MockTunnelSink {
async fn add_client_tunnel(&self, _tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_client_tunnel(
&self,
_tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
assert_eq!(origin, PeerConnectionOrigin::UdpHolePunch);
self.clients.fetch_add(1, Ordering::Relaxed); self.clients.fetch_add(1, Ordering::Relaxed);
Ok(()) Ok(())
} }
async fn add_server_tunnel(&self, _tunnel: Box<dyn Tunnel>) -> anyhow::Result<()> { async fn add_server_tunnel(
&self,
_tunnel: Box<dyn Tunnel>,
origin: PeerConnectionOrigin,
) -> anyhow::Result<()> {
assert_eq!(origin, PeerConnectionOrigin::UdpHolePunch);
self.servers.fetch_add(1, Ordering::Relaxed); self.servers.fetch_add(1, Ordering::Relaxed);
Ok(()) Ok(())
} }
+2 -2
View File
@@ -26,7 +26,7 @@ use crate::{
}, },
events::{CoreEvent, CoreEventSink}, events::{CoreEvent, CoreEventSink},
host::dns::{DnsQuery, DnsResolver}, host::dns::{DnsQuery, DnsResolver},
peers::peer_manager::PeerManagerCore, peers::{PeerConnectionOrigin, peer_manager::PeerManagerCore},
proto::common::TunnelInfo, proto::common::TunnelInfo,
socket::{ socket::{
IpVersion, SocketContext, IpVersion, SocketContext,
@@ -891,7 +891,7 @@ where
let (peer_id, conn_id) = let (peer_id, conn_id) =
with_timeout_budget("handshake", started_at, connect_timeout, async move { with_timeout_budget("handshake", started_at, connect_timeout, async move {
peer_manager peer_manager
.add_client_tunnel_with_peer_id_hint(tunnel, true, None) .add_client_tunnel_with_peer_id_hint(tunnel, PeerConnectionOrigin::Manual, None)
.await .await
.map_err(anyhow::Error::from) .map_err(anyhow::Error::from)
}) })
+2 -2
View File
@@ -113,8 +113,8 @@ async fn setup_data_plane_pair() -> (DataPlaneEndpoint, DataPlaneEndpoint) {
let client_tunnel = registry.connect(listener_id).unwrap().into_tunnel(); let client_tunnel = registry.connect(listener_id).unwrap().into_tunnel();
let server_tunnel = listener.accept().await.unwrap().into_tunnel(); let server_tunnel = listener.accept().await.unwrap().into_tunnel();
let (client, server) = tokio::join!( let (client, server) = tokio::join!(
b.peer_manager.add_client_tunnel(client_tunnel, true), b.peer_manager.add_client_tunnel(client_tunnel),
a.peer_manager.add_tunnel_as_server(server_tunnel, true), a.peer_manager.add_tunnel_as_server(server_tunnel),
); );
client.unwrap(); client.unwrap();
server.unwrap(); server.unwrap();
+1 -4
View File
@@ -46,11 +46,8 @@ where
pub async fn admit_client_tunnel_for_test( pub async fn admit_client_tunnel_for_test(
&self, &self,
tunnel: Box<dyn crate::tunnel::Tunnel>, tunnel: Box<dyn crate::tunnel::Tunnel>,
is_directly_connected: bool,
) -> Result<(crate::config::PeerId, PeerConnId), crate::peers::error::Error> { ) -> Result<(crate::config::PeerId, PeerConnId), crate::peers::error::Error> {
self.peer_manager self.peer_manager.add_client_tunnel(tunnel).await
.add_client_tunnel(tunnel, is_directly_connected)
.await
} }
#[doc(hidden)] #[doc(hidden)]
+1 -1
View File
@@ -63,7 +63,7 @@ impl AcceptedTunnelHandler for PeerAcceptedTunnelHandler {
tracing::error!(error = %error, "handle conn error"); tracing::error!(error = %error, "handle conn error");
return Err(anyhow::anyhow!(error)); return Err(anyhow::anyhow!(error));
}; };
if let Err(error) = peer_manager.add_tunnel_as_server(tunnel, true).await { if let Err(error) = peer_manager.add_tunnel_as_server(tunnel).await {
self.events.emit(CoreEvent::TunnelAdmissionFailed { self.events.emit(CoreEvent::TunnelAdmissionFailed {
local_url, local_url,
remote_url, remote_url,
+54 -10
View File
@@ -293,9 +293,6 @@ pub struct PeerConn {
info: Option<HandshakeRequest>, info: Option<HandshakeRequest>,
is_client: Option<bool>, is_client: Option<bool>,
// remote or local
is_hole_punched: bool,
close_event_notifier: Arc<PeerConnCloseNotify>, close_event_notifier: Arc<PeerConnCloseNotify>,
ctrl_resp_sender: broadcast::Sender<ZCPacket>, ctrl_resp_sender: broadcast::Sender<ZCPacket>,
@@ -333,7 +330,7 @@ impl PeerConn {
tunnel, tunnel,
None, None,
peer_session_store, peer_session_store,
PeerConnectionOrigin::Network, PeerConnectionOrigin::Manual,
) )
} }
@@ -396,8 +393,6 @@ impl PeerConn {
info: None, info: None,
is_client: None, is_client: None,
is_hole_punched: true,
close_event_notifier: Arc::new(PeerConnCloseNotify::new(conn_id)), close_event_notifier: Arc::new(PeerConnCloseNotify::new(conn_id)),
ctrl_resp_sender: ctrl_sender, ctrl_resp_sender: ctrl_sender,
@@ -443,12 +438,23 @@ impl PeerConn {
self.origin == PeerConnectionOrigin::Attached self.origin == PeerConnectionOrigin::Attached
} }
pub fn set_is_hole_punched(&mut self, is_hole_punched: bool) { pub fn is_hole_punched(&self) -> bool {
self.is_hole_punched = is_hole_punched; matches!(
self.origin,
PeerConnectionOrigin::TcpHolePunch | PeerConnectionOrigin::UdpHolePunch
)
} }
pub fn is_hole_punched(&self) -> bool { fn max_ping_interval(&self) -> Duration {
self.is_hole_punched match self.origin {
// TCP hole-punched connections need frequent traffic to stay alive.
PeerConnectionOrigin::TcpHolePunch => Duration::from_secs(1),
PeerConnectionOrigin::Manual
| PeerConnectionOrigin::Direct
| PeerConnectionOrigin::Listener
| PeerConnectionOrigin::UdpHolePunch
| PeerConnectionOrigin::Attached => Duration::from_secs(32),
}
} }
pub fn is_closed(&self) -> bool { pub fn is_closed(&self) -> bool {
@@ -1377,6 +1383,7 @@ impl PeerConn {
self.context.clone(), self.context.clone(),
self.get_conn_info().network_name, self.get_conn_info().network_name,
self.liveness.clone(), self.liveness.clone(),
self.max_ping_interval(),
); );
let close_event_notifier = self.close_event_notifier.clone(); let close_event_notifier = self.close_event_notifier.clone();
@@ -1526,3 +1533,40 @@ impl Drop for PeerConn {
self.close_event_notifier.notify_close(); self.close_event_notifier.notify_close();
} }
} }
#[cfg(test)]
mod tests {
use super::*;
use crate::{peers::test_support::NoopPeerContext, tunnel::ring::create_ring_tunnel_pair};
#[tokio::test]
async fn connection_origin_determines_hole_punch_and_ping_policy() {
for (origin, is_hole_punched, max_interval) in [
(PeerConnectionOrigin::Manual, false, 32),
(PeerConnectionOrigin::Direct, false, 32),
(PeerConnectionOrigin::Listener, false, 32),
(PeerConnectionOrigin::TcpHolePunch, true, 1),
(PeerConnectionOrigin::UdpHolePunch, true, 32),
(PeerConnectionOrigin::Attached, false, 32),
] {
// Admission determines the policy even when the transport is a ring.
let (tunnel, _remote_tunnel) = create_ring_tunnel_pair();
let conn = PeerConn::new_with_peer_id_hint_and_origin(
1,
Arc::new(NoopPeerContext::default()),
tunnel,
None,
Arc::new(PeerSessionStore::new()),
origin,
);
assert_eq!(conn.is_hole_punched(), is_hole_punched, "{origin:?}");
assert_eq!(
conn.max_ping_interval(),
Duration::from_secs(max_interval),
"{origin:?}",
);
assert_eq!(conn.is_attached(), origin == PeerConnectionOrigin::Attached);
}
}
}
+73 -3
View File
@@ -34,6 +34,7 @@ struct PingIntervalController {
loss_counter: Arc<AtomicU32>, loss_counter: Arc<AtomicU32>,
interval: Interval, interval: Interval,
max_interval: Duration,
logic_time: u64, logic_time: u64,
last_send_logic_time: u64, last_send_logic_time: u64,
@@ -53,19 +54,25 @@ impl std::fmt::Debug for PingIntervalController {
.field("last_send_logic_time", &self.last_send_logic_time) .field("last_send_logic_time", &self.last_send_logic_time)
.field("backoff_idx", &self.backoff_idx) .field("backoff_idx", &self.backoff_idx)
.field("max_backoff_idx", &self.max_backoff_idx) .field("max_backoff_idx", &self.max_backoff_idx)
.field("max_interval", &self.max_interval)
.field("last_throughput", &self.last_throughput) .field("last_throughput", &self.last_throughput)
.finish() .finish()
} }
} }
impl PingIntervalController { impl PingIntervalController {
fn new(throughput: Arc<Throughput>, loss_counter: Arc<AtomicU32>) -> Self { fn new(
throughput: Arc<Throughput>,
loss_counter: Arc<AtomicU32>,
max_interval: Duration,
) -> Self {
let last_throughput = (*throughput).clone(); let last_throughput = (*throughput).clone();
Self { Self {
throughput, throughput,
loss_counter, loss_counter,
interval: interval(Duration::from_secs(1)), interval: interval(Duration::from_secs(1)),
max_interval,
logic_time: 0, logic_time: 0,
last_send_logic_time: 0, last_send_logic_time: 0,
@@ -99,7 +106,8 @@ impl PingIntervalController {
self.last_throughput = (*self.throughput).clone(); self.last_throughput = (*self.throughput).clone();
if (self.logic_time - self.last_send_logic_time) < (1 << self.backoff_idx) { let send_interval = Duration::from_secs(1 << self.backoff_idx).min(self.max_interval);
if Duration::from_secs(self.logic_time - self.last_send_logic_time) < send_interval {
return false; return false;
} }
@@ -126,6 +134,7 @@ pub struct PeerConnPinger {
context: ArcPeerContext, context: ArcPeerContext,
network_name: String, network_name: String,
liveness: PeerConnLiveness, liveness: PeerConnLiveness,
max_interval: Duration,
} }
impl std::fmt::Debug for PeerConnPinger { impl std::fmt::Debug for PeerConnPinger {
@@ -150,6 +159,7 @@ impl PeerConnPinger {
context: ArcPeerContext, context: ArcPeerContext,
network_name: String, network_name: String,
liveness: PeerConnLiveness, liveness: PeerConnLiveness,
max_interval: Duration,
) -> Self { ) -> Self {
Self { Self {
my_peer_id, my_peer_id,
@@ -162,6 +172,7 @@ impl PeerConnPinger {
context, context,
network_name, network_name,
liveness, liveness,
max_interval,
} }
} }
@@ -245,8 +256,10 @@ impl PeerConnPinger {
let mut controller_tasks = JoinSet::new(); let mut controller_tasks = JoinSet::new();
let throughput = self.throughput_stats.clone(); let throughput = self.throughput_stats.clone();
let controller_loss_counter = loss_counter.clone(); let controller_loss_counter = loss_counter.clone();
let max_interval = self.max_interval;
controller_tasks.spawn(async move { controller_tasks.spawn(async move {
let mut controller = PingIntervalController::new(throughput, controller_loss_counter); let mut controller =
PingIntervalController::new(throughput, controller_loss_counter, max_interval);
loop { loop {
controller.tick().await; controller.tick().await;
if !controller.should_send_ping() { if !controller.should_send_ping() {
@@ -337,6 +350,61 @@ mod tests {
}, },
}; };
#[cfg(not(target_os = "wasi"))]
#[tokio::test(start_paused = true)]
async fn one_second_limit_disables_ping_backoff() {
let mut controller = PingIntervalController::new(
Arc::new(Throughput::new()),
Arc::new(AtomicU32::new(0)),
Duration::from_secs(1),
);
let started_at = tokio::time::Instant::now();
for second in 0..100 {
controller.tick().await;
assert_eq!(started_at.elapsed(), Duration::from_secs(second));
assert!(controller.should_send_ping());
assert!(!controller.should_send_ping());
}
}
#[cfg(not(target_os = "wasi"))]
#[tokio::test(start_paused = true)]
async fn ping_backoff_respects_non_power_of_two_limit() {
let mut controller = PingIntervalController::new(
Arc::new(Throughput::new()),
Arc::new(AtomicU32::new(0)),
Duration::from_secs(3),
);
for tick in 1..=100 {
controller.tick().await;
assert_eq!(controller.should_send_ping(), tick == 1 || tick % 3 == 0);
}
}
#[cfg(not(target_os = "wasi"))]
#[tokio::test(start_paused = true)]
async fn default_limit_preserves_backoff_and_loss_retries() {
let loss_counter = Arc::new(AtomicU32::new(0));
let mut controller = PingIntervalController::new(
Arc::new(Throughput::new()),
loss_counter.clone(),
Duration::from_secs(32),
);
for second in 1..=14 {
controller.tick().await;
assert_eq!(controller.should_send_ping(), matches!(second, 1 | 3 | 7));
}
loss_counter.store(1, Ordering::Relaxed);
for _ in 0..3 {
controller.tick().await;
assert!(controller.should_send_ping());
}
}
#[tokio::test(flavor = "current_thread")] #[tokio::test(flavor = "current_thread")]
async fn ingress_traffic_does_not_mask_failed_round_trips() { async fn ingress_traffic_does_not_mask_failed_round_trips() {
let (local_tunnel, _remote_tunnel) = create_ring_tunnel_pair(); let (local_tunnel, _remote_tunnel) = create_ring_tunnel_pair();
@@ -354,6 +422,7 @@ mod tests {
Arc::new(NoopPeerContext::default()), Arc::new(NoopPeerContext::default()),
"test".to_owned(), "test".to_owned(),
PeerConnLiveness::new(), PeerConnLiveness::new(),
Duration::from_secs(32),
); );
let ingress = tokio::spawn(async move { let ingress = tokio::spawn(async move {
@@ -412,6 +481,7 @@ mod tests {
Arc::new(NoopPeerContext::default()), Arc::new(NoopPeerContext::default()),
"test".to_owned(), "test".to_owned(),
local_liveness, local_liveness,
Duration::from_secs(32),
); );
let result = timeout(Duration::from_secs(12), pinger.pingpong()).await; let result = timeout(Duration::from_secs(12), pinger.pingpong()).await;
+6 -1
View File
@@ -27,9 +27,14 @@ use tokio::sync::mpsc::error::{SendError, TryRecvError, TrySendError};
use self::conn::peer_conn::PeerConnId; use self::conn::peer_conn::PeerConnId;
use crate::config::PeerId; use crate::config::PeerId;
/// The local entry point that created a peer connection.
#[derive(Clone, Copy, Debug, Eq, PartialEq)] #[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum PeerConnectionOrigin { pub(crate) enum PeerConnectionOrigin {
Network, Manual,
Direct,
Listener,
TcpHolePunch,
UdpHolePunch,
Attached, Attached,
} }
+38 -71
View File
@@ -1565,21 +1565,19 @@ impl PeerManagerCore {
pub async fn add_client_tunnel( pub async fn add_client_tunnel(
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
self.peer_connection_admission self.add_client_tunnel_with_peer_id_hint(tunnel, PeerConnectionOrigin::Manual, None)
.add_client_tunnel(tunnel, is_directly_connected)
.await .await
} }
pub async fn add_client_tunnel_with_peer_id_hint( pub(crate) async fn add_client_tunnel_with_peer_id_hint(
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
is_directly_connected: bool, origin: PeerConnectionOrigin,
peer_id_hint: Option<PeerId>, peer_id_hint: Option<PeerId>,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
self.peer_connection_admission self.peer_connection_admission
.add_client_tunnel_with_peer_id_hint(tunnel, is_directly_connected, peer_id_hint) .add_client_tunnel_with_peer_id_hint(tunnel, origin, peer_id_hint)
.await .await
} }
@@ -1587,18 +1585,23 @@ impl PeerManagerCore {
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
self.peer_connection_admission self.add_client_tunnel_with_peer_id_hint(tunnel, PeerConnectionOrigin::Attached, None)
.add_client_tunnel_with_origin(tunnel, true, None, PeerConnectionOrigin::Attached)
.await .await
} }
pub async fn add_tunnel_as_server( pub async fn add_tunnel_as_server(&self, tunnel: Box<dyn Tunnel>) -> Result<(), Error> {
self.add_tunnel_as_server_with_origin(tunnel, PeerConnectionOrigin::Listener)
.await
.map(|_| ())
}
pub(crate) async fn add_tunnel_as_server_with_origin(
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
is_directly_connected: bool, origin: PeerConnectionOrigin,
) -> Result<(), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
self.peer_connection_admission self.peer_connection_admission
.add_tunnel_as_server(tunnel, is_directly_connected) .add_tunnel_as_server_with_origin(tunnel, origin)
.await .await
} }
@@ -1606,8 +1609,7 @@ impl PeerManagerCore {
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
self.peer_connection_admission self.add_tunnel_as_server_with_origin(tunnel, PeerConnectionOrigin::Attached)
.add_tunnel_as_server_with_origin(tunnel, true, PeerConnectionOrigin::Attached)
.await .await
} }
@@ -1958,36 +1960,11 @@ impl PeerConnectionAdmission {
} }
} }
pub async fn add_client_tunnel(
&self,
tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
) -> Result<(PeerId, PeerConnId), Error> {
self.add_client_tunnel_with_peer_id_hint(tunnel, is_directly_connected, None)
.await
}
pub async fn add_client_tunnel_with_peer_id_hint( pub async fn add_client_tunnel_with_peer_id_hint(
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
peer_id_hint: Option<PeerId>,
) -> Result<(PeerId, PeerConnId), Error> {
self.add_client_tunnel_with_origin(
tunnel,
is_directly_connected,
peer_id_hint,
PeerConnectionOrigin::Network,
)
.await
}
async fn add_client_tunnel_with_origin(
&self,
tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
peer_id_hint: Option<PeerId>,
origin: PeerConnectionOrigin, origin: PeerConnectionOrigin,
peer_id_hint: Option<PeerId>,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
let mut peer = PeerConn::new_with_peer_id_hint_and_origin( let mut peer = PeerConn::new_with_peer_id_hint_and_origin(
self.my_peer_id, self.my_peer_id,
@@ -1997,7 +1974,6 @@ impl PeerConnectionAdmission {
self.peer_session_store.clone(), self.peer_session_store.clone(),
origin, origin,
); );
peer.set_is_hole_punched(!is_directly_connected);
peer.do_handshake_as_client().await?; peer.do_handshake_as_client().await?;
let conn_id = peer.get_conn_id(); let conn_id = peer.get_conn_id();
let peer_id = peer.get_peer_id(); let peer_id = peer.get_peer_id();
@@ -2036,24 +2012,9 @@ impl PeerConnectionAdmission {
} }
#[tracing::instrument(ret, skip(self, tunnel))] #[tracing::instrument(ret, skip(self, tunnel))]
pub async fn add_tunnel_as_server(
&self,
tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
) -> Result<(), Error> {
self.add_tunnel_as_server_with_origin(
tunnel,
is_directly_connected,
PeerConnectionOrigin::Network,
)
.await
.map(|_| ())
}
async fn add_tunnel_as_server_with_origin( async fn add_tunnel_as_server_with_origin(
&self, &self,
tunnel: Box<dyn Tunnel>, tunnel: Box<dyn Tunnel>,
is_directly_connected: bool,
origin: PeerConnectionOrigin, origin: PeerConnectionOrigin,
) -> Result<(PeerId, PeerConnId), Error> { ) -> Result<(PeerId, PeerConnId), Error> {
tracing::info!("add tunnel as server start"); tracing::info!("add tunnel as server start");
@@ -2138,8 +2099,6 @@ impl PeerConnectionAdmission {
)); ));
} }
conn.set_is_hole_punched(!is_directly_connected);
let add_peer_ret = if is_local_network { let add_peer_ret = if is_local_network {
let local_secure_mode = self let local_secure_mode = self
.context .context
@@ -4424,19 +4383,27 @@ mod tests {
#[test] #[test]
fn forged_attached_source_header_does_not_bypass_relay_disable() { fn forged_attached_source_header_does_not_bypass_relay_disable() {
let packet = data_packet(77, 3); let packet = data_packet(77, 3);
let network_ingress = PeerPacketIngress::Peer { for origin in [
peer_id: 2, PeerConnectionOrigin::Manual,
conn_id: PeerConnId::new_v4(), PeerConnectionOrigin::Direct,
origin: PeerConnectionOrigin::Network, PeerConnectionOrigin::Listener,
}; PeerConnectionOrigin::TcpHolePunch,
PeerConnectionOrigin::UdpHolePunch,
] {
let network_ingress = PeerPacketIngress::Peer {
peer_id: 2,
conn_id: PeerConnId::new_v4(),
origin,
};
assert!(should_drop_relay_data( assert!(should_drop_relay_data(
true, true,
&packet, &packet,
1, 1,
network_ingress, network_ingress,
false, false,
)); ));
}
} }
#[test] #[test]
@@ -4450,7 +4417,7 @@ mod tests {
let network_ingress = PeerPacketIngress::Peer { let network_ingress = PeerPacketIngress::Peer {
peer_id: 2, peer_id: 2,
conn_id: PeerConnId::new_v4(), conn_id: PeerConnId::new_v4(),
origin: PeerConnectionOrigin::Network, origin: PeerConnectionOrigin::Listener,
}; };
assert!(!should_drop_relay_data( assert!(!should_drop_relay_data(
-1
View File
@@ -285,7 +285,6 @@ async fn peer_channel_uses_admission_origin_instead_of_packet_header() {
); );
client_ret.unwrap(); client_ret.unwrap();
server_ret.unwrap(); server_ret.unwrap();
server_conn.set_is_hole_punched(false);
let server_conn_id = server_conn.get_conn_id(); let server_conn_id = server_conn.get_conn_id();
let (client_tx, _client_rx) = create_packet_recv_chan(); let (client_tx, _client_rx) = create_packet_recv_chan();
+1 -1
View File
@@ -379,7 +379,7 @@ impl NetworkInstanceManager {
tokio::select! { tokio::select! {
biased; biased;
_ = retiring.cancelled() => None, _ = retiring.cancelled() => None,
result = peer_manager.add_tunnel_as_server(tunnel, true) => Some(result), result = peer_manager.add_tunnel_as_server(tunnel) => Some(result),
} }
} }
+1 -1
View File
@@ -582,7 +582,7 @@ P2P hole punch 的流程:
这个流程不受影响,因为: 这个流程不受影响,因为:
- 打洞信息交换通过管理节点中继(RPC),不经过临时节点 - 打洞信息交换通过管理节点中继(RPC),不经过临时节点
- P2P tunnel 建立后的握手是直连,不通过临时节点的 listener - P2P tunnel 建立后的握手是直连,不通过临时节点的 listener
- `is_directly_connected=false` 的连接(hole punch 结果)可以被临时节点接受 - 来源为 `TcpHolePunch` 或 `UdpHolePunch` 的连接可以被临时节点接受
**设计思路**: 将凭据映射为 ACL Group,复用现有的 group-based ACL 规则系统。 **设计思路**: 将凭据映射为 ACL Group,复用现有的 group-based ACL 规则系统。
+1 -1
View File
@@ -3048,7 +3048,7 @@ async fn assert_peer_admission_blocked(inst: &Instance, url: url::Url) {
url, url,
) )
.await?; .await?;
core.admit_client_tunnel_for_test(tunnel, true) core.admit_client_tunnel_for_test(tunnel)
.await .await
.map(|_| ()) .map(|_| ())
.map_err(anyhow::Error::from) .map_err(anyhow::Error::from)