Merge branch 'main' into feat/add_tokio_console

Author: georgehao

crates/chain-orchestrator/src/event.rs

@@ -2,9 +2,7 @@ use alloy_consensus::Header;
 use alloy_primitives::{Signature, B256};
 use reth_network_peers::PeerId;
 use reth_scroll_primitives::ScrollBlock;
-use rollup_node_primitives::{
-    BatchInfo, BlockInfo, ChainImport, L2BlockInfoWithL1Messages, WithFinalizedBlockNumber,
-};
+use rollup_node_primitives::{BatchInfo, BlockInfo, ChainImport, L2BlockInfoWithL1Messages};
 
 /// An event emitted by the `ChainOrchestrator`.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -40,9 +38,8 @@ pub enum ChainOrchestratorEvent {
         /// The safe L2 block info.
         safe_head: Option<BlockInfo>,
     },
-    /// A batch has been finalized returning an optional finalized L2 block. Also returns a
-    /// [`BatchInfo`] if the finalized event occurred in a finalized L1 block.
-    BatchFinalized(Option<WithFinalizedBlockNumber<BatchInfo>>, Option<BlockInfo>),
+    /// A batch has been finalized returning a list of finalized batches.
+    BatchFinalized(u64, Vec<BatchInfo>),
     /// An L1 block has been finalized returning the L1 block number and the list of finalized
     /// batches.
     L1BlockFinalized(u64, Vec<BatchInfo>),

crates/chain-orchestrator/src/lib.rs

@@ -556,6 +556,7 @@ impl<
                         self.database.clone(),
                         index,
                         block_number,
+                        self.l1_finalized_block_number.clone(),
                     )),
                 ))
             }
@@ -695,10 +696,20 @@ impl<
         database: Arc<Database>,
         batch_index: u64,
         block_number: u64,
+        finalized_block_number: Arc<AtomicU64>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
         // finalize all batches up to `batch_index`.
         database.finalize_batches_up_to_index(batch_index, block_number).await?;
 
+        // Get all unprocessed batches that have been finalized by this L1 block finalization.
+        let finalized_block_number = finalized_block_number.load(Ordering::Relaxed);
+        if finalized_block_number >= block_number {
+            let finalized_batches = database
+                .fetch_and_update_unprocessed_finalized_batches(finalized_block_number)
+                .await?;
+            return Ok(Some(ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches)))
+        }
+
         Ok(None)
     }
 }

crates/manager/src/manager/mod.rs

@@ -175,6 +175,7 @@ where
         sequencer: Option<Sequencer<L1MP>>,
         signer: Option<SignerHandle>,
         block_time: Option<u64>,
+        auto_start: bool,
         chain_orchestrator: ChainOrchestrator<CS, <N as BlockDownloaderProvider>::Client, P>,
         l1_v2_message_queue_start_index: u64,
     ) -> (Self, RollupManagerHandle<N>) {
@@ -193,7 +194,11 @@ where
             event_sender: None,
             sequencer,
             signer,
-            block_building_trigger: block_time.map(delayed_interval),
+            block_building_trigger: if auto_start {
+                block_time.map(delayed_interval)
+            } else {
+                None
+            },
             block_time_config: block_time,
         };
         (rnm, RollupManagerHandle::new(handle_tx))
@@ -202,7 +207,7 @@ where
     /// Returns a new event listener for the rollup node manager.
     pub fn event_listener(&mut self) -> EventStream<RollupManagerEvent> {
         if let Some(event_sender) = &self.event_sender {
-            return event_sender.new_listener()
+            return event_sender.new_listener();
         };
 
         let event_sender = EventSender::new(EVENT_CHANNEL_SIZE);
@@ -269,17 +274,16 @@ where
                 // // push the batch info into the derivation pipeline.
                 // self.derivation_pipeline.push_batch(batch_info, l1_block_number);
             }
