refactor(eth2api): make ValidatorCache immutable in BeaconNodeClient

crates/app/src/node/mod.rs

@@ -378,10 +378,22 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
         verify_fork_schedule(&eth2_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,

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<ValidatorInfo>,
     /// 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<BLSPubKey, BLSPubKey> = HashMap::new();
     let mut fee_recipient_by_pubkey: HashMap<PubKey, ExecutionAddress> = 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())

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<ConsensusWrapper>,
     threshold: u64,
@@ -217,22 +217,14 @@ 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
 /// SigAgg `verifier`. The validator's group pubkey is `pubkey` (which the real
 /// verifier parses and verifies the reconstructed group signature against).
 fn wire_inputs_with(
     eth2_cl: EthBeaconNodeApiClient,
-    beacon_client: BeaconNodeClient,
     pubkey: PubKey,
     consensus: Arc<ConsensusWrapper>,
     threshold: u64,
@@ -245,16 +237,22 @@ 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,
         share_idx: 1,
         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,15 +422,14 @@ 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);
 
     // threshold = 2 (the BLS library rejects threshold <= 1). Two matching
     // 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::<u64>(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

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"),