From a90b27b06ca95a5f5fe86b9eab3ee838b0fd3a9c Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Tue, 25 Aug 2026 16:51:54 +0200 Subject: [PATCH] refactor(eth2api): make ValidatorCache immutable in BeaconNodeClient The validator set is known at node construction (from the cluster validators), so the cache no longer needs to be optional, mutable, or shared behind a lock. Build a single `ValidatorCache` in `node::run` and thread it into both the scheduler and submission `BeaconNodeClient`s at construction. - `BeaconNodeClient::new` now takes the `ValidatorCache` and stores it as a plain field (the type is already `Arc`-backed, so clones share state). - Remove the `Arc>>` wrapping, the `set_validator_cache` setter, and the `NoActiveValidatorCache` error variant. - `validator_cache()` returns `&ValidatorCache`; drops the `.expect`/TODO at the scheduler read site. Closes #482 Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/node/mod.rs | 17 +++++++- crates/app/src/node/wire.rs | 24 +++++------ crates/app/tests/wiring.rs | 70 ++++++++++++------------------- crates/core/src/bcast/mod.rs | 26 +++++++----- crates/core/src/scheduler.rs | 24 +++++++---- crates/eth2api/src/beacon_node.rs | 37 +++++++--------- 6 files changed, 97 insertions(+), 101 deletions(-) diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 9c134c84..f342b504 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -378,10 +378,22 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { verify_fork_schedule(ð2_cl, &lock.fork_version).await?; } - let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone()); + // One pubkey-scoped validator cache shared by the scheduler's beacon + // client, the submission client, and the validator API, so every consumer + // resolves the same cluster validator set. `ValidatorCache` is `Arc`-backed, + // so the clones seeded into each client (and the one wired into the + // per-epoch refresh subscriber in `wire_core_workflow`) share state, letting + // a single refresh update every consumer at once. + let eth2_pubkeys: Vec<_> = validators.iter().map(|v| v.eth2_pubkey).collect(); + let validator_cache = + pluto_eth2api::valcache::ValidatorCache::new(eth2_cl.clone(), eth2_pubkeys); + + let beacon_client = + pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone(), validator_cache.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); + let submission_client = + pluto_eth2api::BeaconNodeClient::new(submission_api, validator_cache.clone()); // ---- Beacon-derived duty-workflow inputs ---- @@ -556,6 +568,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { beacon_client, eth2_cl, submission_client, + validator_cache, validators, consensus: consensus_controller.current_consensus(), builder_enabled: config.builder_api, diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 40dab2ff..85777397 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -258,6 +258,11 @@ pub struct WireInputs { pub eth2_cl: EthBeaconNodeApiClient, /// Submission beacon node client used for broadcasting. pub submission_client: BeaconNodeClient, + /// Pubkey-scoped validator cache shared by the beacon/submission clients + /// and the validator API. A clone of the same `Arc`-backed cache seeded + /// into those clients, so the per-epoch trim + refresh subscriber wired + /// below refreshes every consumer at once. + pub validator_cache: ValidatorCache, /// Per-validator data for this node. pub validators: Vec, /// Current consensus implementation, from the controller. Forwards to the @@ -424,6 +429,7 @@ pub async fn wire_core_workflow( beacon_client, eth2_cl, submission_client, + validator_cache, validators, consensus, builder_enabled, @@ -442,29 +448,19 @@ pub async fn wire_core_workflow( } = inputs; // ---- Derived validator maps ---- - let mut eth2_pubkeys = Vec::with_capacity(validators.len()); // DV root pubkey -> this node's public share (validatorapi wants this flat // map already collapsed for our share index). let mut pub_share_by_pubkey: HashMap = HashMap::new(); let mut fee_recipient_by_pubkey: HashMap = HashMap::new(); for val in &validators { - eth2_pubkeys.push(val.eth2_pubkey); pub_share_by_pubkey.insert(val.eth2_pubkey, val.pubshare); fee_recipient_by_pubkey.insert(val.pubkey, val.fee_recipient); } - // One pubkey-scoped validator cache shared by the scheduler's beacon - // client, the submission client, and the validator API, so every consumer - // resolves the same cluster validator set. Without seeding, the scheduler - // 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); - tokio::join!( - beacon_client.set_validator_cache(validator_cache.clone()), - submission_client.set_validator_cache(validator_cache.clone()), - ); - + // The pubkey-scoped validator cache is built and seeded into the + // beacon/submission clients at construction (in `node::run`), and passed in + // here so the per-epoch trim + refresh subscriber registered below (and the + // validator API) share the same `Arc`-backed state. let fee_recipient_fn: FeeRecipientFunc = { let map = fee_recipient_by_pubkey.clone(); Arc::new(move |pubkey: &PubKey| map.get(pubkey).copied().unwrap_or_default()) diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 6d8aa63c..0d68ec41 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -49,6 +49,7 @@ use pluto_eth2api::{ BeaconNodeClient, EthBeaconNodeApiClient, GetStateValidatorsResponseResponse, GetStateValidatorsResponseResponseDatum, spec::{altair, phase0}, + valcache::ValidatorCache, versioned::{self, AttestationPayload, SignedProposalBlock, VersionedAttestation}, }; use pluto_testutil::BeaconMock; @@ -208,7 +209,6 @@ fn attester_partial(share_idx: u64, share: &pluto_crypto::types::PrivateKey) -> /// path connects, not that BLS verification works). fn wire_inputs( eth2_cl: EthBeaconNodeApiClient, - beacon_client: BeaconNodeClient, pubkey: PubKey, consensus: Arc, threshold: u64, @@ -217,14 +217,7 @@ fn wire_inputs( // eth2 verification is deliberately bypassed here. The bad-partial-signature // test injects the real verifier. let permissive_verifier: VerifyFn = Arc::new(|_pubkey, _data| Box::pin(async { Ok(()) })); - wire_inputs_with( - eth2_cl, - beacon_client, - pubkey, - consensus, - threshold, - permissive_verifier, - ) + wire_inputs_with(eth2_cl, pubkey, consensus, threshold, permissive_verifier) } /// Builds the wiring inputs for a single-validator cluster with a caller-chosen @@ -232,7 +225,6 @@ fn wire_inputs( /// verifier parses and verifies the reconstructed group signature against). fn wire_inputs_with( eth2_cl: EthBeaconNodeApiClient, - beacon_client: BeaconNodeClient, pubkey: PubKey, consensus: Arc, threshold: u64, @@ -245,9 +237,14 @@ fn wire_inputs_with( fee_recipient: [0u8; 20], }]; + // One shared, pubkey-scoped cache seeded into both clients at construction, + // mirroring production wiring (`node::run`). + let eth2_pubkeys = validators.iter().map(|v| v.eth2_pubkey).collect(); + let validator_cache = ValidatorCache::new(eth2_cl.clone(), eth2_pubkeys); + let beacon_client = BeaconNodeClient::new(eth2_cl.clone(), validator_cache.clone()); // The broadcaster's constructor performs beacon-node calls, so the // submission client must point at the mock too. - let submission_client = BeaconNodeClient::new(eth2_cl.clone()); + let submission_client = BeaconNodeClient::new(eth2_cl.clone(), validator_cache.clone()); WireInputs { threshold, @@ -255,6 +252,7 @@ fn wire_inputs_with( beacon_client, eth2_cl, submission_client, + validator_cache, validators, consensus, builder_enabled: false, @@ -329,16 +327,12 @@ async fn wiring_exercises_fetcher_back_edges() { let ct = CancellationToken::new(); let mock = BeaconMock::builder().build().await.expect("beacon mock"); let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let pubkey = PubKey::new([2u8; PK_LEN]); let consensus = build_consensus(&ct); let wired = tokio::time::timeout( GUARD, - wire_core_workflow( - wire_inputs(eth2_cl, beacon_client, pubkey, consensus, 1), - ct.clone(), - ), + wire_core_workflow(wire_inputs(eth2_cl, pubkey, consensus, 1), ct.clone()), ) .await .expect("wire did not deadlock") @@ -428,7 +422,6 @@ async fn wiring_connects_sign_path() { let mock = BeaconMock::builder().build().await.expect("beacon mock"); mount_attestation_submit(mock.server()).await; let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let pubkey = PubKey::new([5u8; PK_LEN]); let consensus = build_consensus(&ct); @@ -436,7 +429,7 @@ async fn wiring_connects_sign_path() { // partial signatures (distinct share indices) cross the threshold and are // aggregated by SigAgg. const THRESHOLD: u64 = 2; - let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD); + let inputs = wire_inputs(eth2_cl, pubkey, consensus, THRESHOLD); let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) .await @@ -551,12 +544,11 @@ async fn wiring_connects_sign_path_proposer() { let mock = BeaconMock::builder().build().await.expect("beacon mock"); mount_submit(mock.server(), "/eth/v2/beacon/blocks").await; let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let pubkey = PubKey::new([6u8; PK_LEN]); let consensus = build_consensus(&ct); const THRESHOLD: u64 = 2; - let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD); + let inputs = wire_inputs(eth2_cl, pubkey, consensus, THRESHOLD); let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) .await @@ -624,12 +616,11 @@ async fn wiring_connects_sign_path_sync_contribution() { let mock = BeaconMock::builder().build().await.expect("beacon mock"); mount_submit(mock.server(), "/eth/v1/validator/contribution_and_proofs").await; let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let pubkey = PubKey::new([8u8; PK_LEN]); let consensus = build_consensus(&ct); const THRESHOLD: u64 = 2; - let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD); + let inputs = wire_inputs(eth2_cl, pubkey, consensus, THRESHOLD); let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) .await @@ -712,7 +703,6 @@ async fn wiring_rejects_bad_partial_signature() { let mock = BeaconMock::builder().build().await.expect("beacon mock"); mount_attestation_submit(mock.server()).await; let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let consensus = build_consensus(&ct); // Real BLS group key: the verifier parses this pubkey and verifies the @@ -728,14 +718,7 @@ async fn wiring_rejects_bad_partial_signature() { let verifier: VerifyFn = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone())); const THRESHOLD: u64 = 2; - let inputs = wire_inputs_with( - eth2_cl, - beacon_client, - pubkey, - consensus, - THRESHOLD, - verifier, - ); + let inputs = wire_inputs_with(eth2_cl, pubkey, consensus, THRESHOLD, verifier); let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) .await @@ -824,12 +807,12 @@ async fn wiring_rejects_bad_partial_signature() { ct.cancel(); } -/// (d) `wire_core_workflow` seeds one pubkey-scoped validator cache into the -/// scheduler's beacon client and the submission client (Charon shares a single +/// (d) One pubkey-scoped validator cache is seeded into the scheduler's beacon +/// client and the submission client at construction (Charon shares a single /// cache across both; the validator API reuses the same instance). The mock's /// POST validators endpoint returns only validators whose pubkey appears in the -/// request-body `ids`, so the unseeded (empty-pubkey) default cache would -/// resolve zero validators — the regression this test guards against. +/// request-body `ids`, so an unseeded (empty-pubkey) cache would resolve zero +/// validators — the regression this test guards against. #[tokio::test] async fn wiring_seeds_shared_validator_cache() { let ct = CancellationToken::new(); @@ -839,13 +822,14 @@ async fn wiring_seeds_shared_validator_cache() { mount_filtered_post_validators(mock.server(), vec![validator_datum(V_IDX, pubkey)]).await; let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let consensus = build_consensus(&ct); - // `BeaconNodeClient` clones share the cache slot, so the seeding performed - // inside `wire_core_workflow` is observable through these probes. - let beacon_probe = beacon_client.clone(); - let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, 1); + // The clients are constructed with the shared, pubkey-seeded cache inside + // `wire_inputs` (mirroring production `node::run`). `BeaconNodeClient` clones + // share the same `Arc`-backed cache, so the seeded pubkeys are observable + // through these probes. + let inputs = wire_inputs(eth2_cl, pubkey, consensus, 1); + let beacon_probe = inputs.beacon_client.clone(); let submission_probe = inputs.submission_client.clone(); let _wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) @@ -882,7 +866,6 @@ async fn wiring_delivers_slot_ticks_to_subscriber() { .expect("beacon mock"); let pubkey = PubKey::new([9u8; PK_LEN]); let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let consensus = build_consensus(&ct); let (tx, mut rx) = tokio::sync::mpsc::channel::(8); @@ -895,7 +878,7 @@ async fn wiring_delivers_slot_ticks_to_subscriber() { }) }); - let mut inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, 1); + let mut inputs = wire_inputs(eth2_cl, pubkey, consensus, 1); inputs.slot_tick = Some(slot_tick); let _wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) @@ -980,9 +963,8 @@ async fn multinode_parsig_exchange_reaches_submission() { let mut nodes = Vec::with_capacity(N); for i in 0..N { let eth2_cl = mock.client().clone(); - let beacon_client = BeaconNodeClient::new(eth2_cl.clone()); let consensus = build_consensus(&ct); - let mut inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD); + let mut inputs = wire_inputs(eth2_cl, pubkey, consensus, THRESHOLD); inputs.parsigex = routed_parsigex_seam(i, Arc::clone(&receivers)); let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) .await diff --git a/crates/core/src/bcast/mod.rs b/crates/core/src/bcast/mod.rs index 99a3f53e..8e341359 100644 --- a/crates/core/src/bcast/mod.rs +++ b/crates/core/src/bcast/mod.rs @@ -868,11 +868,9 @@ mod tests { .mount(beacon.server()) .await; - let client = BeaconNodeClient::new(beacon.client().clone()); - client - .set_validator_cache(ValidatorCache::new(beacon.client().clone(), vec![])) - .await; - client + let api = beacon.client().clone(); + let cache = ValidatorCache::new(api.clone(), vec![]); + BeaconNodeClient::new(api, cache) } fn pubkey(byte: u8) -> PubKey { @@ -890,9 +888,12 @@ mod tests { async fn new_broadcaster() -> (BeaconMock, Broadcaster) { let beacon = BeaconMock::builder().build().await.expect("beacon mock"); mount_submit_successes(beacon.server()).await; - let broadcaster = Broadcaster::new(BeaconNodeClient::new(beacon.client().clone())) - .await - .expect("broadcaster"); + let broadcaster = Broadcaster::new(BeaconNodeClient::new( + beacon.client().clone(), + ValidatorCache::new(beacon.client().clone(), vec![]), + )) + .await + .expect("broadcaster"); (beacon, broadcaster) } @@ -1190,9 +1191,12 @@ mod tests { async fn broadcast_attester_submits_and_swallows_prior_known() { let beacon = BeaconMock::builder().build().await.expect("beacon mock"); mount_prior_attestation_known(beacon.server()).await; - let broadcaster = Broadcaster::new(BeaconNodeClient::new(beacon.client().clone())) - .await - .expect("broadcaster"); + let broadcaster = Broadcaster::new(BeaconNodeClient::new( + beacon.client().clone(), + ValidatorCache::new(beacon.client().clone(), vec![]), + )) + .await + .expect("broadcaster"); let set = signed_set( pubkey(1), VersionedAttestation::new(deneb_attestation()).expect("attestation"), diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index d29b5abe..ab264160 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -458,8 +458,8 @@ impl SchedulerActor { // During this time the Scheduler actor is blocked. // This is the same behavior as in Charon, but it might not be desirable. - let valcache = self.client.validator_cache().await; - let vals = resolve_active_validators(slot.epoch(), &valcache).await?; + let valcache = self.client.validator_cache(); + let vals = resolve_active_validators(slot.epoch(), valcache).await?; SCHEDULER_METRICS.validators_active.set(vals.len() as u64); @@ -1149,11 +1149,21 @@ mod tests { .await; } + /// Builds a [`BeaconNodeClient`] over the mock with an empty-pubkey + /// validator cache. The mock returns its mounted validator datums + /// regardless of the request's `ids` filter, so an empty pubkey set is + /// sufficient for the scheduler tests. + fn test_beacon_client(mock: &BeaconMock) -> BeaconNodeClient { + let api = mock.client().clone(); + let cache = valcache::ValidatorCache::new(api.clone(), Vec::new()); + BeaconNodeClient::new(api, cache) + } + /// Builds an initial [`SchedulerActor`] wired to the mock's client. No /// epoch resolved yet. fn test_actor(mock: &BeaconMock) -> SchedulerActor { SchedulerActor { - client: pluto_eth2api::BeaconNodeClient::new(mock.client().clone()), + client: test_beacon_client(mock), slots_per_epoch: 1, slot_broadcast: sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0, duty_broadcast: sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0, @@ -1221,7 +1231,7 @@ mod tests { let slot_sub = slot_broadcast.subscribe(); let duty_sub = duty_broadcast.subscribe(); - let client = pluto_eth2api::BeaconNodeClient::new(mock.client().clone()); + let client = test_beacon_client(mock); // Cache slots_per_epoch from the mock's spec, mirroring `build`, so // `get_duty_definition`'s epoch math matches the slots the test drives. let (_slot_duration, slots_per_epoch) = client @@ -1263,7 +1273,7 @@ mod tests { let err = fetch_attester_duties( &test_past_slot(0, 1), validator_set_a_mismatched(), - &BeaconNodeClient::new(mock.client().clone()), + &test_beacon_client(&mock), ) .await .expect_err("mismatched pubkey should be rejected"); @@ -1276,7 +1286,7 @@ mod tests { let err = fetch_proposer_duties( &test_past_slot(0, 1), validator_set_a_mismatched(), - &BeaconNodeClient::new(mock.client().clone()), + &test_beacon_client(&mock), ) .await .expect_err("mismatched pubkey should be rejected"); @@ -1289,7 +1299,7 @@ mod tests { let err = fetch_sync_committee_duties( &test_past_slot(0, 1), validator_set_a_mismatched(), - &BeaconNodeClient::new(mock.client().clone()), + &test_beacon_client(&mock), ) .await .expect_err("mismatched pubkey should be rejected"); diff --git a/crates/eth2api/src/beacon_node.rs b/crates/eth2api/src/beacon_node.rs index 5bef37cb..b6286433 100644 --- a/crates/eth2api/src/beacon_node.rs +++ b/crates/eth2api/src/beacon_node.rs @@ -2,8 +2,6 @@ use crate::{ EthBeaconNodeApiClient, valcache::{ActiveValidators, CompleteValidators, ValidatorCache, ValidatorCacheError}, }; -use std::sync::Arc; -use tokio::sync::RwLock; type Result = std::result::Result; @@ -20,18 +18,18 @@ pub enum BeaconNodeClientError { #[derive(Clone)] pub struct BeaconNodeClient { api: EthBeaconNodeApiClient, - // 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>, + /// Pubkey-scoped validator cache, fixed at construction. [`ValidatorCache`] + /// is `Arc`-backed, so clones (including those held by other consumers) + /// share the same underlying cache state. + validator_cache: ValidatorCache, } impl BeaconNodeClient { - /// Creates a new beacon node client. - pub fn new(api: EthBeaconNodeApiClient) -> Self { + /// Creates a new beacon node client backed by the given validator cache. + pub fn new(api: EthBeaconNodeApiClient, validator_cache: ValidatorCache) -> Self { Self { - api: api.clone(), - validator_cache: Arc::new(RwLock::new(ValidatorCache::new(api, Vec::new()))), + api, + validator_cache, } } @@ -40,26 +38,21 @@ impl BeaconNodeClient { &self.api } - /// 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; - } - /// Returns active validators for `head`. pub async fn active_validators(&self) -> Result { - let (active, _) = self.validator_cache().await.get_by_head().await?; + let (active, _) = self.validator_cache.get_by_head().await?; Ok(active) } /// Returns complete validators for `head`. pub async fn complete_validators(&self) -> Result { - let (_, complete) = self.validator_cache().await.get_by_head().await?; + let (_, complete) = self.validator_cache.get_by_head().await?; Ok(complete) } /// Get the validator cache. - pub async fn validator_cache(&self) -> ValidatorCache { - self.validator_cache.read().await.clone() + pub fn validator_cache(&self) -> &ValidatorCache { + &self.validator_cache } } @@ -102,10 +95,8 @@ mod tests { .mount(&mock) .await; - let client = BeaconNodeClient::new(test_client(&mock)); - client - .set_validator_cache(ValidatorCache::new(client.api().clone(), pubkeys)) - .await; + let api = test_client(&mock); + let client = BeaconNodeClient::new(api.clone(), ValidatorCache::new(api, pubkeys)); let active = client.active_validators().await.unwrap(); let complete = client.complete_validators().await.unwrap();