diff --git a/Cargo.lock b/Cargo.lock
index 622ff881837..78b0f335901 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9792,8 +9792,11 @@ dependencies = [
  "futures",
  "graffiti_file",
  "logging",
+ "mockito",
  "parking_lot",
+ "regex",
  "safe_arith",
+ "serde_json",
  "slot_clock",
  "task_executor",
  "tokio",
diff --git a/book/src/help_vc.md b/book/src/help_vc.md
index 2a9936d1d2f..751610931fd 100644
--- a/book/src/help_vc.md
+++ b/book/src/help_vc.md
@@ -205,6 +205,13 @@ Flags:
       --distributed
           Enables functionality required for running the validator in a
           distributed validator cluster.
+      --enable-beacon-head-monitor
+          Enable the beacon head monitor so fallback head updates trigger duties
+          when a lagging        primary is detected. This keeps the attestation
+          service responsive when using multiple beacon nodes, but it relies on
+          the fallback service streaming head events which may increase network
+          usage. When not enabled (default), duties are only triggered on slot
+          boundaries and ignore fallback head changes.
       --enable-doppelganger-protection
           If this flag is set, Lighthouse will delay startup for three epochs
           and monitor for messages on the network by any of the validators
diff --git a/common/eth2/src/lib.rs b/common/eth2/src/lib.rs
index 8746e3c063c..b2d998726a3 100644
--- a/common/eth2/src/lib.rs
+++ b/common/eth2/src/lib.rs
@@ -152,6 +152,7 @@ pub struct BeaconNodeHttpClient {
     client: reqwest::Client,
     server: SensitiveUrl,
     timeouts: Timeouts,
+    index: usize,
 }
 
 impl Eq for BeaconNodeHttpClient {}
@@ -168,6 +169,20 @@ impl BeaconNodeHttpClient {
             client: reqwest::Client::new(),
             server,
             timeouts,
+            index: 0,
+        }
+    }
+
+    pub fn index(&self) -> usize {
+        self.index
+    }
+
+    pub fn new_with_index(server: SensitiveUrl, timeouts: Timeouts, index: usize) -> Self {
+        Self {
+            client: reqwest::Client::new(),
+            server,
+            timeouts,
+            index,
         }
     }
 
@@ -175,11 +190,13 @@ impl BeaconNodeHttpClient {
         server: SensitiveUrl,
         client: reqwest::Client,
         timeouts: Timeouts,
+        index: usize,
     ) -> Self {
         Self {
             client,
             server,
             timeouts,
+            index,
         }
     }
     // Returns a reference to the `SensitiveUrl` of the server.
diff --git a/consensus/types/src/core/slot_epoch.rs b/consensus/types/src/core/slot_epoch.rs
index 97457701b11..cbf9b31a8a5 100644
--- a/consensus/types/src/core/slot_epoch.rs
+++ b/consensus/types/src/core/slot_epoch.rs
@@ -59,6 +59,10 @@ impl Slot {
     pub fn max_value() -> Slot {
         Slot(u64::MAX)
     }
+
+    pub fn is_start_slot_in_epoch(&self, slots_per_epoch: u64) -> bool {
+        self.0.is_multiple_of(slots_per_epoch)
+    }
 }
 
 impl Epoch {
diff --git a/testing/node_test_rig/src/lib.rs b/testing/node_test_rig/src/lib.rs
index e49d11ee1eb..6ad5400ab67 100644
--- a/testing/node_test_rig/src/lib.rs
+++ b/testing/node_test_rig/src/lib.rs
@@ -89,6 +89,7 @@ impl<E: EthSpec> LocalBeaconNode<E> {
             beacon_node_url,
             beacon_node_http_client,
             Timeouts::set_all(HTTP_TIMEOUT),
+            0,
         ))
     }
 }
