fix(connectivity): advertise shared-node running listeners (#2733)

This commit is contained in:
KKRainbow
2026-10-21 09:47:03 +08:00
committed by GitHub
parent 318a0b0686
commit e307dadb48
5 changed files with 210 additions and 5 deletions
@@ -723,6 +723,7 @@ mod tests {
));
let handler = DirectConnectorRpcHandler::new_for_foreign_network_with_stun(
host,
Arc::new(crate::connectivity::NoLocalListeners),
SocketContext::default().with_socket_mark(Some(7)),
Some(Arc::new(FixedStunProvider)),
);
@@ -744,6 +745,14 @@ mod tests {
response.public_ipv4,
Some("198.51.100.7".parse::<std::net::Ipv4Addr>().unwrap().into())
);
assert_eq!(
response
.listeners
.into_iter()
.map(Url::from)
.collect::<Vec<_>>(),
vec!["tcp://192.0.2.1:11010".parse::<Url>().unwrap()]
);
}
fn snapshot() -> HostConnectorEnvironmentSnapshot {
+15 -4
View File
@@ -18,7 +18,7 @@ use crate::{
connectivity::hole_punch::policy::{should_background_p2p_with_peer, should_try_p2p_with_peer},
connectivity::stun::{StunInfoProvider, StunSocketMapper},
connectivity::{
LocalListenerUrls, NoLocalListeners,
LocalListenerUrls,
protocol::{
ClientProtocolUpgrader, ProtocolTransport, protocol_transport, protocol_uses_udp,
},
@@ -915,10 +915,10 @@ where
{
pub fn new_for_foreign_network_with_stun(
host: Arc<H>,
running_listeners: Arc<dyn LocalListenerUrls>,
socket_context: SocketContext,
stun: Option<Arc<dyn StunInfoProvider>>,
) -> Self {
let running_listeners = Arc::new(NoLocalListeners);
Self {
host,
peer_manager: None,
@@ -997,14 +997,23 @@ where
{
host: Arc<H>,
stun: Arc<dyn StunInfoProvider>,
running_listeners: Arc<dyn LocalListenerUrls>,
}
impl<H> ForeignDirectConnectorRpcRegistrar<H>
where
H: DirectConnectorHost,
{
pub fn new(host: Arc<H>, stun: Arc<dyn StunInfoProvider>) -> Self {
Self { host, stun }
pub fn new(
host: Arc<H>,
stun: Arc<dyn StunInfoProvider>,
running_listeners: Arc<dyn LocalListenerUrls>,
) -> Self {
Self {
host,
stun,
running_listeners,
}
}
}
@@ -1022,6 +1031,7 @@ where
GeneratedDirectConnectorRpcServer::new(
DirectConnectorRpcHandler::new_for_foreign_network_with_stun(
self.host.clone(),
self.running_listeners.clone(),
socket_context,
Some(self.stun.clone()),
),
@@ -1201,6 +1211,7 @@ fn is_public_ipv4(ip: Ipv4Addr) -> bool {
#[cfg(test)]
mod tests {
use super::*;
use crate::connectivity::NoLocalListeners;
impl<H> DirectConnectorRpcHandler<H>
where
+2 -1
View File
@@ -531,6 +531,7 @@ where
public_ipv6_host,
public_ipv6_events,
);
let running_listeners = Arc::new(RunningListenerRegistry::default());
let stun = (connectivity_mode == CoreConnectivityMode::Full)
.then(|| Self::prepare_stun(&adapters, &config.connectivity));
let (peer_stun, foreign_rpc_registrar): (
@@ -542,6 +543,7 @@ where
Arc::new(ForeignDirectConnectorRpcRegistrar::new(
adapters.host.clone(),
stun.clone(),
running_listeners.clone(),
)),
),
None => (Arc::new(()), Arc::new(())),
@@ -633,7 +635,6 @@ where
accepted_tunnel_handler.clone(),
)),
};
let running_listeners = Arc::new(RunningListenerRegistry::default());
let PreparedListenerPlan {
transports,
external,
+46
View File
@@ -3747,6 +3747,52 @@ virtual_ip = "10.82.0.2/24"
assert_eq!(state.drop_calls.load(Ordering::Relaxed), 1);
}
#[tokio::test]
async fn foreign_direct_rpc_tracks_running_listeners() {
use crate::{
connectivity::direct::DirectConnectorRpcHandler,
proto::{
peer_rpc::{DirectConnectorRpc, GetIpListRequest},
rpc_types::controller::BaseController,
},
};
let mut config = test_config("foreign-listeners");
config.connectivity.listeners = Some(ListenerRuntimeConfig::new(
vec!["tcp://0.0.0.0:0".parse().unwrap()],
false,
SocketContext::default(),
));
let instance = build_instance(config).unwrap();
let handler = DirectConnectorRpcHandler::new_for_foreign_network_with_stun(
Arc::new(TestHost::default()),
instance.running_listeners.clone(),
SocketContext::default(),
None,
);
let advertised_listeners = || async {
handler
.get_ip_list(BaseController::default(), GetIpListRequest {})
.await
.unwrap()
.listeners
.into_iter()
.map(Url::from)
.collect::<Vec<_>>()
};
assert!(advertised_listeners().await.is_empty());
instance.start().await.unwrap();
assert!(
instance
.running_listeners()
.contains(&"tcp://0.0.0.0:20000".parse().unwrap())
);
assert_eq!(advertised_listeners().await, instance.running_listeners());
instance.stop().await;
assert!(advertised_listeners().await.is_empty());
}
#[tokio::test]
async fn external_listener_uses_core_running_listener_registry() {
let external_url: Url = "unix:///tmp/easytier-external-listener-test"
+138
View File
@@ -2853,6 +2853,144 @@ pub async fn foreign_network_functional_cluster() {
drop_insts(vec![center_inst1, center_inst2, inst1, inst2]).await;
}
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]
async fn foreign_network_advertises_listeners_for_direct_connect(
#[values("tcp", "udp")] protocol: &str,
) {
use easytier_proto::{
peer_rpc::{DirectConnectorRpc, DirectConnectorRpcClientFactory, GetIpListRequest},
rpc_types::controller::BaseController,
};
let mut network = TestNetwork::new();
network.create_bridge("br");
let process_runtime = CoreProcessRuntime::new();
let mut instances = Vec::new();
for (index, name) in ["server_a", "server_b", "client_a", "client_b"]
.into_iter()
.enumerate()
{
network.create_namespace(
name,
&format!("10.81.0.{}/24", index + 1),
&format!("fd81::{}/64", index + 1),
);
network.add_to_bridge("br", name);
let config = TomlConfigLoader::default();
config.set_inst_name(name.to_owned());
config.set_netns(Some(network.namespace(name).to_owned()));
config.set_network_identity(NetworkIdentity::new(
if index < 2 {
"public-net"
} else {
"client-net"
}
.to_owned(),
String::new(),
));
config.set_listeners(if index < 2 {
// Keep listener discovery independent of interface collection in netns.
vec![
format!("{protocol}://10.81.0.{}:11010", index + 1)
.parse()
.unwrap(),
]
} else {
Vec::new()
});
let mut flags = config.get_flags();
flags.no_tun = true;
flags.enable_ipv6 = false;
flags.disable_udp_hole_punching = true;
flags.disable_tcp_hole_punching = true;
flags.relay_all_peer_rpc = true;
if index == 1 {
// The second server only relays RPC, as in the reported topology.
flags.relay_network_whitelist = String::new();
flags.disable_relay_data = true;
}
config.set_flags(flags);
assert!(config.get_mapped_listeners().is_empty());
let mut instance = Instance::new_with_process_runtime(config, process_runtime.clone());
instance.run().await.unwrap();
set_foreign_network_refresh_interval(&instance, 1).await;
instances.push(instance);
}
let [server_a, server_b, client_a, client_b] = instances.as_slice() else {
unreachable!()
};
// Bootstrap only the cluster link and each client's local server.
// Connections to the other server must use its advertised TCP/UDP listener.
server_a.add_connector_url(server_b.ring_listener_url());
client_a.add_connector_url(server_a.ring_listener_url());
client_b.add_connector_url(server_b.ring_listener_url());
for (client, remote_server) in [(client_a, server_b), (client_b, server_a)] {
let client = client.get_core_instance();
let remote_server = remote_server.get_core_instance();
wait_for_condition(
|| async {
remote_server
.foreign_network_snapshots(false)
.await
.contains_key("client-net")
},
Duration::from_secs(10),
)
.await;
let remote_peer_id = remote_server.foreign_network_snapshots(false).await["client-net"]
.my_peer_id_for_this_network;
wait_for_condition(
|| async {
client
.route_snapshots()
.await
.iter()
.any(|route| route.peer_id == remote_peer_id)
},
Duration::from_secs(10),
)
.await;
let rpc = client
.peer_manager()
.get_peer_rpc_mgr()
.rpc_client()
.scoped_client::<DirectConnectorRpcClientFactory<BaseController>>(
client.peer_id(),
remote_peer_id,
"client-net".to_owned(),
);
let listeners = rpc
.get_ip_list(BaseController::default(), GetIpListRequest {})
.await
.unwrap()
.listeners
.into_iter()
.map(url::Url::from)
.collect::<Vec<_>>();
assert_eq!(listeners, remote_server.running_listeners());
wait_for_condition(
|| async {
client.peer_snapshots().await.iter().any(|peer| {
peer.peer_id == remote_peer_id
&& peer.conns.iter().any(|conn| {
conn.is_client
&& conn
.tunnel
.as_ref()
.is_some_and(|tunnel| tunnel.tunnel_type == protocol)
})
})
},
Duration::from_secs(10),
)
.await;
}
drop_insts(instances).await;
}
#[rstest::rstest]
#[tokio::test]
#[serial_test::serial]