fix: wire ImportService

Cargo.toml

@@ -8,43 +8,43 @@ name = "reth_bsc"
 path = "src/lib.rs"
 
 [[bin]]
-name = "reth"
+name = "reth-bsc"
 path = "src/main.rs"
 
 [dependencies]
-reth = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-cli = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-cli-commands = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-basic-payload-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-db = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-chainspec = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-cli-util = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-discv4 = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7", features = ["test-utils"] }
-reth-engine-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-ethereum-forks = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7", features = ["serde"] }
-reth-ethereum-payload-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-ethereum-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-eth-wire = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-eth-wire-types = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-evm-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-node-core = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-revm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-network = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7", features = ["test-utils"] }
-reth-network-p2p = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-network-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-node-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7", features = ["test-utils"] }
-reth-network-peers = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-optimism-rpc = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-payload-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-primitives-traits = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-provider = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7", features = ["test-utils"] }
-reth-rpc-eth-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-rpc-engine-api = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-tracing = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-trie-common = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
-reth-trie-db = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.7" }
+reth = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-cli = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-cli-commands = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-basic-payload-builder = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-db = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-chainspec = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-cli-util = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-discv4 = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7", features = ["test-utils"] }
+reth-engine-primitives = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-ethereum-forks = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7", features = ["serde"] }
+reth-ethereum-payload-builder = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-ethereum-primitives = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-eth-wire = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-eth-wire-types = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-evm = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-evm-ethereum = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-node-core = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-revm = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-network = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7", features = ["test-utils"] }
+reth-network-p2p = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-network-api = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-node-ethereum = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7", features = ["test-utils"] }
+reth-network-peers = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-optimism-rpc = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-payload-primitives = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-primitives = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-primitives-traits = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-provider = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7", features = ["test-utils"] }
+reth-rpc-eth-api = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-rpc-engine-api = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-tracing = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-trie-common = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
+reth-trie-db = { git = "https://github.com/paradigmxyz/reth", rev = "66692a7" }
 revm = { version = "24.0.1", features = ["secp256r1"] }
 
 # alloy dependencies

rustfmt.toml

@@ -0,0 +1,11 @@
+reorder_imports = true
+imports_granularity = "Crate"
+use_small_heuristics = "Max"
+comment_width = 100
+wrap_comments = true
+binop_separator = "Back"
+trailing_comma = "Vertical"
+trailing_semicolon = false
+use_field_init_shorthand = true
+format_code_in_doc_comments = true
+doc_comment_code_block_width = 100

src/main.rs

@@ -2,12 +2,8 @@ use clap::{Args, Parser};
 use reth::builder::NodeHandle;
 use reth_bsc::{
     chainspec::parser::BscChainSpecParser,
-    consensus::ParliaConsensus,
-    node::network::block_import::service::ImportService as BlockImportService,
     node::{cli::Cli, BscNode},
 };
-use std::sync::Arc;
-use tracing::error;
 
 // We use jemalloc for performance reasons
 #[cfg(all(feature = "jemalloc", unix))]
@@ -28,19 +24,11 @@ fn main() -> eyre::Result<()> {
     }
 
     Cli::<BscChainSpecParser, NoArgs>::parse().run(|builder, _| async move {
-        let NodeHandle {
-            node,
-            node_exit_future: exit_future,
-        } = builder.node(BscNode::default()).launch().await?;
-        let provider = node.provider.clone();
-        let consensus = Arc::new(ParliaConsensus { provider });
-        let (service, _) = BlockImportService::new(consensus, node.beacon_engine_handle.clone());
+        let (node, engine_handle_tx) = BscNode::new();
+        let NodeHandle { node, node_exit_future: exit_future } =
+            builder.node(node).launch().await?;
 
-        node.task_executor.spawn(async move {
-            if let Err(e) = service.await {
-                error!("Import service error: {}", e);
-            }
-        });
+        engine_handle_tx.send(node.beacon_engine_handle.clone()).unwrap();
 
         exit_future.await
     })?;

src/node/mod.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::{chainspec::BscChainSpec, node::rpc::BscEthApiBuilder};
 
 use crate::node::rpc::engine_api::{
@@ -14,10 +16,12 @@ use reth::{
         DebugNode, Node, NodeAdapter, NodeComponentsBuilder,
     },
 };
+use reth_engine_primitives::BeaconConsensusEngineHandle;
 use reth_node_ethereum::node::{EthereumPayloadBuilder, EthereumPoolBuilder};
 use reth_primitives::{Block, BlockBody, EthPrimitives};
 use reth_provider::EthStorage;
 use reth_trie_db::MerklePatriciaTrie;
+use tokio::sync::{oneshot, Mutex};
 
 pub mod cli;
 pub mod consensus;
@@ -30,9 +34,18 @@ pub type BscNodeAddOns<N> =
     RpcAddOns<N, BscEthApiBuilder, BscEngineValidatorBuilder, BscEngineApiBuilder>;
 
 /// Type configuration for a regular BSC node.
