Merge branch 'unstable' into into-anchor

common/eth2/Cargo.toml

@@ -7,6 +7,7 @@ edition = { workspace = true }
 
 [dependencies]
 derivative = { workspace = true }
+either = { workspace = true }
 enr = { version = "0.13.0", features = ["ed25519"] }
 eth2_keystore = { workspace = true }
 ethereum_serde_utils = { workspace = true }

common/eth2/src/lib.rs

@@ -18,6 +18,7 @@ use self::mixin::{RequestAccept, ResponseOptional};
 use self::types::{Error as ResponseError, *};
 use ::types::fork_versioned_response::ExecutionOptimisticFinalizedForkVersionedResponse;
 use derivative::Derivative;
+use either::Either;
 use futures::Stream;
 use futures_util::StreamExt;
 use libp2p_identity::PeerId;
@@ -48,6 +49,7 @@ pub const CONSENSUS_BLOCK_VALUE_HEADER: &str = "Eth-Consensus-Block-Value";
 
 pub const CONTENT_TYPE_HEADER: &str = "Content-Type";
 pub const SSZ_CONTENT_TYPE_HEADER: &str = "application/octet-stream";
+pub const JSON_CONTENT_TYPE_HEADER: &str = "application/json";
 
 #[derive(Debug)]
 pub enum Error {
@@ -111,9 +113,9 @@ impl Error {
             Error::InvalidSignatureHeader => None,
             Error::MissingSignatureHeader => None,
             Error::InvalidJson(_) => None,
+            Error::InvalidSsz(_) => None,
             Error::InvalidServerSentEvent(_) => None,
             Error::InvalidHeaders(_) => None,
-            Error::InvalidSsz(_) => None,
             Error::TokenReadError(..) => None,
             Error::NoServerPubkey | Error::NoToken => None,
         }
@@ -1324,9 +1326,9 @@ impl BeaconNodeHttpClient {
     }
 
     /// `POST v2/beacon/pool/attestations`
-    pub async fn post_beacon_pool_attestations_v2(
+    pub async fn post_beacon_pool_attestations_v2<E: EthSpec>(
         &self,
-        attestations: &[SingleAttestation],
+        attestations: Either<Vec<Attestation<E>>, Vec<SingleAttestation>>,
         fork_name: ForkName,
     ) -> Result<(), Error> {
         let mut path = self.eth_path(V2)?;
@@ -1337,13 +1339,26 @@ impl BeaconNodeHttpClient {
             .push("pool")
             .push("attestations");
 
-        self.post_with_timeout_and_consensus_header(
-            path,
-            &attestations,
-            self.timeouts.attestation,
-            fork_name,
-        )
-        .await?;
+        match attestations {
+            Either::Right(attestations) => {
+                self.post_with_timeout_and_consensus_header(
+                    path,
+                    &attestations,
+                    self.timeouts.attestation,
+                    fork_name,
+                )
+                .await?;
+            }
+            Either::Left(attestations) => {
+                self.post_with_timeout_and_consensus_header(
+                    path,
+                    &attestations,
+                    self.timeouts.attestation,
+                    fork_name,
+                )
+                .await?;
+            }
+        };
 
         Ok(())
     }

common/eth2/src/types.rs

@@ -585,12 +585,20 @@ pub struct IdentityData {
     pub metadata: MetaData,
 }
 
+#[superstruct(
+    variants(V2, V3),
+    variant_attributes(derive(Clone, Debug, PartialEq, Serialize, Deserialize))
+)]
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
 pub struct MetaData {
     #[serde(with = "serde_utils::quoted_u64")]
     pub seq_number: u64,
     pub attnets: String,
     pub syncnets: String,
+    #[superstruct(only(V3))]
+    #[serde(with = "serde_utils::quoted_u64")]
+    pub custody_group_count: u64,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -1635,7 +1643,7 @@ impl<E: EthSpec> FullBlockContents<E> {
     }
 
     /// SSZ decode with fork variant determined by slot.
-    pub fn from_ssz_bytes(bytes: &[u8], spec: &ChainSpec) -> Result<Self, ssz::DecodeError> {
+    pub fn from_ssz_bytes(bytes: &[u8], spec: &ChainSpec) -> Result<Self, DecodeError> {
         let slot_len = <Slot as Decode>::ssz_fixed_len();
         let slot_bytes = bytes
             .get(0..slot_len)
@@ -1649,10 +1657,7 @@ impl<E: EthSpec> FullBlockContents<E> {
     }
 
     /// SSZ decode with fork variant passed in explicitly.
-    pub fn from_ssz_bytes_for_fork(
-        bytes: &[u8],
-        fork_name: ForkName,
-    ) -> Result<Self, ssz::DecodeError> {
+    pub fn from_ssz_bytes_for_fork(bytes: &[u8], fork_name: ForkName) -> Result<Self, DecodeError> {
         if fork_name.deneb_enabled() {
             let mut builder = ssz::SszDecoderBuilder::new(bytes);
 
@@ -1807,7 +1812,7 @@ impl<E: EthSpec> PublishBlockRequest<E> {
     }
 
     /// SSZ decode with fork variant determined by `fork_name`.
-    pub fn from_ssz_bytes(bytes: &[u8], fork_name: ForkName) -> Result<Self, ssz::DecodeError> {
+    pub fn from_ssz_bytes(bytes: &[u8], fork_name: ForkName) -> Result<Self, DecodeError> {
         if fork_name.deneb_enabled() {
             let mut builder = ssz::SszDecoderBuilder::new(bytes);
             builder.register_anonymous_variable_length_item()?;
@@ -1816,7 +1821,7 @@ impl<E: EthSpec> PublishBlockRequest<E> {
 
             let mut decoder = builder.build()?;
             let block = decoder.decode_next_with(|bytes| {
-                SignedBeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name)
+                SignedBeaconBlock::from_ssz_bytes_by_fork(bytes, fork_name)
             })?;
             let kzg_proofs = decoder.decode_next()?;
             let blobs = decoder.decode_next()?;
@@ -1825,7 +1830,7 @@ impl<E: EthSpec> PublishBlockRequest<E> {
                 Some((kzg_proofs, blobs)),
             ))
         } else {
-            SignedBeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name)
+            SignedBeaconBlock::from_ssz_bytes_by_fork(bytes, fork_name)
                 .map(|block| PublishBlockRequest::Block(Arc::new(block)))
         }
     }
@@ -1917,6 +1922,24 @@ pub enum FullPayloadContents<E: EthSpec> {
     PayloadAndBlobs(ExecutionPayloadAndBlobs<E>),
 }
 
+impl<E: EthSpec> ForkVersionDecode for FullPayloadContents<E> {
+    fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result<Self, DecodeError> {
+        if fork_name.deneb_enabled() {
+            Ok(Self::PayloadAndBlobs(
+                ExecutionPayloadAndBlobs::from_ssz_bytes_by_fork(bytes, fork_name)?,
+            ))
+        } else if fork_name.bellatrix_enabled() {
+            Ok(Self::Payload(ExecutionPayload::from_ssz_bytes_by_fork(
+                bytes, fork_name,
+            )?))
+        } else {
+            Err(ssz::DecodeError::BytesInvalid(format!(
+                "FullPayloadContents decoding for {fork_name} not implemented"
+            )))
+        }
+    }
+}
+
 impl<E: EthSpec> FullPayloadContents<E> {
     pub fn new(
         execution_payload: ExecutionPayload<E>,
@@ -1983,6 +2006,36 @@ pub struct ExecutionPayloadAndBlobs<E: EthSpec> {
     pub blobs_bundle: BlobsBundle<E>,
 }
 
+impl<E: EthSpec> ForkVersionDecode for ExecutionPayloadAndBlobs<E> {
+    fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result<Self, DecodeError> {
+        let mut builder = ssz::SszDecoderBuilder::new(bytes);
+        builder.register_anonymous_variable_length_item()?;
+        builder.register_type::<BlobsBundle<E>>()?;
+        let mut decoder = builder.build()?;
+
+        if fork_name.deneb_enabled() {
+            let execution_payload = decoder.decode_next_with(|bytes| {
+                ExecutionPayload::from_ssz_bytes_by_fork(bytes, fork_name)
+            })?;
+            let blobs_bundle = decoder.decode_next()?;
+            Ok(Self {
+                execution_payload,
+                blobs_bundle,
+            })
+        } else {
+            Err(DecodeError::BytesInvalid(format!(
+                "ExecutionPayloadAndBlobs decoding for {fork_name} not implemented"
+            )))
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum ContentType {
+    Json,
+    Ssz,
+}
+
 #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize, Encode, Decode)]
 #[serde(bound = "E: EthSpec")]
 pub struct BlobsBundle<E: EthSpec> {