Add block_gossip Beacon API events

beacon_node/beacon_chain/src/block_verification.rs

@@ -67,7 +67,7 @@ use crate::{
     metrics, BeaconChain, BeaconChainError, BeaconChainTypes,
 };
 use derivative::Derivative;
-use eth2::types::{EventKind, PublishBlockRequest};
+use eth2::types::{BlockGossip, EventKind, PublishBlockRequest};
 use execution_layer::PayloadStatus;
 pub use fork_choice::{AttestationFromBlock, PayloadVerificationStatus};
 use parking_lot::RwLockReadGuard;
@@ -975,6 +975,16 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // Validate the block's execution_payload (if any).
         validate_execution_payload_for_gossip(&parent_block, block.message(), chain)?;
 
+        // Beacon API block_gossip events
+        if let Some(event_handler) = chain.event_handler.as_ref() {
+            if event_handler.has_block_gossip_subscribers() {
+                event_handler.register(EventKind::BlockGossip(Box::new(BlockGossip {
+                    slot: block.slot(),
+                    block: block_root,
+                })));
+            }
+        }
+
         // Having checked the proposer index and the block root we can cache them.
         let consensus_context = ConsensusContext::new(block.slot())
             .set_current_block_root(block_root)

beacon_node/beacon_chain/src/events.rs

@@ -23,6 +23,7 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     proposer_slashing_tx: Sender<EventKind<E>>,
     attester_slashing_tx: Sender<EventKind<E>>,
     bls_to_execution_change_tx: Sender<EventKind<E>>,
+    block_gossip_tx: Sender<EventKind<E>>,
     log: Logger,
 }
 
@@ -51,6 +52,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         let (proposer_slashing_tx, _) = broadcast::channel(capacity);
         let (attester_slashing_tx, _) = broadcast::channel(capacity);
         let (bls_to_execution_change_tx, _) = broadcast::channel(capacity);
+        let (block_gossip_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -69,6 +71,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             proposer_slashing_tx,
             attester_slashing_tx,
             bls_to_execution_change_tx,
+            block_gossip_tx,
             log,
         }
     }
@@ -147,6 +150,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .bls_to_execution_change_tx
                 .send(kind)
                 .map(|count| log_count("bls to execution change", count)),
+            EventKind::BlockGossip(_) => self
+                .block_gossip_tx
+                .send(kind)
+                .map(|count| log_count("block gossip", count)),
         };
         if let Err(SendError(event)) = result {
             trace!(self.log, "No receivers registered to listen for event"; "event" => ?event);
@@ -217,6 +224,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.bls_to_execution_change_tx.subscribe()
     }
 
+    pub fn subscribe_block_gossip(&self) -> Receiver<EventKind<E>> {
+        self.block_gossip_tx.subscribe()
+    }
+
     pub fn has_attestation_subscribers(&self) -> bool {
         self.attestation_tx.receiver_count() > 0
     }
@@ -272,4 +283,8 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
     pub fn has_bls_to_execution_change_subscribers(&self) -> bool {
         self.bls_to_execution_change_tx.receiver_count() > 0
     }
+
+    pub fn has_block_gossip_subscribers(&self) -> bool {
+        self.block_gossip_tx.receiver_count() > 0
+    }
 }

beacon_node/http_api/src/lib.rs

@@ -4367,6 +4367,9 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::BlsToExecutionChange => {
                                     event_handler.subscribe_bls_to_execution_change()
                                 }
