Hierarchical state diffs

.github/workflows/test-suite.yml

@@ -63,8 +63,8 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v3
       - name: Install dependencies
-        run: apt update && apt install -y cmake
-      - name: Generate code coverage
+        run: apt update && apt install -y cmake libclang-dev
+      - name: Check for deadlocks
         run: |
           cargo lockbud -k deadlock -b -l tokio_util
 

Cargo.toml

@@ -263,6 +263,8 @@ validator_http_metrics = { path = "validator_client/http_metrics" }
 validator_metrics = { path = "validator_client/validator_metrics" }
 validator_store= { path = "validator_client/validator_store" }
 warp_utils = { path = "common/warp_utils" }
+xdelta3 = { git = "http://github.com/sigp/xdelta3-rs", rev = "50d63cdf1878e5cf3538e9aae5eed34a22c64e4a" }
+zstd = "0.13"
 
 [profile.maxperf]
 inherits = "release"

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -767,7 +767,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             start_slot,
             local_head.beacon_state.clone(),
             local_head.beacon_block_root,
-            &self.spec,
         )?;
 
         Ok(iter.map(|result| result.map_err(Into::into)))
@@ -790,12 +789,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         self.with_head(move |head| {
-            let iter = self.store.forwards_block_roots_iterator_until(
-                start_slot,
-                end_slot,
-                || Ok((head.beacon_state.clone(), head.beacon_block_root)),
-                &self.spec,
-            )?;
+            let iter =
+                self.store
+                    .forwards_block_roots_iterator_until(start_slot, end_slot, || {
+                        Ok((head.beacon_state.clone(), head.beacon_block_root))
+                    })?;
             Ok(iter
                 .map(|result| result.map_err(Into::into))
                 .take_while(move |result| {
@@ -865,7 +863,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             start_slot,
             local_head.beacon_state_root(),
             local_head.beacon_state.clone(),
-            &self.spec,
         )?;
 
         Ok(iter.map(|result| result.map_err(Into::into)))
@@ -882,12 +879,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         end_slot: Slot,
     ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>> + '_, Error> {
         self.with_head(move |head| {
-            let iter = self.store.forwards_state_roots_iterator_until(
-                start_slot,
-                end_slot,
-                || Ok((head.beacon_state.clone(), head.beacon_state_root())),
-                &self.spec,
-            )?;
+            let iter =
+                self.store
+                    .forwards_state_roots_iterator_until(start_slot, end_slot, || {
+                        Ok((head.beacon_state.clone(), head.beacon_state_root()))
+                    })?;
             Ok(iter
                 .map(|result| result.map_err(Into::into))
                 .take_while(move |result| {

beacon_node/beacon_chain/src/block_verification.rs

@@ -839,9 +839,6 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
 
         let block_root = get_block_header_root(block_header);
 
-        // Disallow blocks that conflict with the anchor (weak subjectivity checkpoint), if any.
-        check_block_against_anchor_slot(block.message(), chain)?;
-
         // Do not gossip a block from a finalized slot.
         check_block_against_finalized_slot(block.message(), block_root, chain)?;
 
@@ -1074,9 +1071,6 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
             .fork_name(&chain.spec)
             .map_err(BlockError::InconsistentFork)?;
 
-        // Check the anchor slot before loading the parent, to avoid spurious lookups.
-        check_block_against_anchor_slot(block.message(), chain)?;
-
         let (mut parent, block) = load_parent(block, chain)?;
 
         let state = cheap_state_advance_to_obtain_committees::<_, BlockError>(
@@ -1688,19 +1682,6 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
     }
 }
 
-/// Returns `Ok(())` if the block's slot is greater than the anchor block's slot (if any).
-fn check_block_against_anchor_slot<T: BeaconChainTypes>(
-    block: BeaconBlockRef<'_, T::EthSpec>,
-    chain: &BeaconChain<T>,
-) -> Result<(), BlockError> {
-    if let Some(anchor_slot) = chain.store.get_anchor_slot() {
-        if block.slot() <= anchor_slot {
-            return Err(BlockError::WeakSubjectivityConflict);
-        }
-    }
-    Ok(())
-}
-
 /// Returns `Ok(())` if the block is later than the finalized slot on `chain`.
 ///
 /// Returns an error if the block is earlier or equal to the finalized slot, or there was an error

beacon_node/beacon_chain/src/builder.rs

@@ -363,6 +363,10 @@ where
         store
             .put_block(&beacon_block_root, beacon_block.clone())
             .map_err(|e| format!("Failed to store genesis block: {:?}", e))?;
+        store
+            .store_frozen_block_root_at_skip_slots(Slot::new(0), Slot::new(1), beacon_block_root)
+            .and_then(|ops| store.cold_db.do_atomically(ops))
+            .map_err(|e| format!("Failed to store genesis block root: {e:?}"))?;
 
         // Store the genesis block under the `ZERO_HASH` key.
         store

beacon_node/beacon_chain/src/historical_blocks.rs

@@ -11,8 +11,8 @@ use std::iter;
 use std::time::Duration;
 use store::metadata::DataColumnInfo;
 use store::{
-    chunked_vector::BlockRoots, AnchorInfo, BlobInfo, ChunkWriter, Error as StoreError,
-    KeyValueStore,
+    get_key_for_col, AnchorInfo, BlobInfo, DBColumn, Error as StoreError, KeyValueStore,
+    KeyValueStoreOp,
 };
 use strum::IntoStaticStr;
 use types::{FixedBytesExtended, Hash256, Slot};
@@ -35,8 +35,6 @@ pub enum HistoricalBlockError {
     InvalidSignature,
     /// Transitory error, caller should retry with the same blocks.
     ValidatorPubkeyCacheTimeout,
-    /// No historical sync needed.
-    NoAnchorInfo,
     /// Logic error: should never occur.
     IndexOutOfBounds,
     /// Internal store error
@@ -72,10 +70,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         mut blocks: Vec<AvailableBlock<T::EthSpec>>,
     ) -> Result<usize, HistoricalBlockError> {
-        let anchor_info = self
-            .store
-            .get_anchor_info()
-            .ok_or(HistoricalBlockError::NoAnchorInfo)?;
+        let anchor_info = self.store.get_anchor_info();
         let blob_info = self.store.get_blob_info();
         let data_column_info = self.store.get_data_column_info();
 
@@ -119,8 +114,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         let mut expected_block_root = anchor_info.oldest_block_parent;
         let mut prev_block_slot = anchor_info.oldest_block_slot;
-        let mut chunk_writer =
-            ChunkWriter::<BlockRoots, _, _>::new(&self.store.cold_db, prev_block_slot.as_usize())?;
         let mut new_oldest_blob_slot = blob_info.oldest_blob_slot;
         let mut new_oldest_data_column_slot = data_column_info.oldest_data_column_slot;
 
@@ -158,8 +151,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
 
             // Store block roots, including at all skip slots in the freezer DB.
-            for slot in (block.slot().as_usize()..prev_block_slot.as_usize()).rev() {
-                chunk_writer.set(slot, block_root, &mut cold_batch)?;
+            for slot in (block.slot().as_u64()..prev_block_slot.as_u64()).rev() {
+                cold_batch.push(KeyValueStoreOp::PutKeyValue(
+                    get_key_for_col(DBColumn::BeaconBlockRoots.into(), &slot.to_be_bytes()),
+                    block_root.as_slice().to_vec(),
+                ));
             }
 
             prev_block_slot = block.slot();
@@ -171,15 +167,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             // completion.
             if expected_block_root == self.genesis_block_root {
                 let genesis_slot = self.spec.genesis_slot;
-                for slot in genesis_slot.as_usize()..prev_block_slot.as_usize() {
-                    chunk_writer.set(slot, self.genesis_block_root, &mut cold_batch)?;
+                for slot in genesis_slot.as_u64()..prev_block_slot.as_u64() {
+                    cold_batch.push(KeyValueStoreOp::PutKeyValue(
+                        get_key_for_col(DBColumn::BeaconBlockRoots.into(), &slot.to_be_bytes()),
+                        self.genesis_block_root.as_slice().to_vec(),
+                    ));
                 }
                 prev_block_slot = genesis_slot;
                 expected_block_root = Hash256::zero();
                 break;
             }
         }
-        chunk_writer.write(&mut cold_batch)?;
         // these were pushed in reverse order so we reverse again
         signed_blocks.reverse();
 
@@ -271,7 +269,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let backfill_complete = new_anchor.block_backfill_complete(self.genesis_backfill_slot);
         anchor_and_blob_batch.push(
             self.store
-                .compare_and_set_anchor_info(Some(anchor_info), Some(new_anchor))?,
+                .compare_and_set_anchor_info(anchor_info, new_anchor)?,
         );
         self.store.hot_db.do_atomically(anchor_and_blob_batch)?;
 

beacon_node/beacon_chain/src/metrics.rs

@@ -2004,6 +2004,7 @@ pub fn scrape_for_metrics<T: BeaconChainTypes>(beacon_chain: &BeaconChain<T>) {
     let attestation_stats = beacon_chain.op_pool.attestation_stats();
     let chain_metrics = beacon_chain.metrics();
 
+    // Kept duplicated for backwards compatibility
     set_gauge_by_usize(
         &BLOCK_PROCESSING_SNAPSHOT_CACHE_SIZE,
         beacon_chain.store.state_cache_len(),
@@ -2067,6 +2068,8 @@ pub fn scrape_for_metrics<T: BeaconChainTypes>(beacon_chain: &BeaconChain<T>) {
         .canonical_head
         .fork_choice_read_lock()
         .scrape_for_metrics();
+
+    beacon_chain.store.register_metrics();
 }
 
 /// Scrape the given `state` assuming it's the head state, updating the `DEFAULT_REGISTRY`.