diff --git a/testing/validator_test_rig/src/mock_beacon_node.rs b/testing/validator_test_rig/src/mock_beacon_node.rs
index ff1e772d544..7b7fa8e64ca 100644
--- a/testing/validator_test_rig/src/mock_beacon_node.rs
+++ b/testing/validator_test_rig/src/mock_beacon_node.rs
@@ -18,12 +18,13 @@ pub struct MockBeaconNode<E: EthSpec> {
 }
 
 impl<E: EthSpec> MockBeaconNode<E> {
-    pub async fn new() -> Self {
+    pub async fn new(index: usize) -> Self {
         // mock server logging
         let server = Server::new_async().await;
-        let beacon_api_client = BeaconNodeHttpClient::new(
+        let beacon_api_client = BeaconNodeHttpClient::new_with_index(
             SensitiveUrl::from_str(&server.url()).unwrap(),
             Timeouts::set_all(Duration::from_secs(1)),
+            index,
         );
         Self {
             server,
diff --git a/validator_client/beacon_node_fallback/src/beacon_head_monitor.rs b/validator_client/beacon_node_fallback/src/beacon_head_monitor.rs
new file mode 100644
index 00000000000..4cdf78c0942
--- /dev/null
+++ b/validator_client/beacon_node_fallback/src/beacon_head_monitor.rs
@@ -0,0 +1,372 @@
+use crate::BeaconNodeFallback;
+use eth2::types::{EventKind, EventTopic, SseHead};
+use futures::StreamExt;
+use slot_clock::SlotClock;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use tracing::{info, warn};
+use types::EthSpec;
+
+type CacheHashMap = HashMap<usize, SseHead>;
+
+// This is used send the index derived from `CandidateBeaconNode` to the
+// `AttestationService` for further processing
+#[derive(Debug)]
+pub struct HeadEvent {
+    pub beacon_node_index: usize,
+}
+
+/// Cache to maintain the latest head received from each of the beacon nodes
+/// in the `BeaconNodeFallback`.
+#[derive(Debug)]
+pub struct BeaconHeadCache {
+    cache: RwLock<CacheHashMap>,
+}
+
+impl BeaconHeadCache {
+    pub fn new() -> Self {
+        Self {
+            cache: RwLock::new(HashMap::new()),
+        }
+    }
+
+    pub async fn get(&self, beacon_node_index: usize) -> Option<SseHead> {
+        self.cache.read().await.get(&beacon_node_index).cloned()
+    }
+
+    pub async fn insert(&self, beacon_node_index: usize, head: SseHead) {
+        self.cache.write().await.insert(beacon_node_index, head);
+    }
+
+    pub async fn is_latest(&self, head: &SseHead) -> bool {
+        let cache = self.cache.read().await;
+        cache
+            .values()
+            .all(|cache_head| head.slot >= cache_head.slot)
+    }
+
+    pub async fn purge_cache(&self) {
+        self.cache.write().await.clear();
+    }
+}
+
+impl Default for BeaconHeadCache {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// Runs a non-terminating loop to update the `BeaconHeadCache` with the latest head received
+// from the candidate beacon_nodes. This is an attempt to stream events to beacon nodes and
+// potential start attestion duties earlier as soon as latest head is receive from any of the
+// beacon node in contrast to attest at the 1/3rd mark in the slot.
+//
+//
+// The cache and the candidate BNs list are refresh/purged to avoid dangling reference conditions
+// that arise due to `update_candidates_list`.
+//
+// Starts the service to perpetually stream head events from connected beacon_nodes
+pub async fn poll_head_event_from_beacon_nodes<E: EthSpec, T: SlotClock + 'static>(
+    beacon_nodes: Arc<BeaconNodeFallback<T>>,
+) -> Result<(), String> {
+    let head_cache = beacon_nodes
+        .beacon_head_cache
+        .clone()
+        .expect("Unable to start head monitor without beacon_head_cache");
+    let head_monitor_send = beacon_nodes
+        .head_monitor_send
+        .clone()
+        .expect("Unable to start head monitor without head_monitor_send");
+
+    info!("Starting head monitoring service");
+    let candidates = {
+        let candidates_guard = beacon_nodes.candidates.read().await;
+        candidates_guard.clone()
+    };
+    let mut tasks = vec![];
+
+    for candidate in candidates.iter() {
+        let head_event_stream = candidate
+            .beacon_node
+            .get_events::<E>(&[EventTopic::Head])
+            .await;
+
+        let mut head_event_stream = match head_event_stream {
+            Ok(stream) => stream,
+            Err(e) => {
+                warn!("failed to get head event stream: {:?}", e);
+                continue;
+            }
+        };
+
+        let sender_tx = head_monitor_send.clone();
+        let head_cache_ref = head_cache.clone();
+
+        let stream_fut = async move {
+            while let Some(event_result) = head_event_stream.next().await {
+                if let Ok(EventKind::Head(head)) = event_result {
+                    head_cache_ref
+                        .insert(candidate.beacon_node.index(), head.clone())
+                        .await;
+
+                    if !head_cache_ref.is_latest(&head).await {
+                        continue;
+                    }
+
+                    if sender_tx
+                        .send(HeadEvent {
+                            beacon_node_index: candidate.beacon_node.index(),
+                        })
+                        .await
+                        .is_err()
+                    {
+                        warn!("Head monitoring service channel closed");
+                    }
+                }
+            }
+        };
+
+        tasks.push(stream_fut);
+    }
+
+    if tasks.is_empty() {
+        head_cache.purge_cache().await;
+        return Err(
+            "No beacon nodes available for head event streaming, retry in sometime".to_string(),
+        );
+    }
+
+    futures::future::join_all(tasks).await;
+
+    drop(candidates);
+    head_cache.purge_cache().await;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bls::FixedBytesExtended;
+    use types::Hash256;
+
+    fn create_sse_head(slot: u64, block_root: u8) -> SseHead {
+        SseHead {
+            slot: types::Slot::new(slot),
+            block: Hash256::from_low_u64_be(block_root as u64),
+            state: Hash256::from_low_u64_be(block_root as u64),
+            epoch_transition: false,
+            previous_duty_dependent_root: Hash256::from_low_u64_be(block_root as u64),
+            current_duty_dependent_root: Hash256::from_low_u64_be(block_root as u64),
+            execution_optimistic: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn test_beacon_head_cache_insertion_and_retrieval() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(1, 1);
+        let head_2 = create_sse_head(2, 2);
+
+        cache.insert(0, head_1.clone()).await;
+        cache.insert(1, head_2.clone()).await;
+
+        assert_eq!(cache.get(0).await, Some(head_1));
+        assert_eq!(cache.get(1).await, Some(head_2));
+        assert_eq!(cache.get(2).await, None);
+    }
+
+    #[tokio::test]
+    async fn test_beacon_head_cache_update() {
+        let cache = BeaconHeadCache::new();
+        let head_old = create_sse_head(1, 1);
+        let head_new = create_sse_head(2, 2);
+
+        cache.insert(0, head_old).await;
+        cache.insert(0, head_new.clone()).await;
+
+        assert_eq!(cache.get(0).await, Some(head_new));
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_with_higher_slot() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(1, 1);
+        let head_2 = create_sse_head(2, 2);
+        let head_3 = create_sse_head(3, 3);
+
+        cache.insert(0, head_1).await;
+        cache.insert(1, head_2).await;
+
+        assert!(cache.is_latest(&head_3).await);
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_with_lower_slot() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(1, 1);
+        let head_2 = create_sse_head(2, 2);
+        let head_older = create_sse_head(1, 99);
+
+        cache.insert(0, head_1).await;
+        cache.insert(1, head_2).await;
+
+        assert!(!cache.is_latest(&head_older).await);
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_with_equal_slot() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(5, 1);
+        let head_2 = create_sse_head(5, 2);
+        let head_equal = create_sse_head(5, 3);
+
+        cache.insert(0, head_1).await;
+        cache.insert(1, head_2).await;
+
+        assert!(cache.is_latest(&head_equal).await);
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_empty_cache() {
+        let cache = BeaconHeadCache::new();
+        let head = create_sse_head(1, 1);
+
+        assert!(cache.is_latest(&head).await);
+    }
+
+    #[tokio::test]
+    async fn test_purge_cache_clears_all_entries() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(1, 1);
+        let head_2 = create_sse_head(2, 2);
+
+        cache.insert(0, head_1).await;
+        cache.insert(1, head_2).await;
+
+        assert!(cache.get(0).await.is_some());
+        assert!(cache.get(1).await.is_some());
+
+        cache.purge_cache().await;
+
+        assert!(cache.get(0).await.is_none());
+        assert!(cache.get(1).await.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_head_event_creation() {
+        let event = HeadEvent {
+            beacon_node_index: 42,
+        };
+        assert_eq!(event.beacon_node_index, 42);
+    }
+
+    #[tokio::test]
+    async fn test_cache_caches_multiple_heads_from_different_nodes() {
+        let cache = BeaconHeadCache::new();
+        let head_1 = create_sse_head(10, 1);
+        let head_2 = create_sse_head(5, 2);
+        let head_3 = create_sse_head(8, 3);
+
+        cache.insert(0, head_1.clone()).await;
+        cache.insert(1, head_2.clone()).await;
+        cache.insert(2, head_3.clone()).await;
+
+        // Verify all are stored
+        assert_eq!(cache.get(0).await, Some(head_1));
+        assert_eq!(cache.get(1).await, Some(head_2));
+        assert_eq!(cache.get(2).await, Some(head_3));
+
+        // The latest should be slot 10
+        let head_10 = create_sse_head(10, 99);
+        assert!(cache.is_latest(&head_10).await);
+
+        // Anything with slot > 10 should be latest
+        let head_11 = create_sse_head(11, 99);
+        assert!(cache.is_latest(&head_11).await);
+
+        // Anything with slot < 10 should not be latest
+        let head_9 = create_sse_head(9, 99);
+        assert!(!cache.is_latest(&head_9).await);
+    }
+
+    #[tokio::test]
+    async fn test_cache_handles_concurrent_operations() {
+        let cache = Arc::new(BeaconHeadCache::new());
+        let mut handles = vec![];
+
+        // Spawn multiple tasks that insert heads concurrently
+        for i in 0..10 {
+            let cache_clone = cache.clone();
+            let handle = tokio::spawn(async move {
+                let head = create_sse_head(i as u64, (i % 256) as u8);
+                cache_clone.insert(i, head).await;
+            });
+            handles.push(handle);
+        }
+
+        // Wait for all tasks to complete
+        for handle in handles {
+            handle.await.unwrap();
+        }
+
+        // Verify all heads are cached
+        for i in 0..10 {
+            assert!(cache.get(i).await.is_some());
+        }
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_after_cache_updates() {
+        let cache = BeaconHeadCache::new();
+
+        // Start with head at slot 5
+        let head_5 = create_sse_head(5, 1);
+        cache.insert(0, head_5.clone()).await;
+        assert!(cache.is_latest(&head_5).await);
+
+        // Add a higher slot
+        let head_10 = create_sse_head(10, 2);
+        cache.insert(1, head_10.clone()).await;
+
+        // head_5 should no longer be latest
+        assert!(!cache.is_latest(&head_5).await);
+        // head_10 should be latest
+        assert!(cache.is_latest(&head_10).await);
+
+        // Add an even higher slot
+        let head_15 = create_sse_head(15, 3);
+        cache.insert(2, head_15.clone()).await;
+
+        // head_10 should no longer be latest
+        assert!(!cache.is_latest(&head_10).await);
+        // head_15 should be latest
+        assert!(cache.is_latest(&head_15).await);
+    }
+
+    #[tokio::test]
+    async fn test_cache_default_is_empty() {
+        let cache = BeaconHeadCache::default();
+        assert!(cache.get(0).await.is_none());
+        assert!(cache.get(999).await.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_is_latest_with_multiple_same_slot_heads() {
+        let cache = BeaconHeadCache::new();
+        let head_slot_5_node1 = create_sse_head(5, 1);
+        let head_slot_5_node2 = create_sse_head(5, 2);
+        let head_slot_5_node3 = create_sse_head(5, 3);
+
+        cache.insert(0, head_slot_5_node1).await;
+        cache.insert(1, head_slot_5_node2).await;
+
+        // All heads with slot 5 should be considered latest
+        assert!(cache.is_latest(&head_slot_5_node3).await);
+
+        // But heads with slot 4 should not be latest
+        let head_slot_4 = create_sse_head(4, 4);
+        assert!(!cache.is_latest(&head_slot_4).await);
+    }
+}
diff --git a/validator_client/beacon_node_fallback/src/lib.rs b/validator_client/beacon_node_fallback/src/lib.rs
index 2d75df2fa34..0c130137b99 100644
--- a/validator_client/beacon_node_fallback/src/lib.rs
+++ b/validator_client/beacon_node_fallback/src/lib.rs
@@ -2,7 +2,10 @@
 //! "fallback" behaviour; it will try a request on all of the nodes until one or none of them
 //! succeed.
 
+pub mod beacon_head_monitor;
 pub mod beacon_node_health;
+
+use beacon_head_monitor::{BeaconHeadCache, HeadEvent, poll_head_event_from_beacon_nodes};
 use beacon_node_health::{
     BeaconNodeHealth, BeaconNodeSyncDistanceTiers, ExecutionEngineHealth, IsOptimistic,
     SyncDistanceTier, check_node_health,
@@ -22,7 +25,11 @@ use std::time::{Duration, Instant};
 use std::vec::Vec;
 use strum::VariantNames;
 use task_executor::TaskExecutor;
-use tokio::{sync::RwLock, time::sleep};
+
+use tokio::{
+    sync::{RwLock, mpsc},
+    time::sleep,
+};
 use tracing::{debug, error, warn};
 use types::{ChainSpec, Config as ConfigSpec, EthSpec, Slot};
 use validator_metrics::{ENDPOINT_ERRORS, ENDPOINT_REQUESTS, inc_counter_vec};
@@ -68,6 +75,24 @@ pub fn start_fallback_updater_service<T: SlotClock + 'static, E: EthSpec>(
         return Err("Cannot start fallback updater without slot clock");
     }
 
+    let beacon_nodes_ref = beacon_nodes.clone();
+
+    // the existence of head_monitor_send is overloaded with the predicate of
+    // requirement of starting the head monitoring service or not.
+    if beacon_nodes_ref.head_monitor_send.is_some() {
+        let head_monitor_future = async move {
+            loop {
+                if let Err(err) =
+                    poll_head_event_from_beacon_nodes::<E, T>(beacon_nodes_ref.clone()).await
+                {
+                    warn!(error=?err, "Head service failed");
+                }
+            }
+        };
+
+        executor.spawn(head_monitor_future, "head_monitoring");
+    }
+
     let future = async move {
         loop {
             beacon_nodes.update_all_candidates::<E>().await;
@@ -186,14 +211,13 @@ impl Serialize for CandidateInfo {
 /// for a query.
 #[derive(Clone, Debug)]
 pub struct CandidateBeaconNode {
-    pub index: usize,
     pub beacon_node: BeaconNodeHttpClient,
     pub health: Arc<RwLock<Result<BeaconNodeHealth, CandidateError>>>,
 }
 
 impl PartialEq for CandidateBeaconNode {
     fn eq(&self, other: &Self) -> bool {
-        self.index == other.index && self.beacon_node == other.beacon_node
+        self.beacon_node == other.beacon_node
     }
 }
 
@@ -201,9 +225,8 @@ impl Eq for CandidateBeaconNode {}
 
 impl CandidateBeaconNode {
     /// Instantiate a new node.
-    pub fn new(beacon_node: BeaconNodeHttpClient, index: usize) -> Self {
+    pub fn new(beacon_node: BeaconNodeHttpClient) -> Self {
         Self {
-            index,
             beacon_node,
             health: Arc::new(RwLock::new(Err(CandidateError::Uninitialized))),
         }
@@ -258,7 +281,7 @@ impl CandidateBeaconNode {
                     };
 
                     let new_health = BeaconNodeHealth::from_status(
-                        self.index,
+                        self.beacon_node.index(),
                         sync_distance,
                         head,
                         optimistic_status,
@@ -380,6 +403,8 @@ pub struct BeaconNodeFallback<T> {
     pub candidates: Arc<RwLock<Vec<CandidateBeaconNode>>>,
     distance_tiers: BeaconNodeSyncDistanceTiers,
     slot_clock: Option<T>,
+    beacon_head_cache: Option<Arc<BeaconHeadCache>>,
+    head_monitor_send: Option<Arc<mpsc::Sender<HeadEvent>>>,
     broadcast_topics: Vec<ApiTopic>,
     spec: Arc<ChainSpec>,
 }
@@ -396,6 +421,8 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
             candidates: Arc::new(RwLock::new(candidates)),
             distance_tiers,
             slot_clock: None,
+            beacon_head_cache: None,
+            head_monitor_send: None,
             broadcast_topics,
             spec,
         }
@@ -410,6 +437,15 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
         self.slot_clock = Some(slot_clock);
     }
 
+    /// This the head monitor channel that streams events from all the beacon node that the
+    /// validator client is connected in the `BeaconNodeFallback`. This is also initialize the
+    /// beacon_head_cache under the assumption the beacon_head_cache will always be needed when
+    /// head_monitor_send is set.
+    pub fn set_head_send(&mut self, head_monitor_send: Arc<mpsc::Sender<HeadEvent>>) {
+        self.head_monitor_send = Some(head_monitor_send);
+        self.beacon_head_cache = Some(Arc::new(BeaconHeadCache::new()));
+    }
+
     /// The count of candidates, regardless of their state.
     pub async fn num_total(&self) -> usize {
         self.candidates.read().await.len()
@@ -454,7 +490,7 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
             }
 
             candidate_info.push(CandidateInfo {
-                index: candidate.index,
+                index: candidate.beacon_node.index(),
                 endpoint: candidate.beacon_node.to_string(),
                 health,
             });
@@ -486,13 +522,21 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
             .into_iter()
             .enumerate()
             .map(|(index, url)| {
-                CandidateBeaconNode::new(BeaconNodeHttpClient::new(url, timeouts.clone()), index)
+                CandidateBeaconNode::new(BeaconNodeHttpClient::new_with_index(
+                    url,
+                    timeouts.clone(),
+                    index,
+                ))
             })
             .collect();
 
         let mut candidates = self.candidates.write().await;
         *candidates = new_candidates;
 
+        if let Some(cache) = &self.beacon_head_cache {
+            cache.purge_cache().await;
+        }
+
         Ok(new_list)
     }
 
@@ -596,9 +640,67 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
             .collect()
     }
 
+    /// A wrapper for `first_success_with_index` when the beacon node `index` is not needed.
+    pub async fn first_success<F, O, Err, R>(&self, func: F) -> Result<O, Errors<Err>>
+    where
+        F: Fn(BeaconNodeHttpClient) -> R,
+        R: Future<Output = Result<O, Err>>,
+        Err: Debug,
+    {
+        self.first_success_with_index(func)
+            .await
+            .map(|(val, _)| val)
+    }
+
+    pub async fn first_n_responses<F, O, Err, R, C>(
+        &self,
+        fetch_func: F,
+        mut consensus_check: C,
+    ) -> Result<(O, usize), Errors<Err>>
+    where
+        F: Fn(BeaconNodeHttpClient) -> R,
+        R: Future<Output = Result<O, Err>>,
+        C: FnMut(&(O, usize)) -> bool,
+        O: Eq + Clone + Debug,
+        Err: Debug,
+    {
+        let mut errors = vec![];
+
+        // Collect all responses from all candidates
+        let candidates = self.candidates.read().await;
+        let mut futures = vec![];
+
+        for candidate in candidates.iter() {
+            futures.push(Self::run_on_candidate(
+                candidate.beacon_node.clone(),
+                &fetch_func,
+            ));
+        }
+        drop(candidates);
+
+        // Process futures sequentially, checking consensus after each response
+        for future in futures {
+            match future.await {
+                Ok(val) => {
+                    if consensus_check(&val) {
+                        return Ok(val);
+                    }
+                }
+                Err(e) => {
+                    errors.push(e);
+                }
+            }
+        }
+
+        Err(Errors(errors))
+    }
+
     /// Run `func` against each candidate in `self`, returning immediately if a result is found.
     /// Otherwise, return all the errors encountered along the way.
-    pub async fn first_success<F, O, Err, R>(&self, func: F) -> Result<O, Errors<Err>>
+    pub async fn first_success_with_index<F, O, Err, R>(
+        &self,
+        func: F,
+    ) -> Result<(O, usize), Errors<Err>>
     where
         F: Fn(BeaconNodeHttpClient) -> R,
         R: Future<Output = Result<O, Err>>,
@@ -646,11 +748,46 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
         Err(Errors(errors))
     }
 
+    /// Try `func` on a specific beacon node by index first, then fall back to the normal order.
+    /// Returns immediately if the preferred node succeeds, otherwise falls back to first_success.
+    /// This is an insurance against potential race conditions that may arise.
+    pub async fn first_success_from_index<F, O, Err, R>(
+        &self,
+        preferred_index: Option<usize>,
+        func: F,
+    ) -> Result<(O, usize), Errors<Err>>
+    where
+        F: Fn(BeaconNodeHttpClient) -> R + Clone,
+        R: Future<Output = Result<O, Err>>,
+        Err: Debug,
+    {
+        // Try the preferred beacon node first if it exists
+        if let Some(preferred_idx) = preferred_index
+            && let candidates = self.candidates.read().await
+            && let Some(preferred_candidate) = candidates
+                .iter()
+                .find(|c| c.beacon_node.index() == preferred_idx)
+        {
+            let preferred_node = preferred_candidate.beacon_node.clone();
+            drop(candidates);
+
+            match Self::run_on_candidate(preferred_node, &func).await {
+                Ok(val) => return Ok(val),
+                Err(_) => {
+                    return self.first_success_with_index(func).await;
+                }
+            }
+        }
+
+        // Fall back to normal first_success behavior
+        self.first_success_with_index(func).await
+    }
+
     /// Run the future `func` on `candidate` while reporting metrics.
     async fn run_on_candidate<F, R, Err, O>(
         candidate: BeaconNodeHttpClient,
         func: F,
-    ) -> Result<O, (String, Error<Err>)>
+    ) -> Result<(O, usize), (String, Error<Err>)>
     where
         F: Fn(BeaconNodeHttpClient) -> R,
         R: Future<Output = Result<O, Err>>,
@@ -661,7 +798,7 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
         // There exists a race condition where `func` may be called when the candidate is
         // actually not ready. We deem this an acceptable inefficiency.
         match func(candidate.clone()).await {
-            Ok(val) => Ok(val),
+            Ok(val) => Ok((val, candidate.index())),
             Err(e) => {
                 debug!(
                     node = %candidate,
@@ -807,11 +944,12 @@ mod tests {
         let execution_status = ExecutionEngineHealth::Healthy;
 
         fn new_candidate(index: usize) -> CandidateBeaconNode {
-            let beacon_node = BeaconNodeHttpClient::new(
+            let beacon_node = BeaconNodeHttpClient::new_with_index(
                 SensitiveUrl::parse(&format!("http://example_{index}.com")).unwrap(),
                 Timeouts::set_all(Duration::from_secs(index as u64)),
+                index,
             );
-            CandidateBeaconNode::new(beacon_node, index)
+            CandidateBeaconNode::new(beacon_node)
         }
 
         let candidate_1 = new_candidate(1);
@@ -914,11 +1052,10 @@ mod tests {
         index: usize,
         spec: &ChainSpec,
     ) -> (MockBeaconNode<E>, CandidateBeaconNode) {
-        let mut mock_beacon_node = MockBeaconNode::<E>::new().await;
+        let mut mock_beacon_node = MockBeaconNode::<E>::new(index).await;
         mock_beacon_node.mock_config_spec(spec);
 
-        let beacon_node =
-            CandidateBeaconNode::new(mock_beacon_node.beacon_api_client.clone(), index);
+        let beacon_node = CandidateBeaconNode::new(mock_beacon_node.beacon_api_client.clone());
 
         (mock_beacon_node, beacon_node)
     }
@@ -1073,4 +1210,137 @@ mod tests {
         mock1.expect(3).assert();
         mock2.expect(3).assert();
     }
+
+    #[tokio::test]
+    async fn first_success_from_index_tries_preferred_node_first() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let (mut mock_beacon_node_1, beacon_node_1) = new_mock_beacon_node(0, &spec).await;
+        let (mut mock_beacon_node_2, beacon_node_2) = new_mock_beacon_node(1, &spec).await;
+        let (mut mock_beacon_node_3, beacon_node_3) = new_mock_beacon_node(2, &spec).await;
+
+        let beacon_node_fallback = create_beacon_node_fallback(
+            vec![beacon_node_1, beacon_node_2, beacon_node_3],
+            vec![],
+            spec.clone(),
+        );
+
+        let mock1 = mock_beacon_node_1.mock_offline_node();
+        let _mock2 = mock_beacon_node_2.mock_online_node();
+        let mock3 = mock_beacon_node_3.mock_online_node();
+
+        // Request with preferred_index=1 (beacon_node_2)
+        let result = beacon_node_fallback
+            .first_success_from_index(
+                Some(1),
+                |client| async move { client.get_node_version().await },
+            )
+            .await;
+
+        // Should succeed since beacon_node_2 is online
+        assert!(result.is_ok());
+
+        // mock1 should not be called since preferred node succeeds
+        mock1.expect(0).assert();
+        mock3.expect(0).assert();
+    }
+
+    #[tokio::test]
+    async fn first_success_from_index_falls_back_when_preferred_fails() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let (mut mock_beacon_node_1, beacon_node_1) = new_mock_beacon_node(0, &spec).await;
+        let (mut mock_beacon_node_2, beacon_node_2) = new_mock_beacon_node(1, &spec).await;
+        let (mut mock_beacon_node_3, beacon_node_3) = new_mock_beacon_node(2, &spec).await;
+
+        let beacon_node_fallback = create_beacon_node_fallback(
+            vec![beacon_node_1, beacon_node_2, beacon_node_3],
+            vec![],
+            spec.clone(),
+        );
+
+        let _mock1 = mock_beacon_node_1.mock_online_node();
+        let mock2 = mock_beacon_node_2.mock_offline_node();
+        let _mock3 = mock_beacon_node_3.mock_offline_node();
+
+        // Request with preferred_index=1 (beacon_node_2), but it's offline
+        let result = beacon_node_fallback
+            .first_success_from_index(
+                Some(1),
+                |client| async move { client.get_node_version().await },
+            )
+            .await;
+
+        // Should succeed by falling back to beacon_node_1
+        assert!(result.is_ok());
+
+        // mock2 should be called at least once (the preferred attempt)
+        mock2.expect(1).assert();
+        // since the result was ok we can safely assume that the fallback first_success
+        // behaviour succeeded instead of checking if either of mock1/mock3 received hits
+    }
+
+    #[tokio::test]
+    async fn first_success_from_index_with_none_falls_back_to_first_success() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let (mut mock_beacon_node_1, beacon_node_1) = new_mock_beacon_node(0, &spec).await;
+        let (mut mock_beacon_node_2, beacon_node_2) = new_mock_beacon_node(1, &spec).await;
+        let (mut mock_beacon_node_3, beacon_node_3) = new_mock_beacon_node(2, &spec).await;
+
+        let beacon_node_fallback = create_beacon_node_fallback(
+            vec![beacon_node_1, beacon_node_2, beacon_node_3],
+            vec![],
+            spec.clone(),
+        );
+
+        let _mock1 = mock_beacon_node_1.mock_offline_node();
+        let _mock2 = mock_beacon_node_2.mock_offline_node();
+        let mock3 = mock_beacon_node_3.mock_online_node();
+
+        // Request with preferred_index=None
+        let result = beacon_node_fallback
+            .first_success_from_index(
+                None,
+                |client| async move { client.get_node_version().await },
+            )
+            .await;
+
+        // Should succeed with beacon_node_3 in the first pass
+        assert!(result.is_ok());
+
+        // mock3 should be called once in the first pass
+        mock3.expect(1).assert();
+    }
+
+    #[tokio::test]
+    async fn first_success_from_index_all_offline() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let (mut mock_beacon_node_1, beacon_node_1) = new_mock_beacon_node(0, &spec).await;
+        let (mut mock_beacon_node_2, beacon_node_2) = new_mock_beacon_node(1, &spec).await;
+        let (mut mock_beacon_node_3, beacon_node_3) = new_mock_beacon_node(2, &spec).await;
+
+        let beacon_node_fallback = create_beacon_node_fallback(
+            vec![beacon_node_1, beacon_node_2, beacon_node_3],
+            vec![],
+            spec.clone(),
+        );
+
+        let _mock1 = mock_beacon_node_1.mock_offline_node();
+        let mock2 = mock_beacon_node_2.mock_offline_node();
+        let _mock3 = mock_beacon_node_3.mock_offline_node();
+
+        // Request with preferred_index=1, but all nodes are offline
+        let result = beacon_node_fallback
+            .first_success_from_index(
+                Some(1),
+                |client| async move { client.get_node_version().await },
+            )
+            .await;
+
+        // Should fail since all nodes are offline
+        assert!(result.is_err());
+
+        // Preferred node (mock2) should be called 3 times:
+        // - 1 time for the preferred attempt
+        // - 2 more times from the fallback to first_success (first and second pass)
+        mock2.expect(3).assert();
+    }
 }
diff --git a/validator_client/http_api/src/lib.rs b/validator_client/http_api/src/lib.rs
index a35b4ec6c6d..f8cf0d92acd 100644
--- a/validator_client/http_api/src/lib.rs
+++ b/validator_client/http_api/src/lib.rs
@@ -437,7 +437,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
                 let mut beacon_nodes = Vec::new();
                 for node in &*block_filter.beacon_nodes.candidates.read().await {
                     beacon_nodes.push(CandidateInfo {
-                        index: node.index,
+                        index: node.beacon_node.index(),
                         endpoint: node.beacon_node.to_string(),
                         health: *node.health.read().await,
                     });
@@ -448,7 +448,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
                     let mut proposer_nodes = Vec::new();
                     for node in &*proposer_nodes_list.candidates.read().await {
                         proposer_nodes.push(CandidateInfo {
-                            index: node.index,
+                            index: node.beacon_node.index(),
                             endpoint: node.beacon_node.to_string(),
                             health: *node.health.read().await,
                         });
diff --git a/validator_client/src/cli.rs b/validator_client/src/cli.rs
index 3e1c46097f0..7deac420641 100644
--- a/validator_client/src/cli.rs
+++ b/validator_client/src/cli.rs
@@ -476,6 +476,17 @@ pub struct ValidatorClient {
     )]
     pub beacon_nodes_sync_tolerances: Vec<u64>,
 
+    #[clap(
+        long,
+        help = "Enable the beacon head monitor so fallback head updates trigger duties when a lagging        primary is detected. \
+                This keeps the attestation service responsive when using multiple beacon nodes, but it relies on the \
+                fallback service streaming head events which may increase network usage. \
+                When not enabled (default), duties are only triggered on slot boundaries and ignore fallback head changes.",
+        display_order = 0,
+        help_heading = FLAG_HEADER
+    )]
+    pub enable_beacon_head_monitor: bool,
+
     #[clap(
         long,
         help = "Disable Lighthouse's slashing protection for all web3signer keys. This can \
diff --git a/validator_client/src/config.rs b/validator_client/src/config.rs
index 1a286a74dc1..d89674608ab 100644
--- a/validator_client/src/config.rs
+++ b/validator_client/src/config.rs
@@ -82,6 +82,9 @@ pub struct Config {
     pub broadcast_topics: Vec<ApiTopic>,
     /// Enables a service which attempts to measure latency between the VC and BNs.
     pub enable_latency_measurement_service: bool,
+    /// Enables the beacon head monitor that reacts to fallback head updates.
+    #[serde(default)]
+    pub enable_beacon_head_monitor: bool,
     /// Defines the number of validators per `validator/register_validator` request sent to the BN.
     pub validator_registration_batch_size: usize,
     /// Whether we are running with distributed network support.
@@ -132,6 +135,7 @@ impl Default for Config {
             builder_registration_timestamp_override: None,
             broadcast_topics: vec![ApiTopic::Subscriptions],
             enable_latency_measurement_service: true,
+            enable_beacon_head_monitor: false,
             validator_registration_batch_size: 500,
             distributed: false,
             initialized_validators: <_>::default(),
@@ -377,6 +381,7 @@ impl Config {
         config.validator_store.builder_boost_factor = validator_client_config.builder_boost_factor;
         config.enable_latency_measurement_service =
             !validator_client_config.disable_latency_measurement_service;
+        config.enable_beacon_head_monitor = validator_client_config.enable_beacon_head_monitor;
 
         config.validator_registration_batch_size =
             validator_client_config.validator_registration_batch_size;
diff --git a/validator_client/src/lib.rs b/validator_client/src/lib.rs
index b3cd3425f3d..16cac92a3cf 100644
--- a/validator_client/src/lib.rs
+++ b/validator_client/src/lib.rs
@@ -9,10 +9,12 @@ use metrics::set_gauge;
 use monitoring_api::{MonitoringHttpClient, ProcessType};
 use sensitive_url::SensitiveUrl;
 use slashing_protection::{SLASHING_PROTECTION_FILENAME, SlashingDatabase};
+use tokio::sync::Mutex;
 
 use account_utils::validator_definitions::ValidatorDefinitions;
 use beacon_node_fallback::{
-    BeaconNodeFallback, CandidateBeaconNode, start_fallback_updater_service,
+    BeaconNodeFallback, CandidateBeaconNode, beacon_head_monitor::HeadEvent,
+    start_fallback_updater_service,
 };
 use clap::ArgMatches;
 use doppelganger_service::DoppelgangerService;
@@ -70,6 +72,8 @@ pub const AGGREGATION_PRE_COMPUTE_EPOCHS: u64 = 2;
 /// Number of slots in advance to compute sync selection proofs when in `distributed` mode.
 pub const AGGREGATION_PRE_COMPUTE_SLOTS_DISTRIBUTED: u64 = 1;
 
+const MAX_HEAD_EVENT_QUEUE_LEN: usize = 1_024;
+
 type ValidatorStore<E> = LighthouseValidatorStore<SystemTimeSlotClock, E>;
 
 #[derive(Clone)]
@@ -298,6 +302,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
                 url.clone(),
                 beacon_node_http_client,
                 timeouts,
+                i,
             ))
         };
 
@@ -320,8 +325,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
         // the node in `--beacon_nodes`.
         let candidates = beacon_nodes
             .into_iter()
-            .enumerate()
-            .map(|(index, node)| CandidateBeaconNode::new(node, index))
+            .map(CandidateBeaconNode::new)
             .collect();
 
         let proposer_nodes_num = proposer_nodes.len();
@@ -329,8 +333,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
         // the node in `--proposer_nodes`.
         let proposer_candidates = proposer_nodes
             .into_iter()
-            .enumerate()
-            .map(|(index, node)| CandidateBeaconNode::new(node, index))
+            .map(CandidateBeaconNode::new)
             .collect();
 
         // Set the count for beacon node fallbacks excluding the primary beacon node.
@@ -395,6 +398,17 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
         beacon_nodes.set_slot_clock(slot_clock.clone());
         proposer_nodes.set_slot_clock(slot_clock.clone());
 
+        // Only the beacon_nodes are used for attestation duties and thus biconditionally
+        // proposer_nodes do not need head_send ref.
+        let head_monitor_rx = if config.enable_beacon_head_monitor {
+            let (head_monitor_tx, head_receiver) =
+                mpsc::channel::<HeadEvent>(MAX_HEAD_EVENT_QUEUE_LEN);
+            beacon_nodes.set_head_send(Arc::new(head_monitor_tx));
+            Some(Arc::new(Mutex::new(head_receiver)))
+        } else {
+            None
+        };
+
         let beacon_nodes = Arc::new(beacon_nodes);
         start_fallback_updater_service::<_, E>(context.executor.clone(), beacon_nodes.clone())?;
 
@@ -505,15 +519,18 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
 
         let block_service = block_service_builder.build()?;
 
-        let attestation_service = AttestationServiceBuilder::new()
+        let attestation_builder = AttestationServiceBuilder::new()
             .duties_service(duties_service.clone())
             .slot_clock(slot_clock.clone())
             .validator_store(validator_store.clone())
             .beacon_nodes(beacon_nodes.clone())
             .executor(context.executor.clone())
+            .head_monitor_rx(head_monitor_rx)
+            .consensus_threshold(3)
             .chain_spec(context.eth2_config.spec.clone())
-            .disable(config.disable_attesting)
-            .build()?;
+            .disable(config.disable_attesting);
+
+        let attestation_service = attestation_builder.build()?;
 
         let preparation_service = PreparationServiceBuilder::new()
             .slot_clock(slot_clock.clone())
diff --git a/validator_client/validator_services/Cargo.toml b/validator_client/validator_services/Cargo.toml
index c9149409148..5fc245cd9b9 100644
--- a/validator_client/validator_services/Cargo.toml
+++ b/validator_client/validator_services/Cargo.toml
@@ -22,3 +22,8 @@ tree_hash = { workspace = true }
 types = { workspace = true }
 validator_metrics = { workspace = true }
 validator_store = { workspace = true }
+
+[dev-dependencies]
+mockito = { workspace = true }
+regex = { workspace = true }
+serde_json = { workspace = true }
diff --git a/validator_client/validator_services/src/attestation_data_service.rs b/validator_client/validator_services/src/attestation_data_service.rs
new file mode 100644
index 00000000000..a2eecd0c32a
--- /dev/null
+++ b/validator_client/validator_services/src/attestation_data_service.rs
@@ -0,0 +1,418 @@
+use std::{collections::HashMap, sync::Arc};
+
+use beacon_node_fallback::BeaconNodeFallback;
+use safe_arith::SafeArith;
+use slot_clock::SlotClock;
+use tracing::{Instrument, info_span};
+use types::{AttestationData, Checkpoint, Epoch, Slot};
+
+#[derive(Debug, Clone)]
+pub enum AttestationDataStrategy {
+    Fallback,
+    ByIndex(usize),
+    Consensus((usize, Option<(Checkpoint, usize)>)),
+    HighestScore,
+    IgnoreEpoch(Epoch),
+}
+
+// New trait for aggregation strategies that need parallel queries
+trait ResultAggregator {
+    /// Process a single response and decide whether to continue or stop
+    fn process_result(&mut self, attestation_data: &AttestationData, index: usize) -> bool; // Returns true if we should stop and return
+
+    /// Get the final result after aggregation
+    fn get_result(&self) -> Option<(AttestationData, usize)>;
+}
+
+// Consensus aggregator
+struct ConsensusAggregator {
+    results: HashMap<Checkpoint, Vec<usize>>,
+    threshold: usize,
+    target_checkpoint_and_index: Option<(Checkpoint, usize)>,
+    consensus_result: Option<(AttestationData, usize)>,
+}
+
+impl ConsensusAggregator {
+    fn new(threshold: usize, target_checkpoint_and_index: Option<(Checkpoint, usize)>) -> Self {
+        Self {
+            results: HashMap::new(),
+            threshold,
+            target_checkpoint_and_index,
+            consensus_result: None,
+        }
+    }
+}
+
+impl ResultAggregator for ConsensusAggregator {
+    fn process_result(&mut self, attestation_data: &AttestationData, index: usize) -> bool {
+        if let Some((target_checkpoint, preferred_index)) = self.target_checkpoint_and_index {
+            // If we have a preferred index set, return attestation data from it
+            // TODO(attestation-consensus) this is a small optimization to immediately return data
+            // from the preferred index. We shouldn't need to check the target checkpoint, but maybe
+            // its just safer to do so?
+            if preferred_index == index {
+                self.consensus_result = Some((attestation_data.clone(), index));
+                return true;
+            }
+            // return if fetched data matches the target checkpoint
+            if attestation_data.target == target_checkpoint {
+                self.consensus_result = Some((attestation_data.clone(), index));
+                return true;
+            }
+        }
+        self.results
+            .entry(attestation_data.target)
+            .or_default()
+            .push(index);
+
+        if self
+            .results
+            .get(&attestation_data.target)
+            .is_some_and(|servers| servers.len() >= self.threshold)
+        {
+            // Consensus has been reached
+            self.consensus_result = Some((attestation_data.clone(), index));
+            return true;
+        }
+
+        false
+    }
+
+    fn get_result(&self) -> Option<(AttestationData, usize)> {
+        self.consensus_result.clone()
+    }
+}
+
+// Score aggregator
+struct ScoreAggregator {
+    results: HashMap<usize, (u64, AttestationData)>,
+    // TODO im pretty sure the head slot is just the requested slot
+    // double check the attestation service before deleting this TODO.
+    head_slot: Slot,
+    responses_needed: usize,
+    responses_received: usize,
+}
+
+impl ScoreAggregator {
+    fn new(head_slot: Slot, responses_needed: usize) -> Self {
+        Self {
+            results: HashMap::new(),
+            head_slot,
+            responses_needed,
+            responses_received: 0,
+        }
+    }
+
+    fn calculate_score(&self, attestation_data: &AttestationData) -> u64 {
+        let checkpoint_value = attestation_data.source.epoch + attestation_data.target.epoch;
+        let slot_value = 1 + attestation_data.slot.as_u64() - self.head_slot.as_u64();
+        // TODO unwrap
+        checkpoint_value.as_u64() + 1.safe_div(slot_value).unwrap()
+    }
+}
+
+impl ResultAggregator for ScoreAggregator {
+    fn process_result(&mut self, attestation_data: &AttestationData, index: usize) -> bool {
+        let score = self.calculate_score(attestation_data);
+        self.results
+            .insert(index, (score, attestation_data.clone()));
+        self.responses_received += 1;
+
+        // Stop when we've received enough responses
+        self.responses_received >= self.responses_needed
+    }
+
+    fn get_result(&self) -> Option<(AttestationData, usize)> {
+        self.results
+            .iter()
+            .max_by_key(|(_, (score, _))| score)
+            .map(|(idx, (_, data))| (data.clone(), *idx))
+    }
+}
+
+/// The AttestationDataService is responsible for downloading and caching attestation data at a given slot.
+/// It also helps prevent us from re-downloading identical attestation data.
+pub struct AttestationDataService<T> {
+    beacon_nodes: Arc<BeaconNodeFallback<T>>,
+}
+
+impl<T: SlotClock> AttestationDataService<T> {
+    pub fn new(beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
+        Self { beacon_nodes }
+    }
+
+    async fn data_by_index(
+        &self,
+        request_slot: &Slot,
+        candidate_beacon_node: Option<usize>,
+    ) -> Result<(AttestationData, usize), String> {
+        self.beacon_nodes
+            .first_success_from_index(candidate_beacon_node, |beacon_node| async move {
+                let _timer = validator_metrics::start_timer_vec(
+                    &validator_metrics::ATTESTATION_SERVICE_TIMES,
+                    &[validator_metrics::ATTESTATIONS_HTTP_GET],
+                );
+                beacon_node
+                    .get_validator_attestation_data(*request_slot, 0)
+                    .await
+                    .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
+                    .map(|result| result.data)
+            })
+            .instrument(info_span!("fetch_attestation_data"))
+            .await
+            .map_err(|e| e.to_string())
+    }
+
+    async fn data_with_aggregation(
+        &self,
+        request_slot: &Slot,
+        mut aggregator: impl ResultAggregator,
+    ) -> Result<(AttestationData, usize), String> {
+        self.beacon_nodes
+            .first_n_responses(
+                |beacon_node| async move {
+                    let _timer = validator_metrics::start_timer_vec(
+                        &validator_metrics::ATTESTATION_SERVICE_TIMES,
+                        &[validator_metrics::ATTESTATIONS_HTTP_GET],
+                    );
+                    beacon_node
+                        .get_validator_attestation_data(*request_slot, 0)
+                        .await
+                        .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
+                        .map(|result| result.data)
+                },
+                |(attestation_data, index)| aggregator.process_result(attestation_data, *index),
+            )
+            .instrument(info_span!("fetch_attestation_data"))
+            .await
+            .map_err(|e| e.to_string())?;
+
+        aggregator
+            .get_result()
+            .ok_or_else(|| "No valid attestation data found".to_string())
+    }
+
+    pub async fn download_data(
+        &self,
+        request_slot: &Slot,
+        strategy: &AttestationDataStrategy,
+    ) -> Result<(AttestationData, usize), String> {
+        match strategy {
+            AttestationDataStrategy::Fallback => self.data_by_index(request_slot, None).await,
+            AttestationDataStrategy::ByIndex(index) => {
+                self.data_by_index(request_slot, Some(*index)).await
+            }
+            AttestationDataStrategy::Consensus((threshold, checkpoint_and_index)) => {
+                let consensus_aggregator =
+                    ConsensusAggregator::new(*threshold, *checkpoint_and_index);
+                self.data_with_aggregation(request_slot, consensus_aggregator)
+                    .await
+            }
+            AttestationDataStrategy::IgnoreEpoch(epoch) => Err(format!(
+                "Disabled attestation production for epoch {:?}",
+                epoch
+            )),
+            AttestationDataStrategy::HighestScore => {
+                let aggregator =
+                    ScoreAggregator::new(*request_slot, self.beacon_nodes.num_total().await);
+                self.data_with_aggregation(request_slot, aggregator).await
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{sync::Arc, time::Duration};
+
+    use beacon_node_fallback::{BeaconNodeFallback, CandidateBeaconNode, Config as FallbackConfig};
+    use eth2::{SensitiveUrl, Timeouts};
+    use slot_clock::{SlotClock, TestingSlotClock};
+    use bls::FixedBytesExtended;
+    use types::{
+        AttestationData, Checkpoint, Epoch, EthSpec, Hash256, MainnetEthSpec, MinimalEthSpec, Slot,
+    };
+
+    use crate::attestation_data_service::{AttestationDataService, AttestationDataStrategy};
+
+    fn create_attestation_data(
+        slot: Slot,
+        source_epoch: Epoch,
+        target_epoch: Epoch,
+    ) -> AttestationData {
+        AttestationData {
+            slot,
+            index: 0,
+            beacon_block_root: Hash256::ZERO,
+            source: Checkpoint {
+                epoch: source_epoch,
+                root: Hash256::ZERO,
+            },
+            target: Checkpoint {
+                epoch: target_epoch,
+                root: Hash256::from_low_u64_be(target_epoch.as_u64()),
+            },
+        }
+    }
+
+    // Helper to create a beacon node with mocked attestation endpoint
+    async fn create_mocked_beacon_node(
+        index: usize,
+        slot: Slot,
+        attestation_data: AttestationData,
+    ) -> (mockito::ServerGuard, CandidateBeaconNode) {
+        use eth2::types::GenericResponse;
+        use mockito::{Matcher, Server};
+        use regex::Regex;
+
+        let mut server = Server::new_async().await;
+        let data = GenericResponse::from(attestation_data);
+
+        let path_pattern = Regex::new(&format!(
+            r"^/eth/v1/validator/attestation_data\?slot={}&committee_index=0$",
+            slot.as_u64()
+        ))
+        .unwrap();
+
+        server
+            .mock("GET", Matcher::Regex(path_pattern.to_string()))
+            .with_status(200)
+            .with_body(serde_json::to_string(&data).unwrap())
+            .create();
+
+        let url = SensitiveUrl::parse(&server.url()).unwrap();
+        let client = eth2::BeaconNodeHttpClient::new_with_index(
+            url,
+            Timeouts::set_all(Duration::from_secs(1)),
+            index,
+        );
+        let candidate = CandidateBeaconNode::new(client);
+
+        (server, candidate)
+    }
+
+    async fn create_offline_beacon_node(
+        index: usize,
+    ) -> (mockito::ServerGuard, CandidateBeaconNode) {
+        use mockito::{Matcher, Server};
+        use regex::Regex;
+
+        let mut server = Server::new_async().await;
+        let path_pattern = Regex::new(r"^/eth/v1/validator/attestation_data").unwrap();
+
+        server
+            .mock("GET", Matcher::Regex(path_pattern.to_string()))
+            .with_status(500)
+            .create();
+
+        let url = SensitiveUrl::parse(&server.url()).unwrap();
+        let client = eth2::BeaconNodeHttpClient::new_with_index(
+            url,
+            Timeouts::set_all(Duration::from_secs(1)),
+            index,
+        );
+        let candidate = CandidateBeaconNode::new(client);
+
+        (server, candidate)
+    }
+
+    #[tokio::test]
+    async fn test_download_attestation_data() {
+        let spec = Arc::new(MinimalEthSpec::default_spec());
+        let slot = Slot::new(10);
+        let attestation_data = create_attestation_data(slot, Epoch::new(0), Epoch::new(1));
+
+        let (_server, beacon_node) =
+            create_mocked_beacon_node(0, slot, attestation_data.clone()).await;
+
+        let mut fallback =
+            BeaconNodeFallback::new(vec![beacon_node], FallbackConfig::default(), vec![], spec);
+
+        fallback.set_slot_clock(TestingSlotClock::new(
+            Slot::new(1),
+            Duration::from_secs(0),
+            Duration::from_secs(12),
+        ));
+
+        let service = AttestationDataService::<TestingSlotClock>::new(Arc::new(fallback));
+        let result = service
+            .download_data(&slot, &AttestationDataStrategy::Fallback)
+            .await;
+
+        // Verify download is successful
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), (attestation_data.clone(), 0));
+    }
+
+    #[tokio::test]
+    async fn test_download_attestation_data_all_nodes_offline() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let slot = Slot::new(10);
+
+        // Create two offline nodes
+        let (_server1, beacon_node_1) = create_offline_beacon_node(0).await;
+        let (_server2, beacon_node_2) = create_offline_beacon_node(1).await;
+
+        let mut fallback = BeaconNodeFallback::new(
+            vec![beacon_node_1, beacon_node_2],
+            FallbackConfig::default(),
+            vec![],
+            spec,
+        );
+
+        fallback.set_slot_clock(TestingSlotClock::new(
+            Slot::new(1),
+            Duration::from_secs(0),
+            Duration::from_secs(12),
+        ));
+
+        let service = AttestationDataService::<TestingSlotClock>::new(Arc::new(fallback));
+        let result = service
+            .download_data(&slot, &AttestationDataStrategy::Fallback)
+            .await;
+
+        // Verify all nodes offline
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .contains("Failed to produce attestation data")
+        );
+    }
+
+    #[tokio::test]
+    async fn test_download_attestation_data_node_fallback() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let slot = Slot::new(10);
+        let attestation_data = create_attestation_data(slot, Epoch::new(0), Epoch::new(1));
+
+        // Create one offline node and one working node
+        let (_server1, beacon_node_1) = create_offline_beacon_node(0).await;
+        let (_server2, beacon_node_2) =
+            create_mocked_beacon_node(1, slot, attestation_data.clone()).await;
+        let (_server2, beacon_node_3) =
+            create_mocked_beacon_node(2, slot, attestation_data.clone()).await;
+
+        let mut fallback = BeaconNodeFallback::new(
+            vec![beacon_node_1, beacon_node_2, beacon_node_3],
+            FallbackConfig::default(),
+            vec![],
+            spec,
+        );
+
+        fallback.set_slot_clock(TestingSlotClock::new(
+            Slot::new(1),
+            Duration::from_secs(0),
+            Duration::from_secs(12),
+        ));
+
+        let service = AttestationDataService::<TestingSlotClock>::new(Arc::new(fallback));
+        let result = service
+            .download_data(&slot, &AttestationDataStrategy::Fallback)
+            .await;
+
+        // Verify download is successful and we fell back to the next node
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), (attestation_data.clone(), 1));
+    }
+}
diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index 58b1acfcdf0..e9a68513f6e 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -1,5 +1,8 @@
 use crate::duties_service::{DutiesService, DutyAndProof};
-use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
+use tokio::sync::Mutex;
+
+use crate::attestation_data_service::{AttestationDataService, AttestationDataStrategy};
+use beacon_node_fallback::{ApiTopic, BeaconNodeFallback, beacon_head_monitor::HeadEvent};
 use futures::future::join_all;
 use logging::crit;
 use slot_clock::SlotClock;
@@ -7,10 +10,13 @@ use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::Arc;
 use task_executor::TaskExecutor;
+use tokio::sync::mpsc;
 use tokio::time::{Duration, Instant, sleep, sleep_until};
-use tracing::{Instrument, debug, error, info, info_span, instrument, trace, warn};
+use tracing::{Instrument, debug, error, info, info_span, instrument, warn};
 use tree_hash::TreeHash;
-use types::{Attestation, AttestationData, ChainSpec, CommitteeIndex, EthSpec, Slot};
+use types::{
+    Attestation, AttestationData, ChainSpec, Checkpoint, CommitteeIndex, Epoch, EthSpec, Slot,
+};
 use validator_store::{Error as ValidatorStoreError, ValidatorStore};
 
 /// Builds an `AttestationService`.
@@ -22,6 +28,10 @@ pub struct AttestationServiceBuilder<S: ValidatorStore, T: SlotClock + 'static>
     beacon_nodes: Option<Arc<BeaconNodeFallback<T>>>,
     executor: Option<TaskExecutor>,
     chain_spec: Option<Arc<ChainSpec>>,
+    head_monitor_rx: Option<Arc<Mutex<mpsc::Receiver<HeadEvent>>>>,
+    attestation_data_service: Option<Arc<AttestationDataService<T>>>,
+    latest_target_checkpoint: Arc<Mutex<Option<(Epoch, Checkpoint, usize)>>>,
+    consensus_threshold: Option<usize>,
     disable: bool,
 }
 
@@ -34,6 +44,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
             beacon_nodes: None,
             executor: None,
             chain_spec: None,
+            head_monitor_rx: None,
+            attestation_data_service: None,
+            latest_target_checkpoint: Arc::new(Mutex::new(None)),
+            consensus_threshold: None,
             disable: false,
         }
     }
@@ -54,7 +68,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
     }
 
     pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
-        self.beacon_nodes = Some(beacon_nodes);
+        self.beacon_nodes = Some(beacon_nodes.clone());
+        self.attestation_data_service = Some(Arc::new(AttestationDataService::new(beacon_nodes)));
+
         self
     }
 
@@ -73,6 +89,18 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
         self
     }
 
+    pub fn consensus_threshold(mut self, threshold: usize) -> Self {
+        self.consensus_threshold = Some(threshold);
+        self
+    }
+
+    pub fn head_monitor_rx(
+        mut self,
+        head_monitor_rx: Option<Arc<Mutex<mpsc::Receiver<HeadEvent>>>>,
+    ) -> Self {
+        self.head_monitor_rx = head_monitor_rx;
+        self
+    }
     pub fn build(self) -> Result<AttestationService<S, T>, String> {
         Ok(AttestationService {
             inner: Arc::new(Inner {
@@ -94,7 +122,14 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
                 chain_spec: self
                     .chain_spec
                     .ok_or("Cannot build AttestationService without chain_spec")?,
+                attestation_data_service: self
+                    .attestation_data_service
+                    .ok_or("Cannot build AttestationService without attestation_data_service")?,
+                head_monitor_rx: self.head_monitor_rx,
+                latest_target_checkpoint: self.latest_target_checkpoint,
+                consensus_threshold: self.consensus_threshold,
                 disable: self.disable,
+                latest_attested_slot: Mutex::new(Slot::default()),
             }),
         })
     }
@@ -108,10 +143,16 @@ pub struct Inner<S, T> {
     beacon_nodes: Arc<BeaconNodeFallback<T>>,
     executor: TaskExecutor,
     chain_spec: Arc<ChainSpec>,
+    head_monitor_rx: Option<Arc<Mutex<mpsc::Receiver<HeadEvent>>>>,
+    attestation_data_service: Arc<AttestationDataService<T>>,
+    latest_target_checkpoint: Arc<Mutex<Option<(Epoch, Checkpoint, usize)>>>,
+    consensus_threshold: Option<usize>,
     disable: bool,
+    latest_attested_slot: Mutex<Slot>,
 }
 
-/// Attempts to produce attestations for all known validators 1/3rd of the way through each slot.
+/// Attempts to produce attestations for all known validators 1/3rd of the way through each slot
+/// or when a head event is received from the BNs.
 ///
 /// If any validators are on the same committee, a single attestation will be downloaded and
 /// returned to the beacon node. This attestation will have a signature from each of the
@@ -144,6 +185,13 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             return Ok(());
         }
 
