wip

Author: Tumas

http_api_utils/src/response.rs

@@ -113,6 +113,11 @@ impl<T, M, F> EthResponse<T, M, F> {
         self.data
     }
 
+    #[must_use]
+    pub fn into_data_and_dependent_root(self) -> (T, Option<H256>) {
+        (self.data, self.dependent_root)
+    }
+
     #[must_use]
     pub const fn version(mut self, phase: Phase) -> Self {
         self.version = Some(phase);

validator/src/beacon_node_api.rs

@@ -1,20 +1,40 @@
 use core::future::Future;
 
 use anyhow::Result;
+use http_api_utils::ValidatorAttesterDutyResponse;
 use p2p::BeaconCommitteeSubscription;
 use types::{
     nonstandard::OwnAttestation,
     phase0::{
         containers::AttestationData,
-        primitives::{CommitteeIndex, Slot},
+        primitives::{CommitteeIndex, Epoch, H256, Slot, ValidatorIndex},
     },
     preset::Preset,
 };
 
+pub struct AttesterDuties {
+    pub dependent_root: H256,
+    pub duties: Vec<ValidatorAttesterDutyResponse>,
+}
+
 /// A beacon node the validator can perform duties against.
-///
-/// Implementors also implement `Display`, which names them in logs.
 pub trait BeaconNodeApi<P: Preset> {
+    // A remote beacon node reports the dependent root only alongside duties, so `validator_index`
+    // names the one validator whose duties are requested to obtain it. The built-in node derives
+    // the root without asking anything and is given `None`.
+    fn dependent_root(
+        &self,
+        epoch: Epoch,
+        validator_index: Option<ValidatorIndex>,
+    ) -> impl Future<Output = Result<H256>> + Send;
+
+    /// <https://ethereum.github.io/beacon-APIs/#/Validator/getAttesterDuties>
+    fn attester_duties(
+        &self,
+        epoch: Epoch,
+        validator_indices: &[ValidatorIndex],
+    ) -> impl Future<Output = Result<AttesterDuties>> + Send;
+
     /// <https://ethereum.github.io/beacon-APIs/#/Validator/produceAttestationData>
     fn attestation_data(
         &self,

validator/src/beacon_nodes.rs

@@ -1,4 +1,4 @@
-use core::{convert::identity, fmt::Display, future::Future};
+use core::{convert::identity, fmt::Display, future::Future, ops::Range};
 use std::sync::Arc;
 
 use anyhow::{Error as AnyhowError, Result};
@@ -12,14 +12,17 @@ use types::{
     nonstandard::{OwnAttestation, PublishedDuty},
     phase0::{
         containers::AttestationData,
-        primitives::{CommitteeIndex, Slot},
+        primitives::{CommitteeIndex, Epoch, H256, Slot, ValidatorIndex},
     },
     preset::Preset,
 };
 
 use crate::{
-    beacon_node_api::BeaconNodeApi, local_beacon_node::LocalBeaconNode,
-    remote_beacon_node::RemoteBeaconNode, remote_beacon_nodes::RemoteBeaconNodes,
+    beacon_node_api::{AttesterDuties, BeaconNodeApi},
+    local_beacon_node::LocalBeaconNode,
+    misc,
+    remote_beacon_node::RemoteBeaconNode,
+    remote_beacon_nodes::RemoteBeaconNodes,
 };
 
 pub struct BeaconNodes<P: Preset, W: Wait> {
@@ -48,9 +51,135 @@ impl<P: Preset, W: Wait + Sync> BeaconNodes<P, W> {
             .map(ArcExt::clone_arc)
             .collect()
     }
+
+    #[must_use]
+    pub const fn has_local_node(&self) -> bool {
+        self.local_node.is_some()
+    }
+
+    // The built-in beacon node computes only what is about to be needed; a remote one answers a
+    // whole epoch per request, so nothing is gained by asking for less than the epochs in flight.
+    #[must_use]
+    pub fn prefetch_slots(&self, current_slot: Slot) -> Range<Slot> {
+        if self.has_local_node() {
+            return misc::slots_to_compute_in_advance(current_slot);
+        }
+
+        let current_epoch = helper_functions::misc::compute_epoch_at_slot::<P>(current_slot);
+
+        helper_functions::misc::compute_start_slot_at_epoch::<P>(current_epoch)
+            ..helper_functions::misc::compute_start_slot_at_epoch::<P>(
+                current_epoch.saturating_add(2),
+            )
+    }
+
+    pub async fn attester_duties_at_slots(
+        &self,
+        slots: Range<Slot>,
+        validator_indices: &[ValidatorIndex],
+    ) -> Result<AttesterDuties> {
+        let epoch = helper_functions::misc::compute_epoch_at_slot::<P>(slots.start);
+
+        if let Some(node) = &self.local_node {
+            match node
+                .attester_duties_at_slots(slots, validator_indices)
+                .await
+            {
+                Ok(duties) => return Ok(duties),
+                Err(error) if self.remote_nodes.is_empty() => return Err(error),
+                Err(error) => {
+                    warn_with_peers!(
+                        "{node} beacon node failed to produce attester duties for epoch \
+                         {epoch}: {error:?}",
+                    );
+                }
+            }
+        }
+
+        self.remote_attester_duties(epoch, validator_indices).await
+    }
+
+    async fn remote_attester_duties(
+        &self,
+        epoch: Epoch,
+        validator_indices: &[ValidatorIndex],
+    ) -> Result<AttesterDuties> {
+        let operation = format!("produce attester duties for epoch {epoch}");
+        let mut attempts = Vec::with_capacity(self.remote_nodes.len());
+
+        for node in &self.remote_nodes {
+            attempts.push((
+                node.as_ref(),
+                BeaconNodeApi::<P>::attester_duties(node.as_ref(), epoch, validator_indices),
+            ));
+        }
+
+        let (node, duties) = first_success(&operation, attempts).await?;
+
+        debug_with_peers!(
+            "{node} beacon node produced {} attester duties for epoch {epoch} \
+             under dependent root {:?}",
+            duties.duties.len(),
+            duties.dependent_root,
+        );
+
+        Ok(duties)
+    }
 }
 
 impl<P: Preset, W: Wait + Sync> BeaconNodeApi<P> for BeaconNodes<P, W> {
+    async fn dependent_root(
+        &self,
+        epoch: Epoch,
+        validator_index: Option<ValidatorIndex>,
+    ) -> Result<H256> {
+        let operation = format!("produce the dependent root of epoch {epoch}");
+
+        if let Some(node) = &self.local_node {
+            match node.dependent_root(epoch, validator_index).await {
+                Ok(dependent_root) => return Ok(dependent_root),
+                Err(error) if self.remote_nodes.is_empty() => return Err(error),
+                Err(error) => {
+                    warn_with_peers!("{node} beacon node failed to {operation}: {error:?}");
+                }
+            }
+        }
+
+        let mut attempts = Vec::with_capacity(self.remote_nodes.len());
+
+        for node in &self.remote_nodes {
+            attempts.push((
+                node.as_ref(),
+                BeaconNodeApi::<P>::dependent_root(node.as_ref(), epoch, validator_index),
+            ));
+        }
+
+        first_success(&operation, attempts)
+            .await
+            .map(|(_, dependent_root)| dependent_root)
+    }
+
+    async fn attester_duties(
+        &self,
+        epoch: Epoch,
+        validator_indices: &[ValidatorIndex],
+    ) -> Result<AttesterDuties> {
+        if let Some(node) = &self.local_node {
+            match node.attester_duties(epoch, validator_indices).await {
+                Ok(duties) => return Ok(duties),
+                Err(error) if self.remote_nodes.is_empty() => return Err(error),
+                Err(error) => {
+                    warn_with_peers!(
+                        "{node} beacon node failed to produce attester duties for epoch \
+                         {epoch}: {error:?}",
+                    );
+                }
+            }
+        }
+
+        self.remote_attester_duties(epoch, validator_indices).await
+    }
+
     async fn attestation_data(
         &self,
         slot: Slot,
@@ -67,6 +196,7 @@ impl<P: Preset, W: Wait + Sync> BeaconNodeApi<P> for BeaconNodes<P, W> {
 
                     return Ok(data);
                 }
+                Err(error) if self.remote_nodes.is_empty() => return Err(error),
                 Err(error) => {
                     warn_with_peers!("{node} beacon node failed to {operation}: {error:?}");
                 }
@@ -144,8 +274,7 @@ impl<P: Preset, W: Wait + Sync> BeaconNodeApi<P> for BeaconNodes<P, W> {
         if !subscribe_on.is_empty() {
             let subscriptions = Arc::new(subscriptions.to_vec());
 
-            // Every node is told, not just the first one that accepts: any of them may be asked to
-            // produce or aggregate later in the slot.
+            // Every node is told, as any of them may be asked to produce or aggregate later.
             spawn_broadcast(
                 "update beacon committee subscriptions",
                 subscribe_on,
@@ -174,12 +303,7 @@ impl<P: Preset, W: Wait + Sync> BeaconNodeApi<P> for BeaconNodes<P, W> {
     }
 }
 
-/// Publishes to the first remote node that accepts, without holding up the caller.
-///
-/// The walk is detached rather than each node separately, so that the nodes are still tried in
-/// order: deciding whether to fall through to the next one needs the previous one's result.
-/// Attestations are published from the validator's main loop, so awaiting the walk there would let
-/// a single unresponsive node delay every later duty in the slot.
+// Detached as a whole rather than per node, so that the nodes are still tried in order.
 fn spawn_publish<P: Preset>(
     remotes: Vec<Arc<RemoteBeaconNode>>,
     attestations: Vec<OwnAttestation<P>>,
@@ -203,9 +327,7 @@ fn spawn_publish<P: Preset>(
     });
 }
 
-/// Runs `attempt` against every remote node, without holding up the caller.
-///
-/// Detached for the same reason as [`spawn_publish`].
+// Detached for the same reason as `spawn_publish`.
 fn spawn_broadcast<F, Fut>(operation: &'static str, remotes: Vec<Arc<RemoteBeaconNode>>, attempt: F)
 where
     F: Fn(Arc<RemoteBeaconNode>) -> Fut + Send + 'static,

validator/src/health.rs

@@ -15,8 +15,7 @@ impl Health {
         matches!(self, Self::Ready)
     }
 
-    /// A degraded node is still worth asking, but an [`Self::Incompatible`] one would answer under
-    /// the wrong domain and an [`Self::Unreachable`] one would only burn the deadline.
+    // A degraded node is still worth asking, unlike one on the wrong network or an unreachable one.
     #[must_use]
     pub const fn can_serve(self) -> bool {
         matches!(self, Self::Unusable | Self::Unknown | Self::Ready)