[Merged by Bors] - Deposit Cache Finalization & Fast WS Sync

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -16,6 +16,7 @@ use crate::chain_config::ChainConfig;
 use crate::early_attester_cache::EarlyAttesterCache;
 use crate::errors::{BeaconChainError as Error, BlockProductionError};
 use crate::eth1_chain::{Eth1Chain, Eth1ChainBackend};
+use crate::eth1_finalization_cache::{Eth1FinalizationCache, Eth1FinalizationData};
 use crate::events::ServerSentEventHandler;
 use crate::execution_payload::{get_execution_payload, PreparePayloadHandle};
 use crate::fork_choice_signal::{ForkChoiceSignalRx, ForkChoiceSignalTx, ForkChoiceWaitResult};
@@ -117,6 +118,9 @@ pub const ATTESTATION_CACHE_LOCK_TIMEOUT: Duration = Duration::from_secs(1);
 /// validator pubkey cache.
 pub const VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT: Duration = Duration::from_secs(1);
 
+/// The timeout for the eth1 finalization cache
+pub const ETH1_FINALIZATION_CACHE_LOCK_TIMEOUT: Duration = Duration::from_millis(200);
+
 // These keys are all zero because they get stored in different columns, see `DBColumn` type.
 pub const BEACON_CHAIN_DB_KEY: Hash256 = Hash256::zero();
 pub const OP_POOL_DB_KEY: Hash256 = Hash256::zero();
@@ -359,6 +363,8 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub(crate) snapshot_cache: TimeoutRwLock<SnapshotCache<T::EthSpec>>,
     /// Caches the attester shuffling for a given epoch and shuffling key root.
     pub shuffling_cache: TimeoutRwLock<ShufflingCache>,
+    /// A cache of eth1 deposit data at epoch boundaries for deposit finalization
+    pub eth1_finalization_cache: TimeoutRwLock<Eth1FinalizationCache>,
     /// Caches the beacon block proposer shuffling for a given epoch and shuffling key root.
     pub beacon_proposer_cache: Mutex<BeaconProposerCache>,
     /// Caches a map of `validator_index -> validator_pubkey`.
@@ -2531,9 +2537,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             block,
             block_root,
             state,
-            parent_block: _,
+            parent_block,
             confirmed_state_roots,
             payload_verification_handle,
+            parent_eth1_finalization_data,
         } = execution_pending_block;
 
         let PayloadVerificationOutcome {
@@ -2585,6 +2592,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         confirmed_state_roots,
                         payload_verification_status,
                         count_unrealized,
+                        parent_block,
+                        parent_eth1_finalization_data,
                     )
                 },
                 "payload_verification_handle",
@@ -2599,6 +2608,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ///
     /// An error is returned if the block was unable to be imported. It may be partially imported
     /// (i.e., this function is not atomic).
