From d0ac24da5f6b49b35e8d0507fa339bf571ed8150 Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Wed, 8 Oct 2025 12:36:57 +0800
Subject: [PATCH 1/7] attestation committee index

---
 .../src/attestation_service.rs                | 148 ++++++++++--------
 1 file changed, 85 insertions(+), 63 deletions(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index da6e8f35886..458a410561d 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -180,8 +180,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
-    /// For each each required attestation, spawn a new task that downloads, signs and uploads the
-    /// attestation to the beacon node.
+    /// 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> {
         let slot = self.slot_clock.now().ok_or("Failed to read slot clock")?;
         let duration_to_next_slot = self
@@ -189,6 +190,15 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             .duration_to_next_slot()
             .ok_or("Unable to determine duration to next slot")?;
 
+        // Create and publish an `Attestation` for all validators only once
+        // as the committee_index is the same (index 0) post-Electra
+        let attestation_duties = self.duties_service.attesters(slot).into_iter().collect();
+
+        self.inner.executor.spawn_ignoring_error(
+            self.clone().publish_attestations(slot, attestation_duties),
+            "attestation publish",
+        );
+
         // If a validator needs to publish an aggregate attestation, they must do so at 2/3
         // through the slot. This delay triggers at this time
         let aggregate_production_instant = Instant::now()
@@ -196,7 +206,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                 .checked_sub(slot_duration / 3)
                 .unwrap_or_else(|| Duration::from_secs(0));
 
-        let duties_by_committee_index: HashMap<CommitteeIndex, Vec<DutyAndProof>> = self
+        let aggregate_duties_by_committee_index: HashMap<CommitteeIndex, Vec<DutyAndProof>> = self
             .duties_service
             .attesters(slot)
             .into_iter()
@@ -208,23 +218,21 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             });
 
         // For each committee index for this slot:
