feat: integrate payload builder in engine api impl (#9710)

Cargo.lock

@@ -7144,6 +7144,7 @@ dependencies = [
  "reth-evm-ethereum",
  "reth-exex-types",
  "reth-network-p2p",
+ "reth-payload-builder",
  "reth-payload-validator",
  "reth-primitives",
  "reth-provider",

crates/engine/tree/src/tree/mod.rs

@@ -18,6 +18,7 @@ use reth_consensus::{Consensus, PostExecutionInput};
 use reth_engine_primitives::EngineTypes;
 use reth_errors::{ConsensusError, ProviderResult};
 use reth_evm::execute::{BlockExecutorProvider, Executor};
+use reth_payload_builder::PayloadBuilderHandle;
 use reth_payload_primitives::PayloadTypes;
 use reth_payload_validator::ExecutionPayloadValidator;
 use reth_primitives::{
@@ -39,7 +40,6 @@ use reth_stages_api::ControlFlow;
 use reth_trie::HashedPostState;
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
-    marker::PhantomData,
     sync::{mpsc::Receiver, Arc},
 };
 use tokio::sync::{
@@ -259,8 +259,9 @@ pub struct EngineApiTreeHandlerImpl<P, E, T: EngineTypes> {
     /// Keeps track of the state of the canonical chain that isn't persisted yet.
     /// This is intended to be accessed from external sources, such as rpc.
     canonical_in_memory_state: CanonicalInMemoryState,
-    /// Marker for the engine types.
+    /// Handle to the payload builder that will receive payload attributes for valid forkchoice
-    _marker: PhantomData<T>,
+    /// updates
+    payload_builder: PayloadBuilderHandle<T>,
 }
 
 impl<P, E, T> EngineApiTreeHandlerImpl<P, E, T>
@@ -280,6 +281,7 @@ where
         state: EngineApiTreeState,
         header: SealedHeader,
         persistence: PersistenceHandle,
+        payload_builder: PayloadBuilderHandle<T>,
     ) -> Self {
         Self {
             provider,
@@ -293,7 +295,7 @@ where
             is_backfill_active: false,
             state,
             canonical_in_memory_state: CanonicalInMemoryState::with_head(header),
-            _marker: PhantomData,
+            payload_builder,
         }
     }
 
@@ -308,6 +310,7 @@ where
         payload_validator: ExecutionPayloadValidator,
         incoming: Receiver<FromEngine<BeaconEngineMessage<T>>>,
         persistence: PersistenceHandle,
+        payload_builder: PayloadBuilderHandle<T>,
     ) -> UnboundedReceiver<EngineApiEvent> {
         let (tx, outgoing) = tokio::sync::mpsc::unbounded_channel();
         let state = EngineApiTreeState::new(
@@ -328,6 +331,7 @@ where
             state,
             header,
             persistence,
+            payload_builder,
         );
         std::thread::Builder::new().name("Tree Task".to_string()).spawn(|| task.run()).unwrap();
         outgoing
@@ -1190,6 +1194,7 @@ mod tests {
     use reth_chainspec::{ChainSpecBuilder, MAINNET};
     use reth_ethereum_engine_primitives::EthEngineTypes;
     use reth_evm::test_utils::MockExecutorProvider;
+    use reth_payload_builder::PayloadServiceCommand;
     use reth_provider::test_utils::MockEthProvider;
     use std::sync::mpsc::{channel, Sender};
     use tokio::sync::mpsc::unbounded_channel;
@@ -1199,6 +1204,7 @@ mod tests {
         to_tree_tx: Sender<FromEngine<BeaconEngineMessage<EthEngineTypes>>>,
         blocks: Vec<ExecutedBlock>,
         sf_action_rx: Receiver<StaticFileAction>,
+        payload_command_rx: UnboundedReceiver<PayloadServiceCommand<EthEngineTypes>>,
     }
 
     fn get_default_test_harness(number_of_blocks: u64) -> TestHarness {
@@ -1249,6 +1255,9 @@ mod tests {
         };
 
         let header = blocks.first().unwrap().block().header.clone();
+
+        let (to_payload_service, payload_command_rx) = unbounded_channel();
+        let payload_builder = PayloadBuilderHandle::new(to_payload_service);
         let mut tree = EngineApiTreeHandlerImpl::new(
             provider,
             executor_factory,
@@ -1259,20 +1268,21 @@ mod tests {
             engine_api_tree_state,
             header,
             persistence_handle,
+            payload_builder,
         );
         let last_executed_block = blocks.last().unwrap().clone();
         let pending = Some(BlockState::new(last_executed_block));
         tree.canonical_in_memory_state =
             CanonicalInMemoryState::new(state_by_hash, hash_by_number, pending);
 
-        TestHarness { tree, to_tree_tx, blocks, sf_action_rx }
+        TestHarness { tree, to_tree_tx, blocks, sf_action_rx, payload_command_rx }
     }
 
     #[tokio::test]
     async fn test_tree_persist_blocks() {
         // we need more than PERSISTENCE_THRESHOLD blocks to trigger the
         // persistence task.
-        let TestHarness { tree, to_tree_tx, sf_action_rx, mut blocks } =
+        let TestHarness { tree, to_tree_tx, sf_action_rx, mut blocks, payload_command_rx } =
             get_default_test_harness(PERSISTENCE_THRESHOLD + 1);
         std::thread::Builder::new().name("Tree Task".to_string()).spawn(|| tree.run()).unwrap();
 
@@ -1292,7 +1302,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_in_memory_state_trait_impl() {
-        let TestHarness { tree, to_tree_tx, sf_action_rx, blocks } = get_default_test_harness(10);
+        let TestHarness { tree, to_tree_tx, sf_action_rx, blocks, payload_command_rx } =
+            get_default_test_harness(10);
 
         let head_block = blocks.last().unwrap().block();
         let first_block = blocks.first().unwrap().block();

crates/ethereum/engine/Cargo.toml

@@ -24,6 +24,7 @@ reth-provider.workspace = true
 reth-prune.workspace = true
 reth-stages-api.workspace = true
 reth-tasks.workspace = true
+reth-payload-builder.workspace = true
 
 # async
 futures.workspace = true

crates/ethereum/engine/src/service.rs

@@ -17,6 +17,7 @@ pub use reth_engine_tree::{
 use reth_ethereum_engine_primitives::EthEngineTypes;
 use reth_evm_ethereum::execute::EthExecutorProvider;
 use reth_network_p2p::{bodies::client::BodiesClient, headers::client::HeadersClient};
+use reth_payload_builder::PayloadBuilderHandle;
 use reth_payload_validator::ExecutionPayloadValidator;
 use reth_provider::{providers::BlockchainProvider, ProviderFactory};
 use reth_prune::Pruner;
@@ -66,6 +67,7 @@ where
         provider: ProviderFactory<DB>,
         blockchain_db: BlockchainProvider<DB>,
         pruner: Pruner<DB, ProviderFactory<DB>>,
+        payload_builder: PayloadBuilderHandle<EthEngineTypes>,
     ) -> Self {
         let consensus = Arc::new(EthBeaconConsensus::new(chain_spec.clone()));
         let downloader = BasicBlockDownloader::new(client, consensus.clone());
@@ -83,6 +85,7 @@ where
             payload_validator,
             to_tree_rx,
             persistence_handle,
+            payload_builder,
         );
 
         let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
@@ -176,6 +179,7 @@ mod tests {
         let pruner =
             Pruner::<_, ProviderFactory<_>>::new(provider_factory.clone(), vec![], 0, 0, None, rx);
 
+        let (tx, _rx) = unbounded_channel();
         let _eth_service = EthService::new(
             chain_spec,
             client,
@@ -185,6 +189,7 @@ mod tests {
             provider_factory,
             blockchain_db,
             pruner,
+            PayloadBuilderHandle::new(tx),
         );
     }
 }

crates/ethereum/node/src/launch.rs

@@ -168,6 +168,7 @@ where
             ctx.provider_factory().clone(),
             ctx.blockchain_db().clone(),
             pruner,
+            ctx.components().payload_builder().clone(),
         );
 
         let event_sender = EventSender::default();

crates/payload/builder/src/lib.rs

@@ -115,7 +115,9 @@ pub mod test_utils;
 
 pub use events::Events;
 pub use reth_rpc_types::engine::PayloadId;
-pub use service::{PayloadBuilderHandle, PayloadBuilderService, PayloadStore};
+pub use service::{
+    PayloadBuilderHandle, PayloadBuilderService, PayloadServiceCommand, PayloadStore,
+};
 pub use traits::{KeepPayloadJobAlive, PayloadJob, PayloadJobGenerator};
 
 // re-export the Ethereum engine primitives for convenience