diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 21472442..ad4abe3d 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -192,6 +192,28 @@ lazy_static! { ) .unwrap(); + pub static ref DUTIES_FETCH: IntCounterVec = register_int_counter_vec_with_registry!( + "duties_fetch_total", + "Count of proposer duties fetches by outcome", + &["status"], + &RELAY_METRICS_REGISTRY + ) + .unwrap(); + + pub static ref DUTIES_AGE_SECONDS: Gauge = register_gauge_with_registry!( + "duties_age_seconds", + "Seconds since the last successful proposer duties fetch", + &RELAY_METRICS_REGISTRY + ) + .unwrap(); + + pub static ref DUTIES_LOOKAHEAD_SLOTS: Gauge = register_gauge_with_registry!( + "duties_lookahead_slots", + "Slots between the head and the last duty the relay knows about", + &RELAY_METRICS_REGISTRY + ) + .unwrap(); + //////////////// SIMULATOR //////////////// static ref SIMULATOR_COUNTS: IntCounterVec = register_int_counter_vec_with_registry!( "simulator_count_total", diff --git a/crates/database/src/handle.rs b/crates/database/src/handle.rs index 6423c1b0..2a0d403b 100644 --- a/crates/database/src/handle.rs +++ b/crates/database/src/handle.rs @@ -252,6 +252,13 @@ impl DbHandle { } } + pub fn fetch_validator_registrations(&self, pub_keys: Vec) { + if let Err(err) = self.sender.try_send(DbRequest::FetchValidatorRegistrations { pub_keys }) + { + error!(%err, "failed to send FetchValidatorRegistrations request"); + } + } + pub fn disable_adjustments( &self, block_hash: B256, diff --git a/crates/database/src/lib.rs b/crates/database/src/lib.rs index 3e544901..7ebc5ceb 100644 --- a/crates/database/src/lib.rs +++ b/crates/database/src/lib.rs @@ -14,7 +14,7 @@ use std::{ use helix_common::{RelayConfig, is_local_dev, local_cache}; pub use postgres::postgres_db_service::PostgresDatabaseService; -use tracing::info; +use tracing::{error, info}; pub use types::*; pub use crate::postgres::postgres_db_service::{DbRequest, PendingBlockSubmissionValue}; @@ -77,10 +77,21 @@ pub async fn start_db_service( snapshot::save_known_validators_bg(dir, &set); } let fetch_time = SystemTime::now(); - postgres_db.update_validator_registrations(validator_reg_update_time).await; - validator_reg_update_time = fetch_time; + match postgres_db + .update_validator_registrations(validator_reg_update_time) + .await + { + Ok(()) => validator_reg_update_time = fetch_time, + Err(err) => { + error!(%err, "validator registration update failed, keeping watermark") + } + } if let Some(dir) = &snapshot_dir { - save_validator_registrations_snapshot(&local_cache, dir, fetch_time); + save_validator_registrations_snapshot( + &local_cache, + dir, + validator_reg_update_time, + ); } postgres_db.load_builder_infos(local_cache.clone()).await; } diff --git a/crates/database/src/postgres/postgres_db_row_parsing.rs b/crates/database/src/postgres/postgres_db_row_parsing.rs index 720e4041..7e835421 100644 --- a/crates/database/src/postgres/postgres_db_row_parsing.rs +++ b/crates/database/src/postgres/postgres_db_row_parsing.rs @@ -12,6 +12,7 @@ use helix_types::{ BidTrace, BlsPublicKeyBytes, BlsSignatureBytes, SignedValidatorRegistration, ValidatorRegistration, }; +use tracing::warn; use uuid::Uuid; use crate::{ @@ -387,6 +388,27 @@ pub fn parse_rows(rows: Vec) -> Result, rows.iter().map(|row| T::from_row(row)).collect() } +/// Parse rows, dropping the ones that fail instead of losing the whole batch. +pub fn parse_rows_lossy(rows: Vec, context: &str) -> Vec { + let mut parsed = Vec::with_capacity(rows.len()); + let mut skipped = 0usize; + for row in rows.iter() { + match T::from_row(row) { + Ok(value) => parsed.push(value), + Err(err) => { + if skipped == 0 { + warn!(%err, context, "skipping unparsable row"); + } + skipped += 1; + } + } + } + if skipped > 0 { + warn!(skipped, context, "skipped unparsable rows"); + } + parsed +} + pub fn parse_row(row: &tokio_postgres::Row) -> Result { T::from_row(row) } diff --git a/crates/database/src/postgres/postgres_db_service.rs b/crates/database/src/postgres/postgres_db_service.rs index 3357601d..d2a4b6b1 100644 --- a/crates/database/src/postgres/postgres_db_service.rs +++ b/crates/database/src/postgres/postgres_db_service.rs @@ -40,7 +40,7 @@ use crate::{ postgres_db_init::run_migrations_async, postgres_db_row_parsing::{ parse_bytes_to_hash, parse_bytes_to_pubkey_bytes, parse_i32_to_u64, parse_i64_to_u64, - parse_row, parse_rows, + parse_row, parse_rows, parse_rows_lossy, }, postgres_db_u256_parsing::PostgresNumeric, }, @@ -111,6 +111,9 @@ pub enum DbRequest { SetProposerDuties { duties: Vec, }, + FetchValidatorRegistrations { + pub_keys: Vec, + }, DisableAdjustments { block_hash: B256, failsafe_trigger: Arc, @@ -139,6 +142,8 @@ const MAINNET_VALIDATOR_COUNT: usize = 1_100_000; const DB_CHECK_INTERVAL: Duration = Duration::from_secs(1); static DELIVERED_PAYLOADS_MIG_SLOT: AtomicU64 = AtomicU64::new(0); const POSTGRES_PASSWORD_ENV_VAR: &str = "POSTGRES_PASSWORD"; +/// Covers clock skew between hosts and the delay between `inserted_at` and the commit. +const REGISTRATION_FETCH_OVERLAP: Duration = Duration::from_secs(120); fn new_validator_set() -> FxHashSet { FxHashSet::with_capacity_and_hasher(MAINNET_VALIDATOR_COUNT, Default::default()) @@ -357,24 +362,22 @@ impl PostgresDatabaseService { } #[instrument(skip_all)] - pub async fn update_validator_registrations(&self, last_update_time: SystemTime) { + pub async fn update_validator_registrations( + &self, + last_update_time: SystemTime, + ) -> Result<(), DatabaseError> { let mut record = DbMetricRecord::new("update_validator_registrations"); - match self.fetch_validator_registrations_since(last_update_time).await { - Ok(entries) => { - let num_entries = entries.len(); - entries.into_iter().for_each(|entry| { - self.local_cache - .validator_registration_cache - .insert(entry.registration_info.registration.message.pubkey, entry); - }); - info!("Loaded {} validator registrations", num_entries); - record.record_success(); - } - Err(e) => { - error!("Error loading validator registrations: {}", e); - } - } + let entries = self.fetch_validator_registrations_since(last_update_time).await?; + let num_entries = entries.len(); + entries.into_iter().for_each(|entry| { + self.local_cache + .validator_registration_cache + .insert(entry.registration_info.registration.message.pubkey, entry); + }); + info!("Loaded {} validator registrations", num_entries); + record.record_success(); + Ok(()) } pub async fn start_processors( @@ -692,6 +695,26 @@ impl PostgresDatabaseService { error!(%err, "failed to set proposer duties"); } } + DbRequest::FetchValidatorRegistrations { pub_keys } => { + let refs: Vec<&BlsPublicKeyBytes> = pub_keys.iter().collect(); + match self.get_validator_registrations_for_pub_keys(&refs).await { + Ok(entries) => { + let found = entries.len(); + for entry in entries { + self.local_cache + .validator_registration_cache + .insert(entry.registration_info.registration.message.pubkey, entry); + } + info!( + requested = pub_keys.len(), + found, "fetched registrations missing for upcoming duties" + ); + } + Err(err) => { + error!(%err, "failed to fetch registrations for upcoming duties"); + } + } + } DbRequest::DisableAdjustments { block_hash, failsafe_trigger, adjustments_enabled } => { if let Err(err) = self.disable_adjustments().await { failsafe_trigger.store(true, Ordering::Relaxed); @@ -1159,7 +1182,7 @@ impl PostgresDatabaseService { .await?; record.record_success(); - parse_rows(rows) + Ok(parse_rows_lossy(rows, "validator_registrations")) } #[instrument(skip_all)] @@ -1169,6 +1192,10 @@ impl PostgresDatabaseService { ) -> Result, DatabaseError> { let mut record = DbMetricRecord::new("fetch_validator_registrations_since"); + let since = last_fetch_time + .checked_sub(REGISTRATION_FETCH_OVERLAP) + .unwrap_or(SystemTime::UNIX_EPOCH); + let rows = self .pool .get() @@ -1180,12 +1207,12 @@ impl PostgresDatabaseService { ON validator_registrations.public_key = validator_preferences.public_key WHERE validator_registrations.active = true AND validator_registrations.inserted_at > $1 ", - &[&last_fetch_time], + &[&since], ) .await?; record.record_success(); - parse_rows(rows) + Ok(parse_rows_lossy(rows, "validator_registrations_since")) } #[instrument(skip_all)] diff --git a/crates/relay/src/housekeeper/chain_head.rs b/crates/relay/src/housekeeper/chain_head.rs index b0fe5fbf..49077721 100644 --- a/crates/relay/src/housekeeper/chain_head.rs +++ b/crates/relay/src/housekeeper/chain_head.rs @@ -147,6 +147,10 @@ impl ChainHead { pub fn mark_duties_done(&mut self) { self.duties_done = true; } + /// Re-arm `is_ready()` so a corrective update goes out after a reorg changed the duties. + pub fn mark_duties_changed(&mut self) { + self.sent_was_complete = false; + } pub fn mark_payload_attrs_done(&mut self) { self.payload_attributes_done = true; } @@ -224,6 +228,33 @@ mod tests { assert!(!ch.is_ready()); } + #[test] + fn duties_change_rearms_a_complete_send() { + let (mut ch, ci) = make_head(); + ch.update(head_event(ci.current_slot() + 1)); + ch.mark_duties_done(); + ch.mark_payload_attrs_done(); + ch.mark_il_done(); + ch.sent(); + assert!(!ch.is_ready()); + + ch.mark_duties_changed(); + assert!(ch.is_ready()); + } + + #[test] + fn a_corrective_send_does_not_rearm_again() { + let (mut ch, ci) = make_head(); + ch.update(head_event(ci.current_slot() + 1)); + ch.mark_duties_done(); + ch.mark_payload_attrs_done(); + ch.mark_il_done(); + ch.sent(); + ch.mark_duties_changed(); + ch.sent(); + assert!(!ch.is_ready()); + } + // --- Head event transitions --- #[test] diff --git a/crates/relay/src/housekeeper/duties.rs b/crates/relay/src/housekeeper/duties.rs index 324cec73..c40db715 100644 --- a/crates/relay/src/housekeeper/duties.rs +++ b/crates/relay/src/housekeeper/duties.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use alloy_primitives::B256; use helix_common::{ ProposerDuty, SignedValidatorRegistrationEntry, api::{ @@ -12,25 +13,81 @@ use helix_common::{ validator_preferences::ValidatorPreferences, }; use helix_database::handle::DbHandle; -use helix_types::{BlsPublicKeyBytes, SignedValidatorRegistration, ValidatorRegistration}; -use rustc_hash::FxHashMap; +use helix_types::{BlsPublicKeyBytes, SignedValidatorRegistration, Slot, ValidatorRegistration}; +use rustc_hash::{FxHashMap, FxHashSet}; use tracing::{error, info}; use url::Url; // State machine for the two-epoch proposer duties fetch (current + next epoch). pub enum DutiesFetchState { Current { req: PendingResponse, next_epoch_url: Url }, - Next { current: Vec, req: PendingResponse }, + Next { current: Vec, dependent_root: Option, req: PendingResponse }, Done, } +pub struct DutiesUpdate { + pub duties: Vec, + /// A change means a reorg moved the block the duties were computed from. + pub dependent_root: Option, + /// False when the next-epoch request failed, so `duties` covers one epoch only. + pub has_next_epoch: bool, +} + +fn dependent_root(meta: &FxHashMap) -> Option { + meta.get("dependent_root")?.as_str()?.parse().ok() +} + +/// Keep the duties a partial fetch did not cover, so one failed request does not shrink +/// the lookahead the relay gives builders. +pub fn merge_partial_duties( + update: DutiesUpdate, + known: &[ProposerDuty], + first_uncovered_slot: Slot, +) -> Vec { + if update.has_next_epoch { + return update.duties; + } + let mut duties = update.duties; + let covered: FxHashSet = duties.iter().map(|d| d.slot).collect(); + duties.extend( + known + .iter() + .filter(|d| d.slot >= first_uncovered_slot && !covered.contains(&d.slot)) + .cloned(), + ); + duties.sort_by_key(|d| d.slot); + duties +} + +/// Keeps a missing duty pubkey to one fetch per epoch, not one per slot. +#[derive(Default)] +pub struct RequestedRegistrations { + epoch: u64, + keys: FxHashSet, +} + +impl RequestedRegistrations { + pub fn take_new( + &mut self, + epoch: u64, + keys: impl Iterator, + ) -> Vec { + if epoch != self.epoch { + self.epoch = epoch; + self.keys.clear(); + } + keys.filter(|key| self.keys.insert(*key)).collect() + } +} + impl DutiesFetchState { pub fn new( http_client: &HttpClient, beacon_client: &MultiBeaconClient, epoch: u64, + attempt: usize, ) -> Option { - let c = beacon_client.beacon_clients_by_last_response().next()?; + let c = beacon_client.beacon_clients_by_last_response().nth(attempt)?; let url = c.config.url.join(&format!("/eth/v1/validator/duties/proposer/{epoch}")).ok()?; let next_url = c.config.url.join(&format!("/eth/v1/validator/duties/proposer/{}", epoch + 1)).ok()?; @@ -41,7 +98,7 @@ impl DutiesFetchState { pub fn poll( &mut self, http_client: &HttpClient, - ) -> std::task::Poll, Box>> { + ) -> std::task::Poll>> { use std::task::Poll; loop { match std::mem::replace(self, Self::Done) { @@ -52,30 +109,49 @@ impl DutiesFetchState { return Poll::Pending; } Poll::Ready(Err(e)) => return Poll::Ready(Err(e.into())), - Poll::Ready(Ok(resp)) => match http_client.get(&next_epoch_url) { - Err(e) => { - error!(%e, epoch_offset = 1, "failed to start next duties fetch"); - return Poll::Ready(Ok(resp.data)); - } - Ok(next_req) => { - *self = Self::Next { current: resp.data, req: next_req }; + Poll::Ready(Ok(resp)) => { + let root = dependent_root(&resp.meta); + match http_client.get(&next_epoch_url) { + Err(e) => { + error!(%e, epoch_offset = 1, "failed to start next duties fetch"); + return Poll::Ready(Ok(DutiesUpdate { + duties: resp.data, + dependent_root: root, + has_next_epoch: false, + })); + } + Ok(next_req) => { + *self = Self::Next { + current: resp.data, + dependent_root: root, + req: next_req, + }; + } } - }, + } } } - Self::Next { mut current, mut req } => { + Self::Next { mut current, dependent_root, mut req } => { match req.poll_json::>>() { Poll::Pending => { - *self = Self::Next { current, req }; + *self = Self::Next { current, dependent_root, req }; return Poll::Pending; } Poll::Ready(Err(e)) => { error!(%e, epoch_offset = 1, "failed fetching next epoch duties"); - return Poll::Ready(Ok(current)); + return Poll::Ready(Ok(DutiesUpdate { + duties: current, + dependent_root, + has_next_epoch: false, + })); } Poll::Ready(Ok(mut resp)) => { current.append(&mut resp.data); - return Poll::Ready(Ok(current)); + return Poll::Ready(Ok(DutiesUpdate { + duties: current, + dependent_root, + has_next_epoch: true, + })); } } } @@ -128,11 +204,22 @@ pub fn process_duties( proposer_duties: &[ProposerDuty], local_cache: &Arc, db: &DbHandle, + requested: &mut RequestedRegistrations, + epoch: u64, ) { let pubkeys: Vec = proposer_duties.iter().map(|d| d.pubkey).collect(); let registrations: Vec = local_cache.get_validator_registrations_for_pub_keys(&pubkeys); - let registrations = registrations.into_iter().map(|e| (*e.public_key(), e)).collect(); + let registrations: FxHashMap = + registrations.into_iter().map(|e| (*e.public_key(), e)).collect(); + + let missing = requested + .take_new(epoch, pubkeys.iter().copied().filter(|key| !registrations.contains_key(key))); + if !missing.is_empty() { + info!(missing = missing.len(), "fetching registrations missing for upcoming duties"); + db.fetch_validator_registrations(missing); + } + let formatted: Vec = _build_formatted_duties(proposer_duties, ®istrations); @@ -140,3 +227,154 @@ pub fn process_duties( local_cache.update_proposer_duties(formatted.clone()); db.set_proposer_duties(formatted); } + +#[cfg(test)] +mod tests { + use helix_common::validator_preferences::ValidatorPreferences; + use helix_database::{DbRequest, PendingBlockSubmissionValue}; + use helix_types::Slot; + + use super::*; + + type Requests = crossbeam_channel::Receiver; + + fn pubkey(byte: u8) -> BlsPublicKeyBytes { + BlsPublicKeyBytes::repeat_byte(byte) + } + + fn duty(slot: u64, pubkey: BlsPublicKeyBytes) -> ProposerDuty { + ProposerDuty { pubkey, validator_index: slot, slot: Slot::new(slot) } + } + + fn harness() -> (Arc, DbHandle, Requests) { + let (sender, receiver) = crossbeam_channel::unbounded(); + let (batch_sender, _batch_receiver) = + crossbeam_channel::unbounded::(); + (Arc::new(LocalCache::new()), DbHandle::new(sender, batch_sender), receiver) + } + + fn register(cache: &LocalCache, pubkey: BlsPublicKeyBytes) { + let info = ValidatorRegistrationInfo { + registration: SignedValidatorRegistration { + message: ValidatorRegistration { + fee_recipient: Default::default(), + gas_limit: 60_000_000, + timestamp: 0, + pubkey, + }, + signature: Default::default(), + }, + preferences: ValidatorPreferences::default(), + }; + cache.save_validator_registrations(std::iter::once(info), None); + } + + fn fetched(receiver: &Requests) -> Vec { + let mut keys = Vec::new(); + while let Ok(request) = receiver.try_recv() { + if let DbRequest::FetchValidatorRegistrations { pub_keys } = request { + keys.extend(pub_keys); + } + } + keys.sort(); + keys + } + + #[test] + fn fetches_only_the_duty_pubkeys_missing_from_the_cache() { + let (cache, db, receiver) = harness(); + register(&cache, pubkey(1)); + let duties = vec![duty(1, pubkey(1)), duty(2, pubkey(2)), duty(3, pubkey(3))]; + + process_duties(&duties, &cache, &db, &mut RequestedRegistrations::default(), 0); + + assert_eq!(fetched(&receiver), vec![pubkey(2), pubkey(3)]); + } + + #[test] + fn fetches_nothing_when_every_duty_is_registered() { + let (cache, db, receiver) = harness(); + register(&cache, pubkey(1)); + let duties = vec![duty(1, pubkey(1))]; + + process_duties(&duties, &cache, &db, &mut RequestedRegistrations::default(), 0); + + assert!(fetched(&receiver).is_empty()); + } + + #[test] + fn requests_a_missing_pubkey_once_per_epoch() { + let (cache, db, receiver) = harness(); + let duties = vec![duty(1, pubkey(2))]; + let mut requested = RequestedRegistrations::default(); + + process_duties(&duties, &cache, &db, &mut requested, 7); + assert_eq!(fetched(&receiver), vec![pubkey(2)]); + + process_duties(&duties, &cache, &db, &mut requested, 7); + assert!(fetched(&receiver).is_empty()); + + process_duties(&duties, &cache, &db, &mut requested, 8); + assert_eq!(fetched(&receiver), vec![pubkey(2)]); + } + + #[test] + fn requests_a_pubkey_once_when_it_has_two_duties_in_the_window() { + let (cache, db, receiver) = harness(); + let duties = vec![duty(1, pubkey(2)), duty(40, pubkey(2))]; + + process_duties(&duties, &cache, &db, &mut RequestedRegistrations::default(), 0); + + assert_eq!(fetched(&receiver), vec![pubkey(2)]); + } + + fn update(duties: Vec, has_next_epoch: bool) -> DutiesUpdate { + DutiesUpdate { duties, dependent_root: None, has_next_epoch } + } + + #[test] + fn a_complete_fetch_replaces_the_duty_list() { + let known = vec![duty(40, pubkey(9))]; + + let merged = + merge_partial_duties(update(vec![duty(1, pubkey(1))], true), &known, Slot::new(32)); + + assert_eq!(merged.len(), 1); + assert_eq!(merged[0].slot, Slot::new(1)); + } + + #[test] + fn a_partial_fetch_keeps_the_duties_it_did_not_cover() { + let known = vec![duty(1, pubkey(1)), duty(40, pubkey(9))]; + + let merged = + merge_partial_duties(update(vec![duty(1, pubkey(1))], false), &known, Slot::new(32)); + + let slots: Vec = merged.iter().map(|d| d.slot.as_u64()).collect(); + assert_eq!(slots, vec![1, 40]); + } + + #[test] + fn a_partial_fetch_prefers_the_fresh_duty_for_a_slot_it_covered() { + let known = vec![duty(1, pubkey(9))]; + + let merged = + merge_partial_duties(update(vec![duty(1, pubkey(1))], false), &known, Slot::new(0)); + + assert_eq!(merged.len(), 1); + assert_eq!(merged[0].pubkey, pubkey(1)); + } + + #[test] + fn builds_the_duties_it_can_when_a_registration_is_missing() { + let (cache, db, _receiver) = harness(); + register(&cache, pubkey(1)); + let duties = vec![duty(1, pubkey(1)), duty(2, pubkey(2))]; + + process_duties(&duties, &cache, &db, &mut RequestedRegistrations::default(), 0); + + let stored = cache.get_proposer_duties(); + assert_eq!(stored.len(), 1); + assert_eq!(stored[0].entry.registration.message.pubkey, pubkey(1)); + } +} diff --git a/crates/relay/src/housekeeper/tile.rs b/crates/relay/src/housekeeper/tile.rs index 8a111292..7d82ac26 100644 --- a/crates/relay/src/housekeeper/tile.rs +++ b/crates/relay/src/housekeeper/tile.rs @@ -20,6 +20,7 @@ use helix_common::{ chain_info::ChainInfo, http::client::{HttpClient, PendingResponse, SseStream}, local_cache::LocalCache, + metrics::{DUTIES_AGE_SECONDS, DUTIES_FETCH, DUTIES_LOOKAHEAD_SLOTS}, }; use helix_types::Slot; use rustc_hash::FxHashMap; @@ -29,7 +30,7 @@ use crate::{ DbHandle, HelixSpine, housekeeper::{ chain_head::ChainHead, - duties::{DutiesFetchState, process_duties}, + duties::{DutiesFetchState, RequestedRegistrations, merge_partial_duties, process_duties}, inclusion_list_service::{IL_CUTOFF, IlFetchState}, payload_attrs::process_payload_attributes, primev_service::{ @@ -43,6 +44,8 @@ use crate::{ const KNOWN_VALIDATORS_REFRESH_INTERVAL: Duration = Duration::from_secs(10 * 60); const SYNC_STATUS_CHECK_INTERVAL: Duration = Duration::from_secs(4); +/// Warn once a slot when the duty list is older than this. +const DUTIES_STALE_AFTER: Duration = Duration::from_secs(60); const SYNC_STATUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(2); /// Slot event stored in SharedVector and broadcast via spine. @@ -67,6 +70,8 @@ struct HousekeeperStats { duties_fetch_ok: u32, duties_fetch_empty: u32, duties_fetch_err: u32, + duties_fetch_partial: u32, + duties_dependent_root_changed: u32, il_fetch_ok: u32, il_fetch_timeout: u32, il_fetch_decode_err: u32, @@ -102,6 +107,10 @@ pub struct HousekeeperTile { known_payload_attributes: FxHashMap<(B256, Slot), PayloadAttributesUpdate>, duties: Vec, + requested_registrations: RequestedRegistrations, + last_dependent_root: Option, + duties_fetch_attempt: usize, + last_duties_ok: Instant, // In-flight fetch state machines duties_fetch: Option, @@ -174,6 +183,10 @@ impl HousekeeperTile { db, known_payload_attributes: FxHashMap::default(), duties: Vec::with_capacity(64), + requested_registrations: RequestedRegistrations::default(), + last_dependent_root: None, + duties_fetch_attempt: 0, + last_duties_ok: Instant::now(), duties_fetch: None, primev_builders_fetch: None, primev_validators_fetch: None, @@ -194,8 +207,10 @@ impl HousekeeperTile { let slot = self.chain_head.head(); info!(%slot, "new slot started"); + self.duties_fetch_attempt = 0; self.fetch_duties(); self.maybe_fetch_il(); + self.report_duties_age(); let bid_slot = slot + 1; for d in &self.duties { @@ -218,9 +233,31 @@ impl HousekeeperTile { &self.http_client, &self.beacon_client, self.chain_head.epoch().as_u64(), + self.duties_fetch_attempt, ); } + /// First slot a next-epoch failure leaves uncovered, i.e. the start of the next epoch. + fn first_uncovered_slot(&self) -> Slot { + let slots_per_epoch = self.chain_head.chain_info().slots_per_epoch(); + (self.chain_head.epoch() + 1).start_slot(slots_per_epoch) + } + + fn report_duties_age(&self) { + let age = self.last_duties_ok.elapsed(); + DUTIES_AGE_SECONDS.set(age.as_secs_f64()); + let head = self.chain_head.head(); + let last_duty = self.duties.iter().map(|d| d.slot).max().unwrap_or(head); + DUTIES_LOOKAHEAD_SLOTS.set(last_duty.as_u64().saturating_sub(head.as_u64()) as f64); + if age >= DUTIES_STALE_AFTER { + warn!( + age_secs = age.as_secs(), + lookahead_slots = last_duty.as_u64().saturating_sub(head.as_u64()), + "proposer duties are stale, serving a duty list from an earlier fetch" + ); + } + } + fn maybe_fetch_primev(&mut self) { if let Some(cfg) = &self.primev_config { self.primev_builders_fetch = PrimevBuildersFetch::new(&self.http_client, cfg); @@ -307,20 +344,54 @@ impl Tile for HousekeeperTile { if let Some(result) = duties_result { self.duties_fetch = None; match result { - Ok(proposer_duties) if proposer_duties.is_empty() => { + Ok(update) if update.duties.is_empty() => { self.stats.duties_fetch_empty += 1; + DUTIES_FETCH.with_label_values(&["empty"]).inc(); warn!("no proposer duties found"); } - Ok(proposer_duties) => { + Ok(update) => { self.stats.duties_fetch_ok += 1; - process_duties(&proposer_duties, &self.local_cache, &self.db); - self.duties = proposer_duties; + DUTIES_FETCH.with_label_values(&["ok"]).inc(); + self.last_duties_ok = Instant::now(); + let root_changed = update.dependent_root.is_some() && + self.last_dependent_root.is_some() && + update.dependent_root != self.last_dependent_root; + if update.dependent_root.is_some() { + self.last_dependent_root = update.dependent_root; + } + if !update.has_next_epoch { + self.stats.duties_fetch_partial += 1; + DUTIES_FETCH.with_label_values(&["partial"]).inc(); + } + let duties = + merge_partial_duties(update, &self.duties, self.first_uncovered_slot()); + process_duties( + &duties, + &self.local_cache, + &self.db, + &mut self.requested_registrations, + self.chain_head.epoch().as_u64(), + ); + self.duties = duties; self.chain_head.mark_duties_done(); + if root_changed { + self.stats.duties_dependent_root_changed += 1; + warn!( + dependent_root = ?self.last_dependent_root, + "proposer duties dependent root changed, re-sending slot update" + ); + self.chain_head.mark_duties_changed(); + } self.maybe_fetch_primev(); } Err(e) => { self.stats.duties_fetch_err += 1; - error!(%e, "failed to fetch proposer duties"); + DUTIES_FETCH.with_label_values(&["err"]).inc(); + error!(%e, attempt = self.duties_fetch_attempt, "failed to fetch proposer duties"); + self.duties_fetch_attempt += 1; + if self.duties_fetch_attempt < self.beacon_client.beacon_clients.len() { + self.fetch_duties(); + } } } } @@ -569,6 +640,8 @@ fn send_slot_event( duties_fetch_ok = stats.duties_fetch_ok, duties_fetch_empty = stats.duties_fetch_empty, duties_fetch_err = stats.duties_fetch_err, + duties_fetch_partial = stats.duties_fetch_partial, + duties_dependent_root_changed = stats.duties_dependent_root_changed, il_fetch_ok = stats.il_fetch_ok, il_fetch_timeout = stats.il_fetch_timeout, il_fetch_decode_err = stats.il_fetch_decode_err,