feat(eth2api): add submit_proposal_preparations client method

crates/eth2api/src/extensions.rs

@@ -1,7 +1,9 @@
 use crate::{
     BeaconStateFork, ConsensusVersion, EthBeaconNodeApiClient, EventstreamRequestQueryTopic,
     GetForkScheduleRequest, GetForkScheduleResponse, GetGenesisRequest, GetGenesisResponse,
-    GetGenesisResponseResponseData, GetSpecRequest, GetSpecResponse, ValidatorStatus, spec::phase0,
+    GetGenesisResponseResponseData, GetSpecRequest, GetSpecResponse, PrepareBeaconProposerRequest,
+    PrepareBeaconProposerRequestBodyItem, PrepareBeaconProposerResponse, ValidatorStatus,
+    spec::{bellatrix, phase0},
 };
 use chrono::{DateTime, Utc};
 use eventsource_stream::Eventsource;
@@ -59,6 +61,19 @@ pub struct BeaconNodeEvent {
     pub data: String,
 }
 
+/// A single proposal preparation submitted to the beacon node
+/// (`prepare_beacon_proposer`), associating a validator index with the fee
+/// recipient the node should use when building blocks for it.
+///
+/// Mirrors go-eth2-client's `eth2v1.ProposalPreparation`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ProposalPreparation {
+    /// Index of the validator the preparation applies to.
+    pub validator_index: phase0::ValidatorIndex,
+    /// Execution-layer address that should receive block rewards.
+    pub fee_recipient: bellatrix::ExecutionAddress,
+}
+
 // Ordered oldest-to-newest. `resolve_fork_version` relies on this order to
 // break equal-epoch ties (the latest fork wins), so keep it chronological.
 const FORKS: [ConsensusVersion; 6] = [
@@ -483,6 +498,35 @@ impl EthBeaconNodeApiClient {
 
         Ok(stream)
     }
+
+    /// Submits proposal preparations to the beacon node
+    /// (`POST /eth/v1/validator/prepare_beacon_proposer`).
+    ///
+    /// Each preparation tells the beacon node which fee recipient to use when
+    /// it builds a block for the given validator. The information persists for
+    /// the epoch of submission plus the following two epochs, so callers resend
+    /// it periodically (e.g. once per epoch). Mirrors go-eth2-client's
+    /// `SubmitProposalPreparations`.
+    pub async fn submit_proposal_preparations(
+        &self,
+        preparations: &[ProposalPreparation],
+    ) -> Result<(), EthBeaconNodeApiClientError> {
+        let body = preparations
+            .iter()
+            .map(|preparation| PrepareBeaconProposerRequestBodyItem {
+                validator_index: preparation.validator_index.to_string(),
+                fee_recipient: format!("0x{}", hex::encode(preparation.fee_recipient)),
+            })
+            .collect();
+
+        match self
+            .prepare_beacon_proposer(PrepareBeaconProposerRequest { body })
+            .await?
+        {
+            PrepareBeaconProposerResponse::Ok => Ok(()),
+            _ => Err(EthBeaconNodeApiClientError::UnexpectedResponse),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -672,4 +716,81 @@ mod tests {
         assert_eq!(second.topic, "chain_reorg");
         assert_eq!(second.data, r#"{"slot":"20","depth":"2"}"#);
     }
+
+    #[tokio::test]
+    async fn submit_proposal_preparations_posts_expected_body() {
+        use wiremock::{
+            Mock, MockServer, ResponseTemplate,
+            matchers::{body_json, method, path},
+        };
+
+        let server = MockServer::start().await;
+
+        Mock::given(method("POST"))
+            .and(path("/eth/v1/validator/prepare_beacon_proposer"))
+            .and(body_json(json!([
+                {
+                    "validator_index": "1",
+                    "fee_recipient": "0x0101010101010101010101010101010101010101"
+                },
+                {
+                    "validator_index": "42",
+                    "fee_recipient": "0x2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a2a"
+                }
+            ])))
+            .respond_with(ResponseTemplate::new(200))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let client = EthBeaconNodeApiClient::with_base_url(server.uri()).expect("valid url");
+        client
+            .submit_proposal_preparations(&[
+                ProposalPreparation {
+                    validator_index: 1,
+                    fee_recipient: [0x01; 20],
+                },
+                ProposalPreparation {
+                    validator_index: 42,
+                    fee_recipient: [0x2a; 20],
+                },
+            ])
+            .await
+            .expect("submit succeeds");
+        // The mock's `.expect(1)` verifies on drop that the posted body
+        // matched.
+    }
+
+    #[tokio::test]
+    async fn submit_proposal_preparations_surfaces_error_status() {
+        use wiremock::{
+            Mock, MockServer, ResponseTemplate,
+            matchers::{method, path},
+        };
+
+        let server = MockServer::start().await;
+
+        Mock::given(method("POST"))
+            .and(path("/eth/v1/validator/prepare_beacon_proposer"))
+            .respond_with(ResponseTemplate::new(500).set_body_json(json!({
+                "code": 500,
+                "message": "internal error"
+            })))
+            .mount(&server)
+            .await;
+
+        let client = EthBeaconNodeApiClient::with_base_url(server.uri()).expect("valid url");
+        let error = client
+            .submit_proposal_preparations(&[ProposalPreparation {
+                validator_index: 1,
+                fee_recipient: [0x01; 20],
+            }])
+            .await
+            .expect_err("a 500 response must surface as an error");
+
+        assert!(matches!(
+            error,
+            EthBeaconNodeApiClientError::UnexpectedResponse
+        ));
+    }
 }

crates/testutil/src/beaconmock/mod.rs

@@ -8,6 +8,7 @@ mod defaults;
 mod fuzzer;
 mod headproducer;
 mod options;
+mod proposal;
 mod state;
 
 use std::{sync::Arc, time::Duration};
@@ -151,6 +152,7 @@ impl BeaconMock {
 
         mount_defaults(&server, Arc::clone(&state)).await;
         attestation::mount(&server, Arc::clone(&state)).await;
+        proposal::mount(&server, Arc::clone(&state)).await;
 
         let head_producer =
             HeadProducer::spawn(&server, effective_genesis_time, effective_slot_duration).await;

crates/testutil/src/beaconmock/proposal.rs

@@ -0,0 +1,178 @@
+//! Proposal-preparation recording endpoint used by `BeaconMock`.
+//!
+//! Charon's Go beaconmock swallows `prepare_beacon_proposer` submissions
+//! (testutil/beaconmock/server.go); Pluto instead records them so the
+//! app-level fee-recipient subscriber can be asserted against the mock in
+//! tests.
+
+use std::sync::{Arc, RwLock};
+
+use pluto_eth2api::ProposalPreparation;
+use serde::Deserialize;
+use wiremock::{
+    Mock, MockServer, Request, ResponseTemplate,
+    matchers::{method, path},
+};
+
+use super::{
+    defaults::error_response,
+    state::{MockState, read_lock, write_lock},
+};
+
+/// Priority for the recording route; lower value (higher priority) than the
+/// default 200 handler in `defaults.rs`, so submissions are captured instead of
+/// silently swallowed.
+const PROPOSAL_PREPARATION_PRIORITY: u8 = 100;
+
+/// Records proposal preparations submitted to
+/// `/eth/v1/validator/prepare_beacon_proposer`.
+#[derive(Debug, Default)]
+pub(crate) struct ProposalPreparationStore {
+    submissions: RwLock<Vec<ProposalPreparation>>,
+}
+
+impl ProposalPreparationStore {
+    fn record(&self, preparations: impl IntoIterator<Item = ProposalPreparation>) {
+        write_lock(&self.submissions).extend(preparations);
+    }
+
+    /// Returns every preparation recorded so far, in submission order.
+    pub(crate) fn submissions(&self) -> Vec<ProposalPreparation> {
+        read_lock(&self.submissions).clone()
+    }
+}
+
+/// Wire representation of a single `prepare_beacon_proposer` body item.
+#[derive(Debug, Deserialize)]
+struct ProposalPreparationItem {
+    validator_index: String,
+    fee_recipient: String,
+}
+
+/// Mounts the recording `prepare_beacon_proposer` handler on `server`.
+pub(crate) async fn mount(server: &MockServer, state: Arc<MockState>) {
+    Mock::given(method("POST"))
+        .and(path("/eth/v1/validator/prepare_beacon_proposer"))
+        .respond_with(move |request: &Request| response(&state, request))
+        .with_priority(PROPOSAL_PREPARATION_PRIORITY)
+        .mount(server)
+        .await;
+}
+
+fn response(state: &MockState, request: &Request) -> ResponseTemplate {
+    match parse_body(&request.body) {
+        Ok(preparations) => {
+            state.proposal_preparation_store.record(preparations);
+            ResponseTemplate::new(200)
+        }
+        Err(message) => error_response(400, message),
+    }
+}
+
+fn parse_body(body: &[u8]) -> Result<Vec<ProposalPreparation>, &'static str> {
+    let items: Vec<ProposalPreparationItem> =
+        serde_json::from_slice(body).map_err(|_| "invalid prepare_beacon_proposer body")?;
+
+    items
+        .into_iter()
+        .map(|item| {
+            let validator_index = item
+                .validator_index
+                .parse()
+                .map_err(|_| "invalid validator_index")?;
+            let fee_recipient = parse_execution_address(&item.fee_recipient)?;
+            Ok(ProposalPreparation {
+                validator_index,
+                fee_recipient,
+            })
+        })
+        .collect()
+}
+
+fn parse_execution_address(value: &str) -> Result<[u8; 20], &'static str> {
+    let stripped = value.strip_prefix("0x").unwrap_or(value);
+    let bytes = hex::decode(stripped).map_err(|_| "invalid fee_recipient hex")?;
+    bytes.try_into().map_err(|_| "invalid fee_recipient length")
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::beaconmock::BeaconMock;
+    use pluto_eth2api::ProposalPreparation;
+    use serde_json::json;
+
+    #[tokio::test]
+    async fn records_submitted_preparations() {
+        let mock = BeaconMock::builder()
+            .build()
+            .await
+            .expect("build beacon mock");
+
+        let preparations = vec![
+            ProposalPreparation {
+                validator_index: 1,
+                fee_recipient: [0x11; 20],
+            },
+            ProposalPreparation {
+                validator_index: 7,
+                fee_recipient: [0x22; 20],
+            },
+        ];
+
+        mock.client()
+            .submit_proposal_preparations(&preparations)
+            .await
+            .expect("submit succeeds");
+
+        assert_eq!(mock.state().proposal_preparations(), preparations);
+    }
+
+    #[tokio::test]
+    async fn accumulates_across_submissions() {
+        let mock = BeaconMock::builder()
+            .build()
+            .await
+            .expect("build beacon mock");
+
+        let first = ProposalPreparation {
+            validator_index: 1,
+            fee_recipient: [0x11; 20],
+        };
+        let second = ProposalPreparation {
+            validator_index: 2,
+            fee_recipient: [0x22; 20],
+        };
+
+        mock.client()
+            .submit_proposal_preparations(std::slice::from_ref(&first))
+            .await
+            .expect("first submit succeeds");
+        mock.client()
+            .submit_proposal_preparations(std::slice::from_ref(&second))
+            .await
+            .expect("second submit succeeds");
+
+        assert_eq!(mock.state().proposal_preparations(), vec![first, second]);
+    }
+
+    #[tokio::test]
+    async fn rejects_invalid_body() {
+        let mock = BeaconMock::builder()
+            .build()
+            .await
+            .expect("build beacon mock");
+
+        let response = reqwest::Client::new()
+            .post(format!(
+                "{}/eth/v1/validator/prepare_beacon_proposer",
+                mock.uri()
+            ))
+            .json(&json!([{ "validator_index": "abc", "fee_recipient": "0x00" }]))
+            .send()
+            .await
+            .expect("send");
+
+        assert_eq!(response.status(), 400);
+        assert!(mock.state().proposal_preparations().is_empty());
+    }
+}

crates/testutil/src/beaconmock/state.rs

@@ -6,12 +6,12 @@ use std::{
 };
 
 use pluto_eth2api::{
-    ValidatorResponseValidator, ValidatorStatus,
+    ProposalPreparation, ValidatorResponseValidator, ValidatorStatus,
     spec::phase0::{BLSPubKey, ValidatorIndex},
 };
 use serde_json::Value;
 
-use super::attestation::AttestationStore;
+use super::{attestation::AttestationStore, proposal::ProposalPreparationStore};
 
 pub(crate) const DEFAULT_WITHDRAWAL_CREDENTIALS: &str =
     "0x3132333435363738393031323334353637383930313233343536373839303132";
@@ -152,6 +152,7 @@ pub struct MockState {
     pub(crate) deterministic_proposer_duties: RwLock<Option<u64>>,
     pub(crate) deterministic_sync_comm_duties: RwLock<Option<(u64, u64)>>,
     pub(crate) attestation_store: AttestationStore,
+    pub(crate) proposal_preparation_store: ProposalPreparationStore,
 }
 
 impl MockState {
@@ -164,6 +165,7 @@ impl MockState {
             deterministic_proposer_duties: RwLock::new(None),
             deterministic_sync_comm_duties: RwLock::new(None),
             attestation_store: AttestationStore::default(),
+            proposal_preparation_store: ProposalPreparationStore::default(),
         }
     }
 
@@ -201,6 +203,13 @@ impl MockState {
     pub fn set_validator_set(&self, validator_set: ValidatorSet) {
         *write_lock(&self.validator_set) = validator_set;
     }
+
+    /// Returns every proposal preparation recorded by the
+    /// `prepare_beacon_proposer` endpoint, in submission order.
+    #[must_use]
+    pub fn proposal_preparations(&self) -> Vec<ProposalPreparation> {
+        self.proposal_preparation_store.submissions()
+    }
 }
 
 pub(crate) fn hex_0x(bytes: impl AsRef<[u8]>) -> String {