+        let mut attestation_data_strategy =
+            if let Some(consensus_threshold) = self.consensus_threshold {
+                AttestationDataStrategy::Consensus((consensus_threshold, None))
+            } else {
+                AttestationDataStrategy::Fallback
+            };
+
         let slot_duration = Duration::from_secs(spec.seconds_per_slot);
         let duration_to_next_slot = self
             .slot_clock
@@ -159,19 +207,45 @@ 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;
+                let Some(duration) = self.slot_clock.duration_to_next_slot() else {
+                    error!("Failed to read slot clock");
+                    sleep(slot_duration).await;
+                    continue;
+                };
 
-                    if let Err(e) = self.spawn_attestation_tasks(slot_duration) {
-                        crit!(error = e, "Failed to spawn attestation tasks")
-                    } else {
-                        trace!("Spawned attestation tasks");
+                let slot_trigger_delay = slot_duration / 3;
+                let beacon_node_index = if self.head_monitor_rx.is_some() {
+                    tokio::select! {
+                        _ = sleep(duration + slot_trigger_delay) => None,
+                        event = self.poll_for_head_events() => event.map(|event| event.beacon_node_index),
                     }
                 } else {
-                    error!("Failed to read slot clock");
-                    // If we can't read the slot clock, just wait another slot.
-                    sleep(slot_duration).await;
+                    sleep(duration + slot_trigger_delay).await;
+                    None
+                };
+
+                if let Some(beacon_node_index) = beacon_node_index {
+                    attestation_data_strategy = AttestationDataStrategy::ByIndex(beacon_node_index);
+                };
+
+                let Some(current_slot) = self.slot_clock.now() else {
+                    error!("Failed to read slot clock after trigger");
                     continue;
+                };
+
+                let mut last_slot = self.latest_attested_slot.lock().await;
+
+                if current_slot <= *last_slot {
+                    debug!(?current_slot, "Attestation already initiated for the slot");
+                    continue;
+                }
+
+                match self.spawn_attestation_tasks(slot_duration, attestation_data_strategy.clone())
+                {
+                    Ok(_) => *last_slot = current_slot,
+                    Err(e) => {
+                        crit!(error = e, "Failed to spawn attestation tasks")
+                    }
                 }
             }
         };
