feat(logging): instrument spawned task with current span

RolandSherwin · RolandSherwin · commit d0cd74a15d6c · 2025-06-19T02:44:45.000+05:30
diff --git a/ant-networking/src/cmd.rs b/ant-networking/src/cmd.rs
@@ -35,6 +35,7 @@ use std::{
     time::Duration,
 };
 use tokio::sync::oneshot;
+use tracing::Instrument;
 use xor_name::XorName;
 
 const MAX_CONTINUOUS_HDD_WRITE_ERROR: usize = 5;
@@ -975,7 +976,7 @@ impl SwarmDriver {
                             error!("Failed to get response from one shot channel for Cmd::PeerConsideredAsBad : {err:?}");
                         }
                     }
-                });
+                }.instrument(tracing::Span::current()));
 
                 // request
                 let request = Request::Cmd(Cmd::PeerConsideredAsBad {
diff --git a/ant-networking/src/driver.rs b/ant-networking/src/driver.rs
@@ -48,6 +48,7 @@ use std::{
 };
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio::time::Duration;
+use tracing::Instrument;
 use tracing::warn;
 
 /// 10 is the max number of issues per node we track to avoid mem leaks
@@ -366,7 +367,7 @@ impl SwarmDriver {
                         if let Err(err) = old_cache.sync_and_flush_to_disk() {
                             error!("Failed to save bootstrap cache: {err}");
                         }
-                    });
+                    }.instrument(tracing::Span::current()));
 
                     if current_interval.period() >= bootstrap_cache.config().max_cache_save_duration {
                         continue;
@@ -412,7 +413,7 @@ impl SwarmDriver {
             if let Err(error) = event_sender.send(event).await {
                 error!("SwarmDriver failed to send event: {}", error);
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     /// Sends an event after pushing it off thread so as to be non-blocking
@@ -432,7 +433,7 @@ impl SwarmDriver {
             if let Err(error) = event_sender.send(event).await {
                 error!("SwarmDriver failed to send event: {}", error);
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     /// Get K closest peers to self, from our local RoutingTable.
diff --git a/ant-networking/src/event/kad.rs b/ant-networking/src/event/kad.rs
@@ -12,6 +12,7 @@ use crate::{
 };
 use libp2p::kad::{self, GetClosestPeersError, InboundRequest, QueryResult, K_VALUE};
 use std::collections::hash_map::Entry;
+use tracing::Instrument;
 
 impl SwarmDriver {
     pub(super) fn handle_kad_event(&mut self, kad_event: libp2p::kad::Event) -> Result<()> {
@@ -104,7 +105,7 @@ impl SwarmDriver {
                     PendingGetClosestType::FunctionCall(sender) => {
                         tokio::spawn(async move {
                             let _ = sender.send(vec![]);
-                        });
+                        }.instrument(tracing::Span::current()));
                     }
                 }
             }
diff --git a/ant-networking/src/event/swarm.rs b/ant-networking/src/event/swarm.rs
@@ -24,6 +24,7 @@ use libp2p::{
     Multiaddr, TransportError,
 };
 use tokio::time::Duration;
+use tracing::Instrument;
 
 impl SwarmDriver {
     /// Handle `SwarmEvents`
@@ -192,7 +193,7 @@ impl SwarmDriver {
                                     if let Err(err) = old_cache.sync_and_flush_to_disk() {
                                         error!("Failed to save bootstrap cache: {err}");
                                     }
-                                });
+                                }.instrument(tracing::Span::current()));
                             }
                         }
                     } else if let Some(external_add_manager) =
diff --git a/ant-networking/src/lib.rs b/ant-networking/src/lib.rs
@@ -71,6 +71,7 @@ use tokio::sync::{
     mpsc::{self, Sender},
     oneshot,
 };
+use tracing::Instrument;
 
 /// Majority of a given group (i.e. > 1/2).
 #[inline]
@@ -694,7 +695,7 @@ pub(crate) fn send_local_swarm_cmd(swarm_cmd_sender: Sender<LocalSwarmCmd>, cmd:
         if let Err(error) = swarm_cmd_sender.send(cmd).await {
             error!("Failed to send SwarmCmd: {}", error);
         }
-    });
+    }.instrument(tracing::Span::current()));
 }
 
 pub(crate) fn send_network_swarm_cmd(
@@ -715,5 +716,5 @@ pub(crate) fn send_network_swarm_cmd(
         if let Err(error) = swarm_cmd_sender.send(cmd).await {
             error!("Failed to send SwarmCmd: {}", error);
         }
-    });
+    }.instrument(tracing::Span::current()));
 }
