feat: Let followers import zone blocks from leader sequencers

Cargo.toml

@@ -129,6 +129,7 @@ zone-sequencer = { path = "crates/sequencer" }
 
 # reth
 reth-basic-payload-builder = { git = "https://github.com/paradigmxyz/reth", rev = "1bf2384" }
+reth-chain-state = { git = "https://github.com/paradigmxyz/reth", rev = "1bf2384" }
 reth-chainspec = { git = "https://github.com/paradigmxyz/reth", rev = "1bf2384" }
 reth-cli = { git = "https://github.com/paradigmxyz/reth", rev = "1bf2384" }
 reth-cli-util = { git = "https://github.com/paradigmxyz/reth", rev = "1bf2384" }

crates/node/Cargo.toml

@@ -31,6 +31,7 @@ zone-rpc.workspace = true
 zone-sequencer.workspace = true
 
 # reth
+reth-chain-state.workspace = true
 reth-chainspec.workspace = true
 reth-consensus = { workspace = true, optional = true }
 reth-eth-wire-types.workspace = true
@@ -59,6 +60,7 @@ alloy-genesis.workspace = true
 alloy-network.workspace = true
 alloy-primitives.workspace = true
 alloy-provider = { workspace = true, features = ["ws"] }
+alloy-rlp.workspace = true
 alloy-rpc-client.workspace = true
 alloy-rpc-types-engine.workspace = true
 alloy-rpc-types-eth.workspace = true
@@ -88,12 +90,11 @@ alloy = { workspace = true, features = [
 alloy-contract.workspace = true
 alloy-eips.workspace = true
 alloy-evm.workspace = true
-alloy-rlp.workspace = true
 alloy-signer.workspace = true
 base64.workspace = true
+const-hex.workspace = true
 commonware-codec.workspace = true
 commonware-cryptography.workspace = true
-const-hex.workspace = true
 p256.workspace = true
 rand.workspace = true
 reth-ethereum = { workspace = true, features = ["node", "test-utils"] }

crates/node/src/cli.rs

@@ -129,6 +129,12 @@ fn run_node(mut cli: Cli<ZoneChainSpecParser, ZoneArgs>) -> eyre::Result<()> {
         }
 
         let manifest_mode = p2p_config.is_some();
+        if manifest_mode {
+            // Replicate only durable blocks. Persist every block immediately so followers can
+            // acknowledge each block without waiting for Reth's in-memory buffer to fill.
+            builder.config_mut().engine.persistence_threshold = 0;
+            builder.config_mut().engine.memory_block_buffer_target = 0;
+        }
         let should_sequence_blocks = sequencer_enabled(args.enable_sequencer, manifest_role);
         let sequencer_signer = (should_sequence_blocks || manifest_mode)
             .then(|| {

crates/node/src/lib.rs

@@ -12,6 +12,7 @@ pub mod dev;
 pub mod engine;
 pub mod genesis;
 pub mod node;
+mod replication;
 pub mod rpc;
 
 pub use engine::ZoneEngine;

crates/node/src/node.rs

@@ -5,6 +5,7 @@
 
 use crate::{
     ZoneEngine,
+    replication::{broadcast_persisted_blocks, import_leader_blocks},
     rpc::{ZoneRpc, ZoneRpcApi, rpc_connection_config, start_private_rpc},
 };
 use alloy_primitives::Address;
@@ -67,7 +68,7 @@ use zone_l1::{
         spawn_policy_resolution_task, spawn_pool_prefetch_task,
     },
 };
-use zone_p2p::{P2pConfig, P2pNetworkId, spawn_p2p};
+use zone_p2p::{P2pConfig, P2pNetworkId, Role, spawn_p2p};
 use zone_payload::{
     DEFAULT_WITHDRAWAL_BATCH_INTERVAL_BLOCKS, WithdrawalRevealEncryptor, ZonePayloadAttributes,
     ZonePayloadFactory, ZonePayloadTypes,
@@ -466,14 +467,30 @@ where
             .erased();
 
         self.resolve_and_seed_tokens(&l1_provider).await?;
-        self.spawn_l1_subscriber(&ctx);
+        let p2p_role = self.p2p_config.as_ref().map(P2pConfig::role);
+        if p2p_role == Some(Role::Follower) {
+            // TODO(multi-sequencer): Split L1 observation/cache updates from deposit
+            // enqueueing. Followers import complete blocks from the leader and do not consume
+            // DepositQueue; starting the unified subscriber here would grow that queue forever.
+            // On promotion/restart the subscriber resumes from the tempoBlockNumber persisted in
+            // the follower's imported zone state.
+            info!(target: "reth::cli", "Skipping L1 deposit subscriber on follower");
+        } else {
+            self.spawn_l1_subscriber(&ctx);
+        }
         self.spawn_policy_tasks(&l1_provider, &ctx);
 
         let task_executor = ctx.node.task_executor().clone();
         if let Some(config) = self.p2p_config.take() {
             let network_id =
                 P2pNetworkId::new(l1_provider.get_chain_id().await?, self.portal_address);
-            Self::launch_p2p(config, network_id, &task_executor)?;
+            Self::launch_p2p(
+                config,
+                network_id,
+                &task_executor,
+                ctx.node.provider().clone(),
+                ctx.beacon_engine_handle.clone(),
+            )?;
         }
 
         if let Some(ref config) = self.sequencer_config {
@@ -527,16 +544,39 @@ where
         config: P2pConfig,
         network_id: P2pNetworkId,
         task_executor: &reth_tasks::TaskExecutor,
+        provider: N::Provider,
+        engine: reth_node_builder::ConsensusEngineHandle<ZonePayloadTypes>,
     ) -> eyre::Result<()> {
+        let role = config.role();
         let handle = spawn_p2p(config, network_id)?;
         let zone_p2p::P2pHandleParts {
             shutdown: shutdown_token,
             mut stopped,
             thread,
-            commands: _commands,
-            events: _events,
+            commands,
+            events,
         } = handle.into_parts();
 
+        match role {
+            Role::Leader => {
+                task_executor.spawn_critical_task(
+                    "zone-p2p-block-broadcast",
+                    broadcast_persisted_blocks(provider, commands),
+                );
+                // Leaders do not receive block messages. Dropping this receiver is harmless: the
+                // runtime only emits BlockReceived on followers.
+                drop(events);
+            }
+            Role::Follower => {
+                // Keep the command sender alive so the runtime's command loop remains available
+                // for later ACK/backfill commands even though followers send nothing in this PR.
+                task_executor.spawn_critical_task(
+                    "zone-p2p-block-import",
+                    import_leader_blocks(provider, engine, events, commands),
+                );
+            }
+        }
+
         task_executor.spawn_critical_with_graceful_shutdown_signal(
             "zone-p2p",
             |shutdown| async move {