Merge branch 'main' into feat-penalize-peer-on-duplicate-block

Author: yiweichi

crates/chain-orchestrator/src/handle/command.rs

@@ -3,9 +3,9 @@ use crate::{ChainOrchestratorEvent, ChainOrchestratorStatus};
 use reth_network_api::FullNetwork;
 use reth_scroll_node::ScrollNetworkPrimitives;
 use reth_tokio_util::EventStream;
-use rollup_node_primitives::{BlockInfo, L1MessageEnvelope};
+use rollup_node_primitives::{BlockInfo, ChainImport, L1MessageEnvelope};
 use scroll_db::L1MessageKey;
-use scroll_network::ScrollNetworkHandle;
+use scroll_network::{NewBlockWithPeer, ScrollNetworkHandle};
 use tokio::sync::oneshot;
 
 /// The commands that can be sent to the rollup manager.
@@ -29,6 +29,13 @@ pub enum ChainOrchestratorCommand<N: FullNetwork<Primitives = ScrollNetworkPrimi
     DatabaseQuery(DatabaseQuery),
     /// Revert the rollup node state to the specified L1 block number.
     RevertToL1Block((u64, oneshot::Sender<bool>)),
+    /// Import a block from a remote source.
+    ImportBlock {
+        /// The block to import with peer info
+        block_with_peer: NewBlockWithPeer,
+        /// Response channel
+        response: oneshot::Sender<Result<ChainImport, String>>,
+    },
     /// Enable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     SetGossip((bool, oneshot::Sender<()>)),

crates/chain-orchestrator/src/handle/mod.rs

@@ -5,9 +5,9 @@ use super::ChainOrchestratorEvent;
 use reth_network_api::FullNetwork;
 use reth_scroll_node::ScrollNetworkPrimitives;
 use reth_tokio_util::EventStream;
-use rollup_node_primitives::{BlockInfo, L1MessageEnvelope};
+use rollup_node_primitives::{BlockInfo, ChainImport, L1MessageEnvelope};
 use scroll_db::L1MessageKey;
-use scroll_network::ScrollNetworkHandle;
+use scroll_network::{NewBlockWithPeer, ScrollNetworkHandle};
 use tokio::sync::{mpsc, oneshot};
 use tracing::error;
 
@@ -132,6 +132,16 @@ impl<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> ChainOrchestratorHand
         rx.await
     }
 
+    /// Import a block from a remote source.
+    pub async fn import_block(
+        &self,
+        block_with_peer: NewBlockWithPeer,
+    ) -> Result<Result<ChainImport, String>, oneshot::error::RecvError> {
+        let (tx, rx) = oneshot::channel();
+        self.send_command(ChainOrchestratorCommand::ImportBlock { block_with_peer, response: tx });
+        rx.await
+    }
+
     /// Sends a command to the rollup manager to enable or disable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     pub async fn set_gossip(&self, enabled: bool) -> Result<(), oneshot::error::RecvError> {

crates/chain-orchestrator/src/lib.rs

@@ -292,13 +292,16 @@ impl<
                 }
             }
             SequencerEvent::PayloadReady(payload_id) => {
-                if let Some(block) = self
+                let block = self
                     .sequencer
                     .as_mut()
                     .expect("sequencer must be present")
                     .finalize_payload_building(payload_id, &mut self.engine)
-                    .await?
-                {
+                    .await?;
+
+                self.metric_handler.finish_block_building_recording(block.as_ref());
+
+                if let Some(block) = block {
                     let block_info: L2BlockInfoWithL1Messages = (&block).into();
                     self.database
                         .update_l1_messages_from_l2_blocks(vec![block_info.clone()])
@@ -307,7 +310,6 @@ impl<
                         .as_mut()
                         .expect("signer must be present")
                         .sign_block(block.clone())?;
-                    self.metric_handler.finish_block_building_recording();
                     return Ok(Some(ChainOrchestratorEvent::BlockSequenced(block)));
                 }
             }
@@ -425,6 +427,13 @@ impl<
                 self.notify(ChainOrchestratorEvent::UnwoundToL1Block(block_number));
                 let _ = tx.send(true);
             }
