Send it boy

Author: dapplion

account_manager/src/validator/slashing_protection.rs

@@ -90,7 +90,7 @@ pub fn cli_run<E: EthSpec>(
             let slashing_protection_database =
                 SlashingDatabase::open_or_create(&slashing_protection_db_path).map_err(|e| {
                     format!(
-                        "Unable to open database at {}: {:?}",
+                        "Unable to open slashing protection database at {}: {:?}",
                         slashing_protection_db_path.display(),
                         e
                     )
@@ -198,7 +198,7 @@ pub fn cli_run<E: EthSpec>(
             let slashing_protection_database = SlashingDatabase::open(&slashing_protection_db_path)
                 .map_err(|e| {
                     format!(
-                        "Unable to open database at {}: {:?}",
+                        "Unable to open slashing protection database at {}: {:?}",
                         slashing_protection_db_path.display(),
                         e
                     )

beacon_node/beacon_chain/src/block_verification.rs

@@ -91,7 +91,7 @@ use std::fmt::Debug;
 use std::fs;
 use std::io::Write;
 use std::sync::Arc;
-use store::{Error as DBError, HotStateSummary, KeyValueStore, StoreOp};
+use store::{Error as DBError, KeyValueStore};
 use strum::AsRefStr;
 use task_executor::JoinHandle;
 use types::{
@@ -1483,28 +1483,19 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
                 // processing, but we get early access to it.
                 let state_root = state.update_tree_hash_cache()?;
 
-                // Store the state immediately.
-                let txn_lock = chain.store.hot_db.begin_rw_transaction();
+                // Store the state immediately. States are ONLY deleted on finalization pruning, so
+                // we won't have race conditions where we should have writen a state and didn't.
                 let state_already_exists =
                     chain.store.load_hot_state_summary(&state_root)?.is_some();
 
-                let state_batch = if state_already_exists {
+                if state_already_exists {
                     // If the state exists, we do not need to re-write it.
-                    vec![]
                 } else {
-                    vec![if state.slot() % T::EthSpec::slots_per_epoch() == 0 {
-                        StoreOp::PutState(state_root, &state)
-                    } else {
-                        StoreOp::PutStateSummary(
-                            state_root,
-                            HotStateSummary::new(&state_root, &state)?,
-                        )
-                    }]
+                    // Recycle store codepath to create a state summary and store the state / diff
+                    let mut ops = vec![];
+                    chain.store.store_hot_state(&state_root, &state, &mut ops)?;
+                    chain.store.hot_db.do_atomically(ops)?;
                 };
-                chain
-                    .store
-                    .do_atomically_with_block_and_blobs_cache(state_batch)?;
-                drop(txn_lock);
 
                 state_root
             };

beacon_node/beacon_chain/src/builder.rs

@@ -39,8 +39,8 @@ use std::time::Duration;
 use store::{Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
 use task_executor::{ShutdownReason, TaskExecutor};
 use types::{
-    BeaconBlock, BeaconState, BlobSidecarList, ChainSpec, Checkpoint, Epoch, EthSpec,
-    FixedBytesExtended, Hash256, Signature, SignedBeaconBlock, Slot,
+    BeaconBlock, BeaconState, BlobSidecarList, ChainSpec, Epoch, EthSpec, FixedBytesExtended,
+    Hash256, Signature, SignedBeaconBlock, Slot,
 };
 
 /// An empty struct used to "witness" all the `BeaconChainTypes` traits. It has no user-facing
@@ -387,21 +387,28 @@ where
     }
 
     /// Starts a new chain from a genesis state.
-    pub fn genesis_state(mut self, beacon_state: BeaconState<E>) -> Result<Self, String> {
+    pub fn genesis_state(mut self, mut beacon_state: BeaconState<E>) -> Result<Self, String> {
         let store = self.store.clone().ok_or("genesis_state requires a store")?;
 
+        // Initialize anchor info before attempting to write the genesis state
+        let retain_historic_states = self.chain_config.reconstruct_historic_states;
+        let genesis_beacon_block = genesis_block(&mut beacon_state, &self.spec)?;
+        self.pending_io_batch.push(
+            store
+                .init_anchor_info(
+                    genesis_beacon_block.message().parent_root(),
+                    Slot::new(0),
+                    retain_historic_states,
+                )
+                .map_err(|e| format!("Failed to initialize genesis anchor: {:?}", e))?,
+        );
+
         let (genesis, updated_builder) = self.set_genesis_state(beacon_state)?;
         self = updated_builder;
 
         // Stage the database's metadata fields for atomic storage when `build` is called.
         // Since v4.4.0 we will set the anchor with a dummy state upper limit in order to prevent
         // historic states from being retained (unless `--reconstruct-historic-states` is set).
-        let retain_historic_states = self.chain_config.reconstruct_historic_states;
-        self.pending_io_batch.push(
-            store
-                .init_anchor_info(genesis.beacon_block.message(), retain_historic_states)
-                .map_err(|e| format!("Failed to initialize genesis anchor: {:?}", e))?,
-        );
         self.pending_io_batch.push(
             store
                 .init_blob_info(genesis.beacon_block.slot())
@@ -519,6 +526,14 @@ where
             }
         }
 
+        debug!(
+            log,
+            "Storing split from weak subjectivity state";
+            "slot" => weak_subj_slot,
+            "state_root" => ?weak_subj_state_root,
+            "block_root" => ?weak_subj_block_root,
+        );
+
         // Set the store's split point *before* storing genesis so that genesis is stored
         // immediately in the freezer DB.
         store.set_split(weak_subj_slot, weak_subj_state_root, weak_subj_block_root);
@@ -540,6 +555,19 @@ where
             .do_atomically(block_root_batch)
             .map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
 
+        // Write the anchor to memory before calling `put_state` otherwise hot hdiff can't store
+        // states that do not align with the start_slot grid
+        let retain_historic_states = self.chain_config.reconstruct_historic_states;
+        self.pending_io_batch.push(
+            store
+                .init_anchor_info(
+                    weak_subj_block.message().parent_root(),
+                    weak_subj_slot,
+                    retain_historic_states,
+                )
+                .map_err(|e| format!("Failed to initialize anchor info: {:?}", e))?,
+        );
+
         // Write the state, block and blobs non-atomically, it doesn't matter if they're forgotten
         // about on a crash restart.
         store
@@ -549,6 +577,8 @@ where
                 weak_subj_state.clone(),
             )
             .map_err(|e| format!("Failed to set checkpoint state as finalized state: {:?}", e))?;
+        // Note: post hot hdiff must update the anchor info before attempting to put_state otherwise
+        // the write will fail if the weak_subj_slot is not aligned with the snapshot moduli.
         store
             .put_state(&weak_subj_state_root, &weak_subj_state)
             .map_err(|e| format!("Failed to store weak subjectivity state: {e:?}"))?;
@@ -585,13 +615,7 @@ where
         // Stage the database's metadata fields for atomic storage when `build` is called.
         // This prevents the database from restarting in an inconsistent state if the anchor
         // info or split point is written before the `PersistedBeaconChain`.
-        let retain_historic_states = self.chain_config.reconstruct_historic_states;
         self.pending_io_batch.push(store.store_split_in_batch());
-        self.pending_io_batch.push(
-            store
-                .init_anchor_info(weak_subj_block.message(), retain_historic_states)
-                .map_err(|e| format!("Failed to initialize anchor info: {:?}", e))?,
-        );
         self.pending_io_batch.push(
             store
                 .init_blob_info(weak_subj_block.slot())
@@ -603,13 +627,6 @@ where
                 .map_err(|e| format!("Failed to initialize data column info: {:?}", e))?,
         );
 
-        // Store pruning checkpoint to prevent attempting to prune before the anchor state.
-        self.pending_io_batch
-            .push(store.pruning_checkpoint_store_op(Checkpoint {
-                root: weak_subj_block_root,
-                epoch: weak_subj_state.slot().epoch(E::slots_per_epoch()),
-            }));
-
         let snapshot = BeaconSnapshot {
             beacon_block_root: weak_subj_block_root,
             beacon_block: Arc::new(weak_subj_block),

beacon_node/beacon_chain/src/migrate.rs

@@ -586,11 +586,23 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         let newly_finalized_blocks = state_summaries_dag
             .blocks_of_states(newly_finalized_state_roots.iter())
             .map_err(|e| PruningError::SummariesDagError("blocks of newly finalized", e))?;
+        let newly_finalized_blocks_min_slot = *newly_finalized_blocks
+            .iter()
+            .map(|(_, slot)| slot)
+            .min()
+            .ok_or(PruningError::EmptyFinalizedBlocks)?;
+
+        // Compute the set of finalized state roots that we must keep to make the dynamic HDiff system
+        // work.
+        let required_finalized_diff_state_slots = store
+            .hierarchy_hot
+            .closest_layer_points(new_finalized_slot, store.hot_hdiff_start_slot()?);
 
         // We don't know which blocks are shared among abandoned chains, so we buffer and delete
         // everything in one fell swoop.
         let mut blocks_to_prune: HashSet<Hash256> = HashSet::new();
         let mut states_to_prune: HashSet<(Slot, Hash256)> = HashSet::new();
+        let mut kept_summaries_for_hdiff = vec![];
 
         // Consider the following block tree where we finalize block `[0]` at the checkpoint `(f)`.
         // There's a block `[3]` that descendends from the finalized block but NOT from the
@@ -603,14 +615,38 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         //  \---[3]--|-----------------[4]
         //           |
 
-        for (_, summaries) in state_summaries_dag.summaries_by_slot_ascending() {
-            for (state_root, summary) in summaries {
+        for (slot, summaries) in state_summaries_dag.summaries_by_slot_ascending() {
+            for (state_root, _) in summaries {
                 let should_prune = if finalized_and_descendant_state_roots_of_finalized_checkpoint
                     .contains(&state_root)
                 {
                     // This state is a viable descendant of the finalized checkpoint, so does not
                     // conflict with finality and can be built on or become a head
                     false
+                } else if required_finalized_diff_state_slots.contains(&slot) {
+                    // Keep this state and diff as it's necessary for the finalized portion of the
+                    // HDiff links. `required_finalized_diff_state_slots` tracks the set of slots on
+                    // each diff layer, and by checking `newly_finalized_state_roots` which only
+                    // keep those on the finalized canonical chain. Checking the state root ensures
+                    // we avoid lingering forks.
+
+                    // In the diagram below, `o` are diffs by slot that we must keep. In the prior
+                    // finalized section there's only one chain so we preserve them unconditionally.
+                    // For the newly finalized chain, we check which of is canonical and only keep
+                    // those. Slots below `min_finalized_state_slot` we don't have canonical
+                    // information so we assume they are part of the finalized pruned chain.
+                    //
+                    //                  /-----o----
+                    // o-------o------/-------o----
+                    if slot < newly_finalized_states_min_slot
+                        || newly_finalized_state_roots.contains(&state_root)
+                    {
+                        // Track kept summaries to debug hdiff inconsistencies with "Extra pruning information"
+                        kept_summaries_for_hdiff.push((state_root, slot));
+                        false
+                    } else {
+                        true
+                    }
                 } else {
                     // Everything else, prune
                     true
@@ -619,7 +655,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                 if should_prune {
                     // States are migrated into the cold DB in the migrate step. All hot states
                     // prior to finalized can be pruned from the hot DB columns
-                    states_to_prune.insert((summary.slot, state_root));
+                    states_to_prune.insert((slot, state_root));
                 }
             }
         }
@@ -639,7 +675,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             } else if newly_finalized_blocks.contains(&(block_root, slot)) {
                 // Keep recently finalized blocks
                 false
-            } else if slot < newly_finalized_states_min_slot {
+            } else if slot < newly_finalized_blocks_min_slot {
                 // Keep recently finalized blocks that we know are canonical. Blocks with slots <
                 // that `newly_finalized_blocks_min_slot` we don't have canonical information so we
                 // assume they are part of the finalized pruned chain
@@ -668,7 +704,10 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             "new_finalized_checkpoint" => ?new_finalized_checkpoint,
             "newly_finalized_blocks" => newly_finalized_blocks.len(),
             "newly_finalized_state_roots" => newly_finalized_state_roots.len(),
+            "newly_finalized_blocks_min_slot" => newly_finalized_blocks_min_slot,
             "newly_finalized_states_min_slot" => newly_finalized_states_min_slot,
+            "required_finalized_diff_state_slots" => ?required_finalized_diff_state_slots,
+            "kept_summaries_for_hdiff" => ?kept_summaries_for_hdiff,
             "state_summaries_count" => state_summaries_dag.summaries_count(),
             "state_summaries_dag_roots" => ?state_summaries_dag_roots,
             "finalized_and_descendant_state_roots_of_finalized_checkpoint" => finalized_and_descendant_state_roots_of_finalized_checkpoint.len(),

beacon_node/beacon_chain/src/schema_change.rs

@@ -3,6 +3,7 @@ mod migration_schema_v20;
 mod migration_schema_v21;
 mod migration_schema_v22;
 mod migration_schema_v23;
+mod migration_schema_v24;
 
 use crate::beacon_chain::BeaconChainTypes;
 use slog::Logger;
@@ -68,6 +69,14 @@ pub fn migrate_schema<T: BeaconChainTypes>(
             let ops = migration_schema_v23::downgrade_from_v23::<T>(db.clone(), log)?;
             db.store_schema_version_atomically(to, ops)
         }
+        (SchemaVersion(23), SchemaVersion(24)) => {
+            let ops = migration_schema_v24::upgrade_to_v24::<T>(db.clone(), log)?;
+            db.store_schema_version_atomically(to, ops)
+        }
+        (SchemaVersion(24), SchemaVersion(23)) => {
+            let ops = migration_schema_v24::downgrade_from_v24::<T>(db.clone(), log)?;
+            db.store_schema_version_atomically(to, ops)
+        }
         // Anything else is an error.
         (_, _) => Err(HotColdDBError::UnsupportedSchemaVersion {
             target_version: to,