From 1b585978c1d93cd60700ce21d81f9b60bc178958 Mon Sep 17 00:00:00 2001
From: Eitan Seri-Levi <eserilev@ucsc.edu>
Date: Mon, 3 Mar 2025 21:44:50 -0700
Subject: [PATCH] Manual compaction endpoint (#7072)

* manual compaction endpoint

* linting

---------

Co-authored-by: Michael Sproul <michael@sigmaprime.io>
---
 beacon_node/beacon_chain/src/beacon_chain.rs |  4 ++++
 beacon_node/beacon_chain/src/migrate.rs      | 24 ++++++++++++++++++++
 beacon_node/http_api/src/lib.rs              | 18 +++++++++++++++
 3 files changed, 46 insertions(+)

diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index c2f52d89e7..c9f47bce87 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -1709,6 +1709,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
     }
 
+    pub fn manually_compact_database(&self) {
+        self.store_migrator.process_manual_compaction();
+    }
+
     pub fn manually_finalize_state(
         &self,
         state_root: Hash256,
diff --git a/beacon_node/beacon_chain/src/migrate.rs b/beacon_node/beacon_chain/src/migrate.rs
index ca20321be0..615b19118b 100644
--- a/beacon_node/beacon_chain/src/migrate.rs
+++ b/beacon_node/beacon_chain/src/migrate.rs
@@ -125,6 +125,7 @@ pub enum Notification {
     Reconstruction,
     PruneBlobs(Epoch),
     ManualFinalization(ManualFinalizationNotification),
+    ManualCompaction,
 }
 
 pub struct ManualFinalizationNotification {
@@ -198,6 +199,14 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         Ok(())
     }
 
+    pub fn process_manual_compaction(&self) {
+        if let Some(Notification::ManualCompaction) =
+            self.send_background_notification(Notification::ManualCompaction)
+        {
+            Self::run_manual_compaction(self.db.clone(), &self.log);
+        }
+    }
+
     pub fn process_manual_finalization(&self, notif: ManualFinalizationNotification) {
         if let Some(Notification::ManualFinalization(notif)) =
             self.send_background_notification(Notification::ManualFinalization(notif))
@@ -445,6 +454,15 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         debug!(log, "Database consolidation complete");
     }
 
+    fn run_manual_compaction(db: Arc<HotColdDB<E, Hot, Cold>>, log: &Logger) {
+        debug!(log, "Running manual compaction");
+        if let Err(e) = db.compact() {
+            warn!(log, "Database compaction failed"; "error" => format!("{:?}", e));
+        } else {
+            debug!(log, "Manual compaction completed");
+        }
+    }
+
     /// Spawn a new child thread to run the migration process.
     ///
     /// Return a channel handle for sending requests to the thread.
@@ -459,17 +477,20 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                 let mut reconstruction_notif = None;
                 let mut finalization_notif = None;
                 let mut manual_finalization_notif = None;
+                let mut manual_compaction_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),
+                    Notification::ManualCompaction => manual_compaction_notif = Some(notif),
                 }
                 // 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::ManualCompaction => manual_compaction_notif = Some(notif),
                         Notification::ManualFinalization(fin) => {
                             if let Some(current) = manual_finalization_notif.as_mut() {
                                 if fin.checkpoint.epoch > current.checkpoint.epoch {
@@ -510,6 +531,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                 if reconstruction_notif.is_some() {
                     Self::run_reconstruction(db.clone(), Some(inner_tx.clone()), &log);
                 }
+                if manual_compaction_notif.is_some() {
+                    Self::run_manual_compaction(db.clone(), &log);
+                }
             }
         });
         (tx, thread)
diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs
index 58208d5b79..8e878ca154 100644
--- a/beacon_node/http_api/src/lib.rs
+++ b/beacon_node/http_api/src/lib.rs
@@ -4163,6 +4163,23 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // POST lighthouse/compaction
+    let post_lighthouse_compaction = warp::path("lighthouse")
+        .and(warp::path("compaction"))
+        .and(warp::path::end())
+        .and(task_spawner_filter.clone())
+        .and(chain_filter.clone())
+        .then(
+            |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
+                task_spawner.blocking_json_task(Priority::P0, move || {
+                    chain.manually_compact_database();
+                    Ok(api_types::GenericResponse::from(String::from(
+                        "Triggered manual compaction",
+                    )))
+                })
+            },
+        );
+
     // POST lighthouse/add_peer
     let post_lighthouse_add_peer = warp::path("lighthouse")
         .and(warp::path("add_peer"))
@@ -4968,6 +4985,7 @@ pub fn serve<T: BeaconChainTypes>(
                     .uor(post_lighthouse_ui_validator_metrics)
                     .uor(post_lighthouse_ui_validator_info)
                     .uor(post_lighthouse_finalize)
+                    .uor(post_lighthouse_compaction)
                     .uor(post_lighthouse_add_peer)
                     .recover(warp_utils::reject::handle_rejection),
             ),