diff --git a/ant-networking/src/metrics/bad_node.rs b/ant-networking/src/metrics/bad_node.rs
@@ -18,6 +18,7 @@ use std::{
     time::{Duration, Instant},
 };
 use strum::IntoEnumIterator;
+use tracing::Instrument;
 
 const UPDATE_INTERVAL: Duration = Duration::from_secs(20);
 
@@ -158,7 +159,7 @@ impl BadNodeMetrics {
                     }
                 }
             }
-        });
+        }.instrument(tracing::Span::current()));
         tx
     }
 }
diff --git a/ant-networking/src/metrics/mod.rs b/ant-networking/src/metrics/mod.rs
@@ -27,6 +27,7 @@ use std::collections::HashMap;
 use std::sync::atomic::AtomicU64;
 use sysinfo::{Pid, ProcessRefreshKind, System};
 use tokio::time::Duration;
+use tracing::Instrument;
 
 const UPDATE_INTERVAL: Duration = Duration::from_secs(60);
 const TO_MB: u64 = 1_000_000;
@@ -319,7 +320,7 @@ impl NetworkMetricsRecorder {
                 }
                 sleep(UPDATE_INTERVAL).await;
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     // Records the metric
@@ -339,7 +340,7 @@ impl NetworkMetricsRecorder {
                     {
                         error!("Failed to send shunned report via notifier: {err:?}");
                     }
-                });
+                }.instrument(tracing::Span::current()));
             }
             Marker::QuotingMetrics { quoting_metrics } => {
                 let _ = self.relevant_records.set(
@@ -370,7 +371,7 @@ impl NetworkMetricsRecorder {
             {
                 error!("Failed to send shunned report via notifier: {err:?}");
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     pub(crate) fn update_node_versions(&self, versions: &HashMap<PeerId, String>) {
diff --git a/ant-networking/src/metrics/service.rs b/ant-networking/src/metrics/service.rs
@@ -15,6 +15,7 @@ use std::{
     sync::{Arc, Mutex},
     task::{Context, Poll},
 };
+use tracing::Instrument;
 
 /// The types of metrics that are exposed via the various endpoints.
 #[derive(Default, Debug)]
@@ -41,7 +42,7 @@ pub(crate) fn run_metrics_server(registries: MetricsRegistries, port: u16) {
         if let Err(e) = server.await {
             error!("server error: {}", e);
         }
-    });
+    }.instrument(tracing::Span::current()));
 }
 
 type SharedRegistry = Arc<Mutex<Registry>>;
diff --git a/ant-networking/src/record_store.rs b/ant-networking/src/record_store.rs
@@ -44,6 +44,7 @@ use std::{
     vec,
 };
 use tokio::{sync::mpsc, time::Duration};
+use tracing::Instrument;
 use walkdir::{DirEntry, WalkDir};
 use xor_name::XorName;
 
@@ -358,7 +359,7 @@ impl NodeRecordStore {
                 let mut serialiser = rmp_serde::encode::Serializer::new(&mut file);
                 let _ = historic_quoting_metrics.serialize(&mut serialiser);
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     /// Creates a new `DiskBackedStore` with the given configuration.
@@ -750,7 +751,7 @@ impl NodeRecordStore {
 
                 send_local_swarm_cmd(cloned_cmd_sender, cmd);
             }
-        });
+        }.instrument(tracing::Span::current()));
 
         Ok(())
     }
@@ -925,7 +926,7 @@ impl RecordStore for NodeRecordStore {
             {
                 error!("SwarmDriver failed to send event: {}", error);
             }
-        });
+        }.instrument(tracing::Span::current()));
 
         Ok(())
     }
@@ -962,7 +963,7 @@ impl RecordStore for NodeRecordStore {
                     error!("Error while removing file. filename: {filename}, error: {err:?}");
                 }
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 
     fn records(&self) -> Self::RecordsIter<'_> {
