Resolve merge conflicts

beacon_node/beacon_chain/src/events.rs

@@ -25,6 +25,11 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     attester_slashing_tx: Sender<EventKind<E>>,
     bls_to_execution_change_tx: Sender<EventKind<E>>,
     block_gossip_tx: Sender<EventKind<E>>,
+    execution_payload_tx: Sender<EventKind<E>>,
+    execution_payload_gossip_tx: Sender<EventKind<E>>,
+    execution_payload_available_tx: Sender<EventKind<E>>,
+    execution_payload_bid_tx: Sender<EventKind<E>>,
+    payload_attestation_message_tx: Sender<EventKind<E>>,
 }
 
 impl<E: EthSpec> ServerSentEventHandler<E> {
@@ -51,6 +56,11 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         let (attester_slashing_tx, _) = broadcast::channel(capacity);
         let (bls_to_execution_change_tx, _) = broadcast::channel(capacity);
         let (block_gossip_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_gossip_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_available_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_bid_tx, _) = broadcast::channel(capacity);
+        let (payload_attestation_message_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -71,6 +81,11 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             attester_slashing_tx,
             bls_to_execution_change_tx,
             block_gossip_tx,
+            execution_payload_tx,
+            execution_payload_gossip_tx,
+            execution_payload_available_tx,
+            execution_payload_bid_tx,
+            payload_attestation_message_tx,
         }
     }
 
@@ -155,6 +170,26 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .block_gossip_tx
                 .send(kind)
                 .map(|count| log_count("block gossip", count)),
+            EventKind::ExecutionPayload(_) => self
+                .execution_payload_tx
+                .send(kind)
+                .map(|count| log_count("execution payload", count)),
+            EventKind::ExecutionPayloadGossip(_) => self
+                .execution_payload_gossip_tx
+                .send(kind)
+                .map(|count| log_count("execution payload gossip", count)),
+            EventKind::ExecutionPayloadAvailable(_) => self
+                .execution_payload_available_tx
+                .send(kind)
+                .map(|count| log_count("execution payload available", count)),
+            EventKind::ExecutionPayloadBid(_) => self
+                .execution_payload_bid_tx
+                .send(kind)
+                .map(|count| log_count("execution payload bid", count)),
+            EventKind::PayloadAttestationMessage(_) => self
+                .payload_attestation_message_tx
+                .send(kind)
+                .map(|count| log_count("payload attestation message", count)),
         };
         if let Err(SendError(event)) = result {
             trace!(?event, "No receivers registered to listen for event");
@@ -233,6 +268,26 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.block_gossip_tx.subscribe()
     }
 
+    pub fn subscribe_execution_payload(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_tx.subscribe()
+    }
+
+    pub fn subscribe_execution_payload_gossip(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_gossip_tx.subscribe()
+    }
+
+    pub fn subscribe_execution_payload_available(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_available_tx.subscribe()
+    }
+
+    pub fn subscribe_execution_payload_bid(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_bid_tx.subscribe()
+    }
+
+    pub fn subscribe_payload_attestation_message(&self) -> Receiver<EventKind<E>> {
+        self.payload_attestation_message_tx.subscribe()
+    }
+
     pub fn has_attestation_subscribers(&self) -> bool {
         self.attestation_tx.receiver_count() > 0
     }
@@ -296,4 +351,24 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
     pub fn has_block_gossip_subscribers(&self) -> bool {
         self.block_gossip_tx.receiver_count() > 0
     }
+
+    pub fn has_execution_payload_subscribers(&self) -> bool {
+        self.execution_payload_tx.receiver_count() > 0
+    }
+
+    pub fn has_execution_payload_gossip_subscribers(&self) -> bool {
+        self.execution_payload_gossip_tx.receiver_count() > 0
+    }
+
+    pub fn has_execution_payload_available_subscribers(&self) -> bool {
+        self.execution_payload_available_tx.receiver_count() > 0
+    }
+
+    pub fn has_execution_payload_bid_subscribers(&self) -> bool {
+        self.execution_payload_bid_tx.receiver_count() > 0
+    }
+
+    pub fn has_payload_attestation_message_subscribers(&self) -> bool {
+        self.payload_attestation_message_tx.receiver_count() > 0
+    }
 }

beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use educe::Educe;
+use eth2::types::{EventKind, SseExecutionPayloadGossip};
 use parking_lot::{Mutex, RwLock};
 use store::DatabaseBlock;
 use tracing::{Span, debug};
@@ -10,7 +11,7 @@ use types::{
 };
 
 use crate::{
-    BeaconChain, BeaconChainError, BeaconChainTypes, BeaconStore,
+    BeaconChain, BeaconChainError, BeaconChainTypes, BeaconStore, ServerSentEventHandler,
     beacon_proposer_cache::{self, BeaconProposerCache},
     canonical_head::CanonicalHead,
     payload_envelope_verification::{
@@ -28,6 +29,7 @@ pub struct GossipVerificationContext<'a, T: BeaconChainTypes> {
     pub beacon_proposer_cache: &'a Mutex<BeaconProposerCache>,
     pub validator_pubkey_cache: &'a RwLock<ValidatorPubkeyCache<T>>,
     pub genesis_validators_root: Hash256,
+    pub event_handler: &'a Option<ServerSentEventHandler<T::EthSpec>>,
 }
 
 /// Verify that an execution payload envelope is consistent with its beacon block
@@ -213,6 +215,20 @@ impl<T: BeaconChainTypes> GossipVerifiedEnvelope<T> {
             return Err(EnvelopeError::BadSignature);
         }
 
+        if let Some(event_handler) = ctx.event_handler.as_ref()
+            && event_handler.has_execution_payload_gossip_subscribers()
+        {
+            event_handler.register(EventKind::ExecutionPayloadGossip(
+                SseExecutionPayloadGossip {
+                    slot: block.slot(),
+                    builder_index,
+                    block_hash: signed_envelope.message.payload.block_hash,
+                    block_root: beacon_block_root,
+                    state_root: signed_envelope.message.state_root,
+                },
+            ));
+        }
+
         Ok(Self {
             signed_envelope,
             block,
@@ -235,6 +251,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             beacon_proposer_cache: &self.beacon_proposer_cache,
             validator_pubkey_cache: &self.validator_pubkey_cache,
             genesis_validators_root: self.genesis_validators_root,
+            event_handler: &self.event_handler,
         }
     }
 

beacon_node/beacon_chain/src/test_utils.rs

@@ -223,7 +223,7 @@ pub fn test_da_checker<E: EthSpec>(
     let slot_clock = TestingSlotClock::new(
         Slot::new(0),
         Duration::from_secs(0),
-        Duration::from_secs(spec.seconds_per_slot),
+        spec.get_slot_duration(),
     );
     let kzg = get_kzg(&spec);
     let ordered_custody_column_indices = generate_data_column_indices_rand_order::<E>();