refactor: split coder from blobber

Author: prestwich

crates/blobber/src/builder.rs renamed to crates/blobber/src/blobs/builder.rs

@@ -1,13 +1,15 @@
-use crate::{BlobberError, BlobberResult, Blobs, FetchResult};
+use crate::{BlobFetcher, BlobberError, BlobberResult, Blobs, FetchResult};
 use alloy::consensus::{SidecarCoder, SimpleCoder, Transaction as _};
 use alloy::eips::eip7691::MAX_BLOBS_PER_BLOCK_ELECTRA;
 use alloy::eips::merge::EPOCH_SLOTS;
 use alloy::primitives::{B256, Bytes, keccak256};
+use core::fmt;
 use reth::transaction_pool::TransactionPool;
 use reth::{network::cache::LruMap, primitives::Receipt};
 use signet_extract::ExtractedEvent;
 use signet_zenith::Zenith::BlockSubmitted;
 use signet_zenith::ZenithBlock;
+use std::marker::PhantomData;
 use std::{
     sync::{Arc, Mutex},
     time::Duration,
@@ -37,11 +39,13 @@ enum CacheInst {
 
 /// Handle for the cache.
 #[derive(Debug, Clone)]
-pub struct CacheHandle {
+pub struct CacheHandle<Coder = SimpleCoder> {
     sender: mpsc::Sender<CacheInst>,
+
+    _coder: PhantomData<Coder>,
 }
 
-impl CacheHandle {
+impl<Coder> CacheHandle<Coder> {
     /// Sends a cache instruction.
     async fn send(&self, inst: CacheInst) {
         let _ = self.sender.send(inst).await;
@@ -71,12 +75,14 @@ impl CacheHandle {
 
     /// Fetch the blobs using [`Self::fetch_blobs`] and decode them to get the
     /// Zenith block data using the provided coder.
-    pub async fn fetch_and_decode_with_coder<C: SidecarCoder>(
+    pub async fn fetch_and_decode(
         &self,
         slot: usize,
         extract: &ExtractedEvent<'_, Receipt, BlockSubmitted>,
-        mut coder: C,
-    ) -> BlobberResult<Bytes> {
+    ) -> BlobberResult<Bytes>
+    where
+        Coder: SidecarCoder + Default,
+    {
         let tx_hash = extract.tx_hash();
         let versioned_hashes = extract
             .tx
@@ -87,23 +93,13 @@ impl CacheHandle {
 
         let blobs = self.fetch_blobs(slot, tx_hash, versioned_hashes.to_owned()).await?;
 
-        coder
+        Coder::default()
             .decode_all(blobs.as_ref())
             .ok_or_else(BlobberError::blob_decode_error)?
             .into_iter()
             .find(|data| keccak256(data) == extract.block_data_hash())
             .map(Into::into)
-            .ok_or_else(|| BlobberError::block_data_not_found(tx_hash))
-    }
-
-    /// Fetch the blobs using [`Self::fetch_blobs`] and decode them using
-    /// [`SimpleCoder`] to get the Zenith block data.
-    pub async fn fech_and_decode(
-        &self,
-        slot: usize,
-        extract: &ExtractedEvent<'_, Receipt, BlockSubmitted>,
-    ) -> BlobberResult<Bytes> {
-        self.fetch_and_decode_with_coder(slot, extract, SimpleCoder::default()).await
+            .ok_or_else(|| BlobberError::block_data_not_found(extract.block_data_hash()))
     }
 
     /// Fetch the blobs, decode them using the provided coder, and construct a
@@ -117,15 +113,17 @@ impl CacheHandle {
     ///   decoded (e.g., due to a malformatted blob).
     /// - `Err(FetchError)` if there was an unrecoverable error fetching the
     ///   blobs.
-    pub async fn signet_block_with_coder<C: SidecarCoder>(
+    pub async fn signet_block(
         &self,
         host_block_number: u64,
         slot: usize,
         extract: &ExtractedEvent<'_, Receipt, BlockSubmitted>,
-        coder: C,
-    ) -> FetchResult<ZenithBlock> {
+    ) -> FetchResult<ZenithBlock>
+    where
+        Coder: SidecarCoder + Default,
+    {
         let header = extract.ru_header(host_block_number);
-        let block_data = match self.fetch_and_decode_with_coder(slot, extract, coder).await {
+        let block_data = match self.fetch_and_decode(slot, extract).await {
             Ok(buf) => buf,
             Err(BlobberError::Decode(_)) => {
                 trace!("Failed to decode block data");
@@ -135,44 +133,24 @@ impl CacheHandle {
         };
         Ok(ZenithBlock::from_header_and_data(header, block_data))
     }
-
-    /// Fetch the blobs, decode them using [`SimpleCoder`], and construct a
-    /// Zenith block from the header and data.
-    ///
-    /// # Returns
-    ///
-    /// - `Ok(ZenithBlock)` if the block was successfully fetched and
-    ///   decoded.
-    /// - `Ok(ZenithBlock)` with an EMPTY BLOCK if the block_data could not be
-    ///   decoded (e.g., due to a malformatted blob).
-    /// - `Err(FetchError)` if there was an unrecoverable error fetching the
-    ///   blobs.
-    pub async fn signet_block(
-        &self,
-        host_block_number: u64,
-        slot: usize,
-        extract: &ExtractedEvent<'_, Receipt, BlockSubmitted>,
-    ) -> FetchResult<ZenithBlock> {
-        self.signet_block_with_coder(host_block_number, slot, extract, SimpleCoder::default()).await
-    }
 }
 
 /// Retrieves blobs and stores them in a cache for later use.
 pub struct BlobCacher<Pool> {
-    fetcher: crate::BlobFetcher<Pool>,
+    fetcher: BlobFetcher<Pool>,
 
     cache: Mutex<LruMap<(usize, B256), Blobs>>,
 }
 
-impl<Pool: core::fmt::Debug> core::fmt::Debug for BlobCacher<Pool> {
-    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+impl<Pool: fmt::Debug> fmt::Debug for BlobCacher<Pool> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("BlobCacher").field("fetcher", &self.fetcher).finish_non_exhaustive()
     }
 }
 
 impl<Pool: TransactionPool + 'static> BlobCacher<Pool> {
     /// Creates a new `BlobCacher` with the provided extractor and cache size.
-    pub fn new(fetcher: crate::BlobFetcher<Pool>) -> Self {
+    pub fn new(fetcher: BlobFetcher<Pool>) -> Self {
         Self { fetcher, cache: LruMap::new(BLOB_CACHE_SIZE).into() }
     }
 
@@ -237,10 +215,10 @@ impl<Pool: TransactionPool + 'static> BlobCacher<Pool> {
     ///
     /// # Panics
     /// This function will panic if the cache task fails to spawn.
-    pub fn spawn(self) -> CacheHandle {
+    pub fn spawn<C: SidecarCoder + Default>(self) -> CacheHandle<C> {
         let (sender, inst) = mpsc::channel(CACHE_REQUEST_CHANNEL_SIZE);
         tokio::spawn(Arc::new(self).task_future(inst));
-        CacheHandle { sender }
+        CacheHandle { sender, _coder: PhantomData }
     }
 }
 
@@ -310,7 +288,7 @@ mod tests {
             .unwrap()
             .with_slot_calculator(calc)
             .build_cache()?;
-        let handle = cache.spawn();
+        let handle = cache.spawn::<SimpleCoder>();
 
         let got = handle
             .fetch_blobs(

crates/blobber/src/cache.rs renamed to crates/blobber/src/blobs/cache.rs

@@ -0,0 +1,38 @@
+use alloy::primitives::B256;
+use reth::transaction_pool::BlobStoreError;
+
+/// Result using [`FetchError`] as the default error type.
+pub type FetchResult<T> = Result<T, FetchError>;
+
+/// Unrecoverable blob fetching errors. These result in the node shutting
+/// down. They occur when the blobstore is down or the sidecar is unretrievable.
+#[derive(Debug, thiserror::Error)]
+pub enum FetchError {
+    /// Reqwest error
+    #[error(transparent)]
+    Reqwest(#[from] reqwest::Error),
+    /// Missing sidecar error
+    #[error("Cannot retrieve sidecar for {0} from any source")]
+    MissingSidecar(B256),
+    /// Reth blobstore error.
+    #[error(transparent)]
+    BlobStore(BlobStoreError),
+    /// Url parse error.
+    #[error(transparent)]
+    UrlParse(#[from] url::ParseError),
+    /// Consensus client URL not set error.
+    #[error("Consensus client URL not set")]
+    ConsensusClientUrlNotSet,
+    /// Pylon client URL not set error.
+    #[error("Pylon client URL not set")]
+    PylonClientUrlNotSet,
+}
+
+impl From<BlobStoreError> for FetchError {
+    fn from(err: BlobStoreError) -> Self {
+        match err {
+            BlobStoreError::MissingSidecar(tx) => FetchError::MissingSidecar(tx),
+            _ => FetchError::BlobStore(err),
+        }
+    }
+}

crates/blobber/src/config.rs renamed to crates/blobber/src/blobs/config.rs

@@ -1,5 +1,5 @@
 use crate::{
-    BlobFetcherBuilder, BlobberError, BlobberResult, FetchResult, error::FetchError,
+    BlobFetcherBuilder, BlobberError, BlobberResult, FetchError, FetchResult,
     shim::ExtractableChainShim, utils::extract_blobs_from_bundle,
 };
 use alloy::{

crates/blobber/src/blobs/error.rs

@@ -0,0 +1,14 @@
+mod builder;
+pub use builder::{BlobFetcherBuilder, BuilderError as BlobFetcherBuilderError};
+
+mod cache;
+pub use cache::{BlobCacher, CacheHandle};
+
+mod config;
+pub use config::BlobFetcherConfig;
+
+mod error;
+pub use error::{FetchError, FetchResult};
+
+mod fetch;
+pub use fetch::{BlobFetcher, Blobs};

crates/blobber/src/fetch.rs renamed to crates/blobber/src/blobs/fetch.rs

@@ -0,0 +1,24 @@
+use alloy::{eips::eip2718::Eip2718Error, primitives::B256};
+
+/// Result using [`DecodeError`] as the default error type.
+pub type DecodeResult<T> = Result<T, DecodeError>;
+
+/// Ignorable blob fetching errors. These result in the block being skipped.
+#[derive(Debug, thiserror::Error, Copy, Clone)]
+pub enum DecodeError {
+    /// Incorrect transaction type error
+    #[error("Non-4844 transaction")]
+    Non4844Transaction,
+    /// Decoding error from the internal [`SimpleCoder`]. This indicates the
+    /// blobs are not formatted in the simple coder format.
+    ///
+    /// [`SimpleCoder`]: alloy::consensus::SimpleCoder
+    #[error("Decoding failed")]
+    BlobDecodeError,
+    /// Block data not found in decoded blob
+    #[error("Block data not found in decoded blob. Expected block hash: {0}")]
+    BlockDataNotFound(B256),
+    /// Error while decoding block from blob
+    #[error("Block decode error: {0}")]
+    BlockDecodeError(#[from] Eip2718Error),
+}

crates/blobber/src/blobs/mod.rs

@@ -0,0 +1,5 @@
+mod error;
+pub use error::{DecodeError, DecodeResult};
+
+mod r#trait;
+pub use r#trait::SignetBlockDecoder;

crates/blobber/src/coder/error.rs

@@ -0,0 +1,49 @@
+use crate::{Blobs, DecodeError, DecodeResult};
+use alloy::{
+    consensus::SidecarCoder,
+    primitives::{B256, Bytes, keccak256},
+};
+use signet_zenith::{Zenith, ZenithBlock};
+
+/// A trait for decoding blocks from blob data.
+pub trait SignetBlockDecoder {
+    /// Decodes a block from the given blob bytes.
+    fn decode_block(
+        &mut self,
+        blobs: Blobs,
+        header: Zenith::BlockHeader,
+        data_hash: B256,
+    ) -> DecodeResult<ZenithBlock>;
+
+    /// Decodes a block from the given blob bytes, or returns an empty block.
+    fn decode_block_or_default(
+        &mut self,
+        blobs: Blobs,
+        header: Zenith::BlockHeader,
+        data_hash: B256,
+    ) -> ZenithBlock {
+        self.decode_block(blobs, header, data_hash)
+            .unwrap_or_else(|_| ZenithBlock::from_header_and_data(header, Bytes::new()))
+    }
+}
+
+impl<T> SignetBlockDecoder for T
+where
+    T: SidecarCoder,
+{
+    fn decode_block(
+        &mut self,
+        blobs: Blobs,
+        header: Zenith::BlockHeader,
+        data_hash: B256,
+    ) -> DecodeResult<ZenithBlock> {
+        let block_data = self
+            .decode_all(blobs.as_ref())
+            .ok_or(DecodeError::BlobDecodeError)?
+            .into_iter()
+            .find(|data| keccak256(data) == data_hash)
+            .map(Into::<Bytes>::into)
+            .ok_or(DecodeError::BlockDataNotFound(data_hash))?;
+        Ok(ZenithBlock::from_header_and_data(header, block_data))
+    }
+}

crates/blobber/src/coder/mod.rs

@@ -1,59 +1,10 @@
+use crate::{DecodeError, FetchError};
 use alloy::{eips::eip2718::Eip2718Error, primitives::B256};
 use reth::transaction_pool::BlobStoreError;
 
 /// Result using [`BlobFetcherError`] as the default error type.
 pub type BlobberResult<T, E = BlobberError> = std::result::Result<T, E>;
 
-/// Result using [`FetchError`] as the default error type.
-pub type FetchResult<T> = BlobberResult<T, FetchError>;
-
-/// Result using [`DecodeError`] as the default error type.
-pub type DecodeResult<T> = BlobberResult<T, DecodeError>;
-
-/// Unrecoverable blob fetching errors. These result in the node shutting
-/// down. They occur when the blobstore is down or the sidecar is unretrievable.
-#[derive(Debug, thiserror::Error)]
-pub enum FetchError {
-    /// Reqwest error
-    #[error(transparent)]
-    Reqwest(#[from] reqwest::Error),
-    /// Missing sidecar error
-    #[error("Cannot retrieve sidecar for {0} from any source")]
-    MissingSidecar(B256),
-    /// Reth blobstore error.
-    #[error(transparent)]
-    BlobStore(BlobStoreError),
-    /// Url parse error.
-    #[error(transparent)]
-    UrlParse(#[from] url::ParseError),
-    /// Consensus client URL not set error.
-    #[error("Consensus client URL not set")]
-    ConsensusClientUrlNotSet,
-    /// Pylon client URL not set error.
-    #[error("Pylon client URL not set")]
-    PylonClientUrlNotSet,
-}
-
-/// Ignorable blob fetching errors. These result in the block being skipped.
-#[derive(Debug, thiserror::Error, Copy, Clone)]
-pub enum DecodeError {
-    /// Incorrect transaction type error
-    #[error("Non-4844 transaction")]
-    Non4844Transaction,
-    /// Decoding error from the internal [`SimpleCoder`]. This indicates the
-    /// blobs are not formatted in the simple coder format.
-    ///
-    /// [`SimpleCoder`]: alloy::consensus::SimpleCoder
-    #[error("Decoding failed")]
-    BlobDecodeError,
-    /// Block data not found in decoded blob
-    #[error("Block data not found in decoded blob. Expected block hash: {0}")]
-    BlockDataNotFound(B256),
-    /// Error while decoding block from blob
-    #[error("Block decode error: {0}")]
-    BlockDecodeError(#[from] Eip2718Error),
-}
-
 /// Blob fetching errors
 #[derive(Debug, thiserror::Error)]
 pub enum BlobberError {
@@ -92,8 +43,8 @@ impl BlobberError {
     }
 
     /// Blob decoded, but expected hash not found
-    pub fn block_data_not_found(tx: B256) -> Self {
-        DecodeError::BlockDataNotFound(tx).into()
+    pub fn block_data_not_found(data_hash: B256) -> Self {
+        DecodeError::BlockDataNotFound(data_hash).into()
     }
 
     /// Missing sidecar error
@@ -107,15 +58,6 @@ impl BlobberError {
     }
 }
 
-impl From<BlobStoreError> for FetchError {
-    fn from(err: BlobStoreError) -> Self {
-        match err {
-            BlobStoreError::MissingSidecar(tx) => FetchError::MissingSidecar(tx),
-            _ => FetchError::BlobStore(err),
-        }
-    }
-}
-
 impl From<BlobStoreError> for BlobberError {
     fn from(err: BlobStoreError) -> Self {
         Self::Fetch(err.into())