jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 19:09:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add version to BeaconEngineMessage FCU (#12089)

Browse Source 
Co-authored-by: Dan Cline <6798349+Rjected@users.noreply.github.com>
This commit is contained in:
0xOsiris
2024-10-28 05:00:36 -07:00
committed by GitHub
parent 1b0f625f1d
commit 0d07d27f3c
14 changed files with 107 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
bin/reth/src/commands/debug_cmd/replay_engine.rs
View File

@ -18,7 +18,9 @@ use reth_engine_util::engine_store::{EngineMessageStore, StoredEngineApiMessage}
use reth_fs_util as fs;
use reth_network::{BlockDownloaderProvider, NetworkHandle};
use reth_network_api::NetworkInfo;
use reth_node_api::{NodeTypesWithDB, NodeTypesWithDBAdapter, NodeTypesWithEngine};
use reth_node_api::{
EngineApiMessageVersion, NodeTypesWithDB, NodeTypesWithDBAdapter, NodeTypesWithEngine,
};
use reth_node_ethereum::{EthEngineTypes, EthEvmConfig, EthExecutorProvider};
use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService};
use reth_provider::{
@ -166,8 +168,13 @@ impl<C: ChainSpecParser<ChainSpec = ChainSpec>> Command<C> {
debug!(target: "reth::cli", filepath = %filepath.display(), ?message, "Forwarding Engine API message");
match message {
StoredEngineApiMessage::ForkchoiceUpdated { state, payload_attrs } => {
let response =
beacon_engine_handle.fork_choice_updated(state, payload_attrs).await?;
let response = beacon_engine_handle
.fork_choice_updated(
state,
payload_attrs,
EngineApiMessageVersion::default(),
)
.await?;
debug!(target: "reth::cli", ?response, "Received for forkchoice updated");
}
StoredEngineApiMessage::NewPayload { payload, sidecar } => {

3
crates/consensus/auto-seal/src/task.rs
View File

@ -3,7 +3,7 @@ use alloy_rpc_types_engine::ForkchoiceState;
use futures_util::{future::BoxFuture, FutureExt};
use reth_beacon_consensus::{BeaconEngineMessage, ForkchoiceStatus};
use reth_chainspec::{EthChainSpec, EthereumHardforks};
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_evm::execute::BlockExecutorProvider;
use reth_provider::{CanonChainTracker, StateProviderFactory};
use reth_stages_api::PipelineEvent;
@ -155,6 +155,7 @@ where
state,
payload_attrs: None,
tx,
version: EngineApiMessageVersion::default(),
});
debug!(target: "consensus::auto", ?state, "Sent fork choice update");

7
crates/consensus/beacon/src/engine/handle.rs
View File

@ -8,7 +8,7 @@ use alloy_rpc_types_engine::{
ExecutionPayload, ExecutionPayloadSidecar, ForkchoiceState, ForkchoiceUpdated, PayloadStatus,
};
use futures::TryFutureExt;
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_errors::RethResult;
use reth_tokio_util::{EventSender, EventStream};
use tokio::sync::{mpsc::UnboundedSender, oneshot};
@ -60,9 +60,10 @@ where
&self,
state: ForkchoiceState,
payload_attrs: Option<Engine::PayloadAttributes>,
version: EngineApiMessageVersion,
) -> Result<ForkchoiceUpdated, BeaconForkChoiceUpdateError> {
Ok(self
.send_fork_choice_updated(state, payload_attrs)
.send_fork_choice_updated(state, payload_attrs, version)
.map_err(|_| BeaconForkChoiceUpdateError::EngineUnavailable)
.await??
.await?)
@ -74,12 +75,14 @@ where
&self,
state: ForkchoiceState,
payload_attrs: Option<Engine::PayloadAttributes>,
version: EngineApiMessageVersion,
) -> oneshot::Receiver<RethResult<OnForkChoiceUpdated>> {
let (tx, rx) = oneshot::channel();
let _ = self.to_engine.send(BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
});
rx
}

4
crates/consensus/beacon/src/engine/message.rs
View File

@ -4,7 +4,7 @@ use alloy_rpc_types_engine::{
ForkchoiceUpdateError, ForkchoiceUpdated, PayloadId, PayloadStatus, PayloadStatusEnum,
};
use futures::{future::Either, FutureExt};
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_errors::RethResult;
use reth_payload_primitives::PayloadBuilderError;
use std::{
@ -156,6 +156,8 @@ pub enum BeaconEngineMessage<Engine: EngineTypes> {
state: ForkchoiceState,
/// The payload attributes for block building.
payload_attrs: Option<Engine::PayloadAttributes>,
/// The Engine API Version.
version: EngineApiMessageVersion,
/// The sender for returning forkchoice updated result.
tx: oneshot::Sender<RethResult<OnForkChoiceUpdated>>,
},

17
crates/consensus/beacon/src/engine/mod.rs
View File

@ -10,7 +10,7 @@ use reth_blockchain_tree_api::{
error::{BlockchainTreeError, CanonicalError, InsertBlockError, InsertBlockErrorKind},
BlockStatus, BlockValidationKind, BlockchainTreeEngine, CanonicalOutcome, InsertPayloadOk,
};
use reth_engine_primitives::{EngineTypes, PayloadTypes};
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes, PayloadTypes};
use reth_errors::{BlockValidationError, ProviderResult, RethError, RethResult};
use reth_network_p2p::{
sync::{NetworkSyncUpdater, SyncState},
@ -428,7 +428,12 @@ where
} else if let Some(attrs) = attrs {
// the CL requested to build a new payload on top of this new VALID head
let head = outcome.into_header().unseal();
self.process_payload_attributes(attrs, head, state)
self.process_payload_attributes(
attrs,
head,
state,
EngineApiMessageVersion::default(),
)
} else {
OnForkChoiceUpdated::valid(PayloadStatus::new(
PayloadStatusEnum::Valid,
@ -1160,6 +1165,7 @@ where
attrs: <N::Engine as PayloadTypes>::PayloadAttributes,
head: Header,
state: ForkchoiceState,
_version: EngineApiMessageVersion,
) -> OnForkChoiceUpdated {
// 7. Client software MUST ensure that payloadAttributes.timestamp is greater than timestamp
// of a block referenced by forkchoiceState.headBlockHash. If this condition isn't held
@ -1855,7 +1861,12 @@ where
// sensitive, hence they are polled first.
if let Poll::Ready(Some(msg)) = this.engine_message_stream.poll_next_unpin(cx) {
match msg {
BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx } => {
BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version: _version,
} => {
this.on_forkchoice_updated(state, payload_attrs, tx);
}
BeaconEngineMessage::NewPayload { payload, sidecar, tx } => {

10
crates/consensus/beacon/src/engine/test_utils.rs
View File

@ -19,6 +19,7 @@ use reth_downloaders::{
bodies::bodies::BodiesDownloaderBuilder,
headers::reverse_headers::ReverseHeadersDownloaderBuilder,
};
use reth_engine_primitives::EngineApiMessageVersion;
use reth_ethereum_engine_primitives::EthEngineTypes;
use reth_evm::{either::Either, test_utils::MockExecutorProvider};
use reth_evm_ethereum::execute::EthExecutorProvider;
@ -93,7 +94,9 @@ impl<DB> TestEnv<DB> {
&self,
state: ForkchoiceState,
) -> Result<ForkchoiceUpdated, BeaconForkChoiceUpdateError> {
self.engine_handle.fork_choice_updated(state, None).await
self.engine_handle
.fork_choice_updated(state, None, EngineApiMessageVersion::default())
.await
}
/// Sends the `ForkchoiceUpdated` message to the consensus engine and retries if the engine
@ -103,7 +106,10 @@ impl<DB> TestEnv<DB> {
state: ForkchoiceState,
) -> Result<ForkchoiceUpdated, BeaconForkChoiceUpdateError> {
loop {
let result = self.engine_handle.fork_choice_updated(state, None).await?;
let result = self
.engine_handle
.fork_choice_updated(state, None, EngineApiMessageVersion::default())
.await?;
if !result.is_syncing() {
return Ok(result)
}

4
crates/engine/local/src/miner.rs
View File

@ -6,7 +6,7 @@ use eyre::OptionExt;
use futures_util::{stream::Fuse, StreamExt};
use reth_beacon_consensus::BeaconEngineMessage;
use reth_chainspec::EthereumHardforks;
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_payload_builder::PayloadBuilderHandle;
use reth_payload_primitives::{
BuiltPayload, PayloadAttributesBuilder, PayloadBuilder, PayloadKind, PayloadTypes,
@ -167,6 +167,7 @@ where
state: self.forkchoice_state(),
payload_attrs: None,
tx,
version: EngineApiMessageVersion::default(),
})?;
let res = rx.await??;
@ -193,6 +194,7 @@ where
state: self.forkchoice_state(),
payload_attrs: Some(self.payload_attributes_builder.build(timestamp)),
tx,
version: EngineApiMessageVersion::default(),
})?;
let res = rx.await??.await?;

23
crates/engine/tree/src/tree/mod.rs
View File

@ -26,7 +26,7 @@ use reth_chain_state::{
};
use reth_chainspec::EthereumHardforks;
use reth_consensus::{Consensus, PostExecutionInput};
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_errors::{ConsensusError, ProviderResult};
use reth_evm::execute::BlockExecutorProvider;
use reth_payload_builder::PayloadBuilderHandle;
@ -969,6 +969,7 @@ where
&mut self,
state: ForkchoiceState,
attrs: Option<T::PayloadAttributes>,
version: EngineApiMessageVersion,
) -> ProviderResult<TreeOutcome<OnForkChoiceUpdated>> {
trace!(target: "engine::tree", ?attrs, "invoked forkchoice update");
self.metrics.engine.forkchoice_updated_messages.increment(1);
@ -1018,7 +1019,7 @@ where
// to return an error
ProviderError::HeaderNotFound(state.head_block_hash.into())
})?;
let updated = self.process_payload_attributes(attr, &tip, state);
let updated = self.process_payload_attributes(attr, &tip, state, version);
return Ok(TreeOutcome::new(updated))
}
@ -1038,7 +1039,7 @@ where
}
if let Some(attr) = attrs {
let updated = self.process_payload_attributes(attr, &tip, state);
let updated = self.process_payload_attributes(attr, &tip, state, version);
return Ok(TreeOutcome::new(updated))
}
@ -1054,7 +1055,8 @@ where
if self.engine_kind.is_opstack() {
if let Some(attr) = attrs {
debug!(target: "engine::tree", head = canonical_header.number, "handling payload attributes for canonical head");
let updated = self.process_payload_attributes(attr, &canonical_header, state);
let updated =
self.process_payload_attributes(attr, &canonical_header, state, version);
return Ok(TreeOutcome::new(updated))
}
}
@ -1206,8 +1208,14 @@ where
}
EngineApiRequest::Beacon(request) => {
match request {
BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx } => {
let mut output = self.on_forkchoice_updated(state, payload_attrs);
BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
} => {
let mut output =
self.on_forkchoice_updated(state, payload_attrs, version);
if let Ok(res) = &mut output {
// track last received forkchoice state
@ -2484,6 +2492,7 @@ where
attrs: T::PayloadAttributes,
head: &Header,
state: ForkchoiceState,
_version: EngineApiMessageVersion,
) -> OnForkChoiceUpdated {
// 7. Client software MUST ensure that payloadAttributes.timestamp is greater than timestamp
// of a block referenced by forkchoiceState.headBlockHash. If this condition isn't held
@ -2808,6 +2817,7 @@ mod tests {
state: fcu_state,
payload_attrs: None,
tx,
version: EngineApiMessageVersion::default(),
}
.into(),
))
@ -3097,6 +3107,7 @@ mod tests {
},
payload_attrs: None,
tx,
version: EngineApiMessageVersion::default(),
}
.into(),
))

