Add error handling to iterators

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -319,12 +319,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> {
-        let head = self.head()?;
-
+    pub fn rev_iter_block_roots(
+        &self,
+    ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
+        let head = self.head().map_err(|e| Error::from(e))?;
         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 +343,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 +358,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`.
@@ -373,7 +379,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ) -> Result<Option<Hash256>, Error> {
         Ok(self
             .rev_iter_block_roots_from(block_root)?
-            .find(|(_, ancestor_slot)| *ancestor_slot == slot)
+            .find(|result| match result {
+                Ok((_, ancestor_slot)) => *ancestor_slot == slot,
+                Err(_) => true,
+            })
+            .transpose()?
             .map(|(ancestor_block_root, _)| ancestor_block_root))
     }
 
@@ -386,13 +396,14 @@ 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<Box<dyn 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);
+        Ok(Box::new(iter.map(|result| result.map_err(|e| e.into()))))
     }
 
     /// Returns the block at the given slot, if any. Only returns blocks in the canonical chain.
@@ -406,7 +417,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ) -> Result<Option<SignedBeaconBlock<T::EthSpec>>, Error> {
         let root = self
             .rev_iter_block_roots()?
-            .find(|(_, this_slot)| *this_slot == slot)
+            .find(|result| match result {
+                Ok((_, this_slot)) => *this_slot == slot,
+                Err(_) => true,
+            })
+            .transpose()?
             .map(|(root, _)| root);
 
         if let Some(block_root) = root {
@@ -555,8 +570,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             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)
+                    .take_while(|result| match result {
+                        Ok((_, current_slot)) => *current_slot >= slot,
+                        Err(_) => true,
+                    })
+                    .find(|result| match result {
+                        Ok((_, current_slot)) => *current_slot == slot,
+                        Err(_) => true,
+                    })
+                    .transpose()?
                     .map(|(root, _slot)| root)
                     .ok_or_else(|| Error::NoStateForSlot(slot))?;
 
@@ -635,8 +657,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     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))
+            .find(|result| match result {
+                Ok((_, slot)) => *slot == target_slot,
+                Err(_) => true,
+            })
+            .transpose()?
+            .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/src/router/processor.rs

@@ -12,6 +12,7 @@ use eth2_libp2p::{NetworkGlobals, PeerId, Request, Response};
 use slog::{debug, error, o, trace, warn};
 use ssz::Encode;
 use std::sync::Arc;
+use store::errors::Error as StoreError;
 use store::Store;
 use tokio::sync::mpsc;
 use types::{
@@ -357,20 +358,33 @@ 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)
+        let maybe_block_roots: Result<Vec<Option<Hash256>>, StoreError> = forwards_block_root_iter
+            .take_while(|result| match result {
+                Ok((_, slot)) => slot.as_u64() < req.start_slot + req.count * req.step,
+                Err(_) => true,
+            })
             // 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
+            .map(|result| {
+                result.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<_>>();
+            .collect::<Result<_, _>>();
+
+        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/src/helpers.rs

@@ -120,9 +120,16 @@ pub fn block_root_at_slot<T: BeaconChainTypes>(
 ) -> 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))
+        .take_while(|result| match result {
+            Ok((_, slot)) => *slot >= target,
+            Err(_) => true,
+        })
+        .find(|result| match result {
+            Ok((_, slot)) => *slot == target,
+            Err(_) => true,
+        })
+        .transpose()?
+        .map(|(root, _)| root))
 }
 
 /// Returns a `BeaconState` and it's root in the canonical chain of `beacon_chain` at the given
@@ -193,8 +200,12 @@ pub fn state_root_at_slot<T: BeaconChainTypes>(
         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)
+            .find(|result| match result {
+                Ok((_, s)) => *s == slot,
+                Err(_) => true,
+            })
+            .transpose()?
+            .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");

beacon_node/store/src/errors.rs

@@ -3,6 +3,8 @@ use crate::hot_cold_store::HotColdDBError;
 use ssz::DecodeError;
 use types::{BeaconStateError, Hash256};
 
+pub type Result<T> = std::result::Result<T, Error>;
+
 #[derive(Debug)]
 pub enum Error {
     SszDecodeError(DecodeError),