Fix wait for fully verified head for sync committee duties in local validator

Author: povi

validator/src/slot_head.rs

@@ -3,6 +3,8 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use bls::{PublicKeyBytes, SignatureBytes};
+use eth1_api::ApiController;
+use fork_choice_control::Wait;
 use futures::lock::Mutex;
 use helper_functions::{
     accessors, misc, predicates,
@@ -12,6 +14,7 @@ use itertools::Itertools as _;
 use log::warn;
 use signer::{Signer, SigningMessage, SigningTriple};
 use slashing_protection::SlashingProtector;
+use tap::Pipe as _;
 use types::{
     altair::{
         containers::{SyncAggregatorSelectionData, SyncCommitteeMessage},
@@ -75,6 +78,17 @@ impl<P: Preset> SlotHead<P> {
         self.beacon_state.phase() >= Phase::Altair
     }
 
+    pub fn is_optimistic<W: Wait>(&self, controller: &ApiController<P, W>) -> Result<bool> {
+        if !self.optimistic {
+            return Ok(false);
+        }
+
+        controller
+            .block_by_root(self.beacon_block_root)?
+            .is_none_or(|block| block.status.is_optimistic())
+            .pipe(Ok)
+    }
+
     pub fn subnet_id(&self, slot: Slot, committee_index: CommitteeIndex) -> Result<SubnetId> {
         let committees_per_slot =
             accessors::get_committee_count_per_slot(&self.beacon_state, RelativeEpoch::Current);

validator/src/validator.rs

@@ -743,7 +743,7 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
             return Ok(());
         }
 
-        if slot_head.optimistic {
+        if self.wait_for_fully_validated_head(slot_head).await.is_err() {
             warn!(
                 "validator cannot produce a block because \
                  chain head has not been fully verified by an execution engine",
@@ -975,31 +975,7 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
         wait_group: &W,
         slot_head: &SlotHead<P>,
     ) -> Result<()> {
-        const BLOCK_EVENT_WAIT_TIMEOUT: Duration = Duration::from_secs(1);
-
-        if slot_head.optimistic
-            && timeout(BLOCK_EVENT_WAIT_TIMEOUT, async {
-                loop {
-                    let block_event =
-                        match self.event_channels.receiver_for(Topic::Block).recv().await {
-                            Ok(Event::Block(block_event)) => block_event,
-                            Ok(_) => continue,
-                            Err(error) => {
-                                debug!("error receiving block event: {error:?}");
-                                continue;
-                            }
-                        };
-
-                    if block_event.block == slot_head.beacon_block_root
-                        && !block_event.execution_optimistic
-                    {
-                        break;
-                    }
-                }
-            })
-            .await
-            .is_err()
-        {
+        if self.wait_for_fully_validated_head(slot_head).await.is_err() {
             warn!(
                 "validator cannot participate in attestation because \
                  chain head has not been fully verified by an execution engine",
@@ -1146,7 +1122,7 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
 
     #[expect(clippy::too_many_lines)]
     async fn publish_aggregates_and_proofs(&self, wait_group: &W, slot_head: &SlotHead<P>) {
-        if slot_head.optimistic {
+        if self.wait_for_fully_validated_head(slot_head).await.is_err() {
             warn!(
                 "validators cannot participate in aggregation because \
                  chain head has not been fully verified by an execution engine",
@@ -1303,7 +1279,11 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
             return Ok(());
         }
 
-        if slot_head.optimistic {
+        if self
+            .wait_for_fully_validated_head(&slot_head)
+            .await
+            .is_err()
+        {
             warn!(
                 "validator cannot participate in sync committees because \
                  chain head has not been fully verified by an execution engine",
@@ -1360,7 +1340,7 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
             return;
         }
 
-        if slot_head.optimistic {
+        if self.wait_for_fully_validated_head(slot_head).await.is_err() {
             warn!(
                 "validator cannot participate in sync committees because \
                  chain head has not been fully verified by an execution engine",
@@ -2090,6 +2070,36 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
             validator_statistics.track_collection_metrics().await;
         }
     }
+
+    async fn wait_for_fully_validated_head(&self, slot_head: &SlotHead<P>) -> Result<()> {
+        const BLOCK_EVENT_WAIT_TIMEOUT: Duration = Duration::from_secs(1);
+
+        if !slot_head.is_optimistic(&self.controller)? {
+            return Ok(());
+        }
+
+        timeout(BLOCK_EVENT_WAIT_TIMEOUT, async {
+            loop {
+                let block_event = match self.event_channels.receiver_for(Topic::Block).recv().await
+                {
+                    Ok(Event::Block(block_event)) => block_event,
+                    Ok(_) => continue,
+                    Err(error) => {
+                        warn!("error receiving block event: {error:?}");
+                        continue;
+                    }
+                };
+
+                if block_event.block == slot_head.beacon_block_root
+                    && !block_event.execution_optimistic
+                {
+                    break;
+                }
+            }
+        })
+        .await
+        .map_err(Into::into)
+    }
 }
 
 // Use `BTreeMap` to make grouping deterministic for snapshot testing.