Broadcast VC requests in parallel (#6223)

Squashed commit of the following:

commit f0b4a0a
Author: Michael Sproul <[email protected]>
Date:   Tue Aug 6 11:58:56 2024 +1000

    Try some things

commit e2f1875
Author: Michael Sproul <[email protected]>
Date:   Tue Aug 6 11:50:35 2024 +1000

    Remove outdated comment

commit 763a6ae
Author: Michael Sproul <[email protected]>
Date:   Mon Aug 5 13:27:55 2024 +1000

    Broadcast VC requests in parallel

Author: michaelsproul

common/eth2/src/lib.rs

@@ -121,6 +121,7 @@ impl fmt::Display for Error {
 pub struct Timeouts {
     pub attestation: Duration,
     pub attester_duties: Duration,
+    pub attestation_subscriptions: Duration,
     pub liveness: Duration,
     pub proposal: Duration,
     pub proposer_duties: Duration,
@@ -137,6 +138,7 @@ impl Timeouts {
         Timeouts {
             attestation: timeout,
             attester_duties: timeout,
+            attestation_subscriptions: timeout,
             liveness: timeout,
             proposal: timeout,
             proposer_duties: timeout,
@@ -2515,7 +2517,12 @@ impl BeaconNodeHttpClient {
             .push("validator")
             .push("beacon_committee_subscriptions");
 
-        self.post(path, &subscriptions).await?;
+        self.post_with_timeout(
+            path,
+            &subscriptions,
+            self.timeouts.attestation_subscriptions,
+        )
+        .await?;
 
         Ok(())
     }

validator_client/src/beacon_node_fallback.rs

@@ -184,6 +184,11 @@ impl<E: EthSpec> CandidateBeaconNode<E> {
         spec: &ChainSpec,
         log: &Logger,
     ) -> Result<(), CandidateError> {
+        debug!(
+            log,
+            "Refresh started";
+            "endpoint" => %self.beacon_node,
+        );
         let previous_status = self.status(RequireSynced::Yes).await;
         let was_offline = matches!(previous_status, Err(CandidateError::Offline));
 
@@ -203,6 +208,12 @@ impl<E: EthSpec> CandidateBeaconNode<E> {
         // holding a write-lock whilst we check the online status of the node.
         *self.status.write().await = new_status;
 
+        debug!(
+            log,
+            "Refresh complete";
+            "endpoint" => %self.beacon_node,
+        );
+
         new_status
     }
 
@@ -599,46 +610,42 @@ impl<T: SlotClock, E: EthSpec> BeaconNodeFallback<T, E> {
         F: Fn(&'a BeaconNodeHttpClient) -> R,
         R: Future<Output = Result<O, Err>>,
     {
-        let mut results = vec![];
         let mut to_retry = vec![];
         let mut retry_unsynced = vec![];
 
         // Run `func` using a `candidate`, returning the value or capturing errors.
-        //
-        // We use a macro instead of a closure here since it is not trivial to move `func` into a
-        // closure.
-        macro_rules! try_func {
-            ($candidate: ident) => {{
-                inc_counter_vec(&ENDPOINT_REQUESTS, &[$candidate.beacon_node.as_ref()]);
+        let run_on_candidate = |candidate: &'a CandidateBeaconNode<E>| async {
+            inc_counter_vec(&ENDPOINT_REQUESTS, &[candidate.beacon_node.as_ref()]);
 
-                // 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.beacon_node).await {
-                    Ok(val) => results.push(Ok(val)),
-                    Err(e) => {
-                        // If we have an error on this function, make the client as not-ready.
-                        //
-                        // There exists a race condition where the candidate may have been marked
-                        // as ready between the `func` call and now. We deem this an acceptable
-                        // inefficiency.
-                        if matches!(offline_on_failure, OfflineOnFailure::Yes) {
-                            $candidate.set_offline().await;
-                        }
-                        results.push(Err((
-                            $candidate.beacon_node.to_string(),
-                            Error::RequestFailed(e),
-                        )));
-                        inc_counter_vec(&ENDPOINT_ERRORS, &[$candidate.beacon_node.as_ref()]);
+            // 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.beacon_node).await {
+                Ok(val) => Ok(val),
+                Err(e) => {
+                    // If we have an error on this function, mark the client as not-ready.
+                    //
+                    // There exists a race condition where the candidate may have been marked
+                    // as ready between the `func` call and now. We deem this an acceptable
+                    // inefficiency.
+                    if matches!(offline_on_failure, OfflineOnFailure::Yes) {
+                        candidate.set_offline().await;
                     }
+                    inc_counter_vec(&ENDPOINT_ERRORS, &[candidate.beacon_node.as_ref()]);
+                    Err((candidate.beacon_node.to_string(), Error::RequestFailed(e)))
                 }
-            }};
-        }
+            }
+        };
 
         // First pass: try `func` on all synced and ready candidates.
         //
         // This ensures that we always choose a synced node if it is available.
+        debug!(self.log, "Starting first batch of subscription requests");
+        let mut first_batch_futures = vec![];
         for candidate in &self.candidates {
             match candidate.status(RequireSynced::Yes).await {
+                Ok(_) => {
+                    first_batch_futures.push(run_on_candidate(candidate));
+                }
                 Err(CandidateError::NotSynced) if require_synced == false => {
                     // This client is unsynced we will try it after trying all synced clients
                     retry_unsynced.push(candidate);
@@ -647,22 +654,26 @@ impl<T: SlotClock, E: EthSpec> BeaconNodeFallback<T, E> {
                     // This client was not ready on the first pass, we might try it again later.
                     to_retry.push(candidate);
                 }
-                Ok(_) => try_func!(candidate),
             }
         }
+        let first_batch_results = futures::future::join_all(first_batch_futures).await;
 
         // Second pass: try `func` on ready unsynced candidates. This only runs if we permit
         // unsynced candidates.
         //
         // Due to async race-conditions, it is possible that we will send a request to a candidate
         // that has been set to an offline/unready status. This is acceptable.
-        if require_synced == false {
-            for candidate in retry_unsynced {
-                try_func!(candidate);
-            }
-        }
+        debug!(self.log, "Starting second batch of subscription requests");
+        let second_batch_results = if require_synced == false {
+            futures::future::join_all(retry_unsynced.into_iter().map(run_on_candidate)).await
+        } else {
+            vec![]
+        };
 
         // Third pass: try again, attempting to make non-ready clients become ready.
+        debug!(self.log, "Starting third batch of subscription requests");
+        let mut third_batch_futures = vec![];
+        let mut third_batch_results = vec![];
         for candidate in to_retry {
             // If the candidate hasn't luckily transferred into the correct state in the meantime,
             // force an update of the state.
@@ -676,16 +687,22 @@ impl<T: SlotClock, E: EthSpec> BeaconNodeFallback<T, E> {
             };
 
             match new_status {
-                Ok(()) => try_func!(candidate),
-                Err(CandidateError::NotSynced) if require_synced == false => try_func!(candidate),
-                Err(e) => {
-                    results.push(Err((
-                        candidate.beacon_node.to_string(),
-                        Error::Unavailable(e),
-                    )));
+                Ok(()) => third_batch_futures.push(run_on_candidate(candidate)),
+                Err(CandidateError::NotSynced) if require_synced == false => {
+                    third_batch_futures.push(run_on_candidate(candidate))
                 }
+                Err(e) => third_batch_results.push(Err((
+                    candidate.beacon_node.to_string(),
+                    Error::Unavailable(e),
+                ))),
             }
         }
+        third_batch_results.extend(futures::future::join_all(third_batch_futures).await);
+        debug!(self.log, "Completed all subscription requests");
+
+        let mut results = first_batch_results;
+        results.extend(second_batch_results);
+        results.extend(third_batch_results);
 
         let errors: Vec<_> = results.into_iter().filter_map(|res| res.err()).collect();
 

validator_client/src/lib.rs

@@ -75,6 +75,7 @@ const WAITING_FOR_GENESIS_POLL_TIME: Duration = Duration::from_secs(12);
 /// This can help ensure that proper endpoint fallback occurs.
 const HTTP_ATTESTATION_TIMEOUT_QUOTIENT: u32 = 4;
 const HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
+const HTTP_ATTESTATION_SUBSCRIPTIONS_TIMEOUT_QUOTIENT: u32 = 24;
 const HTTP_LIVENESS_TIMEOUT_QUOTIENT: u32 = 4;
 const HTTP_PROPOSAL_TIMEOUT_QUOTIENT: u32 = 2;
 const HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
@@ -323,6 +324,8 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
                 Timeouts {
                     attestation: slot_duration / HTTP_ATTESTATION_TIMEOUT_QUOTIENT,
                     attester_duties: slot_duration / HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT,
+                    attestation_subscriptions: slot_duration
+                        / HTTP_ATTESTATION_SUBSCRIPTIONS_TIMEOUT_QUOTIENT,
                     liveness: slot_duration / HTTP_LIVENESS_TIMEOUT_QUOTIENT,
                     proposal: slot_duration / HTTP_PROPOSAL_TIMEOUT_QUOTIENT,
                     proposer_duties: slot_duration / HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT,