gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-20 21:34:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Capella eip 4844 cleanup (#3652)

Browse Source 
* add capella gossip boiler plate

* get everything compiling

Co-authored-by: realbigsean <sean@sigmaprime.io
Co-authored-by: Mark Mackey <mark@sigmaprime.io>

* small cleanup

* small cleanup

* cargo fix + some test cleanup

* improve block production

* add fixme for potential panic

Co-authored-by: Mark Mackey <mark@sigmaprime.io>
This commit is contained in:
realbigsean
2022-10-26 15:15:26 -04:00
committed by GitHub
parent 221c433d62
commit 137f230344
52 changed files with 1392 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

349
beacon_node/execution_layer/src/lib.rs
View File

@@ -4,7 +4,7 @@
//! This crate only provides useful functionality for "The Merge", it does not provide any of the
//! deposit-contract functionality that the `beacon_node/eth1` crate already provides.
use crate::json_structures::JsonBlobBundlesV1;
use crate::json_structures::JsonBlobBundles;
use crate::payload_cache::PayloadCache;
use auth::{strip_prefix, Auth, JwtKey};
use builder_client::BuilderHttpClient;
@@ -33,10 +33,12 @@ use tokio::{
time::sleep,
};
use tokio_stream::wrappers::WatchStream;
use types::{AbstractExecPayload, Blob, ExecPayload, ExecutionPayloadEip4844, KzgCommitment};
use types::{
BlindedPayload, BlockType, ChainSpec, Epoch, ExecPayload, ExecutionBlockHash, ForkName,
BlindedPayload, BlockType, ChainSpec, Epoch, ExecutionBlockHash, ForkName,
ProposerPreparationData, PublicKeyBytes, SignedBeaconBlock, Slot,
};
use types::{ExecutionPayload, ExecutionPayloadCapella, ExecutionPayloadMerge};
mod engine_api;
mod engines;
@@ -88,6 +90,70 @@ impl From<ApiError> for Error {
}
}
pub enum BlockProposalContents<T: EthSpec, Payload: AbstractExecPayload<T>> {
Payload(Payload),
PayloadAndBlobs {
payload: Payload,
kzg_commitments: Vec<KzgCommitment>,
blobs: Vec<Blob<T>>,
},
}
impl<T: EthSpec, Payload: AbstractExecPayload<T>> BlockProposalContents<T, Payload> {
pub fn payload(&self) -> &Payload {
match self {
Self::Payload(payload) => payload,
Self::PayloadAndBlobs {
payload,
kzg_commitments: _,
blobs: _,
} => payload,
}
}
pub fn to_payload(self) -> Payload {
match self {
Self::Payload(payload) => payload,
Self::PayloadAndBlobs {
payload,
kzg_commitments: _,
blobs: _,
} => payload,
}
}
pub fn kzg_commitments(&self) -> Option<&[KzgCommitment]> {
match self {
Self::Payload(_) => None,
Self::PayloadAndBlobs {
payload: _,
kzg_commitments,
blobs: _,
} => Some(kzg_commitments),
}
}
pub fn blobs(&self) -> Option<&[Blob<T>]> {
match self {
Self::Payload(_) => None,
Self::PayloadAndBlobs {
payload: _,
kzg_commitments: _,
blobs,
} => Some(blobs),
}
}
pub fn default_at_fork(fork_name: ForkName) -> Self {
match fork_name {
ForkName::Base | ForkName::Altair | ForkName::Merge | ForkName::Capella => {
BlockProposalContents::Payload(Payload::default_at_fork(fork_name))
}
ForkName::Eip4844 => BlockProposalContents::PayloadAndBlobs {
payload: Payload::default_at_fork(fork_name),
blobs: vec![],
kzg_commitments: vec![],
},
}
}
}
#[derive(Clone, PartialEq)]
pub struct ProposerPreparationDataEntry {
update_epoch: Epoch,
@@ -536,7 +602,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
/// The result will be returned from the first node that returns successfully. No more nodes
/// will be contacted.
#[allow(clippy::too_many_arguments)]
pub async fn get_payload<Payload: ExecPayload<T>>(
pub async fn get_payload<Payload: AbstractExecPayload<T>>(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
@@ -545,7 +611,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
forkchoice_update_params: ForkchoiceUpdateParameters,
builder_params: BuilderParams,
spec: &ChainSpec,
) -> Result<Payload, Error> {
) -> Result<BlockProposalContents<T, Payload>, Error> {
let suggested_fee_recipient = self.get_suggested_fee_recipient(proposer_index).await;
match Payload::block_type() {
@@ -583,7 +649,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
}
#[allow(clippy::too_many_arguments)]
async fn get_blinded_payload<Payload: ExecPayload<T>>(
async fn get_blinded_payload<Payload: AbstractExecPayload<T>>(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
@@ -592,7 +658,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
forkchoice_update_params: ForkchoiceUpdateParameters,
builder_params: BuilderParams,
spec: &ChainSpec,
) -> Result<Payload, Error> {
) -> Result<BlockProposalContents<T, Payload>, Error> {
if let Some(builder) = self.builder() {
let slot = builder_params.slot;
let pubkey = builder_params.pubkey;
@@ -635,6 +701,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
Ok(local)
}
(Ok(Some(relay)), Ok(local)) => {
let local_payload = local.payload();
let is_signature_valid = relay.data.verify_signature(spec);
let header = relay.data.message.header;
@@ -668,14 +735,14 @@ impl<T: EthSpec> ExecutionLayer<T> {
falling back to local execution engine."
);
Ok(local)
} else if header.timestamp() != local.timestamp() {
} else if header.timestamp() != local_payload.timestamp() {
warn!(
self.log(),
"Invalid timestamp from connected builder, \
falling back to local execution engine."
);
Ok(local)
} else if header.block_number() != local.block_number() {
} else if header.block_number() != local_payload.block_number() {
warn!(
self.log(),
"Invalid block number from connected builder, \
@@ -706,7 +773,8 @@ impl<T: EthSpec> ExecutionLayer<T> {
not match, using it anyways."
);
}
Ok(header)
//FIXME(sean) the builder API needs to be updated
Ok(BlockProposalContents::Payload(header))
}
}
(relay_result, Err(local_error)) => {
@@ -715,7 +783,10 @@ impl<T: EthSpec> ExecutionLayer<T> {
relay_result
.map_err(Error::Builder)?
.ok_or(Error::NoHeaderFromBuilder)
.map(|d| d.data.message.header)
.map(|d| {
//FIXME(sean) the builder API needs to be updated
BlockProposalContents::Payload(d.data.message.header)
})
}
};
}
@@ -743,14 +814,14 @@ impl<T: EthSpec> ExecutionLayer<T> {
}
/// Get a full payload without caching its result in the execution layer's payload cache.
async fn get_full_payload<Payload: ExecPayload<T>>(
async fn get_full_payload<Payload: AbstractExecPayload<T>>(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
prev_randao: Hash256,
suggested_fee_recipient: Address,
forkchoice_update_params: ForkchoiceUpdateParameters,
) -> Result<Payload, Error> {
) -> Result<BlockProposalContents<T, Payload>, Error> {
self.get_full_payload_with(
parent_hash,
timestamp,
@@ -763,14 +834,14 @@ impl<T: EthSpec> ExecutionLayer<T> {
}
/// Get a full payload and cache its result in the execution layer's payload cache.
async fn get_full_payload_caching<Payload: ExecPayload<T>>(
async fn get_full_payload_caching<Payload: AbstractExecPayload<T>>(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
prev_randao: Hash256,
suggested_fee_recipient: Address,
forkchoice_update_params: ForkchoiceUpdateParameters,
) -> Result<Payload, Error> {
) -> Result<BlockProposalContents<T, Payload>, Error> {
self.get_full_payload_with(
parent_hash,
timestamp,
@@ -782,51 +853,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
.await
}
pub async fn get_blob_bundles(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
prev_randao: Hash256,
proposer_index: u64,
) -> Result<JsonBlobBundlesV1<T>, Error> {
let suggested_fee_recipient = self.get_suggested_fee_recipient(proposer_index).await;
debug!(
self.log(),
"Issuing engine_getBlobsBundle";
"suggested_fee_recipient" => ?suggested_fee_recipient,
"prev_randao" => ?prev_randao,
"timestamp" => timestamp,
"parent_hash" => ?parent_hash,
);
self.engine()
.request(|engine| async move {
let payload_id = if let Some(id) = engine
.get_payload_id(parent_hash, timestamp, prev_randao, suggested_fee_recipient)
.await
{
// The payload id has been cached for this engine.
metrics::inc_counter_vec(
&metrics::EXECUTION_LAYER_PRE_PREPARED_PAYLOAD_ID,
&[metrics::HIT],
);
id
} else {
error!(
self.log(),
"Exec engine unable to produce blobs, did you call get_payload before?",
);
return Err(ApiError::PayloadIdUnavailable);
};
engine.api.get_blobs_bundle_v1::<T>(payload_id).await
})
.await
.map_err(Box::new)
.map_err(Error::EngineError)
}
async fn get_full_payload_with<Payload: ExecPayload<T>>(
async fn get_full_payload_with<Payload: AbstractExecPayload<T>>(
&self,
parent_hash: ExecutionBlockHash,
timestamp: u64,
@@ -834,15 +861,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
suggested_fee_recipient: Address,
forkchoice_update_params: ForkchoiceUpdateParameters,
f: fn(&ExecutionLayer<T>, &ExecutionPayload<T>) -> Option<ExecutionPayload<T>>,
) -> Result<Payload, Error> {
debug!(
self.log(),
"Issuing engine_getPayload";
"suggested_fee_recipient" => ?suggested_fee_recipient,
"prev_randao" => ?prev_randao,
"timestamp" => timestamp,
"parent_hash" => ?parent_hash,
);
) -> Result<BlockProposalContents<T, Payload>, Error> {
self.engine()
.request(|engine| async move {
let payload_id = if let Some(id) = engine
@@ -871,11 +890,13 @@ impl<T: EthSpec> ExecutionLayer<T> {
.finalized_hash
.unwrap_or_else(ExecutionBlockHash::zero),
};
let payload_attributes = PayloadAttributes {
// FIXME: This will have to properly handle forks. To do that,
// withdrawals will need to be passed into this function
let payload_attributes = PayloadAttributes::V1(PayloadAttributesV1 {
timestamp,
prev_randao,
suggested_fee_recipient,
};
});
let response = engine
.notify_forkchoice_updated(
@@ -900,33 +921,64 @@ impl<T: EthSpec> ExecutionLayer<T> {
}
};
engine
.api
.get_payload_v1::<T>(payload_id)
.await
.map(|full_payload| {
if full_payload.fee_recipient != suggested_fee_recipient {
error!(
self.log(),
"Inconsistent fee recipient";
"msg" => "The fee recipient returned from the Execution Engine differs \
from the suggested_fee_recipient set on the beacon node. This could \
indicate that fees are being diverted to another address. Please \
ensure that the value of suggested_fee_recipient is set correctly and \
that the Execution Engine is trusted.",
"fee_recipient" => ?full_payload.fee_recipient,
"suggested_fee_recipient" => ?suggested_fee_recipient,
);
}
if f(self, &full_payload).is_some() {
warn!(
self.log(),
"Duplicate payload cached, this might indicate redundant proposal \
let blob_fut = async {
//FIXME(sean) do a fork check here and return None otherwise
debug!(
self.log(),
"Issuing engine_getBlobsBundle";
"suggested_fee_recipient" => ?suggested_fee_recipient,
"prev_randao" => ?prev_randao,
"timestamp" => timestamp,
"parent_hash" => ?parent_hash,
);
Some(engine.api.get_blobs_bundle_v1::<T>(payload_id).await)
};
let payload_fut = async {
debug!(
self.log(),
"Issuing engine_getPayload";
"suggested_fee_recipient" => ?suggested_fee_recipient,
"prev_randao" => ?prev_randao,
"timestamp" => timestamp,
"parent_hash" => ?parent_hash,
);
engine.api.get_payload_v1::<T>(payload_id).await
};
let (blob, payload) = tokio::join!(blob_fut, payload_fut);
let payload = payload.map(|full_payload| {
if full_payload.fee_recipient() != suggested_fee_recipient {
error!(
self.log(),
"Inconsistent fee recipient";
"msg" => "The fee recipient returned from the Execution Engine differs \
from the suggested_fee_recipient set on the beacon node. This could \
indicate that fees are being diverted to another address. Please \
ensure that the value of suggested_fee_recipient is set correctly and \
that the Execution Engine is trusted.",
"fee_recipient" => ?full_payload.fee_recipient(),
"suggested_fee_recipient" => ?suggested_fee_recipient,
);
}
if f(self, &full_payload).is_some() {
warn!(
self.log(),
"Duplicate payload cached, this might indicate redundant proposal \
attempts."
);
}
full_payload.into()
);
}
full_payload.into()
})?;
if let Some(blob) = blob.transpose()? {
// FIXME(sean) cache blobs
Ok(BlockProposalContents::PayloadAndBlobs {
payload,
blobs: blob.blobs,
kzg_commitments: blob.kzgs,
})
} else {
Ok(BlockProposalContents::Payload(payload))
}
})
.await
.map_err(Box::new)
@@ -958,9 +1010,9 @@ impl<T: EthSpec> ExecutionLayer<T> {
trace!(
self.log(),
"Issuing engine_newPayload";
"parent_hash" => ?execution_payload.parent_hash,
"block_hash" => ?execution_payload.block_hash,
"block_number" => execution_payload.block_number,
"parent_hash" => ?execution_payload.parent_hash(),
"block_hash" => ?execution_payload.block_hash(),
"block_number" => execution_payload.block_number(),
);
let result = self
@@ -975,7 +1027,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
);
}
process_payload_status(execution_payload.block_hash, result, self.log())
process_payload_status(execution_payload.block_hash(), result, self.log())
.map_err(Box::new)
.map_err(Error::EngineError)
}
@@ -1076,9 +1128,9 @@ impl<T: EthSpec> ExecutionLayer<T> {
let payload_attributes = self.payload_attributes(next_slot, head_block_root).await;
// Compute the "lookahead", the time between when the payload will be produced and now.
if let Some(payload_attributes) = payload_attributes {
if let Some(ref payload_attributes) = payload_attributes {
if let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) {
let timestamp = Duration::from_secs(payload_attributes.timestamp);
let timestamp = Duration::from_secs(payload_attributes.timestamp());
if let Some(lookahead) = timestamp.checked_sub(now) {
metrics::observe_duration(
&metrics::EXECUTION_LAYER_PAYLOAD_ATTRIBUTES_LOOKAHEAD,
@@ -1105,11 +1157,16 @@ impl<T: EthSpec> ExecutionLayer<T> {
.set_latest_forkchoice_state(forkchoice_state)
.await;
let payload_attributes_ref = &payload_attributes;
let result = self
.engine()
.request(|engine| async move {
engine
.notify_forkchoice_updated(forkchoice_state, payload_attributes, self.log())
.notify_forkchoice_updated(
forkchoice_state,
payload_attributes_ref.clone(),
self.log(),
)
.await
})
.await;
@@ -1399,7 +1456,8 @@ impl<T: EthSpec> ExecutionLayer<T> {
let _timer = metrics::start_timer(&metrics::EXECUTION_LAYER_GET_PAYLOAD_BY_BLOCK_HASH);
if hash == ExecutionBlockHash::zero() {
return Ok(Some(ExecutionPayload::default()));
// FIXME: how to handle forks properly here?
return Ok(Some(ExecutionPayloadMerge::default().into()));
}
let block = if let Some(block) = engine.api.get_block_by_hash_with_txns::<T>(hash).await? {
@@ -1410,7 +1468,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
let transactions = VariableList::new(
block
.transactions
.transactions()
.into_iter()
.map(|transaction| VariableList::new(transaction.rlp().to_vec()))
.collect::<Result<_, _>>()
@@ -1418,22 +1476,73 @@ impl<T: EthSpec> ExecutionLayer<T> {
)
.map_err(ApiError::DeserializeTransactions)?;
Ok(Some(ExecutionPayload {
parent_hash: block.parent_hash,
fee_recipient: block.fee_recipient,
state_root: block.state_root,
receipts_root: block.receipts_root,
logs_bloom: block.logs_bloom,
prev_randao: block.prev_randao,
block_number: block.block_number,
gas_limit: block.gas_limit,
gas_used: block.gas_used,
timestamp: block.timestamp,
extra_data: block.extra_data,
base_fee_per_gas: block.base_fee_per_gas,
block_hash: block.block_hash,
transactions,
}))
let payload = match block {
ExecutionBlockWithTransactions::Merge(merge_block) => {
ExecutionPayload::Merge(ExecutionPayloadMerge {
parent_hash: merge_block.parent_hash,
fee_recipient: merge_block.fee_recipient,
state_root: merge_block.state_root,
receipts_root: merge_block.receipts_root,
logs_bloom: merge_block.logs_bloom,
prev_randao: merge_block.prev_randao,
block_number: merge_block.block_number,
gas_limit: merge_block.gas_limit,
gas_used: merge_block.gas_used,
timestamp: merge_block.timestamp,
extra_data: merge_block.extra_data,
base_fee_per_gas: merge_block.base_fee_per_gas,
block_hash: merge_block.block_hash,
transactions,
})
}
ExecutionBlockWithTransactions::Capella(capella_block) => {
let withdrawals = VariableList::new(capella_block.withdrawals.clone())
.map_err(ApiError::DeserializeWithdrawals)?;
ExecutionPayload::Capella(ExecutionPayloadCapella {
parent_hash: capella_block.parent_hash,
fee_recipient: capella_block.fee_recipient,
state_root: capella_block.state_root,
receipts_root: capella_block.receipts_root,
logs_bloom: capella_block.logs_bloom,
prev_randao: capella_block.prev_randao,
block_number: capella_block.block_number,
gas_limit: capella_block.gas_limit,
gas_used: capella_block.gas_used,
timestamp: capella_block.timestamp,
extra_data: capella_block.extra_data,
base_fee_per_gas: capella_block.base_fee_per_gas,
block_hash: capella_block.block_hash,
transactions,
withdrawals,
})
}
ExecutionBlockWithTransactions::Eip4844(eip4844_block) => {
let withdrawals = VariableList::new(eip4844_block.withdrawals.clone())
.map_err(ApiError::DeserializeWithdrawals)?;
ExecutionPayload::Eip4844(ExecutionPayloadEip4844 {
parent_hash: eip4844_block.parent_hash,
fee_recipient: eip4844_block.fee_recipient,
state_root: eip4844_block.state_root,
receipts_root: eip4844_block.receipts_root,
logs_bloom: eip4844_block.logs_bloom,
prev_randao: eip4844_block.prev_randao,
block_number: eip4844_block.block_number,
gas_limit: eip4844_block.gas_limit,
gas_used: eip4844_block.gas_used,
timestamp: eip4844_block.timestamp,
extra_data: eip4844_block.extra_data,
base_fee_per_gas: eip4844_block.base_fee_per_gas,
excess_blobs: eip4844_block.excess_blobs,
block_hash: eip4844_block.block_hash,
transactions,
withdrawals,
})
}
};
Ok(Some(payload))
}
pub async fn propose_blinded_beacon_block(