feat(derive): Online Data Source Factory Wiring

crates/derive/src/online/alloy_providers.rs

@@ -27,7 +27,7 @@ const CACHE_SIZE: usize = 16;
 /// **Note**:
 /// This provider fetches data using the `debug_getRawHeader`, `debug_getRawReceipts`, and
 /// `debug_getRawBlock` methods. The RPC must support this namespace.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct AlloyChainProvider<T: Provider<Http<reqwest::Client>>> {
     /// The inner Ethereum JSON-RPC provider.
     inner: T,

crates/derive/src/online/beacon_client.rs

@@ -45,7 +45,7 @@ pub trait BeaconClient {
 }
 
 /// An online implementation of the [BeaconClient] trait.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct OnlineBeaconClient<T: Provider<Http<Client>>> {
     /// The inner Ethereum JSON-RPC provider.
     inner: T,

crates/derive/src/online/blob_provider.rs

@@ -56,7 +56,7 @@ impl From<anyhow::Error> for OnlineBlobProviderError {
 }
 
 /// An online implementation of the [BlobProvider] trait.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct OnlineBlobProvider<T: Provider<Http<Client>>, B: BeaconClient, S: SlotDerivation> {
     /// The inner Ethereum JSON-RPC provider.
     _inner: T,
@@ -168,7 +168,7 @@ impl<T: Provider<Http<Client>>, B: BeaconClient, S: SlotDerivation> OnlineBlobPr
 }
 
 /// Minimal slot derivation implementation.
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct SimpleSlotDerivation;
 
 impl SlotDerivation for SimpleSlotDerivation {

crates/derive/src/online/mod.rs

@@ -1,11 +1,12 @@
 //! Contains "online" implementations for providers.
 
 use crate::{
+    sources::DataSourceFactory,
     stages::{
         AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
         L1Traversal, NextAttributes, StatefulAttributesBuilder,
     },
-    traits::{DataAvailabilityProvider, ResettableStage},
+    traits::ResettableStage,
     types::RollupConfig,
 };
 
@@ -18,7 +19,14 @@ use core::fmt::Debug;
 pub fn new_online_stack(
     rollup_config: Arc<RollupConfig>,
     chain_provider: AlloyChainProvider<ReqwestProvider>,
-    dap_source: impl DataAvailabilityProvider + Send + Sync + Debug,
+    dap_source: DataSourceFactory<
+        AlloyChainProvider<ReqwestProvider>,
+        OnlineBlobProvider<
+            ReqwestProvider,
+            OnlineBeaconClient<ReqwestProvider>,
+            SimpleSlotDerivation,
+        >,
+    >,
     fetcher: AlloyL2ChainProvider<ReqwestProvider>,
     builder: StatefulAttributesBuilder<
         AlloyChainProvider<ReqwestProvider>,

crates/derive/src/stages/batch_queue.rs

@@ -637,14 +637,13 @@ mod tests {
         };
         let res = bq.next_batch(parent).await.unwrap_err();
         let logs = trace_store.get_by_level(Level::INFO);
-        assert_eq!(logs.len(), 4);
+        assert_eq!(logs.len(), 2);
         let str = alloc::format!("Advancing batch queue origin: {:?}", origin);
         assert!(logs[0].contains(&str));
-        assert!(logs[1].contains("need more l1 blocks to check entire origins of span batch"));
-        assert!(logs[2].contains("Deriving next batch for epoch: 16988980031808077784"));
-        assert!(logs[3].contains("need more l1 blocks to check entire origins of span batch"));
+        assert!(logs[1].contains("Deriving next batch for epoch: 16988980031808077784"));
         let warns = trace_store.get_by_level(Level::WARN);
-        assert_eq!(warns.len(), 0);
+        assert_eq!(warns.len(), 1);
+        assert!(warns[0].contains("batch is for future epoch too far ahead, while it has the next timestamp, so it must be invalid"));
         assert_eq!(res, StageError::NotEnoughData);
     }