diff --git a/ant-networking/src/replication_fetcher.rs b/ant-networking/src/replication_fetcher.rs
@@ -19,6 +19,7 @@ use libp2p::{
 };
 use std::collections::{hash_map::Entry, BTreeMap, HashMap, HashSet, VecDeque};
 use tokio::{sync::mpsc, time::Duration};
+use tracing::Instrument;
 
 // Max parallel fetches that can be undertaken at the same time.
 const MAX_PARALLEL_FETCH: usize = 5;
@@ -589,7 +590,7 @@ impl ReplicationFetcher {
             if let Err(error) = event_sender.send(event).await {
                 error!("ReplicationFetcher failed to send event: {}", error);
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 }
 
diff --git a/ant-node/src/bin/antnode/main.rs b/ant-node/src/bin/antnode/main.rs
@@ -42,6 +42,7 @@ use tokio::{
     sync::{broadcast::error::RecvError, mpsc},
     time::sleep,
 };
+use tracing::Instrument;
 use tracing_appender::non_blocking::WorkerGuard;
 
 #[derive(Debug, Clone)]
@@ -417,7 +418,7 @@ You can check your reward balance by running:
         {
             error!("Failed to send node control msg to antnode bin main thread: {err}");
         }
-    });
+    }.instrument(tracing::Span::current()));
 
     // Start up gRPC interface if enabled by user
     if let Some(addr) = rpc {
diff --git a/ant-node/src/bin/antnode/rpc_service.rs b/ant-node/src/bin/antnode/rpc_service.rs
@@ -26,6 +26,7 @@ use std::{
 };
 use tokio::sync::mpsc::{self, Sender};
 use tokio_stream::wrappers::ReceiverStream;
+use tracing::Instrument;
 use tonic::{transport::Server, Code, Request, Response, Status};
 use tracing::{debug, info};
 
@@ -132,7 +133,7 @@ impl AntNode for SafeNodeRpcService {
                     break;
                 }
             }
-        });
+        }.instrument(tracing::Span::current()));
 
         Ok(Response::new(ReceiverStream::new(client_rx)))
     }
@@ -316,5 +317,5 @@ pub(crate) fn start_rpc_service(
         {
             error!("RPC Server failed to start: {e:?}");
         }
-    });
+    }.instrument(tracing::Span::current()));
 }
diff --git a/ant-node/src/node.rs b/ant-node/src/node.rs
@@ -356,7 +356,7 @@ impl Node {
                         let _handle = spawn(async move {
                             Self::try_interval_replication(network);
                             trace!("Periodic replication took {:?}", start.elapsed());
-                        });
+                        }.instrument(tracing::Span::current()));
                     }
                     _ = uptime_metrics_update_interval.tick() => {
                         #[cfg(feature = "open-metrics")]
diff --git a/ant-node/src/replication.rs b/ant-node/src/replication.rs
@@ -19,6 +19,7 @@ use libp2p::{
     PeerId,
 };
 use tokio::task::spawn;
+use tracing::Instrument;
 
 impl Node {
     /// Sends _all_ record keys every interval to all peers within the REPLICATE_RANGE.
@@ -82,7 +83,7 @@ impl Node {
                 } else {
                     debug!("Completed storing Replication Record {pretty_key:?} from holder {holder:?}.");
                 }
-            });
+            }.instrument(tracing::Span::current()));
         }
         Ok(())
     }
@@ -228,6 +229,6 @@ impl Node {
                 node.network()
                     .add_fresh_records_to_the_replication_fetcher(holder, new_keys);
             }
-        });
+        }.instrument(tracing::Span::current()));
     }
 }
diff --git a/autonomi/src/networking/mod.rs b/autonomi/src/networking/mod.rs
@@ -38,6 +38,7 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
+use tracing::Instrument;
 
 /// Result type for tasks responses sent by the [`crate::driver::NetworkDriver`] to the [`crate::Network`]
 pub(in crate::networking) type OneShotTaskResult<T> = oneshot::Sender<Result<T, NetworkError>>;
