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                Poll::Pending
116            }
117            Self::Interval(interval) => {
118                if interval.poll_tick(cx).is_ready() {
119                    return Poll::Ready(())
120                }
121                Poll::Pending
122            }
123            Self::Trigger(trigger) => {
124                if trigger.poll_next_unpin(cx).is_ready() {
125                    return Poll::Ready(())
126                }
127                Poll::Pending
128            }
129        }
130    }
131}
132
133/// Local miner advancing the chain
134#[derive(Debug)]
135pub struct LocalMiner<T: PayloadTypes, B, Pool: TransactionPool + Unpin> {
136    /// The payload attribute builder for the engine
137    payload_attributes_builder: B,
138    /// Sender for events to engine.
139    to_engine: ConsensusEngineHandle<T>,
140    /// The mining mode for the engine
141    mode: MiningMode<Pool>,
142    /// The payload builder for the engine
143    payload_builder: PayloadBuilderHandle<T>,
144    /// Latest block in the chain so far.
145    last_header: SealedHeaderFor<<T::BuiltPayload as BuiltPayload>::Primitives>,
146    /// Stores latest mined blocks.
147    last_block_hashes: VecDeque<B256>,
148    /// Number of confirmations required before a block is finalized.
149    finality_depth: NonZeroUsize,
150    /// Optional sleep duration between initiating payload building and resolving.
151    ///
152    /// When set, the miner sleeps after `fork_choice_updated` before calling
153    /// `resolve_kind`, giving the payload job time for multiple rebuild attempts.
154    payload_wait_time: Option<Duration>,
155}
156
157impl<T, B, Pool> LocalMiner<T, B, Pool>
158where
159    T: PayloadTypes,
160    B: PayloadAttributesBuilder<
161        T::PayloadAttributes,
162        HeaderTy<<T::BuiltPayload as BuiltPayload>::Primitives>,
163    >,
164    Pool: TransactionPool + Unpin,
165{
166    /// Spawns a new [`LocalMiner`] with the given parameters.
167    pub fn new(
168        provider: impl BlockReader<Header = HeaderTy<<T::BuiltPayload as BuiltPayload>::Primitives>>,
169        payload_attributes_builder: B,
170        to_engine: ConsensusEngineHandle<T>,
171        mode: MiningMode<Pool>,
172        payload_builder: PayloadBuilderHandle<T>,
173    ) -> Self {
174        let last_header =
175            provider.sealed_header(provider.best_block_number().unwrap()).unwrap().unwrap();
176
177        Self {
178            payload_attributes_builder,
179            to_engine,
180            mode,
181            payload_builder,
182            last_block_hashes: VecDeque::from([last_header.hash()]),
183            finality_depth: DEFAULT_FINALITY_DEPTH,
184            last_header,
185            payload_wait_time: None,
186        }
187    }
188
189    /// Sets the number of confirmations required before a block is finalized.
190    pub const fn with_finality_depth(mut self, finality_depth: NonZeroUsize) -> Self {
191        self.finality_depth = finality_depth;
192        self
193    }
194
195    /// Sets the payload wait time, if any.
196    pub const fn with_payload_wait_time_opt(mut self, wait_time: Option<Duration>) -> Self {
197        self.payload_wait_time = wait_time;
198        self
199    }
200
201    /// Runs the [`LocalMiner`] in a loop, polling the miner and building payloads.
202    pub async fn run(mut self) {
203        let mut fcu_interval = tokio::time::interval(Duration::from_secs(1));
204        loop {
205            tokio::select! {
206                // Wait for the interval or the pool to receive a transaction
207                _ = &mut self.mode => {
208                    if let Err(e) = self.advance().await {
209                        error!(target: "engine::local", "Error advancing the chain: {:?}", e);
210                    }
211                }
212                // send FCU once in a while
213                _ = fcu_interval.tick() => {
214                    if let Err(e) = self.update_forkchoice_state().await {
215                        error!(target: "engine::local", "Error updating fork choice: {:?}", e);
216                    }
217                }
218            }
219        }
220    }
221
222    /// Returns current forkchoice state.
223    fn forkchoice_state(&self) -> ForkchoiceState {
224        let finality_depth = self.finality_depth.get();
225        ForkchoiceState {
226            head_block_hash: block_hash_at_depth(&self.last_block_hashes, 1),
227            safe_block_hash: block_hash_at_depth(
228                &self.last_block_hashes,
229                finality_depth.div_ceil(2),
230            ),
231            finalized_block_hash: block_hash_at_depth(&self.last_block_hashes, finality_depth),
232        }
233    }
234
235    /// Sends a FCU to the engine.
236    async fn update_forkchoice_state(&self) -> eyre::Result<()> {
237        let state = self.forkchoice_state();
238        let res = self.to_engine.fork_choice_updated(state, None).await?;
239
240        if !res.is_valid() {
241            eyre::bail!("Invalid fork choice update {state:?}: {res:?}");
242        }
243
244        Ok(())
245    }
246
247    /// Generates payload attributes for a new block, passes them to FCU and inserts built payload
248    /// through newPayload.
249    async fn advance(&mut self) -> eyre::Result<()> {
250        let res = self
251            .to_engine
252            .fork_choice_updated(
253                self.forkchoice_state(),
254                Some(self.payload_attributes_builder.build(&self.last_header)),
255            )
256            .await?;
257
258        if !res.is_valid() {
259            eyre::bail!("Invalid payload status");
260        }
261
262        let payload_id = res.payload_id.ok_or_eyre("No payload id")?;
263
264        if let Some(wait_time) = self.payload_wait_time {
265            tokio::time::sleep(wait_time).await;
266        }
267
268        let Some(Ok(payload)) =
269            self.payload_builder.resolve_kind(payload_id, PayloadKind::WaitForPending).await
270        else {
271            eyre::bail!("No payload");
272        };
273
274        let header = payload.block().sealed_header().clone();
275        let res = self.to_engine.new_payload(payload.into()).await?;
276
277        if !res.is_valid() {
278            eyre::bail!("Invalid payload");
279        }
280
281        self.last_block_hashes.push_back(header.hash());
282        self.last_header = header;
283        // Ensure we retain only the hashes needed to calculate the finalized block.
284        if self.last_block_hashes.len() > self.finality_depth.get() {
285            self.last_block_hashes.pop_front();
286        }
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