-            ChainOrchestratorEvent::BatchFinalized(batch_info, ..) => {
+            ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches) => {
                 // Uncomment once we implement issue #273.
                 // // update the fcs on new finalized block.
                 // if let Some(finalized_block) = finalized_block {
                 //     self.engine.set_finalized_block_info(finalized_block);
                 // }
                 // Remove once we implement issue #273.
                 // Update the derivation pipeline on new finalized batch.
-                #[allow(clippy::collapsible_match)]
-                if let Some(batch_info) = batch_info {
-                    self.derivation_pipeline.push_batch(batch_info.inner, batch_info.number);
+                for batch_info in finalized_batches {
+                    self.derivation_pipeline.push_batch(batch_info, block_number);
                 }
             }
             ChainOrchestratorEvent::L1BlockFinalized(l1_block_number, finalized_batches, ..) => {

crates/network/src/manager.rs

@@ -4,7 +4,7 @@ use super::{
     BlockImportOutcome, BlockValidation, NetworkHandleMessage, NetworkManagerEvent,
     NewBlockWithPeer, ScrollNetworkHandle,
 };
-use alloy_primitives::{FixedBytes, Signature, U128};
+use alloy_primitives::{FixedBytes, Signature, B256, U128};
 use futures::{FutureExt, Stream, StreamExt};
 use reth_chainspec::EthChainSpec;
 use reth_eth_wire_types::NewBlock as EthWireNewBlock;
@@ -54,7 +54,9 @@ pub struct ScrollNetworkManager<N, CS> {
     /// The receiver for new blocks received from the network (used to bridge from eth-wire).
     eth_wire_listener: Option<EventStream<RethNewBlockWithPeer<ScrollBlock>>>,
     /// The scroll wire protocol manager.
-    scroll_wire: ScrollWireManager,
+    pub scroll_wire: ScrollWireManager,
+    /// The LRU cache used to track already seen (block,signature) pair.
+    pub blocks_seen: LruCache<(B256, Signature)>,
     /// The constant value that must be added to the block number to get the total difficulty.
     td_constant: U128,
 }
@@ -90,6 +92,8 @@ impl<CS: ScrollHardforks + EthChainSpec + Send + Sync + 'static>
         // Create the scroll-wire protocol manager.
         let scroll_wire = ScrollWireManager::new(events);
 
+        let blocks_seen = LruCache::new(LRU_CACHE_SIZE);
+
         // Spawn the inner network manager.
         tokio::spawn(inner_network_manager);
 
@@ -98,6 +102,7 @@ impl<CS: ScrollHardforks + EthChainSpec + Send + Sync + 'static>
             handle,
             from_handle_rx: from_handle_rx.into(),
             scroll_wire,
+            blocks_seen,
             eth_wire_listener,
             td_constant,
         }
@@ -128,11 +133,14 @@ impl<
 
         let handle = ScrollNetworkHandle::new(to_manager_tx, inner_network_handle);
 
+        let blocks_seen = LruCache::new(LRU_CACHE_SIZE);
+
         Self {
             chain_spec,
             handle,
             from_handle_rx: from_handle_rx.into(),
             scroll_wire,
+            blocks_seen,
             eth_wire_listener,
             td_constant,
         }
@@ -177,11 +185,29 @@ impl<
     }
 
     /// Handler for received events from the [`ScrollWireManager`].
