Add error handling to iterators (#1243)

* Add error handling to iterators

* Review feedback

* Leverage itertools::process_results() in few places

beacon_node/beacon_chain/Cargo.toml

@@ -48,7 +48,7 @@ bls = { path = "../../crypto/bls" }
 safe_arith = { path = "../../consensus/safe_arith" }
 environment = { path = "../../lighthouse/environment" }
 bus = "2.2.3"
+itertools = "0.9.0"
 
 [dev-dependencies]
 lazy_static = "1.4.0"
-

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -23,6 +23,7 @@ use crate::snapshot_cache::SnapshotCache;
 use crate::timeout_rw_lock::TimeoutRwLock;
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::BeaconSnapshot;
+use itertools::process_results;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use slog::{crit, debug, error, info, trace, warn, Logger};
 use slot_clock::SlotClock;
@@ -319,12 +320,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// - Iterator returns `(Hash256, Slot)`.
     /// - As this iterator starts at the `head` of the chain (viz., the best block), the first slot
     ///     returned may be earlier than the wall-clock slot.
-    pub fn rev_iter_block_roots(&self) -> Result<impl Iterator<Item = (Hash256, Slot)>, Error> {
+    pub fn rev_iter_block_roots(
+        &self,
+    ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
         let head = self.head()?;
-
         let iter = BlockRootsIterator::owned(self.store.clone(), head.beacon_state);
-
-        Ok(std::iter::once((head.beacon_block_root, head.beacon_block.slot())).chain(iter))
+        Ok(
+            std::iter::once(Ok((head.beacon_block_root, head.beacon_block.slot())))
+                .chain(iter)
+                .map(|result| result.map_err(|e| e.into())),
+        )
     }
 
     pub fn forwards_iter_block_roots(
@@ -339,7 +344,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             local_head.beacon_state,
             local_head.beacon_block_root,
             &self.spec,
-        ))
+        )?)
     }
 
     /// Traverse backwards from `block_root` to find the block roots of its ancestors.
@@ -354,15 +359,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn rev_iter_block_roots_from(
         &self,
         block_root: Hash256,
-    ) -> Result<impl Iterator<Item = (Hash256, Slot)>, Error> {
+    ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
         let block = self
             .get_block(&block_root)?
             .ok_or_else(|| Error::MissingBeaconBlock(block_root))?;
         let state = self
             .get_state(&block.state_root(), Some(block.slot()))?
             .ok_or_else(|| Error::MissingBeaconState(block.state_root()))?;
         let iter = BlockRootsIterator::owned(self.store.clone(), state);
-        Ok(std::iter::once((block_root, block.slot())).chain(iter))
+        Ok(std::iter::once(Ok((block_root, block.slot())))
+            .chain(iter)
+            .map(|result| result.map_err(|e| e.into())))
     }
 
     /// Traverse backwards from `block_root` to find the root of the ancestor block at `slot`.
@@ -371,10 +378,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block_root: Hash256,
         slot: Slot,
     ) -> Result<Option<Hash256>, Error> {
-        Ok(self
-            .rev_iter_block_roots_from(block_root)?
-            .find(|(_, ancestor_slot)| *ancestor_slot == slot)
-            .map(|(ancestor_block_root, _)| ancestor_block_root))
+        process_results(self.rev_iter_block_roots_from(block_root)?, |mut iter| {
+            iter.find(|(_, ancestor_slot)| *ancestor_slot == slot)
+                .map(|(ancestor_block_root, _)| ancestor_block_root)
+        })
     }
 
     /// Iterates across all `(state_root, slot)` pairs from the head of the chain (inclusive) to
@@ -386,13 +393,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// - Iterator returns `(Hash256, Slot)`.
     /// - As this iterator starts at the `head` of the chain (viz., the best block), the first slot
     ///     returned may be earlier than the wall-clock slot.
