reth_engine_service/
service.rs

1use futures::{Stream, StreamExt};
2use pin_project::pin_project;
3use reth_chainspec::EthChainSpec;
4use reth_consensus::{ConsensusError, FullConsensus};
5use reth_engine_primitives::{BeaconEngineMessage, ConsensusEngineEvent};
6use reth_engine_tree::{
7    backfill::PipelineSync,
8    download::BasicBlockDownloader,
9    engine::{EngineApiKind, EngineApiRequest, EngineApiRequestHandler, EngineHandler},
10    persistence::PersistenceHandle,
11    tree::{EngineApiTreeHandler, EngineValidator, TreeConfig},
12};
13pub use reth_engine_tree::{
14    chain::{ChainEvent, ChainOrchestrator},
15    engine::EngineApiEvent,
16};
17use reth_ethereum_primitives::EthPrimitives;
18use reth_evm::ConfigureEvm;
19use reth_network_p2p::BlockClient;
20use reth_node_types::{BlockTy, NodeTypes};
21use reth_payload_builder::PayloadBuilderHandle;
22use reth_provider::{
23    providers::{BlockchainProvider, ProviderNodeTypes},
24    ProviderFactory,
25};
26use reth_prune::PrunerWithFactory;
27use reth_stages_api::{MetricEventsSender, Pipeline};
28use reth_tasks::TaskSpawner;
29use std::{
30    pin::Pin,
31    sync::Arc,
32    task::{Context, Poll},
33};
34
35/// Alias for consensus engine stream.
36pub type EngineMessageStream<T> = Pin<Box<dyn Stream<Item = BeaconEngineMessage<T>> + Send + Sync>>;
37
38/// Alias for chain orchestrator.
39type EngineServiceType<N, Client> = ChainOrchestrator<
40    EngineHandler<
41        EngineApiRequestHandler<
42            EngineApiRequest<<N as NodeTypes>::Payload, <N as NodeTypes>::Primitives>,
43            <N as NodeTypes>::Primitives,
44        >,
45        EngineMessageStream<<N as NodeTypes>::Payload>,
46        BasicBlockDownloader<Client, BlockTy<N>>,
47    >,
48    PipelineSync<N>,
49>;
50
51/// The type that drives the chain forward and communicates progress.
52#[pin_project]
53#[expect(missing_debug_implementations)]
54// TODO(mattsse): remove hidden once fixed : <https://github.com/rust-lang/rust/issues/135363>
55//  otherwise rustdoc fails to resolve the alias
56#[doc(hidden)]
57pub struct EngineService<N, Client>
58where
59    N: ProviderNodeTypes,
60    Client: BlockClient<Block = BlockTy<N>> + 'static,
61{
62    orchestrator: EngineServiceType<N, Client>,
63}
64
65impl<N, Client> EngineService<N, Client>
66where
67    N: ProviderNodeTypes,
68    Client: BlockClient<Block = BlockTy<N>> + 'static,
69{
70    /// Constructor for `EngineService`.
71    #[expect(clippy::too_many_arguments)]
72    pub fn new<V, C>(
73        consensus: Arc<dyn FullConsensus<N::Primitives, Error = ConsensusError>>,
74        chain_spec: Arc<N::ChainSpec>,
75        client: Client,
76        incoming_requests: EngineMessageStream<N::Payload>,
77        pipeline: Pipeline<N>,
78        pipeline_task_spawner: Box<dyn TaskSpawner>,
79        provider: ProviderFactory<N>,
80        blockchain_db: BlockchainProvider<N>,
81        pruner: PrunerWithFactory<ProviderFactory<N>>,
82        payload_builder: PayloadBuilderHandle<N::Payload>,
83        payload_validator: V,
84        tree_config: TreeConfig,
85        sync_metrics_tx: MetricEventsSender,
86        evm_config: C,
87    ) -> Self
88    where
89        V: EngineValidator<N::Payload>,
90        C: ConfigureEvm<Primitives = N::Primitives> + 'static,
91    {
92        let engine_kind =
93            if chain_spec.is_optimism() { EngineApiKind::OpStack } else { EngineApiKind::Ethereum };
94
95        let downloader = BasicBlockDownloader::new(client, consensus.clone());
96
97        let persistence_handle =
98            PersistenceHandle::<EthPrimitives>::spawn_service(provider, pruner, sync_metrics_tx);
99
100        let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();
101
102        let (to_tree_tx, from_tree) = EngineApiTreeHandler::<N::Primitives, _, _, _, _>::spawn_new(
103            blockchain_db,
104            consensus,
105            payload_validator,
106            persistence_handle,
107            payload_builder,
108            canonical_in_memory_state,
109            tree_config,
110            engine_kind,
111            evm_config,
112        );
113
114        let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
115        let handler = EngineHandler::new(engine_handler, downloader, incoming_requests);
116
117        let backfill_sync = PipelineSync::new(pipeline, pipeline_task_spawner);
118
119        Self { orchestrator: ChainOrchestrator::new(handler, backfill_sync) }
120    }
121
122    /// Returns a mutable reference to the orchestrator.
123    pub fn orchestrator_mut(&mut self) -> &mut EngineServiceType<N, Client> {
124        &mut self.orchestrator
125    }
126}
127
128impl<N, Client> Stream for EngineService<N, Client>
129where
130    N: ProviderNodeTypes,
131    Client: BlockClient<Block = BlockTy<N>> + 'static,
132{
133    type Item = ChainEvent<ConsensusEngineEvent<N::Primitives>>;
134
135    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
136        let mut orchestrator = self.project().orchestrator;
137        StreamExt::poll_next_unpin(&mut orchestrator, cx)
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use reth_chainspec::{ChainSpecBuilder, MAINNET};
145    use reth_engine_primitives::{BeaconEngineMessage, NoopInvalidBlockHook};
146    use reth_engine_tree::{test_utils::TestPipelineBuilder, tree::BasicEngineValidator};
147    use reth_ethereum_consensus::EthBeaconConsensus;
148    use reth_ethereum_engine_primitives::EthEngineTypes;
149    use reth_evm_ethereum::EthEvmConfig;
150    use reth_exex_types::FinishedExExHeight;
151    use reth_network_p2p::test_utils::TestFullBlockClient;
152    use reth_node_ethereum::EthereumEngineValidator;
153    use reth_primitives_traits::SealedHeader;
154    use reth_provider::{
155        providers::BlockchainProvider, test_utils::create_test_provider_factory_with_chain_spec,
156    };
157    use reth_prune::Pruner;
158    use reth_tasks::TokioTaskExecutor;
159    use std::sync::Arc;
160    use tokio::sync::{mpsc::unbounded_channel, watch};
161    use tokio_stream::wrappers::UnboundedReceiverStream;
162
163    #[test]
164    fn eth_chain_orchestrator_build() {
165        let chain_spec = Arc::new(
166            ChainSpecBuilder::default()
167                .chain(MAINNET.chain)
168                .genesis(MAINNET.genesis.clone())
169                .paris_activated()
170                .build(),
171        );
172        let consensus = Arc::new(EthBeaconConsensus::new(chain_spec.clone()));
173
174        let client = TestFullBlockClient::default();
175
176        let (_tx, rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();
177        let incoming_requests = UnboundedReceiverStream::new(rx);
178
179        let pipeline = TestPipelineBuilder::new().build(chain_spec.clone());
180        let pipeline_task_spawner = Box::<TokioTaskExecutor>::default();
181        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
182
183        let blockchain_db =
184            BlockchainProvider::with_latest(provider_factory.clone(), SealedHeader::default())
185                .unwrap();
186        let engine_payload_validator = EthereumEngineValidator::new(chain_spec.clone());
187        let (_tx, rx) = watch::channel(FinishedExExHeight::NoExExs);
188        let pruner = Pruner::new_with_factory(provider_factory.clone(), vec![], 0, 0, None, rx);
189        let evm_config = EthEvmConfig::new(chain_spec.clone());
190
191        let engine_validator = BasicEngineValidator::new(
192            blockchain_db.clone(),
193            consensus.clone(),
194            evm_config.clone(),
195            engine_payload_validator,
196            TreeConfig::default(),
197            Box::new(NoopInvalidBlockHook::default()),
198        );
199
200        let (sync_metrics_tx, _sync_metrics_rx) = unbounded_channel();
201        let (tx, _rx) = unbounded_channel();
202        let _eth_service = EngineService::new(
203            consensus,
204            chain_spec,
205            client,
206            Box::pin(incoming_requests),
207            pipeline,
208            pipeline_task_spawner,
209            provider_factory,
210            blockchain_db,
211            pruner,
212            PayloadBuilderHandle::new(tx),
213            engine_validator,
214            TreeConfig::default(),
215            sync_metrics_tx,
216            evm_config,
217        );
218    }
219}
reth_engine_service/service.rs

reth_engine_service/
service.rs
