Add pruning db tool commands (MystenLabs#12695)

## Description 

Add db tool commands to trigger object and checkpoint pruning.

crates/sui-config/src/node.rs

@@ -473,7 +473,7 @@ pub struct AuthorityStorePruningConfig {
     pub periodic_compaction_threshold_days: Option<usize>,
     /// number of epochs to keep the latest version of transactions and effects for
     #[serde(skip_serializing_if = "Option::is_none")]
-    num_epochs_to_retain_for_checkpoints: Option<u64>,
+    pub num_epochs_to_retain_for_checkpoints: Option<u64>,
 }
 
 impl Default for AuthorityStorePruningConfig {

crates/sui-tool/src/commands.rs

@@ -304,7 +304,7 @@ impl ToolCommand {
             ToolCommand::DbTool { db_path, cmd } => {
                 let path = PathBuf::from(db_path);
                 match cmd {
-                    Some(c) => execute_db_tool_command(path, c)?,
+                    Some(c) => execute_db_tool_command(path, c).await?,
                     None => print_db_all_tables(path)?,
                 }
             }

crates/sui-tool/src/db_tool/db_dump.rs

@@ -4,19 +4,27 @@
 use anyhow::{anyhow, Ok};
 use clap::{Parser, ValueEnum};
 use comfy_table::{Cell, ContentArrangement, Row, Table};
+use prometheus::Registry;
 use rocksdb::MultiThreaded;
 use std::collections::{BTreeMap, HashMap};
 use std::path::PathBuf;
 use std::str;
 use std::sync::Arc;
 use strum_macros::EnumString;
+use sui_archival::reader::ArchiveReaderBalancer;
+use sui_config::node::AuthorityStorePruningConfig;
 use sui_core::authority::authority_per_epoch_store::AuthorityEpochTables;
-use sui_core::authority::authority_store_pruner::AuthorityStorePruner;
+use sui_core::authority::authority_store_pruner::{
+    AuthorityStorePruner, AuthorityStorePruningMetrics,
+};
 use sui_core::authority::authority_store_tables::AuthorityPerpetualTables;
 use sui_core::authority::authority_store_types::{StoreData, StoreObject};
+use sui_core::checkpoints::CheckpointStore;
 use sui_core::epoch::committee_store::CommitteeStoreTables;
+use sui_storage::mutex_table::RwLockTable;
 use sui_storage::IndexStoreTables;
 use sui_types::base_types::{EpochId, ObjectID};
+use tracing::info;
 use typed_store::rocks::{default_db_options, MetricConf};
 use typed_store::traits::{Map, TableSummary};
 
@@ -184,6 +192,71 @@ pub fn compact(db_path: PathBuf) -> anyhow::Result<()> {
     Ok(())
 }
 
+pub async fn prune_objects(db_path: PathBuf) -> anyhow::Result<()> {
+    let perpetual_db = Arc::new(AuthorityPerpetualTables::open(&db_path.join("store"), None));
+    let checkpoint_store = Arc::new(CheckpointStore::open_tables_read_write(
+        db_path.join("checkpoints"),
+        MetricConf::default(),
+        None,
+        None,
+    ));
+    let highest_pruned_checkpoint = checkpoint_store.get_highest_pruned_checkpoint_seq_number()?;
+    let latest_checkpoint = checkpoint_store.get_highest_executed_checkpoint()?;
+    info!(
+        "Latest executed checkpoint sequence num: {}",
+        latest_checkpoint.map(|x| x.sequence_number).unwrap_or(0)
+    );
+    info!("Highest pruned checkpoint: {}", highest_pruned_checkpoint);
+    let metrics = AuthorityStorePruningMetrics::new(&Registry::default());
+    let lock_table = Arc::new(RwLockTable::new(1));
+    info!("Pruning setup for db at path: {:?}", db_path.display());
+    let pruning_config = AuthorityStorePruningConfig {
+        num_epochs_to_retain: 0,
+        ..Default::default()
+    };
+    info!("Starting object pruning");
+    AuthorityStorePruner::prune_objects_for_eligible_epochs(
+        &perpetual_db,
+        &checkpoint_store,
+        &lock_table,
+        pruning_config,
+        metrics,
+        usize::MAX,
+    )
+    .await?;
+    Ok(())
+}
+
+pub async fn prune_checkpoints(db_path: PathBuf) -> anyhow::Result<()> {
+    let perpetual_db = Arc::new(AuthorityPerpetualTables::open(&db_path.join("store"), None));
+    let checkpoint_store = Arc::new(CheckpointStore::open_tables_read_write(
+        db_path.join("checkpoints"),
+        MetricConf::default(),
+        None,
+        None,
+    ));
+    let metrics = AuthorityStorePruningMetrics::new(&Registry::default());
+    let lock_table = Arc::new(RwLockTable::new(1));
+    info!("Pruning setup for db at path: {:?}", db_path.display());
+    let pruning_config = AuthorityStorePruningConfig {
+        num_epochs_to_retain_for_checkpoints: Some(1),
+        ..Default::default()
+    };
+    info!("Starting txns and effects pruning");
+    let archive_readers = ArchiveReaderBalancer::default();
+    AuthorityStorePruner::prune_checkpoints_for_eligible_epochs(
+        &perpetual_db,
+        &checkpoint_store,
+        &lock_table,
+        pruning_config,
+        metrics,
+        usize::MAX,
+        archive_readers,
+    )
+    .await?;
+    Ok(())
+}
+
 // TODO: condense this using macro or trait dyn skills
 pub fn dump_table(
     store_name: StoreName,

crates/sui-tool/src/db_tool/mod.rs

@@ -3,7 +3,7 @@
 
 use self::db_dump::{dump_table, duplicate_objects_summary, list_tables, table_summary, StoreName};
 use self::index_search::{search_index, SearchRange};
-use crate::db_tool::db_dump::{compact, print_table_metadata};
+use crate::db_tool::db_dump::{compact, print_table_metadata, prune_checkpoints, prune_objects};
 use anyhow::bail;
 use clap::Parser;
 use std::path::{Path, PathBuf};
@@ -35,6 +35,8 @@ pub enum DbToolCommand {
     ResetDB,
     RewindCheckpointExecution(RewindCheckpointExecutionOptions),
     Compact,
+    PruneObjects,
+    PruneCheckpoints,
 }
 
 #[derive(Parser)]
@@ -128,7 +130,7 @@ pub struct RewindCheckpointExecutionOptions {
     checkpoint_sequence_number: u64,
 }
 
-pub fn execute_db_tool_command(db_path: PathBuf, cmd: DbToolCommand) -> anyhow::Result<()> {
+pub async fn execute_db_tool_command(db_path: PathBuf, cmd: DbToolCommand) -> anyhow::Result<()> {
     match cmd {
         DbToolCommand::ListTables => print_db_all_tables(db_path),
         DbToolCommand::Dump(d) => print_all_entries(
@@ -154,6 +156,8 @@ pub fn execute_db_tool_command(db_path: PathBuf, cmd: DbToolCommand) -> anyhow::
             rewind_checkpoint_execution(&db_path, d.epoch, d.checkpoint_sequence_number)
         }
         DbToolCommand::Compact => compact(db_path),
+        DbToolCommand::PruneObjects => prune_objects(db_path).await,
+        DbToolCommand::PruneCheckpoints => prune_checkpoints(db_path).await,
         DbToolCommand::IndexSearchKeyRange(rg) => {
             let res = search_index(
                 db_path,