Hierarchical state diffs

Cargo.toml

@@ -240,6 +240,8 @@ unused_port = { path = "common/unused_port" }
 validator_client = { path = "validator_client" }
 validator_dir = { path = "common/validator_dir" }
 warp_utils = { path = "common/warp_utils" }
+xdelta3 = { git = "http://github.com/sigp/xdelta3-rs", rev = "2a06390cd5b61b44ca3eaa89632b4ba3410d3d7f" }
+zstd = "0.13"
 
 [profile.maxperf]
 inherits = "release"

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -769,7 +769,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)))
@@ -794,12 +793,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| {
@@ -869,7 +867,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)))
@@ -886,12 +883,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/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

@@ -10,7 +10,7 @@ use std::borrow::Cow;
 use std::iter;
 use std::time::Duration;
 use store::metadata::DataColumnInfo;
-use store::{chunked_vector::BlockRoots, AnchorInfo, BlobInfo, ChunkWriter, KeyValueStore};
+use store::{get_key_for_col, AnchorInfo, BlobInfo, DBColumn, KeyValueStore, KeyValueStoreOp};
 use types::{FixedBytesExtended, Hash256, Slot};
 
 /// Use a longer timeout on the pubkey cache.
@@ -107,8 +107,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;
 
@@ -147,8 +145,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();
@@ -160,15 +161,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();
 

beacon_node/beacon_chain/src/metrics.rs

@@ -1928,6 +1928,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(),

beacon_node/beacon_chain/src/migrate.rs

@@ -24,6 +24,10 @@ const MAX_COMPACTION_PERIOD_SECONDS: u64 = 604800;
 const MIN_COMPACTION_PERIOD_SECONDS: u64 = 7200;
 /// Compact after a large finality gap, if we respect `MIN_COMPACTION_PERIOD_SECONDS`.
 const COMPACTION_FINALITY_DISTANCE: u64 = 1024;
+/// Maximum number of blocks applied in each reconstruction burst.
+///
+/// This limits the amount of time that the finalization migration is paused for.
+const BLOCKS_PER_RECONSTRUCTION: usize = 8192 * 4;
 
 /// Default number of epochs to wait between finalization migrations.
 pub const DEFAULT_EPOCHS_PER_MIGRATION: u64 = 1;
@@ -188,7 +192,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         if let Some(Notification::Reconstruction) =
             self.send_background_notification(Notification::Reconstruction)
         {
-            Self::run_reconstruction(self.db.clone(), &self.log);
+            // If we are running in foreground mode (as in tests), then this will just run a single
+            // batch. We may need to tweak this in future.
+            Self::run_reconstruction(self.db.clone(), None, &self.log);
         }
     }
 
@@ -200,13 +206,34 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         }
     }
 
-    pub fn run_reconstruction(db: Arc<HotColdDB<E, Hot, Cold>>, log: &Logger) {
-        if let Err(e) = db.reconstruct_historic_states() {
-            error!(
-                log,
-                "State reconstruction failed";
-                "error" => ?e,
-            );
+    pub fn run_reconstruction(
+        db: Arc<HotColdDB<E, Hot, Cold>>,
+        opt_tx: Option<mpsc::Sender<Notification>>,
+        log: &Logger,
+    ) {
+        match db.reconstruct_historic_states(Some(BLOCKS_PER_RECONSTRUCTION)) {
+            Ok(()) => {
+                // Schedule another reconstruction batch if required and we have access to the
+                // channel for requeueing.
+                if let Some(tx) = opt_tx {
+                    if db.get_anchor_info().is_some() {
+                        if let Err(e) = tx.send(Notification::Reconstruction) {
+                            error!(
+                                log,
+                                "Unable to requeue reconstruction notification";
+                                "error" => ?e
+                            );
+                        }
+                    }
+                }
+            }
+            Err(e) => {
+                error!(
+                    log,
+                    "State reconstruction failed";
+                    "error" => ?e,
+                );
+            }
         }
     }
 
@@ -388,6 +415,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         log: Logger,
     ) -> (mpsc::Sender<Notification>, thread::JoinHandle<()>) {
         let (tx, rx) = mpsc::channel();
+        let inner_tx = tx.clone();
         let thread = thread::spawn(move || {
             while let Ok(notif) = rx.recv() {
                 let mut reconstruction_notif = None;
@@ -418,16 +446,17 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                         }
                     }
                 }
-                // If reconstruction is on-going, ignore finalization migration and blob pruning.
+                // Run finalization and blob pruning migrations first, then a reconstruction batch.
+                // This prevents finalization from being starved while reconstruciton runs (a
+                // problem in previous LH versions).
+                if let Some(fin) = finalization_notif {
+                    Self::run_migration(db.clone(), fin, &log);
+                }
+                if let Some(dab) = prune_blobs_notif {
+                    Self::run_prune_blobs(db.clone(), dab, &log);
+                }
                 if reconstruction_notif.is_some() {
-                    Self::run_reconstruction(db.clone(), &log);
-                } else {
-                    if let Some(fin) = finalization_notif {
-                        Self::run_migration(db.clone(), fin, &log);
-                    }
-                    if let Some(dab) = prune_blobs_notif {
-                        Self::run_prune_blobs(db.clone(), dab, &log);
-                    }
+                    Self::run_reconstruction(db.clone(), Some(inner_tx.clone()), &log);
                 }
             }
         });

