Merge branch 'main' into feat/bump-reth

Author: greged93

crates/node/src/add_ons/mod.rs

@@ -24,13 +24,15 @@ use reth_scroll_node::{
 use reth_scroll_primitives::ScrollPrimitives;
 use reth_scroll_rpc::{eth::ScrollEthApiBuilder, ScrollEthApi, ScrollEthApiError};
 use scroll_alloy_evm::ScrollTransactionIntoTxEnv;
+use scroll_wire::ScrollWireEvent;
 
 mod handle;
 pub use handle::ScrollAddOnsHandle;
 
 mod rollup;
 pub use rollup::IsDevChain;
 use rollup::RollupManagerAddOn;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 /// Add-ons for the Scroll follower node.
 #[derive(Debug)]
@@ -57,9 +59,12 @@ where
     ScrollEthApiBuilder: EthApiBuilder<N>,
 {
     /// Create a new instance of [`ScrollRollupNodeAddOns`].
-    pub fn new(config: ScrollRollupNodeConfig) -> Self {
+    pub fn new(
+        config: ScrollRollupNodeConfig,
+        scroll_wire_event: UnboundedReceiver<ScrollWireEvent>,
+    ) -> Self {
         let rpc_add_ons = RpcAddOns::default();
-        let rollup_manager_addon = RollupManagerAddOn::new(config);
+        let rollup_manager_addon = RollupManagerAddOn::new(config, scroll_wire_event);
         Self { rpc_add_ons, rollup_manager_addon }
     }
 }

crates/node/src/add_ons/rollup.rs

@@ -11,8 +11,9 @@ use reth_scroll_node::ScrollNetworkPrimitives;
 use rollup_node_manager::RollupManagerHandle;
 use rollup_node_watcher::L1Notification;
 use scroll_alloy_hardforks::ScrollHardforks;
+use scroll_wire::ScrollWireEvent;
 use std::sync::Arc;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc::{Sender, UnboundedReceiver};
 
 /// Implementing the trait allows the type to return whether it is configured for dev chain.
 #[auto_impl::auto_impl(Arc)]
@@ -32,12 +33,16 @@ impl IsDevChain for ScrollChainSpec {
 #[derive(Debug)]
 pub struct RollupManagerAddOn {
     config: ScrollRollupNodeConfig,
+    scroll_wire_event: UnboundedReceiver<ScrollWireEvent>,
 }
 
 impl RollupManagerAddOn {
     /// Create a new rollup node manager addon.
-    pub const fn new(config: ScrollRollupNodeConfig) -> Self {
-        Self { config }
+    pub const fn new(
+        config: ScrollRollupNodeConfig,
+        scroll_wire_event: UnboundedReceiver<ScrollWireEvent>,
+    ) -> Self {
+        Self { config, scroll_wire_event }
     }
 
     /// Launch the rollup node manager addon.
@@ -54,6 +59,7 @@ impl RollupManagerAddOn {
             .config
             .build(
                 ctx.node.network().clone(),
+                self.scroll_wire_event,
                 rpc.rpc_server_handles,
                 ctx.config.chain.clone(),
                 ctx.config.datadir().db(),

crates/node/src/args.rs

@@ -11,7 +11,7 @@ use alloy_signer_local::PrivateKeySigner;
 use alloy_transport::layers::RetryBackoffLayer;
 use aws_sdk_kms::config::BehaviorVersion;
 use reth_chainspec::EthChainSpec;
-use reth_network::{protocol::IntoRlpxSubProtocol, NetworkProtocols};
+use reth_network::NetworkProtocols;
 use reth_network_api::FullNetwork;
 use reth_node_builder::rpc::RethRpcServerHandles;
 use reth_node_core::primitives::BlockHeader;
@@ -34,9 +34,9 @@ use scroll_db::{Database, DatabaseConnectionProvider, DatabaseOperations};
 use scroll_engine::{genesis_hash_from_chain_spec, EngineDriver, ForkchoiceState};
 use scroll_migration::traits::ScrollMigrator;
 use scroll_network::ScrollNetworkManager;
-use scroll_wire::{ScrollWireConfig, ScrollWireProtocolHandler};
+use scroll_wire::ScrollWireEvent;
 use std::{fs, path::PathBuf, sync::Arc, time::Duration};
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc::{Sender, UnboundedReceiver};
 
 /// A struct that represents the arguments for the rollup node.
 #[derive(Debug, Clone, clap::Args)]
@@ -96,6 +96,7 @@ impl ScrollRollupNodeConfig {
     >(
         self,
         network: N,
+        events: UnboundedReceiver<ScrollWireEvent>,
         rpc_server_handles: RethRpcServerHandles,
         chain_spec: CS,
         db_path: PathBuf,
@@ -112,9 +113,6 @@ impl ScrollRollupNodeConfig {
         Option<Sender<Arc<L1Notification>>>,
     )> {
         // Instantiate the network manager
-        let (scroll_wire_handler, events) =
-            ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
-        network.add_rlpx_sub_protocol(scroll_wire_handler.into_rlpx_sub_protocol());
         let scroll_network_manager = ScrollNetworkManager::from_parts(network.clone(), events);
 
         // Get the rollup node config.

crates/node/src/node.rs

@@ -4,6 +4,7 @@ use crate::args::ScrollRollupNodeConfig;
 use std::time::Duration;
 
 use super::add_ons::ScrollRollupNodeAddOns;
+use reth_network::protocol::IntoRlpxSubProtocol;
 use reth_node_api::NodeTypes;
 use reth_node_builder::{
     components::{BasicPayloadServiceBuilder, ComponentsBuilder},
@@ -13,11 +14,15 @@ use reth_scroll_node::{
     ScrollConsensusBuilder, ScrollExecutorBuilder, ScrollNetworkBuilder, ScrollNode,
     ScrollPayloadBuilderBuilder, ScrollPoolBuilder,
 };
+use scroll_wire::{ScrollWireConfig, ScrollWireEvent, ScrollWireProtocolHandler};
+use std::sync::Arc;
+use tokio::sync::{mpsc::UnboundedReceiver, Mutex};
 
 /// The Scroll node implementation.
 #[derive(Clone, Debug)]
 pub struct ScrollRollupNode {
     config: ScrollRollupNodeConfig,
+    scroll_wire_events: Arc<Mutex<Option<UnboundedReceiver<ScrollWireEvent>>>>,
 }
 
 impl ScrollRollupNode {
@@ -27,7 +32,8 @@ impl ScrollRollupNode {
             .validate()
             .map_err(|e| eyre::eyre!("Configuration validation failed: {}", e))
             .expect("Configuration validation failed");
-        Self { config }
+
+        Self { config, scroll_wire_events: Arc::new(Mutex::new(None)) }
     }
 }
 
@@ -49,18 +55,29 @@ where
     >;
 
     fn components_builder(&self) -> Self::ComponentsBuilder {
-        ScrollNode::components().payload(BasicPayloadServiceBuilder::new(
-            ScrollPayloadBuilderBuilder {
+        let (scroll_wire_handler, events) =
+            ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
+
+        *self.scroll_wire_events.try_lock().unwrap() = Some(events);
+
+        ScrollNode::components()
+            .payload(BasicPayloadServiceBuilder::new(ScrollPayloadBuilderBuilder {
                 payload_building_time_limit: Duration::from_millis(
                     self.config.sequencer_args.payload_building_duration,
                 ),
                 best_transactions: (),
-            },
-        ))
+            }))
+            .network(
+                ScrollNetworkBuilder::new()
+                    .with_sub_protocol(scroll_wire_handler.into_rlpx_sub_protocol()),
+            )
     }
 
     fn add_ons(&self) -> Self::AddOns {
-        ScrollRollupNodeAddOns::new(self.config.clone())
+        ScrollRollupNodeAddOns::new(
+            self.config.clone(),
+            self.scroll_wire_events.try_lock().unwrap().take().unwrap(),
+        )
     }
 }
 

crates/node/tests/e2e.rs

@@ -18,7 +18,7 @@ use rollup_node_sequencer::L1MessageInclusionMode;
 use rollup_node_watcher::L1Notification;
 use scroll_alloy_consensus::TxL1Message;
 use scroll_network::{NewBlockWithPeer, SCROLL_MAINNET};
-use scroll_wire::ScrollWireConfig;
+use scroll_wire::{ScrollWireConfig, ScrollWireProtocolHandler};
 use std::{
     path::PathBuf,
     sync::Arc,
@@ -277,10 +277,12 @@ async fn graceful_shutdown_consolidates_most_recent_batch_on_startup() -> eyre::
             .expect("valid url that will not be used as test batches use calldata"),
     );
 
+    let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
     let (mut rnm, handle, l1_notification_tx) = config
         .clone()
         .build(
             node.inner.network.clone(),
+            events,
             node.inner.add_ons_handle.rpc_handle.rpc_server_handles.clone(),
             chain_spec.clone(),
             path.clone(),
@@ -384,10 +386,12 @@ async fn graceful_shutdown_consolidates_most_recent_batch_on_startup() -> eyre::
     drop(rnm_events);
 
     // Start the RNM again.
+    let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
     let (mut rnm, handle, l1_notification_tx) = config
         .clone()
         .build(
             node.inner.network.clone(),
+            events,
             node.inner.add_ons_handle.rpc_handle.rpc_server_handles.clone(),
             chain_spec,
             path.clone(),