Manual finalization endpoint new pruning

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -33,7 +33,6 @@ use crate::events::ServerSentEventHandler;
 use crate::execution_payload::{get_execution_payload, NotifyExecutionLayer, PreparePayloadHandle};
 use crate::fork_choice_signal::{ForkChoiceSignalRx, ForkChoiceSignalTx, ForkChoiceWaitResult};
 use crate::graffiti_calculator::GraffitiCalculator;
-use crate::head_tracker::{HeadTracker, HeadTrackerReader, SszHeadTracker};
 use crate::kzg_utils::reconstruct_blobs;
 use crate::light_client_finality_update_verification::{
     Error as LightClientFinalityUpdateError, VerifiedLightClientFinalityUpdate,
@@ -42,7 +41,7 @@ use crate::light_client_optimistic_update_verification::{
     Error as LightClientOptimisticUpdateError, VerifiedLightClientOptimisticUpdate,
 };
 use crate::light_client_server_cache::LightClientServerCache;
-use crate::migrate::BackgroundMigrator;
+use crate::migrate::{BackgroundMigrator, ManualFinalizationNotification};
 use crate::naive_aggregation_pool::{
     AggregatedAttestationMap, Error as NaiveAggregationError, NaiveAggregationPool,
     SyncContributionAggregateMap,
@@ -57,7 +56,7 @@ use crate::observed_block_producers::ObservedBlockProducers;
 use crate::observed_data_sidecars::ObservedDataSidecars;
 use crate::observed_operations::{ObservationOutcome, ObservedOperations};
 use crate::observed_slashable::ObservedSlashable;
-use crate::persisted_beacon_chain::{PersistedBeaconChain, DUMMY_CANONICAL_HEAD_BLOCK_ROOT};
+use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::persisted_fork_choice::PersistedForkChoice;
 use crate::pre_finalization_cache::PreFinalizationBlockCache;
 use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
@@ -458,8 +457,6 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// A handler for events generated by the beacon chain. This is only initialized when the
     /// HTTP server is enabled.
     pub event_handler: Option<ServerSentEventHandler<T::EthSpec>>,
-    /// Used to track the heads of the beacon chain.
-    pub(crate) head_tracker: Arc<HeadTracker>,
     /// Caches the attester shuffling for a given epoch and shuffling key root.
     pub shuffling_cache: RwLock<ShufflingCache>,
     /// A cache of eth1 deposit data at epoch boundaries for deposit finalization
@@ -613,57 +610,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         })
     }
 
-    /// Persists the head tracker and fork choice.
+    /// Return a database operation for writing the `PersistedBeaconChain` to disk.
     ///
