diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 35db1fbc..4f273960 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -382,10 +382,8 @@ 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()); // 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); // ---- Beacon-derived duty-workflow inputs ---- @@ -558,9 +556,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { WireInputs { threshold, share_idx, - beacon_client, eth2_cl, - submission_client, + submission_api, 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 1be20c37..e0b3fe63 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -241,12 +241,10 @@ pub struct WireInputs { pub threshold: u64, /// This node's 1-indexed share index. pub share_idx: u64, - /// Beacon node client used for scheduling. - pub beacon_client: BeaconNodeClient, - /// Beacon node API client used for fetching / dutydb / validatorapi. + /// Beacon node API client for everything except broadcasting. pub eth2_cl: EthBeaconNodeApiClient, - /// Submission beacon node client used for broadcasting. - pub submission_client: BeaconNodeClient, + /// Beacon node API client for broadcasting, built with the submit timeout. + pub submission_api: EthBeaconNodeApiClient, /// Per-validator data for this node. pub validators: Vec, /// Current consensus implementation, from the controller. Forwards to the @@ -410,9 +408,8 @@ pub async fn wire_core_workflow( let WireInputs { threshold, share_idx, - beacon_client, eth2_cl, - submission_client, + submission_api, validators, consensus, builder_enabled, @@ -431,34 +428,28 @@ 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()), - ); - let fee_recipient_fn: FeeRecipientFunc = { let map = fee_recipient_by_pubkey.clone(); Arc::new(move |pubkey: &PubKey| map.get(pubkey).copied().unwrap_or_default()) }; + // ---- Beacon node clients ---- + // Both clients, the per-epoch refresher and the validator API share one + // validator cache, so a single refresh serves every consumer. + 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()); + let submission_client = BeaconNodeClient::new(submission_api, validator_cache.clone()); + // ---- Deadliners (one per component) ---- // // Each component gets its own deadliner task sharing the injected diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 41d24820..ac2be463 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -46,8 +46,7 @@ use pluto_core::{ }; use pluto_crypto::tbls; use pluto_eth2api::{ - BeaconNodeClient, EthBeaconNodeApiClient, GetStateValidatorsResponseResponse, - GetStateValidatorsResponseResponseDatum, + EthBeaconNodeApiClient, spec::{altair, phase0}, versioned::{self, AttestationPayload, SignedProposalBlock, VersionedAttestation}, }; @@ -55,8 +54,8 @@ use pluto_testutil::BeaconMock; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use wiremock::{ - Mock, MockServer, Request, ResponseTemplate, - matchers::{method, path, path_regex}, + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, }; const PK_LEN: usize = 48; @@ -118,46 +117,7 @@ async fn wait_for_post(server: &MockServer, submit_path: &'static str) -> usize } }) .await - .unwrap_or_else(|_| panic!("submit endpoint {submit_path} should be hit")) -} - -/// Builds a `/states/{id}/validators` datum for an active validator with the -/// given index and pubkey. -fn validator_datum(index: u64, pubkey: PubKey) -> GetStateValidatorsResponseResponseDatum { - let v = pluto_testutil::Validator::active(index, pubkey_to_eth2(pubkey)); - GetStateValidatorsResponseResponseDatum { - index: v.index.to_string(), - balance: v.balance.to_string(), - status: v.status, - validator: v.validator, - } -} - -/// Mounts POST `/eth/v1/beacon/states/{state_id}/validators` returning ONLY the -/// datums whose pubkey appears in the request-body `ids` — so an unseeded -/// (empty-pubkey) cache resolves zero validators. Cover both `head` and slot -/// state IDs because the scheduler refreshes the cache by slot immediately. -async fn mount_filtered_post_validators( - server: &MockServer, - datums: Vec, -) { - Mock::given(method("POST")) - .and(path_regex(r"^/eth/v1/beacon/states/[^/]+/validators$")) - .respond_with(move |request: &Request| { - let body = String::from_utf8_lossy(&request.body); - let data: Vec<_> = datums - .iter() - .filter(|d| body.contains(&d.validator.pubkey)) - .cloned() - .collect(); - ResponseTemplate::new(200).set_body_json(GetStateValidatorsResponseResponse { - execution_optimistic: false, - finalized: true, - data, - }) - }) - .mount(server) - .await; + .unwrap_or_else(|_| panic!("POST {submit_path} should be hit")) } /// Counts POSTs the mock has received for `submit_path`. @@ -208,7 +168,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 +176,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 +184,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, @@ -246,15 +197,14 @@ fn wire_inputs_with( }]; // 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()); + // submission API must point at the mock too. + let submission_api = eth2_cl.clone(); WireInputs { threshold, share_idx: 1, - beacon_client, eth2_cl, - submission_client, + submission_api, validators, consensus, builder_enabled: false, @@ -329,16 +279,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") @@ -432,7 +378,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); @@ -440,7 +385,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 @@ -555,12 +500,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 @@ -628,12 +572,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 @@ -716,7 +659,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 @@ -732,14 +674,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 @@ -828,46 +763,41 @@ 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 -/// 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. +/// (d) `wire_core_workflow` seeds the shared validator cache with the cluster +/// pubkeys. The scheduler resolves the current slot on start, so its validators +/// request must carry those pubkeys in `ids`; an unseeded cache sends an empty +/// `ids` and resolves zero validators. #[tokio::test] -async fn wiring_seeds_shared_validator_cache() { +async fn wiring_seeds_validator_cache() { let ct = CancellationToken::new(); let mock = BeaconMock::builder().build().await.expect("beacon mock"); let pubkey = PubKey::new([9u8; PK_LEN]); - const V_IDX: u64 = 7; - 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); - let submission_probe = inputs.submission_client.clone(); + let _wired = tokio::time::timeout( + GUARD, + wire_core_workflow(wire_inputs(eth2_cl, pubkey, consensus, 1), ct.clone()), + ) + .await + .expect("wire did not deadlock") + .expect("wire succeeded"); - let _wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone())) + const VALIDATORS: &str = "/eth/v1/beacon/states/head/validators"; + wait_for_post(mock.server(), VALIDATORS).await; + let bodies: Vec<_> = mock + .server() + .received_requests() .await - .expect("wire did not deadlock") - .expect("wire succeeded"); - - for (name, probe) in [("beacon", beacon_probe), ("submission", submission_probe)] { - let active = tokio::time::timeout(GUARD, probe.active_validators()) - .await - .unwrap_or_else(|_| panic!("(d) {name} client active_validators timed out")) - .unwrap_or_else(|e| panic!("(d) {name} client active_validators failed: {e}")); - assert_eq!( - active.get(&V_IDX), - Some(&pubkey_to_eth2(pubkey)), - "(d) the {name} client's cache should be seeded with the cluster pubkeys" - ); - } + .expect("requests") + .iter() + .filter(|r| r.method.as_str() == "POST" && r.url.path() == VALIDATORS) + .map(|r| String::from_utf8_lossy(&r.body).into_owned()) + .collect(); + assert!( + bodies.iter().all(|body| body.contains(&pubkey.to_string())), + "(d) validators requests should carry the cluster pubkeys, got: {bodies:?}" + ); ct.cancel(); } @@ -886,7 +816,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); @@ -899,7 +828,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())) @@ -984,9 +913,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 3946884f..31d3af6f 100644 --- a/crates/core/src/bcast/mod.rs +++ b/crates/core/src/bcast/mod.rs @@ -871,11 +871,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 { @@ -893,9 +891,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) } @@ -1193,9 +1194,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 dfcd6ea5..52816a78 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -460,8 +460,8 @@ impl SchedulerActor { // 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); @@ -1152,11 +1152,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, @@ -1224,7 +1234,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 @@ -1266,7 +1276,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"); @@ -1279,7 +1289,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"); @@ -1292,7 +1302,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();