Add bls_to_execution_change topic to eth1/v1/events

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2554,6 +2554,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         bls_to_execution_change: SigVerifiedOp<SignedBlsToExecutionChange, T::EthSpec>,
         received_pre_capella: ReceivedPreCapella,
     ) -> bool {
+        if let Some(event_handler) = self.event_handler.as_ref() {
+            if event_handler.has_bls_to_execution_change_subscribers() {
+                event_handler.register(EventKind::BlsToExecutionChange(Box::new(
+                    bls_to_execution_change.clone().into_inner(),
+                )));
+            }
+        }
+
         if self.eth1_chain.is_some() {
             self.op_pool
                 .insert_bls_to_execution_change(bls_to_execution_change, received_pre_capella)

beacon_node/beacon_chain/src/events.rs

@@ -22,6 +22,7 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     block_reward_tx: Sender<EventKind<E>>,
     proposer_slashing_tx: Sender<EventKind<E>>,
     attester_slashing_tx: Sender<EventKind<E>>,
+    bls_to_execution_change_tx: Sender<EventKind<E>>,
     log: Logger,
 }
 
@@ -49,6 +50,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         let (block_reward_tx, _) = broadcast::channel(capacity);
         let (proposer_slashing_tx, _) = broadcast::channel(capacity);
         let (attester_slashing_tx, _) = broadcast::channel(capacity);
+        let (bls_to_execution_change_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -66,6 +68,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             block_reward_tx,
             proposer_slashing_tx,
             attester_slashing_tx,
+            bls_to_execution_change_tx,
             log,
         }
     }
@@ -140,6 +143,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .attester_slashing_tx
                 .send(kind)
                 .map(|count| log_count("attester slashing", count)),
+            EventKind::BlsToExecutionChange(_) => self
+                .bls_to_execution_change_tx
+                .send(kind)
+                .map(|count| log_count("bls to execution change", count)),
         };
         if let Err(SendError(event)) = result {
             trace!(self.log, "No receivers registered to listen for event"; "event" => ?event);
@@ -206,6 +213,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.proposer_slashing_tx.subscribe()
     }
 
+    pub fn subscribe_bls_to_execution_change(&self) -> Receiver<EventKind<E>> {
+        self.bls_to_execution_change_tx.subscribe()
+    }
+
     pub fn has_attestation_subscribers(&self) -> bool {
         self.attestation_tx.receiver_count() > 0
     }
@@ -257,4 +268,8 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
     pub fn has_attester_slashing_subscribers(&self) -> bool {
         self.attester_slashing_tx.receiver_count() > 0
     }
+
+    pub fn has_bls_to_execution_change_subscribers(&self) -> bool {
+        self.bls_to_execution_change_tx.receiver_count() > 0
+    }
 }

beacon_node/http_api/src/lib.rs

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

beacon_node/http_api/tests/tests.rs

@@ -70,6 +70,7 @@ struct ApiTester {
     attester_slashing: AttesterSlashing<E>,
     proposer_slashing: ProposerSlashing,
     voluntary_exit: SignedVoluntaryExit,
+    bls_to_execution_change: SignedBlsToExecutionChange,
     network_rx: NetworkReceivers<E>,
     local_enr: Enr,
     external_peer_id: PeerId,
@@ -128,6 +129,7 @@ impl ApiTester {
             })
             .logger(logging::test_logger())
             .deterministic_keypairs(VALIDATOR_COUNT)
+            .deterministic_withdrawal_keypairs(VALIDATOR_COUNT)
             .fresh_ephemeral_store()
             .mock_execution_layer_with_config()
             .build();