beacon_node/beacon_chain/src/schema_change.rs

@@ -1,53 +1,38 @@
 //! Utilities for managing database schema changes.
 mod migration_schema_v20;
 mod migration_schema_v21;
+mod migration_schema_v22;
 
 use crate::beacon_chain::BeaconChainTypes;
-use crate::types::ChainSpec;
 use slog::Logger;
 use std::sync::Arc;
 use store::hot_cold_store::{HotColdDB, HotColdDBError};
 use store::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION};
 use store::Error as StoreError;
+use types::Hash256;
 
 /// Migrate the database from one schema version to another, applying all requisite mutations.
-#[allow(clippy::only_used_in_recursion)] // spec is not used but likely to be used in future
 pub fn migrate_schema<T: BeaconChainTypes>(
     db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
-    deposit_contract_deploy_block: u64,
+    genesis_state_root: Option<Hash256>,
     from: SchemaVersion,
     to: SchemaVersion,
     log: Logger,
-    spec: &ChainSpec,
 ) -> Result<(), StoreError> {
     match (from, to) {
         // Migrating from the current schema version to itself is always OK, a no-op.
         (_, _) if from == to && to == CURRENT_SCHEMA_VERSION => Ok(()),
         // Upgrade across multiple versions by recursively migrating one step at a time.
         (_, _) if from.as_u64() + 1 < to.as_u64() => {
             let next = SchemaVersion(from.as_u64() + 1);
-            migrate_schema::<T>(
-                db.clone(),
-                deposit_contract_deploy_block,
-                from,
-                next,
-                log.clone(),
-                spec,
-            )?;
-            migrate_schema::<T>(db, deposit_contract_deploy_block, next, to, log, spec)
+            migrate_schema::<T>(db.clone(), genesis_state_root, from, next, log.clone())?;
+            migrate_schema::<T>(db, genesis_state_root, next, to, log)
         }
         // Downgrade across multiple versions by recursively migrating one step at a time.
         (_, _) if to.as_u64() + 1 < from.as_u64() => {
             let next = SchemaVersion(from.as_u64() - 1);
-            migrate_schema::<T>(
-                db.clone(),
-                deposit_contract_deploy_block,
-                from,
-                next,
-                log.clone(),
-                spec,
-            )?;
-            migrate_schema::<T>(db, deposit_contract_deploy_block, next, to, log, spec)
+            migrate_schema::<T>(db.clone(), genesis_state_root, from, next, log.clone())?;
+            migrate_schema::<T>(db, genesis_state_root, next, to, log)
         }
 
         //
@@ -69,6 +54,9 @@ pub fn migrate_schema<T: BeaconChainTypes>(
             let ops = migration_schema_v21::downgrade_from_v21::<T>(db.clone(), log)?;
             db.store_schema_version_atomically(to, ops)
         }
+        (SchemaVersion(21), SchemaVersion(22)) => {
+            migration_schema_v22::upgrade_to_v22::<T>(db.clone(), genesis_state_root, log)
+        }
         // Anything else is an error.
         (_, _) => Err(HotColdDBError::UnsupportedSchemaVersion {
             target_version: to,