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

reth_engine_primitives/
message.rs
