refactor(taiko-client-rs): consolidate shared assembly and collapse single-impl abstractions (#21776)

Author: davidtaikocha

packages/taiko-client-rs/Cargo.lock

@@ -1,12 +1,5 @@
-use alloy_consensus::BlobTransactionSidecar;
-use alloy_primitives::B256;
-use async_trait::async_trait;
-use rpc::{
-    blob::{BlobDataError, BlobDataSource},
-    error::RpcClientError,
-};
+use rpc::{blob::BlobDataError, error::RpcClientError};
 use thiserror::Error;
-use tracing::{debug, error};
 
 /// Shasta manifest fetcher implementation.
 pub mod shasta;
@@ -40,60 +33,3 @@ pub enum ManifestFetcherError {
     #[error("invalid shasta manifest: {0}")]
     Invalid(String),
 }
-
-/// Trait describing manifest fetch behaviour for different forks.
-#[async_trait]
-pub trait ManifestFetcher: Send + Sync {
-    /// Fork-specific manifest type produced by the decoder.
-    type Manifest;
-
-    /// Access the blob data source used by this fetcher.
-    fn blob_source(&self) -> &BlobDataSource;
-
-    /// Fetch blobs from the provided source for the given blob hashes.
-    async fn fetch_blobs(
-        &self,
-        timestamp: u64,
-        blob_hashes: &[B256],
-    ) -> Result<Vec<BlobTransactionSidecar>, ManifestFetcherError> {
-        if blob_hashes.is_empty() {
-            return Err(ManifestFetcherError::EmptyBlobHashes);
-        }
-
-        debug!(hash_count = blob_hashes.len(), timestamp, "fetching blob sidecars");
-        let sidecars = self.blob_source().get_blobs(timestamp, blob_hashes).await?;
-        if sidecars.len() != blob_hashes.len() {
-            error!(
-                expected = blob_hashes.len(),
-                actual = sidecars.len(),
-                "blob response count mismatch"
-            );
-            return Err(ManifestFetcherError::BlobCountMismatch {
-                expected: blob_hashes.len(),
-                actual: sidecars.len(),
-            });
-        }
-
-        Ok(sidecars)
-    }
-
-    /// Decode the manifest for the given sidecars.
-    async fn decode_manifest(
-        &self,
-        sidecars: &[BlobTransactionSidecar],
-        offset: usize,
-        max_blocks: usize,
-    ) -> Result<Self::Manifest, ManifestFetcherError>;
-
-    /// Fetch and decode the manifest for the given blob hashes.
-    async fn fetch_and_decode_manifest(
-        &self,
-        timestamp: u64,
-        blob_hashes: &[B256],
-        offset: usize,
-        max_blocks: usize,
-    ) -> Result<Self::Manifest, ManifestFetcherError> {
-        let sidecars = self.fetch_blobs(timestamp, blob_hashes).await?;
-        self.decode_manifest(&sidecars, offset, max_blocks).await
-    }
-}

packages/taiko-client-rs/crates/driver/src/derivation/manifest/fetcher/mod.rs

@@ -1,12 +1,12 @@
 use std::sync::Arc;
 
 use alloy_consensus::BlobTransactionSidecar;
-use async_trait::async_trait;
+use alloy_primitives::B256;
 use protocol::shasta::{BlobCoder, manifest::DerivationSourceManifest};
 use rpc::blob::BlobDataSource;
-use tracing::{debug, instrument};
+use tracing::{debug, error, instrument};
 
-use super::{ManifestFetcher, ManifestFetcherError};
+use super::ManifestFetcherError;
 
 /// Decode a Shasta derivation-source manifest from blob sidecars.
 fn decode_manifest_from_sidecars(
@@ -49,27 +49,45 @@ impl ShastaSourceManifestFetcher {
     pub fn new(blob_source: Arc<BlobDataSource>) -> Self {
         Self { blob_source }
     }
-}
 
-#[async_trait]
-impl ManifestFetcher for ShastaSourceManifestFetcher {
-    /// The type of manifest produced by this fetcher.
-    type Manifest = DerivationSourceManifest;
+    /// Fetch blob sidecars for the given blob hashes.
+    async fn fetch_blobs(
+        &self,
+        timestamp: u64,
+        blob_hashes: &[B256],
+    ) -> Result<Vec<BlobTransactionSidecar>, ManifestFetcherError> {
+        if blob_hashes.is_empty() {
+            return Err(ManifestFetcherError::EmptyBlobHashes);
+        }
+
+        debug!(hash_count = blob_hashes.len(), timestamp, "fetching blob sidecars");
+        let sidecars = self.blob_source.get_blobs(timestamp, blob_hashes).await?;
+        if sidecars.len() != blob_hashes.len() {
+            error!(
+                expected = blob_hashes.len(),
+                actual = sidecars.len(),
+                "blob response count mismatch"
+            );
+            return Err(ManifestFetcherError::BlobCountMismatch {
+                expected: blob_hashes.len(),
+                actual: sidecars.len(),
+            });
+        }
 
-    /// Access the underlying blob data source.
-    fn blob_source(&self) -> &BlobDataSource {
-        &self.blob_source
+        Ok(sidecars)
     }
 
-    /// Decode a manifest from the provided sidecars and offset.
-    #[instrument(skip(self, sidecars), fields(sidecar_count = sidecars.len(), offset))]
-    async fn decode_manifest(
+    /// Fetch and decode the manifest for the given blob hashes.
+    #[instrument(skip(self, blob_hashes), fields(hash_count = blob_hashes.len(), offset))]
+    pub async fn fetch_and_decode_manifest(
         &self,
-        sidecars: &[BlobTransactionSidecar],
+        timestamp: u64,
+        blob_hashes: &[B256],
         offset: usize,
         max_blocks: usize,
-    ) -> Result<Self::Manifest, ManifestFetcherError> {
-        let manifest = decode_manifest_from_sidecars(sidecars, offset, max_blocks)?;
+    ) -> Result<DerivationSourceManifest, ManifestFetcherError> {
+        let sidecars = self.fetch_blobs(timestamp, blob_hashes).await?;
+        let manifest = decode_manifest_from_sidecars(&sidecars, offset, max_blocks)?;
         debug!(offset, max_blocks, "decoded shasta manifest from blob sidecars");
         Ok(manifest)
     }

packages/taiko-client-rs/crates/driver/src/derivation/manifest/fetcher/shasta.rs

@@ -3,4 +3,4 @@
 /// Manifest fetchers and decoding helpers per fork.
 pub mod fetcher;
 
-pub use fetcher::{ManifestFetcher, ManifestFetcherError, ShastaSourceManifestFetcher};
+pub use fetcher::{ManifestFetcherError, ShastaSourceManifestFetcher};

packages/taiko-client-rs/crates/driver/src/derivation/manifest/mod.rs

@@ -1,4 +1,4 @@
 pub mod manifest;
 pub mod pipeline;
 
-pub use pipeline::{DerivationError, DerivationPipeline, ShastaDerivationPipeline};
+pub use pipeline::{DerivationError, ShastaDerivationPipeline};

packages/taiko-client-rs/crates/driver/src/derivation/mod.rs

@@ -1,7 +1,4 @@
-//! Derivation pipeline abstractions shared across protocol forks.
-
-use alloy::rpc::types::Log;
-use async_trait::async_trait;
+//! Derivation pipeline implementations per protocol fork.
 
 /// Shared derivation error type.
 mod error;
@@ -10,33 +7,3 @@ pub mod shasta;
 
 pub use error::DerivationError;
 pub use shasta::ShastaDerivationPipeline;
-
-use crate::sync::engine::{EngineBlockOutcome, PayloadApplier};
-
-/// Trait implemented by derivation pipelines for different protocol forks.
-#[async_trait]
-pub trait DerivationPipeline: Send + Sync {
-    /// Fork-specific manifest types.
-    type Manifest: Send;
-
-    /// Convert a proposal log into a manifest for processing.
-    async fn log_to_manifest(&self, log: &Log) -> Result<Self::Manifest, DerivationError>;
-
-    /// Convert a manifest into execution engine blocks for block production.
-    async fn manifest_to_engine_blocks(
-        &self,
-        manifest: Self::Manifest,
-        applier: &(dyn PayloadApplier + Send + Sync),
-    ) -> Result<Vec<EngineBlockOutcome>, DerivationError>;
-
-    /// Process the provided proposal log, materialising the derived blocks in the execution
-    /// engine.
-    async fn process_proposal(
-        &self,
-        log: &Log,
-        applier: &(dyn PayloadApplier + Send + Sync),
-    ) -> Result<Vec<EngineBlockOutcome>, DerivationError> {
-        let manifest = self.log_to_manifest(log).await?;
-        self.manifest_to_engine_blocks(manifest, applier).await
-    }
-}

packages/taiko-client-rs/crates/driver/src/derivation/pipeline/mod.rs

@@ -12,7 +12,6 @@ use alloy_consensus::{Transaction, TxEnvelope};
 use alloy_provider::RootProvider;
 use alloy_rpc_types::{Transaction as RpcTransaction, eth::Block as RpcBlock};
 use anyhow::anyhow;
-use async_trait::async_trait;
 use bindings::inbox::{IInbox::DerivationSource, Inbox::Proposed};
 use protocol::shasta::{
     constants::{
@@ -26,13 +25,13 @@ use rpc::{blob::BlobDataSource, client::Client};
 use tracing::{debug, info, instrument, warn};
 
 use crate::{
-    derivation::manifest::{ManifestFetcher, fetcher::shasta::ShastaSourceManifestFetcher},
+    derivation::manifest::fetcher::shasta::ShastaSourceManifestFetcher,
     metrics::DriverMetrics,
     sync::engine::{EngineBlockOutcome, PayloadApplier},
 };
 use protocol::shasta::AnchorTxConstructor;
 
-use super::super::{DerivationError, DerivationPipeline};
+use super::super::DerivationError;
 
 /// Decoded Shasta `Proposed` event enriched with the containing L1 block metadata.
 #[derive(Debug, Clone)]
@@ -200,8 +199,7 @@ where
     /// Builder for Shasta anchor transactions.
     anchor_constructor: AnchorTxConstructor<RootProvider>,
     /// Manifest fetcher used to resolve derivation-source blobs.
-    derivation_source_manifest_fetcher:
-        Arc<dyn ManifestFetcher<Manifest = DerivationSourceManifest>>,
+    derivation_source_manifest_fetcher: ShastaSourceManifestFetcher,
     /// Activation timestamp for the Shasta fork on this chain.
     shasta_fork_timestamp: u64,
     /// Minimum base-fee clamp to use for EIP-4396 calculations on this chain.
@@ -226,8 +224,7 @@ where
         blob_source: Arc<BlobDataSource>,
         initial_proposal_id: U256,
     ) -> Result<Self, DerivationError> {
-        let source_manifest_fetcher: Arc<dyn ManifestFetcher<Manifest = DerivationSourceManifest>> =
-            Arc::new(ShastaSourceManifestFetcher::new(blob_source.clone()));
+        let source_manifest_fetcher = ShastaSourceManifestFetcher::new(blob_source.clone());
         let anchor_address = *rpc.shasta.anchor.address();
         let anchor_constructor =
             AnchorTxConstructor::new(rpc.l2_provider.clone(), anchor_address).await?;
@@ -351,19 +348,12 @@ where
         Ok(ProposedEventContext { event, l1_block_number, l1_block_hash, l1_timestamp })
     }
 
-    /// Fetch and decode a single manifest from the blob store.
-    ///
-    /// The caller is responsible for providing the correct fetcher implementation for
-    /// the manifest type.
-    async fn fetch_and_decode_manifest<M>(
+    /// Fetch and decode a single derivation-source manifest from the blob store.
+    async fn fetch_and_decode_manifest(
         &self,
-        fetcher: &dyn ManifestFetcher<Manifest = M>,
         source: &DerivationSource,
         proposal_timestamp: u64,
-    ) -> Result<M, DerivationError>
-    where
-        M: Send,
-    {
+    ) -> Result<DerivationSourceManifest, DerivationError> {
         let hashes = derivation_source_to_blob_hashes(source);
         let offset = source.blobSlice.offset.to::<u64>() as usize;
         let timestamp = source.blobSlice.timestamp.to::<u64>();
@@ -373,8 +363,10 @@ where
             hash_count = hashes.len(),
             offset, timestamp, proposal_timestamp, max_blocks, "fetching manifest sidecars"
         );
-        let manifest =
-            fetcher.fetch_and_decode_manifest(timestamp, &hashes, offset, max_blocks).await?;
+        let manifest = self
+            .derivation_source_manifest_fetcher
+            .fetch_and_decode_manifest(timestamp, &hashes, offset, max_blocks)
+            .await?;
         Ok(manifest)
     }
 
@@ -404,13 +396,7 @@ where
             let manifest = if !is_source_offset_valid(source) {
                 DerivationSourceManifest::default()
             } else {
-                let manifest = self
-                    .fetch_and_decode_manifest(
-                        self.derivation_source_manifest_fetcher.as_ref(),
-                        source,
-                        event.l1_timestamp,
-                    )
-                    .await?;
+                let manifest = self.fetch_and_decode_manifest(source, event.l1_timestamp).await?;
                 validate_forced_inclusion_manifest(proposal_id, source, manifest)
             };
             manifest_segments.push(SourceManifestSegment {
@@ -485,25 +471,15 @@ where
     }
 }
 
-#[async_trait]
-impl<P> DerivationPipeline for ShastaDerivationPipeline<P>
+impl<P> ShastaDerivationPipeline<P>
 where
     P: Provider + Clone + Send + Sync + 'static,
 {
-    type Manifest = ShastaProposalBundle;
-
-    // Convert a proposal log into a manifest for processing.
-    #[instrument(skip(self, log), name = "shasta_manifest_from_log")]
-    async fn log_to_manifest(&self, log: &Log) -> Result<Self::Manifest, DerivationError> {
-        let event = self.decode_log_to_event_context(log).await?;
-        self.event_to_manifest(&event).await
-    }
-
-    // Convert a manifest into execution engine blocks for block production.
+    /// Convert a manifest into execution engine blocks for block production.
     #[instrument(skip(self, manifest, applier), name = "shasta_manifest_to_blocks")]
     async fn manifest_to_engine_blocks(
         &self,
-        manifest: Self::Manifest,
+        manifest: ShastaProposalBundle,
         applier: &(dyn PayloadApplier + Send + Sync),
     ) -> Result<Vec<EngineBlockOutcome>, DerivationError> {
         let ShastaProposalBundle { meta, sources, .. } = manifest;
@@ -549,8 +525,10 @@ where
         Ok(outcomes)
     }
 
+    /// Process the provided proposal log, materialising the derived blocks in the execution
+    /// engine.
     #[instrument(skip(self, log, applier), name = "shasta_process_proposal")]
-    async fn process_proposal(
+    pub async fn process_proposal(
         &self,
         log: &Log,
         applier: &(dyn PayloadApplier + Send + Sync),
@@ -823,9 +801,7 @@ mod tests {
         let pipeline = ShastaDerivationPipeline {
             rpc: client,
             anchor_constructor,
-            derivation_source_manifest_fetcher: Arc::new(ShastaSourceManifestFetcher::new(
-                blob_source,
-            )),
+            derivation_source_manifest_fetcher: ShastaSourceManifestFetcher::new(blob_source),
             shasta_fork_timestamp: 0,
             min_base_fee_to_clamp: min_base_fee_for_chain(TAIKO_MAINNET_CHAIN_ID),
             chain_id: TAIKO_MAINNET_CHAIN_ID,