feat: pipeline modifications

Author: greged93

Cargo.lock

@@ -26,6 +26,7 @@ rollup-node-providers.workspace = true
 scroll-codec.workspace = true
 
 # misc
+futures.workspace = true
 thiserror.workspace = true
 
 [dev-dependencies]

crates/derivation-pipeline/Cargo.toml

@@ -10,6 +10,9 @@ pub enum DerivationPipelineError {
     /// Missing L1 messages cursor.
     #[error("missing l1 message queue cursor")]
     MissingL1MessageQueueCursor,
+    /// Missing L1 message.
+    #[error("missing l1 message")]
+    MissingL1Message,
     /// An error at the L1 provider.
     #[error(transparent)]
     L1Provider(#[from] L1ProviderError),

crates/derivation-pipeline/src/error.rs

@@ -12,30 +12,33 @@ mod error;
 extern crate alloc as std;
 
 use crate::{data_source::CodecDataSource, error::DerivationPipelineError};
-use std::vec::Vec;
+use std::{sync::Arc, vec::Vec};
 
 use alloy_primitives::B256;
 use alloy_rpc_types_engine::PayloadAttributes;
+use futures::{stream, Stream, StreamExt};
 use reth_scroll_chainspec::SCROLL_FEE_VAULT_ADDRESS;
 use rollup_node_primitives::BatchCommitData;
-use rollup_node_providers::{L1Provider, L1ProviderError};
+use rollup_node_providers::L1Provider;
 use scroll_alloy_rpc_types_engine::ScrollPayloadAttributes;
 use scroll_codec::Codec;
 
 /// Returns an iterator over [`ScrollPayloadAttributes`] from the [`BatchCommitData`] and a
 /// [`L1Provider`].
-pub async fn derive<P: L1Provider<Error = L1ProviderError>>(
+pub async fn derive<P: L1Provider>(
     batch: BatchCommitData,
     l1_provider: &mut P,
-) -> Result<impl Iterator<Item = ScrollPayloadAttributes> + use<'_, P>, DerivationPipelineError> {
+) -> Result<
+    impl Stream<Item = Result<ScrollPayloadAttributes, DerivationPipelineError>> + use<'_, P>,
+    DerivationPipelineError,
+> {
     // fetch the blob then decode the input batch.
     let blob = if let Some(hash) = batch.blob_versioned_hash {
         l1_provider.blob(batch.block_timestamp, hash).await?
     } else {
         None
     };
-    let data =
-        CodecDataSource { calldata: batch.calldata.as_ref(), blob: blob.as_ref().map(|v| &**v) };
+    let data = CodecDataSource { calldata: batch.calldata.as_ref(), blob: blob.as_deref() };
     let decoded = Codec::decode(&data)?;
 
     // set the cursor for the l1 provider.
@@ -46,39 +49,45 @@ pub async fn derive<P: L1Provider<Error = L1ProviderError>>(
         l1_provider.set_hash_cursor(*hash);
         // we skip the first l1 message, as we are interested in the one starting after
         // prev_l1_message_queue_hash.
-        let _ = l1_provider.next_l1_message();
+        let _ = l1_provider.next_l1_message().await.map_err(Into::into)?;
     } else {
         return Err(DerivationPipelineError::MissingL1MessageQueueCursor)
     }
 
-    let iter = decoded.data.into_l2_blocks().into_iter().map(|mut block| {
-        // query the appropriate amount of l1 messages.
-        let mut txs = (0..block.context.num_l1_messages)
-            .map(|_| l1_provider.next_l1_message())
-            .map(|tx| {
+    let provider = Arc::new(&*l1_provider);
+    let iter = stream::iter(decoded.data.into_l2_blocks())
+        .map(move |data| (data, provider.clone()))
+        .then(|(mut block, provider)| async move {
+            // query the appropriate amount of l1 messages.
+            let mut txs = Vec::with_capacity(block.context.num_l1_messages as usize);
+            for _ in 0..block.context.num_l1_messages {
+                let l1_message = provider
+                    .next_l1_message()
+                    .await
+                    .map_err(Into::into)?
+                    .ok_or(DerivationPipelineError::MissingL1Message)?;
                 let mut bytes = Vec::new();
-                tx.eip2718_encode(&mut bytes);
-                bytes.into()
+                l1_message.eip2718_encode(&mut bytes);
+                txs.push(bytes.into());
+            }
+
+            // add the block transactions.
+            txs.append(&mut block.transactions);
+
+            // construct the payload attributes.
+            Ok(ScrollPayloadAttributes {
+                payload_attributes: PayloadAttributes {
+                    timestamp: block.context.timestamp,
+                    prev_randao: B256::ZERO,
+                    // TODO: this should be based off the current configuration value.
+                    suggested_fee_recipient: SCROLL_FEE_VAULT_ADDRESS,
+                    withdrawals: None,
+                    parent_beacon_block_root: None,
+                },
+                transactions: Some(txs),
+                no_tx_pool: true,
             })
-            .collect::<Vec<_>>();
-
-        // add the block transactions.
-        txs.append(&mut block.transactions);
-
-        // construct the payload attributes.
-        ScrollPayloadAttributes {
-            payload_attributes: PayloadAttributes {
-                timestamp: block.context.timestamp,
-                prev_randao: B256::ZERO,
-                // TODO: this should be based off the current configuration value.
-                suggested_fee_recipient: SCROLL_FEE_VAULT_ADDRESS,
-                withdrawals: None,
-                parent_beacon_block_root: None,
-            },
-            transactions: Some(txs),
-            no_tx_pool: true,
-        }
-    });
+        });
 
     Ok(iter)
 }
@@ -90,7 +99,7 @@ mod tests {
 
     use alloy_eips::eip4844::Blob;
     use alloy_primitives::{address, b256, bytes, U256};
-    use rollup_node_providers::L1MessageProvider;
+    use rollup_node_providers::{L1MessageProvider, L1ProviderError};
     use scroll_alloy_consensus::TxL1Message;
     use scroll_codec::decoding::test_utils::read_to_bytes;
     use tokio::sync::Mutex;
@@ -99,22 +108,30 @@ mod tests {
         messages: Arc<Mutex<Vec<TxL1Message>>>,
     }
 
+    struct Infallible;
+    impl From<Infallible> for L1ProviderError {
+        fn from(_value: Infallible) -> Self {
+            Self::Other("infallible")
+        }
+    }
+
     #[async_trait::async_trait]
     impl L1BlobProvider for TestL1MessageProvider {
-        type Error = L1ProviderError;
-
         async fn blob(
             &self,
             _block_timestamp: u64,
             _hash: B256,
-        ) -> Result<Option<Arc<Blob>>, Self::Error> {
+        ) -> Result<Option<Arc<Blob>>, L1ProviderError> {
             Ok(None)
         }
     }
 
+    #[async_trait::async_trait]
     impl L1MessageProvider for TestL1MessageProvider {
-        fn next_l1_message(&self) -> TxL1Message {
-            self.messages.try_lock().expect("lock is free").remove(0)
+        type Error = Infallible;
+
+        async fn next_l1_message(&self) -> Result<Option<TxL1Message>, Self::Error> {
+            Ok(Some(self.messages.try_lock().expect("lock is free").remove(0)))
         }
 
         fn set_index_cursor(&mut self, _index: u64) {}
@@ -152,8 +169,10 @@ mod tests {
         }];
         let mut provider = TestL1MessageProvider { messages: Arc::new(Mutex::new(l1_messages)) };
 
-        let mut attributes = derive(batch_data, &mut provider).await?;
-        let attribute = attributes.find(|a| a.payload_attributes.timestamp == 1696935384).unwrap();
+        let attributes: Vec<_> = derive(batch_data, &mut provider).await?.collect().await;
+        let attributes = attributes.into_iter().collect::<Result<Vec<_>, _>>()?;
+        let attribute =
+            attributes.iter().find(|a| a.payload_attributes.timestamp == 1696935384).unwrap();
 
         let expected = ScrollPayloadAttributes{
             payload_attributes: PayloadAttributes{
@@ -166,7 +185,7 @@ mod tests {
             transactions: Some(vec![bytes!("7ef901b7218302904094781e90f1c8fc4611c9b7497c3b47f99ef6969cbc80b901848ef1332e0000000000000000000000007f2b8c31f88b6006c382775eea88297ec1e3e9050000000000000000000000006ea73e05adc79974b931123675ea8f78ffdacdf0000000000000000000000000000000000000000000000000006a94d74f430000000000000000000000000000000000000000000000000000000000000000002100000000000000000000000000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000000000000a4232e8748000000000000000000000000ca266224613396a0e8d4c2497dbc4f33dd6cdeff000000000000000000000000ca266224613396a0e8d4c2497dbc4f33dd6cdeff000000000000000000000000000000000000000000000000006a94d74f4300000000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000947885bcbd5cecef1336b5300fb5186a12ddd8c478"), bytes!("7ef901b7228302904094781e90f1c8fc4611c9b7497c3b47f99ef6969cbc80b901848ef1332e0000000000000000000000007f2b8c31f88b6006c382775eea88297ec1e3e9050000000000000000000000006ea73e05adc79974b931123675ea8f78ffdacdf000000000000000000000000000000000000000000000000000470de4df820000000000000000000000000000000000000000000000000000000000000000002200000000000000000000000000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000000000000a4232e8748000000000000000000000000982fe4a7cbd74bb3422ebe46333c3e8046c12c7f000000000000000000000000982fe4a7cbd74bb3422ebe46333c3e8046c12c7f00000000000000000000000000000000000000000000000000470de4df8200000000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000947885bcbd5cecef1336b5300fb5186a12ddd8c478")]),
             no_tx_pool: true,
         };
-        assert_eq!(attribute, expected);
+        assert_eq!(attribute, &expected);
 
         let attribute = attributes.last().unwrap();
         let expected = ScrollPayloadAttributes{
@@ -180,7 +199,7 @@ mod tests {
             transactions: Some(vec![bytes!("f88c8202658417d7840082a4f294530000000000000000000000000000000000000280a4bede39b500000000000000000000000000000000000000000000000000000001669aa2f583104ec4a07461e6555f927393ebdf5f183738450c3842bc3b86a1db7549d9bee21fadd0b1a06d7ba96897bd9fb8e838a327d3ca34be66da11955f10d1fb2264949071e9e8cd")]),
             no_tx_pool: true,
         };
-        assert_eq!(attribute, expected);
+        assert_eq!(attribute, &expected);
 
         Ok(())
     }