Use events API to eager send attestations

validator_client/validator_services/src/attestation_service.rs

@@ -1,5 +1,7 @@
 use crate::duties_service::{DutiesService, DutyAndProof};
 use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
+use eth2::types::{EventKind, EventTopic};
+use futures::StreamExt;
 use futures::future::join_all;
 use logging::crit;
 use slot_clock::SlotClock;
@@ -160,7 +162,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         let interval_fut = async move {
             loop {
                 if let Some(duration_to_next_slot) = self.slot_clock.duration_to_next_slot() {
-                    sleep(duration_to_next_slot + slot_duration / 3).await;
+                    tokio::select! {
+                        _ = sleep(duration_to_next_slot + slot_duration / 3) => {},
+                        _ = poll_head_event_on_all_beacon_nodes::<T, S::E>(&self.beacon_nodes) => {}
+                    }
 
                     if let Err(e) = self.spawn_attestation_tasks(slot_duration) {
                         crit!(error = e, "Failed to spawn attestation tasks")
@@ -709,6 +714,38 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     }
 }
 
+async fn poll_head_event_on_all_beacon_nodes<T: SlotClock, E: EthSpec>(
+    beacon_nodes: &Arc<BeaconNodeFallback<T>>,
+) -> Result<(), String> {
+    beacon_nodes
+        .first_success(|beacon_node| async move {
+            let mut event_stream = beacon_node
+                .get_events::<E>(&[EventTopic::Head])
+                .await
+                .map_err(|e| format!("Failed to get event stream: {:?}", e))?;
+
+            // Poll once for a head event to trigger early attestation processing
+            if let Some(event_result) = event_stream.next().await {
+                let event =
+                    event_result.map_err(|e| format!("Head event stream error: {:?}", e))?;
+                match event {
+                    EventKind::Head(_) => {
+                        trace!("Received head event, triggering early attestation processing");
+                        Ok(())
+                    }
+                    _ => Err("Received non-head event when expecting head event".to_string()),
+                }
+            } else {
+                Err("No head events received".to_string())
+            }
+        })
+        .await
+        .map_err(|e| {
+            debug!(error = %e, "Failed to get head events from any beacon node");
+            e.to_string()
+        })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;