fix duties not found bug

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

crates/database/src/handle.rs

@@ -252,6 +252,13 @@ impl DbHandle {
         }
     }
 
+    pub fn fetch_validator_registrations(&self, pub_keys: Vec<BlsPublicKeyBytes>) {
+        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,

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;
                 }

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<T: FromRow>(rows: Vec<tokio_postgres::Row>) -> Result<Vec<T>,
     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<T: FromRow>(rows: Vec<tokio_postgres::Row>, context: &str) -> Vec<T> {
+    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<T: FromRow>(row: &tokio_postgres::Row) -> Result<T, DatabaseError> {
     T::from_row(row)
 }

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<BuilderGetValidatorsResponseEntry>,
     },
+    FetchValidatorRegistrations {
+        pub_keys: Vec<BlsPublicKeyBytes>,
+    },
     DisableAdjustments {
         block_hash: B256,
         failsafe_trigger: Arc<AtomicBool>,
@@ -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<BlsPublicKeyBytes> {
     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<Vec<SignedValidatorRegistrationEntry>, 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)]

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]