L1 retrieval

crates/derive/src/stages/l1_retrieval.rs

@@ -56,17 +56,29 @@ where
                 .ok_or_else(|| anyhow!("No block to retrieve data from"))?;
             self.data = Some(
                 self.provider
-                    .open_data(&next, self.prev.system_config.batch_sender)
+                    .open_data(&next, self.prev.system_config.batcher_addr)
                     .await?,
             );
         }
 
         // Fetch next data item from the iterator.
-        let data = self
-            .data
-            .as_mut()
-            .and_then(|d| d.next())
-            .ok_or_else(|| anyhow!("No more data to retrieve"))?;
+        let data = self.data.as_mut().and_then(|d| d.next()).ok_or_else(|| {
+            self.data = None;
+            anyhow!("No more data to retrieve")
+        })?;
         Ok(data.into())
     }
 }
+
+#[async_trait]
+impl<T, DAP, CP> ResettableStage for L1Retrieval<T, DAP, CP>
+where
+    T: Into<Bytes>,
+    DAP: DataAvailabilityProvider + Send,
+    CP: ChainProvider + Send,
+{
+    async fn reset(&mut self, base: BlockInfo, cfg: SystemConfig) -> Result<()> {
+        self.data = Some(self.provider.open_data(&base, cfg.batcher_addr).await?);
+        Ok(())
+    }
+}

crates/derive/src/stages/l1_traversal.rs

@@ -4,7 +4,9 @@ use crate::{
     traits::{ChainProvider, ResettableStage},
     types::{BlockInfo, RollupConfig, SystemConfig},
 };
+use alloc::boxed::Box;
 use anyhow::{anyhow, bail, Result};
+use async_trait::async_trait;
 
 /// The L1 traversal stage of the derivation pipeline.
 #[derive(Debug, Clone, Copy)]
@@ -83,8 +85,9 @@ impl<F: ChainProvider> L1Traversal<F> {
     }
 }
 
-impl<F: ChainProvider> ResettableStage for L1Traversal<F> {
-    fn reset(&mut self, base: BlockInfo, cfg: SystemConfig) -> Result<()> {
+#[async_trait]
+impl<F: ChainProvider + Send> ResettableStage for L1Traversal<F> {
+    async fn reset(&mut self, base: BlockInfo, cfg: SystemConfig) -> Result<()> {
         self.block = Some(base);
         self.done = false;
         self.system_config = cfg;

crates/derive/src/traits/data_sources.rs

@@ -22,7 +22,7 @@ pub trait ChainProvider {
 #[async_trait]
 pub trait DataAvailabilityProvider {
     /// A data iterator for the data source to return.
-    type DataIter<T>: DataIter<T>;
+    type DataIter<T>: DataIter<T> + Send;
 
     /// Returns the data availability for the block with the given hash, or an error if the block does not exist in the
     /// data source.

crates/derive/src/traits/stages.rs

@@ -1,10 +1,13 @@
 //! This module contains common traits for stages within the derivation pipeline.
 
 use crate::types::{BlockInfo, SystemConfig};
+use alloc::boxed::Box;
 use anyhow::Result;
+use async_trait::async_trait;
 
 /// Describes the functionality fo a resettable stage within the derivation pipeline.
+#[async_trait]
 pub trait ResettableStage {
     /// Resets the derivation stage to its initial state.
-    fn reset(&mut self, base: BlockInfo, cfg: SystemConfig) -> Result<()>;
+    async fn reset(&mut self, base: BlockInfo, cfg: SystemConfig) -> Result<()>;
 }

crates/derive/src/types/system_config.rs

@@ -15,14 +15,17 @@ const CONFIG_UPDATE_EVENT_VERSION_0: B256 = B256::ZERO;
 /// Optimism system config contract values
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
 pub struct SystemConfig {
     /// Batch sender address
-    pub batch_sender: Address,
+    pub batcher_addr: Address,
     /// L2 gas limit
     pub gas_limit: U256,
     /// Fee overhead
+    #[cfg_attr(feature = "serde", serde(rename = "overhead"))]
     pub l1_fee_overhead: U256,
     /// Fee scalar
+    #[cfg_attr(feature = "serde", serde(rename = "scalar"))]
     pub l1_fee_scalar: U256,
     /// Sequencer's signer for unsafe blocks
     pub unsafe_block_signer: Address,
@@ -134,7 +137,7 @@ impl SystemConfig {
                 let batcher_address =
                     <sol!(address)>::abi_decode(&log.data.data.as_ref()[64..], true)
                         .map_err(|e| anyhow!("Failed to decode batcher update log"))?;
-                self.batch_sender = batcher_address;
+                self.batcher_addr = batcher_address;
             }
             SystemConfigUpdateType::GasConfig => {
                 if log_data.len() != 128 {
@@ -290,7 +293,7 @@ mod test {
             .unwrap();
 
         assert_eq!(
-            system_config.batch_sender,
+            system_config.batcher_addr,
             address!("000000000000000000000000000000000000bEEF")
         );
     }