Add proposer preferences SSE event (#9308)

This is needed to connect to buildoor (Kurtosis package, acts as a trustless builder)


  


Co-Authored-By: Eitan Seri- Levi <eserilev@gmail.com>

Co-Authored-By: Eitan Seri-Levi <eserilev@ucsc.edu>

Co-Authored-By: Michael Sproul <michaelsproul@users.noreply.github.com>

beacon_node/beacon_chain/src/events.rs

@@ -29,6 +29,7 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     execution_payload_gossip_tx: Sender<EventKind<E>>,
     execution_payload_available_tx: Sender<EventKind<E>>,
     execution_payload_bid_tx: Sender<EventKind<E>>,
+    proposer_preferences_tx: Sender<EventKind<E>>,
     payload_attestation_message_tx: Sender<EventKind<E>>,
 }
 
@@ -60,6 +61,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         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 (proposer_preferences_tx, _) = broadcast::channel(capacity);
         let (payload_attestation_message_tx, _) = broadcast::channel(capacity);
 
         Self {
@@ -85,6 +87,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             execution_payload_gossip_tx,
             execution_payload_available_tx,
             execution_payload_bid_tx,
+            proposer_preferences_tx,
             payload_attestation_message_tx,
         }
     }
@@ -186,6 +189,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .execution_payload_bid_tx
                 .send(kind)
                 .map(|count| log_count("execution payload bid", count)),
+            EventKind::ProposerPreferences(_) => self
+                .proposer_preferences_tx
+                .send(kind)
+                .map(|count| log_count("proposer preferences", count)),
             EventKind::PayloadAttestationMessage(_) => self
                 .payload_attestation_message_tx
                 .send(kind)
@@ -284,6 +291,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.execution_payload_bid_tx.subscribe()
     }
 
+    pub fn subscribe_proposer_preferences(&self) -> Receiver<EventKind<E>> {
+        self.proposer_preferences_tx.subscribe()
+    }
+
     pub fn subscribe_payload_attestation_message(&self) -> Receiver<EventKind<E>> {
         self.payload_attestation_message_tx.subscribe()
     }
@@ -368,6 +379,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.execution_payload_bid_tx.receiver_count() > 0
     }
 
+    pub fn has_proposer_preferences_subscribers(&self) -> bool {
+        self.proposer_preferences_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/proposer_preferences_verification/gossip_verified_proposer_preferences.rs

@@ -6,6 +6,7 @@ use crate::{
         ProposerPreferencesError, proposer_preference_cache::GossipVerifiedProposerPreferenceCache,
     },
 };
+use eth2::types::{EventKind, ForkVersionedResponse};
 use slot_clock::SlotClock;
 use state_processing::signature_sets::{get_pubkey_from_state, proposer_preferences_signature_set};
 use tracing::debug;
@@ -145,6 +146,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     %validator_index,
                     "Successfully verified gossip proposer preferences"
                 );
+
+                if let Some(event_handler) = self.event_handler.as_ref()
+                    && event_handler.has_proposer_preferences_subscribers()
+                {
+                    event_handler.register(EventKind::ProposerPreferences(Box::new(
+                        ForkVersionedResponse {
+                            version: self.spec.fork_name_at_slot::<T::EthSpec>(proposal_slot),
+                            metadata: Default::default(),
+                            data: (*verified.signed_preferences).clone(),
+                        },
+                    )));
+                }
+
                 Ok(verified)
             }
             Err(e) => {

beacon_node/beacon_chain/tests/events.rs

@@ -7,9 +7,9 @@ use eth2::types::{EventKind, SseBlobSidecar, SseDataColumnSidecar};
 use std::sync::Arc;
 use types::data::FixedBlobSidecarList;
 use types::{
-    BlobSidecar, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas, Domain, EthSpec,
-    MinimalEthSpec, PayloadAttestationData, PayloadAttestationMessage, SignedExecutionPayloadBid,
-    SignedRoot, Slot,
+    Address, BlobSidecar, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas, Domain,
+    EthSpec, Hash256, MinimalEthSpec, PayloadAttestationData, PayloadAttestationMessage,
+    ProposerPreferences, SignedExecutionPayloadBid, SignedProposerPreferences, SignedRoot, Slot,
 };
 
 type E = MinimalEthSpec;
@@ -401,3 +401,80 @@ async fn payload_attestation_message_event_on_gossip_verification() {
         panic!("Expected PayloadAttestationMessage event, got {:?}", event);
     }
 }
+
+/// Verifies that a `proposer_preferences` SSE event is emitted when signed proposer preferences
+/// pass gossip verification.
+#[tokio::test]
+async fn proposer_preferences_event_on_gossip_verification() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+
+    let harness = BeaconChainHarness::builder(E::default())
+        .default_spec()
+        .deterministic_keypairs(64)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    let head = harness.chain.canonical_head.cached_head();
+    let head_state = &head.snapshot.beacon_state;
+    let genesis_validators_root = harness.chain.genesis_validators_root;
+
+    // Pick a proposal slot in the next epoch so it is always a valid, future slot. The lookahead
+    // covers 2 epochs: index = epoch_offset * slots_per_epoch + slot_in_epoch.
+    let slots_per_epoch = E::slots_per_epoch() as usize;
+    let proposer_lookahead = head_state
+        .proposer_lookahead()
+        .expect("gloas state should have proposer lookahead");
+    let next_epoch_start = (head_state.current_epoch() + 1).start_slot(E::slots_per_epoch());
+    let proposal_slot = next_epoch_start + 1;
+    let lookahead_index = slots_per_epoch + 1;
+    let validator_index = *proposer_lookahead
+        .get(lookahead_index)
+        .expect("lookahead index should be in range");
+
+    // Build and sign proposer preferences for the proposer of `proposal_slot`.
+    let preferences = ProposerPreferences {
+        dependent_root: Hash256::ZERO,
+        proposal_slot,
+        validator_index,
+        fee_recipient: Address::repeat_byte(0xaa),
+        target_gas_limit: 30_000_000,
+    };
+    let domain = harness.spec.get_domain(
+        proposal_slot.epoch(E::slots_per_epoch()),
+        Domain::ProposerPreferences,
+        &head_state.fork(),
+        genesis_validators_root,
+    );
+    let signature = harness.validator_keypairs[validator_index as usize]
+        .sk
+        .sign(preferences.signing_root(domain));
+    let signed = SignedProposerPreferences {
+        message: preferences.clone(),
+        signature,
+    };
+
+    // Subscribe before verification.
+    let event_handler = harness.chain.event_handler.as_ref().unwrap();
+    let mut receiver = event_handler.subscribe_proposer_preferences();
+
+    // Verify the preferences through the gossip path.
+    harness
+        .chain
+        .verify_proposer_preferences_for_gossip(Arc::new(signed))
+        .expect("verification should succeed");
+
+    // Assert the event was emitted with the expected data.
+    let event = receiver.try_recv().expect("should receive event");
+    if let EventKind::ProposerPreferences(versioned) = event {
+        assert_eq!(versioned.data.message, preferences);
+        assert_eq!(
+            versioned.version,
+            harness.spec.fork_name_at_slot::<E>(proposal_slot)
+        );
+    } else {
+        panic!("Expected ProposerPreferences event, got {:?}", event);
+    }
+}