diff --git a/easytier-core/src/connectivity/connector_host.rs b/easytier-core/src/connectivity/connector_host.rs index c96e5e7b..746ffcb5 100644 --- a/easytier-core/src/connectivity/connector_host.rs +++ b/easytier-core/src/connectivity/connector_host.rs @@ -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::().unwrap().into()) ); + assert_eq!( + response + .listeners + .into_iter() + .map(Url::from) + .collect::>(), + vec!["tcp://192.0.2.1:11010".parse::().unwrap()] + ); } fn snapshot() -> HostConnectorEnvironmentSnapshot { diff --git a/easytier-core/src/connectivity/direct/mod.rs b/easytier-core/src/connectivity/direct/mod.rs index cf6269d7..44a41c12 100644 --- a/easytier-core/src/connectivity/direct/mod.rs +++ b/easytier-core/src/connectivity/direct/mod.rs @@ -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, + running_listeners: Arc, socket_context: SocketContext, stun: Option>, ) -> Self { - let running_listeners = Arc::new(NoLocalListeners); Self { host, peer_manager: None, @@ -997,14 +997,23 @@ where { host: Arc, stun: Arc, + running_listeners: Arc, } impl ForeignDirectConnectorRpcRegistrar where H: DirectConnectorHost, { - pub fn new(host: Arc, stun: Arc) -> Self { - Self { host, stun } + pub fn new( + host: Arc, + stun: Arc, + running_listeners: Arc, + ) -> 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 DirectConnectorRpcHandler where diff --git a/easytier-core/src/instance/mod.rs b/easytier-core/src/instance/mod.rs index 43303d7c..5367c138 100644 --- a/easytier-core/src/instance/mod.rs +++ b/easytier-core/src/instance/mod.rs @@ -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, diff --git a/easytier-core/src/instance/tests.rs b/easytier-core/src/instance/tests.rs index cde3f038..49abe15b 100644 --- a/easytier-core/src/instance/tests.rs +++ b/easytier-core/src/instance/tests.rs @@ -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::>() + }; + + 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" diff --git a/easytier/src/tests/three_node.rs b/easytier/src/tests/three_node.rs index 360cc5ee..c4ef3cde 100644 --- a/easytier/src/tests/three_node.rs +++ b/easytier/src/tests/three_node.rs @@ -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::>( + 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::>(); + 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]