diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index 0b11e3ad..e8ddab3b 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -21,7 +21,7 @@ use libp2p::{relay, swarm::NetworkBehaviour}; use pluto_consensus::qbft; use pluto_core::{gater::DutyGaterFn, types::PubKey}; use pluto_crypto::types::PublicKey; -use pluto_eth2api::EthBeaconNodeApiClient; +use pluto_eth2api::BeaconNodeClient; use pluto_p2p::{ bootnode, force_direct::ForceDirectBehaviour, @@ -80,7 +80,7 @@ pub(crate) async fn wire_p2p( peers: Vec, consensus: Arc, duty_gater: DutyGaterFn, - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, pub_shares_by_key: HashMap>, lock_hash: Vec, builder_enabled: bool, diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 3bdb29ce..bc8a30f4 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -313,6 +313,11 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { }; let eth2_cl = build_api_client(&beacon_node_addr, config.beacon_node_timeout)?; + let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone()); + // Broadcasting uses a separate client with the (distinct) submit timeout. + let submission_api = build_api_client(&beacon_node_addr, config.beacon_node_submit_timeout)?; + let submission_client = pluto_eth2api::BeaconNodeClient::new(submission_api); + // Fail fast if the beacon node is on a different network than the cluster // lock (Charon's `configureEth2Client`, app.go:1022-1053). Both eth2 // clients here target the same endpoint, so a single check suffices — @@ -322,18 +327,19 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // fork version (`build_simnet_beacon_mock`), so a network mismatch is // impossible by construction. if simnet_beacon_mock.is_none() { - verify_fork_schedule(ð2_cl, &lock.fork_version).await?; + verify_fork_schedule(&beacon_client, &lock.fork_version).await?; } - let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone()); - // Broadcasting uses a separate client with the (distinct) submit timeout. - let submission_api = build_api_client(&beacon_node_addr, config.beacon_node_submit_timeout)?; - let submission_client = pluto_eth2api::BeaconNodeClient::new(submission_api); + // Warm both config caches before duty scheduling so duty-path config + // reads start served from cache (refetched at most once per TTL); + // failure aborts startup. The network check above already filled + // `beacon_client`'s schedule slot. + tokio::try_join!(beacon_client.warm(), submission_client.warm())?; // ---- Beacon-derived duty-workflow inputs ---- // Duty admission gate: validates duties against the beacon chain. - let duty_gater: DutyGaterFn = pluto_core::gater::DutyGater::new(ð2_cl) + let duty_gater: DutyGaterFn = pluto_core::gater::DutyGater::new(&beacon_client) .await .map_err(AppError::Gater)? .into_fn(); @@ -341,7 +347,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // Per-component deadline calculator, shared as an `Arc` so a single // beacon-derived instance backs every component's deadliner. let deadline_calc: Arc = Arc::new( - pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) + pluto_core::deadline::DutyDeadlineCalculator::from_client(&beacon_client) .await .map_err(AppError::Deadline)?, ); @@ -361,8 +367,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // Use the mock's echoed slot timing, not the configured value: fuzz mode // overrides it with a 12s default, and the validator mock's ticker must // match the mock. `slots_per_epoch` also feeds the Electra activation slot. - let (fetched_slot_duration, slots_per_epoch) = eth2_cl.fetch_slots_config().await?; - let fork_config = eth2_cl.fetch_fork_config().await?; + let (fetched_slot_duration, slots_per_epoch) = beacon_client.slots_config().await?; + let fork_config = beacon_client.fork_config().await?; let electra_slot = fork_config .get(&pluto_eth2api::ConsensusVersion::Electra) .map(|schedule| schedule.epoch) @@ -429,7 +435,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { peers, Arc::clone(&consensus), Arc::clone(&duty_gater), - eth2_cl.clone(), + beacon_client.clone(), pub_shares_by_key, lock.lock_hash.clone(), config.builder_api, @@ -449,10 +455,9 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // Aggregated-signature verifier: verifies the reconstructed group signature // against the beacon-node signing domain. - let sigagg_verifier = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone())); + let sigagg_verifier = pluto_core::sigagg::new_verifier(beacon_client.clone()); - // The readiness checker uses its own beacon-client clone, taken before - // `eth2_cl` is moved into the workflow inputs below. + // The readiness checker uses its own beacon-client clone. let monitoring_beacon = eth2_cl.clone(); // Readiness observes which DV root pubkeys the validator client references @@ -496,7 +501,6 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { threshold, share_idx, beacon_client, - eth2_cl, submission_client, validators, consensus: Arc::clone(&consensus), @@ -923,14 +927,19 @@ fn parse_execution_address(s: &str) -> Option<[u8; 20]> { /// Fails fast when the beacon node is on a different network than the cluster /// lock. /// -/// Mirrors Charon's `configureEth2Client` fork-schedule guard. +/// Mirrors Charon's `configureEth2Client` fork-schedule guard. Reads the +/// schedule through the client's config cache, so the startup warm-up reuses +/// this fetch. async fn verify_fork_schedule( - eth2_cl: &pluto_eth2api::EthBeaconNodeApiClient, + client: &pluto_eth2api::BeaconNodeClient, lock_fork_version: &[u8], ) -> Result<(), AppError> { - let versions = eth2_cl.fetch_fork_schedule_versions().await?; + let schedule = client.fork_schedule().await?; - if versions.iter().any(|v| v.as_slice() == lock_fork_version) { + if schedule + .iter() + .any(|fork| fork.version.as_slice() == lock_fork_version) + { return Ok(()); } @@ -938,9 +947,11 @@ async fn verify_fork_schedule( Err(AppError::ForkScheduleMismatch { lock_network: network_name_or_hex(lock_fork_version), lock_fork_version: format!("0x{}", hex::encode(lock_fork_version)), - beacon_node_network: versions + // Defensive: `fork_schedule()` rejects empty schedules, so `first()` + // cannot be `None` here. + beacon_node_network: schedule .first() - .map(|v| network_name_or_hex(v)) + .map(|fork| network_name_or_hex(&fork.version)) .unwrap_or_else(|| "unknown".to_string()), }) } @@ -1119,7 +1130,7 @@ async fn build_simnet_validator_mock( Ok(Arc::new( ValidatorMock::builder() - .eth2_cl(vapi_client) + .eth2_cl(pluto_eth2api::BeaconNodeClient::new(vapi_client)) .sign_func(signer) .pubkeys(pubshares) .meta(meta) @@ -1300,7 +1311,7 @@ mod tests { .await .expect("beacon mock"); - verify_fork_schedule(mock.client(), &HOLESKY_FORK_VERSION) + verify_fork_schedule(&mock.beacon_client(), &HOLESKY_FORK_VERSION) .await .expect("matching fork version should pass the startup guard"); } @@ -1315,7 +1326,7 @@ mod tests { .await .expect("beacon mock"); - let err = verify_fork_schedule(mock.client(), &MAINNET_FORK_VERSION) + let err = verify_fork_schedule(&mock.beacon_client(), &MAINNET_FORK_VERSION) .await .expect_err("mismatched fork version should fail the startup guard"); diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 697b2897..f4662a19 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -38,7 +38,7 @@ use pluto_core::{ validatorapi::{self, Component, Handler, SeenPubkeysFn}, }; use pluto_eth2api::{ - BeaconNodeClient, EthBeaconNodeApiClient, + BeaconNodeClient, spec::{bellatrix::ExecutionAddress, phase0::BLSPubKey}, valcache::{ValidatorCache, ValidatorCacheError}, }; @@ -109,10 +109,9 @@ pub struct WireInputs { pub threshold: u64, /// This node's 1-indexed share index. pub share_idx: u64, - /// Beacon node client used for scheduling. + /// Beacon node client used for scheduling, fetching, validatorapi, and + /// signing-domain resolution. pub beacon_client: BeaconNodeClient, - /// Beacon node API client used for fetching / dutydb / validatorapi. - pub eth2_cl: EthBeaconNodeApiClient, /// Submission beacon node client used for broadcasting. pub submission_client: BeaconNodeClient, /// Per-validator data for this node. @@ -267,7 +266,6 @@ pub async fn wire_core_workflow( threshold, share_idx, beacon_client, - eth2_cl, submission_client, validators, consensus, @@ -301,7 +299,7 @@ pub async fn wire_core_workflow( // would resolve duties against an empty (or unfiltered) set. `ValidatorCache` // clones share state, so the per-epoch trim + refresh subscriber registered // below refreshes every consumer at once. - let validator_cache = ValidatorCache::new(eth2_cl.clone(), eth2_pubkeys); + let validator_cache = ValidatorCache::new(beacon_client.api().clone(), eth2_pubkeys); tokio::join!( beacon_client.set_validator_cache(validator_cache.clone()), submission_client.set_validator_cache(validator_cache.clone()), @@ -386,7 +384,7 @@ pub async fn wire_core_workflow( let fetcher = Arc::new( Fetcher::builder() - .eth2_cl(eth2_cl.clone()) + .eth2_cl(beacon_client.clone()) .fee_recipient(Arc::clone(&fee_recipient_fn)) .agg_sig_db(agg_sig_db_fn) .await_att_data(await_att_data_fn) @@ -623,7 +621,7 @@ pub async fn wire_core_workflow( } let (scheduler, scheduler_task) = sched_builder - .build(beacon_client, ct.clone()) + .build(beacon_client.clone(), ct.clone()) .await .map_err(AppError::Scheduler)?; @@ -635,7 +633,7 @@ pub async fn wire_core_workflow( // agg-attestation / sync-contribution / pubkey-by-attestation lookups, and // the scheduler-backed duty-definition lookup. let mut vapi = Component::new( - Arc::new(eth2_cl.clone()), + beacon_client.clone(), Arc::clone(&dutydb), share_idx, pub_share_by_pubkey, @@ -899,7 +897,7 @@ mod tests { use super::*; use pluto_core::types::SlotNumber; use pluto_eth2api::{ - BlindedBlock400Response, GetStateValidatorsResponseResponse, + BlindedBlock400Response, EthBeaconNodeApiClient, GetStateValidatorsResponseResponse, GetStateValidatorsResponseResponseDatum, ValidatorResponseValidator, ValidatorStatus, }; use wiremock::{ diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 9580aa8b..83ab3290 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -253,7 +253,6 @@ fn wire_inputs_with( threshold, share_idx: 1, beacon_client, - eth2_cl, submission_client, validators, consensus, @@ -720,7 +719,8 @@ async fn wiring_rejects_bad_partial_signature() { // REAL eth2 verifier (mirrors production `run`): BeaconMock serves the // signing domain via `/eth/v1/config/spec` + `/eth/v1/beacon/genesis`. - let verifier: VerifyFn = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone())); + let verifier: VerifyFn = + pluto_core::sigagg::new_verifier(BeaconNodeClient::new(eth2_cl.clone())); const THRESHOLD: u64 = 2; let inputs = wire_inputs_with( diff --git a/crates/core/src/bcast/mod.rs b/crates/core/src/bcast/mod.rs index 7cec31e9..8cbe221d 100644 --- a/crates/core/src/bcast/mod.rs +++ b/crates/core/src/bcast/mod.rs @@ -8,8 +8,8 @@ use std::{any::Any, error::Error as StdError}; use chrono::{DateTime, Duration, Utc}; use pluto_crypto::{blst_impl::BlstImpl, tbls::Tbls}; use pluto_eth2api::{ - AttesterDuty, BeaconNodeClient, EthBeaconNodeApiClient, - GetStateValidatorsResponseResponseDatum, ValidatorStatus, data_version_is_before_electra, + AttesterDuty, BeaconNodeClient, GetStateValidatorsResponseResponseDatum, ValidatorStatus, + data_version_is_before_electra, spec::{altair, phase0}, versioned, }; @@ -229,24 +229,20 @@ pub struct Broadcaster { impl Broadcaster { /// Creates a new broadcaster. pub async fn new(client: BeaconNodeClient) -> Result { - let genesis_time = - client - .api() - .fetch_genesis_time() - .await - .map_err(|source| Error::Client { - context: "fetch genesis time", - source: Box::new(source), - })?; - let (slot_duration, _) = - client - .api() - .fetch_slots_config() - .await - .map_err(|source| Error::Client { - context: "fetch slots config", - source: Box::new(source), - })?; + let genesis_time = client + .genesis_time() + .await + .map_err(|source| Error::Client { + context: "fetch genesis time", + source: Box::new(source), + })?; + let (slot_duration, _) = client + .slots_config() + .await + .map_err(|source| Error::Client { + context: "fetch slots config", + source: Box::new(source), + })?; let slot_duration = Duration::from_std(slot_duration).map_err(|_| Error::ArithmeticOverflow { context: "slot duration", @@ -277,7 +273,7 @@ impl Broadcaster { // Use first slot in current epoch for accurate delay calculations while // submitting builder registrations. This is because builder // registrations are submitted in first slot of every epoch. - duty.slot = first_slot_in_current_epoch(self.client.api()).await?; + duty.slot = first_slot_in_current_epoch(&self.client).await?; self.broadcast_builder_registration(&duty, &set).await?; } DutyType::Exit => self.broadcast_exits(&duty, &set).await?, @@ -526,15 +522,16 @@ impl Broadcaster { context: "fetch attester duties", source: Box::new(source), })?; - let domain = self - .client - .api() - .fetch_beacon_attester_domain(epoch) - .await - .map_err(|source| Error::Client { - context: "fetch beacon attester domain", - source: Box::new(source), - })?; + let domain = pluto_eth2util::signing::get_domain( + &self.client, + pluto_eth2util::signing::DomainName::BeaconAttester, + epoch, + ) + .await + .map_err(|source| Error::Client { + context: "fetch beacon attester domain", + source: Box::new(source), + })?; // Try to find the matching attester duty and attestation by verifying the full // aggregated signature of the attestation with the pubkey found in the attester @@ -717,10 +714,10 @@ fn attestation_matches_duty( } async fn first_slot_in_current_epoch( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, ) -> Result { let genesis_time = client - .fetch_genesis_time() + .genesis_time() .await .map_err(|source| Error::Client { context: "fetch genesis time", @@ -728,7 +725,7 @@ async fn first_slot_in_current_epoch( })?; let (slot_duration, slots_per_epoch) = client - .fetch_slots_config() + .slots_config() .await .map_err(|source| Error::Client { context: "fetch slots config", @@ -1231,11 +1228,13 @@ mod tests { .build() .await .expect("beacon mock"); - let domain = beacon - .client() - .fetch_beacon_attester_domain(3) - .await - .expect("domain"); + let domain = pluto_eth2util::signing::get_domain( + &beacon.beacon_client(), + pluto_eth2util::signing::DomainName::BeaconAttester, + 3, + ) + .await + .expect("domain"); let client = cached_client( &beacon, vec![validator_datum( @@ -1271,14 +1270,6 @@ mod tests { pubkey: public_key, }] ); - assert_eq!( - beacon - .client() - .fetch_beacon_attester_domain(3) - .await - .expect("domain"), - domain - ); let broadcaster = Broadcaster::new(client).await.expect("broadcaster"); let mut attestations = vec![attestation]; diff --git a/crates/core/src/deadline/calculator.rs b/crates/core/src/deadline/calculator.rs index 3d75a76f..06e10102 100644 --- a/crates/core/src/deadline/calculator.rs +++ b/crates/core/src/deadline/calculator.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use chrono::{DateTime, Duration, Utc}; -use pluto_eth2api::EthBeaconNodeApiClient; +use pluto_eth2api::BeaconNodeClient; use crate::types::{Duty, DutyType, SlotNumber}; @@ -35,9 +35,9 @@ impl DutyDeadlineCalculator { /// # Errors /// /// Returns an error if fetching genesis time or slots config fails. - pub async fn from_client(client: &EthBeaconNodeApiClient) -> Result { - let genesis_time = client.fetch_genesis_time().await?; - let slots_config = client.fetch_slots_config().await?; + pub async fn from_client(client: &BeaconNodeClient) -> Result { + let genesis_time = client.genesis_time().await?; + let slots_config = client.slots_config().await?; let (slot_duration, _slots_per_epoch) = slots_config; let slot_duration = to_chrono_duration(slot_duration)?; Ok(Self { diff --git a/crates/core/src/deadline/mod.rs b/crates/core/src/deadline/mod.rs index dca033c2..f33ae555 100644 --- a/crates/core/src/deadline/mod.rs +++ b/crates/core/src/deadline/mod.rs @@ -11,10 +11,10 @@ //! deadline::{AddOutcome, DeadlinerTask, DutyDeadlineCalculator}, //! types::{Duty, SlotNumber}, //! }; -//! use pluto_eth2api::EthBeaconNodeApiClient; +//! use pluto_eth2api::BeaconNodeClient; //! use tokio_util::sync::CancellationToken; //! -//! # async fn example(client: &EthBeaconNodeApiClient) -> anyhow::Result<()> { +//! # async fn example(client: &BeaconNodeClient) -> anyhow::Result<()> { //! let cancel_token = CancellationToken::new(); //! let calculator = DutyDeadlineCalculator::from_client(client).await?; //! let (deadliner, mut rx) = DeadlinerTask::start(cancel_token, "example", calculator); @@ -599,9 +599,9 @@ mod tests { let mock = create_mock_beacon_client(genesis_time, slot_duration_secs, slots_per_epoch).await; - let client = mock.client(); + let client = mock.beacon_client(); - let calculator = DutyDeadlineCalculator::from_client(client).await?; + let calculator = DutyDeadlineCalculator::from_client(&client).await?; let duty = Duty::new(SlotNumber::new(100), duty_type); let result = calculator.deadline(&duty)?; @@ -628,7 +628,7 @@ mod tests { let mock = create_mock_beacon_client(genesis_time, slot_duration_secs, slots_per_epoch).await; - let client = mock.client(); + let client = mock.beacon_client(); let slot_duration = Duration::from_secs(slot_duration_secs); let margin = slot_duration.checked_div(12).context("margin overflow")?; @@ -648,7 +648,7 @@ mod tests { .context("slot_start overflow")? }; - let calculator = DutyDeadlineCalculator::from_client(client).await?; + let calculator = DutyDeadlineCalculator::from_client(&client).await?; let expected_duration = match duty_type { DutyType::Proposer | DutyType::Randao => slot_duration diff --git a/crates/core/src/eth2signeddata.rs b/crates/core/src/eth2signeddata.rs index 999ae144..e1a0bd73 100644 --- a/crates/core/src/eth2signeddata.rs +++ b/crates/core/src/eth2signeddata.rs @@ -9,7 +9,7 @@ use std::any::Any; use async_trait::async_trait; use pluto_crypto::types::PublicKey; -use pluto_eth2api::{client::EthBeaconNodeApiClient, spec::phase0::Epoch}; +use pluto_eth2api::{BeaconNodeClient, spec::phase0::Epoch}; use pluto_eth2util::{ helpers::{self, HelperError}, signing::{self, DomainName, SigningError}, @@ -53,21 +53,21 @@ pub trait Eth2SignedData: SignedData { fn domain_name(&self) -> DomainName; /// Returns the epoch at which the signing domain is resolved. - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result; + async fn epoch(&self, config: &BeaconNodeClient) -> Result; } /// Verifies the eth2 signature associated with the given [`Eth2SignedData`]. pub async fn verify_eth2_signed_data( - client: &EthBeaconNodeApiClient, + config: &BeaconNodeClient, data: &dyn Eth2SignedData, pubkey: &PublicKey, ) -> Result<(), Eth2SignedDataError> { let sig_root = data.message_root()?; let signature = data.signature()?; - let epoch = data.epoch(client).await?; + let epoch = data.epoch(config).await?; signing::verify( - client, + config, data.domain_name(), epoch, sig_root, @@ -133,12 +133,12 @@ impl Eth2SignedData for VersionedSignedProposal { DomainName::BeaconProposer } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, config: &BeaconNodeClient) -> Result { if self.0.version == pluto_eth2api::versioned::DataVersion::Unknown { return Err(SignedDataError::UnknownVersion.into()); } - Ok(helpers::epoch_from_slot(client, self.0.block.slot()).await?) + Ok(helpers::epoch_from_slot(config, self.0.block.slot()).await?) } } @@ -148,7 +148,7 @@ impl Eth2SignedData for Attestation { DomainName::BeaconAttester } - async fn epoch(&self, _client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, _config: &BeaconNodeClient) -> Result { Ok(self.0.data.target.epoch) } } @@ -159,7 +159,7 @@ impl Eth2SignedData for VersionedAttestation { DomainName::BeaconAttester } - async fn epoch(&self, _client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, _config: &BeaconNodeClient) -> Result { let version = self.0.version; if version == pluto_eth2api::versioned::DataVersion::Unknown { return Err(SignedDataError::UnknownVersion.into()); @@ -182,7 +182,7 @@ impl Eth2SignedData for SignedVoluntaryExit { DomainName::VoluntaryExit } - async fn epoch(&self, _client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, _config: &BeaconNodeClient) -> Result { Ok(self.0.message.epoch) } } @@ -193,7 +193,7 @@ impl Eth2SignedData for VersionedSignedValidatorRegistration { DomainName::ApplicationBuilder } - async fn epoch(&self, _client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, _config: &BeaconNodeClient) -> Result { // Always use epoch 0 for DomainApplicationBuilder. Ok(0) } @@ -205,7 +205,7 @@ impl Eth2SignedData for SignedRandao { DomainName::Randao } - async fn epoch(&self, _client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, _config: &BeaconNodeClient) -> Result { Ok(self.0.epoch) } } @@ -216,8 +216,8 @@ impl Eth2SignedData for BeaconCommitteeSelection { DomainName::SelectionProof } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { - Ok(helpers::epoch_from_slot(client, self.0.slot).await?) + async fn epoch(&self, config: &BeaconNodeClient) -> Result { + Ok(helpers::epoch_from_slot(config, self.0.slot).await?) } } @@ -227,8 +227,8 @@ impl Eth2SignedData for SignedAggregateAndProof { DomainName::AggregateAndProof } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { - Ok(helpers::epoch_from_slot(client, self.0.message.aggregate.data.slot).await?) + async fn epoch(&self, config: &BeaconNodeClient) -> Result { + Ok(helpers::epoch_from_slot(config, self.0.message.aggregate.data.slot).await?) } } @@ -238,10 +238,10 @@ impl Eth2SignedData for VersionedSignedAggregateAndProof { DomainName::AggregateAndProof } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { + async fn epoch(&self, config: &BeaconNodeClient) -> Result { let slot = self.0.slot().ok_or(SignedDataError::UnknownVersion)?; - Ok(helpers::epoch_from_slot(client, slot).await?) + Ok(helpers::epoch_from_slot(config, slot).await?) } } @@ -251,8 +251,8 @@ impl Eth2SignedData for SignedSyncMessage { DomainName::SyncCommittee } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { - Ok(helpers::epoch_from_slot(client, self.0.slot).await?) + async fn epoch(&self, config: &BeaconNodeClient) -> Result { + Ok(helpers::epoch_from_slot(config, self.0.slot).await?) } } @@ -262,8 +262,8 @@ impl Eth2SignedData for SignedSyncContributionAndProof { DomainName::ContributionAndProof } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { - Ok(helpers::epoch_from_slot(client, self.0.message.contribution.slot).await?) + async fn epoch(&self, config: &BeaconNodeClient) -> Result { + Ok(helpers::epoch_from_slot(config, self.0.message.contribution.slot).await?) } } @@ -273,8 +273,8 @@ impl Eth2SignedData for SyncCommitteeSelection { DomainName::SyncCommitteeSelectionProof } - async fn epoch(&self, client: &EthBeaconNodeApiClient) -> Result { - Ok(helpers::epoch_from_slot(client, self.0.slot).await?) + async fn epoch(&self, config: &BeaconNodeClient) -> Result { + Ok(helpers::epoch_from_slot(config, self.0.slot).await?) } } @@ -331,7 +331,7 @@ mod tests { /// Mirrors Go's `TestVerifyEth2SignedData`: resolve the epoch and message /// root, BLS-sign the signing-domain data root, inject the signature, and /// assert verification succeeds. - async fn assert_verifies(client: &EthBeaconNodeApiClient, data: T) + async fn assert_verifies(client: &BeaconNodeClient, data: T) where T: Eth2SignedData + Clone, { @@ -360,7 +360,7 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: VersionedSignedProposal = load("TestJSONSerialisation_VersionedSignedProposal.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -368,14 +368,14 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: VersionedAttestation = load("TestJSONSerialisation_VersionedAttestation.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] async fn verify_randao() { let mock = BeaconMock::builder().build().await.unwrap(); let data: SignedRandao = load("TestJSONSerialisation_SignedRandao.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -383,7 +383,7 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: SignedVoluntaryExit = load("TestJSONSerialisation_SignedVoluntaryExit.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -391,7 +391,7 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: VersionedSignedValidatorRegistration = load("VersionedSignedValidatorRegistration.v1.json"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -399,7 +399,7 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: BeaconCommitteeSelection = load("TestJSONSerialisation_BeaconCommitteeSelection.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -407,14 +407,14 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: VersionedSignedAggregateAndProof = load("TestJSONSerialisation_VersionedSignedAggregateAndProof.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] async fn verify_phase0_attestation() { let mock = BeaconMock::builder().build().await.unwrap(); let data = Attestation::new(sample_phase0_attestation()); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -428,14 +428,14 @@ mod tests { }, signature: [0x66; 96], }); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] async fn verify_sync_committee_message() { let mock = BeaconMock::builder().build().await.unwrap(); let data: SignedSyncMessage = load("TestJSONSerialisation_SignedSyncMessage.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -443,7 +443,7 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: SignedSyncContributionAndProof = load("TestJSONSerialisation_SignedSyncContributionAndProof.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] @@ -451,13 +451,13 @@ mod tests { let mock = BeaconMock::builder().build().await.unwrap(); let data: SyncCommitteeSelection = load("TestJSONSerialisation_SyncCommitteeSelection.json.golden"); - assert_verifies(mock.client(), data).await; + assert_verifies(&mock.beacon_client(), data).await; } #[tokio::test] async fn verify_rejects_wrong_pubkey() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = &mock.beacon_client(); let data: SignedRandao = load("TestJSONSerialisation_SignedRandao.json.golden"); let epoch = data.epoch(client).await.unwrap(); @@ -485,7 +485,7 @@ mod tests { #[tokio::test] async fn verify_rejects_zero_signature() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = &mock.beacon_client(); let data: SignedRandao = load("TestJSONSerialisation_SignedRandao.json.golden"); let pubkey = [0x11; 48]; diff --git a/crates/core/src/fetcher/mod.rs b/crates/core/src/fetcher/mod.rs index dbf3b80f..fb58a1c6 100644 --- a/crates/core/src/fetcher/mod.rs +++ b/crates/core/src/fetcher/mod.rs @@ -9,7 +9,7 @@ pub use graffiti::{GraffitiBuilder, GraffitiError}; use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc}; use pluto_eth2api::{ - EthBeaconNodeApiClient, EthBeaconNodeApiClientError, GetAggregatedAttestationV2Request, + BeaconNodeClient, EthBeaconNodeApiClientError, GetAggregatedAttestationV2Request, GetAggregatedAttestationV2Response, GetAggregatedAttestationV2ResponseResponseData, ProduceAttestationDataRequest, ProduceAttestationDataResponse, ProduceBlockV3Request, ProduceBlockV3Response, ProduceSyncCommitteeContributionRequest, @@ -142,7 +142,7 @@ pub struct Fetcher { /// builder's `subscribe` method (zero or more times). #[builder(field)] subs: Vec, - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, fee_recipient: FeeRecipientFunc, agg_sig_db: AggSigDbFunc, await_att_data: AwaitAttDataFunc, @@ -360,6 +360,7 @@ impl Fetcher { let response = match self .eth2_cl + .api() .produce_block_v3(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)? @@ -452,6 +453,7 @@ impl Fetcher { match self .eth2_cl + .api() .produce_attestation_data(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)? @@ -479,6 +481,7 @@ impl Fetcher { let ok = match self .eth2_cl + .api() .get_aggregated_attestation_v2(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)? @@ -513,6 +516,7 @@ impl Fetcher { match self .eth2_cl + .api() .produce_sync_committee_contribution(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)? @@ -1016,7 +1020,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(stub_await_att_data()) @@ -1094,7 +1098,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(stub_agg_sig_db()) .await_att_data(stub_await_att_data()) @@ -1309,7 +1313,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetch = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(await_att_data) @@ -1364,7 +1368,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetch = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(await_att_data) @@ -1410,7 +1414,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(await_att_data) @@ -1447,7 +1451,7 @@ mod tests { let (agg_sig_db, await_att_data) = aggregator_funcs(std::slice::from_ref(&att_a)); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(await_att_data) @@ -1548,7 +1552,7 @@ mod tests { let captured: Captured = Arc::new(Mutex::new(None)); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(stub_await_att_data()) @@ -1621,7 +1625,7 @@ mod tests { }); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(stub_await_att_data()) @@ -1657,7 +1661,7 @@ mod tests { }); let fetcher = Fetcher::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .fee_recipient(stub_fee_recipient()) .agg_sig_db(agg_sig_db) .await_att_data(stub_await_att_data()) diff --git a/crates/core/src/gater.rs b/crates/core/src/gater.rs index 4afafeee..9fc9cf88 100644 --- a/crates/core/src/gater.rs +++ b/crates/core/src/gater.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use chrono::{DateTime, Utc}; -use pluto_eth2api::{EthBeaconNodeApiClient, EthBeaconNodeApiClientError}; +use pluto_eth2api::{BeaconNodeClient, EthBeaconNodeApiClientError}; use crate::{ clock::{ChronoClock, Clock}, @@ -58,7 +58,7 @@ impl DutyGater { /// Builds a gater from a beacon node client using production defaults: a /// real wall clock and a `DEFAULT_ALLOWED_FUTURE_EPOCHS` future-epoch /// budget. - pub async fn new(client: &EthBeaconNodeApiClient) -> Result { + pub async fn new(client: &BeaconNodeClient) -> Result { Self::with_options(client, Box::new(ChronoClock), DEFAULT_ALLOWED_FUTURE_EPOCHS).await } @@ -66,12 +66,12 @@ impl DutyGater { /// single fetch path shared with [`DutyGater::new`]; the overrides /// exist for tests. async fn with_options( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, clock: Box, allowed_future_epochs: u64, ) -> Result { - let genesis_time = client.fetch_genesis_time().await?; - let (slot_duration, slots_per_epoch) = client.fetch_slots_config().await?; + let genesis_time = client.genesis_time().await?; + let (slot_duration, slots_per_epoch) = client.slots_config().await?; // Work in whole milliseconds. `as_millis()` is u128 (SECONDS_PER_SLOT // keeps it tiny); reject a zero (sub-millisecond) or overflowing value @@ -201,7 +201,7 @@ mod tests { .await .expect("build beacon mock"); - let gater = DutyGater::with_options(bmock.client(), fixed_clock(now), 2) + let gater = DutyGater::with_options(&bmock.beacon_client(), fixed_clock(now), 2) .await .expect("build gater"); @@ -246,7 +246,9 @@ mod tests { .await .expect("build beacon mock"); - let gater = DutyGater::new(bmock.client()).await.expect("build gater"); + let gater = DutyGater::new(&bmock.beacon_client()) + .await + .expect("build gater"); assert!(gater.allows(&attester(0))); // current epoch assert!(gater.allows(&attester(95))); // epoch 2 (= budget) diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index 3b0f56e2..cd0f298c 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -230,11 +230,12 @@ impl SchedulerBuilder { .await .ok_or(SchedulerError::Terminated)??; - // Cached once here since the node is synced at this point; see the + // Read once here since the node is synced at this point; see the // `slots_per_epoch` field on `SchedulerActor`. - let (_slot_duration, slots_per_epoch) = client.api().fetch_slots_config().await?; + let genesis_time = client.genesis_time().await?; + let (slot_duration, slots_per_epoch) = client.slots_config().await?; - let slot_rx = new_slot_ticker(&client, ct.clone()).await?; + let slot_rx = new_slot_ticker(genesis_time, slot_duration, slots_per_epoch, ct.clone()); let actor = SchedulerActor { client: client.clone(), @@ -298,10 +299,7 @@ impl SchedulerHandle { struct SchedulerActor { client: pluto_eth2api::BeaconNodeClient, - /// Cached chain constant: number of slots per epoch. Fetched once at build - /// time. Charon reads this from the memoized beacon-node spec; Pluto's - /// `fetch_slots_config` is not memoized, so we cache it here to avoid a - /// beacon-node round-trip on every duty lookup. + /// Chain constant, read once at build time from the memoized config. slots_per_epoch: u64, slot_broadcast: sync::broadcast::Sender, @@ -621,12 +619,12 @@ impl SchedulerActor { /// /// The production of slots is cancelled when the provided [`CancellationToken`] /// is cancelled. -async fn new_slot_ticker( - client: &pluto_eth2api::BeaconNodeClient, +fn new_slot_ticker( + genesis_time: chrono::DateTime, + slot_duration: std::time::Duration, + slots_per_epoch: u64, ct: CancellationToken, -) -> Result> { - let genesis_time = client.api().fetch_genesis_time().await?; - let (slot_duration, slots_per_epoch) = client.api().fetch_slots_config().await?; +) -> sync::mpsc::Receiver { let slot_duration = chrono::Duration::from_std(slot_duration).expect("within range"); let current_slot = move || { @@ -690,7 +688,7 @@ async fn new_slot_ticker( } }); - Ok(rx) + rx } struct Validator { @@ -789,7 +787,7 @@ async fn resolve_active_validators( /// Blocks until the beacon chain has started. async fn wait_chain_start(client: &pluto_eth2api::BeaconNodeClient) -> Result<()> { - let fetch = || client.api().fetch_genesis_time(); + let fetch = || client.genesis_time(); let backoff = crate::expbackoff::fast(); let genesis_time = fetch .retry(backoff) diff --git a/crates/core/src/sigagg.rs b/crates/core/src/sigagg.rs index 944d7094..19c06530 100644 --- a/crates/core/src/sigagg.rs +++ b/crates/core/src/sigagg.rs @@ -4,7 +4,7 @@ use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc}; use pluto_crypto::{blst_impl::BlstImpl, tbls::Tbls, types::PublicKey}; -use pluto_eth2api::client::EthBeaconNodeApiClient; +use pluto_eth2api::BeaconNodeClient; use tracing::{debug, error, info_span}; use crate::{ @@ -255,7 +255,7 @@ impl Aggregator { /// Returns a [`VerifyFn`] that verifies the aggregated signature against the /// beacon chain. -pub fn new_verifier(eth2_cl: Arc) -> VerifyFn { +pub fn new_verifier(eth2_cl: BeaconNodeClient) -> VerifyFn { Arc::new(move |pubkey: &PubKey, data: &dyn SignedData| { let eth2_cl = eth2_cl.clone(); // The future must be `'static`, so clone the borrowed inputs out of the @@ -300,7 +300,7 @@ mod tests { #[tokio::test] async fn new_verifier_rejects_non_eth2_data() { let mock = pluto_testutil::BeaconMock::builder().build().await.unwrap(); - let eth2_cl = Arc::new(mock.client().clone()); + let eth2_cl = mock.beacon_client(); let verify = new_verifier(eth2_cl); let data = MockSignedData { diff --git a/crates/core/src/validatorapi/component.rs b/crates/core/src/validatorapi/component.rs index 8dc24108..8e4ff63b 100644 --- a/crates/core/src/validatorapi/component.rs +++ b/crates/core/src/validatorapi/component.rs @@ -9,7 +9,7 @@ use std::{any::Any, collections::HashMap, future::Future, pin::Pin, sync::Arc, t use async_trait::async_trait; use axum::http::StatusCode; use pluto_eth2api::{ - EthBeaconNodeApiClient, GetAttesterDutiesRequest, GetAttesterDutiesResponse, + BeaconNodeClient, GetAttesterDutiesRequest, GetAttesterDutiesResponse, GetProposerDutiesRequest, GetProposerDutiesResponse, GetStateValidatorsResponseResponse, GetSyncCommitteeDutiesRequest, GetSyncCommitteeDutiesResponse, PostStateValidatorsRequest, PostStateValidatorsRequestPath, PostStateValidatorsResponse, ValidatorRequestBody, @@ -18,7 +18,7 @@ use pluto_eth2api::{ versioned::{DataVersion, SignedBlindedProposalBlock, SignedProposalBlock}, }; use pluto_eth2util::{ - helpers::epoch_from_slot, + helpers::{HelperError, epoch_from_slot}, signing::{self, DomainName, SigningError}, }; use tokio::time::error::Elapsed; @@ -162,8 +162,9 @@ const PROPOSAL_TIMEOUT: Duration = Duration::from_secs(24); /// validator client, and emits partial-signed-data to subscribers on submit /// endpoints. pub struct Component { - /// Upstream beacon-node API client. - eth2_cl: Arc, + /// Upstream beacon-node API client with cached static chain config + /// (spec, genesis, fork schedule) for signing-domain resolution. + eth2_cl: BeaconNodeClient, /// Per-epoch active-validators cache. Submit handlers consult this to /// translate a validator-client-supplied `validator_index` into the /// cluster's DV root public key. @@ -209,7 +210,7 @@ pub struct Component { impl Component { /// Builds a new component. pub fn new( - eth2_cl: Arc, + eth2_cl: BeaconNodeClient, dutydb: Arc, share_idx: u64, pub_share_by_pubkey: HashMap, @@ -241,7 +242,7 @@ impl Component { /// to bypass signature checks. #[cfg(test)] pub fn new_insecure( - eth2_cl: Arc, + eth2_cl: BeaconNodeClient, dutydb: Arc, share_idx: u64, validator_cache: Arc, @@ -450,8 +451,7 @@ impl Component { "unknown validator public key for partial signature", ), VerifyPartialSigError::Signing(inner) => { - ApiError::new(StatusCode::BAD_REQUEST, "invalid partial signature") - .with_source(inner) + signing_error_to_api_error(inner, "invalid partial signature") } }) } @@ -746,11 +746,10 @@ impl Component { StatusCode::INTERNAL_SERVER_ERROR, format!("{endpoint}: unknown validator public key"), ), - VerifyPartialSigError::Signing(inner) => ApiError::new( - StatusCode::BAD_REQUEST, + VerifyPartialSigError::Signing(inner) => signing_error_to_api_error( + inner, format!("{endpoint}: invalid partial signature"), - ) - .with_source(inner), + ), }) } @@ -915,7 +914,7 @@ impl Handler for Component { let response = tokio::time::timeout( UPSTREAM_REQUEST_TIMEOUT, - self.eth2_cl.get_proposer_duties(request), + self.eth2_cl.api().get_proposer_duties(request), ) .await .map_err(|_| upstream_timeout("proposer duties"))? @@ -965,7 +964,7 @@ impl Handler for Component { let response = tokio::time::timeout( UPSTREAM_REQUEST_TIMEOUT, - self.eth2_cl.get_attester_duties(request), + self.eth2_cl.api().get_attester_duties(request), ) .await .map_err(|_| upstream_timeout("attester duties"))? @@ -1018,7 +1017,7 @@ impl Handler for Component { let response = tokio::time::timeout( UPSTREAM_REQUEST_TIMEOUT, - self.eth2_cl.get_sync_committee_duties(request), + self.eth2_cl.api().get_sync_committee_duties(request), ) .await .map_err(|_| upstream_timeout("sync committee duties"))? @@ -1363,11 +1362,10 @@ impl Handler for Component { signing::verify_aggregate_and_proof_selection(&self.eth2_cl, eth2_pubkey, &agg.0) .await .map_err(|err| { - ApiError::new( - StatusCode::BAD_REQUEST, + signing_error_to_api_error( + err, "aggregate selection proof verification failed", ) - .with_source(err) })?; } @@ -1643,7 +1641,7 @@ impl Handler for Component { let response = tokio::time::timeout( UPSTREAM_REQUEST_TIMEOUT, - self.eth2_cl.post_state_validators(request), + self.eth2_cl.api().post_state_validators(request), ) .await .map_err(|_| upstream_timeout("validators"))? @@ -1726,12 +1724,12 @@ impl Handler for Component { // so we resolve it once here too rather than letting // `verify_partial_sig` fan out 2N domain-lookup calls. let (slot_duration, _) = - tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.fetch_slots_config()) + tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.slots_config()) .await .map_err(|_| upstream_timeout("slots config"))? .map_err(|err| upstream_call_failed("slots config", err.into()))?; let genesis_time = - tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.fetch_genesis_time()) + tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.genesis_time()) .await .map_err(|_| upstream_timeout("genesis time"))? .map_err(|err| upstream_call_failed("genesis time", err.into()))?; @@ -1766,7 +1764,7 @@ impl Handler for Component { // Duty slot = slots_per_epoch * epoch. let (_, slots_per_epoch) = - tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.fetch_slots_config()) + tokio::time::timeout(UPSTREAM_REQUEST_TIMEOUT, self.eth2_cl.slots_config()) .await .map_err(|_| upstream_timeout("slots config"))? .map_err(|err| upstream_call_failed("slots config", err.into()))?; @@ -1910,11 +1908,7 @@ impl Handler for Component { ) .await .map_err(|err| { - ApiError::new( - StatusCode::BAD_REQUEST, - "invalid sync committee selection proof", - ) - .with_source(err) + signing_error_to_api_error(err, "invalid sync committee selection proof") })?; } @@ -2268,11 +2262,39 @@ fn pubkey_to_bls(pk: &PubKey) -> BLSPubKey { out } -/// Maps a [`VerifyPartialSigError`] into the `ApiError` returned to the -/// client. `UnknownPubKey` is a misconfiguration (500), `Signing` is a -/// validator-client mistake (400) — both keep the underlying error as a -/// `source` so the debug log retains it while the client sees a generic -/// message. +/// Maps a [`SigningError`] to the client-facing `ApiError`: upstream +/// beacon-node failures are 502 (no signature was checked, so 400 would +/// mislead the VC); failures attributable to the submitted signature are 400 +/// with `invalid_msg`; anything else — e.g. a public key from the cluster +/// lock or validator cache that is not a valid BLS point — is server-side +/// state, so 500. +fn signing_error_to_api_error(err: SigningError, invalid_msg: impl Into) -> ApiError { + use pluto_crypto::types::Error as CryptoError; + + match err { + SigningError::BeaconNode(_) | SigningError::Helper(HelperError::GettingSpec(_)) => { + ApiError::new( + StatusCode::BAD_GATEWAY, + "beacon node lookup failed during signature verification", + ) + .with_source(err) + } + SigningError::ZeroSignature + | SigningError::UnknownAggregateAndProofVersion + | SigningError::Verification( + CryptoError::InvalidSignature(_) | CryptoError::VerificationFailed(_), + ) => ApiError::new(StatusCode::BAD_REQUEST, invalid_msg).with_source(err), + _ => ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "internal error during signature verification", + ) + .with_source(err), + } +} + +/// Maps a [`VerifyPartialSigError`] to the client-facing `ApiError`: +/// `UnknownPubKey` is a misconfiguration (500), `Signing` follows +/// [`signing_error_to_api_error`]. fn verify_partial_sig_error(err: VerifyPartialSigError) -> ApiError { match err { VerifyPartialSigError::UnknownPubKey => ApiError::new( @@ -2280,11 +2302,9 @@ fn verify_partial_sig_error(err: VerifyPartialSigError) -> ApiError { "unknown public key for partial signature verification", ) .with_source(err), - VerifyPartialSigError::Signing(_) => ApiError::new( - StatusCode::BAD_REQUEST, - "partial signature verification failed", - ) - .with_source(err), + VerifyPartialSigError::Signing(inner) => { + signing_error_to_api_error(inner, "partial signature verification failed") + } } } @@ -2708,10 +2728,13 @@ mod tests { use chrono::{DateTime, Utc}; use pluto_crypto::{blst_impl::BlstImpl, tbls::Tbls}; - use pluto_eth2api::spec::altair::{ - ContributionAndProof, SignedContributionAndProof as AltairSignedContributionAndProof, - SyncCommitteeContribution as AltairSyncCommitteeContribution, - SyncCommitteeMessage as AltairSyncCommitteeMessage, + use pluto_eth2api::{ + EthBeaconNodeApiClient, + spec::altair::{ + ContributionAndProof, SignedContributionAndProof as AltairSignedContributionAndProof, + SyncCommitteeContribution as AltairSyncCommitteeContribution, + SyncCommitteeMessage as AltairSyncCommitteeMessage, + }, }; use pluto_ssz::BitVector; use pluto_testutil::BeaconMock; @@ -2788,8 +2811,7 @@ mod tests { // `evict_rx` doesn't observe a closed channel. let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let component = Component::new_insecure(eth2_cl, Arc::clone(&dutydb), 1, TestValidatorCache::empty()); (component, dutydb) @@ -3031,8 +3053,7 @@ mod tests { DeadlinerTask::start(cancel.clone(), "validatorapi-tests", FarFutureCalculator); let (trim_tx, trim_rx) = channel::(8); let dutydb = Arc::new(MemDB::new(deadliner, trim_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let component = Component::new_insecure(eth2_cl, Arc::clone(&dutydb), 1, TestValidatorCache::empty()); @@ -3199,6 +3220,16 @@ mod tests { // Plumbing tests — Subscribe / Register* / verify_partial_sig // ==================================================================== + /// Beacon client over the given (mock server) URL. + fn beacon_client_at(url: &str) -> BeaconNodeClient { + BeaconNodeClient::new(EthBeaconNodeApiClient::with_base_url(url).unwrap()) + } + + /// Beacon client pinned to an unroutable endpoint. + fn dummy_beacon_client() -> BeaconNodeClient { + beacon_client_at("http://127.0.0.1:0") + } + fn dv_pubkey(byte: u8) -> BLSPubKey { [byte; 48] } @@ -3218,8 +3249,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); Component::new(eth2_cl, dutydb, 1, map, false, TestValidatorCache::empty()) } @@ -3458,7 +3488,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new(eth2_cl, dutydb, 1, map, false, TestValidatorCache::empty()); (component, mock) } @@ -3485,10 +3515,14 @@ mod tests { // Compute the signing root the same way `signing::verify` does, then // sign it with the share's secret. - let signing_root = - pluto_eth2util::signing::get_data_root(mock.client(), domain, epoch, message_root) - .await - .unwrap(); + let signing_root = pluto_eth2util::signing::get_data_root( + &mock.beacon_client(), + domain, + epoch, + message_root, + ) + .await + .unwrap(); let good_signature = BlstImpl.sign(&secret, &signing_root).unwrap(); component @@ -3540,8 +3574,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let component = Component::new_insecure(eth2_cl, dutydb, 1, TestValidatorCache::empty()); component @@ -3556,6 +3589,93 @@ mod tests { .expect("insecure_test mode skips verification"); } + /// A beacon-node failure during domain resolution must map to 502, not + /// the 400 the VC would misread as an invalid signature. + #[tokio::test] + async fn beacon_outage_during_verification_maps_to_502_not_400() { + let secret = BlstImpl + .generate_insecure_secret(rand::rngs::OsRng) + .unwrap(); + let pubshare = BlstImpl.secret_to_public_key(&secret).unwrap(); + let dv_root = dv_pubkey(0xAB); + + // Unroutable beacon node: domain resolution fails before any BLS + // verification. Non-zero signature avoids the zero-sig short-circuit. + let cancel = CancellationToken::new(); + let (deadliner, _deadliner_rx) = DeadlinerTask::start( + cancel.clone(), + "validatorapi-outage-tests", + FarFutureCalculator, + ); + let (_evict_tx, evict_rx) = mpsc::channel(1); + let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); + let eth2_cl = dummy_beacon_client(); + let component = Component::new( + eth2_cl, + dutydb, + 1, + HashMap::from([(dv_root, pubshare)]), + false, + TestValidatorCache::empty(), + ); + + let err = component + .verify_partial_sig( + &dv_root, + DomainName::BeaconAttester, + 0, + [0x42; 32], + &[0x11; 96], + ) + .await + .unwrap_err(); + assert!( + matches!( + err, + VerifyPartialSigError::Signing(SigningError::BeaconNode(_)) + ), + "expected upstream signing error, got {err:?}" + ); + assert_eq!( + verify_partial_sig_error(err).status_code, + StatusCode::BAD_GATEWAY + ); + } + + /// Genuine signature failures keep mapping to 400. + #[test] + fn verify_partial_sig_error_maps_signature_failures_to_400() { + let err = VerifyPartialSigError::Signing(SigningError::ZeroSignature); + assert_eq!( + verify_partial_sig_error(err).status_code, + StatusCode::BAD_REQUEST + ); + } + + /// A configured pubshare that is not a valid BLS point comes from the + /// cluster lock, not the VC, so it must map to 500 rather than 400. + #[tokio::test] + async fn invalid_configured_pubshare_maps_to_500_not_400() { + let dv_root = dv_pubkey(0xAC); + let bad_share: BLSPubKey = [0x11; 48]; + let (component, _mock) = make_verify_component(HashMap::from([(dv_root, bad_share)])).await; + + let err = component + .verify_partial_sig( + &dv_root, + DomainName::BeaconAttester, + 0, + [0x42; 32], + &[0x11; 96], + ) + .await + .unwrap_err(); + assert_eq!( + verify_partial_sig_error(err).status_code, + StatusCode::INTERNAL_SERVER_ERROR + ); + } + // CachedValidatorsProvider plumbing // ==================================================================== @@ -3571,8 +3691,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let expected = HashMap::from([(1u64, dv_pubkey(0xA1)), (7u64, dv_pubkey(0xA7))]); let component = Component::new_insecure( @@ -3619,8 +3738,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let component = Component::new_insecure(eth2_cl, dutydb, 1, Arc::new(FailingCache)); let err = component.fetch_active_validators().await.unwrap_err(); @@ -3665,7 +3783,7 @@ mod tests { DeadlinerTask::start(cancel.clone(), "selections-tests", FarFutureCalculator); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new_insecure( eth2_cl, dutydb, @@ -3698,7 +3816,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new( eth2_cl, dutydb, @@ -4195,7 +4313,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let mut component = Component::new( eth2_cl, dutydb, @@ -4351,7 +4469,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new( eth2_cl, dutydb, @@ -4483,7 +4601,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new(eth2_cl, dutydb, 1, map, true, TestValidatorCache::empty()); let reg = make_signed_registration(dv_root, 24, [0x42; 96]); @@ -4520,8 +4638,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let mut component = Component::new_insecure(eth2_cl, dutydb, 7, TestValidatorCache::arc(active)); @@ -4788,8 +4905,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = - Arc::new(EthBeaconNodeApiClient::with_base_url("http://127.0.0.1:0").unwrap()); + let eth2_cl = dummy_beacon_client(); let component = Component::new_insecure(eth2_cl, dutydb, 1, Arc::new(FailingCache)); let err = component @@ -4864,7 +4980,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); // Empty share map: lookup for `dv_root` will return // `VerifyPartialSigError::UnknownPubKey`, which the handler maps // to 400. @@ -4903,7 +5019,7 @@ mod tests { // Resolve the same signing root the handler will compute (epoch=0 // since slot/SLOTS_PER_EPOCH=1/16=0). let signing_root = pluto_eth2util::signing::get_data_root( - mock.client(), + &mock.beacon_client(), DomainName::SyncCommittee, 0, beacon_block_root, @@ -4921,7 +5037,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let active: HashMap = HashMap::from([(7, dv_root)]); let mut component = Component::new( eth2_cl, @@ -4979,12 +5095,12 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new(eth2_cl, dutydb, 1, map, false, TestValidatorCache::empty()); let message_root: Root = [0xCD; 32]; let signing_root = pluto_eth2util::signing::get_data_root( - mock.client(), + &mock.beacon_client(), DomainName::SyncCommittee, 0, message_root, @@ -5011,7 +5127,12 @@ mod tests { /// `verify_partial_sig_for` is reached for the contribution path too. #[tokio::test] async fn submit_sync_committee_contributions_rejects_invalid_partial_sig() { - let dv_root = [0xCD_u8; 48]; + // A valid BLS point: an unparseable root pubkey would map to 500 + // (server-side state), not the 400 under test. + let secret = BlstImpl + .generate_insecure_secret(rand::rngs::OsRng) + .unwrap(); + let dv_root = BlstImpl.secret_to_public_key(&secret).unwrap(); let mock = mock_beacon_for_signing().await; let cancel = CancellationToken::new(); let (deadliner, _deadliner_rx) = DeadlinerTask::start( @@ -5021,7 +5142,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); // `insecure_test = false` but no share registered for `dv_root`. The // inner selection-proof verify runs first; because the selection // proof is a zero-byte signature here it will be rejected with 400 @@ -5036,9 +5157,8 @@ mod tests { TestValidatorCache::arc(active), ); - // The dummy fixture's `selection_proof` is `[0x50; 96]` — a random - // non-zero garbage signature, so `signing::verify` returns - // `VerifyFailed`, which we map to 400. + // The dummy fixture's `selection_proof` is `[0x50; 96]` — non-zero + // garbage, so `signing::verify` rejects it and maps to 400. let err = component .submit_sync_committee_contributions(vec![dummy_signed_contribution_and_proof(1, 7, 0)]) .await @@ -5089,7 +5209,7 @@ mod tests { } .selection_proof_message_root(); let selection_proof_signing_root = pluto_eth2util::signing::get_data_root( - mock.client(), + &mock.beacon_client(), DomainName::SyncCommitteeSelectionProof, 0, selection_proof_root, @@ -5114,7 +5234,7 @@ mod tests { } .message_root(); let outer_signing_root = pluto_eth2util::signing::get_data_root( - mock.client(), + &mock.beacon_client(), DomainName::ContributionAndProof, 0, outer_root, @@ -5132,7 +5252,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let active: HashMap = HashMap::from([(aggregator_index, root_pubkey)]); let mut component = Component::new( @@ -5228,7 +5348,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component::new_insecure(eth2_cl, Arc::clone(&dutydb), 1, TestValidatorCache::empty()); (component, mock) @@ -5891,7 +6011,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let mut component = Component::new( eth2_cl, Arc::clone(&dutydb), @@ -6075,7 +6195,7 @@ mod tests { DeadlinerTask::start(cancel.clone(), "validatorapi-tests", FarFutureCalculator); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(server.uri()).unwrap()); + let eth2_cl = beacon_client_at(&server.uri()); Component::new( eth2_cl, dutydb, @@ -6535,7 +6655,7 @@ mod tests { ); let (_evict_tx, evict_rx) = mpsc::channel(1); let dutydb = Arc::new(MemDB::new(deadliner, evict_rx, &cancel)); - let eth2_cl = Arc::new(EthBeaconNodeApiClient::with_base_url(mock.uri()).unwrap()); + let eth2_cl = beacon_client_at(&mock.uri()); let component = Component { eth2_cl, dutydb, diff --git a/crates/eth2api/src/beacon_node.rs b/crates/eth2api/src/beacon_node.rs index 5bef37cb..22ac90c3 100644 --- a/crates/eth2api/src/beacon_node.rs +++ b/crates/eth2api/src/beacon_node.rs @@ -1,11 +1,19 @@ use crate::{ - EthBeaconNodeApiClient, + ConsensusVersion, EthBeaconNodeApiClient, EthBeaconNodeApiClientError, + confcache::ConfigCache, + extensions::{ + self, ForkSchedule, GenesisInfo, compute_builder_domain, domain_from_config, + fork_schedule_from_spec, resolve_domain_type, + }, + spec::phase0, valcache::{ActiveValidators, CompleteValidators, ValidatorCache, ValidatorCacheError}, }; -use std::sync::Arc; +use chrono::{DateTime, Utc}; +use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; use tokio::sync::RwLock; type Result = std::result::Result; +type ConfigResult = std::result::Result; /// Errors returned by [`BeaconNodeClient`]. #[derive(Debug, thiserror::Error)] @@ -15,34 +23,120 @@ pub enum BeaconNodeClientError { ValidatorCache(#[from] ValidatorCacheError), } -/// Beacon node client with Charon/Pluto convenience state layered on top of the -/// generated Beacon API client. -#[derive(Clone)] -pub struct BeaconNodeClient { +/// Shared state behind every [`BeaconNodeClient`] clone. +struct Inner { api: EthBeaconNodeApiClient, + config: ConfigCache, // TODO: Find the concrete usages of the `validator_cache` and consider if we can make it // immutable, that is, set it once at construction and not have to deal with the possibility of // it being unset later. - validator_cache: Arc>, + validator_cache: RwLock, +} + +/// Beacon node client layering a per-epoch validator cache and the static +/// chain-config cache (backing signing-domain resolution) over the generated +/// API client. Clones share one `Arc`. +#[derive(Clone)] +pub struct BeaconNodeClient(Arc); + +impl fmt::Debug for BeaconNodeClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BeaconNodeClient") + .field("base_url", &self.0.api.base_url.as_str()) + .finish_non_exhaustive() + } } impl BeaconNodeClient { /// Creates a new beacon node client. pub fn new(api: EthBeaconNodeApiClient) -> Self { - Self { - api: api.clone(), - validator_cache: Arc::new(RwLock::new(ValidatorCache::new(api, Vec::new()))), - } + Self(Arc::new(Inner { + config: ConfigCache::default(), + validator_cache: RwLock::new(ValidatorCache::new(api.clone(), Vec::new())), + api, + })) } /// Returns the generated Beacon API client. pub fn api(&self) -> &EthBeaconNodeApiClient { - &self.api + &self.0.api + } + + /// Warms the static-config cache (spec, genesis, fork schedule). Called + /// at startup, before duty scheduling, failing fast. + pub async fn warm(&self) -> ConfigResult<()> { + tokio::try_join!(self.spec(), self.genesis(), self.fork_schedule())?; + Ok(()) + } + + /// Returns the chain spec as a JSON object (cached). + pub async fn spec(&self) -> ConfigResult> { + self.0.config.spec(&self.0.api).await + } + + /// Returns the parsed genesis data (cached). + pub(crate) async fn genesis(&self) -> ConfigResult> { + self.0.config.genesis(&self.0.api).await + } + + /// Returns the parsed fork-schedule entries, in server order (cached). + /// The first entry is the genesis fork version, which identifies the + /// beacon node's network. + pub async fn fork_schedule(&self) -> ConfigResult>> { + self.0.config.fork_schedule(&self.0.api).await + } + + /// Returns the genesis time (cached). + pub async fn genesis_time(&self) -> ConfigResult> { + Ok(self.genesis().await?.time) + } + + /// Returns the slot duration and slots per epoch (cached). + pub async fn slots_config(&self) -> ConfigResult<(Duration, u64)> { + let spec = self.spec().await?; + extensions::slots_config_from_spec(&spec) + } + + /// Returns the spec-derived fork schedule for all known forks (cached). + pub async fn fork_config(&self) -> ConfigResult> { + let spec = self.spec().await?; + fork_schedule_from_spec(&spec) + } + + /// Returns the domain type with the provided config/spec key (cached). + pub async fn domain_type(&self, spec_key: &str) -> ConfigResult { + let spec = self.spec().await?; + resolve_domain_type(&spec, spec_key) + } + + /// Returns the genesis (builder) domain for the provided domain type + /// (cached). + pub async fn genesis_domain( + &self, + domain_type: phase0::DomainType, + ) -> ConfigResult { + let genesis = self.genesis().await?; + + Ok(compute_builder_domain(domain_type, genesis.fork_version)) + } + + /// Returns the resolved beacon domain for the provided domain type and + /// epoch (cached); see [`domain_from_config`] for the derivation rules. + pub async fn domain( + &self, + domain_type: phase0::DomainType, + epoch: phase0::Epoch, + ) -> ConfigResult { + let spec = self.spec().await?; + let genesis = self.genesis().await?; + let schedule = self.fork_schedule().await?; + + domain_from_config(&spec, &genesis, &schedule, domain_type, epoch) } /// Sets the validator cache used by cached validator methods. pub async fn set_validator_cache(&self, validator_cache: ValidatorCache) { - *self.validator_cache.write().await = validator_cache; + *self.0.validator_cache.write().await = validator_cache; } /// Returns active validators for `head`. @@ -59,7 +153,7 @@ impl BeaconNodeClient { /// Get the validator cache. pub async fn validator_cache(&self) -> ValidatorCache { - self.validator_cache.read().await.clone() + self.0.validator_cache.read().await.clone() } } diff --git a/crates/eth2api/src/confcache.rs b/crates/eth2api/src/confcache.rs new file mode 100644 index 00000000..13a0e51b --- /dev/null +++ b/crates/eth2api/src/confcache.rs @@ -0,0 +1,355 @@ +//! Cache of static beacon-node chain config (spec, genesis, fork schedule), +//! so signing-domain resolution does not hit the beacon node per operation. +//! The generated [`EthBeaconNodeApiClient`] cannot hold state, so the cache +//! lives in [`BeaconNodeClient`](crate::BeaconNodeClient), which exposes the +//! derived lookups over this module's mechanism. + +use crate::{ + EthBeaconNodeApiClient, EthBeaconNodeApiClientError, + extensions::{ForkSchedule, GenesisInfo, parse_fork_schedule, parse_genesis}, +}; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; +use tokio::sync::RwLock; + +type Result = std::result::Result; + +/// How long a cached value is served before re-fetching, so fork-schedule +/// changes (e.g. a beacon-node upgrade) are picked up within minutes. +const STATIC_CONFIG_TTL: Duration = Duration::from_secs(5 * 60); + +/// One cached config value and its fetch time. +#[derive(Debug)] +struct ConfigEntry { + value: RwLock, Instant)>>, +} + +impl Default for ConfigEntry { + fn default() -> Self { + Self { + value: RwLock::new(None), + } + } +} + +impl ConfigEntry { + /// Returns the cached value, fetching when absent or older than `ttl`. + /// The fetch runs under the write lock, so concurrent cold callers + /// coalesce into one request (a caller cancelled mid-fetch releases the + /// lock and the next caller retries). Failures are never cached — and + /// never masked by an expired value, which caps how stale the config can + /// get (a stale fork schedule would derive wrong signing domains across + /// a fork activation) — so the error propagates and the next caller + /// retries. + async fn get_or_fetch(&self, ttl: Duration, fetch: F) -> Result> + where + F: FnOnce() -> Fut, + Fut: Future>, + { + { + let cached = self.value.read().await; + if let Some((value, fetched_at)) = &*cached + && fetched_at.elapsed() < ttl + { + return Ok(Arc::clone(value)); + } + } + + let mut cached = self.value.write().await; + // Re-check: a caller holding the write lock before us may have + // filled the entry while we were blocked acquiring it. + if let Some((value, fetched_at)) = &*cached + && fetched_at.elapsed() < ttl + { + return Ok(Arc::clone(value)); + } + + let value = Arc::new(fetch().await?); + *cached = Some((Arc::clone(&value), Instant::now())); + + Ok(value) + } +} + +/// The cached static beacon-node config: spec, genesis, and fork schedule. +/// The owning client passes its API handle into the fetching getters. +#[derive(Debug, Default)] +pub(crate) struct ConfigCache { + spec: ConfigEntry, + genesis: ConfigEntry, + fork_schedule: ConfigEntry>, +} + +impl ConfigCache { + /// Returns the chain spec as a JSON object. + pub(crate) async fn spec( + &self, + api: &EthBeaconNodeApiClient, + ) -> Result> { + self.spec + .get_or_fetch(STATIC_CONFIG_TTL, || api.fetch_spec_data()) + .await + } + + /// Returns the parsed genesis data (parsed at fetch time, so malformed + /// responses are never cached). + pub(crate) async fn genesis(&self, api: &EthBeaconNodeApiClient) -> Result> { + self.genesis + .get_or_fetch(STATIC_CONFIG_TTL, || async { + parse_genesis(&api.fetch_genesis_data().await?) + }) + .await + } + + /// Returns the parsed fork-schedule entries, in server order. + pub(crate) async fn fork_schedule( + &self, + api: &EthBeaconNodeApiClient, + ) -> Result>> { + self.fork_schedule + .get_or_fetch(STATIC_CONFIG_TTL, || async { + parse_fork_schedule(&api.fetch_fork_schedule_data().await?) + }) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::BeaconNodeClient; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + const SPEC_PATH: &str = "/eth/v1/config/spec"; + const GENESIS_PATH: &str = "/eth/v1/beacon/genesis"; + const FORK_SCHEDULE_PATH: &str = "/eth/v1/config/fork_schedule"; + + fn spec_body() -> serde_json::Value { + json!({ "data": { + "SECONDS_PER_SLOT": "12", + "SLOTS_PER_EPOCH": "32", + "DOMAIN_BEACON_ATTESTER": "0x01000000", + "DOMAIN_VOLUNTARY_EXIT": "0x04000000", + "ALTAIR_FORK_VERSION": "0x01000000", + "ALTAIR_FORK_EPOCH": "10", + "BELLATRIX_FORK_VERSION": "0x02000000", + "BELLATRIX_FORK_EPOCH": "20", + "CAPELLA_FORK_VERSION": "0x03000000", + "CAPELLA_FORK_EPOCH": "30", + "DENEB_FORK_VERSION": "0x04000000", + "DENEB_FORK_EPOCH": "40", + "ELECTRA_FORK_VERSION": "0x05000000", + "ELECTRA_FORK_EPOCH": "50", + "FULU_FORK_VERSION": "0x06000000", + "FULU_FORK_EPOCH": "60", + }}) + } + + fn genesis_body() -> serde_json::Value { + json!({ "data": { + "genesis_time": "1606824023", + "genesis_validators_root": + "0x4b363db94e286120d76eb905340fdd4e54bfe9f06bf33ff6cf5ad27f511bfe95", + "genesis_fork_version": "0x00000000", + }}) + } + + fn fork_schedule_body() -> serde_json::Value { + json!({ "data": [ + { + "previous_version": "0x00000000", + "current_version": "0x00000000", + "epoch": "0" + }, + { + "previous_version": "0x00000000", + "current_version": "0x01000000", + "epoch": "10" + }, + ]}) + } + + async fn mount_config(server: &MockServer, expect: u64) { + Mock::given(method("GET")) + .and(path(SPEC_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(spec_body())) + .expect(expect) + .mount(server) + .await; + Mock::given(method("GET")) + .and(path(GENESIS_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(genesis_body())) + .expect(expect) + .mount(server) + .await; + Mock::given(method("GET")) + .and(path(FORK_SCHEDULE_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(fork_schedule_body())) + .expect(expect) + .mount(server) + .await; + } + + fn client_over(server: &MockServer) -> BeaconNodeClient { + BeaconNodeClient::new( + EthBeaconNodeApiClient::with_base_url(server.uri()).expect("valid mock server URL"), + ) + } + + #[tokio::test] + async fn repeated_lookups_fetch_each_endpoint_once() { + let server = MockServer::start().await; + mount_config(&server, 1).await; + let client = client_over(&server); + + client.warm().await.unwrap(); + + // Every lookup after warm() is served from cache (`.expect(1)` mocks). + let attester = client.domain_type("DOMAIN_BEACON_ATTESTER").await.unwrap(); + let first = client.domain(attester, 20).await.unwrap(); + let second = client.domain(attester, 20).await.unwrap(); + assert_eq!(first, second); + client.genesis_domain(attester).await.unwrap(); + assert_eq!( + client.slots_config().await.unwrap(), + (Duration::from_secs(12), 32) + ); + client.fork_config().await.unwrap(); + client.genesis_time().await.unwrap(); + + // Fork selection: epoch 5 resolves the first schedule entry, epoch 20 + // the second, so the domains differ. + assert_ne!(client.domain(attester, 5).await.unwrap(), first); + } + + #[tokio::test] + async fn concurrent_cold_lookups_coalesce_into_one_fetch() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(SPEC_PATH)) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(spec_body()) + // Force overlap with the first in-flight fetch. + .set_delay(Duration::from_millis(100)), + ) + .expect(1) + .mount(&server) + .await; + let client = client_over(&server); + + // Spawn all tasks before awaiting any (a lazy `map` would run them + // sequentially) so they race on the cold entry. + let lookups: Vec<_> = (0..16) + .map(|_| { + let client = client.clone(); + tokio::spawn(async move { client.spec().await }) + }) + .collect(); + for lookup in lookups { + lookup.await.unwrap().unwrap(); + } + } + + #[tokio::test] + async fn fetch_failures_are_not_cached() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(GENESIS_PATH)) + .respond_with(ResponseTemplate::new(500)) + .expect(1) + .up_to_n_times(1) + .mount(&server) + .await; + + let client = client_over(&server); + client.genesis().await.unwrap_err(); + + // Not cached: the next lookup after recovery succeeds. + Mock::given(method("GET")) + .and(path(GENESIS_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(genesis_body())) + .expect(1) + .mount(&server) + .await; + client.genesis().await.unwrap(); + } + + /// An empty fork schedule is a fetch failure: cached, it would break all + /// non-builder domain resolution until TTL expiry. + #[tokio::test] + async fn empty_fork_schedule_is_rejected_and_not_cached() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(FORK_SCHEDULE_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "data": [] }))) + .expect(1) + .up_to_n_times(1) + .mount(&server) + .await; + + let client = client_over(&server); + client.fork_schedule().await.unwrap_err(); + + // Not cached: the next lookup after recovery succeeds. + Mock::given(method("GET")) + .and(path(FORK_SCHEDULE_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(fork_schedule_body())) + .expect(1) + .mount(&server) + .await; + assert_eq!(client.fork_schedule().await.unwrap().len(), 2); + } + + #[tokio::test] + async fn expired_values_are_refetched() { + let entry = ConfigEntry::::default(); + let fetches = AtomicUsize::new(0); + + for _ in 0..2 { + entry + .get_or_fetch(Duration::ZERO, || async { + fetches.fetch_add(1, Ordering::SeqCst); + Ok(0) + }) + .await + .unwrap(); + } + + assert_eq!(fetches.load(Ordering::SeqCst), 2); + } + + /// A failed refresh propagates the error rather than serving the expired + /// value, capping config staleness at one TTL; the next call retries. + #[tokio::test] + async fn refresh_failure_propagates_and_next_call_retries() { + let entry = ConfigEntry::::default(); + + entry + .get_or_fetch(Duration::ZERO, || async { Ok(7) }) + .await + .unwrap(); + + // Expired (zero TTL) and the refresh fails: the error surfaces. + entry + .get_or_fetch(Duration::ZERO, || async { + Err(EthBeaconNodeApiClientError::UnexpectedResponse) + }) + .await + .unwrap_err(); + + // The failure was not cached: the next call fetches again. + let value = entry + .get_or_fetch(Duration::ZERO, || async { Ok(8) }) + .await + .unwrap(); + assert_eq!(*value, 8); + } +} diff --git a/crates/eth2api/src/extensions.rs b/crates/eth2api/src/extensions.rs index 0130aae9..aa2f1f39 100644 --- a/crates/eth2api/src/extensions.rs +++ b/crates/eth2api/src/extensions.rs @@ -111,6 +111,57 @@ pub(crate) fn decode_fixed_hex String>( .map_err(|_| EthBeaconNodeApiClientError::ParseError(step())) } +/// Genesis response data parsed into typed values (once, at fetch time, so +/// malformed responses never enter the config cache). +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct GenesisInfo { + /// The genesis time. + pub time: DateTime, + /// The genesis fork version. + pub fork_version: phase0::Version, + /// The genesis validators root. + pub validators_root: phase0::Root, +} + +pub(crate) fn parse_genesis( + genesis: &GetGenesisResponseResponseData, +) -> Result { + let (fork_version, validators_root) = parse_genesis_fork_version_and_validators_root(genesis)?; + + Ok(GenesisInfo { + time: genesis_time_from_data(genesis)?, + fork_version, + validators_root, + }) +} + +/// Parses fork-schedule entries, preserving server order (which +/// [`fork_version_from_schedule`] relies on). Empty schedules are rejected +/// before they can enter the config cache; no fork version resolves from one. +pub(crate) fn parse_fork_schedule( + entries: &[BeaconStateFork], +) -> Result, EthBeaconNodeApiClientError> { + if entries.is_empty() { + return Err(EthBeaconNodeApiClientError::ParseError( + "empty fork schedule".to_string(), + )); + } + + entries + .iter() + .map(|fork| { + Ok(ForkSchedule { + version: decode_fixed_hex(&fork.current_version, || { + "decode fork schedule current_version".to_string() + })?, + epoch: fork.epoch.parse::().map_err(|_| { + EthBeaconNodeApiClientError::ParseError("parse fork schedule epoch".to_string()) + })?, + }) + }) + .collect() +} + fn parse_genesis_fork_version_and_validators_root( genesis_data: &GetGenesisResponseResponseData, ) -> Result<(phase0::Version, phase0::Root), EthBeaconNodeApiClientError> { @@ -124,7 +175,7 @@ fn parse_genesis_fork_version_and_validators_root( Ok((fork_version, validators_root)) } -fn fork_schedule_from_spec( +pub(crate) fn fork_schedule_from_spec( spec_data: &serde_json::Value, ) -> Result, EthBeaconNodeApiClientError> { fn fetch_fork( @@ -236,7 +287,7 @@ pub fn resolve_fork_version( /// static fork schedule unchanged), and cross-client signature verification /// only works when both sides derive the fork version the same way. fn fork_version_from_schedule( - schedule: &[BeaconStateFork], + schedule: &[ForkSchedule], epoch: phase0::Epoch, ) -> Result { let mut current = schedule.first().ok_or_else(|| { @@ -244,18 +295,13 @@ fn fork_version_from_schedule( })?; for fork in schedule { - let fork_epoch = fork.epoch.parse::().map_err(|_| { - EthBeaconNodeApiClientError::ParseError("parse fork schedule epoch".to_string()) - })?; - if fork_epoch > epoch { + if fork.epoch > epoch { break; } current = fork; } - decode_fixed_hex(¤t.current_version, || { - "decode fork schedule current_version".to_string() - }) + Ok(current.version) } /// Returns the fork version for voluntary-exit domains: EIP-7044 pins them to @@ -271,6 +317,32 @@ fn voluntary_exit_fork_version( .unwrap_or(genesis_fork_version)) } +/// The single domain derivation, over already-fetched config: non-exit +/// domains resolve the fork version from the fork-schedule entries (see +/// [`fork_version_from_schedule`]); voluntary exits stay pinned to Capella +/// per EIP-7044. +pub(crate) fn domain_from_config( + spec: &serde_json::Value, + genesis: &GenesisInfo, + fork_schedule: &[ForkSchedule], + domain_type: phase0::DomainType, + epoch: phase0::Epoch, +) -> Result { + let voluntary_exit_domain_type = resolve_domain_type(spec, "DOMAIN_VOLUNTARY_EXIT")?; + + let fork_version = if domain_type == voluntary_exit_domain_type { + voluntary_exit_fork_version(spec, genesis.fork_version)? + } else { + fork_version_from_schedule(fork_schedule, epoch)? + }; + + Ok(compute_domain( + domain_type, + fork_version, + genesis.validators_root, + )) +} + impl ValidatorStatus { /// Returns true if the validator is in one of the active states. pub fn is_active(&self) -> bool { @@ -283,15 +355,46 @@ impl ValidatorStatus { } } +/// Parses the genesis time out of a genesis response. +pub(crate) fn genesis_time_from_data( + genesis: &GetGenesisResponseResponseData, +) -> Result, EthBeaconNodeApiClientError> { + genesis + .genesis_time + .parse() + .map_err(|_| EthBeaconNodeApiClientError::ParseError("parse genesis_time".into())) + .and_then(|timestamp| { + DateTime::from_timestamp(timestamp, 0).ok_or_else(|| { + EthBeaconNodeApiClientError::ParseError("convert genesis_time to timestamp".into()) + }) + }) +} + +/// Parses the slot duration and slots per epoch out of the chain spec. +pub(crate) fn slots_config_from_spec( + spec: &serde_json::Value, +) -> Result<(time::Duration, u64), EthBeaconNodeApiClientError> { + let slot_duration = time::Duration::from_secs(parse_u64_field(spec, "SECONDS_PER_SLOT")?); + let slots_per_epoch = parse_u64_field(spec, "SLOTS_PER_EPOCH")?; + + if slot_duration == time::Duration::ZERO || slots_per_epoch == 0 { + return Err(EthBeaconNodeApiClientError::ZeroSlotDurationOrSlotsPerEpoch); + } + + Ok((slot_duration, slots_per_epoch)) +} + impl EthBeaconNodeApiClient { - async fn fetch_spec_data(&self) -> Result { + pub(crate) async fn fetch_spec_data( + &self, + ) -> Result { match self.get_spec(GetSpecRequest {}).await? { GetSpecResponse::Ok(spec) => Ok(spec.data), _ => Err(EthBeaconNodeApiClientError::UnexpectedResponse), } } - async fn fetch_genesis_data( + pub(crate) async fn fetch_genesis_data( &self, ) -> Result { match self.get_genesis(GetGenesisRequest {}).await? { @@ -303,26 +406,7 @@ impl EthBeaconNodeApiClient { /// Fetches the genesis time. pub async fn fetch_genesis_time(&self) -> Result, EthBeaconNodeApiClientError> { let genesis = self.fetch_genesis_data().await?; - - genesis - .genesis_time - .parse() - .map_err(|_| EthBeaconNodeApiClientError::ParseError("parse genesis_time".into())) - .and_then(|timestamp| { - DateTime::from_timestamp(timestamp, 0).ok_or_else(|| { - EthBeaconNodeApiClientError::ParseError( - "convert genesis_time to timestamp".into(), - ) - }) - }) - } - - /// Fetches the raw chain spec as a JSON object. - pub async fn fetch_spec(&self) -> Result { - match self.get_spec(GetSpecRequest {}).await? { - GetSpecResponse::Ok(resp) => Ok(resp.data), - _ => Err(EthBeaconNodeApiClientError::UnexpectedResponse), - } + genesis_time_from_data(&genesis) } /// Fetches the slot duration and slots per epoch. @@ -330,15 +414,7 @@ impl EthBeaconNodeApiClient { &self, ) -> Result<(time::Duration, u64), EthBeaconNodeApiClientError> { let spec = self.fetch_spec_data().await?; - - let slot_duration = time::Duration::from_secs(parse_u64_field(&spec, "SECONDS_PER_SLOT")?); - let slots_per_epoch = parse_u64_field(&spec, "SLOTS_PER_EPOCH")?; - - if slot_duration == time::Duration::ZERO || slots_per_epoch == 0 { - return Err(EthBeaconNodeApiClientError::ZeroSlotDurationOrSlotsPerEpoch); - } - - Ok((slot_duration, slots_per_epoch)) + slots_config_from_spec(&spec) } /// Fetches the fork schedule for all known forks. @@ -349,52 +425,8 @@ impl EthBeaconNodeApiClient { fork_schedule_from_spec(&spec) } - /// Fetches the domain type with the provided config/spec key. - pub async fn fetch_domain_type( - &self, - spec_key: &str, - ) -> Result { - let spec = self.fetch_spec_data().await?; - resolve_domain_type(&spec, spec_key) - } - - /// Fetches the genesis domain for the provided domain type. - pub async fn fetch_genesis_domain( - &self, - domain_type: phase0::DomainType, - ) -> Result { - let genesis = self.fetch_genesis_data().await?; - let (genesis_fork_version, _) = parse_genesis_fork_version_and_validators_root(&genesis)?; - - Ok(compute_domain( - domain_type, - genesis_fork_version, - phase0::Root::default(), - )) - } - - /// Fetches the genesis validators root from the beacon node. - pub async fn fetch_genesis_validators_root( - &self, - ) -> Result { - let genesis = self.fetch_genesis_data().await?; - let (_, validators_root) = parse_genesis_fork_version_and_validators_root(&genesis)?; - - Ok(validators_root) - } - - /// Fetches the genesis fork version from the beacon node. - pub async fn fetch_genesis_fork_version( - &self, - ) -> Result { - let genesis = self.fetch_genesis_data().await?; - let (fork_version, _) = parse_genesis_fork_version_and_validators_root(&genesis)?; - - Ok(fork_version) - } - /// Fetches the fork schedule entries from `/eth/v1/config/fork_schedule`. - async fn fetch_fork_schedule_data( + pub(crate) async fn fetch_fork_schedule_data( &self, ) -> Result, EthBeaconNodeApiClientError> { match self.get_fork_schedule(GetForkScheduleRequest {}).await? { @@ -403,55 +435,6 @@ impl EthBeaconNodeApiClient { } } - /// Fetches the `current_version` of every entry in the beacon node's fork - /// schedule (`/eth/v1/config/fork_schedule`), decoded and returned in the - /// order provided by the endpoint (oldest-to-newest per spec). The first - /// entry is the genesis fork version, which identifies the beacon node's - /// network. - pub async fn fetch_fork_schedule_versions( - &self, - ) -> Result, EthBeaconNodeApiClientError> { - self.fetch_fork_schedule_data() - .await? - .iter() - .map(|fork| { - decode_fixed_hex(&fork.current_version, || { - "decode fork schedule current_version".to_string() - }) - }) - .collect() - } - - /// Fetches the resolved beacon domain for the provided domain type and - /// epoch. Non-exit domains resolve the fork version from the - /// fork-schedule endpoint (go-eth2-client parity, see - /// [`fork_version_from_schedule`]); voluntary exits stay pinned to the - /// Capella fork per EIP-7044. - pub async fn fetch_domain( - &self, - domain_type: phase0::DomainType, - epoch: phase0::Epoch, - ) -> Result { - let spec = self.fetch_spec_data().await?; - let genesis = self.fetch_genesis_data().await?; - let (genesis_fork_version, genesis_validators_root) = - parse_genesis_fork_version_and_validators_root(&genesis)?; - let voluntary_exit_domain_type = resolve_domain_type(&spec, "DOMAIN_VOLUNTARY_EXIT")?; - - let fork_version = if domain_type == voluntary_exit_domain_type { - voluntary_exit_fork_version(&spec, genesis_fork_version)? - } else { - let schedule = self.fetch_fork_schedule_data().await?; - fork_version_from_schedule(&schedule, epoch)? - }; - - Ok(compute_domain( - domain_type, - fork_version, - genesis_validators_root, - )) - } - /// Subscribes to the beacon node SSE stream (`GET /eth/v1/events`) for the /// given topics. /// @@ -623,7 +606,7 @@ mod tests { #[test] fn fork_version_from_schedule_picks_last_activated_entry() { - let schedule = schedule_fixture(); + let schedule = parse_fork_schedule(&schedule_fixture()).unwrap(); // Same-epoch ties resolve to the last listed entry (server order). assert_eq!( diff --git a/crates/eth2api/src/lib.rs b/crates/eth2api/src/lib.rs index ee14e70a..709a4bf4 100644 --- a/crates/eth2api/src/lib.rs +++ b/crates/eth2api/src/lib.rs @@ -36,6 +36,10 @@ pub mod v1; /// Versioned wrappers for signeddata-related payloads. pub mod versioned; +/// Cache of static chain configuration retrieved from the Beacon node. +/// Internal: exposed through [`BeaconNodeClient`]'s cached config methods. +mod confcache; + /// Cache of Validators retrieved from the Beacon node. pub mod valcache; diff --git a/crates/eth2api/src/validator_duty.rs b/crates/eth2api/src/validator_duty.rs index c29b4ae4..b528440b 100644 --- a/crates/eth2api/src/validator_duty.rs +++ b/crates/eth2api/src/validator_duty.rs @@ -70,21 +70,6 @@ impl EthBeaconNodeApiClient { } } - /// Fetches the beacon attester signing domain. - pub async fn fetch_beacon_attester_domain( - &self, - epoch: phase0::Epoch, - ) -> Result { - let domain_type = self - .fetch_domain_type("DOMAIN_BEACON_ATTESTER") - .await - .map_err(error_message)?; - - self.fetch_domain(domain_type, epoch) - .await - .map_err(error_message) - } - /// Submits signed attestations to the beacon node. pub async fn submit_attestations( &self, diff --git a/crates/eth2util/src/eth2exp.rs b/crates/eth2util/src/eth2exp.rs index e9d18b6a..323287b9 100644 --- a/crates/eth2util/src/eth2exp.rs +++ b/crates/eth2util/src/eth2exp.rs @@ -1,9 +1,7 @@ //! Aggregator selection for attestation and sync committee duties. use k256::sha2::{Digest, Sha256}; -use pluto_eth2api::{ - EthBeaconNodeApiClient, EthBeaconNodeApiClientError, spec::phase0::BLSSignature, -}; +use pluto_eth2api::{BeaconNodeClient, EthBeaconNodeApiClientError, spec::phase0::BLSSignature}; /// Error type for aggregator selection operations. #[derive(Debug, thiserror::Error)] @@ -47,11 +45,11 @@ pub enum Eth2ExpError { /// Returns true if the validator is the attestation aggregator for the given /// committee. Refer: pub async fn is_att_aggregator( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, comm_len: u64, slot_sig: BLSSignature, ) -> Result { - let spec = client.fetch_spec().await?; + let spec = client.spec().await?; let aggs_per_comm = spec .as_object() @@ -71,10 +69,10 @@ pub async fn is_att_aggregator( /// Returns true if the validator is the aggregator for the provided sync /// subcommittee. Refer: pub async fn is_sync_comm_aggregator( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, sig: BLSSignature, ) -> Result { - let spec = client.fetch_spec().await?; + let spec = client.spec().await?; let comm_size = spec .as_object() @@ -152,7 +150,7 @@ mod tests { #[tokio::test] async fn is_att_aggregator() { let mock = default_client().await; - let client = mock.client(); + let client = &mock.beacon_client(); // comm_len=3, TARGET_AGGREGATORS_PER_COMMITTEE=16 → modulo=max(3/16,1)=1 → // always true assert!( @@ -165,7 +163,7 @@ mod tests { #[tokio::test] async fn is_not_att_aggregator() { let mock = default_client().await; - let client = mock.client(); + let client = &mock.beacon_client(); // comm_len=64, TARGET_AGGREGATORS_PER_COMMITTEE=16 → modulo=4 → false assert!( !super::is_att_aggregator(client, 64, decode_sig(ATT_SIG_HEX)) @@ -187,7 +185,7 @@ mod tests { #[tokio::test] async fn is_sync_comm_aggregator(sig_hex: &str, expected: bool) { let mock = default_client().await; - let client = mock.client(); + let client = &mock.beacon_client(); let result = super::is_sync_comm_aggregator(client, decode_sig(sig_hex)) .await .unwrap(); diff --git a/crates/eth2util/src/helpers.rs b/crates/eth2util/src/helpers.rs index 9519f7a2..40e4fe0d 100644 --- a/crates/eth2util/src/helpers.rs +++ b/crates/eth2util/src/helpers.rs @@ -157,12 +157,9 @@ pub fn slot_from_timestamp( } /// Returns epoch calculated from given slot. -pub async fn epoch_from_slot( - client: &pluto_eth2api::client::EthBeaconNodeApiClient, - slot: u64, -) -> Result { +pub async fn epoch_from_slot(client: &pluto_eth2api::BeaconNodeClient, slot: u64) -> Result { let (_, slots_per_epoch) = client - .fetch_slots_config() + .slots_config() .await .map_err(|e| HelperError::GettingSpec(e.to_string()))?; diff --git a/crates/eth2util/src/signing.rs b/crates/eth2util/src/signing.rs index 006aa553..06596e73 100644 --- a/crates/eth2util/src/signing.rs +++ b/crates/eth2util/src/signing.rs @@ -4,7 +4,7 @@ use pluto_crypto::{ types::{PublicKey, Signature}, }; use pluto_eth2api::{ - EthBeaconNodeApiClient, EthBeaconNodeApiClientError, + BeaconNodeClient, EthBeaconNodeApiClientError, spec::phase0::{Domain, Epoch, Root, SigningData}, versioned::VersionedSignedAggregateAndProof, }; @@ -103,23 +103,23 @@ pub(crate) fn compute_signing_root(message_root: Root, domain: Domain) -> Root { /// Returns the beacon domain for the provided type. pub async fn get_domain( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, name: DomainName, epoch: Epoch, ) -> Result { - let domain_type = client.fetch_domain_type(name.as_spec_key()).await?; + let domain_type = client.domain_type(name.as_spec_key()).await?; if name == DomainName::ApplicationBuilder { - return Ok(client.fetch_genesis_domain(domain_type).await?); + return Ok(client.genesis_domain(domain_type).await?); } - Ok(client.fetch_domain(domain_type, epoch).await?) + Ok(client.domain(domain_type, epoch).await?) } /// Wraps the message root with the resolved domain and returns the signing-data /// root. pub async fn get_data_root( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, name: DomainName, epoch: Epoch, root: Root, @@ -132,7 +132,7 @@ pub async fn get_data_root( /// Verifies a signature against the resolved eth2 domain signing root. pub async fn verify( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, domain_name: DomainName, epoch: Epoch, message_root: Root, @@ -173,7 +173,7 @@ pub fn verify_with_domain( /// Verifies the selection proof embedded in an aggregate-and-proof payload. pub async fn verify_aggregate_and_proof_selection( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, pubkey: &PublicKey, agg: &VersionedSignedAggregateAndProof, ) -> Result<()> { @@ -282,9 +282,9 @@ mod tests { #[tokio::test] async fn get_domain_matches_builder_vector() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); - let domain = get_domain(client, DomainName::ApplicationBuilder, 1_000) + let domain = get_domain(&client, DomainName::ApplicationBuilder, 1_000) .await .unwrap(); @@ -297,9 +297,9 @@ mod tests { #[tokio::test] async fn get_domain_uses_capella_for_voluntary_exit() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); - let domain = get_domain(client, DomainName::VoluntaryExit, 1_000) + let domain = get_domain(&client, DomainName::VoluntaryExit, 1_000) .await .unwrap(); @@ -312,7 +312,7 @@ mod tests { #[tokio::test] async fn get_data_root_matches_registration_vector() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let fee_recipient: ExecutionAddress = hex::decode("000000000000000000000000000000000000dead") @@ -335,7 +335,7 @@ mod tests { }; let signing_root = get_data_root( - client, + &client, DomainName::ApplicationBuilder, 0, message.message_root(), @@ -352,7 +352,7 @@ mod tests { #[tokio::test] async fn verify_accepts_valid_signature() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let pubkey = BlstImpl.secret_to_public_key(&secret).unwrap(); @@ -369,13 +369,13 @@ mod tests { pubkey, }; let message_root = message.message_root(); - let signing_root = get_data_root(client, DomainName::ApplicationBuilder, 0, message_root) + let signing_root = get_data_root(&client, DomainName::ApplicationBuilder, 0, message_root) .await .unwrap(); let signature = BlstImpl.sign(&secret, &signing_root).unwrap(); verify( - client, + &client, DomainName::ApplicationBuilder, 0, message_root, @@ -389,10 +389,10 @@ mod tests { #[tokio::test] async fn verify_rejects_zero_signature() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let pubkey = [0x11; 48]; let err = verify( - client, + &client, DomainName::ApplicationBuilder, 0, [0x22; 32], @@ -408,12 +408,12 @@ mod tests { #[tokio::test] async fn verify_with_domain_accepts_valid_signature() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let pubkey = BlstImpl.secret_to_public_key(&secret).unwrap(); let message_root = [0x55; 32]; - let domain = get_domain(client, DomainName::ApplicationBuilder, 0) + let domain = get_domain(&client, DomainName::ApplicationBuilder, 0) .await .unwrap(); let signing_root = compute_signing_root(message_root, domain); @@ -425,8 +425,8 @@ mod tests { #[tokio::test] async fn verify_with_domain_rejects_zero_signature() { let mock = mock_beacon_client().await; - let client = mock.client(); - let domain = get_domain(client, DomainName::ApplicationBuilder, 0) + let client = mock.beacon_client(); + let domain = get_domain(&client, DomainName::ApplicationBuilder, 0) .await .unwrap(); let pubkey = [0x11; 48]; @@ -439,20 +439,20 @@ mod tests { #[tokio::test] async fn verify_rejects_wrong_pubkey() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let wrong_secret = secret_key("01477d4bfbbcebe1fef8d4d6f624ecbb6e3178558bb1b0d6286c816c66842a6d"); let pubkey = BlstImpl.secret_to_public_key(&wrong_secret).unwrap(); let message_root = [0x55; 32]; - let signing_root = get_data_root(client, DomainName::ApplicationBuilder, 0, message_root) + let signing_root = get_data_root(&client, DomainName::ApplicationBuilder, 0, message_root) .await .unwrap(); let signature = BlstImpl.sign(&secret, &signing_root).unwrap(); let err = verify( - client, + &client, DomainName::ApplicationBuilder, 0, message_root, @@ -468,14 +468,14 @@ mod tests { #[tokio::test] async fn verify_rejects_wrong_message_root() { let mock = mock_beacon_client().await; - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let pubkey = BlstImpl.secret_to_public_key(&secret).unwrap(); let signed_message_root = [0x55; 32]; let verified_message_root = [0x66; 32]; let signing_root = get_data_root( - client, + &client, DomainName::ApplicationBuilder, 0, signed_message_root, @@ -485,7 +485,7 @@ mod tests { let signature = BlstImpl.sign(&secret, &signing_root).unwrap(); let err = verify( - client, + &client, DomainName::ApplicationBuilder, 0, verified_message_root, diff --git a/crates/parsigex/src/behaviour.rs b/crates/parsigex/src/behaviour.rs index 8dc6806c..b497ef0a 100644 --- a/crates/parsigex/src/behaviour.rs +++ b/crates/parsigex/src/behaviour.rs @@ -28,7 +28,7 @@ use pluto_core::{ types::{Duty, ParSignedData, ParSignedDataSet, PubKey}, }; use pluto_crypto::types::PublicKey; -use pluto_eth2api::EthBeaconNodeApiClient; +use pluto_eth2api::BeaconNodeClient; use pluto_p2p::p2p_context::P2PContext; use super::{Handler, encode_message}; @@ -60,7 +60,7 @@ pub type Verifier = /// /// Ports Charon's `parsigex.NewEth2Verifier` pub fn new_eth2_verifier( - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, pub_shares_by_key: HashMap>, ) -> Verifier { let pub_shares_by_key = Arc::new(pub_shares_by_key); @@ -597,7 +597,7 @@ mod eth2_verifier_tests { tbls::Tbls, types::{Index, PrivateKey, PublicKey}, }; - use pluto_eth2api::{EthBeaconNodeApiClient, spec::phase0}; + use pluto_eth2api::{BeaconNodeClient, spec::phase0}; use pluto_eth2util::signing::{DomainName, get_data_root}; use pluto_testutil::BeaconMock; @@ -637,7 +637,7 @@ mod eth2_verifier_tests { /// Signs the eth2 signing root of `data` for the given domain/epoch with /// `secret`, returning a copy of `data` carrying that signature. async fn sign( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, secret: &PrivateKey, data: &T, domain: DomainName, @@ -674,7 +674,7 @@ mod eth2_verifier_tests { #[tokio::test] async fn accepts_partial_signature_against_correct_share() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let group_pubkey = PubKey::new(BlstImpl.secret_to_public_key(&secret).unwrap()); @@ -684,7 +684,7 @@ mod eth2_verifier_tests { let share_idx: Index = 2; let att = sample_attestation(4); let signed = sign( - client, + &client, &shares[&share_idx], &att, DomainName::BeaconAttester, @@ -705,7 +705,7 @@ mod eth2_verifier_tests { #[tokio::test] async fn rejects_partial_signature_against_wrong_share() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let group_pubkey = PubKey::new(BlstImpl.secret_to_public_key(&secret).unwrap()); @@ -714,7 +714,7 @@ mod eth2_verifier_tests { // Sign with share 2's secret but claim share index 3, so the verifier // looks up share 3's public key and the signature fails to verify. let att = sample_attestation(4); - let signed = sign(client, &shares[&2], &att, DomainName::BeaconAttester, 4).await; + let signed = sign(&client, &shares[&2], &att, DomainName::BeaconAttester, 4).await; let par = ParSignedData::new(signed, 3); let mut pub_shares_by_key = HashMap::new(); @@ -731,14 +731,14 @@ mod eth2_verifier_tests { #[tokio::test] async fn rejects_unknown_pubkey() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let group_pubkey = PubKey::new(BlstImpl.secret_to_public_key(&secret).unwrap()); let (shares, _pub_shares) = split_shares(&secret); let att = sample_attestation(4); - let signed = sign(client, &shares[&1], &att, DomainName::BeaconAttester, 4).await; + let signed = sign(&client, &shares[&1], &att, DomainName::BeaconAttester, 4).await; let par = ParSignedData::new(signed, 1); // Empty map: the validator public key is not part of the cluster lock. @@ -755,14 +755,14 @@ mod eth2_verifier_tests { #[tokio::test] async fn rejects_missing_share_index() { let mock = BeaconMock::builder().build().await.unwrap(); - let client = mock.client(); + let client = mock.beacon_client(); let secret = secret_key("345768c0245f1dc702df9e50e811002f61ebb2680b3d5931527ef59f96cbaf9b"); let group_pubkey = PubKey::new(BlstImpl.secret_to_public_key(&secret).unwrap()); let (shares, pub_shares) = split_shares(&secret); let att = sample_attestation(4); - let signed = sign(client, &shares[&1], &att, DomainName::BeaconAttester, 4).await; + let signed = sign(&client, &shares[&1], &att, DomainName::BeaconAttester, 4).await; // Claim a share index that was never produced by the split. let par = ParSignedData::new(signed, TOTAL_SHARES + 1); diff --git a/crates/testutil/src/beaconmock/mod.rs b/crates/testutil/src/beaconmock/mod.rs index 3f127956..41294c70 100644 --- a/crates/testutil/src/beaconmock/mod.rs +++ b/crates/testutil/src/beaconmock/mod.rs @@ -47,6 +47,7 @@ pub type Result = std::result::Result; pub struct BeaconMock { server: MockServer, client: EthBeaconNodeApiClient, + beacon_client: pluto_eth2api::BeaconNodeClient, state: Arc, // Held to keep the slot ticker alive; dropped with `BeaconMock`. _head_producer: HeadProducer, @@ -165,10 +166,12 @@ impl BeaconMock { } let client = EthBeaconNodeApiClient::with_base_url(server.uri()).map_err(Error::Client)?; + let beacon_client = pluto_eth2api::BeaconNodeClient::new(client.clone()); Ok(Self { server, client, + beacon_client, state, _head_producer: head_producer, }) @@ -180,6 +183,13 @@ impl BeaconMock { &self.client } + /// Returns the beacon node client over this mock's client. Clones share + /// one cache, so repeated calls do not shard cached state. + #[must_use] + pub fn beacon_client(&self) -> pluto_eth2api::BeaconNodeClient { + self.beacon_client.clone() + } + /// Returns the backing mock server for mounting test-specific endpoints. #[must_use] pub fn server(&self) -> &MockServer { diff --git a/crates/testutil/src/validatormock/attest.rs b/crates/testutil/src/validatormock/attest.rs index 472a64a8..6e083f56 100644 --- a/crates/testutil/src/validatormock/attest.rs +++ b/crates/testutil/src/validatormock/attest.rs @@ -32,7 +32,7 @@ use std::{collections::HashMap, sync::Arc}; use pluto_eth2api::{ - ConsensusVersion, ETH_CONSENSUS_VERSION, EthBeaconNodeApiClient, EthBeaconNodeApiClientError, + BeaconNodeClient, ConsensusVersion, ETH_CONSENSUS_VERSION, EthBeaconNodeApiClientError, GetAggregatedAttestationV2Request, GetAggregatedAttestationV2Response, GetAttesterDutiesRequest, GetAttesterDutiesResponse, ProduceAttestationDataRequest, ProduceAttestationDataResponse, SubmitBeaconCommitteeSelectionsRequest, @@ -103,7 +103,7 @@ pub struct BeaconCommitteeSelection { /// `OnceCell`s (one per stage) acting as Go's `chan struct{}` ready signals. #[derive(Debug, Clone)] pub struct SlotAttester { - eth2_cl: Arc, + eth2_cl: BeaconNodeClient, slot: Slot, #[allow(dead_code)] // matched against duties via the active-validator map pubkeys: Vec, @@ -129,7 +129,7 @@ impl SlotAttester { /// and safe to share between the scheduler tasks. #[must_use] pub fn new( - eth2_cl: Arc, + eth2_cl: BeaconNodeClient, slot: Slot, sign_func: SignFunc, pubkeys: Vec, @@ -161,7 +161,7 @@ impl SlotAttester { /// already-closed channel only triggering an explicit panic; here we /// prefer idempotence. pub async fn prepare(&self) -> Result<()> { - let vals = super::validators::active_validators(&self.eth2_cl).await?; + let vals = super::validators::active_validators(self.eth2_cl.api()).await?; let duties = prepare_attesters(&self.eth2_cl, &vals, self.slot).await?; self.set_prepare_duties(vals, duties.clone()).await; @@ -245,7 +245,7 @@ impl SlotAttester { // --------------------------------------------------------------------------- async fn prepare_attesters( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, vals: &ActiveValidators, slot: Slot, ) -> Result> { @@ -264,6 +264,7 @@ async fn prepare_attesters( .map_err(EthBeaconNodeApiClientError::RequestError)?; let response = eth2_cl + .api() .get_attester_duties(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -317,7 +318,7 @@ fn parse_duty( // --------------------------------------------------------------------------- async fn prepare_aggregators( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, sign_func: &SignFunc, _state: &Arc>, duties: &[AttesterDuty], @@ -353,6 +354,7 @@ async fn prepare_aggregators( .map_err(EthBeaconNodeApiClientError::RequestError)?; let response = eth2_cl + .api() .submit_beacon_committee_selections(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -392,7 +394,7 @@ async fn prepare_aggregators( // --------------------------------------------------------------------------- async fn attest( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, sign_func: &SignFunc, slot: Slot, duties: &[AttesterDuty], @@ -429,6 +431,7 @@ async fn attest( .map_err(EthBeaconNodeApiClientError::RequestError)?; let response = eth2_cl + .api() .produce_attestation_data(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -477,7 +480,7 @@ async fn attest( // --------------------------------------------------------------------------- async fn aggregate( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, sign_func: &SignFunc, slot: Slot, vals: &ActiveValidators, @@ -543,7 +546,7 @@ async fn aggregate( } async fn get_aggregate_attestation( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, datas: &[AttestationData], comm_idx: CommitteeIndex, ) -> Result { @@ -561,6 +564,7 @@ async fn get_aggregate_attestation( .map_err(EthBeaconNodeApiClientError::RequestError)?; let response = eth2_cl + .api() .get_aggregated_attestation_v2(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -600,7 +604,7 @@ async fn get_aggregate_attestation( const ERROR_BODY_TRUNCATE: usize = 1024; async fn submit_attestations( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, atts: &[electra::SingleAttestation], ) -> Result<()> { const ENDPOINT: &str = "/eth/v2/beacon/pool/attestations"; @@ -608,7 +612,7 @@ async fn submit_attestations( } async fn submit_aggregate_attestations( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, aggs: &[electra::SignedAggregateAndProof], ) -> Result<()> { const ENDPOINT: &str = "/eth/v2/validator/aggregate_and_proofs"; @@ -618,11 +622,11 @@ async fn submit_aggregate_attestations( } async fn submit_json( - eth2_cl: &EthBeaconNodeApiClient, + eth2_cl: &BeaconNodeClient, endpoint: &'static str, body: &T, ) -> Result<()> { - let mut url = eth2_cl.base_url.clone(); + let mut url = eth2_cl.api().base_url.clone(); { let mut segments = url.path_segments_mut().map_err(|()| { Error::Malformed(format!("base url has no path segments for {endpoint}")) @@ -634,6 +638,7 @@ async fn submit_json( } let response = eth2_cl + .api() .client .post(url) // The v2 pool/aggregate submit endpoints require the consensus-version @@ -788,12 +793,7 @@ mod tests { .expect("fetch slots config"); let sign_func: SignFunc = Arc::new(PubkeyEchoSigner); - let attester = SlotAttester::new( - Arc::new(mock.client().clone()), - slots_per_epoch, - sign_func, - pubkeys, - ); + let attester = SlotAttester::new(mock.beacon_client(), slots_per_epoch, sign_func, pubkeys); attester.prepare().await.expect("prepare"); attester.attest().await.expect("attest"); diff --git a/crates/testutil/src/validatormock/component.rs b/crates/testutil/src/validatormock/component.rs index e8b1fa85..0a129f25 100644 --- a/crates/testutil/src/validatormock/component.rs +++ b/crates/testutil/src/validatormock/component.rs @@ -18,7 +18,7 @@ use std::{ }; use pluto_core::types::DutyType; -use pluto_eth2api::{EthBeaconNodeApiClient, spec::phase0::BLSPubKey}; +use pluto_eth2api::{BeaconNodeClient, spec::phase0::BLSPubKey}; use tokio::{ sync::{Mutex, mpsc}, task::JoinHandle, @@ -61,7 +61,7 @@ pub struct Component { } struct Inner { - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, sign_func: SignFunc, pubkeys: Vec, meta: SpecMeta, @@ -96,7 +96,7 @@ impl Component { /// `clock` defaults to [`SystemClock`] when omitted. #[builder] pub fn new( - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, sign_func: SignFunc, pubkeys: Vec, meta: SpecMeta, @@ -213,7 +213,7 @@ impl Component { async fn start_attesters(&self, epoch: MetaEpoch) { for slot in epoch.slots() { let attester = Arc::new(SlotAttester::new( - Arc::new(self.inner.eth2_cl.clone()), + self.inner.eth2_cl.clone(), slot.slot, Arc::clone(&self.inner.sign_func), self.inner.pubkeys.clone(), @@ -548,7 +548,7 @@ mod tests { .await; let component = Component::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .sign_func(Signer::arc(&[]).expect("empty signer")) .pubkeys(Vec::new()) .meta(meta_at(genesis)) @@ -616,7 +616,7 @@ mod tests { let clock = FakeClock::new(genesis); let meta = meta_at(genesis); let component = Component::builder() - .eth2_cl(mock.client().clone()) + .eth2_cl(mock.beacon_client()) .sign_func(Signer::arc(&[]).expect("empty signer")) .pubkeys(Vec::new()) .meta(meta) diff --git a/crates/testutil/src/validatormock/propose.rs b/crates/testutil/src/validatormock/propose.rs index 429db21e..f2251e9e 100644 --- a/crates/testutil/src/validatormock/propose.rs +++ b/crates/testutil/src/validatormock/propose.rs @@ -14,16 +14,16 @@ //! Fulu range — and their blinded variants — is implemented in full. use pluto_eth2api::{ - BlockRequestBody, BlockRequestBodyObject, BlockRequestBodyObject2, BlockRequestBodyObject3, - BlockRequestBodyObject4, BlockRequestBodyObject5, ConsensusVersion, - DenebSignedBlockContentsSignedBlock, EthBeaconNodeApiClient, - GetBlindedBlockResponseResponseData, GetBlindedBlockResponseResponseDataObject, - GetBlindedBlockResponseResponseDataObject2, GetBlindedBlockResponseResponseDataObject3, - GetBlindedBlockResponseResponseDataObject4, GetProposerDutiesRequest, - GetProposerDutiesResponse, ProduceBlockV3Request, ProduceBlockV3Response, - ProduceBlockV3ResponseResponse, PublishBlindedBlockV2Request, PublishBlockV2Request, - PublishBlockV2Response, RegisterValidatorRequest, RegisterValidatorRequestBodyItem, - RegisterValidatorResponse, SignedBlockContentsSignedBlock, SignedValidatorRegistrationMessage, + BeaconNodeClient, BlockRequestBody, BlockRequestBodyObject, BlockRequestBodyObject2, + BlockRequestBodyObject3, BlockRequestBodyObject4, BlockRequestBodyObject5, ConsensusVersion, + DenebSignedBlockContentsSignedBlock, GetBlindedBlockResponseResponseData, + GetBlindedBlockResponseResponseDataObject, GetBlindedBlockResponseResponseDataObject2, + GetBlindedBlockResponseResponseDataObject3, GetBlindedBlockResponseResponseDataObject4, + GetProposerDutiesRequest, GetProposerDutiesResponse, ProduceBlockV3Request, + ProduceBlockV3Response, ProduceBlockV3ResponseResponse, PublishBlindedBlockV2Request, + PublishBlockV2Request, PublishBlockV2Response, RegisterValidatorRequest, + RegisterValidatorRequestBodyItem, RegisterValidatorResponse, SignedBlockContentsSignedBlock, + SignedValidatorRegistrationMessage, spec::{ BuilderVersion, bellatrix, capella, deneb, electra, phase0::{BLSPubKey, BLSSignature, Root, Slot}, @@ -59,14 +59,14 @@ pub type VersionedValidatorRegistration = VersionedSignedValidatorRegistration; /// [`super::sign`]; in production it wraps real BLS secrets, in tests a stub /// that copies the pubkey bytes into the signature suffices. pub async fn propose_block( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, signer: &super::SignFunc, slot: Slot, ) -> Result<()> { // Ensure active validators are queryable. Mirrors Go's // `eth2Cl.ActiveValidators` call: surfaces beacon-node errors before duty // lookups proceed. - let _ = active_validators(client).await?; + let _ = active_validators(client.api()).await?; let epoch = epoch_from_slot(client, slot).await?; @@ -75,7 +75,7 @@ pub async fn propose_block( .build() .map_err(|err| Error::Malformed(format!("build proposer duties request: {err}")))?; - let duties = match client.get_proposer_duties(request).await { + let duties = match client.api().get_proposer_duties(request).await { Ok(GetProposerDutiesResponse::Ok(resp)) => resp.data, Ok(_) => return Err(Error::Malformed("proposer duties response".to_string())), Err(err) => return Err(Error::Malformed(format!("proposer duties: {err}"))), @@ -107,7 +107,7 @@ pub async fn propose_block( .build() .map_err(|err| Error::Malformed(format!("build produce-block request: {err}")))?; - let proposal_resp = match client.produce_block_v3(proposal_request).await { + let proposal_resp = match client.api().produce_block_v3(proposal_request).await { Ok(ProduceBlockV3Response::Ok(resp)) => resp, Ok(_) => { return Err(Error::Malformed( @@ -132,7 +132,7 @@ pub async fn propose_block( .build() .map_err(|err| Error::Malformed(format!("build blinded-publish request: {err}")))?; - match client.publish_blinded_block_v2(request).await { + match client.api().publish_blinded_block_v2(request).await { Ok(PublishBlockV2Response::Ok | PublishBlockV2Response::Accepted) => Ok(()), Ok(_) => Err(Error::Malformed( "publish-blinded-block-v2 unexpected response".to_string(), @@ -147,7 +147,7 @@ pub async fn propose_block( .build() .map_err(|err| Error::Malformed(format!("build publish-block request: {err}")))?; - match client.publish_block_v2(request).await { + match client.api().publish_block_v2(request).await { Ok(PublishBlockV2Response::Ok | PublishBlockV2Response::Accepted) => Ok(()), Ok(_) => Err(Error::Malformed( "publish-block-v2 unexpected response".to_string(), @@ -167,7 +167,7 @@ pub async fn propose_block( /// any non-V1 variant lands here we surface [`Error::UnsupportedVariant`] /// instead of mis-tagging the signed payload. pub async fn register( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, signer: &super::SignFunc, registration: &VersionedValidatorRegistration, pubshare: BLSPubKey, @@ -200,7 +200,7 @@ pub async fn register( .build() .map_err(|err| Error::Malformed(format!("build register request: {err}")))?; - match client.register_validator(request).await { + match client.api().register_validator(request).await { Ok(RegisterValidatorResponse::Ok) => Ok(()), Ok(_) => Err(Error::Malformed( "register-validator unexpected response".to_string(), @@ -216,7 +216,7 @@ async fn build_block_body( resp: &ProduceBlockV3ResponseResponse, pubkey: &BLSPubKey, signer: &super::SignFunc, - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, epoch: u64, ) -> Result { let block_value = serde_json::to_value(&resp.data) @@ -294,7 +294,7 @@ async fn build_blinded_body( resp: &ProduceBlockV3ResponseResponse, pubkey: &BLSPubKey, signer: &super::SignFunc, - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, epoch: u64, ) -> Result { let block_value = serde_json::to_value(&resp.data) @@ -356,7 +356,7 @@ async fn build_blinded_body( async fn sign_with_proposer( signer: &super::SignFunc, pubkey: &BLSPubKey, - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, epoch: u64, message_root: Root, ) -> Result { @@ -684,7 +684,7 @@ mod tests { ) .await; - propose_block(mock.client(), &stub_signer(), slot) + propose_block(&mock.beacon_client(), &stub_signer(), slot) .await .expect("propose_block"); @@ -738,7 +738,7 @@ mod tests { ) .await; - propose_block(mock.client(), &stub_signer(), slot) + propose_block(&mock.beacon_client(), &stub_signer(), slot) .await .expect("propose_block blinded"); @@ -801,7 +801,7 @@ mod tests { ) .await; - propose_block(mock.client(), &stub_signer(), slot) + propose_block(&mock.beacon_client(), &stub_signer(), slot) .await .expect("propose_block fulu"); @@ -826,7 +826,7 @@ mod tests { .mount(mock.server()) .await; - propose_block(mock.client(), &stub_signer(), slot) + propose_block(&mock.beacon_client(), &stub_signer(), slot) .await .expect("propose_block must be a no-op when not the slot proposer"); } @@ -857,7 +857,7 @@ mod tests { ) .await; - register(mock.client(), &stub_signer(), ®istration, pubkey) + register(&mock.beacon_client(), &stub_signer(), ®istration, pubkey) .await .expect("register"); diff --git a/crates/testutil/src/validatormock/synccomm.rs b/crates/testutil/src/validatormock/synccomm.rs index 85d5b733..4cce06a4 100644 --- a/crates/testutil/src/validatormock/synccomm.rs +++ b/crates/testutil/src/validatormock/synccomm.rs @@ -23,7 +23,7 @@ use std::{ }; use pluto_eth2api::{ - EthBeaconNodeApiClient, EthBeaconNodeApiClientError, GetBlockRootRequest, GetBlockRootResponse, + BeaconNodeClient, EthBeaconNodeApiClientError, GetBlockRootRequest, GetBlockRootResponse, GetSyncCommitteeDutiesRequest, GetSyncCommitteeDutiesResponse, GetSyncCommitteeDutiesResponseResponseDatum, PrepareSyncCommitteeSubnetsRequest, ProduceSyncCommitteeContributionRequest, ProduceSyncCommitteeContributionResponse, @@ -92,7 +92,7 @@ struct Mutable { /// [`SyncCommMember::aggregate`]. pub struct SyncCommMember { // Immutable state. - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, epoch: Epoch, #[allow(dead_code)] pubkeys: Vec, @@ -108,7 +108,7 @@ impl SyncCommMember { /// `NewSyncCommMember`. #[must_use] pub fn new( - eth2_cl: EthBeaconNodeApiClient, + eth2_cl: BeaconNodeClient, epoch: Epoch, sign_func: SignFunc, pubkeys: Vec, @@ -219,7 +219,7 @@ impl SyncCommMember { /// Resolves sync committee duties for this epoch and submits subscriptions /// covering the next epoch. pub async fn prepare_epoch(&self) -> Result<()> { - let vals = active_validators(&self.eth2_cl).await?; + let vals = active_validators(self.eth2_cl.api()).await?; let duties = prepare_sync_comm_duties(&self.eth2_cl, &vals, self.epoch).await?; self.set_duties(vals, duties.clone()); subscribe_sync_comm_subnets(&self.eth2_cl, self.epoch, &duties).await?; @@ -282,7 +282,7 @@ impl SyncCommMember { // -- helper functions (mirror the lowercase Go helpers). -- async fn prepare_sync_comm_duties( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, vals: &ActiveValidators, epoch: Epoch, ) -> Result> { @@ -298,6 +298,7 @@ async fn prepare_sync_comm_duties( .map_err(|e| Error::Malformed(format!("build sync committee duties request: {e}")))?; let response = client + .api() .get_sync_committee_duties(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -349,7 +350,7 @@ fn parse_pubkey(s: &str) -> Result { } async fn subscribe_sync_comm_subnets( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, epoch: Epoch, duties: &[SyncCommitteeDuty], ) -> Result<()> { @@ -379,6 +380,7 @@ async fn subscribe_sync_comm_subnets( .map_err(|e| Error::Malformed(format!("build sync committee subscriptions: {e}")))?; client + .api() .prepare_sync_committee_subnets(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -389,7 +391,7 @@ async fn subscribe_sync_comm_subnets( } async fn prepare_sync_selections( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, sign_func: &SignFunc, duties: &[SyncCommitteeDuty], slot: Slot, @@ -434,6 +436,7 @@ async fn prepare_sync_selections( .map_err(|e| Error::Malformed(format!("build sync committee selections: {e}")))?; let response = client + .api() .submit_sync_committee_selections(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -517,10 +520,10 @@ fn hex_0x(bytes: impl AsRef<[u8]>) -> String { /// `getSubcommittees`: `idx / (SYNC_COMMITTEE_SIZE / /// SYNC_COMMITTEE_SUBNET_COUNT)`. pub(crate) async fn get_subcommittees( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, duty: &SyncCommitteeDuty, ) -> Result> { - let spec = client.fetch_spec().await.map_err(Error::BeaconNode)?; + let spec = client.spec().await.map_err(Error::BeaconNode)?; let comm_size = spec_u64(&spec, "SYNC_COMMITTEE_SIZE")?; let subnet_count = spec_u64(&spec, "SYNC_COMMITTEE_SUBNET_COUNT")?; @@ -554,13 +557,14 @@ fn spec_u64(spec: &serde_json::Value, field: &str) -> Result { .map_err(|_| Error::Malformed(format!("parse spec field {field}"))) } -async fn fetch_head_block_root(client: &EthBeaconNodeApiClient) -> Result { +async fn fetch_head_block_root(client: &BeaconNodeClient) -> Result { let request = GetBlockRootRequest::builder() .block_id("head".to_string()) .build() .map_err(|e| Error::Malformed(format!("build block root request: {e}")))?; let response = client + .api() .get_block_root(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -575,7 +579,7 @@ async fn fetch_head_block_root(client: &EthBeaconNodeApiClient) -> Result } async fn submit_sync_messages( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, slot: Slot, block_root: Root, sign_func: &SignFunc, @@ -613,6 +617,7 @@ async fn submit_sync_messages( .map_err(|e| Error::Malformed(format!("build sync committee messages: {e}")))?; client + .api() .submit_pool_sync_committee_signatures(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -623,7 +628,7 @@ async fn submit_sync_messages( } async fn agg_contributions( - client: &EthBeaconNodeApiClient, + client: &BeaconNodeClient, sign_func: &SignFunc, slot: Slot, vals: &ActiveValidators, @@ -648,6 +653,7 @@ async fn agg_contributions( .map_err(|e| Error::Malformed(format!("build produce contribution: {e}")))?; let response = client + .api() .produce_sync_committee_contribution(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -715,6 +721,7 @@ async fn agg_contributions( .map_err(|e| Error::Malformed(format!("build contribution and proofs request: {e}")))?; client + .api() .publish_contribution_and_proofs(request) .await .map_err(EthBeaconNodeApiClientError::RequestError)?; @@ -764,7 +771,7 @@ mod tests { validator_sync_committee_indices: vec![75, 133, 289, 491], }; - let subcommittees = get_subcommittees(mock.client(), &duty) + let subcommittees = get_subcommittees(&mock.beacon_client(), &duty) .await .expect("get_subcommittees");