+            ChainOrchestratorCommand::ImportBlock { block_with_peer, response } => {
+                let result = self
+                    .import_chain(vec![block_with_peer.block.clone()], block_with_peer)
+                    .await
+                    .map_err(|e| e.to_string());
+                let _ = response.send(result);
+            }
             #[cfg(feature = "test-utils")]
             ChainOrchestratorCommand::SetGossip((enabled, tx)) => {
                 self.network.handle().set_gossip(enabled).await;
@@ -1229,6 +1238,7 @@ impl<
             chain,
             peer_id: block_with_peer.peer_id,
             signature: block_with_peer.signature,
+            result,
         })
     }
 

crates/chain-orchestrator/src/metrics.rs

@@ -1,5 +1,6 @@
 use metrics::Histogram;
 use metrics_derive::Metrics;
+use reth_scroll_primitives::ScrollBlock;
 use std::{collections::HashMap, time::Instant};
 use strum::{EnumIter, IntoEnumIterator};
 
@@ -20,17 +21,33 @@ impl MetricsHandler {
 
     /// Starts tracking a new block building task.
     pub(crate) fn start_block_building_recording(&mut self) {
-        if self.block_building_meter.start.is_some() {
+        if self.block_building_meter.block_building_start.is_some() {
             tracing::warn!(target: "scroll::chain_orchestrator", "block building recording is already ongoing, overwriting");
         }
-        self.block_building_meter.start = Some(Instant::now());
+        self.block_building_meter.block_building_start = Some(Instant::now());
     }
 
-    /// The duration of the current block building task if any.
-    pub(crate) fn finish_block_building_recording(&mut self) {
-        let duration = self.block_building_meter.start.take().map(|start| start.elapsed());
-        if let Some(duration) = duration {
-            self.block_building_meter.metric.block_building_duration.record(duration.as_secs_f64());
+    /// Finishes tracking the current block building task.
+    pub(crate) fn finish_block_building_recording(&mut self, block: Option<&ScrollBlock>) {
+        let now = Instant::now();
+        if let Some(t) = self.block_building_meter.block_building_start.take() {
+            let elapsed = now.duration_since(t).as_secs_f64();
+            self.block_building_meter.metric.all_block_building_duration.record(elapsed);
+
+            // Record only if it's not an empty block
+            let is_empty_block = block.map(|b| b.body.transactions.is_empty()).unwrap_or(true);
+            if !is_empty_block {
+                self.block_building_meter.metric.block_building_duration.record(elapsed);
+            }
+        }
+
+        if block.is_some() {
+            if let Some(t) = self.block_building_meter.last_block_building_time.replace(now) {
+                self.block_building_meter
+                    .metric
+                    .consecutive_block_interval
+                    .record(now.duration_since(t).as_secs_f64());
+            }
         }
     }
 }
@@ -104,13 +121,18 @@ pub(crate) struct ChainOrchestratorMetrics {
 #[derive(Debug, Default)]
 pub(crate) struct BlockBuildingMeter {
     metric: BlockBuildingMetric,
-    start: Option<Instant>,
+    block_building_start: Option<Instant>,
+    last_block_building_time: Option<Instant>,
 }
 
 /// Block building related metric.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "chain_orchestrator")]
 pub(crate) struct BlockBuildingMetric {
-    /// The duration of the block building task.
+    /// The duration of the block building task without empty block
     block_building_duration: Histogram,
+    /// The duration of the block building task for all blocks include empty block
+    all_block_building_duration: Histogram,
+    /// The duration of the block interval include empty block
+    consecutive_block_interval: Histogram,
 }

crates/node/Cargo.toml

@@ -19,9 +19,11 @@ async-trait.workspace = true
 
 # alloy
 alloy-chains.workspace = true
+alloy-eips.workspace = true
 alloy-primitives.workspace = true
 alloy-provider.workspace = true
 alloy-rpc-client.workspace = true
+alloy-rpc-types-engine.workspace = true
 alloy-signer-local.workspace = true
 alloy-signer-aws = "1.0.30"
 alloy-signer = "1.0.30"
@@ -59,6 +61,7 @@ reth-rpc-api.workspace = true
 reth-rpc-eth-api.workspace = true
 reth-rpc-eth-types.workspace = true
 reth-tasks.workspace = true
+reth-tokio-util.workspace = true
 reth-transaction-pool.workspace = true
 reth-trie-db.workspace = true
 
@@ -76,16 +79,13 @@ aws-config = "1.8.0"
 aws-sdk-kms = "1.76.0"
 
 # test-utils
-alloy-eips = { workspace = true, optional = true }
 alloy-rpc-types-eth = { workspace = true, optional = true }
-alloy-rpc-types-engine = { workspace = true, optional = true }
 reth-e2e-test-utils = { workspace = true, optional = true }
 reth-engine-local = { workspace = true, optional = true }
 reth-provider = { workspace = true, optional = true }
 reth-rpc-layer = { workspace = true, optional = true }
 reth-rpc-server-types = { workspace = true, optional = true }
 reth-storage-api = { workspace = true, optional = true }
-reth-tokio-util = { workspace = true, optional = true }
 scroll-alloy-rpc-types-engine = { workspace = true, optional = true }
 scroll-alloy-rpc-types.workspace = true
 
@@ -154,14 +154,11 @@ test-utils = [
     "reth-e2e-test-utils",
     "reth-rpc-server-types",
     "reth-rpc-layer",
-    "reth-tokio-util",
     "scroll-alloy-rpc-types-engine",
-    "alloy-rpc-types-engine",
     "reth-primitives-traits/test-utils",
     "reth-network-p2p/test-utils",
     "rollup-node-chain-orchestrator/test-utils",
     "scroll-network/test-utils",
-    "alloy-eips",
     "reth-storage-api",
     "alloy-rpc-types-eth",
 ]

crates/node/src/add_ons/mod.rs

@@ -32,6 +32,9 @@ use std::sync::Arc;
 mod handle;
 pub use handle::ScrollAddOnsHandle;
 
+mod remote_block_source;
+pub use remote_block_source::RemoteBlockSourceAddOn;
+
 mod rpc;
 pub use rpc::{
     RollupNodeAdminApiClient, RollupNodeAdminApiServer, RollupNodeApiClient, RollupNodeApiServer,
@@ -133,6 +136,8 @@ where
 
         let (tx, rx) = tokio::sync::oneshot::channel();
         let rpc_config = rollup_node_manager_addon.config().rpc_args.clone();
+        let remote_block_source_config =
+            rollup_node_manager_addon.config().remote_block_source_args.clone();
 
         // Register rollupNode API and rollupNodeAdmin API if enabled
         let rollup_node_rpc_ext = Arc::new(RollupNodeRpcExt::<N::Network>::new(rx));
@@ -161,6 +166,22 @@ where
                 .map_err(|_| eyre::eyre!("failed to send rollup manager handle"))?;
         }
 
+        // Launch remote block source if enabled
+        if remote_block_source_config.enabled {
+            let remote_source = RemoteBlockSourceAddOn::new(
+                remote_block_source_config,
+                rollup_manager_handle.clone(),
+            )
+            .await?;
+            ctx.node
+                .task_executor()
+                .spawn_critical_with_shutdown_signal("remote_block_source", |shutdown| async move {
+                    if let Err(e) = remote_source.run_until_shutdown(shutdown).await {
+                        tracing::error!(target: "scroll::remote_source", ?e, "Remote block source failed");
+                    }
+                });
+        }
+
         Ok(ScrollAddOnsHandle { rollup_manager_handle, rpc_handle })
     }
 }