Fix todos in deneb code (#4547)

* Low hanging fruits

* Remove unnecessary todo

I think it's fine to not handle this since the calling functions handle the error.
No specific reason imo to handle it in the function as well.

* Rename BlobError to GossipBlobError

I feel this signified better what the error is for. The BlobError was only for failures when gossip
verifying a blob. We cannot get this error when doing rpc validation

* Remove the BlockError::BlobValidation variant

This error was only there to appease gossip verification before publish.
It's unclear how to peer score this error since this cannot actually occur during any
block verification flows.
This commit introuduces an additional error type BlockContentsError to better represent the
Error type

* Add docs for peer scoring (or lack thereof) of AvailabilityCheck errors

* I do not see a non-convoluted way of doing this. Okay to have some redundant code here

* Removing this to catch the failure red handed

* Fix compilation

* Cannot be deleted because some tests assume the trait impl

Also useful to have around for testing in the future imo

* Add some metrics and logs

* Only process `Imported` variant in sync_methods

The only additional thing for other variants that might be useful is logging. We can do that
later if required

* Convert to TryFrom

Not really sure where this would be used, but just did what the comment says.
Could consider just returning the Block variant for a deneb block in the From version

* Unlikely to change now

* This is fine as this is max_rpc_size per rpc chunk (for blobs, it would be 128kb max)

* Log count instead of individual blobs, can delete log later if it becomes too annoying.

* Add block production blob verification timer

* Extend block_straemer test to deneb

* Remove dbg statement

* Fix tests

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -5,11 +5,12 @@ use crate::{
     sync::SyncMessage,
 };
 
-use beacon_chain::blob_verification::{BlobError, GossipVerifiedBlob};
+use beacon_chain::blob_verification::{GossipBlobError, GossipVerifiedBlob};
 use beacon_chain::block_verification_types::AsBlock;
 use beacon_chain::store::Error;
 use beacon_chain::{
     attestation_verification::{self, Error as AttnError, VerifiedAttestation},
+    data_availability_checker::AvailabilityCheckError,
     light_client_finality_update_verification::Error as LightClientFinalityUpdateError,
     light_client_optimistic_update_verification::Error as LightClientOptimisticUpdateError,
     observed_operations::ObservationOutcome,
@@ -598,7 +599,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         }
     }
 