-        //
-        // - Create and publish an `Attestation` for all required validators.
-        // - Create and publish `SignedAggregateAndProof` for all aggregating validators.
-        duties_by_committee_index
-            .into_iter()
-            .for_each(|(committee_index, validator_duties)| {
+        // Create and publish `SignedAggregateAndProof` for all aggregating validators.
+        aggregate_duties_by_committee_index.into_iter().for_each(
+            |(committee_index, validator_duties)| {
                 // Spawn a separate task for each attestation.
                 self.inner.executor.spawn_ignoring_error(
-                    self.clone().publish_attestations_and_aggregates(
+                    self.clone().publish_aggregates(
                         slot,
                         committee_index,
                         validator_duties,
                         aggregate_production_instant,
                     ),
-                    "attestation publish",
+                    "aggregate publish",
                 );
-            });
+            },
+        );
 
         // Schedule pruning of the slashing protection database once all unaggregated
         // attestations have (hopefully) been signed, i.e. at the same time as aggregate
@@ -234,84 +242,101 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
-    /// Performs the first step of the attesting process: downloading `Attestation` objects,
-    /// signing them and returning them to the validator.
-    ///
-    /// https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/validator.md#attesting
-    ///
-    /// ## Detail
-    ///
-    /// The given `validator_duties` should already be filtered to only contain those that match
-    /// `slot` and `committee_index`. Critical errors will be logged if this is not the case.
-    async fn publish_attestations_and_aggregates(
+    /// Produce and publish unaggregated attestations for all validators.
+    /// This is only called once per slot post-Electra as the committee index will be 0.
+    async fn publish_attestations(
         self,
         slot: Slot,
-        committee_index: CommitteeIndex,
         validator_duties: Vec<DutyAndProof>,
-        aggregate_production_instant: Instant,
     ) -> Result<(), ()> {
         let attestations_timer = validator_metrics::start_timer_vec(
             &validator_metrics::ATTESTATION_SERVICE_TIMES,
             &[validator_metrics::ATTESTATIONS],
         );
 
-        // There's not need to produce `Attestation` or `SignedAggregateAndProof` if we do not have
-        // any validators for the given `slot` and `committee_index`.
+        // There's not need to produce `Attestation` if we do not have
+        // any validators for the given `slot`.
         if validator_duties.is_empty() {
             return Ok(());
         }
 
-        // Step 1.
-        //
-        // Download, sign and publish an `Attestation` for each validator.
-        let attestation_opt = self
-            .produce_and_publish_attestations(slot, committee_index, &validator_duties)
+        // Download, sign and publish an `Attestation` for all validators at once
+        self.produce_and_publish_attestations(slot, &validator_duties)
             .await
             .map_err(move |e| {
                 crit!(
                     error = format!("{:?}", e),
-                    committee_index,
                     slot = slot.as_u64(),
                     "Error during attestation routine"
                 )
             })?;
 
         drop(attestations_timer);
+        Ok(())
+    }
 
-        // Step 2.
-        //
-        // If an attestation was produced, make an aggregate.
-        if let Some(attestation_data) = attestation_opt {
-            // First, wait until the `aggregation_production_instant` (2/3rds
-            // of the way though the slot). As verified in the
-            // `delay_triggers_when_in_the_past` test, this code will still run
-            // even if the instant has already elapsed.
-            sleep_until(aggregate_production_instant).await;
-
-            // Start the metrics timer *after* we've done the delay.
-            let _aggregates_timer = validator_metrics::start_timer_vec(
-                &validator_metrics::ATTESTATION_SERVICE_TIMES,
-                &[validator_metrics::AGGREGATES],
-            );
-
-            // Then download, sign and publish a `SignedAggregateAndProof` for each
-            // validator that is elected to aggregate for this `slot` and
-            // `committee_index`.
-            self.produce_and_publish_aggregates(
-                &attestation_data,
-                committee_index,
-                &validator_duties,
-            )
+    /// Produce and publish aggregated attestations for validators
+    async fn publish_aggregates(
+        self,
+        slot: Slot,
+        committee_index: CommitteeIndex,
+        validator_duties: Vec<DutyAndProof>,
+        aggregate_production_instant: Instant,
+    ) -> Result<(), ()> {
+        // There's not need to produce `SignedAggregateAndProof` if we do not have
+        // any validators for the given `slot` and `committee_index`.
+        if validator_duties.is_empty() {
+            return Ok(());
+        }
+
+        // Wait until the `aggregation_production_instant` (2/3rds
+        // of the way though the slot). As verified in the
+        // `delay_triggers_when_in_the_past` test, this code will still run
+        // even if the instant has already elapsed.
+        sleep_until(aggregate_production_instant).await;
+
+        // Start the metrics timer *after* we've done the delay.
+        let _aggregates_timer = validator_metrics::start_timer_vec(
+            &validator_metrics::ATTESTATION_SERVICE_TIMES,
+            &[validator_metrics::AGGREGATES],
+        );
+
+        let attestation_data = self
+            .beacon_nodes
+            .first_success(|beacon_node| async move {
+                beacon_node
+                    .get_validator_attestation_data(slot, 0)
+                    .await
+                    .map_err(|e| {
+                        format!(
+                            "Failed to produce attestation data for aggregation: {:?}",
+                            e
+                        )
+                    })
+                    .map(|result| result.data)
+            })
+            .await
+            .map_err(|e| {
+                error!(
+                    error = %e,
+                    slot = slot.as_u64(),
+                    "Failed to produce attestation data for aggregation"
+                );
+            })?;
+
+        // Download, sign and publish a `SignedAggregateAndProof` for each
+        // validator that is elected to aggregate for this `slot` and
+        // `committee_index`.
+        self.produce_and_publish_aggregates(&attestation_data, committee_index, &validator_duties)
             .await
             .map_err(move |e| {
                 crit!(
                     error = format!("{:?}", e),
                     committee_index,
                     slot = slot.as_u64(),
-                    "Error during attestation routine"
+                    "Error during aggregate attestation routine"
                 )
             })?;
-        }
 
         Ok(())
     }
@@ -331,7 +356,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     async fn produce_and_publish_attestations(
         &self,
         slot: Slot,
-        committee_index: CommitteeIndex,
         validator_duties: &[DutyAndProof],
     ) -> Result<Option<AttestationData>, String> {
         if validator_duties.is_empty() {
@@ -352,7 +376,8 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     &[validator_metrics::ATTESTATIONS_HTTP_GET],
                 );
                 beacon_node
-                    .get_validator_attestation_data(slot, committee_index)
+                    // Post-Electra, the request to beacon node for AttestationData must have a committee_index of 0
+                    .get_validator_attestation_data(slot, 0)
                     .await
                     .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
                     .map(|result| result.data)
@@ -372,7 +397,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     validator = ?duty.pubkey,
                     duty_slot = %duty.slot,
                     attestation_slot = %attestation_data.slot,
-                    duty_index = duty.committee_index,
                     attestation_index = attestation_data.index,
                     "Inconsistent validator duties during signing"
                 );
@@ -418,7 +442,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                         info = "a validator may have recently been removed from this VC",
                         pubkey = ?pubkey,
                         validator = ?duty.pubkey,
-                        committee_index = committee_index,
                         slot = slot.as_u64(),
                         "Missing pubkey for attestation"
                     );
@@ -428,7 +451,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     crit!(
                         error = ?e,
                         validator = ?duty.pubkey,
-                        committee_index,
                         slot = slot.as_u64(),
                         "Failed to sign attestation"
                     );

From 7b6def0bc94311b74f412f75651a5a173b3d72ab Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Thu, 16 Oct 2025 10:00:25 +0800
Subject: [PATCH 2/7] refactor to reduce calling the bn

---
 .../src/attestation_service.rs                | 150 +++++++++---------
 1 file changed, 71 insertions(+), 79 deletions(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index 458a410561d..d75fa20baba 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -192,10 +192,41 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
 
         // Create and publish an `Attestation` for all validators only once
         // as the committee_index is the same (index 0) post-Electra
-        let attestation_duties = self.duties_service.attesters(slot).into_iter().collect();
+        let attestation_duties: Vec<_> = self.duties_service.attesters(slot).into_iter().collect();
 
+        let attestation_data_cache = Arc::new(tokio::sync::RwLock::new(None));
+        let attestation_data_cache_clone = attestation_data_cache.clone();
+
+        let attestation_service = self.clone();
         self.inner.executor.spawn_ignoring_error(
-            self.clone().publish_attestations(slot, attestation_duties),
+            async move {
+                let attestation_data = attestation_service
+                    .beacon_nodes
+                    .first_success(|beacon_node| async move {
+                        beacon_node
+                            .get_validator_attestation_data(slot, 0)
+                            .await
+                            .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
+                            .map(|result| result.data)
+                    })
+                    .await;
+
+                match attestation_data {
+                    Ok(attestation_data) => {
+                        *attestation_data_cache_clone.write().await =
+                            Some(attestation_data.clone());
+
+                        attestation_service
+                            .publish_attestations(slot, &attestation_duties, attestation_data)
+                            .await
+                            .ok();
+                    }
+                    Err(e) => {
+                        error!(error = %e, slot = slot.as_u64(), "Failed to get attestation data from beacon nodes");
+                    }
+                }
+                Ok(())
+            },
             "attestation publish",
         );
 
@@ -221,13 +252,15 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         // Create and publish `SignedAggregateAndProof` for all aggregating validators.
         aggregate_duties_by_committee_index.into_iter().for_each(
             |(committee_index, validator_duties)| {
+                let attestation_data = attestation_data_cache.clone();
                 // Spawn a separate task for each attestation.
                 self.inner.executor.spawn_ignoring_error(
-                    self.clone().publish_aggregates(
+                    self.clone().handle_aggregates(
                         slot,
                         committee_index,
                         validator_duties,
                         aggregate_production_instant,
+                        attestation_data,
                     ),
                     "aggregate publish",
                 );
@@ -242,46 +275,14 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
-    /// Produce and publish unaggregated attestations for all validators.
-    /// This is only called once per slot post-Electra as the committee index will be 0.
-    async fn publish_attestations(
-        self,
-        slot: Slot,
-        validator_duties: Vec<DutyAndProof>,
-    ) -> Result<(), ()> {
-        let attestations_timer = validator_metrics::start_timer_vec(
-            &validator_metrics::ATTESTATION_SERVICE_TIMES,
-            &[validator_metrics::ATTESTATIONS],
-        );
-
-        // There's not need to produce `Attestation` if we do not have
-        // any validators for the given `slot`.
-        if validator_duties.is_empty() {
-            return Ok(());
-        }
-
-        // Download, sign and publish an `Attestation` for all validators at once
-        self.produce_and_publish_attestations(slot, &validator_duties)
-            .await
-            .map_err(move |e| {
-                crit!(
-                    error = format!("{:?}", e),
-                    slot = slot.as_u64(),
-                    "Error during attestation routine"
-                )
-            })?;
-
-        drop(attestations_timer);
-        Ok(())
-    }
-
     /// Produce and publish aggregated attestations for validators
-    async fn publish_aggregates(
+    async fn handle_aggregates(
         self,
         slot: Slot,
         committee_index: CommitteeIndex,
         validator_duties: Vec<DutyAndProof>,
         aggregate_production_instant: Instant,
+        attestation_data: Arc<tokio::sync::RwLock<Option<AttestationData>>>,
     ) -> Result<(), ()> {
         // There's not need to produce `SignedAggregateAndProof` if we do not have
         // any validators for the given `slot` and `committee_index`.
@@ -301,28 +302,29 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             &[validator_metrics::AGGREGATES],
         );
 
-        let attestation_data = self
-            .beacon_nodes
-            .first_success(|beacon_node| async move {
-                beacon_node
-                    .get_validator_attestation_data(slot, 0)
+        let attestation_data = {
+            let attestation_data_opt = attestation_data.read().await;
+            if let Some(attestation_data) = attestation_data_opt.as_ref() {
+                attestation_data.clone()
+            } else {
+                self.beacon_nodes
+                    .first_success(|beacon_node| async move {
+                        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| {
-                        format!(
-                            "Failed to produce attestation data for aggregation: {:?}",
-                            e
-                        )
-                    })
-                    .map(|result| result.data)
-            })
-            .await
-            .map_err(|e| {
-                error!(
-                    error = %e,
-                    slot = slot.as_u64(),
-                    "Failed to produce attestation data for aggregation"
-                );
-            })?;
+                        error!(
+                            error = %e,
+                            slot = slot.as_u64(),
+                            "Failed to get attestation data from beacon nodes"
+                        );
+                    })?
+            }
+        };
 
         // Download, sign and publish a `SignedAggregateAndProof` for each
         // validator that is elected to aggregate for this `slot` and
@@ -353,13 +355,19 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     ///
     /// Only one `Attestation` is downloaded from the BN. It is then cloned and signed by each
     /// validator and the list of individually-signed `Attestation` objects is returned to the BN.
-    async fn produce_and_publish_attestations(
+    async fn publish_attestations(
         &self,
         slot: Slot,
         validator_duties: &[DutyAndProof],
-    ) -> Result<Option<AttestationData>, String> {
+        attestation_data: AttestationData,
+    ) -> Result<(), String> {
+        let attestations_timer = validator_metrics::start_timer_vec(
+            &validator_metrics::ATTESTATION_SERVICE_TIMES,
+            &[validator_metrics::ATTESTATIONS],
+        );
+
         if validator_duties.is_empty() {
-            return Ok(None);
+            return Ok(());
         }
 
         let current_epoch = self
@@ -368,23 +376,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             .ok_or("Unable to determine current slot from clock")?
             .epoch(S::E::slots_per_epoch());
 
-        let attestation_data = self
-            .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
-                    // Post-Electra, the request to beacon node for AttestationData must have a committee_index of 0
-                    .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())?;
-
         // Create futures to produce signed `Attestation` objects.
         let attestation_data_ref = &attestation_data;
         let signing_futures = validator_duties.iter().map(|duty_and_proof| async move {
@@ -468,7 +459,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
 
         if attestations.is_empty() {
             warn!("No attestations were published");
-            return Ok(None);
+            return Ok(());
         }
         let fork_name = self
             .chain_spec
@@ -529,7 +520,8 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             ),
         }
 
-        Ok(Some(attestation_data))
+        drop(attestations_timer);
+        Ok(())
     }
 
     /// Performs the second step of the attesting process: downloading an aggregated `Attestation`,

From a11bca01dd3259bff212b47cc7b251116b901c04 Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Thu, 16 Oct 2025 11:37:54 +0800
Subject: [PATCH 3/7] revise

---
 validator_client/validator_services/src/attestation_service.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index d75fa20baba..ba11760558f 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -275,7 +275,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
-    /// Produce and publish aggregated attestations for validators
+    /// If an attestation was produced, make an aggregate
     async fn handle_aggregates(
         self,
         slot: Slot,

From 0967696831ce1c666baa6e85137c69483396f1d9 Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Fri, 17 Oct 2025 12:29:34 +0800
Subject: [PATCH 4/7] revise

---
 .../src/attestation_service.rs                | 123 +++++++++---------
 1 file changed, 59 insertions(+), 64 deletions(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index ba11760558f..47fd59287b7 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -193,12 +193,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         // Create and publish an `Attestation` for all validators only once
         // as the committee_index is the same (index 0) post-Electra
         let attestation_duties: Vec<_> = self.duties_service.attesters(slot).into_iter().collect();
-
-        let attestation_data_cache = Arc::new(tokio::sync::RwLock::new(None));
-        let attestation_data_cache_clone = attestation_data_cache.clone();
-
         let attestation_service = self.clone();
-        self.inner.executor.spawn_ignoring_error(
+
+        let attestation_handle = self.inner.executor.spawn_handle(
             async move {
                 let attestation_data = attestation_service
                     .beacon_nodes
@@ -209,23 +206,21 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
                             .map(|result| result.data)
                     })
-                    .await;
-
-                match attestation_data {
-                    Ok(attestation_data) => {
-                        *attestation_data_cache_clone.write().await =
-                            Some(attestation_data.clone());
+                    .await
+                    .map_err(|e| e.to_string())?;
 
-                        attestation_service
-                            .publish_attestations(slot, &attestation_duties, attestation_data)
-                            .await
-                            .ok();
-                    }
-                    Err(e) => {
-                        error!(error = %e, slot = slot.as_u64(), "Failed to get attestation data from beacon nodes");
-                    }
-                }
-                Ok(())
+                attestation_service
+                    .publish_attestations(slot, &attestation_duties, attestation_data.clone())
+                    .await
+                    .map_err(|e| {
+                        crit!(
+                            error = format!("{:?}", e),
+                            slot = slot.as_u64(),
+                            "Error during attestation routine"
+                        );
+                        e
+                    })?;
+                Ok::<AttestationData, String>(attestation_data)
             },
             "attestation publish",
         );
@@ -248,23 +243,47 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                 map
             });
 
-        // For each committee index for this slot:
-        // Create and publish `SignedAggregateAndProof` for all aggregating validators.
-        aggregate_duties_by_committee_index.into_iter().for_each(
-            |(committee_index, validator_duties)| {
-                let attestation_data = attestation_data_cache.clone();
-                // Spawn a separate task for each attestation.
-                self.inner.executor.spawn_ignoring_error(
-                    self.clone().handle_aggregates(
-                        slot,
-                        committee_index,
-                        validator_duties,
-                        aggregate_production_instant,
-                        attestation_data,
-                    ),
-                    "aggregate publish",
-                );
+        // Spawn a task that awaits the attestation handle and then spawns aggregate tasks
+        let attestation_service_clone = self.clone();
+        let executor = self.inner.executor.clone();
+        self.inner.executor.spawn(
+            async move {
+                // Await the attestation handle to get the attestation data
+                let Some(handle) = attestation_handle else {
+                    error!(slot = slot.as_u64(), "Failed to spawn attestation task");
+                    return;
+                };
+
+                let Ok(Some(Ok(attestation_data))) = handle.await else {
+                    error!(slot = slot.as_u64(), "Failed to spawn attestation task");
+                    return;
+                };
+
+                // For each committee index for this slot:
+                // Create and publish `SignedAggregateAndProof` for all aggregating validators.
+                aggregate_duties_by_committee_index.into_iter().for_each(
+                    |(committee_index, validator_duties)| {
+                        let attestation_service = attestation_service_clone.clone();
+                        let attestation_data = attestation_data.clone();
+                        // Spawn a separate task for each committee's aggregates.
+                        executor.spawn_ignoring_error(
+                            async move {
+                                attestation_service
+                                    .handle_aggregates(
+                                        slot,
+                                        committee_index,
+                                        validator_duties,
+                                        aggregate_production_instant,
+                                        attestation_data,
+                                    )
+                                    .await
+                            },
+                            "aggregate publish",
+                        );
+                    },
+                )
             },
+            "attestation and aggregate publish",
         );
 
         // Schedule pruning of the slashing protection database once all unaggregated
@@ -282,7 +301,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         committee_index: CommitteeIndex,
         validator_duties: Vec<DutyAndProof>,
         aggregate_production_instant: Instant,
-        attestation_data: Arc<tokio::sync::RwLock<Option<AttestationData>>>,
+        attestation_data: AttestationData,
     ) -> Result<(), ()> {
         // There's not need to produce `SignedAggregateAndProof` if we do not have
         // any validators for the given `slot` and `committee_index`.
@@ -302,30 +321,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             &[validator_metrics::AGGREGATES],
         );
 
-        let attestation_data = {
-            let attestation_data_opt = attestation_data.read().await;
-            if let Some(attestation_data) = attestation_data_opt.as_ref() {
-                attestation_data.clone()
-            } else {
-                self.beacon_nodes
-                    .first_success(|beacon_node| async move {
-                        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| {
-                        error!(
-                            error = %e,
-                            slot = slot.as_u64(),
-                            "Failed to get attestation data from beacon nodes"
-                        );
-                    })?
-            }
-        };
-
         // Download, sign and publish a `SignedAggregateAndProof` for each
         // validator that is elected to aggregate for this `slot` and
         // `committee_index`.
@@ -361,7 +356,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         validator_duties: &[DutyAndProof],
         attestation_data: AttestationData,
     ) -> Result<(), String> {
-        let attestations_timer = validator_metrics::start_timer_vec(
+        let _attestations_timer = validator_metrics::start_timer_vec(
             &validator_metrics::ATTESTATION_SERVICE_TIMES,
             &[validator_metrics::ATTESTATIONS],
         );
@@ -388,6 +383,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     validator = ?duty.pubkey,
                     duty_slot = %duty.slot,
                     attestation_slot = %attestation_data.slot,
+                    duty_index = duty.committee_index,
                     attestation_index = attestation_data.index,
                     "Inconsistent validator duties during signing"
                 );
@@ -520,7 +516,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             ),
         }
 
-        drop(attestations_timer);
         Ok(())
     }
 

From 99901223bf860a75b60df9c389a757d9aa689cd0 Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Fri, 17 Oct 2025 13:51:05 +0800
Subject: [PATCH 5/7] revise

---
 .../src/attestation_service.rs                 | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index 47fd59287b7..1d41214ba39 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -191,15 +191,19 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             .ok_or("Unable to determine duration to next slot")?;
 
         // Create and publish an `Attestation` for all validators only once
-        // as the committee_index is the same (index 0) post-Electra
+        // as the committee_index is not included in AttestationData post-Electra
         let attestation_duties: Vec<_> = self.duties_service.attesters(slot).into_iter().collect();
-        let attestation_service = self.clone();
+        let attestation_service: AttestationService<S, T> = self.clone();
 
-        let attestation_handle = self.inner.executor.spawn_handle(
+        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
@@ -243,13 +247,13 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                 map
             });
 
-        // Spawn a task that awaits the attestation handle and then spawns aggregate tasks
+        // Spawn a task that awaits the attestation data handle and then spawns aggregate tasks
         let attestation_service_clone = self.clone();
         let executor = self.inner.executor.clone();
         self.inner.executor.spawn(
             async move {
-                // Await the attestation handle to get the attestation data
-                let Some(handle) = attestation_handle else {
+                // Log an error if the handle fails and return, skipping aggregates
+                let Some(handle) = attestation_data_handle else {
                     error!(slot = slot.as_u64(), "Failed to spawn attestation task");
                     return;
                 };
@@ -265,7 +269,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     |(committee_index, validator_duties)| {
                         let attestation_service = attestation_service_clone.clone();
                         let attestation_data = attestation_data.clone();
-                        // Spawn a separate task for each committee's aggregates.
                         executor.spawn_ignoring_error(
                             async move {
                                 attestation_service
@@ -294,7 +297,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         Ok(())
     }
 
-    /// If an attestation was produced, make an aggregate
     async fn handle_aggregates(
         self,
         slot: Slot,

From ed9ef193da427bd514229394c3156dffe8b2e29a Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Fri, 17 Oct 2025 16:43:01 +0800
Subject: [PATCH 6/7] simplify

---
 validator_client/validator_services/src/attestation_service.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index 1d41214ba39..ed069ceb4a6 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -193,7 +193,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         // Create and publish an `Attestation` for all validators only once
         // as the committee_index is not included in AttestationData post-Electra
         let attestation_duties: Vec<_> = self.duties_service.attesters(slot).into_iter().collect();
-        let attestation_service: AttestationService<S, T> = self.clone();
+        let attestation_service = self.clone();
 
         let attestation_data_handle = self.inner.executor.spawn_handle(
             async move {

From 5f9ff2dcb238148615c668d9757a7a11fe1684d4 Mon Sep 17 00:00:00 2001
From: Tan Chee Keong <tanck@sigmaprime.io>
Date: Mon, 20 Oct 2025 07:02:24 +0800
Subject: [PATCH 7/7] simplify

---
 .../src/attestation_service.rs                 | 18 +++++++-----------
 1 file changed, 7 insertions(+), 11 deletions(-)

diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs
index ed069ceb4a6..d3828606961 100644
--- a/validator_client/validator_services/src/attestation_service.rs
+++ b/validator_client/validator_services/src/attestation_service.rs
@@ -270,17 +270,13 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                         let attestation_service = attestation_service_clone.clone();
                         let attestation_data = attestation_data.clone();
                         executor.spawn_ignoring_error(
-                            async move {
-                                attestation_service
-                                    .handle_aggregates(
-                                        slot,
-                                        committee_index,
-                                        validator_duties,
-                                        aggregate_production_instant,
-                                        attestation_data,
-                                    )
-                                    .await
-                            },
+                            attestation_service.handle_aggregates(
+                                slot,
+                                committee_index,
+                                validator_duties,
+                                aggregate_production_instant,
+                                attestation_data,
+                            ),
                             "aggregate publish",
                         );
                     },