Deprecate pruning checkpoint

dapplion · dapplion · commit c8d82f08b5c2 · 2024-12-25T01:25:27.000+08:00
diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs
@@ -38,8 +38,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
@@ -602,13 +602,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),
diff --git a/beacon_node/beacon_chain/src/migrate.rs b/beacon_node/beacon_chain/src/migrate.rs
@@ -15,9 +15,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use store::hot_cold_store::{migrate_database, HotColdDBError};
 use store::{DBColumn, Error, HotStateSummary, ItemStore, StoreItem, StoreOp};
 pub use store::{HotColdDB, MemoryStore};
-use types::{
-    BeaconState, BeaconStateHash, Checkpoint, Epoch, EthSpec, FixedBytesExtended, Hash256, Slot,
-};
+use types::{BeaconState, BeaconStateHash, Checkpoint, Epoch, EthSpec, Hash256, Slot};
 
 /// Compact at least this frequently, finalization permitting (7 days).
 const MAX_COMPACTION_PERIOD_SECONDS: u64 = 604800;
@@ -89,7 +87,7 @@ pub struct PrevMigration {
 pub enum PruningOutcome {
     /// The pruning succeeded and updated the pruning checkpoint from `old_finalized_checkpoint`.
     Successful {
-        old_finalized_checkpoint: Checkpoint,
+        old_finalized_checkpoint_epoch: Epoch,
     },
     /// The run was aborted because the new finalized checkpoint is older than the previous one.
     OutOfOrderFinalization {
@@ -119,6 +117,8 @@ pub enum PruningError {
     MissingSummaryForFinalizedCheckpoint(Hash256),
     MissingBlindedBlock(Hash256),
     SummariesDagError(summaries_dag::Error),
+    EmptyFinalizedStates,
+    EmptyFinalizedBlocks,
 }
 
 /// Message sent to the migration thread containing the information it needs to run.
@@ -356,7 +356,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             }
         };
 
-        let old_finalized_checkpoint = match Self::prune_hot_db(
+        let old_finalized_checkpoint_epoch = match Self::prune_hot_db(
             db.clone(),
             finalized_state_root.into(),
             &finalized_state,
@@ -365,8 +365,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             log,
         ) {
             Ok(PruningOutcome::Successful {
-                old_finalized_checkpoint,
-            }) => old_finalized_checkpoint,
+                old_finalized_checkpoint_epoch,
+            }) => old_finalized_checkpoint_epoch,
             Ok(PruningOutcome::DeferredConcurrentHeadTrackerMutation) => {
                 warn!(
                     log,
@@ -397,7 +397,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         // Finally, compact the database so that new free space is properly reclaimed.
         if let Err(e) = Self::run_compaction(
             db,
-            old_finalized_checkpoint.epoch,
+            old_finalized_checkpoint_epoch,
             notif.finalized_checkpoint.epoch,
             log,
         ) {
@@ -475,17 +475,6 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         genesis_block_root: Hash256,
         log: &Logger,
     ) -> Result<PruningOutcome, BeaconChainError> {
-        let old_finalized_checkpoint =
-            store
-                .load_pruning_checkpoint()?
-                .unwrap_or_else(|| Checkpoint {
-                    epoch: Epoch::new(0),
-                    root: Hash256::zero(),
-                });
-
-        let old_finalized_slot = old_finalized_checkpoint
-            .epoch
-            .start_slot(E::slots_per_epoch());
         let new_finalized_slot = new_finalized_checkpoint
             .epoch
             .start_slot(E::slots_per_epoch());
@@ -500,21 +489,16 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             .into());
         }
 
-        // The new finalized state must be newer than the previous finalized state.
-        // I think this can happen sometimes currently due to `fork_choice` running in parallel
-        // with itself and sending us notifications out of order.
-        if old_finalized_slot > new_finalized_slot {
-            return Ok(PruningOutcome::OutOfOrderFinalization {
-                old_finalized_checkpoint,
-                new_finalized_checkpoint,
-            });
-        }
+        // TODO(hdiff): if we remove the check of `old_finalized_slot > new_finalized_slot` can we
+        // ensure that a single pruning operation is running at once? If a pruning run is triggered
+        // with an old finalized checkpoint it can derive a stale hdiff set of slots and delete
+        // future ones that are necessary breaking the DB.
 
         debug!(
             log,
             "Starting database pruning";
-            "old_finalized_checkpoint" => ?old_finalized_checkpoint,
             "new_finalized_checkpoint" => ?new_finalized_checkpoint,
+            "new_finalized_state_hash" => ?new_finalized_state_hash,
         );
 
         let (state_summaries_dag, block_summaries_dag) = {
@@ -580,9 +564,18 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         );
 
         // Note: ancestors_of includes the finalized state root
-        let newly_finalized_state_roots = state_summaries_dag
+        let newly_finalized_state_summaries = state_summaries_dag
             .ancestors_of(new_finalized_state_hash)
             .map_err(PruningError::SummariesDagError)?;
+        let newly_finalized_state_roots = newly_finalized_state_summaries
+            .iter()
+            .map(|(root, _)| *root)
+            .collect::<HashSet<Hash256>>();
+        let newly_finalized_states_min_slot = *newly_finalized_state_summaries
+            .iter()
+            .map(|(_, slot)| slot)
+            .min()
+            .ok_or(PruningError::EmptyFinalizedStates)?;
 
         // Note: ancestors_of includes the finalized block
         let newly_finalized_blocks = block_summaries_dag
@@ -592,6 +585,11 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             .iter()
             .map(|(root, _)| *root)
             .collect::<HashSet<Hash256>>();
+        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.
@@ -622,11 +620,12 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                         // 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.
+                        // 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 < old_finalized_slot
+                        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"
@@ -660,9 +659,13 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             } else if newly_finalized_block_roots.contains(&block_root) {
                 // Keep recently finalized blocks
                 false
-            } else if slot <= old_finalized_slot {
-                // Keep blocks that are prior to the last finalized slot.
-                // newly_finalized_block_roots only contains blocks over `old_finalized_slot`.
+            } else if slot < newly_finalized_blocks_min_slot
+                || newly_finalized_block_roots.contains(&block_root)
+            {
+                // 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
+                //
                 // Pruning those risks breaking the DB by deleting canonical blocks once the HDiff
                 // grid advances. If the pruning routine is correct this condition should never hit.
                 false
@@ -679,10 +682,11 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         debug!(
             log,
             "Extra pruning information";
-            "old_finalized_checkpoint" => ?old_finalized_checkpoint,
             "new_finalized_checkpoint" => ?new_finalized_checkpoint,
             "newly_finalized_blocks" => newly_finalized_blocks.len(),
+            "newly_finalized_blocks_min_slot" => newly_finalized_blocks_min_slot,
             "newly_finalized_state_roots" => newly_finalized_state_roots.len(),
+            "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(),
@@ -719,11 +723,6 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             persisted_head.as_kv_store_op(BEACON_CHAIN_DB_KEY),
         ));
 
-        // Persist the new finalized checkpoint as the pruning checkpoint.
-        batch.push(StoreOp::KeyValueOp(
-            store.pruning_checkpoint_store_op(new_finalized_checkpoint),
-        ));
-
         // Prune sync committee branches of non-checkpoint canonical finalized blocks
         Self::prune_non_checkpoint_sync_committee_branches(&newly_finalized_blocks, &mut batch);
         // Prune all payloads of the canonical finalized blocks
@@ -736,7 +735,10 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         debug!(log, "Database pruning complete");
 
         Ok(PruningOutcome::Successful {
-            old_finalized_checkpoint,
+            // TODO(hdiff): approximation of the previous finalized checkpoint. Only used in the
+            // compaction to compute time, can we use something else?
+            old_finalized_checkpoint_epoch: newly_finalized_blocks_min_slot
+                .epoch(E::slots_per_epoch()),
         })
     }
 
