reth_engine_primitives/
message.rs

1use crate::{
2    error::BeaconForkChoiceUpdateError, BeaconOnNewPayloadError, ExecutionPayload, ForkchoiceStatus,
3};
4use alloy_eips::eip4895::Withdrawal;
5use alloy_primitives::{Bytes, B256};
6use alloy_rpc_types_engine::{
7    ExecutionData, ForkChoiceUpdateResult, ForkchoiceState, ForkchoiceUpdateError,
8    ForkchoiceUpdated, PayloadId, PayloadStatus, PayloadStatusEnum,
9};
10use core::{
11    fmt::{self, Display},
12    future::Future,
13    pin::Pin,
14    task::{ready, Context, Poll},
15};
16use futures::{future::Either, FutureExt, TryFutureExt};
17use reth_errors::RethResult;
18use reth_payload_builder_primitives::PayloadBuilderError;
19use reth_payload_primitives::PayloadTypes;
20use std::time::{Duration, Instant};
21use tokio::sync::{mpsc::UnboundedSender, oneshot};
22
23/// Type alias for backwards compat
24#[deprecated(note = "Use ConsensusEngineHandle instead")]
25pub type BeaconConsensusEngineHandle<Payload> = ConsensusEngineHandle<Payload>;
26
27/// Represents the outcome of forkchoice update.
28///
29/// This is a future that resolves to [`ForkChoiceUpdateResult`]
30#[must_use = "futures do nothing unless you `.await` or poll them"]
31#[derive(Debug)]
32pub struct OnForkChoiceUpdated {
33    /// Represents the status of the forkchoice update.
34    ///
35    /// Note: This is separate from the response `fut`, because we still can return an error
36    /// depending on the payload attributes, even if the forkchoice update itself is valid.
37    forkchoice_status: ForkchoiceStatus,
38    /// Returns the result of the forkchoice update.
39    fut: Either<futures::future::Ready<ForkChoiceUpdateResult>, PendingPayloadId>,
40}
41
42// === impl OnForkChoiceUpdated ===
43
44impl OnForkChoiceUpdated {
45    /// Returns the determined status of the received `ForkchoiceState`.
46    pub const fn forkchoice_status(&self) -> ForkchoiceStatus {
47        self.forkchoice_status
48    }
49
50    /// Creates a new instance of `OnForkChoiceUpdated` for the `SYNCING` state
51    pub fn syncing() -> Self {
52        let status = PayloadStatus::from_status(PayloadStatusEnum::Syncing);
53        Self {
54            forkchoice_status: ForkchoiceStatus::from_payload_status(&status.status),
55            fut: Either::Left(futures::future::ready(Ok(ForkchoiceUpdated::new(status)))),
56        }
57    }
58
59    /// Creates a new instance of `OnForkChoiceUpdated` if the forkchoice update succeeded and no
60    /// payload attributes were provided.
61    pub fn valid(status: PayloadStatus) -> Self {
62        Self {
63            forkchoice_status: ForkchoiceStatus::from_payload_status(&status.status),
64            fut: Either::Left(futures::future::ready(Ok(ForkchoiceUpdated::new(status)))),
65        }
66    }
67
68    /// Creates a new instance of `OnForkChoiceUpdated` with the given payload status, if the
69    /// forkchoice update failed due to an invalid payload.
70    pub fn with_invalid(status: PayloadStatus) -> Self {
71        Self {
72            forkchoice_status: ForkchoiceStatus::from_payload_status(&status.status),
73            fut: Either::Left(futures::future::ready(Ok(ForkchoiceUpdated::new(status)))),
74        }
75    }
76
77    /// Creates a new instance of `OnForkChoiceUpdated` if the forkchoice update failed because the
78    /// given state is considered invalid
79    pub fn invalid_state() -> Self {
80        Self {
81            forkchoice_status: ForkchoiceStatus::Invalid,
82            fut: Either::Left(futures::future::ready(Err(ForkchoiceUpdateError::InvalidState))),
83        }
84    }
85
86    /// Creates a new instance of `OnForkChoiceUpdated` if the forkchoice update failed because the
87    /// requested reorg to the head block exceeds the supported reorg depth.
88    pub fn too_deep_reorg() -> Self {
89        Self {
90            forkchoice_status: ForkchoiceStatus::Invalid,
91            fut: Either::Left(futures::future::ready(Err(ForkchoiceUpdateError::TooDeepReorg))),
92        }
93    }
94
95    /// Creates a new instance of `OnForkChoiceUpdated` if the forkchoice update was successful but
96    /// payload attributes were invalid.
97    pub fn invalid_payload_attributes() -> Self {
98        Self {
99            // This is valid because this is only reachable if the state and payload is valid
100            forkchoice_status: ForkchoiceStatus::Valid,
101            fut: Either::Left(futures::future::ready(Err(
102                ForkchoiceUpdateError::UpdatedInvalidPayloadAttributes,
103            ))),
104        }
105    }
106
107    /// If the forkchoice update was successful and no payload attributes were provided, this method
108    pub const fn updated_with_pending_payload_id(
109        payload_status: PayloadStatus,
110        pending_payload_id: oneshot::Receiver<Result<PayloadId, PayloadBuilderError>>,
111    ) -> Self {
112        Self {
113            forkchoice_status: ForkchoiceStatus::from_payload_status(&payload_status.status),
114            fut: Either::Right(PendingPayloadId {
115                payload_status: Some(payload_status),
116                pending_payload_id,
117            }),
118        }
119    }
120}
121
122impl Future for OnForkChoiceUpdated {
123    type Output = ForkChoiceUpdateResult;
124
125    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
126        self.get_mut().fut.poll_unpin(cx)
127    }
128}
129
130/// A future that returns the payload id of a yet to be initiated payload job after a successful
131/// forkchoice update
132#[derive(Debug)]
133struct PendingPayloadId {
134    payload_status: Option<PayloadStatus>,
135    pending_payload_id: oneshot::Receiver<Result<PayloadId, PayloadBuilderError>>,
136}
137
138impl Future for PendingPayloadId {
139    type Output = ForkChoiceUpdateResult;
140
141    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
142        let this = self.get_mut();
143        let res = ready!(this.pending_payload_id.poll_unpin(cx));
144        match res {
145            Ok(Ok(payload_id)) => Poll::Ready(Ok(ForkchoiceUpdated {
146                payload_status: this.payload_status.take().expect("Polled after completion"),
147                payload_id: Some(payload_id),
148            })),
149            Err(_) | Ok(Err(_)) => {
150                // failed to initiate a payload build job
151                Poll::Ready(Err(ForkchoiceUpdateError::UpdatedInvalidPayloadAttributes))
152            }
153        }
154    }
155}
156
157/// Timing breakdown for `reth_newPayload` responses.
158#[derive(Debug, Clone, Copy)]
159pub struct NewPayloadTimings {
160    /// Server-side execution latency.
161    pub latency: Duration,
162    /// Time spent waiting on persistence, including both time this message spent queued
163    /// due to persistence backpressure and, when `wait_for_persistence` was requested,
164    /// the explicit wait for in-flight persistence to complete.
165    pub persistence_wait: Duration,
166    /// Time spent waiting for the execution cache lock.
167    ///
168    /// `None` when wasn't asked to wait for execution cache.
169    pub execution_cache_wait: Option<Duration>,
170    /// Time spent waiting for the sparse trie cache lock.
171    ///
172    /// `None` when wasn't asked to wait for sparse trie cache.
173    pub sparse_trie_wait: Option<Duration>,
174}
175
176/// Additional data for big block payloads that merge multiple real blocks.
177///
178/// This is used by the `reth_newPayload` endpoint to pass environment switches
179/// and prior block hashes needed for correct multi-segment execution.
180#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
181pub struct BigBlockData<ExecutionData> {
182    /// Environment switches at block boundaries.
183    /// Each entry is `(cumulative_tx_count, execution_data_of_next_block)`.
184    ///
185    /// The first entry at index 0 represents the **original unmutated** base block's
186    /// `ExecutionData`, which must be used to derive the initial EVM environment.
187    pub env_switches: Vec<ExecutionData>,
188    /// Block number → real block hash for blocks covered by previous big blocks in a sequence.
189    /// When replaying chained big blocks, the BLOCKHASH opcode needs real hashes for blocks
190    /// that were merged into earlier big blocks (and thus not individually persisted).
191    pub prior_block_hashes: Vec<(u64, alloy_primitives::B256)>,
192    /// Block number for this big block.
193    pub block_number: u64,
194    /// Merged block access list for this big block.
195    #[serde(default, skip_serializing_if = "Option::is_none")]
196    pub merged_block_access_list: Option<Bytes>,
197}
198
199impl ExecutionPayload for BigBlockData<ExecutionData> {
200    fn parent_hash(&self) -> B256 {
201        self.env_switches[0].parent_hash()
202    }
203
204    fn block_hash(&self) -> B256 {
205        self.env_switches.last().unwrap().block_hash()
206    }
207
208    fn block_number(&self) -> u64 {
209        self.block_number
210    }
211
212    fn withdrawals(&self) -> Option<&Vec<Withdrawal>> {
213        self.env_switches[0].withdrawals()
214    }
215
216    fn block_access_list(&self) -> Option<&Bytes> {
217        self.merged_block_access_list.as_ref()
218    }
219
220    fn parent_beacon_block_root(&self) -> Option<B256> {
221        self.env_switches[0].parent_beacon_block_root()
222    }
223
224    fn timestamp(&self) -> u64 {
225        self.env_switches[0].timestamp()
226    }
227
228    fn gas_used(&self) -> u64 {
229        self.env_switches.iter().map(|data| data.gas_used()).sum()
230    }
231
232    fn gas_limit(&self) -> u64 {
233        self.env_switches.iter().map(|data| data.gas_limit()).sum()
234    }
235
236    fn transaction_count(&self) -> usize {
237        self.env_switches.iter().map(|data| data.transaction_count()).sum()
238    }
239
240    fn slot_number(&self) -> Option<u64> {
241        self.env_switches[0].payload.slot_number()
242    }
243}
244
245/// A message for the beacon engine from other components of the node (engine RPC API invoked by the
246/// consensus layer).
247#[derive(Debug)]
248pub enum BeaconEngineMessage<Payload: PayloadTypes> {
249    /// Message with new payload.
250    NewPayload {
251        /// The execution payload received by Engine API.
252        payload: Payload::ExecutionData,
253        /// The sender for returning payload status result.
254        tx: oneshot::Sender<Result<PayloadStatus, BeaconOnNewPayloadError>>,
255    },
256    /// Message with new payload used by `reth_newPayload` endpoint.
257    ///
258    /// Supports independent control over waiting for persistence and cache locks before
259    /// processing, providing unbiased timing measurements when enabled.
260    ///
261    /// Returns detailed timing breakdown alongside the payload status.
262    RethNewPayload {
263        /// The execution payload received by Engine API.
264        payload: Payload::ExecutionData,
265        /// Whether to wait for in-flight persistence to complete before processing.
266        wait_for_persistence: bool,
267        /// Whether to wait for execution cache and sparse trie locks before processing.
268        wait_for_caches: bool,
269        /// The sender for returning payload status result and timing breakdown.
270        tx: oneshot::Sender<Result<(PayloadStatus, NewPayloadTimings), BeaconOnNewPayloadError>>,
271        /// When this message was enqueued, used to measure backpressure wait time.
272        enqueued_at: Instant,
273    },
274    /// Message with updated forkchoice state.
275    ForkchoiceUpdated {
276        /// The updated forkchoice state.
277        state: ForkchoiceState,
278        /// The payload attributes for block building.
279        payload_attrs: Option<Payload::PayloadAttributes>,
280        /// The sender for returning forkchoice updated result.
281        tx: oneshot::Sender<RethResult<OnForkChoiceUpdated>>,
282    },
283}
284
285impl<Payload: PayloadTypes> Display for BeaconEngineMessage<Payload> {
286    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
287        match self {
288            Self::NewPayload { payload, .. } => {
289                write!(
290                    f,
291                    "NewPayload(parent: {}, number: {}, hash: {})",
292                    payload.parent_hash(),
293                    payload.block_number(),
294                    payload.block_hash()
295                )
296            }
297            Self::RethNewPayload { payload, .. } => {
298                write!(
299                    f,
300                    "RethNewPayload(parent: {}, number: {}, hash: {})",
301                    payload.parent_hash(),
302                    payload.block_number(),
303                    payload.block_hash()
304                )
305            }
306            Self::ForkchoiceUpdated { state, payload_attrs, .. } => {
307                // we don't want to print the entire payload attributes, because for OP this
308                // includes all txs
309                write!(
310                    f,
311                    "ForkchoiceUpdated {{ state: {state:?}, has_payload_attributes: {} }}",
312                    payload_attrs.is_some()
313                )
314            }
315        }
316    }
317}
318
319/// A cloneable sender type that can be used to send engine API messages.
320///
321/// This type mirrors consensus related functions of the engine API.
322#[derive(Debug, Clone)]
323pub struct ConsensusEngineHandle<Payload>
324where
325    Payload: PayloadTypes,
326{
327    to_engine: UnboundedSender<BeaconEngineMessage<Payload>>,
328}
329
330impl<Payload> ConsensusEngineHandle<Payload>
331where
332    Payload: PayloadTypes,
333{
334    /// Creates a new beacon consensus engine handle.
335    pub const fn new(to_engine: UnboundedSender<BeaconEngineMessage<Payload>>) -> Self {
336        Self { to_engine }
337    }
338
339    /// Sends a new payload message to the beacon consensus engine and waits for a response.
340    ///
341    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_newpayloadv2>
342    pub async fn new_payload(
343        &self,
344        payload: Payload::ExecutionData,
345    ) -> Result<PayloadStatus, BeaconOnNewPayloadError> {
346        let (tx, rx) = oneshot::channel();
347        let _ = self.to_engine.send(BeaconEngineMessage::NewPayload { payload, tx });
348        rx.await.map_err(|_| BeaconOnNewPayloadError::EngineUnavailable)?
349    }
350
351    /// Sends a new payload message used by `reth_newPayload` endpoint.
352    ///
353    /// `wait_for_persistence`: waits for in-flight persistence to complete.
354    /// `wait_for_caches`: waits for execution cache and sparse trie locks.
355    ///
356    /// Returns detailed timing breakdown alongside the payload status.
357    pub async fn reth_new_payload(
358        &self,
359        payload: Payload::ExecutionData,
360        wait_for_persistence: bool,
361        wait_for_caches: bool,
362    ) -> Result<(PayloadStatus, NewPayloadTimings), BeaconOnNewPayloadError> {
363        let (tx, rx) = oneshot::channel();
364        let _ = self.to_engine.send(BeaconEngineMessage::RethNewPayload {
365            payload,
366            wait_for_persistence,
367            wait_for_caches,
368            tx,
369            enqueued_at: Instant::now(),
370        });
371        rx.await.map_err(|_| BeaconOnNewPayloadError::EngineUnavailable)?
372    }
373
374    /// Sends a forkchoice update message to the beacon consensus engine and waits for a response.
375    ///
376    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_forkchoiceupdatedv2>
377    pub async fn fork_choice_updated(
378        &self,
379        state: ForkchoiceState,
380        payload_attrs: Option<Payload::PayloadAttributes>,
381    ) -> Result<ForkchoiceUpdated, BeaconForkChoiceUpdateError> {
382        Ok(self
383            .send_fork_choice_updated(state, payload_attrs)
384            .map_err(|_| BeaconForkChoiceUpdateError::EngineUnavailable)
385            .await?
386            .map_err(BeaconForkChoiceUpdateError::internal)?
387            .await?)
388    }
389
390    /// Sends a forkchoice update message to the beacon consensus engine and returns the receiver to
391    /// wait for a response.
392    fn send_fork_choice_updated(
393        &self,
394        state: ForkchoiceState,
395        payload_attrs: Option<Payload::PayloadAttributes>,
396    ) -> oneshot::Receiver<RethResult<OnForkChoiceUpdated>> {
397        let (tx, rx) = oneshot::channel();
398        let _ = self.to_engine.send(BeaconEngineMessage::ForkchoiceUpdated {
399            state,
400            payload_attrs,
401            tx,
402        });
403        rx
404    }
405}
reth_engine_primitives/message.rs

reth_engine_primitives/
message.rs
