feat: L1 retrieval

crates/derive/src/stages/l1_retrieval.rs

@@ -1 +1,84 @@
+//! Contains the [L1Retrieval] stage of the derivation pipeline.]
 
+use super::L1Traversal;
+use crate::{
+    traits::{ChainProvider, DataAvailabilityProvider, DataIter, ResettableStage},
+    types::{BlockInfo, SystemConfig},
+};
+use alloc::boxed::Box;
+use alloy_primitives::Bytes;
+use anyhow::{anyhow, Result};
+use async_trait::async_trait;
+
+/// The L1 retrieval stage of the derivation pipeline.
+#[derive(Debug)]
+pub struct L1Retrieval<T, DAP, CP>
+where
+    DAP: DataAvailabilityProvider,
+    CP: ChainProvider,
+{
+    /// The previous stage in the pipeline.
+    pub prev: L1Traversal<CP>,
+    /// The data availability provider to use for the L1 retrieval stage.
+    pub provider: DAP,
+    /// The current data iterator.
+    data: Option<DAP::DataIter<T>>,
+}
+
+impl<T, DAP, CP> L1Retrieval<T, DAP, CP>
+where
+    T: Into<Bytes>,
+    DAP: DataAvailabilityProvider,
+    CP: ChainProvider,
+{
+    /// Creates a new L1 retrieval stage with the given data availability provider and previous stage.
+    pub fn new(prev: L1Traversal<CP>, provider: DAP) -> Self {
+        Self {
+            prev,
+            provider,
+            data: None,
+        }
+    }
+
+    /// Returns the current L1 block in the traversal stage, if it exists.
+    pub fn origin(&self) -> Option<&BlockInfo> {
+        self.prev.origin()
+    }
+
+    /// Retrieves the next data item from the L1 retrieval stage.
+    /// If there is data, it pushes it into the next stage.
+    /// If there is no data, it returns an error.
+    pub async fn next_data(&mut self) -> Result<Bytes> {
+        if self.data.is_none() {
+            let next = self
+                .prev
+                .next_l1_block()
+                .ok_or_else(|| anyhow!("No block to retrieve data from"))?;
+            self.data = Some(
+                self.provider
+                    .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(|| {
+            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)]
@@ -16,9 +18,9 @@ pub struct L1Traversal<Provider: ChainProvider> {
     /// Signals whether or not the traversal stage has been completed.
     done: bool,
     /// The system config
-    system_config: SystemConfig,
+    pub system_config: SystemConfig,
     /// The rollup config
-    rollup_config: RollupConfig,
+    pub rollup_config: RollupConfig,
 }
 
 impl<F: ChainProvider> L1Traversal<F> {
@@ -44,6 +46,11 @@ impl<F: ChainProvider> L1Traversal<F> {
         }
     }
 
+    /// Returns the current L1 block in the traversal stage, if it exists.
+    pub fn origin(&self) -> Option<&BlockInfo> {
+        self.block.as_ref()
+    }
+
     /// Advances the internal state of the [L1Traversal] stage to the next L1 block.
     pub async fn advance_l1_block(&mut self) -> Result<()> {
         let block = self.block.ok_or(anyhow!("No block to advance from"))?;
@@ -78,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

@@ -3,7 +3,7 @@
 // use alloy_rpc_types::Block;
 use crate::types::{BlockInfo, Receipt};
 use alloc::{boxed::Box, vec::Vec};
-use alloy_primitives::B256;
+use alloy_primitives::{Address, B256};
 use anyhow::Result;
 use async_trait::async_trait;
 
@@ -17,3 +17,24 @@ pub trait ChainProvider {
     /// source.
     async fn receipts_by_hash(&self, hash: B256) -> Result<Vec<Receipt>>;
 }
+
+/// Describes the functionality of a data source that can provide data availability information.
+#[async_trait]
+pub trait DataAvailabilityProvider {
+    /// A data iterator for the data source to return.
+    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.
+    async fn open_data<T>(
+        &self,
+        block_ref: &BlockInfo,
+        batcher_address: Address,
+    ) -> Result<Self::DataIter<T>>;
+}
+
+/// Describes the behavior of a data iterator.
+pub trait DataIter<T> {
+    /// Returns the next item in the iterator, or `None` if the iterator is exhausted.
+    fn next(&self) -> Option<T>;
+}

crates/derive/src/traits/mod.rs

@@ -1,7 +1,7 @@
 //! This module contains all of the traits describing functionality of portions of the derivation pipeline.
 
 mod data_sources;
-pub use data_sources::ChainProvider;
+pub use data_sources::{ChainProvider, DataAvailabilityProvider, DataIter};
 
 mod stages;
 pub use stages::ResettableStage;

crates/derive/src/traits/stages.rs

@@ -1,11 +1,13 @@
 //! This module contains common traits for stages within the derivation pipeline.
 
-use anyhow::Result;
-
 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")
         );
     }