Add blob_sidecar event to SSE (#4790)

* Add `blob_sidecar` event to SSE.

* Return 202 if a block is published but failed blob validation when validation level is `Gossip`.

* Move `BlobSidecar` event to `process_gossip_blob` and add test.

* Emit `BlobSidecar` event when blobs are received over rpc.

* Improve test assertions on `SseBlobSidecar`s.

* Add quotes to blob index serialization in `SseBlobSidecar`

Co-authored-by: realbigsean <seananderson33@GMAIL.com>

---------

Co-authored-by: realbigsean <seananderson33@GMAIL.com>

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -70,7 +70,7 @@ use crate::{
     kzg_utils, metrics, AvailabilityPendingExecutedBlock, BeaconChainError, BeaconForkChoiceStore,
     BeaconSnapshot, CachedHead,
 };
-use eth2::types::{EventKind, SseBlock, SseExtendedPayloadAttributes, SyncDuty};
+use eth2::types::{EventKind, SseBlobSidecar, SseBlock, SseExtendedPayloadAttributes, SyncDuty};
 use execution_layer::{
     BlockProposalContents, BuilderParams, ChainHealth, ExecutionLayer, FailedCondition,
     PayloadAttributes, PayloadStatus,
@@ -2809,6 +2809,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown);
         }
 
+        if let Some(event_handler) = self.event_handler.as_ref() {
+            if event_handler.has_blob_sidecar_subscribers() {
+                event_handler.register(EventKind::BlobSidecar(SseBlobSidecar::from_blob_sidecar(
+                    blob.as_blob(),
+                )));
+            }
+        }
+
         self.data_availability_checker
             .notify_gossip_blob(blob.as_blob().slot, block_root, &blob);
         let r = self.check_gossip_blob_availability_and_import(blob).await;
@@ -2833,6 +2841,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown);
         }
 
+        if let Some(event_handler) = self.event_handler.as_ref() {
+            if event_handler.has_blob_sidecar_subscribers() {
+                for blob in blobs.iter().filter_map(|maybe_blob| maybe_blob.as_ref()) {
+                    event_handler.register(EventKind::BlobSidecar(
+                        SseBlobSidecar::from_blob_sidecar(blob),
+                    ));
+                }
+            }
+        }
+
         self.data_availability_checker
             .notify_rpc_blobs(slot, block_root, &blobs);
         let r = self

beacon_node/beacon_chain/src/blob_verification.rs

@@ -10,7 +10,7 @@ use crate::block_verification::cheap_state_advance_to_obtain_committees;
 use crate::data_availability_checker::AvailabilityCheckError;
 use crate::kzg_utils::{validate_blob, validate_blobs};
 use crate::{metrics, BeaconChainError};
-use kzg::Kzg;
+use kzg::{Kzg, KzgCommitment};
 use slog::{debug, warn};
 use ssz_derive::{Decode, Encode};
 use ssz_types::VariableList;
