reth_engine_local/
service.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
//! Provides a local dev service engine that can be used to run a dev chain.
//!
//! [`LocalEngineService`] polls the payload builder based on a mining mode
//! which can be set to `Instant` or `Interval`. The `Instant` mode will
//! constantly poll the payload builder and initiate block building
//! with a single transaction. The `Interval` mode will initiate block
//! building at a fixed interval.

use core::fmt;
use std::{
    fmt::{Debug, Formatter},
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};

use crate::miner::{LocalMiner, MiningMode};
use futures_util::{Stream, StreamExt};
use reth_beacon_consensus::{BeaconConsensusEngineEvent, EngineNodeTypes};
use reth_chainspec::EthChainSpec;
use reth_consensus::Consensus;
use reth_engine_primitives::BeaconEngineMessage;
use reth_engine_service::service::EngineMessageStream;
use reth_engine_tree::{
    chain::{ChainEvent, HandlerEvent},
    engine::{
        EngineApiKind, EngineApiRequest, EngineApiRequestHandler, EngineRequestHandler, FromEngine,
        RequestHandlerEvent,
    },
    persistence::{PersistenceHandle, PersistenceNodeTypes},
    tree::{EngineApiTreeHandler, InvalidBlockHook, TreeConfig},
};
use reth_evm::execute::BlockExecutorProvider;
use reth_payload_builder::PayloadBuilderHandle;
use reth_payload_primitives::{PayloadAttributesBuilder, PayloadTypes};
use reth_payload_validator::ExecutionPayloadValidator;
use reth_provider::{providers::BlockchainProvider2, ChainSpecProvider, ProviderFactory};
use reth_prune::PrunerWithFactory;
use reth_stages_api::MetricEventsSender;
use tokio::sync::mpsc::UnboundedSender;
use tracing::error;

/// Provides a local dev service engine that can be used to drive the
/// chain forward.
///
/// This service both produces and consumes [`BeaconEngineMessage`]s. This is done to allow
/// modifications of the stream
pub struct LocalEngineService<N>
where
    N: EngineNodeTypes,
{
    /// Processes requests.
    ///
    /// This type is responsible for processing incoming requests.
    handler: EngineApiRequestHandler<EngineApiRequest<N::Engine>>,
    /// Receiver for incoming requests (from the engine API endpoint) that need to be processed.
    incoming_requests: EngineMessageStream<N::Engine>,
}

impl<N> LocalEngineService<N>
where
    N: EngineNodeTypes + PersistenceNodeTypes,
{
    /// Constructor for [`LocalEngineService`].
    #[allow(clippy::too_many_arguments)]
    pub fn new<B>(
        consensus: Arc<dyn Consensus>,
        executor_factory: impl BlockExecutorProvider,
        provider: ProviderFactory<N>,
        blockchain_db: BlockchainProvider2<N>,
        pruner: PrunerWithFactory<ProviderFactory<N>>,
        payload_builder: PayloadBuilderHandle<N::Engine>,
        tree_config: TreeConfig,
        invalid_block_hook: Box<dyn InvalidBlockHook>,
        sync_metrics_tx: MetricEventsSender,
        to_engine: UnboundedSender<BeaconEngineMessage<N::Engine>>,
        from_engine: EngineMessageStream<N::Engine>,
        mode: MiningMode,
        payload_attributes_builder: B,
    ) -> Self
    where
        B: PayloadAttributesBuilder<<N::Engine as PayloadTypes>::PayloadAttributes>,
    {
        let chain_spec = provider.chain_spec();
        let engine_kind =
            if chain_spec.is_optimism() { EngineApiKind::OpStack } else { EngineApiKind::Ethereum };

        let persistence_handle =
            PersistenceHandle::spawn_service(provider, pruner, sync_metrics_tx);
        let payload_validator = ExecutionPayloadValidator::new(chain_spec);

        let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();

        let (to_tree_tx, from_tree) = EngineApiTreeHandler::spawn_new(
            blockchain_db.clone(),
            executor_factory,
            consensus,
            payload_validator,
            persistence_handle,
            payload_builder.clone(),
            canonical_in_memory_state,
            tree_config,
            invalid_block_hook,
            engine_kind,
        );

        let handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);

        LocalMiner::spawn_new(
            blockchain_db,
            payload_attributes_builder,
            to_engine,
            mode,
            payload_builder,
        );

        Self { handler, incoming_requests: from_engine }
    }
}

impl<N> Stream for LocalEngineService<N>
where
    N: EngineNodeTypes,
{
    type Item = ChainEvent<BeaconConsensusEngineEvent>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();

        if let Poll::Ready(ev) = this.handler.poll(cx) {
            return match ev {
                RequestHandlerEvent::HandlerEvent(ev) => match ev {
                    HandlerEvent::BackfillAction(_) => {
                        error!(target: "engine::local", "received backfill request in local engine");
                        Poll::Ready(Some(ChainEvent::FatalError))
                    }
                    HandlerEvent::Event(ev) => Poll::Ready(Some(ChainEvent::Handler(ev))),
                    HandlerEvent::FatalError => Poll::Ready(Some(ChainEvent::FatalError)),
                },
                RequestHandlerEvent::Download(_) => {
                    error!(target: "engine::local", "received download request in local engine");
                    Poll::Ready(Some(ChainEvent::FatalError))
                }
            }
        }

        // forward incoming requests to the handler
        while let Poll::Ready(Some(req)) = this.incoming_requests.poll_next_unpin(cx) {
            this.handler.on_event(FromEngine::Request(req.into()));
        }

        Poll::Pending
    }
}

impl<N: EngineNodeTypes> Debug for LocalEngineService<N> {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.debug_struct("LocalEngineService").finish_non_exhaustive()
    }
}