Implement gloas proposer preference vc duty (#9208)

Allow for the vc to submit its proposer preferences to the network


  


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

Co-Authored-By: Jimmy Chen <jchen.tc@gmail.com>

beacon_node/http_api/src/lib.rs

@@ -1490,7 +1490,7 @@ pub fn serve<T: BeaconChainTypes>(
     // POST beacon/pool/payload_attestations
     let post_beacon_pool_payload_attestations = post_beacon_pool_payload_attestations(
         &network_tx_filter,
-        optional_consensus_version_header_filter,
+        optional_consensus_version_header_filter.clone(),
         &beacon_pool_path,
     );
 
@@ -1510,6 +1510,22 @@ pub fn serve<T: BeaconChainTypes>(
     let post_beacon_pool_bls_to_execution_changes =
         post_beacon_pool_bls_to_execution_changes(&network_tx_filter, &beacon_pool_path);
 
+    // POST validator/proposer_preferences (JSON)
+    let post_validator_proposer_preferences = post_validator_proposer_preferences(
+        eth_v1.clone(),
+        task_spawner_filter.clone(),
+        chain_filter.clone(),
+        network_tx_filter.clone(),
+    );
+
+    // POST validator/proposer_preferences (SSZ)
+    let post_validator_proposer_preferences_ssz = post_validator_proposer_preferences_ssz(
+        eth_v1.clone(),
+        task_spawner_filter.clone(),
+        chain_filter.clone(),
+        network_tx_filter.clone(),
+    );
+
     // POST beacon/execution_payload_envelope
     let post_beacon_execution_payload_envelope = post_beacon_execution_payload_envelope(
         eth_v1.clone(),
@@ -3416,7 +3432,8 @@ pub fn serve<T: BeaconChainTypes>(
                             .uor(post_beacon_blinded_blocks_ssz)
                             .uor(post_beacon_blinded_blocks_v2_ssz)
                             .uor(post_beacon_execution_payload_envelope_ssz)
-                            .uor(post_beacon_pool_payload_attestations_ssz),
+                            .uor(post_beacon_pool_payload_attestations_ssz)
+                            .uor(post_validator_proposer_preferences_ssz),
                     )
                     .uor(post_beacon_blocks)
                     .uor(post_beacon_blinded_blocks)
@@ -3429,6 +3446,7 @@ pub fn serve<T: BeaconChainTypes>(
                     .uor(post_beacon_pool_sync_committees)
                     .uor(post_beacon_pool_payload_attestations)
                     .uor(post_beacon_pool_bls_to_execution_changes)
+                    .uor(post_validator_proposer_preferences)
                     .uor(post_beacon_execution_payload_envelope)
                     .uor(post_beacon_state_validators)
                     .uor(post_beacon_state_validator_balances)

beacon_node/http_api/src/validator/mod.rs

@@ -9,8 +9,11 @@ use crate::utils::{
 use crate::version::{V1, V2, V3, unsupported_version_rejection};
 use crate::{StateId, attester_duties, proposer_duties, ptc_duties, sync_committees};
 use beacon_chain::attestation_verification::VerifiedAttestation;
+use beacon_chain::proposer_preferences_verification::ProposerPreferencesError;
 use beacon_chain::{AttestationError, BeaconChain, BeaconChainError, BeaconChainTypes};
 use bls::PublicKeyBytes;
+use bytes::Bytes;
+use eth2::CONSENSUS_VERSION_HEADER;
 use eth2::types::{
     Accept, BeaconCommitteeSubscription, EndpointVersion, Failure, GenericResponse,
     StandardLivenessResponseData, StateId as CoreStateId, ValidatorAggregateAttestationQuery,
@@ -20,14 +23,15 @@ use lighthouse_network::PubsubMessage;
 use network::{NetworkMessage, ValidatorSubscriptionMessage};
 use reqwest::StatusCode;
 use slot_clock::SlotClock;
+use ssz::Decode;
 use std::sync::Arc;
 use tokio::sync::mpsc::{Sender, UnboundedSender};
 use tokio::sync::oneshot;
 use tracing::{debug, error, info, warn};
 use types::{
-    BeaconState, Epoch, EthSpec, ProposerPreparationData, SignedAggregateAndProof,
-    SignedContributionAndProof, SignedValidatorRegistrationData, Slot, SyncContributionData,
-    ValidatorSubscription,
+    BeaconState, Epoch, EthSpec, ForkName, ProposerPreparationData, SignedAggregateAndProof,
+    SignedContributionAndProof, SignedProposerPreferences, SignedValidatorRegistrationData, Slot,
+    SyncContributionData, ValidatorSubscription,
 };
 use warp::{Filter, Rejection, Reply};
 use warp_utils::reject::convert_rejection;
@@ -1144,3 +1148,117 @@ pub fn get_validator_duties_proposer<T: BeaconChainTypes>(
         )
         .boxed()
 }
+
+/// POST validator/proposer_preferences (JSON)
+pub fn post_validator_proposer_preferences<T: BeaconChainTypes>(
+    eth_v1: EthV1Filter,
+    task_spawner_filter: TaskSpawnerFilter<T>,
+    chain_filter: ChainFilter<T>,
+    network_tx_filter: NetworkTxFilter<T>,
+) -> ResponseFilter {
+    eth_v1
+        .and(warp::path("validator"))
+        .and(warp::path("proposer_preferences"))
+        .and(warp::path::end())
+        .and(warp_utils::json::json())
+        .and(warp::header::<ForkName>(CONSENSUS_VERSION_HEADER))
+        .and(task_spawner_filter)
+        .and(chain_filter)
+        .and(network_tx_filter)
+        .then(
+            |preferences: Vec<SignedProposerPreferences>,
+             _fork_name: ForkName,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             chain: Arc<BeaconChain<T>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
+                task_spawner.blocking_response_task(Priority::P0, move || {
+                    publish_proposer_preferences(&chain, &network_tx, preferences)?;
+                    Ok(warp::reply())
+                })
+            },
+        )
+        .boxed()
+}
+
+/// POST validator/proposer_preferences (SSZ)
+pub fn post_validator_proposer_preferences_ssz<T: BeaconChainTypes>(
+    eth_v1: EthV1Filter,
+    task_spawner_filter: TaskSpawnerFilter<T>,
+    chain_filter: ChainFilter<T>,
+    network_tx_filter: NetworkTxFilter<T>,
+) -> ResponseFilter {
+    eth_v1
+        .and(warp::path("validator"))
+        .and(warp::path("proposer_preferences"))
+        .and(warp::path::end())
+        .and(warp::body::bytes())
+        .and(warp::header::<ForkName>(CONSENSUS_VERSION_HEADER))
+        .and(task_spawner_filter)
+        .and(chain_filter)
+        .and(network_tx_filter)
+        .then(
+            |body_bytes: Bytes,
+             _fork_name: ForkName,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             chain: Arc<BeaconChain<T>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
+                task_spawner.blocking_response_task(Priority::P0, move || {
+                    let preferences = Vec::<SignedProposerPreferences>::from_ssz_bytes(&body_bytes)
+                        .map_err(|e| {
+                            warp_utils::reject::custom_bad_request(format!("invalid SSZ: {e:?}"))
+                        })?;
+                    publish_proposer_preferences(&chain, &network_tx, preferences)?;
+                    Ok(warp::reply())
+                })
+            },
+        )
+        .boxed()
+}
+
+fn publish_proposer_preferences<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
+    preferences_list: Vec<SignedProposerPreferences>,
+) -> Result<(), warp::Rejection> {
+    let mut failures = vec![];
+    let mut num_already_known = 0;
+
+    for (index, preferences) in preferences_list.into_iter().enumerate() {
+        let validator_index = preferences.message.validator_index;
+        match chain.verify_proposer_preferences_for_gossip(Arc::new(preferences)) {
+            Ok(verified) => {
+                crate::utils::publish_pubsub_message(
+                    network_tx,
+                    PubsubMessage::ProposerPreferences(verified.signed_preferences),
+                )?;
+            }
+            Err(ProposerPreferencesError::AlreadySeen { .. }) => {
+                num_already_known += 1;
+            }
+            Err(e) => {
+                error!(
+                    error = ?e,
+                    %validator_index,
+                    "Failure verifying proposer preferences for gossip"
+                );
+                failures.push(Failure::new(index, format!("{e:?}")));
+            }
+        }
+    }
+
+    if num_already_known > 0 {
+        debug!(
+            count = num_already_known,
+            "Some proposer preferences already known"
+        );
+    }
+
+    if failures.is_empty() {
+        Ok(())
+    } else {
+        Err(warp_utils::reject::indexed_bad_request(
+            "error processing proposer preferences".to_string(),
+            failures,
+        ))
+    }
+}

beacon_node/http_api/tests/tests.rs

@@ -48,9 +48,10 @@ use tokio::time::Duration;
 use tree_hash::TreeHash;
 use types::ApplicationDomain;
 use types::{
-    Domain, EthSpec, ExecutionBlockHash, Hash256, MainnetEthSpec, RelativeEpoch, SelectionProof,
-    SignedExecutionPayloadEnvelope, SignedRoot, SingleAttestation, Slot,
-    attestation::AttestationBase, consts::gloas::BUILDER_INDEX_SELF_BUILD,
+    Address, Domain, EthSpec, ExecutionBlockHash, Hash256, MainnetEthSpec, ProposerPreferences,
+    RelativeEpoch, SelectionProof, SignedExecutionPayloadEnvelope, SignedProposerPreferences,
+    SignedRoot, SingleAttestation, Slot, attestation::AttestationBase,
+    consts::gloas::BUILDER_INDEX_SELF_BUILD,
 };
 
 type E = MainnetEthSpec;
@@ -2898,6 +2899,162 @@ impl ApiTester {
         self
     }
 
+    fn make_valid_signed_proposer_preferences(
+        &self,
+        slot_offset: usize,
+    ) -> SignedProposerPreferences {
+        let head = self.chain.head_snapshot();
+        let head_slot = head.beacon_block.slot();
+        let head_state = &head.beacon_state;
+        let genesis_validators_root = self.chain.genesis_validators_root;
+
+        let proposer_lookahead = head_state
+            .proposer_lookahead()
+            .expect("should get proposer_lookahead");
+
+        // Pick a future slot in the next epoch to ensure it's always valid.
+        // 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 next_epoch = head_slot.epoch(E::slots_per_epoch()) + 1;
+        let next_epoch_start = next_epoch.start_slot(E::slots_per_epoch());
+        let proposal_slot = next_epoch_start + Slot::new((slot_offset % slots_per_epoch) as u64);
+
+        let lookahead_index = slots_per_epoch + (slot_offset % slots_per_epoch);
+        let validator_index = *proposer_lookahead
+            .get(lookahead_index)
+            .expect("slot index should be in lookahead") as usize;
+
+        let preferences = ProposerPreferences {
+            dependent_root: Hash256::ZERO,
+            proposal_slot,
+            validator_index: validator_index as u64,
+            fee_recipient: Address::repeat_byte(0xaa),
+            gas_limit: 30_000_000,
+        };
+
+        let epoch = proposal_slot.epoch(E::slots_per_epoch());
+        let fork = head_state.fork();
+        let domain = self.chain.spec.get_domain(
+            epoch,
+            Domain::ProposerPreferences,
+            &fork,
+            genesis_validators_root,
+        );
+        let signing_root = preferences.signing_root(domain);
+        let sk = &self.validator_keypairs()[validator_index].sk;
+        let signature = sk.sign(signing_root);
+
+        SignedProposerPreferences {
+            message: preferences,
+            signature,
+        }
+    }
+
+    // Each sub-test uses a unique slot_offset (1-5) because the gossip cache deduplicates on
+    // (slot, dependent_root, validator_index). Reusing an offset from an earlier test would hit
+    // "already seen" instead of testing the intended condition.
+    pub async fn test_post_validator_proposer_preferences_valid(mut self) -> Self {
+        let signed = self.make_valid_signed_proposer_preferences(1);
+        let fork_name = self
+            .chain
+            .spec
+            .fork_name_at_slot::<E>(signed.message.proposal_slot);
+
+        self.client
+            .post_validator_proposer_preferences(&[signed], fork_name)
+            .await
+            .unwrap();
+
+        assert!(
+            self.network_rx.network_recv.recv().await.is_some(),
+            "valid proposer preferences should be sent to network"
+        );
+
+        self
+    }
+
+    pub async fn test_post_validator_proposer_preferences_valid_ssz(mut self) -> Self {
+        let signed = self.make_valid_signed_proposer_preferences(2);
+        let fork_name = self
+            .chain
+            .spec
+            .fork_name_at_slot::<E>(signed.message.proposal_slot);
+
+        self.client
+            .post_validator_proposer_preferences_ssz(&vec![signed], fork_name)
+            .await
+            .unwrap();
+
+        assert!(
+            self.network_rx.network_recv.recv().await.is_some(),
+            "valid proposer preferences (SSZ) should be sent to network"
+        );
+
+        self
+    }
+
+    pub async fn test_post_validator_proposer_preferences_invalid_sig(self) -> Self {
+        let mut signed = self.make_valid_signed_proposer_preferences(3);
+        signed.signature = Signature::empty();
+        let fork_name = self
+            .chain
+            .spec
+            .fork_name_at_slot::<E>(signed.message.proposal_slot);
+
+        let result = self
+            .client
+            .post_validator_proposer_preferences(&[signed], fork_name)
+            .await;
+
+        assert!(result.is_err(), "invalid signature should be rejected");
+
+        self
+    }
+
+    pub async fn test_post_validator_proposer_preferences_invalid_sig_ssz(self) -> Self {
+        let mut signed = self.make_valid_signed_proposer_preferences(4);
+        signed.signature = Signature::empty();
+        let fork_name = self
+            .chain
+            .spec
+            .fork_name_at_slot::<E>(signed.message.proposal_slot);
+
+        let result = self
+            .client
+            .post_validator_proposer_preferences_ssz(&vec![signed], fork_name)
+            .await;
+
+        assert!(
+            result.is_err(),
+            "invalid signature should be rejected via SSZ route"
+        );
+
+        self
+    }
+
+    pub async fn test_post_validator_proposer_preferences_duplicate(mut self) -> Self {
+        let signed = self.make_valid_signed_proposer_preferences(5);
+        let fork_name = self
+            .chain
+            .spec
+            .fork_name_at_slot::<E>(signed.message.proposal_slot);
+
+        // First submission should succeed.
+        self.client
+            .post_validator_proposer_preferences(std::slice::from_ref(&signed), fork_name)
+            .await
+            .unwrap();
+        self.network_rx.network_recv.recv().await;
+
+        // Second submission of the same preferences should return 200 (already known, not an error).
+        self.client
+            .post_validator_proposer_preferences(&[signed], fork_name)
+            .await
+            .unwrap();
+
+        self
+    }
+
     pub async fn test_get_config_fork_schedule(self) -> Self {
         let result = self.client.get_config_fork_schedule().await.unwrap().data;
 
@@ -9199,3 +9356,22 @@ async fn get_validator_blocks_v3_http_api_path() {
         .get_validator_blocks_v3_path_graffiti_policy()
         .await;
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn post_validator_proposer_preferences() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+    ApiTester::new_with_hard_forks()
+        .await
+        .test_post_validator_proposer_preferences_valid()
+        .await
+        .test_post_validator_proposer_preferences_valid_ssz()
+        .await
+        .test_post_validator_proposer_preferences_invalid_sig()
+        .await
+        .test_post_validator_proposer_preferences_invalid_sig_ssz()
+        .await
+        .test_post_validator_proposer_preferences_duplicate()
+        .await;
+}