@@ -138,7 +139,7 @@ impl Network {
         // run the network driver in a background task
         tokio::spawn(async move {
             let _ = driver.run().await;
-        });
+        }.instrument(tracing::Span::current()));
 
         let network = Self {
             task_sender: Arc::new(task_sender),

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ use std::{`
`35`	`35`	`time::Duration,`
`36`	`36`	`};`
`37`	`37`	`use tokio::sync::oneshot;`
	`38`	`+use tracing::Instrument;`
`38`	`39`	`use xor_name::XorName;`
`39`	`40`
`40`	`41`	`const MAX_CONTINUOUS_HDD_WRITE_ERROR: usize = 5;`
`@@ -975,7 +976,7 @@ impl SwarmDriver {`
`975`	`976`	`error!("Failed to get response from one shot channel for Cmd::PeerConsideredAsBad : {err:?}");`
`976`	`977`	`}`
`977`	`978`	`}`
`978`		`- });`
	`979`	`+ }.instrument(tracing::Span::current()));`
`979`	`980`
`980`	`981`	`// request`
`981`	`982`	`let request = Request::Cmd(Cmd::PeerConsideredAsBad {`
Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@ use std::{`
`48`	`48`	`};`
`49`	`49`	`use tokio::sync::{mpsc, oneshot, watch};`
`50`	`50`	`use tokio::time::Duration;`
	`51`	`+use tracing::Instrument;`
`51`	`52`	`use tracing::warn;`
`52`	`53`
`53`	`54`	`/// 10 is the max number of issues per node we track to avoid mem leaks`
`@@ -366,7 +367,7 @@ impl SwarmDriver {`
`366`	`367`	`if let Err(err) = old_cache.sync_and_flush_to_disk() {`
`367`	`368`	`error!("Failed to save bootstrap cache: {err}");`
`368`	`369`	`}`
`369`		`- });`
	`370`	`+ }.instrument(tracing::Span::current()));`
`370`	`371`
`371`	`372`	`if current_interval.period() >= bootstrap_cache.config().max_cache_save_duration {`
`372`	`373`	`continue;`
`@@ -412,7 +413,7 @@ impl SwarmDriver {`
`412`	`413`	`if let Err(error) = event_sender.send(event).await {`
`413`	`414`	`error!("SwarmDriver failed to send event: {}", error);`
`414`	`415`	`}`
`415`		`- });`
	`416`	`+ }.instrument(tracing::Span::current()));`
`416`	`417`	`}`
`417`	`418`
`418`	`419`	`/// Sends an event after pushing it off thread so as to be non-blocking`
`@@ -432,7 +433,7 @@ impl SwarmDriver {`
`432`	`433`	`if let Err(error) = event_sender.send(event).await {`
`433`	`434`	`error!("SwarmDriver failed to send event: {}", error);`
`434`	`435`	`}`
`435`		`- });`
	`436`	`+ }.instrument(tracing::Span::current()));`
`436`	`437`	`}`
`437`	`438`
`438`	`439`	`/// Get K closest peers to self, from our local RoutingTable.`
Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,7 @@ use crate::{`
`12`	`12`	`};`
`13`	`13`	`use libp2p::kad::{self, GetClosestPeersError, InboundRequest, QueryResult, K_VALUE};`
`14`	`14`	`use std::collections::hash_map::Entry;`
	`15`	`+use tracing::Instrument;`
`15`	`16`
`16`	`17`	`impl SwarmDriver {`
`17`	`18`	`pub(super) fn handle_kad_event(&mut self, kad_event: libp2p::kad::Event) -> Result<()> {`
`@@ -104,7 +105,7 @@ impl SwarmDriver {`
`104`	`105`	`PendingGetClosestType::FunctionCall(sender) => {`
`105`	`106`	`tokio::spawn(async move {`
`106`	`107`	`let _ = sender.send(vec![]);`
`107`		`- });`
	`108`	`+ }.instrument(tracing::Span::current()));`
`108`	`109`	`}`
`109`	`110`	`}`
`110`	`111`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,7 @@ use libp2p::{`
`24`	`24`	`Multiaddr, TransportError,`
`25`	`25`	`};`
`26`	`26`	`use tokio::time::Duration;`
	`27`	`+use tracing::Instrument;`