-    /// We do it atomically even though no guarantees need to be made about blocks from
-    /// the head tracker also being present in fork choice.
-    pub fn persist_head_and_fork_choice(&self) -> Result<(), Error> {
-        let mut batch = vec![];
-
-        let _head_timer = metrics::start_timer(&metrics::PERSIST_HEAD);
-
-        // Hold a lock to head_tracker until it has been persisted to disk. Otherwise there's a race
-        // condition with the pruning thread which can result in a block present in the head tracker
-        // but absent in the DB. This inconsistency halts pruning and dramastically increases disk
-        // size. Ref: https://github.com/sigp/lighthouse/issues/4773
-        let head_tracker = self.head_tracker.0.read();
-        batch.push(self.persist_head_in_batch(&head_tracker));
-
-        let _fork_choice_timer = metrics::start_timer(&metrics::PERSIST_FORK_CHOICE);
-        batch.push(self.persist_fork_choice_in_batch());
-
-        self.store.hot_db.do_atomically(batch)?;
-        drop(head_tracker);
-
-        Ok(())
-    }
-
-    /// Return a `PersistedBeaconChain` without reference to a `BeaconChain`.
-    pub fn make_persisted_head(
-        genesis_block_root: Hash256,
-        head_tracker_reader: &HeadTrackerReader,
-    ) -> PersistedBeaconChain {
-        PersistedBeaconChain {
-            _canonical_head_block_root: DUMMY_CANONICAL_HEAD_BLOCK_ROOT,
-            genesis_block_root,
-            ssz_head_tracker: SszHeadTracker::from_map(head_tracker_reader),
-        }
-    }
-
-    /// Return a database operation for writing the beacon chain head to disk.
-    pub fn persist_head_in_batch(
-        &self,
-        head_tracker_reader: &HeadTrackerReader,
-    ) -> KeyValueStoreOp {
-        Self::persist_head_in_batch_standalone(self.genesis_block_root, head_tracker_reader)
-    }
-
-    pub fn persist_head_in_batch_standalone(
-        genesis_block_root: Hash256,
-        head_tracker_reader: &HeadTrackerReader,
-    ) -> KeyValueStoreOp {
-        Self::make_persisted_head(genesis_block_root, head_tracker_reader)
-            .as_kv_store_op(BEACON_CHAIN_DB_KEY)
+    /// These days the `PersistedBeaconChain` is only used to store the genesis block root, so it
+    /// should only ever be written once at startup. It used to be written more frequently, but
+    /// this is no longer necessary.
+    pub fn persist_head_in_batch_standalone(genesis_block_root: Hash256) -> KeyValueStoreOp {
+        PersistedBeaconChain { genesis_block_root }.as_kv_store_op(BEACON_CHAIN_DB_KEY)
     }
 
     /// Load fork choice from disk, returning `None` if it isn't found.
@@ -1457,12 +1410,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ///
     /// Returns `(block_root, block_slot)`.
     pub fn heads(&self) -> Vec<(Hash256, Slot)> {
-        self.head_tracker.heads()
-    }
-
-    /// Only used in tests.
-    pub fn knows_head(&self, block_hash: &SignedBeaconBlockHash) -> bool {
-        self.head_tracker.contains_head((*block_hash).into())
+        self.canonical_head
+            .fork_choice_read_lock()
+            .proto_array()
+            .heads_descended_from_finalization::<T::EthSpec>()
+            .iter()
+            .map(|node| (node.root, node.slot))
+            .collect()
     }
 
     /// Returns the `BeaconState` at the given slot.
@@ -1707,6 +1661,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
     }
 
+    pub fn manually_finalize_state(&self, state_root: Hash256, checkpoint: Checkpoint) {
+        let notif = ManualFinalizationNotification {
+            state_root: state_root.into(),
+            checkpoint,
+        };
+        self.store_migrator.process_manual_finalization(notif)
+    }
+
     /// Returns an aggregated `Attestation`, if any, that has a matching `attestation.data`.
     ///
     /// The attestation will be obtained from `self.naive_aggregation_pool`.
@@ -2974,10 +2936,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// Only completed sampling results are received. Blocks are unavailable by default and should
     /// be pruned on finalization, on a timeout or by a max count.
     pub async fn process_sampling_completed(self: &Arc<Self>, block_root: Hash256) {
-        // TODO(das): update fork-choice
+        // TODO(das): update fork-choice, act on sampling result, adjust log level
         // NOTE: It is possible that sampling complets before block is imported into fork choice,
         // in that case we may need to update availability cache.
-        // TODO(das): These log levels are too high, reduce once DAS matures
         info!(self.log, "Sampling completed"; "block_root" => %block_root);
     }
 
@@ -4024,9 +3985,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // about it.
         let block_time_imported = timestamp_now();
 
-        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(),
@@ -4047,9 +4005,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 });
         }
 
-        self.head_tracker
-            .register_block(block_root, parent_root, slot);
-
         metrics::stop_timer(db_write_timer);
 
         metrics::inc_counter(&metrics::BLOCK_PROCESSING_SUCCESSES);
