reth_engine_tree/
launch.rs

1//! Engine orchestrator launch helper.
2//!
3//! Provides [`EngineOrchestratorBuilder`] which wires
4//! together all engine components and builds a
5//! [`ChainOrchestrator`] ready to be polled as a `Stream`, with the staged pipeline or a
6//! caller-supplied [`BackfillSync`] as backfill.
7
8use crate::{
9    backfill::{BackfillSync, PipelineSync},
10    chain::ChainOrchestrator,
11    download::BasicBlockDownloader,
12    engine::{EngineApiKind, EngineApiRequest, EngineApiRequestHandler, EngineHandler},
13    persistence::PersistenceHandle,
14    tree::{EngineApiTreeHandler, EngineValidator, TreeConfig, WaitForCaches},
15};
16use futures::Stream;
17use reth_consensus::FullConsensus;
18use reth_engine_primitives::BeaconEngineMessage;
19use reth_evm::ConfigureEvm;
20use reth_network_p2p::{BlockAccessListsClient, BlockClient};
21use reth_payload_builder::PayloadBuilderHandle;
22use reth_primitives_traits::NodePrimitives;
23use reth_provider::{
24    providers::{BlockchainProvider, ProviderNodeTypes},
25    ProviderFactory,
26};
27use reth_prune::PrunerWithFactory;
28use reth_stages_api::{MetricEventsSender, Pipeline};
29use reth_storage_overlay::OverlayManager;
30use reth_tasks::Runtime;
31use std::{convert::Infallible, sync::Arc};
32
33/// The [`ChainOrchestrator`] built by [`EngineOrchestratorBuilder`].
34pub type EngineOrchestrator<T, N, Client, S, B> = ChainOrchestrator<
35    EngineHandler<
36        EngineApiRequestHandler<EngineApiRequest<T, N>, N>,
37        S,
38        BasicBlockDownloader<Client, <N as NodePrimitives>::Block>,
39    >,
40    B,
41>;
42
43// An [`EngineOrchestrator`], or the error its backfill failed to build with.
44type EngineOrchestratorResult<T, N, Client, S, B, E> =
45    Result<EngineOrchestrator<T, N, Client, S, B>, E>;
46
47/// Components needed to build the engine [`ChainOrchestrator`] that drives the chain forward.
48///
49/// [`build`](Self::build) spawns and wires together the following components:
50///
51/// - **[`BasicBlockDownloader`]** — downloads blocks on demand from the network during live sync.
52/// - **[`PersistenceHandle`]** — spawns the persistence service on a background thread for writing
53///   blocks and performing pruning outside the critical consensus path.
54/// - **[`EngineApiTreeHandler`]** — spawns the tree handler that processes engine API requests
55///   (`newPayload`, `forkchoiceUpdated`) and maintains the in-memory chain state.
56/// - **[`EngineApiRequestHandler`]** + **[`EngineHandler`]** — glue that routes incoming CL
57///   messages to the tree handler and manages download requests.
58/// - **[`PipelineSync`]** — wraps the staged sync [`Pipeline`] for backfill sync when the node
59///   needs to catch up over large block ranges.
60///
61/// The returned orchestrator implements [`Stream`] and yields
62/// [`ChainEvent`]s.
63///
64/// [`ChainEvent`]: crate::chain::ChainEvent
65#[derive(Debug)]
66pub struct EngineOrchestratorBuilder<N, Client, S, V, C>
67where
68    N: ProviderNodeTypes,
69{
70    /// The engine API flavor to run.
71    pub engine_kind: EngineApiKind,
72    /// Consensus used to validate downloaded and incoming blocks.
73    pub consensus: Arc<dyn FullConsensus<N::Primitives>>,
74    /// Network client used to download blocks during live sync.
75    pub client: Client,
76    /// Incoming consensus layer messages.
77    pub incoming_requests: S,
78    /// Staged sync pipeline used for backfill.
79    pub pipeline: Pipeline<N>,
80    /// Runtime the pipeline runs on.
81    pub pipeline_task_spawner: Runtime,
82    /// Provider factory handed to the persistence service.
83    pub provider: ProviderFactory<N>,
84    /// Blockchain provider backing the engine tree.
85    pub blockchain_db: BlockchainProvider<N>,
86    /// Pruner run by the persistence service.
87    pub pruner: PrunerWithFactory<ProviderFactory<N>>,
88    /// Handle to the payload builder service.
89    pub payload_builder: PayloadBuilderHandle<N::Payload>,
90    /// Validator for incoming payloads.
91    pub payload_validator: V,
92    /// Overlay manager for state on top of the database.
93    pub overlay_manager: OverlayManager<N::Primitives>,
94    /// Engine tree configuration.
95    pub tree_config: TreeConfig,
96    /// Sender for sync metric events.
97    pub sync_metrics_tx: MetricEventsSender,
98    /// EVM configuration used to execute payloads.
99    pub evm_config: C,
100    /// Runtime used to spawn engine tree tasks.
101    pub runtime: Runtime,
102}
103
104impl<N, Client, S, V, C> EngineOrchestratorBuilder<N, Client, S, V, C>
105where
106    N: ProviderNodeTypes,
107    Client: BlockClient<Block = <N::Primitives as NodePrimitives>::Block>
108        + BlockAccessListsClient
109        + 'static,
110    S: Stream<Item = BeaconEngineMessage<N::Payload>> + Send + Sync + Unpin + 'static,
111    V: EngineValidator<N::Payload> + WaitForCaches,
112    C: ConfigureEvm<Primitives = N::Primitives> + 'static,
113{
114    /// Spawns the engine services and returns the [`ChainOrchestrator`] driving them.
115    pub fn build(
116        self,
117    ) -> EngineOrchestrator<N::Payload, N::Primitives, Client, S, PipelineSync<N>> {
118        let Ok(orchestrator) = self.build_with_backfill(|pipeline, runtime| {
119            Ok::<_, Infallible>(PipelineSync::new(pipeline, runtime))
120        });
121        orchestrator
122    }
123
124    /// Like [`Self::build`], backfilling through the [`BackfillSync`] `backfill` makes from the
125    /// staged [`Pipeline`] and its runtime, instead of [`PipelineSync`].
126    ///
127    /// The backfill is built before any engine service is spawned, so its error aborts the build.
128    pub fn build_with_backfill<B: BackfillSync + Unpin, E>(
129        self,
130        backfill: impl FnOnce(Pipeline<N>, Runtime) -> Result<B, E>,
131    ) -> EngineOrchestratorResult<N::Payload, N::Primitives, Client, S, B, E> {
132        let Self {
133            engine_kind,
134            consensus,
135            client,
136            incoming_requests,
137            pipeline,
138            pipeline_task_spawner,
139            provider,
140            blockchain_db,
141            pruner,
142            payload_builder,
143            payload_validator,
144            overlay_manager,
145            tree_config,
146            sync_metrics_tx,
147            evm_config,
148            runtime,
149        } = self;
150
151        let backfill_sync = backfill(pipeline, pipeline_task_spawner)?;
152        let downloader = BasicBlockDownloader::new(client, consensus.clone());
153
154        let persistence_handle =
155            PersistenceHandle::<N::Primitives>::spawn_service(provider, pruner, sync_metrics_tx);
156
157        let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();
158
159        let (to_tree_tx, from_tree) = EngineApiTreeHandler::spawn_new(
160            blockchain_db,
161            consensus,
162            payload_validator,
163            persistence_handle,
164            payload_builder,
165            canonical_in_memory_state,
166            overlay_manager,
167            tree_config,
168            engine_kind,
169            evm_config,
170            runtime,
171        );
172
173        let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
174        let handler = EngineHandler::new(engine_handler, downloader, incoming_requests);
175
176        Ok(ChainOrchestrator::new(handler, backfill_sync))
177    }
178}
reth_engine_tree/launch.rs

reth_engine_tree/
launch.rs
