From cd12f9b542ed45e719a9acf2abc4e385a7a49729 Mon Sep 17 00:00:00 2001 From: owen Date: Fri, 11 Sep 2026 16:26:51 +0100 Subject: [PATCH] fix duties not found bug Currently we have two problems with proposer duties. 1 - registration fails to sync via the db 2 - stale duties from beacon node This PR aims to fix these by avoiding updating the watermark on the db side and making use of the dependant root on the beacon api side. --- crates/common/src/metrics.rs | 22 ++ crates/database/src/handle.rs | 7 + crates/database/src/lib.rs | 19 +- .../src/postgres/postgres_db_row_parsing.rs | 22 ++ .../src/postgres/postgres_db_service.rs | 67 +++-- crates/relay/src/housekeeper/chain_head.rs | 31 ++ crates/relay/src/housekeeper/duties.rs | 274 ++++++++++++++++-- crates/relay/src/housekeeper/tile.rs | 85 +++++- 8 files changed, 479 insertions(+), 48 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 214724428..ad4abe3de 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 6423c1b01..2a0d403b0 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 3e5449018..7ebc5ceb4 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 720e40415..7e8354218 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 3357601d5..d2a4b6b19 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 b0fe5fbfa..49077721b 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 324cec737..c40db7155 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 8a1112929..7d82ac262 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,