-    pub fn rev_iter_state_roots(&self) -> Result<impl Iterator<Item = (Hash256, Slot)>, Error> {
+    pub fn rev_iter_state_roots(
+        &self,
+    ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
         let head = self.head()?;
         let slot = head.beacon_state.slot;
-
         let iter = StateRootsIterator::owned(self.store.clone(), head.beacon_state);
-
-        Ok(std::iter::once((head.beacon_state_root, slot)).chain(iter))
+        let iter = std::iter::once(Ok((head.beacon_state_root, slot)))
+            .chain(iter)
+            .map(|result| result.map_err(Into::into));
+        Ok(iter)
     }
 
     /// Returns the block at the given slot, if any. Only returns blocks in the canonical chain.
@@ -404,10 +414,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         slot: Slot,
     ) -> Result<Option<SignedBeaconBlock<T::EthSpec>>, Error> {
-        let root = self
-            .rev_iter_block_roots()?
-            .find(|(_, this_slot)| *this_slot == slot)
-            .map(|(root, _)| root);
+        let root = process_results(self.rev_iter_block_roots()?, |mut iter| {
+            iter.find(|(_, this_slot)| *this_slot == slot)
+                .map(|(root, _)| root)
+        })?;
 
         if let Some(block_root) = root {
             Ok(self.store.get_item(&block_root)?)
@@ -553,12 +563,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 Ok(state)
             }
             Ordering::Less => {
-                let state_root = self
-                    .rev_iter_state_roots()?
-                    .take_while(|(_root, current_slot)| *current_slot >= slot)
-                    .find(|(_root, current_slot)| *current_slot == slot)
-                    .map(|(root, _slot)| root)
-                    .ok_or_else(|| Error::NoStateForSlot(slot))?;
+                let state_root = process_results(self.rev_iter_state_roots()?, |iter| {
+                    iter.take_while(|(_, current_slot)| *current_slot >= slot)
+                        .find(|(_, current_slot)| *current_slot == slot)
+                        .map(|(root, _slot)| root)
+                })?
+                .ok_or_else(|| Error::NoStateForSlot(slot))?;
 
                 Ok(self
                     .get_state(&state_root, Some(slot))?
@@ -633,10 +643,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ///
     /// Returns None if a block doesn't exist at the slot.
     pub fn root_at_slot(&self, target_slot: Slot) -> Result<Option<Hash256>, Error> {
-        Ok(self
-            .rev_iter_block_roots()?
-            .find(|(_root, slot)| *slot == target_slot)
-            .map(|(root, _slot)| root))
+        process_results(self.rev_iter_block_roots()?, |mut iter| {
+            iter.find(|(_, slot)| *slot == target_slot)
+                .map(|(root, _)| root)
+        })
     }
 
     /// Returns the block proposer for a given slot.

beacon_node/beacon_chain/src/migrate.rs

@@ -84,9 +84,10 @@ pub trait Migrate<E: EthSpec>: Send + Sync + 'static {
                 .ok_or_else(|| BeaconStateError::MissingBeaconBlock(head_hash.into()))?
                 .state_root();
 
-            let iterator = std::iter::once((head_hash, head_state_hash, head_slot))
+            let iter = std::iter::once(Ok((head_hash, head_state_hash, head_slot)))
                 .chain(RootsIterator::from_block(Arc::clone(&store), head_hash)?);
-            for (block_hash, state_hash, slot) in iterator {
+            for maybe_tuple in iter {
+                let (block_hash, state_hash, slot) = maybe_tuple?;
                 if slot < old_finalized_slot {
                     // We must assume here any candidate chains include old_finalized_block_hash,
                     // i.e. there aren't any forks starting at a block that is a strict ancestor of

beacon_node/beacon_chain/tests/store_tests.rs

@@ -392,6 +392,7 @@ fn delete_blocks_and_states() {
         .expect("faulty head state exists");
 
     let states_to_delete = StateRootsIterator::new(store.clone(), &faulty_head_state)
+        .map(Result::unwrap)
         .take_while(|(_, slot)| *slot > unforked_blocks)
         .collect::<Vec<_>>();
 
@@ -409,6 +410,7 @@ fn delete_blocks_and_states() {
 
     // Deleting the blocks from the fork should remove them completely
     let blocks_to_delete = BlockRootsIterator::new(store.clone(), &faulty_head_state)
+        .map(Result::unwrap)
         // Extra +1 here accounts for the skipped slot that started this fork
         .take_while(|(_, slot)| *slot > unforked_blocks + 1)
         .collect::<Vec<_>>();
@@ -424,6 +426,7 @@ fn delete_blocks_and_states() {
         .chain
         .rev_iter_state_roots()
         .expect("rev iter ok")
+        .map(Result::unwrap)
         .filter(|(_, slot)| *slot < split_slot);
 
     for (state_root, slot) in finalized_states {
@@ -659,11 +662,12 @@ fn check_shuffling_compatible(
     let previous_pivot_slot =
         (head_state.previous_epoch() - shuffling_lookahead).end_slot(E::slots_per_epoch());
 
-    for (block_root, slot) in harness
+    for maybe_tuple in harness
         .chain
         .rev_iter_block_roots_from(head_block_root)
         .unwrap()
     {
+        let (block_root, slot) = maybe_tuple.unwrap();
         // Shuffling is compatible targeting the current epoch,
         // iff slot is greater than or equal to the current epoch pivot block
         assert_eq!(
@@ -1364,6 +1368,8 @@ fn check_chain_dump(harness: &TestHarness, expected_len: u64) {
         head.beacon_block_root,
         &harness.spec,
     )
+    .unwrap()
+    .map(Result::unwrap)
     .collect::<Vec<_>>();
 
     // Drop the block roots for skipped slots.
@@ -1387,6 +1393,7 @@ fn check_iterators(harness: &TestHarness) {
             .rev_iter_state_roots()
             .expect("should get iter")
             .last()
+            .map(Result::unwrap)
             .map(|(_, slot)| slot),
         Some(Slot::new(0))
     );
@@ -1396,6 +1403,7 @@ fn check_iterators(harness: &TestHarness) {
             .rev_iter_block_roots()
             .expect("should get iter")
             .last()
+            .map(Result::unwrap)
             .map(|(_, slot)| slot),
         Some(Slot::new(0))
     );

beacon_node/beacon_chain/tests/tests.rs

@@ -73,11 +73,13 @@ fn iterators() {
         .chain
         .rev_iter_block_roots()
         .expect("should get iter")
+        .map(Result::unwrap)
         .collect();
     let state_roots: Vec<(Hash256, Slot)> = harness
         .chain
         .rev_iter_state_roots()
         .expect("should get iter")
+        .map(Result::unwrap)
         .collect();
 
     assert_eq!(

beacon_node/network/Cargo.toml

@@ -37,3 +37,4 @@ rlp = "0.4.5"
 lazy_static = "1.4.0"
 lighthouse_metrics = { path = "../../common/lighthouse_metrics" }
 environment = { path = "../../lighthouse/environment" }
+itertools = "0.9.0"

beacon_node/network/src/router/processor.rs

@@ -9,6 +9,7 @@ use beacon_chain::{
 };
 use eth2_libp2p::rpc::*;
 use eth2_libp2p::{NetworkGlobals, PeerId, Request, Response};
+use itertools::process_results;
 use slog::{debug, error, o, trace, warn};
 use ssz::Encode;
 use std::sync::Arc;
@@ -357,20 +358,29 @@ impl<T: BeaconChainTypes> Processor<T> {
 
         // pick out the required blocks, ignoring skip-slots and stepping by the step parameter;
         let mut last_block_root = None;
-        let block_roots = forwards_block_root_iter
-            .take_while(|(_root, slot)| slot.as_u64() < req.start_slot + req.count * req.step)
-            // map skip slots to None
-            .map(|(root, _slot)| {
-                let result = if Some(root) == last_block_root {
-                    None
-                } else {
-                    Some(root)
-                };
-                last_block_root = Some(root);
-                result
-            })
-            .step_by(req.step as usize)
-            .collect::<Vec<_>>();
+        let maybe_block_roots = process_results(forwards_block_root_iter, |iter| {
+            iter.take_while(|(_, slot)| slot.as_u64() < req.start_slot + req.count * req.step)
+                // map skip slots to None
+                .map(|(root, _)| {
+                    let result = if Some(root) == last_block_root {
+                        None
+                    } else {
+                        Some(root)
+                    };
+                    last_block_root = Some(root);
+                    result
+                })
+                .step_by(req.step as usize)
+                .collect::<Vec<Option<Hash256>>>()
+        });
+
+        let block_roots = match maybe_block_roots {
+            Ok(block_roots) => block_roots,
+            Err(e) => {
+                error!(self.log, "Error during iteration over blocks"; "error" => format!("{:?}", e));
+                return;
+            }
+        };
 
         // remove all skip slots
         let block_roots = block_roots

beacon_node/rest_api/Cargo.toml

@@ -39,6 +39,7 @@ rayon = "1.3.0"
 environment = { path = "../../lighthouse/environment" }
 uhttp_sse = "0.5.1"
 bus = "2.2.3"
+itertools = "0.9.0"
 
 [dev-dependencies]
 assert_matches = "1.3.0"

beacon_node/rest_api/src/helpers.rs

@@ -5,6 +5,7 @@ use eth2_libp2p::PubsubMessage;
 use hex;
 use http::header;
 use hyper::{Body, Request};
+use itertools::process_results;
 use network::NetworkMessage;
 use ssz::Decode;
 use store::{iter::AncestorIter, Store};
@@ -118,11 +119,14 @@ pub fn block_root_at_slot<T: BeaconChainTypes>(
     beacon_chain: &BeaconChain<T>,
     target: Slot,
 ) -> Result<Option<Hash256>, ApiError> {
-    Ok(beacon_chain
-        .rev_iter_block_roots()?
-        .take_while(|(_root, slot)| *slot >= target)
-        .find(|(_root, slot)| *slot == target)
-        .map(|(root, _slot)| root))
+    Ok(process_results(
+        beacon_chain.rev_iter_block_roots()?,
+        |iter| {
+            iter.take_while(|(_, slot)| *slot >= target)
+                .find(|(_, slot)| *slot == target)
+                .map(|(root, _)| root)
+        },
+    )?)
 }
 
 /// Returns a `BeaconState` and it's root in the canonical chain of `beacon_chain` at the given
@@ -190,12 +194,15 @@ pub fn state_root_at_slot<T: BeaconChainTypes>(
         //
         // Iterate through the state roots on the head state to find the root for that
         // slot. Once the root is found, load it from the database.
-        Ok(head_state
-            .try_iter_ancestor_roots(beacon_chain.store.clone())
-            .ok_or_else(|| ApiError::ServerError("Failed to create roots iterator".to_string()))?
-            .find(|(_root, s)| *s == slot)
-            .map(|(root, _slot)| root)
-            .ok_or_else(|| ApiError::NotFound(format!("Unable to find state at slot {}", slot)))?)
+        process_results(
+            head_state
+                .try_iter_ancestor_roots(beacon_chain.store.clone())
+                .ok_or_else(|| {
+                    ApiError::ServerError("Failed to create roots iterator".to_string())
+                })?,
+            |mut iter| iter.find(|(_, s)| *s == slot).map(|(root, _)| root),
+        )?
+        .ok_or_else(|| ApiError::NotFound(format!("Unable to find state at slot {}", slot)))
     } else {
         // 4. The request slot is later than the head slot.
         //

beacon_node/rest_api/tests/test.rs

@@ -759,6 +759,7 @@ fn get_genesis_state_root() {
         .expect("should have beacon chain")
         .rev_iter_state_roots()
         .expect("should get iter")
+        .map(Result::unwrap)
         .find(|(_cur_root, cur_slot)| slot == *cur_slot)
         .map(|(cur_root, _)| cur_root)
         .expect("chain should have state root at slot");
@@ -786,6 +787,7 @@ fn get_genesis_block_root() {
         .expect("should have beacon chain")
         .rev_iter_block_roots()
         .expect("should get iter")
+        .map(Result::unwrap)
         .find(|(_cur_root, cur_slot)| slot == *cur_slot)
         .map(|(cur_root, _)| cur_root)
         .expect("chain should have state root at slot");