7
crates/engine/util/src/engine_store.rs
View File

@ -64,7 +64,12 @@ impl EngineMessageStore {
fs::create_dir_all(&self.path)?; // ensure that store path had been created
let timestamp = received_at.duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis();
match msg {
BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx: _tx } => {
BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx: _tx,
version: _version,
} => {
let filename = format!("{}-fcu-{}.json", timestamp, state.head_block_hash);
fs::write(
self.path.join(filename),

20
crates/engine/util/src/reorg.rs
View File

@ -8,7 +8,7 @@ use alloy_rpc_types_engine::{
use futures::{stream::FuturesUnordered, Stream, StreamExt, TryFutureExt};
use itertools::Either;
use reth_beacon_consensus::{BeaconEngineMessage, BeaconOnNewPayloadError, OnForkChoiceUpdated};
use reth_engine_primitives::EngineTypes;
use reth_engine_primitives::{EngineApiMessageVersion, EngineTypes};
use reth_errors::{BlockExecutionError, BlockValidationError, RethError, RethResult};
use reth_ethereum_forks::EthereumHardforks;
use reth_evm::{
@ -211,18 +211,32 @@ where
state: reorg_forkchoice_state,
payload_attrs: None,
tx: reorg_fcu_tx,
version: EngineApiMessageVersion::default(),
},
]);
*this.state = EngineReorgState::Reorg { queue };
continue
}
(Some(BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx }), _) => {
(
Some(BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
}),
_,
) => {
// Record last forkchoice state forwarded to the engine.
// We do not care if it's valid since engine should be able to handle
// reorgs that rely on invalid forkchoice state.
*this.last_forkchoice_state = Some(state);
*this.forkchoice_states_forwarded += 1;
Some(BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx })
Some(BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
})
}
(item, _) => item,
};

