diff --git a/crates/eth2api/src/extensions.rs b/crates/eth2api/src/extensions.rs index 2dd90b12..393797bc 100644 --- a/crates/eth2api/src/extensions.rs +++ b/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 + )); + } } diff --git a/crates/testutil/src/beaconmock/mod.rs b/crates/testutil/src/beaconmock/mod.rs index aaea9e36..e4eb38ef 100644 --- a/crates/testutil/src/beaconmock/mod.rs +++ b/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; diff --git a/crates/testutil/src/beaconmock/proposal.rs b/crates/testutil/src/beaconmock/proposal.rs new file mode 100644 index 00000000..393b45ca --- /dev/null +++ b/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>, +} + +impl ProposalPreparationStore { + fn record(&self, preparations: impl IntoIterator) { + write_lock(&self.submissions).extend(preparations); + } + + /// Returns every preparation recorded so far, in submission order. + pub(crate) fn submissions(&self) -> Vec { + 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) { + 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, &'static str> { + let items: Vec = + 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()); + } +} diff --git a/crates/testutil/src/beaconmock/state.rs b/crates/testutil/src/beaconmock/state.rs index ed9de55f..33618ea1 100644 --- a/crates/testutil/src/beaconmock/state.rs +++ b/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>, pub(crate) deterministic_sync_comm_duties: RwLock>, 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 { + self.proposal_preparation_store.submissions() + } } pub(crate) fn hex_0x(bytes: impl AsRef<[u8]>) -> String {