`27`	`28`
`28`	`29`	`impl SwarmDriver {`
`29`	`30`	/// Handle `SwarmEvents`
`@@ -192,7 +193,7 @@ impl SwarmDriver {`
`192`	`193`	`if let Err(err) = old_cache.sync_and_flush_to_disk() {`
`193`	`194`	`error!("Failed to save bootstrap cache: {err}");`
`194`	`195`	`}`
`195`		`- });`
	`196`	`+ }.instrument(tracing::Span::current()));`
`196`	`197`	`}`
`197`	`198`	`}`
`198`	`199`	`} else if let Some(external_add_manager) =`
Original file line number	Diff line number	Diff line change
`@@ -71,6 +71,7 @@ use tokio::sync::{`
`71`	`71`	`mpsc::{self, Sender},`
`72`	`72`	`oneshot,`
`73`	`73`	`};`
	`74`	`+use tracing::Instrument;`
`74`	`75`
`75`	`76`	`/// Majority of a given group (i.e. > 1/2).`
`76`	`77`	`#[inline]`
`@@ -694,7 +695,7 @@ pub(crate) fn send_local_swarm_cmd(swarm_cmd_sender: Sender<LocalSwarmCmd>, cmd:`
`694`	`695`	`if let Err(error) = swarm_cmd_sender.send(cmd).await {`
`695`	`696`	`error!("Failed to send SwarmCmd: {}", error);`
`696`	`697`	`}`
`697`		`- });`
	`698`	`+ }.instrument(tracing::Span::current()));`
`698`	`699`	`}`
`699`	`700`
`700`	`701`	`pub(crate) fn send_network_swarm_cmd(`
`@@ -715,5 +716,5 @@ pub(crate) fn send_network_swarm_cmd(`
`715`	`716`	`if let Err(error) = swarm_cmd_sender.send(cmd).await {`
`716`	`717`	`error!("Failed to send SwarmCmd: {}", error);`
`717`	`718`	`}`
`718`		`- });`
	`719`	`+ }.instrument(tracing::Span::current()));`
`719`	`720`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@ use std::{`
`18`	`18`	`time::{Duration, Instant},`
`19`	`19`	`};`
`20`	`20`	`use strum::IntoEnumIterator;`
	`21`	`+use tracing::Instrument;`
`21`	`22`
`22`	`23`	`const UPDATE_INTERVAL: Duration = Duration::from_secs(20);`
`23`	`24`
`@@ -158,7 +159,7 @@ impl BadNodeMetrics {`
`158`	`159`	`}`
`159`	`160`	`}`
`160`	`161`	`}`
`161`		`- });`
	`162`	`+ }.instrument(tracing::Span::current()));`
`162`	`163`	`tx`
`163`	`164`	`}`
`164`	`165`	`}`
Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,7 @@ use std::collections::HashMap;`
`27`	`27`	`use std::sync::atomic::AtomicU64;`
`28`	`28`	`use sysinfo::{Pid, ProcessRefreshKind, System};`
`29`	`29`	`use tokio::time::Duration;`
	`30`	`+use tracing::Instrument;`
`30`	`31`
`31`	`32`	`const UPDATE_INTERVAL: Duration = Duration::from_secs(60);`
`32`	`33`	`const TO_MB: u64 = 1_000_000;`
`@@ -319,7 +320,7 @@ impl NetworkMetricsRecorder {`
`319`	`320`	`}`
`320`	`321`	`sleep(UPDATE_INTERVAL).await;`
`321`	`322`	`}`
`322`		`- });`
	`323`	`+ }.instrument(tracing::Span::current()));`
`323`	`324`	`}`
`324`	`325`
`325`	`326`	`// Records the metric`
`@@ -339,7 +340,7 @@ impl NetworkMetricsRecorder {`
`339`	`340`	`{`
`340`	`341`	`error!("Failed to send shunned report via notifier: {err:?}");`
`341`	`342`	`}`
`342`		`- });`
	`343`	`+ }.instrument(tracing::Span::current()));`
`343`	`344`	`}`
`344`	`345`	`Marker::QuotingMetrics { quoting_metrics } => {`
`345`	`346`	`let _ = self.relevant_records.set(`
`@@ -370,7 +371,7 @@ impl NetworkMetricsRecorder {`
`370`	`371`	`{`
`371`	`372`	`error!("Failed to send shunned report via notifier: {err:?}");`
`372`	`373`	`}`
`373`		`- });`
	`374`	`+ }.instrument(tracing::Span::current()));`
