fix blob processing code

Author: realbigsean

consensus/state_processing/src/per_block_processing/eip4844/eip4844.rs

@@ -1,61 +1,71 @@
 use crate::BlockProcessingError;
 use eth2_hashing::{hash, hash_fixed};
-use itertools::Itertools;
+use itertools::{EitherOrBoth, Itertools};
 use ssz::Decode;
 use ssz_types::VariableList;
+use std::slice::Iter;
 use std::vec::IntoIter;
 use types::consts::eip4844::{BLOB_TX_TYPE, VERSIONED_HASH_VERSION_KZG};
-use types::{EthSpec, ExecPayload, KzgCommitment, Transaction, Transactions, VersionedHash};
+use types::{
+    AbstractExecPayload, BeaconBlockBodyRef, EthSpec, ExecPayload, FullPayload, FullPayloadRef,
+    KzgCommitment, Transaction, Transactions, VersionedHash,
+};
 
-pub fn process_blob_kzg_commitments<T: EthSpec, Payload: ExecPayload<T>>(
+pub fn process_blob_kzg_commitments<T: EthSpec, Payload: AbstractExecPayload<T>>(
     block_body: BeaconBlockBodyRef<T, Payload>,
 ) -> Result<(), BlockProcessingError> {
-    if let (Ok(transactions), Ok(kzg_commitments)) = (
-        block_body.execution_payload_header().transactions(),
+    if let (Ok(payload), Ok(kzg_commitments)) = (
+        block_body.execution_payload(),
         block_body.blob_kzg_commitments(),
     ) {
-        if !verify_kzg_commitments_against_transactions(transactions, kzg_commitments) {
-            return Err(BlockProcessingError::BlobVersionHashMismatch);
+        if let Some(transactions) = payload.transactions() {
+            if !verify_kzg_commitments_against_transactions::<T>(transactions, kzg_commitments)? {
+                return Err(BlockProcessingError::BlobVersionHashMismatch);
+            }
         }
     }
 
     Ok(())
 }
 
 pub fn verify_kzg_commitments_against_transactions<T: EthSpec>(
-    transactions: Transactions<T>,
+    transactions: &Transactions<T>,
     kzg_commitments: &VariableList<KzgCommitment, T::MaxBlobsPerBlock>,
 ) -> Result<bool, BlockProcessingError> {
-    let tx_versioned_hashes = transactions
+    let nested_iter = transactions
         .into_iter()
         .filter(|tx| {
             tx.get(0)
-                .map(|tx_type| tx_type == BLOB_TX_TYPE)
+                .map(|tx_type| *tx_type == BLOB_TX_TYPE)
                 .unwrap_or(false)
         })
-        .map(|tx| tx_peek_blob_versioned_hashes(tx))
-        .flatten()
-        .collect();
+        .map(|tx| tx_peek_blob_versioned_hashes::<T>(tx));
 
-    //FIXME(sean) Need to check lengths are equal here, just zipping first hides if one iter is shorter
-    // and `itertools::zip_eq` panics
-    if tx_versioned_hashes.len() != kzg_commitments.len() {
-        return Err(BlockProcessingError::BlobVersionHashMismatch);
-    }
+    itertools::process_results(nested_iter, |mut iter| {
+        let zipped_iter = iter
+            .flatten()
+            // Need to use `itertools::zip_longest` here because just zipping hides if one iter is shorter
+            // and `itertools::zip_eq` panics.
+            .zip_longest(kzg_commitments.into_iter())
+            .map(|next| match next {
+                EitherOrBoth::Both(hash, commitmnet) => Ok((hash?, commitmnet)),
+                _ => Err(BlockProcessingError::BlobVersionHashMismatch),
+            });
 
-    tx_versioned_hashes
-        .into_iter()
-        .zip(kzg_commitments.into_iter())
-        .all(|(tx_versioned_hash, commitment)| {
-            tx_versioned_hash == kzg_commitment_to_versioned_hash(committment)
+        itertools::process_results(zipped_iter, |mut iter| {
+            iter.all(|(tx_versioned_hash, commitment)| {
+                tx_versioned_hash == kzg_commitment_to_versioned_hash(commitment)
+            })
         })
+    })?
 }
 
 fn tx_peek_blob_versioned_hashes<T: EthSpec>(
-    opaque_tx: &Transaction<T>,
-) -> IntoIter<VersionedHash> {
-    //FIXME(sean) there was a first byte check for blob tx type but I think it's redundant, will raise a spec PR
-
+    opaque_tx: &Transaction<T::MaxBytesPerTransaction>,
+) -> Result<
+    impl IntoIterator<Item = Result<VersionedHash, BlockProcessingError>> + '_,
+    BlockProcessingError,
+> {
     let tx_len = opaque_tx.len();
     let message_offset = 1 + u32::from_ssz_bytes(opaque_tx.get(1..5).ok_or(
         BlockProcessingError::BlobVersionHashIndexOutOfBounds {
@@ -64,30 +74,34 @@ fn tx_peek_blob_versioned_hashes<T: EthSpec>(
         },
     )?)?;
 
+    let message_offset_usize = message_offset as usize;
+
     // field offset: 32 + 8 + 32 + 32 + 8 + 4 + 32 + 4 + 4 = 156
     let blob_versioned_hashes_offset = message_offset
         + u32::from_ssz_bytes(
             opaque_tx
-                .get((message_offset + 156)..(message_offset + 160))
+                .get((message_offset_usize + 156)..(message_offset_usize + 160))
                 .ok_or(BlockProcessingError::BlobVersionHashIndexOutOfBounds {
                     length: tx_len,
                     index: 160,
                 })?,
-        );
+        )?;
 
-    opaque_tx
-        .get(blob_versioned_hashes_offset..tx_len)
-        .ok_or(BlockProcessingError::BlobVersionHashIndexOutOfBounds {
-            length: tx_len,
-            index: 160,
-        })?
-        .into_iter()
-        .chunks(32)
-        .map(|chunk| VersionedHash::from(chunk))
+    let num_hashes = (tx_len - blob_versioned_hashes_offset as usize) / 32;
+
+    Ok((0..num_hashes).into_iter().map(move |i| {
+        let bytes = opaque_tx.get(i..i + 32).ok_or(
+            BlockProcessingError::BlobVersionHashIndexOutOfBounds {
+                length: tx_len,
+                index: i + 32,
+            },
+        )?;
+        Ok(VersionedHash::from_slice(bytes))
+    }))
 }
 
-fn kzg_commitment_to_versioned_hash(kzg_commitment: &KZGCommitment) -> VersionedHash {
-    let mut hashed_commitment = hash_fixed(kzg_commitment);
+fn kzg_commitment_to_versioned_hash(kzg_commitment: &KzgCommitment) -> VersionedHash {
+    let mut hashed_commitment = hash_fixed(&kzg_commitment.0);
     hashed_commitment[0] = VERSIONED_HASH_VERSION_KZG;
     VersionedHash::from(hashed_commitment)
 }

consensus/state_processing/src/per_block_processing/errors.rs

@@ -1,6 +1,7 @@
 use super::signature_sets::Error as SignatureSetError;
 use merkle_proof::MerkleTreeError;
 use safe_arith::ArithError;
+use ssz::DecodeError;
 use types::*;
 
 /// The error returned from the `per_block_processing` function. Indicates that a block is either
@@ -53,6 +54,7 @@ pub enum BlockProcessingError {
     BeaconStateError(BeaconStateError),
     SignatureSetError(SignatureSetError),
     SszTypesError(ssz_types::Error),
+    SszDecodeError(DecodeError),
     MerkleTreeError(MerkleTreeError),
     ArithError(ArithError),
     InconsistentBlockFork(InconsistentFork),
@@ -95,6 +97,12 @@ impl From<ssz_types::Error> for BlockProcessingError {
     }
 }
 
+impl From<DecodeError> for BlockProcessingError {
+    fn from(error: DecodeError) -> Self {
+        BlockProcessingError::SszDecodeError(error)
+    }
+}
+
 impl From<ArithError> for BlockProcessingError {
     fn from(e: ArithError) -> Self {
         BlockProcessingError::ArithError(e)

consensus/types/src/kzg_commitment.rs

@@ -9,7 +9,7 @@ use tree_hash::{PackedEncoding, TreeHash};
 
 #[derive(Derivative, Debug, Clone, Serialize, Deserialize)]
 #[derivative(PartialEq, Eq, Hash)]
-pub struct KzgCommitment(#[serde(with = "BigArray")] [u8; 48]);
+pub struct KzgCommitment(#[serde(with = "BigArray")] pub [u8; 48]);
 
 impl Display for KzgCommitment {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {

consensus/types/src/kzg_proof.rs

@@ -1,9 +1,9 @@
 use crate::test_utils::{RngCore, TestRandom};
 use serde::{Deserialize, Serialize};
+use serde_big_array::BigArray;
 use ssz::{Decode, DecodeError, Encode};
 use std::fmt;
 use tree_hash::{PackedEncoding, TreeHash};
-use serde_big_array::BigArray;
 
 const KZG_PROOF_BYTES_LEN: usize = 48;
 

consensus/types/src/payload.rs

@@ -34,6 +34,7 @@ pub trait ExecPayload<T: EthSpec>: Debug + Clone + PartialEq + Hash + TreeHash +
     fn block_hash(&self) -> ExecutionBlockHash;
     fn fee_recipient(&self) -> Address;
     fn gas_limit(&self) -> u64;
+    fn transactions(&self) -> Option<&Transactions<T>>;
 
     // Is this a default payload? (pre-merge)
     fn is_default(&self) -> bool;
@@ -191,6 +192,10 @@ impl<T: EthSpec> ExecPayload<T> for FullPayloadMerge<T> {
         self.execution_payload.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        Some(&self.execution_payload.transactions)
+    }
+
     // TODO: can this function be optimized?
     fn is_default(&self) -> bool {
         self.execution_payload == ExecutionPayloadMerge::default()
@@ -235,6 +240,10 @@ impl<T: EthSpec> ExecPayload<T> for FullPayloadCapella<T> {
         self.execution_payload.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        Some(&self.execution_payload.transactions)
+    }
+
     // TODO: can this function be optimized?
     fn is_default(&self) -> bool {
         self.execution_payload == ExecutionPayloadCapella::default()
@@ -279,6 +288,10 @@ impl<T: EthSpec> ExecPayload<T> for FullPayloadEip4844<T> {
         self.execution_payload.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        Some(&self.execution_payload.transactions)
+    }
+
     // TODO: can this function be optimized?
     fn is_default(&self) -> bool {
         self.execution_payload == ExecutionPayloadEip4844::default()
@@ -347,6 +360,13 @@ impl<T: EthSpec> ExecPayload<T> for FullPayload<T> {
         })
     }
 
+    fn transactions<'a>(&'a self) -> Option<&'a Transactions<T>> {
+        map_full_payload_ref!(&'a _, self.to_ref(), move |payload, cons| {
+            cons(payload);
+            Some(&payload.execution_payload.transactions)
+        })
+    }
+
     fn is_default(&self) -> bool {
         match self {
             Self::Merge(payload) => payload.is_default(),
@@ -428,6 +448,13 @@ impl<'b, T: EthSpec> ExecPayload<T> for FullPayloadRef<'b, T> {
         })
     }
 
+    fn transactions<'a>(&'a self) -> Option<&'a Transactions<T>> {
+        map_full_payload_ref!(&'a _, self, move |payload, cons| {
+            cons(payload);
+            Some(&payload.execution_payload.transactions)
+        })
+    }
+
     // TODO: can this function be optimized?
     fn is_default<'a>(&'a self) -> bool {
         match self {
@@ -687,6 +714,10 @@ impl<T: EthSpec> ExecPayload<T> for BlindedPayload<T> {
         }
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        None
+    }
+
     // TODO: can this function be optimized?
     fn is_default(&self) -> bool {
         match self {
@@ -773,6 +804,10 @@ impl<'b, T: EthSpec> ExecPayload<T> for BlindedPayloadRef<'b, T> {
         }
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        None
+    }
+
     // TODO: can this function be optimized?
     fn is_default<'a>(&'a self) -> bool {
         match self {
@@ -828,6 +863,10 @@ impl<T: EthSpec> ExecPayload<T> for BlindedPayloadMerge<T> {
         self.execution_payload_header.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        None
+    }
+
     fn is_default(&self) -> bool {
         self.execution_payload_header == ExecutionPayloadHeaderMerge::default()
     }
@@ -871,6 +910,10 @@ impl<T: EthSpec> ExecPayload<T> for BlindedPayloadCapella<T> {
         self.execution_payload_header.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        None
+    }
+
     fn is_default(&self) -> bool {
         self.execution_payload_header == ExecutionPayloadHeaderCapella::default()
     }
@@ -914,6 +957,10 @@ impl<T: EthSpec> ExecPayload<T> for BlindedPayloadEip4844<T> {
         self.execution_payload_header.gas_limit
     }
 
+    fn transactions(&self) -> Option<&Transactions<T>> {
+        None
+    }
+
     fn is_default(&self) -> bool {
         self.execution_payload_header == ExecutionPayloadHeaderEip4844::default()
     }