STR-1684: EE P2P broadcast/gossip of head block hash

Cargo.toml

@@ -59,7 +59,6 @@ members = [
   "crates/test-utils/test-utils",
   "crates/test-utils/tx-indexer",
   "crates/util/mmr",
-  "crates/db-tests",
   "provers/risc0",
   "provers/sp1",
 
@@ -240,6 +239,7 @@ reth-db = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-engine-local = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-engine-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-errors = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
+reth-eth-wire = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-ethereum-engine-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-ethereum-forks = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-ethereum-payload-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
@@ -248,7 +248,9 @@ reth-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-evm-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-exex = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-ipc = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
+reth-network = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-network-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
+reth-network-p2p = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-node-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-node-core = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
@@ -262,6 +264,7 @@ reth-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0",
 reth-primitives-traits = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0", default-features = false }
 reth-provider = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-revm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
+reth-rlp = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-rpc = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-rpc-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
 reth-rpc-eth-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.6.0" }
@@ -349,6 +352,7 @@ terrors = "0.3.0"
 thiserror = "2.0.11"
 threadpool = "1.8"
 tokio = { version = "1.37", features = ["full"] }
+tokio-stream = "0.1.17"
 toml = "0.5"
 tower = "0.5"
 tracing = "0.1"

bin/alpen-reth/Cargo.toml

@@ -33,10 +33,16 @@ reth-cli-commands.workspace = true
 reth-cli-runner.workspace = true
 reth-cli-util.workspace = true
 reth-db.workspace = true
+reth-network.workspace = true
+reth-network-api.workspace = true
 reth-node-builder.workspace = true
 reth-node-core.workspace = true
+reth-transaction-pool.workspace = true
+reth-primitives.workspace = true
+futures-util.workspace = true
 rockbound = { workspace = true, optional = true }
 sled = { workspace = true, optional = true }
 strata-db-store-rocksdb = { workspace = true, optional = true }
+tokio.workspace = true
 tracing.workspace = true
 typed-sled = { workspace = true, optional = true }

bin/alpen-reth/src/main.rs

@@ -12,19 +12,32 @@ compile_error!(
 
 mod init_db;
 
-use std::sync::Arc;
+use std::{collections::HashMap, process, sync::Arc};
 
 use alpen_chainspec::{chain_value_parser, AlpenChainSpecParser};
 use alpen_reth_exex::{ProverWitnessGenerator, StateDiffGenerator};
-use alpen_reth_node::{args::AlpenNodeArgs, AlpenEthereumNode};
+use alpen_reth_node::{
+    args::AlpenNodeArgs,
+    head_gossip::{
+        connection::HeadGossipCommand,
+        handler::{HeadGossipEvent, HeadGossipProtocolHandler, HeadGossipState},
+    },
+    AlpenEthereumNode,
+};
 use alpen_reth_rpc::{AlpenRPC, StrataRpcApiServer};
 use clap::Parser;
+use futures_util::StreamExt;
 use init_db::init_witness_db;
 use reth_chainspec::ChainSpec;
 use reth_cli_commands::{launcher::FnLauncher, node::NodeCommand};
 use reth_cli_runner::CliRunner;
+use reth_network::{protocol::IntoRlpxSubProtocol, NetworkProtocols};
+use reth_network_api::PeerId;
 use reth_node_builder::{NodeBuilder, WithLaunchContext};
 use reth_node_core::args::LogArgs;
+use reth_primitives::Header;
+use reth_transaction_pool::{FullTransactionEvent, TransactionPool};
+use tokio::sync::mpsc;
 use tracing::info;
 
 fn main() {
@@ -93,11 +106,66 @@ fn main() {
             });
 
             let handle = node_builder.launch().await?;
+
+            // Create and add the custom RLPx subprotocol
+            // See: crates/reth/node/src/head_gossip/
+            let (gossip_tx, mut gossip_rx) = mpsc::unbounded_channel();
+            let handler = HeadGossipProtocolHandler {
+                state: HeadGossipState { events: gossip_tx },
+            };
+            handle
+                .node
+                .network
+                .add_rlpx_sub_protocol(handler.into_rlpx_sub_protocol());
+
+            // This task will handle gossip events
+            let mut tx_events = handle.node.pool.all_transactions_event_listener();
+            tokio::spawn(async move {
+                let mut connections: HashMap<PeerId, mpsc::UnboundedSender<HeadGossipCommand>> =
+                    HashMap::new();
+                loop {
+                    tokio::select! {
+                        Some(event) = gossip_rx.recv() => {
+                            match event {
+                                HeadGossipEvent::Established {
+                                    peer_id,
+                                    to_connection,
+                                    ..
+                                } => {
+                                    connections.insert(peer_id, to_connection);
+                                }
+                                HeadGossipEvent::Closed { peer_id } => {
+                                    connections.remove(&peer_id);
+                                }
+                                HeadGossipEvent::HeadHash { peer_id, header } => {
+                                    // TODO(@storopoli): do something when we receive a single `Header`.
+                                    info!(target: "head-gossip", "Received head hash from peer {}: {:?}", peer_id, header.hash_slow());
+                                }
+                                HeadGossipEvent::HeadHashes { peer_id, headers } => {
+                                    // TODO(@storopoli): do something when we receive multiple `Header`s.
+                                    info!(target: "head-gossip", "Received {} head hashes from peer {}", headers.len(), peer_id);
+                                }
+                            }
+                        },
+                        Some(event) = tx_events.next() => {
+                            // TODO(@storopoli): How to get a `Header`?
+                            if let FullTransactionEvent::Mined { block_hash, .. } = event {
+                                info!(target: "head-gossip", "New block mined: {:?}, broadcasting to {} peers", block_hash, connections.len());
+                                let header = Header::default();
+                                for sender in connections.values() {
+                                    let _ = sender.send(HeadGossipCommand::SendHeadHash(header.clone()));
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+
             handle.node_exit_future.await
         },
     ) {
         eprintln!("Error: {err:?}");
-        std::process::exit(1);
+        process::exit(1);
     }
 }
 

crates/reth/node/Cargo.toml

@@ -14,17 +14,23 @@ alpen-reth-rpc.workspace = true
 alloy-consensus.workspace = true
 alloy-eips.workspace = true
 alloy-network.workspace = true
+alloy-primitives.workspace = true
+alloy-rlp.workspace = true
 alloy-rpc-types.workspace = true
 eyre.workspace = true
+futures.workspace = true
 reth-basic-payload-builder.workspace = true
 reth-chain-state.workspace = true
 reth-chainspec.workspace = true
 reth-errors.workspace = true
+reth-eth-wire.workspace = true
 reth-ethereum-engine-primitives.workspace = true
 reth-ethereum-payload-builder.workspace = true
 reth-ethereum-primitives.workspace = true
 reth-evm.workspace = true
 reth-evm-ethereum.workspace = true
+reth-network.workspace = true
+reth-network-api.workspace = true
 reth-node-api.workspace = true
 reth-node-builder.workspace = true
 reth-node-ethereum.workspace = true
@@ -38,4 +44,6 @@ reth-trie-db.workspace = true
 revm.workspace = true
 revm-primitives.workspace = true
 serde.workspace = true
+tokio.workspace = true
+tokio-stream.workspace = true
 tracing.workspace = true