Fix errors in duties request

sigp · Sep 17, 2020 · 6aa9f30 · 6aa9f30
1 parent 40d285a
commit 6aa9f30
Show file tree

Hide file tree

Showing 4 changed files with 58 additions and 13 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs
@@ -21,16 +21,17 @@ use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use slog::{crit, error, info, trace, warn, Logger};
 use state_id::StateId;
+use state_processing::per_slot_processing;
 use std::borrow::Cow;
 use std::convert::TryInto;
 use std::future::Future;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
 use types::{
-    Attestation, AttestationDuty, AttesterSlashing, CommitteeCache, Epoch, EthSpec,
-    ProposerSlashing, PublicKey, RelativeEpoch, SignedAggregateAndProof, SignedBeaconBlock,
-    SignedVoluntaryExit, Slot, YamlConfig,
+    Attestation, AttestationDuty, AttesterSlashing, CloneConfig, CommitteeCache, Epoch, EthSpec,
+    Hash256, ProposerSlashing, PublicKey, RelativeEpoch, SignedAggregateAndProof,
+    SignedBeaconBlock, SignedVoluntaryExit, Slot, YamlConfig,
 };
 use warp::Filter;
 
@@ -1083,17 +1084,13 @@ pub fn serve<T: BeaconChainTypes>(
                         .epoch()
                         .map_err(warp_utils::reject::beacon_chain_error)?;
 
-                    dbg!(epoch, current_epoch);
-
                     if epoch > current_epoch + 1 {
                         return Err(warp_utils::reject::custom_bad_request(format!(
                             "request epoch {} is more than one epoch past the current epoch {}",
                             epoch, current_epoch
                         )));
                     }
 
-                    println!("passed: {}, {}", epoch, current_epoch);
-
                     let validator_count = StateId::head()
                         .map_state(&chain, |state| Ok(state.validators.len() as u64))?;
 
@@ -1172,19 +1169,53 @@ pub fn serve<T: BeaconChainTypes>(
                             })
                             .collect::<Result<Vec<_>, warp::Rejection>>()?
                     } else {
-                        // Slow path.
-                        let mut state =
-                            StateId::slot(epoch.start_slot(T::EthSpec::slots_per_epoch()))
-                                .state(&chain)?;
+                        // If the head state is equal to or earlier than the request epoch, use it.
+                        let mut state = chain
+                            .with_head(|head| {
+                                if head.beacon_state.current_epoch() <= epoch {
+                                    Ok(Some(
+                                        head.beacon_state
+                                            .clone_with(CloneConfig::committee_caches_only()),
+                                    ))
+                                } else {
+                                    Ok(None)
+                                }
+                            })
+                            .map_err(warp_utils::reject::beacon_chain_error)?
+                            .map(Result::Ok)
+                            .unwrap_or_else(|| {
+                                StateId::slot(epoch.start_slot(T::EthSpec::slots_per_epoch()))
+                                    .state(&chain)
+                            })?;
+
+                        // Only skip forward to the epoch prior to the request, since we have a
+                        // one-epoch look-ahead on shuffling.
+                        while state.next_epoch() < epoch {
+                            // Don't calculate state roots since they aren't required for calculating
+                            // shuffling (achieved by providing Hash256::zero()).
+                            per_slot_processing(&mut state, Some(Hash256::zero()), &chain.spec)
+                                .map_err(warp_utils::reject::slot_processing_error)?;
+                        }
+
+                        let relative_epoch =
+                            RelativeEpoch::from_epoch(state.current_epoch(), epoch).map_err(
+                                |e| {
+                                    warp_utils::reject::custom_server_error(format!(
+                                        "unable to obtain suitable state: {:?}",
+                                        e
+                                    ))
+                                },
+                            )?;
+
                         state
-                            .build_committee_cache(RelativeEpoch::Current, &chain.spec)
+                            .build_committee_cache(relative_epoch, &chain.spec)
                             .map_err(warp_utils::reject::beacon_state_error)?;
                         pubkeys
                             .into_iter()
                             .filter_map(|(i, pubkey)| {
                                 Some(
                                     state
-                                        .get_attestation_duties(i as usize, RelativeEpoch::Current)
+                                        .get_attestation_duties(i as usize, relative_epoch)
                                         .transpose()?
                                         .map_err(warp_utils::reject::beacon_state_error)
                                         .map(|duty| convert(i, pubkey, duty)),

diff --git a/common/warp_utils/Cargo.toml b/common/warp_utils/Cargo.toml
@@ -11,3 +11,4 @@ warp = "0.2.5"
 eth2 = { path = "../eth2" }
 types = { path = "../../consensus/types" }
 beacon_chain = { path = "../../beacon_node/beacon_chain" }
+state_processing = { path = "../../consensus/state_processing" }
diff --git a/common/warp_utils/src/reject.rs b/common/warp_utils/src/reject.rs
@@ -20,6 +20,15 @@ pub fn beacon_state_error(e: types::BeaconStateError) -> warp::reject::Rejection
     warp::reject::custom(BeaconStateError(e))
 }
 
+#[derive(Debug)]
+pub struct SlotProcessingError(pub state_processing::SlotProcessingError);
+
+impl Reject for SlotProcessingError {}
+
+pub fn slot_processing_error(e: state_processing::SlotProcessingError) -> warp::reject::Rejection {
+    warp::reject::custom(SlotProcessingError(e))
+}
+
 #[derive(Debug)]
 pub struct BlockProductionError(pub beacon_chain::BlockProductionError);
 
@@ -95,6 +104,9 @@ pub async fn handle_rejection(err: warp::Rejection) -> Result<impl warp::Reply,
     } else if let Some(e) = err.find::<crate::reject::BeaconStateError>() {
         code = StatusCode::INTERNAL_SERVER_ERROR;
         message = format!("UNHANDLED_ERROR: {:?}", e.0);
+    } else if let Some(e) = err.find::<crate::reject::SlotProcessingError>() {
+        code = StatusCode::INTERNAL_SERVER_ERROR;
+        message = format!("UNHANDLED_ERROR: {:?}", e.0);
     } else if let Some(e) = err.find::<crate::reject::BlockProductionError>() {
         code = StatusCode::INTERNAL_SERVER_ERROR;
         message = format!("UNHANDLED_ERROR: {:?}", e.0);