From fe24902c2068d8ff172befb4b2a7422d9e72dbca Mon Sep 17 00:00:00 2001 From: "emlautarom1-agent[bot]" <292495798+emlautarom1-agent[bot]@users.noreply.github.com> Date: Wed, 15 Jul 2026 12:59:46 -0300 Subject: [PATCH 1/2] feat(app): trim and refresh the validator cache each epoch Add a scheduler slot subscriber that trims and refreshes the shared ValidatorCache on each epoch's first slot, the Rust analog of Charon's inline refresh loop (app/app.go:484-532). Without it the validator set is frozen at startup: validators activating later never get duties scheduled, and exited validators keep being scheduled. The single shared cache means one refresh covers the scheduler, broadcaster, and validator-API consumers. The new ValidatorCacheRefresher mirrors Charon's firstValCacheRefresh / refreshedBySlot bookkeeping: it skips mid-epoch slots once refreshed by slot, forces a refresh on the first tick and after any head fallback, and re-fetches the epoch's first slot (rather than the current slot) when the previous fetch fell back to head. Closes #529 --- crates/app/src/node/wire.rs | 393 +++++++++++++++++++++++++++++++++++- 1 file changed, 389 insertions(+), 4 deletions(-) diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 1f749919..a77f68c8 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -37,14 +37,14 @@ use pluto_core::{ scheduler::SchedulerBuilder, sigagg::{Aggregator, VerifyFn}, signeddata::{SyncContribution, VersionedAggregatedAttestation}, - types::{Duty, ParSignedData, ParSignedDataSet, PubKey, SignedData, SignedDataSet}, + types::{Duty, ParSignedData, ParSignedDataSet, PubKey, SignedData, SignedDataSet, Slot}, unsigneddata::{self, UnsignedDataSet}, validatorapi::{self, Component, Handler}, }; use pluto_eth2api::{ BeaconNodeClient, EthBeaconNodeApiClient, spec::{bellatrix::ExecutionAddress, phase0::BLSPubKey}, - valcache::ValidatorCache, + valcache::{ValidatorCache, ValidatorCacheError}, }; use tokio_util::sync::CancellationToken; @@ -169,6 +169,81 @@ pub struct WiredComponents { pub validator_api_router: axum::Router, } +/// Per-epoch trim + refresh for the shared [`ValidatorCache`], the Rust analog +/// of Charon's inline validator-cache refresh subscriber +/// (`app/app.go:484-532`). +/// +/// Registered as a scheduler slot subscriber by [`wire_core_workflow`]. The +/// seeded cache is otherwise frozen at startup; this refresher re-fetches the +/// cluster validator set on each epoch's first slot so validators activating +/// after startup get duties scheduled and exited validators stop being +/// resolved. +struct ValidatorCacheRefresher { + cache: ValidatorCache, + bookkeeping: tokio::sync::Mutex, +} + +/// Mirrors Charon's `firstValCacheRefresh` / `refreshedBySlot` locals +/// (`app.go:484-485`), which the refresh closure reads and writes under a lock. +struct RefreshBookkeeping { + /// Whether the cache has never been refreshed. Forces the first tick to + /// refresh regardless of the slot's position in the epoch. + first_refresh: bool, + /// Whether the previous refresh fetched by slot (`true`) or fell back to + /// the head state (`false`). A head fallback forces the next tick to + /// refresh and to re-fetch the epoch's first slot. + refreshed_by_slot: bool, +} + +impl ValidatorCacheRefresher { + fn new(cache: ValidatorCache) -> Self { + Self { + cache, + // Charon initializes both flags to `true` (`app.go:484-485`). + bookkeeping: tokio::sync::Mutex::new(RefreshBookkeeping { + first_refresh: true, + refreshed_by_slot: true, + }), + } + } + + /// Trims and refreshes the cache for `slot` when required, mirroring + /// Charon's `shouldUpdateCache` gate (`app.go:489-498`) and the `GetBySlot` + /// head-fallback re-fetch (`app.go:512-529`). + async fn refresh(&self, slot: &Slot) -> Result<(), ValidatorCacheError> { + let mut bk = self.bookkeeping.lock().await; + + // shouldUpdateCache: skip mid-epoch slots once the cache has been + // refreshed at least once by slot. + if !slot.first_in_epoch() && !bk.first_refresh && bk.refreshed_by_slot { + return Ok(()); + } + + tracing::info!( + slot = %slot.slot, + first_refresh = bk.first_refresh, + "Refreshing validator cache" + ); + + // If the previous refresh fell back to head, fetch the epoch's first + // slot rather than the current slot. `epoch * slots_per_epoch <= slot`, + // so the multiply never actually saturates. + let slot_to_fetch = if bk.refreshed_by_slot { + slot.slot.inner() + } else { + slot.epoch().saturating_mul(slot.slots_per_epoch) + }; + + self.cache.trim().await; + let (_, _, refreshed_by_slot) = self.cache.get_by_slot(slot_to_fetch).await?; + + bk.refreshed_by_slot = refreshed_by_slot; + bk.first_refresh = false; + + Ok(()) + } +} + /// Constructs and wires the ten core duty-workflow components. /// /// Reproduces the data-flow graph from `core/interfaces.go:337-357`. The 13 @@ -216,8 +291,9 @@ pub async fn wire_core_workflow( // resolves the same cluster validator set. Charon seeds a single cache // into both clients (app.go:481-482 and app.go:598); without seeding, the // scheduler would resolve duties against an empty (or unfiltered) set. - // `ValidatorCache` clones share state; the per-epoch trim + refresh - // subscriber is a planned follow-up. + // `ValidatorCache` clones share state, so the per-epoch trim + refresh + // subscriber registered below (Charon app.go:480-532) 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()), @@ -488,6 +564,24 @@ pub async fn wire_core_workflow( "consensus", ); } + // Slot subscriber: per-epoch validator cache trim + refresh (Charon + // app.go:480-532). The shared cache seeded above stays frozen at startup + // until this runs — it re-fetches the cluster validator set on each epoch's + // first slot so late-activating validators get scheduled and exited ones + // stop being resolved. A single refresh covers every consumer (scheduler, + // broadcaster, validator API) since they share this cache. + { + let refresher = Arc::new(ValidatorCacheRefresher::new(validator_cache.clone())); + sched_builder.subscribe_slot( + move |slot: &Slot| { + let refresher = Arc::clone(&refresher); + let slot = slot.clone(); + async move { refresher.refresh(&slot).await } + }, + "validator_cache", + ); + } + let (scheduler, scheduler_task) = sched_builder .build(beacon_client, ct.clone()) .await @@ -604,3 +698,294 @@ pub async fn wire_core_workflow( validator_api_router, }) } + +#[cfg(test)] +mod tests { + //! Unit tests for the per-epoch [`ValidatorCacheRefresher`] (Charon + //! `app.go:484-532`). They drive the refresher directly against a + //! `wiremock`-backed [`ValidatorCache`] — the same cache the scheduler + //! reads via `get_by_head` when resolving duties — and assert the + //! refreshed active set, so they mirror what the scheduler would + //! observe. + + use super::*; + use pluto_core::types::SlotNumber; + use pluto_eth2api::{ + BlindedBlock400Response, GetStateValidatorsResponseResponse, + GetStateValidatorsResponseResponseDatum, ValidatorResponseValidator, ValidatorStatus, + }; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + const FAR_FUTURE_EPOCH: &str = "18446744073709551615"; + + fn test_pubkey(seed: u8) -> BLSPubKey { + let mut bytes = [0u8; 48]; + bytes[0] = seed; + bytes + } + + fn format_pubkey(pubkey: &BLSPubKey) -> String { + format!("0x{}", hex::encode(pubkey)) + } + + fn test_datum( + index: u64, + pubkey: &BLSPubKey, + status: ValidatorStatus, + ) -> GetStateValidatorsResponseResponseDatum { + GetStateValidatorsResponseResponseDatum { + index: index.to_string(), + balance: "32000000000".to_string(), + status, + validator: ValidatorResponseValidator { + pubkey: format_pubkey(pubkey), + withdrawal_credentials: + "0x0000000000000000000000000000000000000000000000000000000000000000".to_string(), + effective_balance: "32000000000".to_string(), + slashed: false, + activation_eligibility_epoch: "0".to_string(), + activation_epoch: "0".to_string(), + exit_epoch: FAR_FUTURE_EPOCH.to_string(), + withdrawable_epoch: FAR_FUTURE_EPOCH.to_string(), + }, + } + } + + /// An unmounted `POST /states/{state_id}/validators` mock returning `data`. + fn post_validators_ok( + state_id: impl AsRef, + data: Vec, + ) -> Mock { + Mock::given(method("POST")) + .and(path(format!( + "/eth/v1/beacon/states/{}/validators", + state_id.as_ref() + ))) + .respond_with(ResponseTemplate::new(200).set_body_json( + GetStateValidatorsResponseResponse { + execution_optimistic: false, + finalized: true, + data, + }, + )) + } + + /// An unmounted `POST /states/{state_id}/validators` mock returning 404, so + /// `get_by_slot` falls back to the head state. + fn post_validators_not_found(state_id: impl AsRef) -> Mock { + Mock::given(method("POST")) + .and(path(format!( + "/eth/v1/beacon/states/{}/validators", + state_id.as_ref() + ))) + .respond_with( + ResponseTemplate::new(404).set_body_json(BlindedBlock400Response { + code: 404.0, + message: "State not found".to_string(), + stacktraces: None, + }), + ) + } + + fn test_cache(server: &MockServer, pubkeys: Vec) -> ValidatorCache { + let client = + EthBeaconNodeApiClient::with_base_url(server.uri()).expect("valid mock server URL"); + ValidatorCache::new(client, pubkeys) + } + + /// A [`Slot`]. Only `slot`/`slots_per_epoch` matter to the refresher; + /// `time` and `slot_duration` are placeholders. + fn test_slot(slot: u64, slots_per_epoch: u64) -> Slot { + Slot { + slot: SlotNumber::new(slot), + time: chrono::Utc::now(), + slot_duration: chrono::Duration::seconds(12), + slots_per_epoch, + } + } + + const SPE: u64 = 32; + + /// A validator that is inactive at startup and activates later is picked up + /// on the next epoch's first slot. + #[tokio::test] + async fn refresh_observes_validator_activated_after_startup() { + let pk = test_pubkey(5); + let mock = MockServer::start().await; + // Startup epoch: still pending (not active). + post_validators_ok( + "0", + vec![test_datum(5, &pk, ValidatorStatus::PendingQueued)], + ) + .mount(&mock) + .await; + // Next epoch: activated. + post_validators_ok( + "32", + vec![test_datum(5, &pk, ValidatorStatus::ActiveOngoing)], + ) + .mount(&mock) + .await; + + let cache = test_cache(&mock, vec![pk]); + let refresher = ValidatorCacheRefresher::new(cache.clone()); + + // First tick (epoch 0, first slot): fetches slot 0 — validator inactive. + refresher + .refresh(&test_slot(0, SPE)) + .await + .expect("refresh slot 0"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert!( + active.is_empty(), + "validator should be inactive at startup, got {active:?}" + ); + + // Next epoch's first slot: fetches slot 32 — validator now active. + refresher + .refresh(&test_slot(SPE, SPE)) + .await + .expect("refresh slot 32"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert_eq!( + active.len(), + 1, + "activated validator should now be resolved" + ); + assert!(active.contains_key(&5)); + } + + /// A validator that exits stops being resolved after the next epoch tick. + #[tokio::test] + async fn refresh_stops_resolving_exited_validator() { + let pk = test_pubkey(5); + let mock = MockServer::start().await; + post_validators_ok( + "0", + vec![test_datum(5, &pk, ValidatorStatus::ActiveOngoing)], + ) + .mount(&mock) + .await; + post_validators_ok( + "32", + vec![test_datum(5, &pk, ValidatorStatus::ExitedUnslashed)], + ) + .mount(&mock) + .await; + + let cache = test_cache(&mock, vec![pk]); + let refresher = ValidatorCacheRefresher::new(cache.clone()); + + refresher + .refresh(&test_slot(0, SPE)) + .await + .expect("refresh slot 0"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert_eq!(active.len(), 1, "validator active at startup"); + + refresher + .refresh(&test_slot(SPE, SPE)) + .await + .expect("refresh slot 32"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert!( + active.is_empty(), + "exited validator should no longer be resolved, got {active:?}" + ); + } + + /// Once refreshed by slot, mid-epoch ticks are skipped + /// (`shouldUpdateCache`): slot 0 is fetched exactly once and no + /// mid-epoch re-fetch is issued. + #[tokio::test] + async fn refresh_skips_mid_epoch_slot_once_refreshed_by_slot() { + let pk = test_pubkey(5); + let mock = MockServer::start().await; + // Only slot 0 is served, and it must be hit exactly once. Slot 1 is left + // unmounted: any mid-epoch fetch would 404 → head fallback → error. + post_validators_ok( + "0", + vec![test_datum(5, &pk, ValidatorStatus::ActiveOngoing)], + ) + .expect(1) + .mount(&mock) + .await; + + let cache = test_cache(&mock, vec![pk]); + let refresher = ValidatorCacheRefresher::new(cache.clone()); + + refresher + .refresh(&test_slot(0, SPE)) + .await + .expect("refresh slot 0"); + + // Mid-epoch tick: skipped, so it returns Ok without any fetch and the + // cache is untouched. + refresher + .refresh(&test_slot(1, SPE)) + .await + .expect("mid-epoch refresh is a skipped no-op"); + + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert_eq!(active.len(), 1, "cache retained the slot-0 set"); + } + + /// After a head fallback (`refreshed_by_slot == false`), the next tick is + /// forced to refresh even mid-epoch, and it re-fetches the epoch's first + /// slot rather than the current slot (`slot.Epoch() * slot.SlotsPerEpoch`). + #[tokio::test] + async fn refresh_refetches_epoch_first_slot_after_head_fallback() { + let pk = test_pubkey(1); + let mock = MockServer::start().await; + // First tick is at mid-epoch slot 5, fetched by slot; it 404s and falls + // back to head, which reports the validator as still pending. + post_validators_not_found("5").mount(&mock).await; + post_validators_ok( + "head", + vec![test_datum(1, &pk, ValidatorStatus::PendingQueued)], + ) + .mount(&mock) + .await; + // The epoch's first slot (0) reports the validator active. Slot 6 (the + // current slot on the second tick) is deliberately left unmounted: were + // it fetched, it would 404 → head → empty active set, failing the assert. + post_validators_ok( + "0", + vec![test_datum(1, &pk, ValidatorStatus::ActiveOngoing)], + ) + .mount(&mock) + .await; + + let cache = test_cache(&mock, vec![pk]); + let refresher = ValidatorCacheRefresher::new(cache.clone()); + + // First tick at slot 5: get_by_slot(5) 404s → head fallback (pending). + refresher + .refresh(&test_slot(5, SPE)) + .await + .expect("refresh slot 5 via head fallback"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert!( + active.is_empty(), + "head fallback reported pending validator" + ); + + // Second tick at mid-epoch slot 6: forced to refresh because the prior + // refresh fell back to head, and it fetches epoch-first slot 0 (active), + // not slot 6. + refresher + .refresh(&test_slot(6, SPE)) + .await + .expect("refresh slot 6 refetches epoch-first slot"); + let (active, _) = cache.get_by_head().await.expect("read cache"); + assert_eq!( + active.len(), + 1, + "refetched the epoch's first slot (0), where the validator is active" + ); + assert!(active.contains_key(&1)); + } +} From ec5f014c9f8a4a29ff49ecc09a51f9e9a8788780 Mon Sep 17 00:00:00 2001 From: "emlautarom1-agent[bot]" <292495798+emlautarom1-agent[bot]@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:53:11 -0300 Subject: [PATCH 2/2] refactor(app): address review feedback on valcache refresher - Reference Charon at the function level (`wireCoreWorkflow`, `shouldUpdateCache`, `GetBySlot`) instead of by line number. - Rename bookkeeping fields to match Charon's names in Rust convention: first_refresh -> first_val_cache_refresh, refreshed_by_slot -> refresh_by_slot. - Trim the over-explanatory subscriber comment and drop the test-module doc comment. --- crates/app/src/node/wire.rs | 55 ++++++++++++++----------------------- 1 file changed, 21 insertions(+), 34 deletions(-) diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index a77f68c8..eddd736e 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -170,8 +170,7 @@ pub struct WiredComponents { } /// Per-epoch trim + refresh for the shared [`ValidatorCache`], the Rust analog -/// of Charon's inline validator-cache refresh subscriber -/// (`app/app.go:484-532`). +/// of Charon's inline validator-cache refresh subscriber in `wireCoreWorkflow`. /// /// Registered as a scheduler slot subscriber by [`wire_core_workflow`]. The /// seeded cache is otherwise frozen at startup; this refresher re-fetches the @@ -183,62 +182,62 @@ struct ValidatorCacheRefresher { bookkeeping: tokio::sync::Mutex, } -/// Mirrors Charon's `firstValCacheRefresh` / `refreshedBySlot` locals -/// (`app.go:484-485`), which the refresh closure reads and writes under a lock. +/// Mirrors the `firstValCacheRefresh` / `refreshedBySlot` locals in Charon's +/// `wireCoreWorkflow`, which the refresh closure reads and writes under a lock. struct RefreshBookkeeping { /// Whether the cache has never been refreshed. Forces the first tick to /// refresh regardless of the slot's position in the epoch. - first_refresh: bool, + first_val_cache_refresh: bool, /// Whether the previous refresh fetched by slot (`true`) or fell back to /// the head state (`false`). A head fallback forces the next tick to /// refresh and to re-fetch the epoch's first slot. - refreshed_by_slot: bool, + refresh_by_slot: bool, } impl ValidatorCacheRefresher { fn new(cache: ValidatorCache) -> Self { Self { cache, - // Charon initializes both flags to `true` (`app.go:484-485`). + // Charon initializes both flags to `true`. bookkeeping: tokio::sync::Mutex::new(RefreshBookkeeping { - first_refresh: true, - refreshed_by_slot: true, + first_val_cache_refresh: true, + refresh_by_slot: true, }), } } /// Trims and refreshes the cache for `slot` when required, mirroring - /// Charon's `shouldUpdateCache` gate (`app.go:489-498`) and the `GetBySlot` - /// head-fallback re-fetch (`app.go:512-529`). + /// Charon's `shouldUpdateCache` gate and the `GetBySlot` head-fallback + /// re-fetch. async fn refresh(&self, slot: &Slot) -> Result<(), ValidatorCacheError> { let mut bk = self.bookkeeping.lock().await; // shouldUpdateCache: skip mid-epoch slots once the cache has been // refreshed at least once by slot. - if !slot.first_in_epoch() && !bk.first_refresh && bk.refreshed_by_slot { + if !slot.first_in_epoch() && !bk.first_val_cache_refresh && bk.refresh_by_slot { return Ok(()); } tracing::info!( slot = %slot.slot, - first_refresh = bk.first_refresh, + first_refresh = bk.first_val_cache_refresh, "Refreshing validator cache" ); // If the previous refresh fell back to head, fetch the epoch's first // slot rather than the current slot. `epoch * slots_per_epoch <= slot`, // so the multiply never actually saturates. - let slot_to_fetch = if bk.refreshed_by_slot { + let slot_to_fetch = if bk.refresh_by_slot { slot.slot.inner() } else { slot.epoch().saturating_mul(slot.slots_per_epoch) }; self.cache.trim().await; - let (_, _, refreshed_by_slot) = self.cache.get_by_slot(slot_to_fetch).await?; + let (_, _, refresh_by_slot) = self.cache.get_by_slot(slot_to_fetch).await?; - bk.refreshed_by_slot = refreshed_by_slot; - bk.first_refresh = false; + bk.refresh_by_slot = refresh_by_slot; + bk.first_val_cache_refresh = false; Ok(()) } @@ -292,8 +291,7 @@ pub async fn wire_core_workflow( // into both clients (app.go:481-482 and app.go:598); 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 (Charon app.go:480-532) refreshes every - // consumer at once. + // 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()), @@ -564,12 +562,8 @@ pub async fn wire_core_workflow( "consensus", ); } - // Slot subscriber: per-epoch validator cache trim + refresh (Charon - // app.go:480-532). The shared cache seeded above stays frozen at startup - // until this runs — it re-fetches the cluster validator set on each epoch's - // first slot so late-activating validators get scheduled and exited ones - // stop being resolved. A single refresh covers every consumer (scheduler, - // broadcaster, validator API) since they share this cache. + // Slot subscriber: per-epoch validator cache trim + refresh (Charon's + // `wireCoreWorkflow`). { let refresher = Arc::new(ValidatorCacheRefresher::new(validator_cache.clone())); sched_builder.subscribe_slot( @@ -701,13 +695,6 @@ pub async fn wire_core_workflow( #[cfg(test)] mod tests { - //! Unit tests for the per-epoch [`ValidatorCacheRefresher`] (Charon - //! `app.go:484-532`). They drive the refresher directly against a - //! `wiremock`-backed [`ValidatorCache`] — the same cache the scheduler - //! reads via `get_by_head` when resolving duties — and assert the - //! refreshed active set, so they mirror what the scheduler would - //! observe. - use super::*; use pluto_core::types::SlotNumber; use pluto_eth2api::{ @@ -933,9 +920,9 @@ mod tests { assert_eq!(active.len(), 1, "cache retained the slot-0 set"); } - /// After a head fallback (`refreshed_by_slot == false`), the next tick is + /// After a head fallback (`refresh_by_slot == false`), the next tick is /// forced to refresh even mid-epoch, and it re-fetches the epoch's first - /// slot rather than the current slot (`slot.Epoch() * slot.SlotsPerEpoch`). + /// slot rather than the current slot. #[tokio::test] async fn refresh_refetches_epoch_first_slot_after_head_fallback() { let pk = test_pubkey(1);