Manual finalization endpoint

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -42,7 +42,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,
@@ -1707,6 +1707,16 @@ 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,
+            head_tracker: self.head_tracker.clone(),
+            genesis_block_root: self.genesis_block_root,
+        };
+        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`.

beacon_node/beacon_chain/src/migrate.rs

@@ -124,6 +124,14 @@ pub enum Notification {
     Finalization(FinalizationNotification),
     Reconstruction,
     PruneBlobs(Epoch),
+    ManualFinalization(ManualFinalizationNotification),
+}
+
+pub struct ManualFinalizationNotification {
+    pub state_root: BeaconStateHash,
+    pub checkpoint: Checkpoint,
+    pub head_tracker: Arc<HeadTracker>,
+    pub genesis_block_root: Hash256,
 }
 
 pub struct FinalizationNotification {
@@ -190,6 +198,10 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         Ok(())
     }
 
+    pub fn process_manual_finalization(&self, notif: ManualFinalizationNotification) {
+        let _ = self.send_background_notification(Notification::ManualFinalization(notif));
+    }
+
     pub fn process_reconstruction(&self) {
         if let Some(Notification::Reconstruction) =
             self.send_background_notification(Notification::Reconstruction)
@@ -289,6 +301,103 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         }
     }
 
+    fn run_manual_migration(
+        db: Arc<HotColdDB<E, Hot, Cold>>,
+        notif: ManualFinalizationNotification,
+        log: &Logger,
+    ) {
+        let state_root = notif.state_root;
+        let block_root = notif.checkpoint.root;
+
+        let manually_finalized_state = match db.get_state(&state_root.into(), None) {
+            Ok(Some(state)) => state,
+            other => {
+                error!(
+                    log,
+                    "Manual migrator failed to load state";
+                    "state_root" => ?state_root,
+                    "error" => ?other
+                );
+                return;
+            }
+        };
+
+        let old_finalized_checkpoint = match Self::prune_abandoned_forks(
+            db.clone(),
+            notif.head_tracker,
+            state_root,
+            &manually_finalized_state,
+            notif.checkpoint,
+            notif.genesis_block_root,
+            log,
+        ) {
+            Ok(PruningOutcome::Successful {
+                old_finalized_checkpoint,
+            }) => old_finalized_checkpoint,
+            Ok(PruningOutcome::DeferredConcurrentHeadTrackerMutation) => {
+                warn!(
+                    log,
+                    "Pruning deferred because of a concurrent mutation";
+                    "message" => "this is expected only very rarely!"
+                );
+                return;
+            }
+            Ok(PruningOutcome::OutOfOrderFinalization {
+                old_finalized_checkpoint,
+                new_finalized_checkpoint,
+            }) => {
+                warn!(
+                    log,
+                    "Ignoring out of order finalization request";
+                    "old_finalized_epoch" => old_finalized_checkpoint.epoch,
+                    "new_finalized_epoch" => new_finalized_checkpoint.epoch,
+                    "message" => "this is expected occasionally due to a (harmless) race condition"
+                );
+                return;
+            }
+            Err(e) => {
+                warn!(log, "Block pruning failed"; "error" => ?e);
+                return;
+            }
+        };
+
+        match migrate_database(
+            db.clone(),
+            state_root.into(),
+            block_root,
+            &manually_finalized_state,
+        ) {
+            Ok(()) => {}
+            Err(Error::HotColdDBError(HotColdDBError::FreezeSlotUnaligned(slot))) => {
+                debug!(
+                    log,
+                    "Database migration postponed, unaligned finalized block";
+                    "slot" => slot.as_u64()
+                );
+            }
+            Err(e) => {
+                warn!(
+                    log,
+                    "Database migration failed";
+                    "error" => format!("{:?}", e)
+                );
+                return;
+            }
+        };
+
+        // Finally, compact the database so that new free space is properly reclaimed.
+        if let Err(e) = Self::run_compaction(
+            db,
+            old_finalized_checkpoint.epoch,
+            notif.checkpoint.epoch,
+            log,
+        ) {
+            warn!(log, "Database compaction failed"; "error" => format!("{:?}", e));
+        }
+
+        debug!(log, "Database consolidation complete");
+    }
+
     /// Perform the actual work of `process_finalization`.
     fn run_migration(
         db: Arc<HotColdDB<E, Hot, Cold>>,
@@ -422,16 +531,27 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             while let Ok(notif) = rx.recv() {
                 let mut reconstruction_notif = None;
                 let mut finalization_notif = None;
+                let mut manual_finalization_notif = None;
                 let mut prune_blobs_notif = None;
                 match notif {
                     Notification::Reconstruction => reconstruction_notif = Some(notif),
                     Notification::Finalization(fin) => finalization_notif = Some(fin),
+                    Notification::ManualFinalization(fin) => manual_finalization_notif = Some(fin),
                     Notification::PruneBlobs(dab) => prune_blobs_notif = Some(dab),
                 }
                 // Read the rest of the messages in the channel, taking the best of each type.
                 for notif in rx.try_iter() {
                     match notif {
                         Notification::Reconstruction => reconstruction_notif = Some(notif),
+                        Notification::ManualFinalization(fin) => {
+                            if let Some(current) = manual_finalization_notif.as_mut() {
+                                if fin.checkpoint.epoch > current.checkpoint.epoch {
+                                    *current = fin;
+                                }
+                            } else {
+                                manual_finalization_notif = Some(fin);
+                            }
+                        }
                         Notification::Finalization(fin) => {
                             if let Some(current) = finalization_notif.as_mut() {
                                 if fin.finalized_checkpoint.epoch
@@ -454,6 +574,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                 if let Some(fin) = finalization_notif {
                     Self::run_migration(db.clone(), fin, &log);
                 }
+                if let Some(fin) = manual_finalization_notif {
+                    Self::run_manual_migration(db.clone(), fin, &log);
+                }
                 if let Some(dab) = prune_blobs_notif {
                     Self::run_prune_blobs(db.clone(), dab, &log);
                 }

beacon_node/http_api/src/lib.rs

@@ -86,11 +86,11 @@ use tokio_stream::{
 };
 use types::{
     fork_versioned_response::EmptyMetadata, Attestation, AttestationData, AttestationShufflingId,
-    AttesterSlashing, BeaconStateError, ChainSpec, CommitteeCache, ConfigAndPreset, Epoch, EthSpec,
-    ForkName, ForkVersionedResponse, Hash256, ProposerPreparationData, ProposerSlashing,
-    RelativeEpoch, SignedAggregateAndProof, SignedBlindedBeaconBlock, SignedBlsToExecutionChange,
-    SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
-    SyncCommitteeMessage, SyncContributionData,
+    AttesterSlashing, BeaconStateError, ChainSpec, Checkpoint, CommitteeCache, ConfigAndPreset,
+    Epoch, EthSpec, ForkName, ForkVersionedResponse, Hash256, ProposerPreparationData,
+    ProposerSlashing, RelativeEpoch, SignedAggregateAndProof, SignedBlindedBeaconBlock,
+    SignedBlsToExecutionChange, SignedContributionAndProof, SignedValidatorRegistrationData,
+    SignedVoluntaryExit, Slot, SyncCommitteeMessage, SyncContributionData,
 };
 use validator::pubkey_to_validator_index;
 use version::{
@@ -4133,6 +4133,29 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // POST lighthouse/finalize
+    let post_lighthouse_finalize = warp::path("lighthouse")
+        .and(warp::path("finalize"))
+        .and(warp::path::end())
+        .and(warp_utils::json::json())
+        .and(task_spawner_filter.clone())
+        .and(chain_filter.clone())
+        .then(
+            |request_data: api_types::ManualFinalizationRequestData,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             chain: Arc<BeaconChain<T>>| {
+                task_spawner.blocking_json_task(Priority::P0, move || {
+                    let checkpoint = Checkpoint {
+                        epoch: request_data.epoch,
+                        root: request_data.block_root,
+                    };
+                    chain.manually_finalize_state(request_data.state_root, checkpoint);
+
+                    Ok(api_types::GenericResponse::from(request_data))
+                })
+            },
+        );
+
     // POST lighthouse/liveness
     let post_lighthouse_liveness = warp::path("lighthouse")
         .and(warp::path("liveness"))
@@ -4932,6 +4955,7 @@ pub fn serve<T: BeaconChainTypes>(
                     .uor(post_lighthouse_block_rewards)
                     .uor(post_lighthouse_ui_validator_metrics)
                     .uor(post_lighthouse_ui_validator_info)
+                    .uor(post_lighthouse_finalize)
                     .recover(warp_utils::reject::handle_rejection),
             ),
         )

common/eth2/src/types.rs

@@ -1424,6 +1424,13 @@ pub struct StandardLivenessResponseData {
     pub is_live: bool,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ManualFinalizationRequestData {
+    pub state_root: Hash256,
+    pub epoch: Epoch,
+    pub block_root: Hash256,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct LivenessRequestData {
     pub epoch: Epoch,