@@ -182,6 +182,12 @@ impl<T: BeaconChainTypes> GossipVerifiedBlob<T> {
     pub fn slot(&self) -> Slot {
         self.blob.message.slot
     }
+    pub fn index(&self) -> u64 {
+        self.blob.message.index
+    }
+    pub fn kzg_commitment(&self) -> KzgCommitment {
+        self.blob.message.kzg_commitment
+    }
     pub fn proposer_index(&self) -> u64 {
         self.blob.message.proposer_index
     }

beacon_node/beacon_chain/src/events.rs

@@ -9,6 +9,7 @@ const DEFAULT_CHANNEL_CAPACITY: usize = 16;
 pub struct ServerSentEventHandler<T: EthSpec> {
     attestation_tx: Sender<EventKind<T>>,
     block_tx: Sender<EventKind<T>>,
+    blob_sidecar_tx: Sender<EventKind<T>>,
     finalized_tx: Sender<EventKind<T>>,
     head_tx: Sender<EventKind<T>>,
     exit_tx: Sender<EventKind<T>>,
@@ -31,6 +32,7 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
     pub fn new_with_capacity(log: Logger, capacity: usize) -> Self {
         let (attestation_tx, _) = broadcast::channel(capacity);
         let (block_tx, _) = broadcast::channel(capacity);
+        let (blob_sidecar_tx, _) = broadcast::channel(capacity);
         let (finalized_tx, _) = broadcast::channel(capacity);
         let (head_tx, _) = broadcast::channel(capacity);
         let (exit_tx, _) = broadcast::channel(capacity);
@@ -43,6 +45,7 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         Self {
             attestation_tx,
             block_tx,
+            blob_sidecar_tx,
             finalized_tx,
             head_tx,
             exit_tx,
@@ -73,6 +76,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
                 .block_tx
                 .send(kind)
                 .map(|count| log_count("block", count)),
+            EventKind::BlobSidecar(_) => self
+                .blob_sidecar_tx
+                .send(kind)
+                .map(|count| log_count("blob sidecar", count)),
             EventKind::FinalizedCheckpoint(_) => self
                 .finalized_tx
                 .send(kind)
@@ -119,6 +126,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.block_tx.subscribe()
     }
 
+    pub fn subscribe_blob_sidecar(&self) -> Receiver<EventKind<T>> {
+        self.blob_sidecar_tx.subscribe()
+    }
+
     pub fn subscribe_finalized(&self) -> Receiver<EventKind<T>> {
         self.finalized_tx.subscribe()
     }
@@ -159,6 +170,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.block_tx.receiver_count() > 0
     }
 
+    pub fn has_blob_sidecar_subscribers(&self) -> bool {
+        self.blob_sidecar_tx.receiver_count() > 0
+    }
+
     pub fn has_finalized_subscribers(&self) -> bool {
         self.finalized_tx.receiver_count() > 0
     }

beacon_node/beacon_chain/tests/events.rs

@@ -0,0 +1,99 @@
+use beacon_chain::blob_verification::GossipVerifiedBlob;
+use beacon_chain::test_utils::BeaconChainHarness;
+use bls::Signature;
+use eth2::types::{EventKind, SseBlobSidecar};
+use rand::rngs::StdRng;
+use rand::SeedableRng;
+use std::marker::PhantomData;
+use std::sync::Arc;
+use types::blob_sidecar::FixedBlobSidecarList;
+use types::{BlobSidecar, EthSpec, ForkName, MinimalEthSpec, SignedBlobSidecar};
+
+type E = MinimalEthSpec;
+
+/// Verifies that a blob event is emitted when a gossip verified blob is received via gossip or the publish block API.  
+#[tokio::test]
+async fn blob_sidecar_event_on_process_gossip_blob() {
+    let spec = ForkName::Deneb.make_genesis_spec(E::default_spec());
+    let harness = BeaconChainHarness::builder(E::default())
+        .spec(spec)
+        .deterministic_keypairs(8)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    // subscribe to blob sidecar events
+    let event_handler = harness.chain.event_handler.as_ref().unwrap();
+    let mut blob_event_receiver = event_handler.subscribe_blob_sidecar();
+
+    // build and process a gossip verified blob
+    let kzg = harness.chain.kzg.as_ref().unwrap();
+    let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
+    let signed_sidecar = SignedBlobSidecar {
+        message: BlobSidecar::random_valid(&mut rng, kzg)
+            .map(Arc::new)
+            .unwrap(),
+        signature: Signature::empty(),
+        _phantom: PhantomData,
+    };
+    let gossip_verified_blob = GossipVerifiedBlob::__assumed_valid(signed_sidecar);
+    let expected_sse_blobs = SseBlobSidecar::from_blob_sidecar(gossip_verified_blob.as_blob());
+
+    let _ = harness
+        .chain
+        .process_gossip_blob(gossip_verified_blob)
+        .await
+        .unwrap();
+
+    let sidecar_event = blob_event_receiver.try_recv().unwrap();
+    assert_eq!(sidecar_event, EventKind::BlobSidecar(expected_sse_blobs));
+}
+
+/// Verifies that a blob event is emitted when blobs are received via RPC.  
+#[tokio::test]
+async fn blob_sidecar_event_on_process_rpc_blobs() {
+    let spec = ForkName::Deneb.make_genesis_spec(E::default_spec());
+    let harness = BeaconChainHarness::builder(E::default())
+        .spec(spec)
+        .deterministic_keypairs(8)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    // subscribe to blob sidecar events
+    let event_handler = harness.chain.event_handler.as_ref().unwrap();
+    let mut blob_event_receiver = event_handler.subscribe_blob_sidecar();
+
+    // build and process multiple rpc blobs
+    let kzg = harness.chain.kzg.as_ref().unwrap();
+    let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
+
+    let blob_1 = BlobSidecar::random_valid(&mut rng, kzg)
+        .map(Arc::new)
+        .unwrap();
+    let blob_2 = Arc::new(BlobSidecar {
+        index: 1,
+        ..BlobSidecar::random_valid(&mut rng, kzg).unwrap()
+    });
+    let blobs = FixedBlobSidecarList::from(vec![Some(blob_1.clone()), Some(blob_2.clone())]);
+    let expected_sse_blobs = vec![
+        SseBlobSidecar::from_blob_sidecar(blob_1.as_ref()),
+        SseBlobSidecar::from_blob_sidecar(blob_2.as_ref()),
+    ];
+
+    let _ = harness
+        .chain
+        .process_rpc_blobs(blob_1.slot, blob_1.block_root, blobs)
+        .await
+        .unwrap();
+
+    let mut sse_blobs: Vec<SseBlobSidecar> = vec![];
+    while let Ok(sidecar_event) = blob_event_receiver.try_recv() {
+        if let EventKind::BlobSidecar(sse_blob_sidecar) = sidecar_event {
+            sse_blobs.push(sse_blob_sidecar);
+        } else {
+            panic!("`BlobSidecar` event kind expected.");
+        }
+    }
+    assert_eq!(sse_blobs, expected_sse_blobs);
+}

beacon_node/beacon_chain/tests/main.rs

@@ -2,6 +2,7 @@ mod attestation_production;
 mod attestation_verification;
 mod block_verification;
 mod capella;
+mod events;
 mod merge;
 mod op_verification;
 mod payload_invalidation;