chore: refactor polling of duties in metadata service

shane-moore · shane-moore · commit 8aba0c426973 · 2026-01-23T17:17:57.000-08:00
diff --git a/anchor/common/ssv_types/src/message.rs b/anchor/common/ssv_types/src/message.rs
@@ -454,7 +454,7 @@ impl SignedSSVMessage {
                     }
                 })
             })
-            .collect();
+            .collect::<Result<Vec<_>, _>>()?;
 
         // Then convert the Vec of VariableLists to VariableList<VariableList<u8, U256>, U13>
         // This can fail if we have more than 13 signatures
diff --git a/anchor/validator_store/src/lib.rs b/anchor/validator_store/src/lib.rs
@@ -968,6 +968,16 @@ pub enum SpecificError {
     AggregatorInfoSlotPassed,
     /// Watch channel for AggregationAssignments has been closed
     AggregatorInfoChannelClosed,
+    /// produce_selection_proof called for validator not in VotingAssignments.attesting_committees
+    ValidatorNotAttesting {
+        validator_pubkey: PublicKeyBytes,
+        slot: Slot,
+    },
+    /// produce_sync_selection_proof called for validator not in VotingAssignments.sync_validators_by_subnet
+    ValidatorNotInSyncCommittee {
+        validator_pubkey: PublicKeyBytes,
+        slot: Slot,
+    },
 }
 
 impl From<CollectionError> for SpecificError {
@@ -1531,6 +1541,21 @@ impl<T: SlotClock, E: EthSpec> ValidatorStore for AnchorValidatorStore<T, E> {
                 let committee_id = cluster.committee_id();
                 let voting_assignments = self.get_voting_assignments(slot).await?;
 
+                // Defensive check: validator should be in `VotingAssignments` since both this
+                // function call and `VotingAssignments` are derived from `DutiesService`. If not,
+                // there's an inconsistency (e.g., stale cache after poll timeout) and we
+                // should not participate with a wrong `num_signatures_to_collect`.
+                if !voting_assignments
+                    .attesting_committees
+                    .contains_key(&validator_pubkey)
+                {
+                    return Err(SpecificError::ValidatorNotAttesting {
+                        validator_pubkey,
+                        slot,
+                    }
+                    .into());
+                }
+
                 // Build a set of validator indices in this committee
                 // This handles divergent operator views, since we only count validators we have
                 // shares
@@ -1629,6 +1654,24 @@ impl<T: SlotClock, E: EthSpec> ValidatorStore for AnchorValidatorStore<T, E> {
                 let committee_id = cluster.committee_id();
                 let voting_assignments = self.get_voting_assignments(slot).await?;
 
+                // Defensive check: validator should be in `VotingAssignments` since both this
+                // function call and `VotingAssignments` are derived from `DutiesService`. If not,
+                // there's an inconsistency (e.g., stale cache after poll timeout) and we
+                // should not participate with a wrong `num_signatures_to_collect`.
+                let validator_index = validator
+                    .index
+                    .ok_or(SpecificError::MissingIndex)?;
+                if !voting_assignments
+                    .sync_validators_by_subnet
+                    .contains_key(&validator_index)
+                {
+                    return Err(SpecificError::ValidatorNotInSyncCommittee {
+                        validator_pubkey: *validator_pubkey,
+                        slot,
+                    }
+                    .into());
+                }
+
                 // Build a set of validator indices in this committee
                 // This handles divergent operator views, since we only count validators we have
                 // shares for
diff --git a/anchor/validator_store/src/metadata_service.rs b/anchor/validator_store/src/metadata_service.rs
@@ -1,21 +1,21 @@
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use beacon_node_fallback::BeaconNodeFallback;
 use slot_clock::SlotClock;
 use ssv_types::{ValidatorIndex, consensus::BeaconVote};
 use task_executor::TaskExecutor;
 use tokio::{sync::watch, time::sleep};
-use tracing::{error, info, trace};
+use tracing::{error, info, trace, warn};
 use types::{ChainSpec, EthSpec, Slot, SyncSubnetId};
 use validator_services::duties_service::DutiesService;
 
 use crate::{
     AggregationAssignments, AnchorValidatorStore, ContributionWaiter, VotingAssignments,
-    VotingContext,
+    VotingContext, metrics,
 };
 
 #[derive(Clone)]
@@ -72,12 +72,24 @@ impl<E: EthSpec, T: SlotClock + 'static> MetadataService<E, T> {
         // PHASE 1: VotingAssignments (slot start)
         // Caches voting assignments for use by both selection proofs AND voting context.
         // Waits for DutiesService to complete polling before reading duties.
+        //
+        // RESILIENCE: We wait up to 3.5s for poll signals, but always proceed to
+        // read from duties cache regardless of poll outcome. This handles:
+        // - Normal operation: Signals arrive quickly (< 100ms), we read fresh duties
+        // - Beacon node slow: Signal arrives late (< 3s), we still get fresh duties
+        // - Beacon node down: Timeout after 3.5s, we read from cache (stale or empty)
+        //
+        // The 3.5s timeout is chosen because:
+        // - Lighthouse BN API timeout is 3 seconds (slot_duration / 4)
+        // - Phase 2 starts at 4 seconds (1/3 slot)
+        // - This gives 500ms buffer after BN timeout before Phase 2 deadline
         // ═══════════════════════════════════════════════════════════════════════
         let self_clone_phase1 = self.clone();
         executor.spawn(
             async move {
                 let mut attesters_poll_rx = self_clone_phase1.attesters_poll_rx.clone();
                 let mut sync_poll_rx = self_clone_phase1.sync_poll_rx.clone();
+                let poll_timeout = Duration::from_millis(3500);
 
                 loop {
                     if let Some(duration_to_next_slot) =
@@ -87,20 +99,71 @@ impl<E: EthSpec, T: SlotClock + 'static> MetadataService<E, T> {
                         sleep(duration_to_next_slot).await;
 
                         let slot: Slot = self_clone_phase1.slot_clock.now().unwrap_or_default();
-
-                        // Wait for BOTH DutiesService polls to complete for this slot.
-                        // This guarantees duties are cached before we read them.
-                        // If poll channels are closed, skip this slot to avoid signing with stale
-                        // data.
-                        if attesters_poll_rx.wait_for(|&s| s >= slot).await.is_err() {
-                            error!(%slot, "Attesters poll channel closed, skipping slot");
-                            continue;
+                        let poll_start = Instant::now();
+
+                        // Wait for poll signals with timeout (parallel execution).
+                        let (attesters_ready, sync_ready) = tokio::join!(
+                            async {
+                                tokio::time::timeout(
+                                    poll_timeout,
+                                    attesters_poll_rx.wait_for(|&s| s >= slot),
+                                )
+                                .await
+                                .is_ok_and(|r| r.is_ok())
+                            },
+                            async {
+                                tokio::time::timeout(
+                                    poll_timeout,
+                                    sync_poll_rx.wait_for(|&s| s >= slot),
+                                )
+                                .await
+                                .is_ok_and(|r| r.is_ok())
+                            }
+                        );
+
+                        let poll_duration = poll_start.elapsed();
+
+                        // Log poll failures
+                        if !attesters_ready {
+                            warn!(
+                                %slot,
+                                poll_duration_ms = poll_duration.as_millis(),
+                                "Attesters poll failed or timed out - will use cached duties"
+                            );
                         }
-                        if sync_poll_rx.wait_for(|&s| s >= slot).await.is_err() {
-                            error!(%slot, "Sync poll channel closed, skipping slot");
-                            continue;
+                        if !sync_ready {
+                            warn!(
+                                %slot,
+                                poll_duration_ms = poll_duration.as_millis(),
+                                "Sync poll failed or timed out - will use cached duties"
+                            );
                         }
 
+                        // Record poll telemetry
+                        metrics::inc_counter_vec(
+                            &metrics::METADATA_SERVICE_POLL_TOTAL,
+                            &[
+                                metrics::ATTESTERS,
+                                if attesters_ready { metrics::SUCCESS } else { metrics::FAILED },
+                            ],
+                        );
+                        metrics::inc_counter_vec(
+                            &metrics::METADATA_SERVICE_POLL_TOTAL,
+                            &[
+                                metrics::SYNC,
+                                if sync_ready { metrics::SUCCESS } else { metrics::FAILED },
+                            ],
+                        );
+                        metrics::observe(
+                            &metrics::METADATA_SERVICE_POLL_DURATION,
+                            poll_duration.as_secs_f64(),
+                        );
+
+                        // Always proceed to build VotingAssignments regardless of poll results.
+                        // Even if polls failed, the duties cache may have:
+                        // - Fresh data from a previous poll this slot
+                        // - Stale data from previous slot/epoch (better than nothing)
+                        // - Empty data (only if poll timed out AND this is a fresh restart)
                         if let Err(err) = self_clone_phase1.update_voting_assignments() {
                             error!(err, "Failed to update validator voting assignments");
                         }
@@ -177,7 +240,7 @@ impl<E: EthSpec, T: SlotClock + 'static> MetadataService<E, T> {
         let slot = self.slot_clock.now().ok_or("Failed to read slot clock")?;
 
         // Get attestation validators
-        let (attesting_validators, attesting_committees) = self
+        let (attesting_validators, attesting_committees): (Vec<_>, HashMap<_, _>) = self
             .duties_service
             .attesters(slot)
             .into_iter()
@@ -222,6 +285,9 @@ impl<E: EthSpec, T: SlotClock + 'static> MetadataService<E, T> {
             })
             .unwrap_or_default();
 
+        let attester_count = attesting_validators.len();
+        let sync_count = sync_validators_by_subnet.len();
+
         let voting_assignments = VotingAssignments {
             slot,
             attesting_validators,
@@ -232,7 +298,20 @@ impl<E: EthSpec, T: SlotClock + 'static> MetadataService<E, T> {
         self.validator_store
             .update_voting_assignments(voting_assignments);
 
-        trace!(%slot, "Published VotingAssignments at slot start");
+        // Record validator count metrics
+        metrics::set_gauge(
+            &metrics::METADATA_SERVICE_ATTESTING_VALIDATORS,
+            attester_count as i64,
+        );
+        metrics::set_gauge(
+            &metrics::METADATA_SERVICE_SYNC_VALIDATORS,
+            sync_count as i64,
+        );
+        if attester_count == 0 && sync_count == 0 {
+            metrics::inc_counter(&metrics::METADATA_SERVICE_EMPTY_ASSIGNMENTS_TOTAL);
+        }
+
+        trace!(%slot, attester_count, sync_count, "Published VotingAssignments at slot start");
         Ok(())
     }
 
diff --git a/anchor/validator_store/src/metrics.rs b/anchor/validator_store/src/metrics.rs
@@ -24,3 +24,56 @@ pub static SIGNED_RANDAO_REVEALS_TOTAL: LazyLock<Result<IntCounterVec>> = LazyLo
         &["status"],
     )
 });
+
+// ═══════════════════════════════════════════════════════════════════════════════
+// MetadataService metrics
+// ═══════════════════════════════════════════════════════════════════════════════
+
+// Poll outcome labels
+pub const ATTESTERS: &str = "attesters";
+pub const SYNC: &str = "sync";
+pub const SUCCESS: &str = "success";
+pub const FAILED: &str = "failed";
+
+/// Count of poll attempts by type (attesters/sync) and outcome (success/failed)
+pub static METADATA_SERVICE_POLL_TOTAL: LazyLock<Result<IntCounterVec>> = LazyLock::new(|| {
+    try_create_int_counter_vec(
+        "anchor_metadata_service_poll_total",
+        "Count of DutiesService poll wait attempts",
+        &["type", "outcome"],
+    )
+});
+
+/// Duration of poll wait phase (both polls in parallel)
+pub static METADATA_SERVICE_POLL_DURATION: LazyLock<Result<Histogram>> = LazyLock::new(|| {
+    try_create_histogram(
+        "anchor_metadata_service_poll_duration_seconds",
+        "Duration waiting for DutiesService poll signals",
+    )
+});
+
+/// Current count of attesting validators in VotingAssignments
+pub static METADATA_SERVICE_ATTESTING_VALIDATORS: LazyLock<Result<IntGauge>> =
+    LazyLock::new(|| {
+        try_create_int_gauge(
+            "anchor_metadata_service_attesting_validators",
+            "Count of validators with attestation duties this slot",
+        )
+    });
+
+/// Current count of sync committee validators in VotingAssignments
+pub static METADATA_SERVICE_SYNC_VALIDATORS: LazyLock<Result<IntGauge>> = LazyLock::new(|| {
+    try_create_int_gauge(
+        "anchor_metadata_service_sync_validators",
+        "Count of validators with sync committee duties this slot",
+    )
+});
+
+/// Count of slots where VotingAssignments was empty
+pub static METADATA_SERVICE_EMPTY_ASSIGNMENTS_TOTAL: LazyLock<Result<IntCounter>> =
+    LazyLock::new(|| {
+        try_create_int_counter(
+            "anchor_metadata_service_empty_assignments_total",
+            "Count of slots where VotingAssignments had no duties",
+        )
+    });

Original file line number	Diff line number	Diff line change
`@@ -454,7 +454,7 @@ impl SignedSSVMessage {`
`454`	`454`	`}`
`455`	`455`	`})`
`456`	`456`	`})`
`457`		`- .collect();`
	`457`	`+ .collect::<Result<Vec<_>, _>>()?;`
`458`	`458`
`459`	`459`	`// Then convert the Vec of VariableLists to VariableList<VariableList<u8, U256>, U13>`
`460`	`460`	`// This can fail if we have more than 13 signatures`