Persist eth1 cache (#760)

* Add intermediate structures for bytes conversion

* Expose byte conversion methods from `Eth1Service`

* Add eth1 ssz containers

* Fix type errors

* Load eth1 cache on restart

* Fix compile errors

* Update Cargo.lock

* Add comments and minor formatting

* Add test for eth1 cache persistence

* Restrict Deposit and Block cache field visibility

* Add checks

* Fix `SszDepositCache` check

* Implement Encode/Decode directly on `BlockCache`

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -116,7 +116,7 @@ pub trait BeaconChainTypes: Send + Sync + 'static {
     type StoreMigrator: store::Migrate<Self::Store, Self::EthSpec>;
     type SlotClock: slot_clock::SlotClock;
     type LmdGhost: LmdGhost<Self::Store, Self::EthSpec>;
-    type Eth1Chain: Eth1ChainBackend<Self::EthSpec>;
+    type Eth1Chain: Eth1ChainBackend<Self::EthSpec, Self::Store>;
     type EthSpec: types::EthSpec;
     type EventHandler: EventHandler<Self::EthSpec>;
 }
@@ -135,7 +135,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// inclusion in a block.
     pub op_pool: OperationPool<T::EthSpec>,
     /// Provides information from the Ethereum 1 (PoW) chain.
-    pub eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec>>,
+    pub eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec, T::Store>>,
     /// Stores a "snapshot" of the chain at the time the head-of-the-chain block was received.
     pub(crate) canonical_head: TimeoutRwLock<CheckPoint<T::EthSpec>>,
     /// The root of the genesis block.
@@ -190,6 +190,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             genesis_block_root: self.genesis_block_root,
             ssz_head_tracker: self.head_tracker.to_ssz_container(),
             fork_choice: self.fork_choice.as_ssz_container(),
+            eth1_cache: self.eth1_chain.as_ref().map(|x| x.as_ssz_container()),
             block_root_tree: self.block_root_tree.as_ssz_container(),
         };
 

beacon_node/beacon_chain/src/builder.rs

@@ -57,7 +57,7 @@ where
     TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