-#[derive(Debug, Default, Clone)]
-#[non_exhaustive]
-pub struct BscNode;
+#[derive(Debug, Clone)]
+pub struct BscNode {
+    engine_handle_rx:
+        Arc<Mutex<Option<oneshot::Receiver<BeaconConsensusEngineHandle<BscPayloadTypes>>>>>,
+}
+
+impl BscNode {
+    pub fn new() -> (Self, oneshot::Sender<BeaconConsensusEngineHandle<BscPayloadTypes>>) {
+        let (tx, rx) = oneshot::channel();
+        (Self { engine_handle_rx: Arc::new(Mutex::new(Some(rx))) }, tx)
+    }
+}
 
 impl BscNode {
     pub fn components<Node>(
@@ -59,21 +72,14 @@ impl BscNode {
             .pool(EthereumPoolBuilder::default())
             .executor(BscExecutorBuilder::default())
             .payload(BasicPayloadServiceBuilder::default())
-            .network(BscNetworkBuilder::default())
+            .network(BscNetworkBuilder { engine_handle_rx: self.engine_handle_rx.clone() })
             .consensus(BscConsensusBuilder::default())
     }
 }
 
 impl<N> Node<N> for BscNode
 where
-    N: FullNodeTypes<
-        Types: NodeTypes<
-            Payload = BscPayloadTypes,
-            ChainSpec = BscChainSpec,
-            Primitives = EthPrimitives,
-            Storage = EthStorage,
-        >,
-    >,
+    N: FullNodeTypes<Types = Self>,
 {
     type ComponentsBuilder = ComponentsBuilder<
         N,

src/node/network/block_import/handle.rs

@@ -14,32 +14,33 @@ use super::service::{BlockMsg, ImportEvent, IncomingBlock, Outcome};
 /// [`super::service::ImportService`]:
 /// - Blocks can be sent to the service for import via [`send_block`](ImportHandle::send_block)
 /// - Import outcomes can be received via [`poll_outcome`](ImportHandle::poll_outcome)`
-pub struct ImportHandle<T: PayloadTypes> {
+#[derive(Debug)]
+pub struct ImportHandle {
     /// Send the new block to the service
-    to_import: UnboundedSender<IncomingBlock<T>>,
+    to_import: UnboundedSender<IncomingBlock>,
     /// Receive the event(Announcement/Outcome) of the import
-    import_outcome: UnboundedReceiver<ImportEvent<T>>,
+    import_outcome: UnboundedReceiver<ImportEvent>,
 }
 
-impl<T: PayloadTypes> ImportHandle<T> {
+impl ImportHandle {
     /// Create a new handle with the provided channels
     pub fn new(
-        to_import: UnboundedSender<IncomingBlock<T>>,
-        import_outcome: UnboundedReceiver<ImportEvent<T>>,
+        to_import: UnboundedSender<IncomingBlock>,
+        import_outcome: UnboundedReceiver<ImportEvent>,
     ) -> Self {
         Self { to_import, import_outcome }
     }
 
     /// Sends the block to import to the service.
     /// Returns a [`BlockImportError`] if the channel to the import service is closed.
-    pub fn send_block(&self, block: BlockMsg<T>, peer_id: PeerId) -> Result<(), BlockImportError> {
+    pub fn send_block(&self, block: BlockMsg, peer_id: PeerId) -> Result<(), BlockImportError> {
         self.to_import
             .send((block, peer_id))
             .map_err(|_| BlockImportError::Other("block import service channel closed".into()))
     }
 
     /// Poll for the next import event
-    pub fn poll_outcome(&mut self, cx: &mut Context<'_>) -> Poll<Option<ImportEvent<T>>> {
+    pub fn poll_outcome(&mut self, cx: &mut Context<'_>) -> Poll<Option<ImportEvent>> {
         self.import_outcome.poll_recv(cx)
     }
 }

src/node/network/block_import/mod.rs

@@ -11,39 +11,33 @@ use std::{
     task::{ready, Context, Poll},
 };
 
+use crate::node::network::BscNewBlock;
+
 pub mod handle;
 pub mod service;
 
-pub struct BscBlockImport<T: PayloadTypes> {
-    handle: ImportHandle<T>,
+#[derive(Debug)]
+pub struct BscBlockImport {
+    handle: ImportHandle,
 }
 
-impl<T: PayloadTypes> BscBlockImport<T> {
-    pub fn new(handle: ImportHandle<T>) -> Self {
+impl BscBlockImport {
+    pub fn new(handle: ImportHandle) -> Self {
         Self { handle }
     }
 }
 
-impl<T: PayloadTypes> BlockImport<BscBlock<T>> for BscBlockImport<T> {
-    fn on_new_block(&mut self, peer_id: PeerId, incoming_block: NewBlockEvent<BscBlock<T>>) {
+impl BlockImport<BscNewBlock> for BscBlockImport {
+    fn on_new_block(&mut self, peer_id: PeerId, incoming_block: NewBlockEvent<BscNewBlock>) {
         if let NewBlockEvent::Block(block) = incoming_block {
             let _ = self.handle.send_block(block, peer_id);
         }
     }
 
-    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<ImportEvent<T>> {
+    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<ImportEvent> {
         match ready!(self.handle.poll_outcome(cx)) {
             Some(outcome) => Poll::Ready(outcome),
             None => Poll::Pending,
         }
     }
 }
-
-impl<T: PayloadTypes> fmt::Debug for BscBlockImport<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("BscBlockImport")
-            .field("engine_handle", &"BeaconConsensusEngineHandle")
-            .field("service_handle", &"BscBlockImportHandle")
-            .finish()
-    }
-}