diff --git a/commons/zenoh-test/src/lib.rs b/commons/zenoh-test/src/lib.rs index 1399787dd9..4f2b973452 100644 --- a/commons/zenoh-test/src/lib.rs +++ b/commons/zenoh-test/src/lib.rs @@ -312,7 +312,7 @@ impl TestSessions { listener_runtime.start().await.unwrap(); let locators = listener_runtime - .get_locators() + .get_locators_noloopback() .into_iter() .map(|l| l.to_endpoint()) .collect(); diff --git a/io/zenoh-link-commons/src/listener.rs b/io/zenoh-link-commons/src/listener.rs index 3d48199c0e..96743928f3 100644 --- a/io/zenoh-link-commons/src/listener.rs +++ b/io/zenoh-link-commons/src/listener.rs @@ -138,7 +138,14 @@ impl ListenersUnicastIP { .collect() } - fn get_locators_impl(&self, noloopback: bool) -> Vec { + /// Returns the set of listener locators across all listener endpoints. + /// + /// When resolving unspecified addresses, loopback addresses are excluded if + /// and only if `exclude_unspecified_lo` is set to `true`. + /// + /// Note that `exclude_unspecified_lo` does not impact "explicit" loopback locators + /// (e.g. `tcp/127.0.0.1:9000` or `udp/localhost:0`.) + fn get_locators_impl(&self, exclude_unspecified_lo: bool) -> Vec { let mut locators = vec![]; let guard = zread!(self.listeners); @@ -150,8 +157,12 @@ impl ListenersUnicastIP { // Either ipv4/0.0.0.0 or ipv6/[::] if kip.is_unspecified() { let mut addrs = match kip { - IpAddr::V4(_) => zenoh_util::net::get_ipv4_ipaddrs(iface, noloopback), - IpAddr::V6(_) => zenoh_util::net::get_ipv6_ipaddrs(iface, noloopback), + IpAddr::V4(_) => { + zenoh_util::net::get_ipv4_ipaddrs(iface, exclude_unspecified_lo) + } + IpAddr::V6(_) => { + zenoh_util::net::get_ipv6_ipaddrs(iface, exclude_unspecified_lo) + } }; let iter = addrs.drain(..).map(|x| { Locator::new( @@ -162,7 +173,7 @@ impl ListenersUnicastIP { .unwrap() }); locators.extend(iter); - } else if !noloopback || !kip.is_loopback() { + } else { locators.push(value.endpoint.to_locator()); } } @@ -170,10 +181,13 @@ impl ListenersUnicastIP { locators } + /// Returns the set of listener locators across all listener endpoints. pub fn get_locators(&self) -> Vec { self.get_locators_impl(false) } + /// Returns the set of listener locators across all listener endpoints; + /// excludes loopback locators obtained by resolving unspecified listener endpoints. pub fn get_locators_noloopback(&self) -> Vec { self.get_locators_impl(true) } diff --git a/io/zenoh-links/zenoh-link-quic/src/unicast.rs b/io/zenoh-links/zenoh-link-quic/src/unicast.rs index a77fe701a1..808c9e54d6 100644 --- a/io/zenoh-links/zenoh-link-quic/src/unicast.rs +++ b/io/zenoh-links/zenoh-link-quic/src/unicast.rs @@ -340,10 +340,12 @@ impl LinkManagerUnicastTrait for LinkManagerUnicastQuic { self.listeners.get_endpoints() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators`]. async fn get_locators(&self) -> Vec { self.listeners.get_locators() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators_noloopback`]. async fn get_locators_noloopback(&self) -> Vec { self.listeners.get_locators_noloopback() } diff --git a/io/zenoh-links/zenoh-link-quic_datagram/src/unicast.rs b/io/zenoh-links/zenoh-link-quic_datagram/src/unicast.rs index fb87a58dbd..0bf571a01a 100644 --- a/io/zenoh-links/zenoh-link-quic_datagram/src/unicast.rs +++ b/io/zenoh-links/zenoh-link-quic_datagram/src/unicast.rs @@ -324,10 +324,12 @@ impl LinkManagerUnicastTrait for LinkManagerUnicastQuicDatagram { self.listeners.get_endpoints() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators`]. async fn get_locators(&self) -> Vec { self.listeners.get_locators() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators_noloopback`]. async fn get_locators_noloopback(&self) -> Vec { self.listeners.get_locators_noloopback() } diff --git a/io/zenoh-links/zenoh-link-tcp/src/unicast.rs b/io/zenoh-links/zenoh-link-tcp/src/unicast.rs index c590638ac2..d76e1b02f8 100644 --- a/io/zenoh-links/zenoh-link-tcp/src/unicast.rs +++ b/io/zenoh-links/zenoh-link-tcp/src/unicast.rs @@ -387,10 +387,12 @@ impl LinkManagerUnicastTrait for LinkManagerUnicastTcp { self.listeners.get_endpoints() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators`]. async fn get_locators(&self) -> Vec { self.listeners.get_locators() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators_noloopback`]. async fn get_locators_noloopback(&self) -> Vec { self.listeners.get_locators_noloopback() } diff --git a/io/zenoh-links/zenoh-link-tls/src/unicast.rs b/io/zenoh-links/zenoh-link-tls/src/unicast.rs index 77b5a9ed6e..1f8a0a3088 100644 --- a/io/zenoh-links/zenoh-link-tls/src/unicast.rs +++ b/io/zenoh-links/zenoh-link-tls/src/unicast.rs @@ -482,10 +482,12 @@ impl LinkManagerUnicastTrait for LinkManagerUnicastTls { self.listeners.get_endpoints() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators`]. async fn get_locators(&self) -> Vec { self.listeners.get_locators() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators_noloopback`]. async fn get_locators_noloopback(&self) -> Vec { self.listeners.get_locators_noloopback() } diff --git a/io/zenoh-links/zenoh-link-udp/src/unicast.rs b/io/zenoh-links/zenoh-link-udp/src/unicast.rs index 084078a48d..d49df17f5b 100644 --- a/io/zenoh-links/zenoh-link-udp/src/unicast.rs +++ b/io/zenoh-links/zenoh-link-udp/src/unicast.rs @@ -606,10 +606,12 @@ impl LinkManagerUnicastTrait for LinkManagerUnicastUdp { self.listeners.get_endpoints() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators`]. async fn get_locators(&self) -> Vec { self.listeners.get_locators() } + /// See [`zenoh_link_commons::ListenersUnicastIP::get_locators_noloopback`]. async fn get_locators_noloopback(&self) -> Vec { self.listeners.get_locators_noloopback() } diff --git a/io/zenoh-transport/src/unicast/manager.rs b/io/zenoh-transport/src/unicast/manager.rs index 0fc6fc0037..b899bea6ac 100644 --- a/io/zenoh-transport/src/unicast/manager.rs +++ b/io/zenoh-transport/src/unicast/manager.rs @@ -512,19 +512,19 @@ impl TransportManager { ); tracing::trace!("{}", e); let (l, asl) = link.fail(); - return Err(InitTransportError::Link(( + return Err(InitTransportError::Link(Box::new(( e.into(), l, asl, close::reason::INVALID, - ))); + )))); } // Add the link to the transport let (start_tx, start_rx, ack, add_link_guard) = transport .add_link(link, other_initial_sn, other_lease) .await - .map_err(InitTransportError::Link)?; + .map_err(|e| InitTransportError::Link(Box::new(e)))?; // complete establish procedure let c_link = ack.link(); @@ -605,7 +605,7 @@ impl TransportManager { Ok(output) => output, Err(e) => { let (l, asl) = link.fail(); - return Err(InitTransportError::Link((e, l, asl, $reason))); + return Err(InitTransportError::Link(Box::new((e, l, asl, $reason)))); } } }; @@ -616,12 +616,12 @@ impl TransportManager { let e = zerror!("{} Attempt to establish transport to itself", self.zid()); tracing::warn!("{e}"); let (l, asl) = link.fail(); - return Err(InitTransportError::Link(( + return Err(InitTransportError::Link(Box::new(( e.into(), l, asl, close::reason::CONNECTION_TO_SELF, - ))); + )))); } // Verify that we haven't reached the transport number limit @@ -633,12 +633,12 @@ impl TransportManager { ); tracing::trace!("{e}"); let (l, asl) = link.fail(); - return Err(InitTransportError::Link(( + return Err(InitTransportError::Link(Box::new(( e.into(), l, asl, close::reason::INVALID, - ))); + )))); } // Create the transport @@ -712,7 +712,7 @@ impl TransportManager { Ok(val) => val, Err(e) => { let _ = t.close(e.3).await; - return Err(InitTransportError::Link(e)); + return Err(InitTransportError::Link(Box::new(e))); } }; @@ -825,7 +825,8 @@ impl TransportManager { match init_result { Ok(transport) => Ok(TransportUnicast(Arc::downgrade(&transport))), - Err(InitTransportError::Link((e, link, associated_link, reason))) => { + Err(InitTransportError::Link(error)) => { + let (e, link, associated_link, reason) = *error; let _ = link.close(Some(reason)).await; if let Some(asl) = associated_link { let _ = asl.close(Some(reason)).await; diff --git a/io/zenoh-transport/src/unicast/transport_unicast_inner.rs b/io/zenoh-transport/src/unicast/transport_unicast_inner.rs index 4f41a71209..9a15b2ed18 100644 --- a/io/zenoh-transport/src/unicast/transport_unicast_inner.rs +++ b/io/zenoh-transport/src/unicast/transport_unicast_inner.rs @@ -38,7 +38,7 @@ pub(crate) type LinkError = ( ); pub(crate) type TransportError = (zenoh_result::Error, Arc, u8); pub(crate) enum InitTransportError { - Link(LinkError), + Link(Box), Transport(TransportError), } diff --git a/zenoh/src/api/info.rs b/zenoh/src/api/info.rs index d193493d7a..8d59b9735e 100644 --- a/zenoh/src/api/info.rs +++ b/zenoh/src/api/info.rs @@ -121,7 +121,7 @@ impl SessionInfo { /// ``` #[zenoh_macros::unstable] pub fn locators(&self) -> impl Resolve> + '_ { - ResolveClosure::new(|| self.session.runtime().get_locators()) + ResolveClosure::new(|| self.session.runtime().get_locators_noloopback()) } /// Return information about currently opened transport sessions. Transport session is a connection to another zenoh node. diff --git a/zenoh/src/net/runtime/adminspace.rs b/zenoh/src/net/runtime/adminspace.rs index 3143b689be..cc1d10ac02 100644 --- a/zenoh/src/net/runtime/adminspace.rs +++ b/zenoh/src/net/runtime/adminspace.rs @@ -769,7 +769,7 @@ fn metrics(prefix: &keyexpr, context: &AdminContext, query: Query) { ), zid = context.runtime.state.zid, whatami = context.runtime.state.whatami, - version = &*LONG_VERSION, + version = *LONG_VERSION, ); #[cfg(feature = "stats")] let mut metrics = String::new(); diff --git a/zenoh/tests/scouting.rs b/zenoh/tests/scouting.rs index e98122927e..45a505ebbe 100644 --- a/zenoh/tests/scouting.rs +++ b/zenoh/tests/scouting.rs @@ -22,7 +22,7 @@ use zenoh::config::WhatAmI; use zenoh::sample::SampleKind; use zenoh_config::{Config, ModeDependentValue, WhatAmIMatcher}; use zenoh_link::EndPoint; -use zenoh_test::get_free_udp_port; +use zenoh_test::{get_free_tcp_port, get_free_udp_port}; #[tokio::test(flavor = "multi_thread")] async fn multicast_scouting_works_on_loopback() { @@ -99,6 +99,90 @@ async fn multicast_scouting_works_on_loopback() { responder.close().await.unwrap(); } +#[cfg(all(feature = "unstable", feature = "transport_tcp"))] +#[tokio::test(flavor = "multi_thread")] +async fn gossip_autoconnect_works_on_loopback() { + zenoh::init_log_from_env_or("error"); + + let port_a = get_free_tcp_port(); + let port_b = get_free_tcp_port(); + let port_c = get_free_tcp_port(); + let endpoint_a = format!("tcp/127.0.0.1:{port_a}"); + let endpoint_b = format!("tcp/127.0.0.1:{port_b}"); + let endpoint_c = format!("tcp/127.0.0.1:{port_c}"); + + let peer_config = |listen: &str, connect: Option<&str>| { + let mut config = Config::default(); + config.set_mode(Some(WhatAmI::Peer)).unwrap(); + config.scouting.multicast.set_enabled(Some(false)).unwrap(); + config.scouting.gossip.set_enabled(Some(true)).unwrap(); + config + .listen + .endpoints + .set(vec![listen.parse::().unwrap()]) + .unwrap(); + if let Some(connect) = connect { + config + .connect + .endpoints + .set(vec![connect.parse::().unwrap()]) + .unwrap(); + } + config + }; + + let peer_a = zenoh::open(peer_config(&endpoint_a, None)).await.unwrap(); + let peer_a_events = peer_a + .info() + .transport_events_listener() + .with(flume::bounded(32)) + .await + .unwrap(); + + let peer_b = zenoh::open(peer_config(&endpoint_b, Some(&endpoint_a))) + .await + .unwrap(); + let peer_b_zid = peer_b.zid(); + + timeout(Duration::from_secs(5), async { + loop { + let event = peer_a_events.recv_async().await.unwrap(); + if event.kind() == SampleKind::Put && event.transport().zid() == &peer_b_zid { + break; + } + } + }) + .await + .expect("timed out waiting for the initial gossip connection"); + + // C only connects to A. It must discover B through gossip, including B's + // loopback listener address. + let peer_c = zenoh::open(peer_config(&endpoint_c, Some(&endpoint_a))) + .await + .unwrap(); + let peer_c_events = peer_c + .info() + .transport_events_listener() + .history(true) + .await + .unwrap(); + + timeout(Duration::from_secs(5), async { + loop { + let event = peer_c_events.recv_async().await.unwrap(); + if event.kind() == SampleKind::Put && event.transport().zid() == &peer_b_zid { + break; + } + } + }) + .await + .expect("timed out waiting for gossip discovery on loopback"); + + peer_c.close().await.unwrap(); + peer_b.close().await.unwrap(); + peer_a.close().await.unwrap(); +} + #[cfg(all(feature = "unstable", feature = "transport_tcp"))] #[tokio::test(flavor = "multi_thread")] async fn multicast_autoconnect_works_on_loopback() {