@@ -223,6 +225,7 @@ impl ApiTester {
         let attester_slashing = harness.make_attester_slashing(vec![0, 1]);
         let proposer_slashing = harness.make_proposer_slashing(2);
         let voluntary_exit = harness.make_voluntary_exit(3, harness.chain.epoch().unwrap());
+        let bls_to_execution_change = harness.make_bls_to_execution_change(4, Address::zero());
 
         let chain = harness.chain.clone();
 
@@ -289,6 +292,7 @@ impl ApiTester {
             attester_slashing,
             proposer_slashing,
             voluntary_exit,
+            bls_to_execution_change,
             network_rx,
             local_enr,
             external_peer_id,
@@ -301,6 +305,7 @@ impl ApiTester {
             BeaconChainHarness::builder(MainnetEthSpec)
                 .default_spec()
                 .deterministic_keypairs(VALIDATOR_COUNT)
+                .deterministic_withdrawal_keypairs(VALIDATOR_COUNT)
                 .fresh_ephemeral_store()
                 .build(),
         );
@@ -336,6 +341,7 @@ impl ApiTester {
         let attester_slashing = harness.make_attester_slashing(vec![0, 1]);
         let proposer_slashing = harness.make_proposer_slashing(2);
         let voluntary_exit = harness.make_voluntary_exit(3, harness.chain.epoch().unwrap());
+        let bls_to_execution_change = harness.make_bls_to_execution_change(4, Address::zero());
 
         let chain = harness.chain.clone();
 
@@ -373,6 +379,7 @@ impl ApiTester {
             attester_slashing,
             proposer_slashing,
             voluntary_exit,
+            bls_to_execution_change,
             network_rx,
             local_enr,
             external_peer_id,
@@ -5216,6 +5223,7 @@ impl ApiTester {
             EventTopic::FinalizedCheckpoint,
             EventTopic::AttesterSlashing,
             EventTopic::ProposerSlashing,
+            EventTopic::BlsToExecutionChange,
         ];
         let mut events_future = self
             .client
@@ -5258,6 +5266,20 @@ impl ApiTester {
             &[EventKind::VoluntaryExit(self.voluntary_exit.clone())]
         );
 
+        // Produce a BLS to execution change event
+        self.client
+            .post_beacon_pool_bls_to_execution_changes(&[self.bls_to_execution_change.clone()])
+            .await
+            .unwrap();
+
+        let bls_events = poll_events(&mut events_future, 1, Duration::from_millis(10000)).await;
+        assert_eq!(
+            bls_events.as_slice(),
+            &[EventKind::BlsToExecutionChange(Box::new(
+                self.bls_to_execution_change.clone()
+            ))]
+        );
+
         // Submit the next block, which is on an epoch boundary, so this will produce a finalized
         // checkpoint event, head event, and block event
         let block_root = self.next_block.signed_block().canonical_root();

common/eth2/src/types.rs

@@ -1082,6 +1082,7 @@ pub enum EventKind<E: EthSpec> {
     PayloadAttributes(VersionedSsePayloadAttributes),
     ProposerSlashing(Box<ProposerSlashing>),
     AttesterSlashing(Box<AttesterSlashing<E>>),
+    BlsToExecutionChange(Box<SignedBlsToExecutionChange>),
 }
 
 impl<E: EthSpec> EventKind<E> {
@@ -1103,6 +1104,7 @@ impl<E: EthSpec> EventKind<E> {
             EventKind::BlockReward(_) => "block_reward",
             EventKind::ProposerSlashing(_) => "proposer_slashing",
             EventKind::AttesterSlashing(_) => "attester_slashing",
+            EventKind::BlsToExecutionChange(_) => "bls_to_execution_change",
         }
     }
 
@@ -1193,6 +1195,11 @@ impl<E: EthSpec> EventKind<E> {
                     ServerError::InvalidServerSentEvent(format!("Proposer Slashing: {:?}", e))
                 })?,
             )),
+            "bls_to_execution_change" => Ok(EventKind::BlsToExecutionChange(
+                serde_json::from_str(data).map_err(|e| {
+                    ServerError::InvalidServerSentEvent(format!("Bls To Execution Change: {:?}", e))
+                })?,
+            )),
             _ => Err(ServerError::InvalidServerSentEvent(
                 "Could not parse event tag".to_string(),
             )),
@@ -1226,6 +1233,7 @@ pub enum EventTopic {
     BlockReward,
     AttesterSlashing,
     ProposerSlashing,
+    BlsToExecutionChange,
 }
 
 impl FromStr for EventTopic {
@@ -1249,6 +1257,7 @@ impl FromStr for EventTopic {
             "block_reward" => Ok(EventTopic::BlockReward),
             "attester_slashing" => Ok(EventTopic::AttesterSlashing),
             "proposer_slashing" => Ok(EventTopic::ProposerSlashing),
+            "bls_to_execution_change" => Ok(EventTopic::BlsToExecutionChange),
             _ => Err("event topic cannot be parsed.".to_string()),
         }
     }
@@ -1273,6 +1282,7 @@ impl fmt::Display for EventTopic {
             EventTopic::BlockReward => write!(f, "block_reward"),
             EventTopic::AttesterSlashing => write!(f, "attester_slashing"),
             EventTopic::ProposerSlashing => write!(f, "proposer_slashing"),
+            EventTopic::BlsToExecutionChange => write!(f, "bls_to_execution_change"),
         }
     }
 }