`374`	`375`	`}`
`375`	`376`
`376`	`377`	`pub(crate) fn update_node_versions(&self, versions: &HashMap<PeerId, String>) {`
Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,7 @@ use std::{`
`15`	`15`	`sync::{Arc, Mutex},`
`16`	`16`	`task::{Context, Poll},`
`17`	`17`	`};`
	`18`	`+use tracing::Instrument;`
`18`	`19`
`19`	`20`	`/// The types of metrics that are exposed via the various endpoints.`
`20`	`21`	`#[derive(Default, Debug)]`
`@@ -41,7 +42,7 @@ pub(crate) fn run_metrics_server(registries: MetricsRegistries, port: u16) {`
`41`	`42`	`if let Err(e) = server.await {`
`42`	`43`	`error!("server error: {}", e);`
`43`	`44`	`}`
`44`		`- });`
	`45`	`+ }.instrument(tracing::Span::current()));`
`45`	`46`	`}`
`46`	`47`
`47`	`48`	`type SharedRegistry = Arc<Mutex<Registry>>;`
Original file line number	Diff line number	Diff line change
`@@ -44,6 +44,7 @@ use std::{`
`44`	`44`	`vec,`
`45`	`45`	`};`
`46`	`46`	`use tokio::{sync::mpsc, time::Duration};`
	`47`	`+use tracing::Instrument;`
`47`	`48`	`use walkdir::{DirEntry, WalkDir};`
`48`	`49`	`use xor_name::XorName;`
`49`	`50`
`@@ -358,7 +359,7 @@ impl NodeRecordStore {`
`358`	`359`	`let mut serialiser = rmp_serde::encode::Serializer::new(&mut file);`
`359`	`360`	`let _ = historic_quoting_metrics.serialize(&mut serialiser);`
`360`	`361`	`}`
`361`		`- });`
	`362`	`+ }.instrument(tracing::Span::current()));`
`362`	`363`	`}`
`363`	`364`
`364`	`365`	/// Creates a new `DiskBackedStore` with the given configuration.
`@@ -750,7 +751,7 @@ impl NodeRecordStore {`
`750`	`751`
`751`	`752`	`send_local_swarm_cmd(cloned_cmd_sender, cmd);`
`752`	`753`	`}`
`753`		`- });`
	`754`	`+ }.instrument(tracing::Span::current()));`
`754`	`755`
`755`	`756`	`Ok(())`
`756`	`757`	`}`
`@@ -925,7 +926,7 @@ impl RecordStore for NodeRecordStore {`
`925`	`926`	`{`
`926`	`927`	`error!("SwarmDriver failed to send event: {}", error);`
`927`	`928`	`}`
`928`		`- });`
	`929`	`+ }.instrument(tracing::Span::current()));`
`929`	`930`
`930`	`931`	`Ok(())`
`931`	`932`	`}`
`@@ -962,7 +963,7 @@ impl RecordStore for NodeRecordStore {`
`962`	`963`	`error!("Error while removing file. filename: {filename}, error: {err:?}");`
`963`	`964`	`}`
`964`	`965`	`}`
`965`		`- });`
	`966`	`+ }.instrument(tracing::Span::current()));`
`966`	`967`	`}`
`967`	`968`
`968`	`969`	`fn records(&self) -> Self::RecordsIter<'_> {`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@ use libp2p::{`
`19`	`19`	`};`
`20`	`20`	`use std::collections::{hash_map::Entry, BTreeMap, HashMap, HashSet, VecDeque};`
`21`	`21`	`use tokio::{sync::mpsc, time::Duration};`
	`22`	`+use tracing::Instrument;`
`22`	`23`
`23`	`24`	`// Max parallel fetches that can be undertaken at the same time.`
`24`	`25`	`const MAX_PARALLEL_FETCH: usize = 5;`
`@@ -589,7 +590,7 @@ impl ReplicationFetcher {`
`589`	`590`	`if let Err(error) = event_sender.send(event).await {`
`590`	`591`	`error!("ReplicationFetcher failed to send event: {}", error);`
`591`	`592`	`}`
`592`		`- });`
	`593`	`+ }.instrument(tracing::Span::current()));`
`593`	`594`	`}`
`594`	`595`	`}`
`595`	`596`