diff --git a/beacon_node/beacon_chain/src/summaries_dag.rs b/beacon_node/beacon_chain/src/summaries_dag.rs
@@ -178,7 +178,7 @@ impl StateSummariesDAG {
 
     /// Returns all ancestors of `state_root` INCLUDING `state_root` until the next parent is not
     /// known.
-    pub fn ancestors_of(&self, mut state_root: Hash256) -> Result<Vec<Hash256>, Error> {
+    pub fn ancestors_of(&self, mut state_root: Hash256) -> Result<Vec<(Hash256, Slot)>, Error> {
         // Sanity check that the first summary exists
         if !self.state_summaries_by_state_root.contains_key(&state_root) {
             return Err(Error::MissingStateSummary(state_root));
@@ -187,7 +187,7 @@ impl StateSummariesDAG {
         let mut ancestors = vec![];
         loop {
             if let Some(summary) = self.state_summaries_by_state_root.get(&state_root) {
-                ancestors.push(state_root);
+                ancestors.push((state_root, summary.slot));
                 state_root = summary.previous_state_root
             } else {
                 return Ok(ancestors);
diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs
@@ -6,10 +6,10 @@ use crate::iter::{BlockRootsIterator, ParentRootBlockIterator, RootsIterator};
 use crate::leveldb_store::{BytesKey, LevelDB};
 use crate::memory_store::MemoryStore;
 use crate::metadata::{
-    AnchorInfo, BlobInfo, CompactionTimestamp, DataColumnInfo, PruningCheckpoint, SchemaVersion,
+    AnchorInfo, BlobInfo, CompactionTimestamp, DataColumnInfo, SchemaVersion,
     ANCHOR_FOR_ARCHIVE_NODE, ANCHOR_INFO_KEY, ANCHOR_UNINITIALIZED, BLOB_INFO_KEY,
     COMPACTION_TIMESTAMP_KEY, CONFIG_KEY, CURRENT_SCHEMA_VERSION, DATA_COLUMN_INFO_KEY,
-    PRUNING_CHECKPOINT_KEY, SCHEMA_VERSION_KEY, SPLIT_KEY, STATE_UPPER_LIMIT_NO_RETAIN,
+    SCHEMA_VERSION_KEY, SPLIT_KEY, STATE_UPPER_LIMIT_NO_RETAIN,
 };
 use crate::state_cache::{PutStateOutcome, StateCache};
 use crate::{
@@ -2694,25 +2694,6 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
         self.config.compact_on_prune
     }
 
-    /// Load the checkpoint to begin pruning from (the "old finalized checkpoint").
-    pub fn load_pruning_checkpoint(&self) -> Result<Option<Checkpoint>, Error> {
-        Ok(self
-            .hot_db
-            .get(&PRUNING_CHECKPOINT_KEY)?
-            .map(|pc: PruningCheckpoint| pc.checkpoint))
-    }
-
-    /// Store the checkpoint to begin pruning from (the "old finalized checkpoint").
-    pub fn store_pruning_checkpoint(&self, checkpoint: Checkpoint) -> Result<(), Error> {
-        self.hot_db
-            .do_atomically(vec![self.pruning_checkpoint_store_op(checkpoint)])
-    }
-
-    /// Create a staged store for the pruning checkpoint.
-    pub fn pruning_checkpoint_store_op(&self, checkpoint: Checkpoint) -> KeyValueStoreOp {
-        PruningCheckpoint { checkpoint }.as_kv_store_op(PRUNING_CHECKPOINT_KEY)
-    }
-
     /// Load the timestamp of the last compaction as a `Duration` since the UNIX epoch.
     pub fn load_compaction_timestamp(&self) -> Result<Option<Duration>, Error> {
         Ok(self
diff --git a/beacon_node/store/src/metadata.rs b/beacon_node/store/src/metadata.rs
@@ -2,7 +2,7 @@ use crate::{DBColumn, Error, StoreItem};
 use serde::{Deserialize, Serialize};
 use ssz::{Decode, Encode};
 use ssz_derive::{Decode, Encode};
-use types::{Checkpoint, Hash256, Slot};
+use types::{Hash256, Slot};
 
 pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(23);
 
@@ -12,7 +12,8 @@ pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(23);
 pub const SCHEMA_VERSION_KEY: Hash256 = Hash256::repeat_byte(0);
 pub const CONFIG_KEY: Hash256 = Hash256::repeat_byte(1);
 pub const SPLIT_KEY: Hash256 = Hash256::repeat_byte(2);
-pub const PRUNING_CHECKPOINT_KEY: Hash256 = Hash256::repeat_byte(3);
+// DEPRECATED
+// pub const PRUNING_CHECKPOINT_KEY: Hash256 = Hash256::repeat_byte(3);
 pub const COMPACTION_TIMESTAMP_KEY: Hash256 = Hash256::repeat_byte(4);
 pub const ANCHOR_INFO_KEY: Hash256 = Hash256::repeat_byte(5);
 pub const BLOB_INFO_KEY: Hash256 = Hash256::repeat_byte(6);
@@ -65,30 +66,6 @@ impl StoreItem for SchemaVersion {
     }
 }
 
-/// The checkpoint used for pruning the database.
-///
-/// Updated whenever pruning is successful.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct PruningCheckpoint {
-    pub checkpoint: Checkpoint,
-}
-
-impl StoreItem for PruningCheckpoint {
-    fn db_column() -> DBColumn {
-        DBColumn::BeaconMeta
-    }
-
-    fn as_store_bytes(&self) -> Vec<u8> {
-        self.checkpoint.as_ssz_bytes()
-    }
-
-    fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
-        Ok(PruningCheckpoint {
-            checkpoint: Checkpoint::from_ssz_bytes(bytes)?,
-        })
-    }
-}
-
 /// The last time the database was compacted.
 pub struct CompactionTimestamp(pub u64);
 