+    #[allow(clippy::too_many_arguments)]
     fn import_block(
         &self,
         signed_block: Arc<SignedBeaconBlock<T::EthSpec>>,
@@ -2607,6 +2617,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         confirmed_state_roots: Vec<Hash256>,
         payload_verification_status: PayloadVerificationStatus,
         count_unrealized: CountUnrealized,
+        parent_block: SignedBlindedBeaconBlock<T::EthSpec>,
+        parent_eth1_finalization_data: Eth1FinalizationData,
     ) -> Result<Hash256, BlockError<T::EthSpec>> {
         let current_slot = self.slot()?;
         let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
@@ -2987,6 +2999,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let parent_root = block.parent_root();
         let slot = block.slot();
 
+        let current_eth1_finalization_data = Eth1FinalizationData {
+            eth1_data: state.eth1_data().clone(),
+            eth1_deposit_index: state.eth1_deposit_index(),
+        };
+        let current_finalized_checkpoint = state.finalized_checkpoint();
         self.snapshot_cache
             .try_write_for(BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT)
             .ok_or(Error::SnapshotCacheLockTimeout)
@@ -3060,6 +3077,57 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
+        // Do not write to eth1 finalization cache for blocks older than 5 epochs
+        // this helps reduce noise during sync
+        if block_delay_total
+            < self.slot_clock.slot_duration() * 5 * (T::EthSpec::slots_per_epoch() as u32)
+        {
+            let parent_block_epoch = parent_block.slot().epoch(T::EthSpec::slots_per_epoch());
+            if parent_block_epoch < current_epoch {
+                // we've crossed epoch boundary, store Eth1FinalizationData
+                let (checkpoint, eth1_finalization_data) =
+                    if current_slot % T::EthSpec::slots_per_epoch() == 0 {
+                        // current block is the checkpoint
+                        (
+                            Checkpoint {
+                                epoch: current_epoch,
+                                root: block_root,
+                            },
+                            current_eth1_finalization_data,
+                        )
+                    } else {
+                        // parent block is the checkpoint
+                        (
+                            Checkpoint {
+                                epoch: current_epoch,
+                                root: parent_block.canonical_root(),
+                            },
+                            parent_eth1_finalization_data,
+                        )
+                    };
+
+                if let Some(finalized_eth1_data) = self
+                    .eth1_finalization_cache
+                    .try_write_for(ETH1_FINALIZATION_CACHE_LOCK_TIMEOUT)
+                    .and_then(|mut cache| {
+                        cache.insert(checkpoint, eth1_finalization_data);
+                        cache.finalize(&current_finalized_checkpoint)
+                    })
+                {
+                    if let Some(eth1_chain) = self.eth1_chain.as_ref() {
+                        let finalized_deposit_count = finalized_eth1_data.deposit_count;
+                        eth1_chain.finalize_eth1_data(finalized_eth1_data);
+                        debug!(
+                            self.log,
+                            "called eth1_chain.finalize_eth1_data()";
+                            "epoch" => current_finalized_checkpoint.epoch,
+                            "deposit count" => finalized_deposit_count,
+                        );
+                    }
+                }
+            }
+        }
+
         // Inform the unknown block cache, in case it was waiting on this block.
         self.pre_finalization_block_cache
             .block_processed(block_root);

beacon_node/beacon_chain/src/block_verification.rs

@@ -42,6 +42,7 @@
 //!            END
 //!
 //! ```
+use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::execution_payload::{
     is_optimistic_candidate_block, validate_execution_payload_for_gossip, validate_merge_block,
     AllowOptimisticImport, PayloadNotifier,
@@ -622,6 +623,7 @@ pub struct ExecutionPendingBlock<T: BeaconChainTypes> {
     pub block_root: Hash256,
     pub state: BeaconState<T::EthSpec>,
     pub parent_block: SignedBeaconBlock<T::EthSpec, BlindedPayload<T::EthSpec>>,
+    pub parent_eth1_finalization_data: Eth1FinalizationData,
     pub confirmed_state_roots: Vec<Hash256>,
     pub payload_verification_handle: PayloadVerificationHandle<T::EthSpec>,
 }
@@ -1164,6 +1166,11 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
             .into());
         }
 
+        let parent_eth1_finalization_data = Eth1FinalizationData {
+            eth1_data: state.eth1_data().clone(),
+            eth1_deposit_index: state.eth1_deposit_index(),
+        };
+
         let distance = block.slot().as_u64().saturating_sub(state.slot().as_u64());
         for _ in 0..distance {
             let state_root = if parent.beacon_block.slot() == state.slot() {
@@ -1419,6 +1426,7 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
             block_root,
             state,
             parent_block: parent.beacon_block,
+            parent_eth1_finalization_data,
             confirmed_state_roots,
             payload_verification_handle,
         })

beacon_node/beacon_chain/src/builder.rs

@@ -1,5 +1,6 @@
 use crate::beacon_chain::{CanonicalHead, BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, OP_POOL_DB_KEY};
 use crate::eth1_chain::{CachingEth1Backend, SszEth1};
