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

Clean up blockv3 metadata and client (#5015)

Browse Source 
* Improve block production v3 client

* Delete wayward line

* Overhaul JSON endpoint as well

* Rename timeout param

* Update tests

* I broke everything

* Ah this is an insane fix

* Remove unnecessary optionals

* Doc fix
This commit is contained in:
Michael Sproul
2023-12-23 01:39:17 +11:00
committed by GitHub
parent a7e5926a1f
commit af11e78ae1
12 changed files with 451 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
common/eth2/src/lib.rs
View File

@@ -31,6 +31,7 @@ use serde::{de::DeserializeOwned, Serialize};
use ssz::Encode;
use std::convert::TryFrom;
use std::fmt;
use std::future::Future;
use std::iter::Iterator;
use std::path::PathBuf;
use std::time::Duration;
@@ -67,6 +68,8 @@ pub enum Error {
InvalidJson(serde_json::Error),
/// The server returned an invalid server-sent event.
InvalidServerSentEvent(String),
/// The server sent invalid response headers.
InvalidHeaders(String),
/// The server returned an invalid SSZ response.
InvalidSsz(ssz::DecodeError),
/// An I/O error occurred while loading an API token from disk.
@@ -97,6 +100,7 @@ impl Error {
Error::MissingSignatureHeader => None,
Error::InvalidJson(_) => None,
Error::InvalidServerSentEvent(_) => None,
Error::InvalidHeaders(_) => None,
Error::InvalidSsz(_) => None,
Error::TokenReadError(..) => None,
Error::NoServerPubkey | Error::NoToken => None,
@@ -124,7 +128,7 @@ pub struct Timeouts {
pub get_beacon_blocks_ssz: Duration,
pub get_debug_beacon_states: Duration,
pub get_deposit_snapshot: Duration,
pub get_validator_block_ssz: Duration,
pub get_validator_block: Duration,
}
impl Timeouts {
@@ -140,7 +144,7 @@ impl Timeouts {
get_beacon_blocks_ssz: timeout,
get_debug_beacon_states: timeout,
get_deposit_snapshot: timeout,
get_validator_block_ssz: timeout,
get_validator_block: timeout,
}
}
}
@@ -273,27 +277,28 @@ impl BeaconNodeHttpClient {
}
/// Perform a HTTP GET request using an 'accept' header, returning `None` on a 404 error.
pub async fn get_bytes_response_with_response_headers<U: IntoUrl>(
pub async fn get_response_with_response_headers<U: IntoUrl, F, T>(
&self,
url: U,
accept_header: Accept,
timeout: Duration,
) -> Result<(Option<Vec<u8>>, Option<HeaderMap>), Error> {
parser: impl FnOnce(Response, HeaderMap) -> F,
) -> Result<Option<T>, Error>
where
F: Future<Output = Result<T, Error>>,
{
let opt_response = self
.get_response(url, |b| b.accept(accept_header).timeout(timeout))
.await
.optional()?;
// let headers = opt_response.headers();
match opt_response {
Some(resp) => {
let response_headers = resp.headers().clone();
Ok((
Some(resp.bytes().await?.into_iter().collect::<Vec<_>>()),
Some(response_headers),
))
let parsed_response = parser(resp, response_headers).await?;
Ok(Some(parsed_response))
}
None => Ok((None, None)),
None => Ok(None),
}
}
@@ -1816,7 +1821,7 @@ impl BeaconNodeHttpClient {
}
/// returns `GET v3/validator/blocks/{slot}` URL path
pub async fn get_validator_blocks_v3_path<T: EthSpec>(
pub async fn get_validator_blocks_v3_path(
&self,
slot: Slot,
randao_reveal: &SignatureBytes,
@@ -1853,7 +1858,7 @@ impl BeaconNodeHttpClient {
slot: Slot,
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
) -> Result<ForkVersionedBeaconBlockType<T>, Error> {
) -> Result<(JsonProduceBlockV3Response<T>, ProduceBlockV3Metadata), Error> {
self.get_validator_blocks_v3_modular(
slot,
randao_reveal,
@@ -1870,35 +1875,41 @@ impl BeaconNodeHttpClient {
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
skip_randao_verification: SkipRandaoVerification,
) -> Result<ForkVersionedBeaconBlockType<T>, Error> {
) -> Result<(JsonProduceBlockV3Response<T>, ProduceBlockV3Metadata), Error> {
let path = self
.get_validator_blocks_v3_path::<T>(
slot,
randao_reveal,
graffiti,
skip_randao_verification,
.get_validator_blocks_v3_path(slot, randao_reveal, graffiti, skip_randao_verification)
.await?;
let opt_result = self
.get_response_with_response_headers(
path,
Accept::Json,
self.timeouts.get_validator_block,
|response, headers| async move {
let header_metadata = ProduceBlockV3Metadata::try_from(&headers)
.map_err(Error::InvalidHeaders)?;
if header_metadata.execution_payload_blinded {
let blinded_response = response
.json::<ForkVersionedResponse<BlindedBeaconBlock<T>,
ProduceBlockV3Metadata>>()
.await?
.map_data(ProduceBlockV3Response::Blinded);
Ok((blinded_response, header_metadata))
} else {
let full_block_response= response
.json::<ForkVersionedResponse<FullBlockContents<T>,
ProduceBlockV3Metadata>>()
.await?
.map_data(ProduceBlockV3Response::Full);
Ok((full_block_response, header_metadata))
}
},
)
.await?;
let response = self.get_response(path, |b| b).await?;
let is_blinded_payload = response
.headers()
.get(EXECUTION_PAYLOAD_BLINDED_HEADER)
.map(|value| value.to_str().unwrap_or_default().to_lowercase() == "true")
.unwrap_or(false);
if is_blinded_payload {
let blinded_payload = response
.json::<ForkVersionedResponse<BlindedBeaconBlock<T>>>()
.await?;
Ok(ForkVersionedBeaconBlockType::Blinded(blinded_payload))
} else {
let full_payload = response
.json::<ForkVersionedResponse<FullBlockContents<T>>>()
.await?;
Ok(ForkVersionedBeaconBlockType::Full(full_payload))
}
// Generic handler is optional but this route should never 404 unless unimplemented, so
// treat that as an error.
opt_result.ok_or(Error::StatusCode(StatusCode::NOT_FOUND))
}
/// `GET v3/validator/blocks/{slot}` in ssz format
@@ -1907,7 +1918,7 @@ impl BeaconNodeHttpClient {
slot: Slot,
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
) -> Result<(Option<Vec<u8>>, bool), Error> {
) -> Result<(ProduceBlockV3Response<T>, ProduceBlockV3Metadata), Error> {
self.get_validator_blocks_v3_modular_ssz::<T>(
slot,
randao_reveal,
@@ -1924,33 +1935,48 @@ impl BeaconNodeHttpClient {
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
skip_randao_verification: SkipRandaoVerification,
) -> Result<(Option<Vec<u8>>, bool), Error> {
) -> Result<(ProduceBlockV3Response<T>, ProduceBlockV3Metadata), Error> {
let path = self
.get_validator_blocks_v3_path::<T>(
slot,
randao_reveal,
graffiti,
skip_randao_verification,
)
.get_validator_blocks_v3_path(slot, randao_reveal, graffiti, skip_randao_verification)
.await?;
let (response_content, response_headers) = self
.get_bytes_response_with_response_headers(
let opt_response = self
.get_response_with_response_headers(
path,
Accept::Ssz,
self.timeouts.get_validator_block_ssz,
self.timeouts.get_validator_block,
|response, headers| async move {
let metadata = ProduceBlockV3Metadata::try_from(&headers)
.map_err(Error::InvalidHeaders)?;
let response_bytes = response.bytes().await?;
// Parse bytes based on metadata.
let response = if metadata.execution_payload_blinded {
ProduceBlockV3Response::Blinded(
BlindedBeaconBlock::from_ssz_bytes_for_fork(
&response_bytes,
metadata.consensus_version,
)
.map_err(Error::InvalidSsz)?,
)
} else {
ProduceBlockV3Response::Full(
FullBlockContents::from_ssz_bytes_for_fork(
&response_bytes,
metadata.consensus_version,
)
.map_err(Error::InvalidSsz)?,
)
};
Ok((response, metadata))
},
)
.await?;
let is_blinded_payload = match response_headers {
Some(headers) => headers
.get(EXECUTION_PAYLOAD_BLINDED_HEADER)
.map(|value| value.to_str().unwrap_or_default().to_lowercase() == "true")
.unwrap_or(false),
None => false,
};
Ok((response_content, is_blinded_payload))
// Generic handler is optional but this route should never 404 unless unimplemented, so
// treat that as an error.
opt_response.ok_or(Error::StatusCode(StatusCode::NOT_FOUND))
}
/// `GET v2/validator/blocks/{slot}` in ssz format
@@ -1981,7 +2007,7 @@ impl BeaconNodeHttpClient {
.get_validator_blocks_path::<T>(slot, randao_reveal, graffiti, skip_randao_verification)
.await?;
self.get_bytes_opt_accept_header(path, Accept::Ssz, self.timeouts.get_validator_block_ssz)
self.get_bytes_opt_accept_header(path, Accept::Ssz, self.timeouts.get_validator_block)
.await
}
@@ -2085,7 +2111,7 @@ impl BeaconNodeHttpClient {
)
.await?;
self.get_bytes_opt_accept_header(path, Accept::Ssz, self.timeouts.get_validator_block_ssz)
self.get_bytes_opt_accept_header(path, Accept::Ssz, self.timeouts.get_validator_block)
.await
}

98
common/eth2/src/types.rs
View File

@@ -1,9 +1,13 @@
//! This module exposes a superset of the `types` crate. It adds additional types that are only
//! required for the HTTP API.
use crate::Error as ServerError;
use crate::{
Error as ServerError, CONSENSUS_BLOCK_VALUE_HEADER, CONSENSUS_VERSION_HEADER,
EXECUTION_PAYLOAD_BLINDED_HEADER, EXECUTION_PAYLOAD_VALUE_HEADER,
};
use lighthouse_network::{ConnectionDirection, Enr, Multiaddr, PeerConnectionStatus};
use mediatype::{names, MediaType, MediaTypeList};
use reqwest::header::HeaderMap;
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::Value;
use ssz::{Decode, DecodeError};
@@ -1430,11 +1434,6 @@ pub mod serde_status_code {
}
}
pub enum ForkVersionedBeaconBlockType<T: EthSpec> {
Full(ForkVersionedResponse<FullBlockContents<T>>),
Blinded(ForkVersionedResponse<BlindedBeaconBlock<T>>),
}
#[cfg(test)]
mod tests {
use super::*;
@@ -1521,6 +1520,9 @@ pub enum ProduceBlockV3Response<E: EthSpec> {
Blinded(BlindedBeaconBlock<E>),
}
pub type JsonProduceBlockV3Response<E> =
ForkVersionedResponse<ProduceBlockV3Response<E>, ProduceBlockV3Metadata>;
/// A wrapper over a [`BeaconBlock`] or a [`BlockContents`].
#[derive(Debug, Encode, Serialize, Deserialize)]
#[serde(untagged)]
@@ -1535,6 +1537,27 @@ pub enum FullBlockContents<T: EthSpec> {
pub type BlockContentsTuple<T> = (BeaconBlock<T>, Option<(KzgProofs<T>, BlobsList<T>)>);
// This value should never be used
fn dummy_consensus_version() -> ForkName {
ForkName::Base
}
/// Metadata about a `ProduceBlockV3Response` which is returned in the body & headers.
#[derive(Debug, Deserialize, Serialize)]
pub struct ProduceBlockV3Metadata {
// The consensus version is serialized & deserialized by `ForkVersionedResponse`.
#[serde(
skip_serializing,
skip_deserializing,
default = "dummy_consensus_version"
)]
pub consensus_version: ForkName,
pub execution_payload_blinded: bool,
pub execution_payload_value: Uint256,
#[serde(with = "serde_utils::quoted_u64")]
pub consensus_block_value: u64,
}
impl<T: EthSpec> FullBlockContents<T> {
pub fn new(block: BeaconBlock<T>, blob_data: Option<(KzgProofs<T>, BlobsList<T>)>) -> Self {
match blob_data {
@@ -1556,13 +1579,19 @@ impl<T: EthSpec> FullBlockContents<T> {
len: bytes.len(),
expected: slot_len,
})?;
let slot = Slot::from_ssz_bytes(slot_bytes)?;
let fork_at_slot = spec.fork_name_at_slot::<T>(slot);
Self::from_ssz_bytes_for_fork(bytes, fork_at_slot)
}
match fork_at_slot {
/// SSZ decode with fork variant passed in explicitly.
pub fn from_ssz_bytes_for_fork(
bytes: &[u8],
fork_name: ForkName,
) -> Result<Self, ssz::DecodeError> {
match fork_name {
ForkName::Base | ForkName::Altair | ForkName::Merge | ForkName::Capella => {
BeaconBlock::from_ssz_bytes(bytes, spec)
BeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name)
.map(|block| FullBlockContents::Block(block))
}
ForkName::Deneb => {
@@ -1573,8 +1602,9 @@ impl<T: EthSpec> FullBlockContents<T> {
builder.register_type::<BlobsList<T>>()?;
let mut decoder = builder.build()?;
let block =
decoder.decode_next_with(|bytes| BeaconBlock::from_ssz_bytes(bytes, spec))?;
let block = decoder.decode_next_with(|bytes| {
BeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name)
})?;
let kzg_proofs = decoder.decode_next()?;
let blobs = decoder.decode_next()?;
@@ -1643,6 +1673,52 @@ impl<T: EthSpec> Into<BeaconBlock<T>> for FullBlockContents<T> {
pub type SignedBlockContentsTuple<T> = (SignedBeaconBlock<T>, Option<(KzgProofs<T>, BlobsList<T>)>);
fn parse_required_header<T>(
headers: &HeaderMap,
header_name: &str,
parse: impl FnOnce(&str) -> Result<T, String>,
) -> Result<T, String> {
let str_value = headers
.get(header_name)
.ok_or_else(|| format!("missing required header {header_name}"))?
.to_str()
.map_err(|e| format!("invalid value in {header_name}: {e}"))?;
parse(str_value)
}
impl TryFrom<&HeaderMap> for ProduceBlockV3Metadata {
type Error = String;
fn try_from(headers: &HeaderMap) -> Result<Self, Self::Error> {
let consensus_version = parse_required_header(headers, CONSENSUS_VERSION_HEADER, |s| {
s.parse::<ForkName>()
.map_err(|e| format!("invalid {CONSENSUS_VERSION_HEADER}: {e:?}"))
})?;
let execution_payload_blinded =
parse_required_header(headers, EXECUTION_PAYLOAD_BLINDED_HEADER, |s| {
s.parse::<bool>()
.map_err(|e| format!("invalid {EXECUTION_PAYLOAD_BLINDED_HEADER}: {e:?}"))
})?;
let execution_payload_value =
parse_required_header(headers, EXECUTION_PAYLOAD_VALUE_HEADER, |s| {
s.parse::<Uint256>()
.map_err(|e| format!("invalid {EXECUTION_PAYLOAD_VALUE_HEADER}: {e:?}"))
})?;
let consensus_block_value =
parse_required_header(headers, CONSENSUS_BLOCK_VALUE_HEADER, |s| {
s.parse::<u64>()
.map_err(|e| format!("invalid {CONSENSUS_BLOCK_VALUE_HEADER}: {e:?}"))
})?;
Ok(ProduceBlockV3Metadata {
consensus_version,
execution_payload_blinded,
execution_payload_value,
consensus_block_value,
})
}
}
/// A wrapper over a [`SignedBeaconBlock`] or a [`SignedBlockContents`].
#[derive(Clone, Debug, Encode, Serialize, Deserialize)]
#[serde(untagged)]