feat!: update to rust edition 2024

dignifiedquire · dignifiedquire · commit 1d9dcdf7a1f5 · 2025-07-12T15:49:11.000+02:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -40,3 +40,8 @@ unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)", "cfg(iroh_l
 
 [workspace.lints.clippy]
 unused-async = "warn"
+
+
+[patch.crates-io]
+portmapper = { git = "https://github.com/n0-computer/net-tools", branch = "deps-update" }
+netwatch = { git = "https://github.com/n0-computer/net-tools", branch = "deps-update" }
diff --git a/iroh-dns-server/src/store/signed_packets.rs b/iroh-dns-server/src/store/signed_packets.rs
@@ -151,7 +151,7 @@ impl Actor {
                             Message::Upsert { packet, res } => {
                                 let key = PublicKeyBytes::from_signed_packet(&packet);
                                 trace!("upsert {}", key);
-                                let replaced = if let Some(existing) = get_packet(&tables.signed_packets, &key)? {
+                                let replaced = match get_packet(&tables.signed_packets, &key)? { Some(existing) => {
                                     if existing.more_recent_than(&packet) {
                                         res.send(false).ok();
                                         continue;
@@ -160,9 +160,9 @@ impl Actor {
                                         tables.update_time.remove(&existing.timestamp().to_bytes(), key.as_bytes()).e()?;
                                         true
                                     }
-                                } else {
+                                } _ => {
                                     false
-                                };
+                                }};
                                 let value = packet.serialize();
                                 tables.signed_packets
                                     .insert(key.as_bytes(), &value[..]).e()?;
@@ -177,14 +177,14 @@ impl Actor {
                             }
                             Message::Remove { key, res } => {
                                 trace!("remove {}", key);
-                                let updated = if let Some(row) = tables.signed_packets.remove(key.as_bytes()).e()? {
+                                let updated = match tables.signed_packets.remove(key.as_bytes()).e()? { Some(row) => {
                                     let packet = SignedPacket::deserialize(row.value()).e()?;
                                     tables.update_time.remove(&packet.timestamp().to_bytes(), key.as_bytes()).e()?;
                                     self.metrics.store_packets_removed.inc();
                                     true
-                                } else {
+                                } _ => {
                                     false
-                                };
+                                }};
                                 res.send(updated).ok();
                             }
                             Message::Snapshot { res } => {
@@ -193,7 +193,7 @@ impl Actor {
                             }
                             Message::CheckExpired { key, time } => {
                                 trace!("check expired {} at {}", key, fmt_time(time));
-                                if let Some(packet) = get_packet(&tables.signed_packets, &key)? {
+                                match get_packet(&tables.signed_packets, &key)? { Some(packet) => {
                                     let expired = Timestamp::now() - expiry_us;
                                     if packet.timestamp() < expired {
                                         tables.update_time.remove(&time.to_bytes(), key.as_bytes()).e()?;
@@ -204,10 +204,10 @@ impl Actor {
                                         debug!("packet {key} is no longer expired, removing obsolete expiry entry");
                                         tables.update_time.remove(&time.to_bytes(), key.as_bytes()).e()?;
                                     }
-                                } else {
+                                } _ => {
                                     debug!("expired packet {key} not found, remove from expiry table");
                                     tables.update_time.remove(&time.to_bytes(), key.as_bytes()).e()?;
-                                }
+                                }}
                             }
                         }
                     }
diff --git a/iroh-relay/src/server/http_server.rs b/iroh-relay/src/server/http_server.rs
@@ -518,10 +518,13 @@ impl RelayService {
                     async move {
                         match hyper::upgrade::on(&mut req).await {
                             Ok(upgraded) => {
-                                if let Err(err) = this.0.relay_connection_handler(upgraded).await {
-                                    warn!("error accepting upgraded connection: {err:#}",);
-                                } else {
-                                    debug!("upgraded connection completed");
+                                match this.0.relay_connection_handler(upgraded).await {
+                                    Err(err) => {
+                                        warn!("error accepting upgraded connection: {err:#}",);
+                                    }
+                                    _ => {
+                                        debug!("upgraded connection completed");
+                                    }
                                 };
                             }
                             Err(err) => warn!("upgrade error: {err:#}"),
diff --git a/iroh-relay/src/server/streams.rs b/iroh-relay/src/server/streams.rs
@@ -173,10 +173,10 @@ impl AsyncRead for MaybeTlsStream {
         buf: &mut tokio::io::ReadBuf<'_>,
     ) -> Poll<std::io::Result<()>> {
         match &mut *self {
-            MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_read(cx, buf),
-            MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_read(cx, buf),
+            MaybeTlsStream::Plain(s) => Pin::new(s).poll_read(cx, buf),
+            MaybeTlsStream::Tls(s) => Pin::new(s).poll_read(cx, buf),
             #[cfg(test)]
-            MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_read(cx, buf),
+            MaybeTlsStream::Test(s) => Pin::new(s).poll_read(cx, buf),
         }
     }
 }
@@ -187,10 +187,10 @@ impl AsyncWrite for MaybeTlsStream {
         cx: &mut Context<'_>,
     ) -> Poll<std::result::Result<(), std::io::Error>> {
         match &mut *self {
-            MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_flush(cx),
-            MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_flush(cx),
+            MaybeTlsStream::Plain(s) => Pin::new(s).poll_flush(cx),
+            MaybeTlsStream::Tls(s) => Pin::new(s).poll_flush(cx),
             #[cfg(test)]
-            MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_flush(cx),
+            MaybeTlsStream::Test(s) => Pin::new(s).poll_flush(cx),
         }
     }
 
@@ -199,10 +199,10 @@ impl AsyncWrite for MaybeTlsStream {
         cx: &mut Context<'_>,
     ) -> Poll<std::result::Result<(), std::io::Error>> {
         match &mut *self {
-            MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_shutdown(cx),
-            MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_shutdown(cx),
+            MaybeTlsStream::Plain(s) => Pin::new(s).poll_shutdown(cx),
+            MaybeTlsStream::Tls(s) => Pin::new(s).poll_shutdown(cx),
             #[cfg(test)]
-            MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_shutdown(cx),
+            MaybeTlsStream::Test(s) => Pin::new(s).poll_shutdown(cx),
         }
     }
 
@@ -212,10 +212,10 @@ impl AsyncWrite for MaybeTlsStream {
         buf: &[u8],
     ) -> Poll<std::result::Result<usize, std::io::Error>> {
         match &mut *self {
-            MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_write(cx, buf),
-            MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_write(cx, buf),
+            MaybeTlsStream::Plain(s) => Pin::new(s).poll_write(cx, buf),
+            MaybeTlsStream::Tls(s) => Pin::new(s).poll_write(cx, buf),
             #[cfg(test)]
-            MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_write(cx, buf),
+            MaybeTlsStream::Test(s) => Pin::new(s).poll_write(cx, buf),
         }
     }
 
@@ -225,10 +225,10 @@ impl AsyncWrite for MaybeTlsStream {
         bufs: &[std::io::IoSlice<'_>],
     ) -> Poll<std::result::Result<usize, std::io::Error>> {
         match &mut *self {
-            MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),
-            MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),
+            MaybeTlsStream::Plain(s) => Pin::new(s).poll_write_vectored(cx, bufs),
+            MaybeTlsStream::Tls(s) => Pin::new(s).poll_write_vectored(cx, bufs),
             #[cfg(test)]
-            MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),
+            MaybeTlsStream::Test(s) => Pin::new(s).poll_write_vectored(cx, bufs),
         }
     }
 }
diff --git a/iroh/src/discovery.rs b/iroh/src/discovery.rs
@@ -699,7 +699,7 @@ impl DiscoverySubscribers {
         }
     }
 
-    pub(crate) fn subscribe(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> {
+    pub(crate) fn subscribe(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> + use<> {
         use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream};
         let recv = self.inner.subscribe();
         BroadcastStream::new(recv).map_err(|BroadcastStreamRecvError::Lagged(n)| Lagged { val: n })
diff --git a/iroh/src/discovery/pkarr.rs b/iroh/src/discovery/pkarr.rs
@@ -353,24 +353,27 @@ impl PublisherService {
                 break; // disconnected
             };
             if let Some(info) = info {
-                if let Err(err) = self.publish_current(info).await {
-                    failed_attempts += 1;
-                    // Retry after increasing timeout
-                    let retry_after = Duration::from_secs(failed_attempts);
-                    republish.as_mut().reset(Instant::now() + retry_after);
-                    warn!(
-                        err = %format!("{err:#}"),
-                        url = %self.pkarr_client.pkarr_relay_url ,
-                        ?retry_after,
-                        %failed_attempts,
-                        "Failed to publish to pkarr",
-                    );
-                } else {
-                    failed_attempts = 0;
-                    // Republish after fixed interval
-                    republish
-                        .as_mut()
-                        .reset(Instant::now() + self.republish_interval);
+                match self.publish_current(info).await {
+                    Err(err) => {
+                        failed_attempts += 1;
+                        // Retry after increasing timeout
+                        let retry_after = Duration::from_secs(failed_attempts);
+                        republish.as_mut().reset(Instant::now() + retry_after);
+                        warn!(
+                            err = %format!("{err:#}"),
+                            url = %self.pkarr_client.pkarr_relay_url ,
+                            ?retry_after,
+                            %failed_attempts,
+                            "Failed to publish to pkarr",
+                        );
+                    }
+                    _ => {
+                        failed_attempts = 0;
+                        // Republish after fixed interval
+                        republish
+                            .as_mut()
+                            .reset(Instant::now() + self.republish_interval);
+                    }
                 }
             }
             // Wait until either the retry/republish timeout is reached, or the node info changed.
diff --git a/iroh/src/endpoint.rs b/iroh/src/endpoint.rs
@@ -908,7 +908,7 @@ impl Endpoint {
     /// # }
     /// ```
     #[cfg(not(wasm_browser))]
-    pub fn node_addr(&self) -> impl n0_watcher::Watcher<Value = Option<NodeAddr>> {
+    pub fn node_addr(&self) -> impl n0_watcher::Watcher<Value = Option<NodeAddr>> + use<> {
         let watch_addrs = self.direct_addresses();
         let watch_relay = self.home_relay();
         let node_id = self.node_id();
@@ -973,7 +973,7 @@ impl Endpoint {
     /// let _relay_url = mep.home_relay().initialized().await.unwrap();
     /// # });
     /// ```
-    pub fn home_relay(&self) -> impl n0_watcher::Watcher<Value = Vec<RelayUrl>> {
+    pub fn home_relay(&self) -> impl n0_watcher::Watcher<Value = Vec<RelayUrl>> + use<> {
         self.msock.home_relay()
     }
 
@@ -1043,7 +1043,7 @@ impl Endpoint {
     /// # });
     /// ```
     #[doc(hidden)]
-    pub fn net_report(&self) -> impl Watcher<Value = Option<Report>> {
+    pub fn net_report(&self) -> impl Watcher<Value = Option<Report>> + use<> {
         self.msock.net_report()
     }
 
@@ -1086,7 +1086,7 @@ impl Endpoint {
     /// connection was ever made or is even possible.
     ///
     /// See also [`Endpoint::remote_info`] to only retrieve information about a single node.
-    pub fn remote_info_iter(&self) -> impl Iterator<Item = RemoteInfo> {
+    pub fn remote_info_iter(&self) -> impl Iterator<Item = RemoteInfo> + use<> {
         self.msock.list_remote_infos().into_iter()
     }
 
@@ -1114,7 +1114,7 @@ impl Endpoint {
     ///
     /// [`MdnsDiscovery`]: crate::discovery::mdns::MdnsDiscovery
     /// [`StaticProvider`]: crate::discovery::static_provider::StaticProvider
-    pub fn discovery_stream(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> {
+    pub fn discovery_stream(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> + use<> {
         self.msock.discovery_subscribers().subscribe()
     }
 
diff --git a/iroh/src/magicsock.rs b/iroh/src/magicsock.rs
@@ -280,7 +280,7 @@ impl MagicSock {
         &self.ip_bind_addrs
     }
 
-    fn ip_local_addrs(&self) -> impl Iterator<Item = SocketAddr> {
+    fn ip_local_addrs(&self) -> impl Iterator<Item = SocketAddr> + use<> {
         self.local_addr()
             .into_iter()
             .filter_map(|addr| addr.into_socket_addr())
@@ -336,7 +336,7 @@ impl MagicSock {
     ///
     /// [`Watcher`]: n0_watcher::Watcher
     /// [`Watcher::initialized`]: n0_watcher::Watcher::initialized
-    pub(crate) fn net_report(&self) -> impl Watcher<Value = Option<Report>> {
+    pub(crate) fn net_report(&self) -> impl Watcher<Value = Option<Report>> + use<> {
         self.net_report
             .watch()
             .map(|(r, _)| r)
@@ -347,7 +347,7 @@ impl MagicSock {
     ///
     /// Note that this can be used to wait for the initial home relay to be known using
     /// [`Watcher::initialized`].
-    pub(crate) fn home_relay(&self) -> impl Watcher<Value = Vec<RelayUrl>> {
+    pub(crate) fn home_relay(&self) -> impl Watcher<Value = Vec<RelayUrl>> + use<> {
         let res = self.local_addrs_watch.clone().map(|addrs| {
             addrs
                 .into_iter()
diff --git a/iroh/src/magicsock/transports.rs b/iroh/src/magicsock/transports.rs
@@ -2,7 +2,7 @@ use std::{
     io::{self, IoSliceMut},
     net::{IpAddr, Ipv6Addr, SocketAddr, SocketAddrV6},
     pin::Pin,
-    sync::{atomic::AtomicUsize, Arc},
+    sync::{Arc, atomic::AtomicUsize},
     task::{Context, Poll},
 };
 
@@ -420,7 +420,7 @@ impl UdpSender {
         match destination {
             #[cfg(wasm_browser)]
             Addr::Ip(..) => {
-                return Poll::Ready(Err(io::Error::other("IP is unsupported in browser")))
+                return Poll::Ready(Err(io::Error::other("IP is unsupported in browser")));
             }
             #[cfg(not(wasm_browser))]
             Addr::Ip(addr) => {
diff --git a/iroh/src/magicsock/transports/relay/actor.rs b/iroh/src/magicsock/transports/relay/actor.rs
@@ -500,7 +500,7 @@ impl ActiveRelayActor {
     /// connections.  It currently does not ever return `Err` as the retries continue
     /// forever.
     // This is using `impl Future` to return a future without a reference to self.
-    fn dial_relay(&self) -> impl Future<Output = Result<Client, DialError>> {
+    fn dial_relay(&self) -> impl Future<Output = Result<Client, DialError>> + use<> {
         let client_builder = self.relay_client_builder.clone();
         async move {
             match time::timeout(CONNECT_TIMEOUT, client_builder.connect()).await {
@@ -981,7 +981,10 @@ impl RelayActor {
     /// If the datagram can not be sent immediately, because the destination channel is
     /// full, a future is returned that will complete once the datagrams have been sent to
     /// the [`ActiveRelayActor`].
-    async fn try_send_datagram(&mut self, item: RelaySendItem) -> Option<impl Future<Output = ()>> {
+    async fn try_send_datagram(
+        &mut self,
+        item: RelaySendItem,
+    ) -> Option<impl Future<Output = ()> + use<>> {
         let url = item.url.clone();
         let handle = self
             .active_relay_handle_for_node(&item.url, &item.remote_node)
@@ -1216,7 +1219,7 @@ impl RelayActor {
         });
     }
 
-    fn active_relay_sorted(&self) -> impl Iterator<Item = RelayUrl> {
+    fn active_relay_sorted(&self) -> impl Iterator<Item = RelayUrl> + use<> {
         let mut ids: Vec<_> = self.active_relays.keys().cloned().collect();
         ids.sort();
 
diff --git a/iroh/src/net_report.rs b/iroh/src/net_report.rs
@@ -160,7 +160,7 @@ impl QadConns {
         reports
     }
 
-    fn watch_v4(&self) -> impl n0_future::Stream<Item = Option<QadProbeReport>> + Unpin {
+    fn watch_v4(&self) -> impl n0_future::Stream<Item = Option<QadProbeReport>> + Unpin + use<> {
         let watcher = self.v4.as_ref().map(|(_url, conn)| conn.observer.watch());
 
         if let Some(watcher) = watcher {
@@ -170,7 +170,7 @@ impl QadConns {
         }
     }
 
-    fn watch_v6(&self) -> impl n0_future::Stream<Item = Option<QadProbeReport>> + Unpin {
+    fn watch_v6(&self) -> impl n0_future::Stream<Item = Option<QadProbeReport>> + Unpin + use<> {
         let watcher = self.v6.as_ref().map(|(_url, conn)| conn.observer.watch());
         if let Some(watcher) = watcher {
             watcher.stream_updates_only().boxed()
diff --git a/iroh/src/test_utils.rs b/iroh/src/test_utils.rs
@@ -483,7 +483,7 @@ pub(crate) mod pkarr_dns_state {
         node_info: &NodeInfo,
         origin: &str,
         ttl: u32,
-    ) -> impl Iterator<Item = hickory_resolver::proto::rr::Record> + 'static {
+    ) -> impl Iterator<Item = hickory_resolver::proto::rr::Record> + 'static + use<> {
         let txt_strings = node_info.to_txt_strings();
         let records = to_hickory_records(txt_strings, node_info.node_id, origin, ttl);
         records.collect::<Vec<_>>().into_iter()

Original file line number	Diff line number	Diff line change
`@@ -173,10 +173,10 @@ impl AsyncRead for MaybeTlsStream {`
`173`	`173`	`buf: &mut tokio::io::ReadBuf<'_>,`
`174`	`174`	`) -> Poll<std::io::Result<()>> {`
`175`	`175`	`match &mut *self {`
`176`		`- MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_read(cx, buf),`
`177`		`- MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_read(cx, buf),`
	`176`	`+ MaybeTlsStream::Plain(s) => Pin::new(s).poll_read(cx, buf),`
	`177`	`+ MaybeTlsStream::Tls(s) => Pin::new(s).poll_read(cx, buf),`
`178`	`178`	`#[cfg(test)]`
`179`		`- MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_read(cx, buf),`
	`179`	`+ MaybeTlsStream::Test(s) => Pin::new(s).poll_read(cx, buf),`
`180`	`180`	`}`
`181`	`181`	`}`
`182`	`182`	`}`
`@@ -187,10 +187,10 @@ impl AsyncWrite for MaybeTlsStream {`
`187`	`187`	`cx: &mut Context<'_>,`
`188`	`188`	`) -> Poll<std::result::Result<(), std::io::Error>> {`
`189`	`189`	`match &mut *self {`
`190`		`- MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_flush(cx),`
`191`		`- MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_flush(cx),`
	`190`	`+ MaybeTlsStream::Plain(s) => Pin::new(s).poll_flush(cx),`
	`191`	`+ MaybeTlsStream::Tls(s) => Pin::new(s).poll_flush(cx),`
`192`	`192`	`#[cfg(test)]`
`193`		`- MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_flush(cx),`
	`193`	`+ MaybeTlsStream::Test(s) => Pin::new(s).poll_flush(cx),`
`194`	`194`	`}`
`195`	`195`	`}`
`196`	`196`
`@@ -199,10 +199,10 @@ impl AsyncWrite for MaybeTlsStream {`
`199`	`199`	`cx: &mut Context<'_>,`
`200`	`200`	`) -> Poll<std::result::Result<(), std::io::Error>> {`
`201`	`201`	`match &mut *self {`
`202`		`- MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_shutdown(cx),`
`203`		`- MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_shutdown(cx),`
	`202`	`+ MaybeTlsStream::Plain(s) => Pin::new(s).poll_shutdown(cx),`
	`203`	`+ MaybeTlsStream::Tls(s) => Pin::new(s).poll_shutdown(cx),`
`204`	`204`	`#[cfg(test)]`
`205`		`- MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_shutdown(cx),`
	`205`	`+ MaybeTlsStream::Test(s) => Pin::new(s).poll_shutdown(cx),`
`206`	`206`	`}`
`207`	`207`	`}`
`208`	`208`
`@@ -212,10 +212,10 @@ impl AsyncWrite for MaybeTlsStream {`
`212`	`212`	`buf: &[u8],`
`213`	`213`	`) -> Poll<std::result::Result<usize, std::io::Error>> {`
`214`	`214`	`match &mut *self {`
`215`		`- MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_write(cx, buf),`
`216`		`- MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_write(cx, buf),`
	`215`	`+ MaybeTlsStream::Plain(s) => Pin::new(s).poll_write(cx, buf),`
	`216`	`+ MaybeTlsStream::Tls(s) => Pin::new(s).poll_write(cx, buf),`
`217`	`217`	`#[cfg(test)]`
`218`		`- MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_write(cx, buf),`
	`218`	`+ MaybeTlsStream::Test(s) => Pin::new(s).poll_write(cx, buf),`
`219`	`219`	`}`
`220`	`220`	`}`
`221`	`221`
`@@ -225,10 +225,10 @@ impl AsyncWrite for MaybeTlsStream {`
`225`	`225`	`bufs: &[std::io::IoSlice<'_>],`
`226`	`226`	`) -> Poll<std::result::Result<usize, std::io::Error>> {`
`227`	`227`	`match &mut *self {`
`228`		`- MaybeTlsStream::Plain(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),`
`229`		`- MaybeTlsStream::Tls(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),`
	`228`	`+ MaybeTlsStream::Plain(s) => Pin::new(s).poll_write_vectored(cx, bufs),`
	`229`	`+ MaybeTlsStream::Tls(s) => Pin::new(s).poll_write_vectored(cx, bufs),`
`230`	`230`	`#[cfg(test)]`
`231`		`- MaybeTlsStream::Test(ref mut s) => Pin::new(s).poll_write_vectored(cx, bufs),`
	`231`	`+ MaybeTlsStream::Test(s) => Pin::new(s).poll_write_vectored(cx, bufs),`
`232`	`232`	`}`
`233`	`233`	`}`
`234`	`234`	`}`
Original file line number	Diff line number	Diff line change
`@@ -699,7 +699,7 @@ impl DiscoverySubscribers {`
`699`	`699`	`}`
`700`	`700`	`}`
`701`	`701`
`702`		`- pub(crate) fn subscribe(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> {`
	`702`	`+ pub(crate) fn subscribe(&self) -> impl Stream<Item = Result<DiscoveryItem, Lagged>> + use<> {`
`703`	`703`	`use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream};`
`704`	`704`	`let recv = self.inner.subscribe();`
`705`	`705`	`BroadcastStream::new(recv).map_err(\|BroadcastStreamRecvError::Lagged(n)\| Lagged { val: n })`