+use crate::eth1_finalization_cache::Eth1FinalizationCache;
 use crate::fork_choice_signal::ForkChoiceSignalTx;
 use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
 use crate::head_tracker::HeadTracker;
@@ -795,6 +796,7 @@ where
                 head_for_snapshot_cache,
             )),
             shuffling_cache: TimeoutRwLock::new(ShufflingCache::new()),
+            eth1_finalization_cache: TimeoutRwLock::new(Eth1FinalizationCache::new(log.clone())),
             beacon_proposer_cache: <_>::default(),
             block_times_cache: <_>::default(),
             pre_finalization_block_cache: <_>::default(),

beacon_node/beacon_chain/src/eth1_chain.rs

@@ -16,7 +16,6 @@ use store::{DBColumn, Error as StoreError, StoreItem};
 use task_executor::TaskExecutor;
 use types::{
     BeaconState, BeaconStateError, ChainSpec, Deposit, Eth1Data, EthSpec, Hash256, Slot, Unsigned,
-    DEPOSIT_TREE_DEPTH,
 };
 
 type BlockNumber = u64;
@@ -170,8 +169,8 @@ fn get_sync_status<T: EthSpec>(
 
 #[derive(Encode, Decode, Clone)]
 pub struct SszEth1 {
-    use_dummy_backend: bool,
-    backend_bytes: Vec<u8>,
+    pub use_dummy_backend: bool,
+    pub backend_bytes: Vec<u8>,
 }
 
 impl StoreItem for SszEth1 {
@@ -305,6 +304,12 @@ where
         }
     }
 
+    /// Set in motion the finalization of `Eth1Data`. This method is called during block import
+    /// so it should be fast.
+    pub fn finalize_eth1_data(&self, eth1_data: Eth1Data) {
+        self.backend.finalize_eth1_data(eth1_data);
+    }
+
     /// Consumes `self`, returning the backend.
     pub fn into_backend(self) -> T {
         self.backend
@@ -335,6 +340,10 @@ pub trait Eth1ChainBackend<T: EthSpec>: Sized + Send + Sync {
     /// beacon node eth1 cache is.
     fn latest_cached_block(&self) -> Option<Eth1Block>;
 
+    /// Set in motion the finalization of `Eth1Data`. This method is called during block import
+    /// so it should be fast.
+    fn finalize_eth1_data(&self, eth1_data: Eth1Data);
+
     /// Returns the block at the head of the chain (ignoring follow distance, etc). Used to obtain
     /// an idea of how up-to-date the remote eth1 node is.
     fn head_block(&self) -> Option<Eth1Block>;
@@ -389,6 +398,8 @@ impl<T: EthSpec> Eth1ChainBackend<T> for DummyEth1ChainBackend<T> {
         None
     }
 
+    fn finalize_eth1_data(&self, _eth1_data: Eth1Data) {}
+
     fn head_block(&self) -> Option<Eth1Block> {
         None
     }
@@ -547,7 +558,7 @@ impl<T: EthSpec> Eth1ChainBackend<T> for CachingEth1Backend<T> {
                     .deposits()
                     .read()
                     .cache
-                    .get_deposits(next, last, deposit_count, DEPOSIT_TREE_DEPTH)
+                    .get_deposits(next, last, deposit_count)
                     .map_err(|e| Error::BackendError(format!("Failed to get deposits: {:?}", e)))
                     .map(|(_deposit_root, deposits)| deposits)
             }
@@ -558,6 +569,12 @@ impl<T: EthSpec> Eth1ChainBackend<T> for CachingEth1Backend<T> {
         self.core.latest_cached_block()
     }
 
+    /// This only writes the eth1_data to a temporary cache so that the service
+    /// thread can later do the actual finalizing of the deposit tree.
+    fn finalize_eth1_data(&self, eth1_data: Eth1Data) {
+        self.core.set_to_finalize(Some(eth1_data));
+    }
+
     fn head_block(&self) -> Option<Eth1Block> {
         self.core.head_block()
     }