-    // TODO: docs
     #[allow(clippy::too_many_arguments)]
     pub async fn process_gossip_blob(
         self: &Arc<Self>,
@@ -635,7 +635,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             }
             Err(err) => {
                 match err {
-                    BlobError::BlobParentUnknown(blob) => {
+                    GossipBlobError::BlobParentUnknown(blob) => {
                         debug!(
                             self.log,
                             "Unknown parent hash for blob";
@@ -645,11 +645,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                         );
                         self.send_sync_message(SyncMessage::UnknownParentBlob(peer_id, blob));
                     }
-                    BlobError::ProposerSignatureInvalid
-                    | BlobError::UnknownValidator(_)
-                    | BlobError::ProposerIndexMismatch { .. }
-                    | BlobError::BlobIsNotLaterThanParent { .. }
-                    | BlobError::InvalidSubnet { .. } => {
+                    GossipBlobError::ProposerSignatureInvalid
+                    | GossipBlobError::UnknownValidator(_)
+                    | GossipBlobError::ProposerIndexMismatch { .. }
+                    | GossipBlobError::BlobIsNotLaterThanParent { .. }
+                    | GossipBlobError::InvalidSubnet { .. } => {
                         warn!(
                             self.log,
                             "Could not verify blob sidecar for gossip. Rejecting the blob sidecar";
@@ -670,10 +670,10 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                             MessageAcceptance::Reject,
                         );
                     }
-                    BlobError::FutureSlot { .. }
-                    | BlobError::BeaconChainError(_)
-                    | BlobError::RepeatBlob { .. }
-                    | BlobError::PastFinalizedSlot { .. } => {
+                    GossipBlobError::FutureSlot { .. }
+                    | GossipBlobError::BeaconChainError(_)
+                    | GossipBlobError::RepeatBlob { .. }
+                    | GossipBlobError::PastFinalizedSlot { .. } => {
                         warn!(
                             self.log,
                             "Could not verify blob sidecar for gossip. Ignoring the blob sidecar";
@@ -710,11 +710,24 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         let blob_slot = verified_blob.slot();
         let blob_index = verified_blob.id().index;
         match self.chain.process_blob(verified_blob).await {
-            Ok(AvailabilityProcessingStatus::Imported(_hash)) => {
-                //TODO(sean) add metrics and logging
+            Ok(AvailabilityProcessingStatus::Imported(hash)) => {
+                // Note: Reusing block imported metric here
+                metrics::inc_counter(&metrics::BEACON_PROCESSOR_GOSSIP_BLOCK_IMPORTED_TOTAL);
+                info!(
+                    self.log,
+                    "Gossipsub blob processed, imported fully available block";
+                    "hash" => %hash
+                );
                 self.chain.recompute_head_at_current_slot().await;
             }
             Ok(AvailabilityProcessingStatus::MissingComponents(slot, block_hash)) => {
+                debug!(
+                    self.log,
+                    "Missing block components for gossip verified blob";
+                    "slot" => %blob_slot,
+                    "blob_index" => %blob_index,
+                    "blob_root" => %blob_root,
+                );
                 self.send_sync_message(SyncMessage::MissingGossipBlockComponents(
                     slot, peer_id, block_hash,
                 ));
@@ -954,14 +967,12 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                 );
                 return None;
             }
-            Err(e @ BlockError::BlobValidation(_)) | Err(e @ BlockError::AvailabilityCheck(_)) => {
-                warn!(self.log, "Could not verify block against known blobs in gossip. Rejecting the block";
-                            "error" => %e);
-                self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Reject);
-                self.gossip_penalize_peer(
-                    peer_id,
-                    PeerAction::LowToleranceError,
-                    "gossip_blob_low",
+            // Note: This error variant cannot be reached when doing gossip validation
+            // as we do not do availability checks here.
+            Err(e @ BlockError::AvailabilityCheck(_)) => {
+                crit!(self.log, "Internal block gossip validation error. Availability check during
+                 gossip validation";
+                    "error" => %e
                 );
                 return None;
             }
@@ -1142,6 +1153,43 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                     "error" => %e
                 );
             }
+            Err(BlockError::AvailabilityCheck(err)) => {
+                match err {
+                    AvailabilityCheckError::KzgNotInitialized
+                    | AvailabilityCheckError::SszTypes(_)
+                    | AvailabilityCheckError::MissingBlobs
+                    | AvailabilityCheckError::StoreError(_)
+                    | AvailabilityCheckError::DecodeError(_) => {
+                        crit!(
+                            self.log,
+                            "Internal availability check error";
+                            "error" => ?err,
+                        );
+                    }
+                    AvailabilityCheckError::Kzg(_)
+                    | AvailabilityCheckError::KzgVerificationFailed
+                    | AvailabilityCheckError::NumBlobsMismatch { .. }
+                    | AvailabilityCheckError::TxKzgCommitmentMismatch(_)
+                    | AvailabilityCheckError::BlobIndexInvalid(_)
+                    | AvailabilityCheckError::UnorderedBlobs { .. }
+                    | AvailabilityCheckError::BlockBlobRootMismatch { .. }
+                    | AvailabilityCheckError::BlockBlobSlotMismatch { .. }
+                    | AvailabilityCheckError::KzgCommitmentMismatch { .. } => {
+                        // Note: we cannot penalize the peer that sent us the block
+                        // over gossip here because these errors imply either an issue
+                        // with:
+                        // 1. Blobs we have received over non-gossip sources
+                        //    (from potentially other peers)
+                        // 2. The proposer being malicious and sending inconsistent
+                        //    blocks and blobs.
+                        warn!(
+                            self.log,
+                            "Received invalid blob or malicious proposer";
+                            "error" => ?err
+                        );
+                    }
+                }
+            }
             other => {
                 debug!(
                     self.log,

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -222,8 +222,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
 
         // RPC block imported, regardless of process type
-        //TODO(sean) do we need to do anything here for missing blobs? or is passing the result
-        // along to sync enough?
         if let &Ok(AvailabilityProcessingStatus::Imported(hash)) = &result {
             info!(self.log, "New RPC block received"; "slot" => slot, "hash" => %hash);
 

beacon_node/network/src/sync/block_lookups/mod.rs

@@ -928,8 +928,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
                     BlockError::AvailabilityCheck(
                         AvailabilityCheckError::KzgVerificationFailed,
                     )
-                    | BlockError::AvailabilityCheck(AvailabilityCheckError::Kzg(_))
-                    | BlockError::BlobValidation(_) => {
+                    | BlockError::AvailabilityCheck(AvailabilityCheckError::Kzg(_)) => {
                         warn!(self.log, "Blob validation failure"; "root" => %root, "peer_id" => %peer_id);
                         if let Ok(blob_peer) = request_ref.processing_peer(ResponseType::Blob) {
                             cx.report_peer(

beacon_node/network/src/sync/block_lookups/tests.rs

@@ -1143,7 +1143,7 @@ fn test_same_chain_race_condition() {
 
 mod deneb_only {
     use super::*;
-    use beacon_chain::blob_verification::BlobError;
+    use beacon_chain::data_availability_checker::AvailabilityCheckError;
     use std::ops::IndexMut;
     use std::str::FromStr;
 
@@ -1509,8 +1509,8 @@ mod deneb_only {
         fn invalid_blob_processed(mut self) -> Self {
             self.bl.single_block_component_processed(
                 self.blob_req_id.expect("blob request id"),
-                BlockProcessingResult::Err(BlockError::BlobValidation(
-                    BlobError::ProposerSignatureInvalid,
+                BlockProcessingResult::Err(BlockError::AvailabilityCheck(
+                    AvailabilityCheckError::KzgVerificationFailed,
                 )),
                 ResponseType::Blob,
                 &mut self.cx,