@@ -180,10 +254,28 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
+    async fn poll_for_head_events(&self) -> Option<HeadEvent> {
+        let Some(receiver) = &self.head_monitor_rx else {
+            return None;
+        };
+        let mut receiver = receiver.lock().await;
+        match receiver.recv().await {
+            Some(head_event) => Some(head_event),
+            None => {
+                warn!("Head monitor channel closed unexpectedly");
+                None
+            }
+        }
+    }
+
     /// Spawn only one new task for attestation post-Electra
     /// For each required aggregates, spawn a new task that downloads, signs and uploads the
     /// aggregates to the beacon node.
-    fn spawn_attestation_tasks(&self, slot_duration: Duration) -> Result<(), String> {
+    fn spawn_attestation_tasks(
+        &self,
+        slot_duration: Duration,
+        attestation_data_strategy: AttestationDataStrategy,
+    ) -> Result<(), String> {
         let slot = self.slot_clock.now().ok_or("Failed to read slot clock")?;
         let duration_to_next_slot = self
             .slot_clock
@@ -201,26 +293,62 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
 
         let attestation_service = self.clone();
 
+        // If we're using the Consensus strategy we need to handle the following situations:
+        // - The first slot in the epoch
+        // - A slot thats within an attestable epoch
+        // - A slot thats not within an attestable epoch
+        // - A slot that is not the first in an epoch and no target checkpoint to compare
+        let attestation_data_strategy =
+            if let AttestationDataStrategy::Consensus((threshold, _)) = attestation_data_strategy {
+                if slot.is_start_slot_in_epoch(S::E::slots_per_epoch()) {
+                    // if the current slot is the first slot in the epoch use the default consensus strategy
+                    attestation_data_strategy
+                } else if let Some((latest_attestable_epoch, target_checkpoint, preferred_index)) =
+                    *attestation_service.latest_target_checkpoint.blocking_lock()
+                {
+                    if slot.epoch(S::E::slots_per_epoch()) == latest_attestable_epoch {
+                        // If the current slot is within the latest attestable epoch, we can attest
+                        // using the `preferred_index` or nodes that have a matching `target_checkpoint`
+                        AttestationDataStrategy::Consensus((
+                            threshold,
+                            Some((target_checkpoint, preferred_index)),
+                        ))
+                    } else {
+                        // If the current slot is not within the latest attestable epoch, we cannot attest
+                        AttestationDataStrategy::IgnoreEpoch(slot.epoch(S::E::slots_per_epoch()))
+                    }
+                } else {
+                    // If the current slot is not the first slot in an epoch and there is no target checkpoint to compare,
+                    // run the default consensus strategy. This can happen if the attestation service was initially
+                    // launched in the middle of an epoch.
+                    attestation_data_strategy
+                }
+            } else {
+                attestation_data_strategy
+            };
+
         let attestation_data_handle = self
             .inner
             .executor
             .spawn_handle(
                 async move {
-                    let attestation_data = attestation_service
-                        .beacon_nodes
-                        .first_success(|beacon_node| async move {
-                            let _timer = validator_metrics::start_timer_vec(
-                                &validator_metrics::ATTESTATION_SERVICE_TIMES,
-                                &[validator_metrics::ATTESTATIONS_HTTP_GET],
-                            );
-                            beacon_node
-                                .get_validator_attestation_data(slot, 0)
-                                .await
-                                .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
-                                .map(|result| result.data)
-                        })
-                        .await
-                        .map_err(|e| e.to_string())?;
+                    let (attestation_data, index) = attestation_service
+                        .attestation_data_service
+                        .download_data(&slot, &attestation_data_strategy)
+                        .await?;
+
+                    // If we're using the consensus strategy and we fetched attestation data for the first slot of an epoch
+                    // update `latest_target_checkpoint` so we can query subsequent epochs
+                    if let AttestationDataStrategy::Consensus(_) = attestation_data_strategy
+                        && slot.is_start_slot_in_epoch(S::E::slots_per_epoch())
+                    {
+                        let mut guard = attestation_service.latest_target_checkpoint.lock().await;
+                        *guard = Some((
+                            slot.epoch(S::E::slots_per_epoch()),
+                            attestation_data.target,
+                            index,
+                        ));
+                    }
 
                     attestation_service
                         .sign_and_publish_attestations(
diff --git a/validator_client/validator_services/src/block_service.rs b/validator_client/validator_services/src/block_service.rs
index 625f8db7cb9..0a22072d3d1 100644
--- a/validator_client/validator_services/src/block_service.rs
+++ b/validator_client/validator_services/src/block_service.rs
@@ -182,7 +182,7 @@ impl<T: SlotClock> ProposerFallback<T> {
 
         match (beacon_nodes_result, &self.proposer_nodes) {
             // The non-proposer node call succeed, return the result.
-            (Ok(success), _) => Ok(success),
+            (Ok(data), _) => Ok(data),
             // The non-proposer node call failed, but we don't have any proposer nodes. Return an error.
             (Err(e), None) => Err(e),
             // The non-proposer node call failed, try the same call on the proposer nodes.
diff --git a/validator_client/validator_services/src/lib.rs b/validator_client/validator_services/src/lib.rs
index 3b8bd9ae14b..e0e05711786 100644
--- a/validator_client/validator_services/src/lib.rs
+++ b/validator_client/validator_services/src/lib.rs
@@ -1,3 +1,4 @@
+pub mod attestation_data_service;
 pub mod attestation_service;
 pub mod block_service;
 pub mod duties_service;