-    TEth1Backend: Eth1ChainBackend<TEthSpec> + 'static,
+    TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
 {
@@ -87,7 +87,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     genesis_block_root: Option<Hash256>,
     op_pool: Option<OperationPool<T::EthSpec>>,
     fork_choice: Option<ForkChoice<T>>,
-    eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec>>,
+    eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec, T::Store>>,
     event_handler: Option<T::EventHandler>,
     slot_clock: Option<T::SlotClock>,
     persisted_beacon_chain: Option<PersistedBeaconChain<T>>,
@@ -114,7 +114,7 @@ where
     TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
-    TEth1Backend: Eth1ChainBackend<TEthSpec> + 'static,
+    TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
 {
@@ -175,7 +175,7 @@ where
     /// Attempt to load an existing chain from the builder's `Store`.
     ///
     /// May initialize several components; including the op_pool and finalized checkpoints.
-    pub fn resume_from_db(mut self) -> Result<Self, String> {
+    pub fn resume_from_db(mut self, config: Eth1Config) -> Result<Self, String> {
         let log = self
             .log
             .as_ref()
@@ -226,6 +226,10 @@ where
             HeadTracker::from_ssz_container(&p.ssz_head_tracker)
                 .map_err(|e| format!("Failed to decode head tracker for database: {:?}", e))?,
         );
+        self.eth1_chain = match &p.eth1_cache {
+            Some(cache) => Some(Eth1Chain::from_ssz_container(cache, config, store, log)?),
+            None => None,
+        };
         self.block_root_tree = Some(Arc::new(p.block_root_tree.clone().into()));
         self.persisted_beacon_chain = Some(p);
 
@@ -422,7 +426,7 @@ where
     TStore: Store<TEthSpec> + 'static,
     TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
-    TEth1Backend: Eth1ChainBackend<TEthSpec> + 'static,
+    TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
 {
@@ -541,7 +545,7 @@ where
     TStore: Store<TEthSpec> + 'static,
     TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
     TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
-    TEth1Backend: Eth1ChainBackend<TEthSpec> + 'static,
+    TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
 {
@@ -583,7 +587,7 @@ where
     TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
-    TEth1Backend: Eth1ChainBackend<TEthSpec> + 'static,
+    TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
 {
     /// Sets the `BeaconChain` event handler to `NullEventHandler`.

beacon_node/beacon_chain/src/eth1_chain.rs

@@ -6,6 +6,7 @@ use futures::Future;
 use integer_sqrt::IntegerSquareRoot;
 use rand::prelude::*;
 use slog::{crit, debug, error, trace, Logger};
+use ssz_derive::{Decode, Encode};
 use state_processing::per_block_processing::get_new_eth1_data;
 use std::collections::HashMap;
 use std::iter::DoubleEndedIterator;
@@ -48,23 +49,31 @@ pub enum Error {
     UnknownPreviousEth1BlockHash,
 }
 
+#[derive(Encode, Decode, Clone)]
+pub struct SszEth1 {
+    use_dummy_backend: bool,
+    backend_bytes: Vec<u8>,
+}
+
 /// Holds an `Eth1ChainBackend` and serves requests from the `BeaconChain`.
-pub struct Eth1Chain<T, E>
+pub struct Eth1Chain<T, E, S>
 where
-    T: Eth1ChainBackend<E>,
+    T: Eth1ChainBackend<E, S>,
     E: EthSpec,
+    S: Store<E>,
 {
     backend: T,
     /// When `true`, the backend will be ignored and dummy data from the 2019 Canada interop method
     /// will be used instead.
     pub use_dummy_backend: bool,
-    _phantom: PhantomData<E>,
+    _phantom: PhantomData<(E, S)>,
 }
 
-impl<T, E> Eth1Chain<T, E>
+impl<T, E, S> Eth1Chain<T, E, S>
 where
-    T: Eth1ChainBackend<E>,
+    T: Eth1ChainBackend<E, S>,
     E: EthSpec,
+    S: Store<E>,
 {
     pub fn new(backend: T) -> Self {
         Self {
@@ -82,7 +91,8 @@ where
         spec: &ChainSpec,
     ) -> Result<Eth1Data, Error> {
         if self.use_dummy_backend {
-            DummyEth1ChainBackend::default().eth1_data(state, spec)
+            let dummy_backend: DummyEth1ChainBackend<E, S> = DummyEth1ChainBackend::default();
+            dummy_backend.eth1_data(state, spec)
         } else {
             self.backend.eth1_data(state, spec)
         }
@@ -103,14 +113,41 @@ where
         spec: &ChainSpec,
     ) -> Result<Vec<Deposit>, Error> {
         if self.use_dummy_backend {
-            DummyEth1ChainBackend::default().queued_deposits(state, eth1_data_vote, spec)
+            let dummy_backend: DummyEth1ChainBackend<E, S> = DummyEth1ChainBackend::default();
+            dummy_backend.queued_deposits(state, eth1_data_vote, spec)
         } else {
             self.backend.queued_deposits(state, eth1_data_vote, spec)
         }
     }
+
+    /// Instantiate `Eth1Chain` from a persisted `SszEth1`.
+    ///
+    /// The `Eth1Chain` will have the same caches as the persisted `SszEth1`.
+    pub fn from_ssz_container(
+        ssz_container: &SszEth1,
+        config: Eth1Config,
+        store: Arc<S>,
+        log: &Logger,
+    ) -> Result<Self, String> {
+        let backend =
+            Eth1ChainBackend::from_bytes(&ssz_container.backend_bytes, config, store, log.clone())?;
+        Ok(Self {
+            use_dummy_backend: ssz_container.use_dummy_backend,
+            backend,
+            _phantom: PhantomData,
+        })
+    }
+
+    /// Return a `SszEth1` containing the state of `Eth1Chain`.
+    pub fn as_ssz_container(&self) -> SszEth1 {
+        SszEth1 {
+            use_dummy_backend: self.use_dummy_backend,
+            backend_bytes: self.backend.as_bytes(),
+        }
+    }
 }
 
-pub trait Eth1ChainBackend<T: EthSpec>: Sized + Send + Sync {
+pub trait Eth1ChainBackend<T: EthSpec, S: Store<T>>: Sized + Send + Sync {
     /// Returns the `Eth1Data` that should be included in a block being produced for the given
     /// `state`.
     fn eth1_data(&self, beacon_state: &BeaconState<T>, spec: &ChainSpec)
@@ -129,6 +166,17 @@ pub trait Eth1ChainBackend<T: EthSpec>: Sized + Send + Sync {
         eth1_data_vote: &Eth1Data,
         spec: &ChainSpec,
     ) -> Result<Vec<Deposit>, Error>;
+
+    /// Encode the `Eth1ChainBackend` instance to bytes.
+    fn as_bytes(&self) -> Vec<u8>;
+
+    /// Create a `Eth1ChainBackend` instance given encoded bytes.
+    fn from_bytes(
+        bytes: &[u8],
+        config: Eth1Config,
+        store: Arc<S>,
+        log: Logger,
+    ) -> Result<Self, String>;
 }
 
 /// Provides a simple, testing-only backend that generates deterministic, meaningless eth1 data.
@@ -136,9 +184,9 @@ pub trait Eth1ChainBackend<T: EthSpec>: Sized + Send + Sync {
 /// Never creates deposits, therefore the validator set is static.
 ///
 /// This was used in the 2019 Canada interop workshops.
-pub struct DummyEth1ChainBackend<T: EthSpec>(PhantomData<T>);
+pub struct DummyEth1ChainBackend<T: EthSpec, S: Store<T>>(PhantomData<(T, S)>);
 
-impl<T: EthSpec> Eth1ChainBackend<T> for DummyEth1ChainBackend<T> {
+impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T, S> for DummyEth1ChainBackend<T, S> {
     /// Produce some deterministic junk based upon the current epoch.
     fn eth1_data(&self, state: &BeaconState<T>, _spec: &ChainSpec) -> Result<Eth1Data, Error> {
         let current_epoch = state.current_epoch();
@@ -164,9 +212,24 @@ impl<T: EthSpec> Eth1ChainBackend<T> for DummyEth1ChainBackend<T> {
     ) -> Result<Vec<Deposit>, Error> {
         Ok(vec![])
     }
+
+    /// Return empty Vec<u8> for dummy backend.
+    fn as_bytes(&self) -> Vec<u8> {
+        Vec::new()
+    }
+
+    /// Create dummy eth1 backend.
+    fn from_bytes(
+        _bytes: &[u8],
+        _config: Eth1Config,
+        _store: Arc<S>,
+        _log: Logger,
+    ) -> Result<Self, String> {
+        Ok(Self(PhantomData))
+    }
 }
 
-impl<T: EthSpec> Default for DummyEth1ChainBackend<T> {
+impl<T: EthSpec, S: Store<T>> Default for DummyEth1ChainBackend<T, S> {
     fn default() -> Self {
         Self(PhantomData)
     }
@@ -214,7 +277,7 @@ impl<T: EthSpec, S: Store<T>> CachingEth1Backend<T, S> {
     }
 }
 
-impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T> for CachingEth1Backend<T, S> {
+impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T, S> for CachingEth1Backend<T, S> {
     fn eth1_data(&self, state: &BeaconState<T>, spec: &ChainSpec) -> Result<Eth1Data, Error> {
         // Note: we do not return random junk if this function call fails as it would be caused by
         // an internal error.
@@ -346,6 +409,27 @@ impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T> for CachingEth1Backend<T, S> {
                 .map(|(_deposit_root, deposits)| deposits)
         }
     }
+
+    /// Return encoded byte representation of the block and deposit caches.
+    fn as_bytes(&self) -> Vec<u8> {
+        self.core.as_bytes()
+    }
+
+    /// Recover the cached backend from encoded bytes.
+    fn from_bytes(
+        bytes: &[u8],
+        config: Eth1Config,
+        store: Arc<S>,
+        log: Logger,
+    ) -> Result<Self, String> {
+        let inner = HttpService::from_bytes(bytes, config, log.clone())?;
+        Ok(Self {
+            core: inner,
+            store,
+            log,
+            _phantom: PhantomData,
+        })
+    }
 }
 
 /// Produces an `Eth1Data` with all fields sourced from `rand::thread_rng()`.
@@ -584,7 +668,7 @@ mod test {
         use store::MemoryStore;
         use types::test_utils::{generate_deterministic_keypair, TestingDepositBuilder};
 
-        fn get_eth1_chain() -> Eth1Chain<CachingEth1Backend<E, MemoryStore<E>>, E> {
+        fn get_eth1_chain() -> Eth1Chain<CachingEth1Backend<E, MemoryStore<E>>, E, MemoryStore<E>> {
             let eth1_config = Eth1Config {
                 ..Eth1Config::default()
             };

beacon_node/beacon_chain/src/persisted_beacon_chain.rs

@@ -1,3 +1,4 @@
+use crate::eth1_chain::SszEth1;
 use crate::fork_choice::SszForkChoice;
 use crate::head_tracker::SszHeadTracker;
 use crate::{BeaconChainTypes, CheckPoint};
@@ -18,6 +19,7 @@ pub struct PersistedBeaconChain<T: BeaconChainTypes> {
     pub genesis_block_root: Hash256,
     pub ssz_head_tracker: SszHeadTracker,
     pub fork_choice: SszForkChoice,
+    pub eth1_cache: Option<SszEth1>,
     pub block_root_tree: SszBlockRootTree,
 }
 

beacon_node/beacon_chain/src/test_utils.rs

@@ -4,6 +4,7 @@ use crate::{
     events::NullEventHandler,
     AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, BlockProcessingOutcome,
 };
+use eth1::Config as Eth1Config;
 use genesis::interop_genesis_state;
 use lmd_ghost::ThreadSafeReducedTree;
 use rayon::prelude::*;
@@ -175,7 +176,7 @@ impl<E: EthSpec> BeaconChainHarness<DiskHarnessType<E>> {
             .custom_spec(spec.clone())
             .store(store.clone())
             .store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(store))
-            .resume_from_db()
+            .resume_from_db(Eth1Config::default())
             .expect("should resume beacon chain from db")
             .dummy_eth1_backend()
             .expect("should build dummy backend")