introduce NetworkManager component

Author: frisitano

Cargo.toml

@@ -6,8 +6,8 @@ license = "MIT OR Apache-2.0"
 exclude = [".github/"]
 
 [workspace]
-members = [
+members = [ "crates/network",
     "crates/scroll-wire"
 ]
 
-resolver = "2"
+resolver = "2"

crates/network/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "network"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+exclude.workspace = true
+
+[dependencies]
+reth-eth-wire-types = { path = "../../../reth/crates/net/eth-wire-types" }
+reth-network = { path = "../../../reth/crates/net/network" }
+reth-network-api = { path = "../../../reth/crates/net/network-api" }
+reth-network-types = { path = "../../../reth/crates/net/network-types" }
+reth-network-peers = { path = "../../../reth/crates/net/peers" }
+reth-primitives = { path = "../../../reth/crates/primitives" }
+reth-discv5 = { path = "../../../reth/crates/net/discv5" }
+reth-tasks = { path = "../../../reth/crates/tasks" }
+reth-storage-api = { path = "../../../reth/crates/storage/storage-api" }
+tokio = {version = "1.39", default-features = false, features = ["full"] }
+tokio-stream = "0.1"
+scroll-wire = { path = "../scroll-wire" }
+alloy-primitives = { version = "0.8.15", default-features = false, features = [
+    "map-foldhash",
+] }
+secp256k1 = { version = "0.29", default-features = false, features = ["global-context", "rand-std", "recovery"] }
+parking_lot = "0.12"
+futures = "0.3"
+reth-scroll-chainspec = { path = "../../../reth/crates/scroll/chainspec" }
+

crates/network/src/handle.rs

@@ -0,0 +1,82 @@
+use alloy_primitives::PrimitiveSignature;
+use reth_network::NetworkHandle as RethNetworkHandle;
+use reth_network_peers::PeerId;
+use reth_primitives::Block;
+use secp256k1::SecretKey;
+use std::sync::Arc;
+use tokio::sync::{mpsc::UnboundedSender, oneshot};
+
+/// A _sharable_ frontend used to communicate with the [`NetworkManager`].
+#[derive(Debug, Clone)]
+pub struct NetworkHandle {
+    /// A reference to the inner network handle.
+    pub(crate) inner: Arc<NetworkInner>,
+}
+
+impl NetworkHandle {
+    /// Creates a new [`NetworkHandle`] instance from the given [`UnboundedSender`] and [`RethNetworkHandle`].
+    pub fn new(
+        to_manager_tx: UnboundedSender<NetworkHandleMessage>,
+        inner_network_handle: RethNetworkHandle,
+    ) -> Self {
+        let inner = NetworkInner {
+            to_manager_tx,
+            inner_network_handle,
+        };
+        Self {
+            inner: Arc::new(inner),
+        }
+    }
+}
+
+/// The inner state of the [`NetworkHandle`].
+#[derive(Debug)]
+pub struct NetworkInner {
+    /// The sender half of the channel set up between this type and the [`NetworkManager`].
+    pub(crate) to_manager_tx: UnboundedSender<NetworkHandleMessage>,
+    /// A reference to the inner network handle which is used to communicate with the inner network.
+    pub inner_network_handle: RethNetworkHandle,
+}
+
+impl NetworkHandle {
+    /// Returns a reference to the inner network handle.
+    pub fn inner(&self) -> &RethNetworkHandle {
+        &self.inner.inner_network_handle
+    }
+
+    /// Returns the peer id of the network handle.
+    pub fn peer_id(&self) -> &PeerId {
+        &self.inner.inner_network_handle.peer_id()
+    }
+
+    /// Sends a message to the network manager.
+    pub fn send_message(&self, msg: NetworkHandleMessage) {
+        let _ = self.inner.to_manager_tx.send(msg);
+    }
+
+    /// Announces a block to the network.
+    pub fn announce_block(&self, block: Block, signature: PrimitiveSignature) {
+        self.send_message(NetworkHandleMessage::AnnounceBlock { block, signature });
+    }
+
+    /// Shuts down the network handle.
+    pub async fn shutdown(&self) -> Result<(), oneshot::error::RecvError> {
+        let (tx, rx) = oneshot::channel();
+        self.send_message(NetworkHandleMessage::Shutdown(tx));
+        rx.await
+    }
+
+    /// Returns the secret key of the network handle.
+    pub fn secret_key(&self) -> &SecretKey {
+        &self.inner.inner_network_handle.secret_key()
+    }
+}
+
+/// A message type used for communication between the [`NetworkHandle`] and the [`super::NetworkManager`].
+pub enum NetworkHandleMessage {
+    AnnounceBlock {
+        block: Block,
+        signature: PrimitiveSignature,
+    },
+    Shutdown(oneshot::Sender<()>),
+}

crates/network/src/import.rs

@@ -0,0 +1,69 @@
+use alloy_primitives::PrimitiveSignature;
+use reth_network_peers::PeerId;
+use scroll_wire::NewBlock;
+use std::task::{Context, Poll};
+
+/// A trait for importing new blocks from the network.
+pub trait BlockImport: std::fmt::Debug + Send + Sync {
+    /// Called when a new block is received from the network.
+    fn on_new_block(
+        &mut self,
+        peer_id: PeerId,
+        block: reth_primitives::Block,
+        signature: PrimitiveSignature,
+    );
+
+    /// Polls the block import type for results of block import.
+    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BlockImportOutcome>;
+}
+
+/// The outcome of a block import operation.
+pub struct BlockImportOutcome {
+    /// The peer that the block was received from.
+    pub peer: PeerId,
+    /// The result of the block import operation.
+    pub result: Result<BlockValidation, BlockImportError>,
+}
+
+/// The result of a block validation operation.
+pub enum BlockValidation {
+    /// The block header is valid.
+    ValidHeader { new_block: NewBlock },
+    /// The block is valid.
+    ValidBlock { new_block: NewBlock },
+}
+
+/// An error that can occur during block import.
+pub enum BlockImportError {
+    /// An error occurred during consensus.
+    Consensus(ConsensusError),
+}
+
+/// A consensus related error that can occur during block import.
+pub enum ConsensusError {
+    /// The block is invalid.
+    InvalidBlock,
+    /// The state root is invalid.
+    InvalidStateRoot,
+    /// The signature is invalid.
+    InvalidSignature,
+}
+
+/// A block import type that does nothing.
+#[derive(Debug)]
+pub struct NoopBlockImport;
+
+impl BlockImport for NoopBlockImport {
+    fn on_new_block(
+        &mut self,
+        _peer_id: PeerId,
+        _block: reth_primitives::Block,
+        _signature: PrimitiveSignature,
+    ) {
+        println!("received new block");
+    }
+
+    fn poll(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<BlockImportOutcome> {
+        std::task::Poll::Pending
+    }
+}

crates/network/src/lib.rs

@@ -0,0 +1,10 @@
+mod import;
+pub use import::{BlockImport, BlockImportOutcome, BlockValidation, NoopBlockImport};
+
+mod handle;
+use handle::{NetworkHandle, NetworkHandleMessage};
+
+mod manager;
+pub use manager::NetworkManager;
+
+pub use reth_network::NetworkConfigBuilder;

crates/network/src/main.rs

@@ -0,0 +1,43 @@
+use alloy_primitives::PrimitiveSignature;
+use network::{NetworkConfigBuilder, NetworkManager, NoopBlockImport};
+use reth_network::{NetworkInfo, Peers};
+use reth_scroll_chainspec::SCROLL_MAINNET;
+
+#[tokio::main]
+async fn main() {
+    let config_1 =
+        NetworkConfigBuilder::<reth_network::EthNetworkPrimitives>::with_rng_secret_key()
+            .disable_discovery()
+            .build_with_noop_provider((*SCROLL_MAINNET).clone());
+    let network_1 = NetworkManager::new(config_1, NoopBlockImport).await;
+    let network_1_handle = network_1.handle();
+    let peer_1_id = *network_1_handle.peer_id();
+    let peer_1_addr = network_1_handle.inner().local_addr();
+
+    let config_2 =
+        NetworkConfigBuilder::<reth_network::EthNetworkPrimitives>::with_rng_secret_key()
+            .disable_discovery()
+            .listener_addr(std::net::SocketAddr::V4(std::net::SocketAddrV4::new(
+                std::net::Ipv4Addr::UNSPECIFIED,
+                0,
+            )))
+            .build_with_noop_provider((*SCROLL_MAINNET).clone());
+    let network_2 = NetworkManager::new(config_2, NoopBlockImport).await;
+    let network_2_handle = network_2.handle();
+    let peer_2_id = *network_2_handle.peer_id();
+    let peer_2_addr = network_2_handle.inner().local_addr();
+
+    tokio::spawn(network_1);
+    tokio::spawn(network_2);
+
+    network_1_handle.inner().add_peer(peer_2_id, peer_2_addr);
+    network_2_handle.inner().add_peer(peer_1_id, peer_1_addr);
+
+    let signature = PrimitiveSignature::try_from(&[0u8; 65][..]).unwrap();
+    let block = reth_primitives::Block::default();
+
+    for _ in 0..100 {
+        network_1_handle.announce_block(block.clone(), signature);
+        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+    }
+}