-    fn on_scroll_wire_event(&mut self, event: ScrollWireEvent) -> NetworkManagerEvent {
+    fn on_scroll_wire_event(&mut self, event: ScrollWireEvent) -> Option<NetworkManagerEvent> {
         match event {
             ScrollWireEvent::NewBlock { peer_id, block, signature } => {
-                trace!(target: "scroll::network::manager", peer_id = ?peer_id, block = ?block.hash_slow(), signature = ?signature, "Received new block");
-                NetworkManagerEvent::NewBlock(NewBlockWithPeer { peer_id, block, signature })
+                let block_hash = block.hash_slow();
+                trace!(target: "scroll::network::manager", peer_id = ?peer_id, block = ?block_hash, signature = ?signature, "Received new block");
+                if self.blocks_seen.contains(&(block_hash, signature)) {
+                    None
+                } else {
+                    // Update the state of the peer cache i.e. peer has seen this block.
+                    self.scroll_wire
+                        .state_mut()
+                        .entry(peer_id)
+                        .or_insert_with(|| LruCache::new(LRU_CACHE_SIZE))
+                        .insert(block_hash);
+                    // Update the state of the block cache i.e. we have seen this block.
+                    self.blocks_seen.insert((block.hash_slow(), signature));
+
+                    Some(NetworkManagerEvent::NewBlock(NewBlockWithPeer {
+                        peer_id,
+                        block,
+                        signature,
+                    }))
+                }
             }
             // Only `NewBlock` events are expected from the scroll-wire protocol.
             _ => {
@@ -214,12 +240,6 @@ impl<
             Ok(BlockValidation::ValidBlock { new_block: msg }) |
             Ok(BlockValidation::ValidHeader { new_block: msg }) => {
                 trace!(target: "scroll::network::manager", peer_id = ?peer, block = ?msg.block, "Block import successful - announcing block to network");
-                let hash = msg.block.hash_slow();
-                self.scroll_wire
-                    .state_mut()
-                    .entry(peer)
-                    .or_insert_with(|| LruCache::new(LRU_CACHE_SIZE))
-                    .insert(hash);
                 self.announce_block(msg);
             }
             Err(BlockImportError::Consensus(err)) => {
@@ -241,14 +261,6 @@ impl<
         block: reth_network_api::block::NewBlockWithPeer<ScrollBlock>,
     ) -> Option<NetworkManagerEvent> {
         let reth_network_api::block::NewBlockWithPeer { peer_id, mut block } = block;
-        let block_hash = block.hash_slow();
-        self.scroll_wire
-            .state_mut()
-            .entry(peer_id)
-            .or_insert_with(|| LruCache::new(LRU_CACHE_SIZE))
-            .insert(block_hash);
-
-        trace!(target: "scroll::bridge::import", peer_id = %peer_id, block_hash = %block_hash, "Received new block from eth-wire protocol");
 
         // We purge the extra data field post euclid v2 to align with protocol specification.
         let extra_data = if self.chain_spec.is_euclid_v2_active_at_timestamp(block.timestamp) {
@@ -268,6 +280,21 @@ impl<
             .checked_sub(ECDSA_SIGNATURE_LEN)
             .and_then(|i| Signature::from_raw(&extra_data[i..]).ok())
         {
+            let block_hash = block.hash_slow();
+            if self.blocks_seen.contains(&(block_hash, signature)) {
+                return None;
+            }
+            trace!(target: "scroll::bridge::import", peer_id = %peer_id, block_hash = %block_hash, "Received new block from eth-wire protocol");
+
+            // Update the state of the peer cache i.e. peer has seen this block.
+            self.scroll_wire
+                .state_mut()
+                .entry(peer_id)
+                .or_insert_with(|| LruCache::new(LRU_CACHE_SIZE))
+                .insert(block_hash);
+
+            // Update the state of the block cache i.e. we have seen this block.
+            self.blocks_seen.insert((block_hash, signature));
             Some(NetworkManagerEvent::NewBlock(NewBlockWithPeer { peer_id, block, signature }))
         } else {
             tracing::warn!(target: "scroll::bridge::import", peer_id = %peer_id, "Failed to extract signature from block extra data, penalizing peer");
@@ -306,7 +333,9 @@ impl<
 
         // Next we handle the scroll-wire events.
         if let Poll::Ready(event) = this.scroll_wire.poll_unpin(cx) {
-            return Poll::Ready(Some(this.on_scroll_wire_event(event)));
+            if let Some(event) = this.on_scroll_wire_event(event) {
+                return Poll::Ready(Some(event));
+            }
         }
 
         // Handle blocks received from the eth-wire protocol.

crates/node/src/args.rs

@@ -307,7 +307,7 @@ impl ScrollRollupNodeConfig {
 
         // Construct the Sequencer.
         let chain_config = chain_spec.chain_config();
-        let (sequencer, block_time) = if self.sequencer_args.sequencer_enabled {
+        let (sequencer, block_time, auto_start) = if self.sequencer_args.sequencer_enabled {
             let args = &self.sequencer_args;
             let sequencer = Sequencer::new(
                 Arc::new(l1_messages_provider),
@@ -317,9 +317,9 @@ impl ScrollRollupNodeConfig {
                 0,
                 self.sequencer_args.l1_message_inclusion_mode,
             );
-            (Some(sequencer), (args.block_time != 0).then_some(args.block_time))
+            (Some(sequencer), (args.block_time != 0).then_some(args.block_time), args.auto_start)
         } else {
-            (None, None)
+            (None, None, false)
         };
 
         // Instantiate the signer
@@ -366,6 +366,7 @@ impl ScrollRollupNodeConfig {
             sequencer,
             signer,
             block_time,
+            auto_start,
             chain_orchestrator,
             l1_v2_message_queue_start_index,
         )
@@ -549,6 +550,9 @@ pub struct SequencerArgs {
     /// Enable the scroll block sequencer.
     #[arg(long = "sequencer.enabled", default_value_t = false)]
     pub sequencer_enabled: bool,
+    /// Whether the sequencer should start sequencing automatically on startup.
+    #[arg(long = "sequencer.auto-start", default_value_t = false)]
+    pub auto_start: bool,
     /// The block time for the sequencer.
     #[arg(long = "sequencer.block-time", id = "sequencer_block_time", value_name = "SEQUENCER_BLOCK_TIME", default_value_t = constants::DEFAULT_BLOCK_TIME)]
     pub block_time: u64,

crates/node/src/test_utils.rs

@@ -185,6 +185,7 @@ pub fn default_sequencer_test_scroll_rollup_node_config() -> ScrollRollupNodeCon
         },
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             payload_building_duration: 40,
             fee_recipient: Default::default(),

crates/node/tests/e2e.rs

@@ -62,6 +62,7 @@ async fn can_bridge_l1_messages() -> eyre::Result<()> {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             allow_empty_blocks: true,
@@ -157,6 +158,7 @@ async fn can_sequence_and_gossip_blocks() {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             payload_building_duration: 1000,
@@ -255,6 +257,7 @@ async fn can_penalize_peer_for_invalid_block() {
         engine_driver_args: EngineDriverArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             payload_building_duration: 1000,

crates/node/tests/sync.rs

@@ -197,6 +197,7 @@ async fn test_should_consolidate_after_optimistic_sync() -> eyre::Result<()> {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             allow_empty_blocks: true,
@@ -445,6 +446,7 @@ async fn test_consolidation() -> eyre::Result<()> {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             allow_empty_blocks: true,
@@ -618,6 +620,7 @@ async fn test_chain_orchestrator_shallow_reorg_with_gap() -> eyre::Result<()> {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             allow_empty_blocks: true,

crates/sequencer/tests/e2e.rs

@@ -502,6 +502,7 @@ async fn can_sequence_blocks_with_private_key_file() -> eyre::Result<()> {
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             payload_building_duration: 1000,
@@ -594,6 +595,7 @@ async fn can_sequence_blocks_with_hex_key_file_without_prefix() -> eyre::Result<
         chain_orchestrator_args: ChainOrchestratorArgs::default(),
         sequencer_args: SequencerArgs {
             sequencer_enabled: true,
+            auto_start: true,
             block_time: 0,
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             payload_building_duration: 1000,

tests/launch_rollup_node_sequencer.bash

@@ -7,6 +7,7 @@ exec rollup-node node --chain dev --datadir=/l2reth --metrics=0.0.0.0:6060 --net
   --log.stdout.format log-fmt -vvv \
   --test \
   --sequencer.enabled \
+  --sequencer.auto-start \
   --sequencer.block-time 250 \
   --sequencer.payload-building-duration 230 \
   --sequencer.allow-empty-blocks \