14
crates/engine/util/src/skip_fcu.rs
View File

@ -45,7 +45,12 @@ where
loop {
let next = ready!(this.stream.poll_next_unpin(cx));
let item = match next {
Some(BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx }) => {
Some(BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
}) => {
if this.skipped < this.threshold {
*this.skipped += 1;
tracing::warn!(target: "engine::stream::skip_fcu", ?state, ?payload_attrs, threshold=this.threshold, skipped=this.skipped, "Skipping FCU");
@ -53,7 +58,12 @@ where
continue
}
*this.skipped = 0;
Some(BeaconEngineMessage::ForkchoiceUpdated { state, payload_attrs, tx })
Some(BeaconEngineMessage::ForkchoiceUpdated {
state,
payload_attrs,
tx,
version,
})
}
next => next,
};

11
crates/payload/primitives/src/lib.rs
View File

@ -324,22 +324,23 @@ where
}
/// The version of Engine API message.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default)]
pub enum EngineApiMessageVersion {
/// Version 1
V1,
V1 = 1,
/// Version 2
///
/// Added in the Shanghai hardfork.
V2,
V2 = 2,
/// Version 3
///
/// Added in the Cancun hardfork.
V3,
#[default]
V3 = 3,
/// Version 4
///
/// Added in the Prague hardfork.
V4,
V4 = 4,
}
/// Determines how we should choose the payload to return.

2
crates/payload/primitives/src/traits.rs
View File

@ -84,7 +84,7 @@ pub trait PayloadBuilderAttributes: Send + Sync + std::fmt::Debug {
/// Creates a new payload builder for the given parent block and the attributes.
///
/// Derives the unique [`PayloadId`] for the given parent and attributes
/// Derives the unique [`PayloadId`] for the given parent, attributes and version.
fn try_new(
parent: B256,
rpc_payload_attributes: Self::RpcPayloadAttributes,

5
crates/rpc/rpc-engine-api/src/engine_api.rs
View File

@ -616,7 +616,8 @@ where
// To do this, we set the payload attrs to `None` if attribute validation failed, but
// we still apply the forkchoice update.
if let Err(err) = attr_validation_res {
let fcu_res = self.inner.beacon_consensus.fork_choice_updated(state, None).await?;
let fcu_res =
self.inner.beacon_consensus.fork_choice_updated(state, None, version).await?;
// TODO: decide if we want this branch - the FCU INVALID response might be more
// useful than the payload attributes INVALID response
if fcu_res.is_invalid() {
@ -626,7 +627,7 @@ where
}
}
Ok(self.inner.beacon_consensus.fork_choice_updated(state, payload_attrs).await?)
Ok(self.inner.beacon_consensus.fork_choice_updated(state, payload_attrs, version).await?)
}
}