reth_engine_local/
miner.rs

1//! Contains the implementation of the mining mode for the local engine.
2
3use alloy_primitives::{TxHash, B256};
4use alloy_rpc_types_engine::ForkchoiceState;
5use eyre::OptionExt;
6use futures_util::{stream::Fuse, Stream, StreamExt};
7use reth_engine_primitives::ConsensusEngineHandle;
8use reth_payload_builder::PayloadBuilderHandle;
9use reth_payload_primitives::{BuiltPayload, PayloadAttributesBuilder, PayloadKind, PayloadTypes};
10use reth_primitives_traits::{HeaderTy, SealedHeaderFor};
11use reth_storage_api::BlockReader;
12use reth_transaction_pool::TransactionPool;
13use std::{
14    collections::VecDeque,
15    fmt,
16    future::Future,
17    num::NonZeroUsize,
18    pin::Pin,
19    task::{Context, Poll},
20    time::Duration,
21};
22use tokio::time::Interval;
23use tokio_stream::wrappers::ReceiverStream;
24use tracing::error;
25
26/// Default number of confirmations required before a block is finalized in dev mode.
27pub const DEFAULT_FINALITY_DEPTH: NonZeroUsize = NonZeroUsize::new(64).unwrap();
28
29/// A mining mode for the local dev engine.
30pub enum MiningMode<Pool: TransactionPool + Unpin> {
31    /// In this mode a block is built as soon as
32    /// a valid transaction reaches the pool.
33    /// If `max_transactions` is set, a block is built when that many transactions have
34    /// accumulated.
35    Instant {
36        /// The transaction pool.
37        pool: Pool,
38        /// Stream of transaction notifications.
39        rx: Fuse<ReceiverStream<TxHash>>,
40        /// Maximum number of transactions to accumulate before mining a block.
41        /// If None, mine immediately when any transaction arrives.
42        max_transactions: Option<usize>,
43        /// Counter for accumulated transactions (only used when `max_transactions` is set).
44        accumulated: usize,
45    },
46    /// In this mode a block is built at a fixed interval.
47    Interval(Interval),
48    /// In this mode a block is built when the trigger stream yields a value.
49    ///
50    /// This is a general-purpose trigger that can be fired on demand, for example via a channel
51    /// or any other [`Stream`] implementation.
52    Trigger(Pin<Box<dyn Stream<Item = ()> + Send + Sync>>),
53}
54
55impl<Pool: TransactionPool + Unpin> fmt::Debug for MiningMode<Pool> {
56    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
57        match self {
58            Self::Instant { max_transactions, accumulated, .. } => f
59                .debug_struct("Instant")
60                .field("max_transactions", max_transactions)
61                .field("accumulated", accumulated)
62                .finish(),
63            Self::Interval(interval) => f.debug_tuple("Interval").field(interval).finish(),
64            Self::Trigger(_) => f.debug_tuple("Trigger").finish(),
65        }
66    }
67}
68
69impl<Pool: TransactionPool + Unpin> MiningMode<Pool> {
70    /// Constructor for a [`MiningMode::Instant`]
71    pub fn instant(pool: Pool, max_transactions: Option<usize>) -> Self {
72        let rx = pool.pending_transactions_listener();
73        Self::Instant { pool, rx: ReceiverStream::new(rx).fuse(), max_transactions, accumulated: 0 }
74    }
75
76    /// Constructor for a [`MiningMode::Interval`]
77    pub fn interval(duration: Duration) -> Self {
78        let start = tokio::time::Instant::now() + duration;
79        Self::Interval(tokio::time::interval_at(start, duration))
80    }
81
82    /// Constructor for a [`MiningMode::Trigger`]
83    ///
84    /// Accepts any stream that yields `()` values, each of which triggers a new block to be
85    /// mined. This can be backed by a channel, a custom stream, or any other async source.
86    pub fn trigger(trigger: impl Stream<Item = ()> + Send + Sync + 'static) -> Self {
87        Self::Trigger(Box::pin(trigger))
88    }
89}
90
91impl<Pool: TransactionPool + Unpin> Future for MiningMode<Pool> {
92    type Output = ();
93
94    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
95        let this = self.get_mut();
96        match this {
97            Self::Instant { pool, rx, max_transactions, accumulated } => {
98                // Poll for new transaction notifications
99                while let Poll::Ready(Some(_)) = rx.poll_next_unpin(cx) {
100                    if pool.pending_and_queued_txn_count().0 == 0 {
101                        continue;
102                    }
103                    if let Some(max_tx) = max_transactions {
104                        *accumulated += 1;
105                        // If we've reached the max transactions threshold, mine a block
106                        if *accumulated >= *max_tx {
107                            *accumulated = 0; // Reset counter for next block
108                            return Poll::Ready(());
109                        }
110                    } else {
111                        // If no max_transactions is set, mine immediately
112                        return Poll::Ready(());
113                    }
114                }
115            }
116            Self::Interval(interval) => {
117                if interval.poll_tick(cx).is_ready() {
118                    return Poll::Ready(())
119                }
120            }
121            Self::Trigger(trigger) => {
122                if trigger.poll_next_unpin(cx).is_ready() {
123                    return Poll::Ready(())
124                }
125            }
126        }
127        Poll::Pending
128    }
129}
130
131/// Local miner advancing the chain
132#[derive(Debug)]
133pub struct LocalMiner<T: PayloadTypes, B, Pool: TransactionPool + Unpin> {
134    /// The payload attribute builder for the engine
135    payload_attributes_builder: B,
136    /// Sender for events to engine.
137    to_engine: ConsensusEngineHandle<T>,
138    /// The mining mode for the engine
139    mode: MiningMode<Pool>,
140    /// The payload builder for the engine
141    payload_builder: PayloadBuilderHandle<T>,
142    /// Latest block in the chain so far.
143    last_header: SealedHeaderFor<<T::BuiltPayload as BuiltPayload>::Primitives>,
144    /// Stores latest mined blocks.
145    last_block_hashes: VecDeque<B256>,
146    /// Number of confirmations required before a block is finalized.
147    finality_depth: NonZeroUsize,
148    /// Optional sleep duration between initiating payload building and resolving.
149    ///
150    /// When set, the miner sleeps after `fork_choice_updated` before calling
151    /// `resolve_kind`, giving the payload job time for multiple rebuild attempts.
152    payload_wait_time: Option<Duration>,
153}
154
155impl<T, B, Pool> LocalMiner<T, B, Pool>
156where
157    T: PayloadTypes,
158    B: PayloadAttributesBuilder<
159        T::PayloadAttributes,
160        HeaderTy<<T::BuiltPayload as BuiltPayload>::Primitives>,
161    >,
162    Pool: TransactionPool + Unpin,
163{
164    /// Spawns a new [`LocalMiner`] with the given parameters.
165    pub fn new(
166        provider: impl BlockReader<Header = HeaderTy<<T::BuiltPayload as BuiltPayload>::Primitives>>,
167        payload_attributes_builder: B,
168        to_engine: ConsensusEngineHandle<T>,
169        mode: MiningMode<Pool>,
170        payload_builder: PayloadBuilderHandle<T>,
171    ) -> Self {
172        let last_header =
173            provider.sealed_header(provider.best_block_number().unwrap()).unwrap().unwrap();
174
175        Self {
176            payload_attributes_builder,
177            to_engine,
178            mode,
179            payload_builder,
180            last_block_hashes: VecDeque::from([last_header.hash()]),
181            finality_depth: DEFAULT_FINALITY_DEPTH,
182            last_header,
183            payload_wait_time: None,
184        }
185    }
186
187    /// Sets the number of confirmations required before a block is finalized.
188    pub const fn with_finality_depth(mut self, finality_depth: NonZeroUsize) -> Self {
189        self.finality_depth = finality_depth;
190        self
191    }
192
193    /// Sets the payload wait time, if any.
194    pub const fn with_payload_wait_time_opt(mut self, wait_time: Option<Duration>) -> Self {
195        self.payload_wait_time = wait_time;
196        self
197    }
198
199    /// Runs the [`LocalMiner`] in a loop, polling the miner and building payloads.
200    pub async fn run(mut self) {
201        let mut fcu_interval = tokio::time::interval(Duration::from_secs(1));
202        loop {
203            tokio::select! {
204                // Wait for the interval or the pool to receive a transaction
205                _ = &mut self.mode => {
206                    if let Err(e) = self.advance().await {
207                        error!(target: "engine::local", "Error advancing the chain: {:?}", e);
208                    }
209                }
210                // send FCU once in a while
211                _ = fcu_interval.tick() => {
212                    if let Err(e) = self.update_forkchoice_state().await {
213                        error!(target: "engine::local", "Error updating fork choice: {:?}", e);
214                    }
215                }
216            }
217        }
218    }
219
220    /// Returns current forkchoice state.
221    fn forkchoice_state(&self) -> ForkchoiceState {
222        let finality_depth = self.finality_depth.get();
223        ForkchoiceState {
224            head_block_hash: block_hash_at_depth(&self.last_block_hashes, 1),
225            safe_block_hash: block_hash_at_depth(
226                &self.last_block_hashes,
227                finality_depth.div_ceil(2),
228            ),
229            finalized_block_hash: block_hash_at_depth(&self.last_block_hashes, finality_depth),
230        }
231    }
232
233    /// Sends a FCU to the engine.
234    async fn update_forkchoice_state(&self) -> eyre::Result<()> {
235        let state = self.forkchoice_state();
236        let res = self.to_engine.fork_choice_updated(state, None).await?;
237
238        if !res.is_valid() {
239            eyre::bail!("Invalid fork choice update {state:?}: {res:?}");
240        }
241
242        Ok(())
243    }
244
245    /// Generates payload attributes for a new block, passes them to FCU and inserts built payload
246    /// through newPayload.
247    async fn advance(&mut self) -> eyre::Result<()> {
248        let res = self
249            .to_engine
250            .fork_choice_updated(
251                self.forkchoice_state(),
252                Some(self.payload_attributes_builder.build(&self.last_header)),
253            )
254            .await?;
255
256        if !res.is_valid() {
257            eyre::bail!("Invalid payload status");
258        }
259
260        let payload_id = res.payload_id.ok_or_eyre("No payload id")?;
261
262        if let Some(wait_time) = self.payload_wait_time {
263            tokio::time::sleep(wait_time).await;
264        }
265
266        let Some(Ok(payload)) =
267            self.payload_builder.resolve_kind(payload_id, PayloadKind::WaitForPending).await
268        else {
269            eyre::bail!("No payload");
270        };
271
272        let header = payload.block().sealed_header().clone();
273        let res = self.to_engine.new_payload(payload.into()).await?;
274
275        if !res.is_valid() {
276            eyre::bail!("Invalid payload");
277        }
278
279        self.last_block_hashes.push_back(header.hash());
280        self.last_header = header;
281        // Ensure we retain only the hashes needed to calculate the finalized block.
282        if self.last_block_hashes.len() > self.finality_depth.get() {
283            self.last_block_hashes.pop_front();
284        }
285
286        self.update_forkchoice_state().await?;
287
288        Ok(())
289    }
290}
291
292fn block_hash_at_depth(block_hashes: &VecDeque<B256>, depth: usize) -> B256 {
293    *block_hashes.get(block_hashes.len().saturating_sub(depth)).expect("at least 1 block exists")
294}
295
296#[cfg(test)]
297mod tests {
298    use super::*;
299
300    #[test]
301    fn block_hash_at_configured_depth() {
302        let block_hashes = (0..64).map(B256::with_last_byte).collect();
303
304        assert_eq!(block_hash_at_depth(&block_hashes, 1), B256::with_last_byte(63));
305        assert_eq!(block_hash_at_depth(&block_hashes, 32), B256::with_last_byte(32));
306        assert_eq!(block_hash_at_depth(&block_hashes, 64), B256::with_last_byte(0));
307        assert_eq!(block_hash_at_depth(&block_hashes, 128), B256::with_last_byte(0));
308    }
309}
reth_engine_local/miner.rs

reth_engine_local/
miner.rs
