feat: L1 message delay (#64)

* test: move calldata to files

* feat: batch header decoding

* feat: improve codec interface

* chore: manifests fixes

* feat: revert some codec changes

* feat: wip derivation pipeline

* feat: batch header v7

* feat: add batch abstraction

* feat: basic derivation pipeline

* feat: implement batch data hash

* feat: move PayloadData

* feat: improve batch data hash computation

* test: derivation

* chore: cleaning

* fix: lints

* fix: lints

* fix: skip wasm for derivation pipeline

* fix: data hash computation for batch

* fix: lint

* fix: lint

* fix: lints

* fix: answer comments

* fix: lints

* feat: wip

* feat: changes to the data model

* fix: codec issue

* feat: modify derivation pipeline to fetch blob

* test: move test data to file

* fix: lints

* fix: answer comments

* test: fix migration

* fix: comments

* feat: providers crate

* feat: l1 providers crate

* feat: l1 provider implementation

* feat: l1 message provider

* feat: pipeline modifications

* fix: avoid iterating the cache

* chore: simplify derivation pipeline interface

* fix: lints

* fix: rebasing

* fix: answer comments

* feat: add sequencer

* sequencer initial implementation

* sequencer implementation

* sequencer implementation

* refactor sequencer provider

* sequencer implementation

* lint

* lint deps

* address comments

* feat: sequencer integration

* remove redundant file

* add std feature for scroll-reth-primitives

* comments and clean up

* feat: L1 message delay

* address feedback

---------

Co-authored-by: Gregory Edison <[email protected]>

Author: frisitano

bin/rollup/src/args.rs

@@ -45,7 +45,7 @@ pub struct L1ProviderArgs {
     pub initial_backoff: u64,
 }
 
-#[derive(Debug, clap::Args)]
+#[derive(Debug, Clone, clap::Args)]
 pub struct SequencerArgs {
     /// The block time for the sequencer.
     #[arg(long)]

bin/rollup/src/network.rs

@@ -154,17 +154,21 @@ where
         // Construct the l1 provider.
         let beacon_provider = beacon_provider(l1_provider_args.beacon_rpc_url.to_string());
         let l1_messages_provider = DatabaseL1MessageProvider::new(db.clone(), 0);
-        let l1_provider =
-            OnlineL1Provider::new(beacon_provider, PROVIDER_BLOB_CACHE_SIZE, l1_messages_provider)
-                .await;
+        let l1_provider = OnlineL1Provider::new(
+            beacon_provider,
+            PROVIDER_BLOB_CACHE_SIZE,
+            l1_messages_provider.clone(),
+        )
+        .await;
 
         // Construct the Sequencer.
         let (sequencer, block_time) = if let Some(args) = self.config.sequencer_args {
-            let message_provider = DatabaseL1MessageProvider::new(db.clone(), 0);
             let sequencer = Sequencer::new(
-                Arc::new(message_provider),
+                Arc::new(l1_messages_provider),
                 args.fee_recipient.unwrap_or_default(),
                 args.max_l1_messages_per_block,
+                0,
+                0,
             );
             (Some(sequencer), Some(args.block_time))
         } else {

crates/node/src/lib.rs

@@ -3,7 +3,7 @@
 use futures::StreamExt;
 use reth_tokio_util::{EventSender, EventStream};
 use rollup_node_indexer::{Indexer, IndexerEvent};
-use rollup_node_sequencer::{Sequencer, SequencerL1MessageProvider};
+use rollup_node_sequencer::Sequencer;
 use rollup_node_watcher::L1Notification;
 use scroll_alloy_provider::ScrollEngineApi;
 use scroll_engine::{EngineDriver, EngineDriverEvent};
@@ -31,7 +31,7 @@ pub use consensus::PoAConsensus;
 mod consensus;
 
 use consensus::Consensus;
-use rollup_node_providers::{ExecutionPayloadProvider, L1Provider};
+use rollup_node_providers::{ExecutionPayloadProvider, L1MessageProvider, L1Provider};
 use scroll_db::Database;
 use scroll_derivation_pipeline::DerivationPipeline;
 
@@ -51,7 +51,7 @@ const EVENT_CHANNEL_SIZE: usize = 100;
 /// - `forkchoice_state`: The forkchoice state of the rollup node.
 /// - `pending_block_imports`: A collection of pending block imports.
 /// - `event_sender`: An event sender for sending events to subscribers of the rollup node manager.
-pub struct RollupNodeManager<C, EC, P, L1P, SMP> {
+pub struct RollupNodeManager<C, EC, P, L1P, L1MP> {
     /// The network manager that manages the scroll p2p network.
     network: NetworkManager,
     /// The engine driver used to communicate with the engine.
@@ -69,13 +69,13 @@ pub struct RollupNodeManager<C, EC, P, L1P, SMP> {
     /// An event sender for sending events to subscribers of the rollup node manager.
     event_sender: Option<EventSender<RollupEvent>>,
     /// The sequencer which is responsible for sequencing transactions and producing new blocks.
-    sequencer: Option<Sequencer<SMP>>,
+    sequencer: Option<Sequencer<L1MP>>,
     /// The trigger for the block building process.
     block_building_trigger: Option<Interval>,
 }
 
-impl<C: Debug, EC: Debug, P: Debug, L1P: Debug, SMP: Debug> Debug
-    for RollupNodeManager<C, EC, P, L1P, SMP>
+impl<C: Debug, EC: Debug, P: Debug, L1P: Debug, L1MP: Debug> Debug
+    for RollupNodeManager<C, EC, P, L1P, L1MP>
 {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("RollupNodeManager")
@@ -93,13 +93,13 @@ impl<C: Debug, EC: Debug, P: Debug, L1P: Debug, SMP: Debug> Debug
     }
 }
 
-impl<C, EC, P, L1P, SMP> RollupNodeManager<C, EC, P, L1P, SMP>
+impl<C, EC, P, L1P, L1MP> RollupNodeManager<C, EC, P, L1P, L1MP>
 where
     C: Consensus + Unpin,
     EC: ScrollEngineApi + Unpin + Sync + Send + 'static,
     P: ExecutionPayloadProvider + Unpin + Send + Sync + 'static,
     L1P: L1Provider + Clone + Send + Sync + 'static,
-    SMP: SequencerL1MessageProvider + Unpin + Send + Sync + 'static,
+    L1MP: L1MessageProvider + Unpin + Send + Sync + 'static,
 {
     /// Create a new [`RollupNodeManager`] instance.
     #[allow(clippy::too_many_arguments)]
@@ -111,7 +111,7 @@ where
         l1_notification_rx: Option<Receiver<Arc<L1Notification>>>,
         consensus: C,
         new_block_rx: Option<UnboundedReceiver<NewBlockWithPeer>>,
-        sequencer: Option<Sequencer<SMP>>,
+        sequencer: Option<Sequencer<L1MP>>,
         block_time: Option<u64>,
     ) -> Self {
         let indexer = Indexer::new(database.clone());
@@ -224,13 +224,13 @@ where
     }
 }
 
-impl<C, EC, P, L1P, SMP> Future for RollupNodeManager<C, EC, P, L1P, SMP>
+impl<C, EC, P, L1P, L1MP> Future for RollupNodeManager<C, EC, P, L1P, L1MP>
 where
     C: Consensus + Unpin,
     EC: ScrollEngineApi + Unpin + Sync + Send + 'static,
     P: ExecutionPayloadProvider + Unpin + Send + Sync + 'static,
     L1P: L1Provider + Clone + Unpin + Send + Sync + 'static,
-    SMP: SequencerL1MessageProvider + Unpin + Send + Sync + 'static,
+    L1MP: L1MessageProvider + Unpin + Send + Sync + 'static,
 {
     type Output = ();
 

crates/providers/src/l1/message/database.rs

@@ -49,81 +49,3 @@ impl<DB: DatabaseConnectionProvider + Sync> L1MessageProvider for DatabaseL1Mess
         self.index.fetch_add(1, Ordering::Relaxed);
     }
 }
-
-/// A provider that can provide L1 messages with a delay.
-/// This provider is used to delay the L1 messages by a certain number of blocks which builds
-/// confidence in the L1 message not being reorged.
-#[derive(Debug)]
-pub struct DatabaseL1MessageDelayProvider<DB> {
-    /// The database L1 message provider.
-    l1_message_provider: DatabaseL1MessageProvider<DB>,
-    /// The current L1 block number.
-    l1_tip: AtomicU64,
-    /// The number of blocks to wait for before including a L1 message in a block.
-    l1_message_delay: u64,
-}
-
-impl<DB> DatabaseL1MessageDelayProvider<DB> {
-    /// Returns a new instance of the [`DatabaseL1MessageDelayProvider`].
-    pub fn new(
-        l1_message_provider: DatabaseL1MessageProvider<DB>,
-        l1_tip_block_number: u64,
-        l1_message_delay: u64,
-    ) -> Self {
-        Self { l1_message_provider, l1_tip: l1_tip_block_number.into(), l1_message_delay }
-    }
-
-    /// Sets the block number of the current L1 head.
-    pub fn set_l1_head(&self, block_number: u64) {
-        self.l1_tip.store(block_number, Ordering::Relaxed);
-    }
-}
-
-/// A trait that allows the L1 message delay provider to set the current head number.
-pub trait L1MessageDelayProvider {
-    /// Set the number of the current L1 head block number.
-    fn set_l1_head(&self, _block_number: u64) {}
-}
-
-impl<DB> L1MessageDelayProvider for DatabaseL1MessageDelayProvider<DB> {
-    fn set_l1_head(&self, block_number: u64) {
-        Self::set_l1_head(self, block_number);
-    }
-}
-
-impl<DB> L1MessageDelayProvider for DatabaseL1MessageProvider<DB> {}
-
-#[async_trait::async_trait]
-impl<DB: DatabaseConnectionProvider + Sync> L1MessageProvider
-    for DatabaseL1MessageDelayProvider<DB>
-{
-    type Error = L1ProviderError;
-
-    async fn get_l1_message_with_block_number(
-        &self,
-    ) -> Result<Option<L1MessageWithBlockNumber>, Self::Error> {
-        let msg_w_bn = self.l1_message_provider.get_l1_message_with_block_number().await?;
-        let result = if let Some(msg_w_bn) = msg_w_bn {
-            let tx_block_number = msg_w_bn.block_number;
-            let depth = self.l1_tip.load(Ordering::Relaxed) - tx_block_number;
-            (depth >= self.l1_message_delay).then_some(msg_w_bn)
-        } else {
-            None
-        };
-
-        Ok(result)
-    }
-
-    fn set_index_cursor(&self, index: u64) {
-        self.l1_message_provider.set_index_cursor(index);
-    }
-
-    fn set_hash_cursor(&self, _hash: B256) {
-        // TODO: issue 43
-        todo!()
-    }
-
-    fn increment_cursor(&self) {
-        self.l1_message_provider.increment_cursor();
-    }
-}

crates/providers/src/l1/message/mod.rs

@@ -6,9 +6,7 @@ use scroll_alloy_consensus::TxL1Message;
 use scroll_db::{DatabaseConnectionProvider, DatabaseOperations};
 
 mod database;
-pub use database::{
-    DatabaseL1MessageDelayProvider, DatabaseL1MessageProvider, L1MessageDelayProvider,
-};
+pub use database::DatabaseL1MessageProvider;
 
 /// An instance of the trait can provide L1 messages using a cursor approach. Set the cursor for the
 /// provider using the queue index or hash and then call
@@ -37,12 +35,37 @@ pub trait L1MessageProvider {
         }
     }
 
+    /// Returns the L1 message with block number at the current cursor and advances the cursor if
+    /// the predicate is satisfied.
+    async fn next_l1_message_with_block_number_and_predicate(
+        &self,
+        predicate: impl Fn(L1MessageWithBlockNumber) -> bool + Send,
+    ) -> Result<Option<L1MessageWithBlockNumber>, Self::Error> {
+        match self.get_l1_message_with_block_number().await? {
+            Some(message) if predicate(message.clone()) => {
+                self.increment_cursor();
+                Ok(Some(message))
+            }
+            _ => Ok(None),
+        }
+    }
+
     /// Returns the L1 message with block number at the current cursor and advances the cursor.
     async fn next_l1_message(&self) -> Result<Option<TxL1Message>, Self::Error> {
         let message = self.next_l1_message_with_block_number().await?;
         Ok(message.map(|message| message.transaction))
     }
 
+    /// Returns the L1 message with block number at the current cursor and advances the cursor if
+    /// the predicate is satisfied.
+    async fn next_l1_message_with_predicate(
+        &self,
+        predicate: impl Fn(L1MessageWithBlockNumber) -> bool + Send,
+    ) -> Result<Option<TxL1Message>, Self::Error> {
+        let message = self.next_l1_message_with_block_number_and_predicate(predicate).await?;
+        Ok(message.map(|message| message.transaction))
+    }
+
     /// Returns the L1 message at the current cursor.
     /// This method does not advance the cursor.
     async fn get_l1_message(&self) -> Result<Option<TxL1Message>, Self::Error> {

crates/providers/src/lib.rs

@@ -12,10 +12,7 @@ mod execution_payload;
 
 pub use l1::{
     blob::L1BlobProvider,
-    message::{
-        DatabaseL1MessageDelayProvider, DatabaseL1MessageProvider, L1MessageDelayProvider,
-        L1MessageProvider,
-    },
+    message::{DatabaseL1MessageProvider, L1MessageProvider},
     L1Provider, L1ProviderError, OnlineL1Provider,
 };
 mod l1;

crates/sequencer/Cargo.toml

@@ -26,6 +26,7 @@ reth-scroll-primitives.workspace = true
 # rollup-node
 scroll-db = { workspace = true }
 rollup-node-providers.workspace = true
+rollup-node-primitives.workspace = true
 
 # misc
 futures.workspace = true

crates/sequencer/src/lib.rs

@@ -12,7 +12,8 @@ use alloy_eips::eip2718::Encodable2718;
 use alloy_primitives::Address;
 use alloy_rpc_types_engine::PayloadAttributes;
 use futures::{task::AtomicWaker, Stream};
-use rollup_node_providers::{L1MessageDelayProvider, L1MessageProvider};
+use rollup_node_primitives::L1MessageWithBlockNumber;
+use rollup_node_providers::L1MessageProvider;
 use scroll_alloy_rpc_types_engine::ScrollPayloadAttributes;
 use std::task::{Context, Poll};
 
@@ -23,34 +24,42 @@ pub use error::SequencerError;
 pub type PayloadBuildingJobFuture =
     Pin<Box<dyn Future<Output = Result<ScrollPayloadAttributes, SequencerError>> + Send>>;
 
-/// A trait used to define the L1 message provider for the sequencer.
-pub trait SequencerL1MessageProvider: L1MessageProvider + L1MessageDelayProvider {}
-impl<T> SequencerL1MessageProvider for T where T: L1MessageProvider + L1MessageDelayProvider {}
-
 /// The sequencer is responsible for sequencing transactions and producing new blocks.
-pub struct Sequencer<SMP> {
+pub struct Sequencer<P> {
     /// A reference to the database
-    provider: Arc<SMP>,
+    provider: Arc<P>,
     /// The fee recipient
     fee_recipient: Address,
     /// The number of L1 messages to include in each block.
     max_l1_messages_per_block: u64,
+    /// The current l1 block number.
+    l1_block_number: u64,
+    /// The L1 block depth at which L1 messages should be included in the payload.
+    l1_block_depth: u64,
     /// The inflight payload attributes job
     payload_attributes_job: Option<PayloadBuildingJobFuture>,
     /// A waker to notify when the Sequencer should be polled.
     waker: AtomicWaker,
 }
 
-impl<SMP> Sequencer<SMP>
+impl<P> Sequencer<P>
 where
-    SMP: SequencerL1MessageProvider + Unpin + Send + Sync + 'static,
+    P: L1MessageProvider + Unpin + Send + Sync + 'static,
 {
     /// Creates a new sequencer.
-    pub fn new(provider: Arc<SMP>, fee_recipient: Address, max_l1_messages_per_block: u64) -> Self {
+    pub fn new(
+        provider: Arc<P>,
+        fee_recipient: Address,
+        max_l1_messages_per_block: u64,
+        l1_block_number: u64,
+        l1_block_depth: u64,
+    ) -> Self {
         Self {
             provider,
             fee_recipient,
             max_l1_messages_per_block,
+            l1_block_number,
+            l1_block_depth,
             payload_attributes_job: None,
             waker: AtomicWaker::new(),
         }
@@ -79,23 +88,32 @@ where
         };
         let max_l1_messages = self.max_l1_messages_per_block;
         let database = self.provider.clone();
+        let l1_block_number = self.l1_block_number;
+        let l1_block_depth = self.l1_block_depth;
 
         self.payload_attributes_job = Some(Box::pin(async move {
-            build_payload_attributes(database, max_l1_messages, payload_attributes).await
+            build_payload_attributes(
+                database,
+                max_l1_messages,
+                payload_attributes,
+                l1_block_number,
+                l1_block_depth,
+            )
+            .await
         }));
 
         self.waker.wake();
     }
 
     /// Handle a reorg event.
-    pub fn handle_reorg(&mut self, queue_index: u64, block_number: u64) {
+    pub fn handle_reorg(&mut self, queue_index: u64, l1_block_number: u64) {
         self.provider.set_index_cursor(queue_index);
-        self.provider.set_l1_head(block_number);
+        self.l1_block_number = l1_block_number;
     }
 
     /// Handle a new L1 block.
     pub fn handle_new_l1_block(&mut self, block_number: u64) {
-        self.provider.set_l1_head(block_number);
+        self.l1_block_number = block_number;
     }
 }
 
@@ -127,17 +145,21 @@ impl<SMP> Stream for Sequencer<SMP> {
 /// Builds the payload attributes for the sequencer using the given L1 message provider.
 /// It collects the L1 messages to include in the payload and returns a `ScrollPayloadAttributes`
 /// instance.
-async fn build_payload_attributes<
-    SMP: SequencerL1MessageProvider + Unpin + Send + Sync + 'static,
->(
-    provider: Arc<SMP>,
+async fn build_payload_attributes<P: L1MessageProvider + Unpin + Send + Sync + 'static>(
+    provider: Arc<P>,
     max_l1_messages: u64,
     payload_attributes: PayloadAttributes,
+    current_l1_block_number: u64,
+    l1_block_depth: u64,
 ) -> Result<ScrollPayloadAttributes, SequencerError> {
+    let predicate = |message: L1MessageWithBlockNumber| {
+        message.block_number + l1_block_depth <= current_l1_block_number
+    };
+
     // Collect L1 messages to include in payload.
     let mut l1_messages = vec![];
     for _ in 0..max_l1_messages {
-        match provider.next_l1_message().await.map_err(Into::into)? {
+        match provider.next_l1_message_with_predicate(predicate).await.map_err(Into::into)? {
             Some(l1_message) => {
                 l1_messages.push(l1_message.encoded_2718().into());
             }