@@ -7312,7 +7267,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 impl<T: BeaconChainTypes> Drop for BeaconChain<T> {
     fn drop(&mut self) {
         let drop = || -> Result<(), Error> {
-            self.persist_head_and_fork_choice()?;
+            self.persist_fork_choice()?;
             self.persist_op_pool()?;
             self.persist_eth1_cache()
         };

beacon_node/beacon_chain/src/block_verification.rs

@@ -1782,7 +1782,12 @@ pub fn check_block_is_finalized_checkpoint_or_descendant<
     fork_choice: &BeaconForkChoice<T>,
     block: B,
 ) -> Result<B, BlockError> {
-    if fork_choice.is_finalized_checkpoint_or_descendant(block.parent_root()) {
+    // If we have a split block newer than finalization then we also ban blocks which are not
+    // descended from that split block.
+    let split = chain.store.get_split_info();
+    if fork_choice.is_finalized_checkpoint_or_descendant(block.parent_root())
+        && fork_choice.is_descendant(split.block_root, block.parent_root())
+    {
         Ok(block)
     } else {
         // If fork choice does *not* consider the parent to be a descendant of the finalized block,

beacon_node/beacon_chain/src/builder.rs

@@ -8,7 +8,6 @@ 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::graffiti_calculator::{GraffitiCalculator, GraffitiOrigin};
-use crate::head_tracker::HeadTracker;
 use crate::kzg_utils::blobs_to_data_column_sidecars;
 use crate::light_client_server_cache::LightClientServerCache;
 use crate::migrate::{BackgroundMigrator, MigratorConfig};
@@ -92,7 +91,6 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     slot_clock: Option<T::SlotClock>,
     shutdown_sender: Option<Sender<ShutdownReason>>,
     light_client_server_tx: Option<Sender<LightClientProducerEvent<T::EthSpec>>>,
-    head_tracker: Option<HeadTracker>,
     validator_pubkey_cache: Option<ValidatorPubkeyCache<T>>,
     spec: Arc<ChainSpec>,
     chain_config: ChainConfig,
@@ -136,7 +134,6 @@ where
             slot_clock: None,
             shutdown_sender: None,
             light_client_server_tx: None,
-            head_tracker: None,
             validator_pubkey_cache: None,
             spec: Arc::new(E::default_spec()),
             chain_config: ChainConfig::default(),
@@ -325,10 +322,6 @@ where
 
         self.genesis_block_root = Some(chain.genesis_block_root);
         self.genesis_state_root = Some(genesis_block.state_root());
-        self.head_tracker = Some(
-            HeadTracker::from_ssz_container(&chain.ssz_head_tracker)
-                .map_err(|e| format!("Failed to decode head tracker for database: {:?}", e))?,
-        );
         self.validator_pubkey_cache = Some(pubkey_cache);
         self.fork_choice = Some(fork_choice);
 
@@ -746,7 +739,6 @@ where
             .genesis_state_root
             .ok_or("Cannot build without a genesis state root")?;
         let validator_monitor_config = self.validator_monitor_config.unwrap_or_default();
-        let head_tracker = Arc::new(self.head_tracker.unwrap_or_default());
         let beacon_proposer_cache: Arc<Mutex<BeaconProposerCache>> = <_>::default();
 
         let mut validator_monitor = ValidatorMonitor::new(
@@ -791,8 +783,6 @@ where
                         &log,
                     )?;
 
-                    // Update head tracker.
-                    head_tracker.register_block(block_root, block.parent_root(), block.slot());
                     (block_root, block, true)
                 }
                 Err(e) => return Err(descriptive_db_error("head block", &e)),
@@ -848,12 +838,7 @@ where
         })?;
 
         let migrator_config = self.store_migrator_config.unwrap_or_default();
-        let store_migrator = BackgroundMigrator::new(
-            store.clone(),
-            migrator_config,
-            genesis_block_root,
-            log.clone(),
-        );
+        let store_migrator = BackgroundMigrator::new(store.clone(), migrator_config, log.clone());
 
         if let Some(slot) = slot_clock.now() {
             validator_monitor.process_valid_state(
@@ -878,11 +863,10 @@ where
         //
         // This *must* be stored before constructing the `BeaconChain`, so that its `Drop` instance
         // doesn't write a `PersistedBeaconChain` without the rest of the batch.
-        let head_tracker_reader = head_tracker.0.read();
         self.pending_io_batch.push(BeaconChain::<
             Witness<TSlotClock, TEth1Backend, E, THotStore, TColdStore>,
         >::persist_head_in_batch_standalone(
-            genesis_block_root, &head_tracker_reader
+            genesis_block_root
         ));
         self.pending_io_batch.push(BeaconChain::<
             Witness<TSlotClock, TEth1Backend, E, THotStore, TColdStore>,
@@ -893,7 +877,6 @@ where
             .hot_db
             .do_atomically(self.pending_io_batch)
             .map_err(|e| format!("Error writing chain & metadata to disk: {:?}", e))?;
-        drop(head_tracker_reader);
 
         let genesis_validators_root = head_snapshot.beacon_state.genesis_validators_root();
         let genesis_time = head_snapshot.beacon_state.genesis_time();
@@ -974,7 +957,6 @@ where
             fork_choice_signal_tx,
             fork_choice_signal_rx,
             event_handler: self.event_handler,
-            head_tracker,
             shuffling_cache: RwLock::new(ShufflingCache::new(
                 shuffling_cache_size,
                 head_shuffling_ids,

beacon_node/beacon_chain/src/canonical_head.rs

@@ -53,7 +53,7 @@ use slot_clock::SlotClock;
 use state_processing::AllCaches;
 use std::sync::Arc;
 use std::time::Duration;
-use store::{iter::StateRootsIterator, KeyValueStoreOp, StoreItem};
+use store::{iter::StateRootsIterator, KeyValueStore, KeyValueStoreOp, StoreItem};
 use task_executor::{JoinHandle, ShutdownReason};
 use types::*;
 
@@ -848,7 +848,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         );
 
         if is_epoch_transition || reorg_distance.is_some() {
-            self.persist_head_and_fork_choice()?;
+            self.persist_fork_choice()?;
             self.op_pool.prune_attestations(self.epoch()?);
         }
 
@@ -992,7 +992,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.store_migrator.process_finalization(
             new_finalized_state_root.into(),
             new_view.finalized_checkpoint,
-            self.head_tracker.clone(),
         )?;
 
         // Prune blobs in the background.
@@ -1007,6 +1006,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(())
     }
 
+    /// Persist fork choice to disk, writing immediately.
+    pub fn persist_fork_choice(&self) -> Result<(), Error> {
+        let _fork_choice_timer = metrics::start_timer(&metrics::PERSIST_FORK_CHOICE);
+        let batch = vec![self.persist_fork_choice_in_batch()];
+        self.store.hot_db.do_atomically(batch)?;
+        Ok(())
+    }
+
     /// Return a database operation for writing fork choice to disk.
     pub fn persist_fork_choice_in_batch(&self) -> KeyValueStoreOp {
         Self::persist_fork_choice_in_batch_standalone(&self.canonical_head.fork_choice_read_lock())

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -110,15 +110,6 @@ impl<E: EthSpec> PendingComponents<E> {
         self.get_cached_blobs().iter().flatten().count()
     }
 
-    /// Checks if a data column of a given index exists in the cache.
-    ///
-    /// Returns:
-    /// - `true` if a data column for the given index exists.
-    /// - `false` otherwise.
-    fn data_column_exists(&self, data_column_index: u64) -> bool {
-        self.get_cached_data_column(data_column_index).is_some()
-    }
-
     /// Returns the number of data columns that have been received and are stored in the cache.
     pub fn num_received_data_columns(&self) -> usize {
         self.verified_data_columns.len()
@@ -182,8 +173,7 @@ impl<E: EthSpec> PendingComponents<E> {
         kzg_verified_data_columns: I,
     ) -> Result<(), AvailabilityCheckError> {
         for data_column in kzg_verified_data_columns {
-            // TODO(das): Add equivalent checks for data columns if necessary
-            if !self.data_column_exists(data_column.index()) {
+            if self.get_cached_data_column(data_column.index()).is_none() {
                 self.verified_data_columns.push(data_column);
             }
         }