refactor: improve tracing of blob cache by passing span to task (#32)

Author: prestwich

crates/blobber/src/cache.rs

@@ -13,7 +13,7 @@ use std::{
     time::Duration,
 };
 use tokio::sync::{mpsc, oneshot};
-use tracing::{error, info, instrument, warn};
+use tracing::{Instrument, debug, error, info, instrument};
 
 const BLOB_CACHE_SIZE: u32 = (MAX_BLOBS_PER_BLOCK_ELECTRA * EPOCH_SLOTS) as u32;
 const CACHE_REQUEST_CHANNEL_SIZE: usize = (MAX_BLOBS_PER_BLOCK_ELECTRA * 2) as usize;
@@ -26,7 +26,13 @@ const BETWEEN_RETRIES: Duration = Duration::from_millis(250);
 /// retrieving blobs.
 #[derive(Debug)]
 enum CacheInst {
-    Retrieve { slot: usize, tx_hash: B256, version_hashes: Vec<B256>, resp: oneshot::Sender<Blobs> },
+    Retrieve {
+        slot: usize,
+        tx_hash: B256,
+        version_hashes: Vec<B256>,
+        resp: oneshot::Sender<Blobs>,
+        span: tracing::Span,
+    },
 }
 
 /// Handle for the cache.
@@ -51,7 +57,14 @@ impl CacheHandle {
     ) -> FetchResult<Blobs> {
         let (resp, receiver) = oneshot::channel();
 
-        self.send(CacheInst::Retrieve { slot, tx_hash, version_hashes, resp }).await;
+        self.send(CacheInst::Retrieve {
+            slot,
+            tx_hash,
+            version_hashes,
+            resp,
+            span: tracing::Span::current(),
+        })
+        .await;
 
         receiver.await.map_err(|_| BlobFetcherError::missing_sidecar(tx_hash))
     }
@@ -164,7 +177,7 @@ impl<Pool: TransactionPool + 'static> BlobCacher<Pool> {
                     return Ok(blobs);
                 }
                 Err(BlobFetcherError::Ignorable(e)) => {
-                    warn!(target: "signet_blobber::BlobCacher", attempt, %e, "Blob fetch attempt failed.");
+                    debug!(target: "signet_blobber::BlobCacher", attempt, %e, "Blob fetch attempt failed.");
                     tokio::time::sleep(BETWEEN_RETRIES).await;
                     continue;
                 }
@@ -178,8 +191,10 @@ impl<Pool: TransactionPool + 'static> BlobCacher<Pool> {
     /// Processes the cache instructions.
     async fn handle_inst(self: Arc<Self>, inst: CacheInst) {
         match inst {
-            CacheInst::Retrieve { slot, tx_hash, version_hashes, resp } => {
-                if let Ok(blobs) = self.fetch_blobs(slot, tx_hash, version_hashes).await {
+            CacheInst::Retrieve { slot, tx_hash, version_hashes, resp, span } => {
+                if let Ok(blobs) =
+                    self.fetch_blobs(slot, tx_hash, version_hashes).instrument(span).await
+                {
                     // if listener has gone away, that's okay, we just won't send the response
                     let _ = resp.send(blobs);
                 }

crates/blobber/src/fetch.rs

@@ -16,7 +16,7 @@ use signet_extract::{ExtractedEvent, Extracts};
 use signet_zenith::{Zenith::BlockSubmitted, ZenithBlock};
 use std::{ops::Deref, sync::Arc};
 use tokio::select;
-use tracing::{error, instrument, trace};
+use tracing::{instrument, trace};
 
 /// Blobs which may be a local shared sidecar, or a list of blobs from an
 /// external source.
@@ -161,7 +161,7 @@ where
     }
 
     /// Fetch blobs from the local txpool, or fall back to remote sources
-    #[instrument(skip(self, versioned_hashes))]
+    #[instrument(skip(self))]
     pub(crate) async fn fetch_blobs(
         &self,
         slot: usize,
@@ -185,7 +185,6 @@ where
                 Ok(blobs)
             }
             else => {
-                error!(%tx_hash, "Blobs not available from any source");
                 Err(BlobFetcherError::missing_sidecar(tx_hash))
             }
         }