+                                api_types::EventTopic::BlockGossip => {
+                                    event_handler.subscribe_block_gossip()
+                                }
                             };
 
                             receivers.push(

beacon_node/http_api/tests/tests.rs

@@ -64,6 +64,7 @@ struct ApiTester {
     chain: Arc<BeaconChain<EphemeralHarnessType<E>>>,
     client: BeaconNodeHttpClient,
     next_block: PublishBlockRequest<E>,
+    block_gossip: PublishBlockRequest<E>,
     reorg_block: PublishBlockRequest<E>,
     attestations: Vec<Attestation<E>>,
     contribution_and_proofs: Vec<SignedContributionAndProof<E>>,
@@ -174,6 +175,11 @@ impl ApiTester {
             .await;
         let next_block = PublishBlockRequest::from(next_block);
 
+        let (next_block_gossip, _next_state_gossip) = harness
+            .make_block(head.beacon_state.clone(), harness.chain.slot().unwrap())
+            .await;
+        let block_gossip = PublishBlockRequest::from(next_block_gossip);
+
         // `make_block` adds random graffiti, so this will produce an alternate block
         let (reorg_block, _reorg_state) = harness
             .make_block(head.beacon_state.clone(), harness.chain.slot().unwrap() + 1)
@@ -286,6 +292,7 @@ impl ApiTester {
             chain,
             client,
             next_block,
+            block_gossip,
             reorg_block,
             attestations,
             contribution_and_proofs,
@@ -319,6 +326,11 @@ impl ApiTester {
             .await;
         let next_block = PublishBlockRequest::from(next_block);
 
+        let (next_block_gossip, _next_state_gossip) = harness
+            .make_block(head.beacon_state.clone(), harness.chain.slot().unwrap())
+            .await;
+        let block_gossip = PublishBlockRequest::from(next_block_gossip);
+
         // `make_block` adds random graffiti, so this will produce an alternate block
         let (reorg_block, _reorg_state) = harness
             .make_block(head.beacon_state.clone(), harness.chain.slot().unwrap())
@@ -373,6 +385,7 @@ impl ApiTester {
             chain,
             client,
             next_block,
+            block_gossip,
             reorg_block,
             attestations,
             contribution_and_proofs: vec![],
@@ -5219,6 +5232,7 @@ impl ApiTester {
             EventTopic::Attestation,
             EventTopic::VoluntaryExit,
             EventTopic::Block,
+            EventTopic::BlockGossip,
             EventTopic::Head,
             EventTopic::FinalizedCheckpoint,
             EventTopic::AttesterSlashing,
@@ -5340,6 +5354,22 @@ impl ApiTester {
             &[expected_block, expected_head, expected_finalized]
         );
 
+        // Test a block gossip event
+        self.client
+            .post_beacon_blocks(&self.block_gossip)
+            .await
+            .unwrap();
+
+        let block_gossip_events =
+            poll_events(&mut events_future, 1, Duration::from_millis(10000)).await;
+        assert_eq!(
+            block_gossip_events.as_slice(),
+            &[EventKind::BlockGossip(Box::new(BlockGossip {
+                slot: next_slot,
+                block: block_root,
+            }))]
+        );
+
         // Test a reorg event
         let mut chain_reorg_event_future = self
             .client

common/eth2/src/types.rs

@@ -954,6 +954,12 @@ pub struct SseHead {
     pub execution_optimistic: bool,
 }
 
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+
+pub struct BlockGossip {
+    pub slot: Slot,
+    pub block: Hash256,
+}
 #[derive(PartialEq, Debug, Serialize, Deserialize, Clone)]
 pub struct SseChainReorg {
     pub slot: Slot,
@@ -1083,6 +1089,7 @@ pub enum EventKind<E: EthSpec> {
     ProposerSlashing(Box<ProposerSlashing>),
     AttesterSlashing(Box<AttesterSlashing<E>>),
     BlsToExecutionChange(Box<SignedBlsToExecutionChange>),
+    BlockGossip(Box<BlockGossip>),
 }
 
 impl<E: EthSpec> EventKind<E> {
@@ -1105,6 +1112,7 @@ impl<E: EthSpec> EventKind<E> {
             EventKind::ProposerSlashing(_) => "proposer_slashing",
             EventKind::AttesterSlashing(_) => "attester_slashing",
             EventKind::BlsToExecutionChange(_) => "bls_to_execution_change",
+            EventKind::BlockGossip(_) => "block_gossip",
         }
     }
 
@@ -1200,6 +1208,9 @@ impl<E: EthSpec> EventKind<E> {
                     ServerError::InvalidServerSentEvent(format!("Bls To Execution Change: {:?}", e))
                 })?,
             )),
+            "block_gossip" => Ok(EventKind::BlockGossip(serde_json::from_str(data).map_err(
+                |e| ServerError::InvalidServerSentEvent(format!("Block Gossip: {:?}", e)),
+            )?)),
             _ => Err(ServerError::InvalidServerSentEvent(
                 "Could not parse event tag".to_string(),
             )),
@@ -1234,6 +1245,7 @@ pub enum EventTopic {
     AttesterSlashing,
     ProposerSlashing,
     BlsToExecutionChange,
+    BlockGossip,
 }
 
 impl FromStr for EventTopic {
@@ -1258,6 +1270,7 @@ impl FromStr for EventTopic {
             "attester_slashing" => Ok(EventTopic::AttesterSlashing),
             "proposer_slashing" => Ok(EventTopic::ProposerSlashing),
             "bls_to_execution_change" => Ok(EventTopic::BlsToExecutionChange),
+            "block_gossip" => Ok(EventTopic::BlockGossip),
             _ => Err("event topic cannot be parsed.".to_string()),
         }
     }
@@ -1283,6 +1296,7 @@ impl fmt::Display for EventTopic {
             EventTopic::AttesterSlashing => write!(f, "attester_slashing"),
             EventTopic::ProposerSlashing => write!(f, "proposer_slashing"),
             EventTopic::BlsToExecutionChange => write!(f, "bls_to_execution_change"),
+            EventTopic::BlockGossip => write!(f, "block_gossip"),
         }
     }
 }