upstream(core): Introduce indexes pruner to authority store (#8536)

# Description of change

- Upstream range: [v1.44.3, v1.45.3)
- Port commit: 
-
MystenLabs/sui@5bc3504
- Description:
Make indexes prunable.


## Links to any relevant issues

Part of #6153 

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [ ] Patch-specific tests (correctness, functionality coverage)
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have checked that new and existing unit tests pass locally with
my changes

Author: jkrvivian

crates/iota-config/src/node.rs

@@ -987,6 +987,8 @@ pub struct AuthorityStorePruningConfig {
     /// may result in some old versions that will never be pruned.
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub enable_compaction_filter: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub num_epochs_to_retain_for_indexes: Option<u64>,
 }
 
 fn default_num_latest_epoch_dbs_to_retain() -> usize {
@@ -1026,6 +1028,7 @@ impl Default for AuthorityStorePruningConfig {
             num_epochs_to_retain_for_checkpoints: if cfg!(msim) { Some(2) } else { None },
             smooth: true,
             enable_compaction_filter: cfg!(test) || cfg!(msim),
+            num_epochs_to_retain_for_indexes: None,
         }
     }
 }

crates/iota-core/src/authority.rs

@@ -2827,6 +2827,7 @@ impl AuthorityState {
             store.perpetual_tables.clone(),
             checkpoint_store.clone(),
             rest_index.clone(),
+            indexes.clone(),
             config.authority_store_pruning_config.clone(),
             epoch_store.committee().authority_exists(&name),
             epoch_store.epoch_start_state().epoch_duration_ms(),

crates/iota-core/src/authority/authority_store_pruner.rs

@@ -41,6 +41,7 @@ use super::authority_store_tables::{AuthorityPerpetualTables, AuthorityPrunerTab
 use crate::{
     authority::authority_store_types::{StoreObject, StoreObjectWrapper},
     checkpoints::{CheckpointStore, CheckpointWatermark},
+    jsonrpc_index::IndexStore,
     rest_index::RestIndexStore,
 };
 
@@ -58,6 +59,7 @@ static PERIODIC_PRUNING_TABLES: Lazy<BTreeSet<String>> = Lazy::new(|| {
     .collect()
 });
 pub const EPOCH_DURATION_MS_FOR_TESTING: u64 = 24 * 60 * 60 * 1000;
+pub const MIN_EPOCHS_TO_RETAIN_FOR_INDEXES: u64 = 7;
 
 /// The `AuthorityStorePruner` manages the pruning process for object stores
 /// within the `AuthorityStore`. It includes a cancellation handle that can be
@@ -75,6 +77,7 @@ pub struct AuthorityStorePruningMetrics {
     pub num_pruned_objects: IntCounter,
     pub num_pruned_tombstones: IntCounter,
     pub last_pruned_effects_checkpoint: IntGauge,
+    pub last_pruned_indexes_transaction: IntGauge,
     pub num_epochs_to_retain_for_objects: IntGauge,
     pub num_epochs_to_retain_for_checkpoints: IntGauge,
 }
@@ -109,6 +112,12 @@ impl AuthorityStorePruningMetrics {
                 registry
             )
             .unwrap(),
+            last_pruned_indexes_transaction: register_int_gauge_with_registry!(
+                "last_pruned_indexes_transaction",
+                "Last pruned indexes transaction",
+                registry
+            )
+            .unwrap(),
             num_epochs_to_retain_for_objects: register_int_gauge_with_registry!(
                 "num_epochs_to_retain_for_objects",
                 "Number of epochs to retain for objects",
@@ -552,6 +561,32 @@ impl AuthorityStorePruner {
         Ok(())
     }
 
+    fn prune_indexes(
+        indexes: Option<&IndexStore>,
+        config: &AuthorityStorePruningConfig,
+        epoch_duration_ms: u64,
+        metrics: &AuthorityStorePruningMetrics,
+    ) -> anyhow::Result<()> {
+        if let (Some(mut epochs_to_retain), Some(indexes)) =
+            (config.num_epochs_to_retain_for_indexes, indexes)
+        {
+            if epochs_to_retain < MIN_EPOCHS_TO_RETAIN_FOR_INDEXES {
+                warn!("num_epochs_to_retain_for_indexes is too low. Resetting it to 7");
+                epochs_to_retain = MIN_EPOCHS_TO_RETAIN_FOR_INDEXES;
+            }
+            let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis();
+            if let Some(cut_time_ms) =
+                u64::try_from(now)?.checked_sub(epochs_to_retain * epoch_duration_ms)
+            {
+                let transaction_id = indexes.prune(cut_time_ms)?;
+                metrics
+                    .last_pruned_indexes_transaction
+                    .set(transaction_id as i64);
+            }
+        }
+        Ok(())
+    }
+
     /// Identifies and compacts the next eligible SST file in the
     /// `AuthorityStore` that meets the specified conditions for manual
     /// compaction. This function checks each SST file's metadata, including
@@ -650,6 +685,7 @@ impl AuthorityStorePruner {
         perpetual_db: Arc<AuthorityPerpetualTables>,
         checkpoint_store: Arc<CheckpointStore>,
         rest_index: Option<Arc<RestIndexStore>>,
+        jsonrpc_index: Option<Arc<IndexStore>>,
         pruner_db: Option<Arc<AuthorityPrunerTables>>,
         metrics: Arc<AuthorityStorePruningMetrics>,
         archive_readers: ArchiveReaderBalancer,
@@ -671,6 +707,8 @@ impl AuthorityStorePruner {
             tokio::time::interval_at(Instant::now() + pruning_initial_delay, tick_duration);
         let mut checkpoints_prune_interval =
             tokio::time::interval_at(Instant::now() + pruning_initial_delay, tick_duration);
+        let mut indexes_prune_interval =
+            tokio::time::interval_at(Instant::now() + pruning_initial_delay, tick_duration);
 
         let perpetual_db_for_compaction = perpetual_db.clone();
         if let Some(delay_days) = config.periodic_compaction_threshold_days {
@@ -719,6 +757,11 @@ impl AuthorityStorePruner {
                             error!("Failed to prune checkpoints: {:?}", err);
                         }
                     },
+                    _ = indexes_prune_interval.tick(), if config.num_epochs_to_retain_for_indexes.is_some() => {
+                        if let Err(err) = Self::prune_indexes(jsonrpc_index.as_deref(), &config, epoch_duration_ms, &metrics) {
+                            error!("Failed to prune indexes: {:?}", err);
+                        }
+                    }
                     _ = &mut recv => break,
                 }
             }
@@ -732,6 +775,7 @@ impl AuthorityStorePruner {
         perpetual_db: Arc<AuthorityPerpetualTables>,
         checkpoint_store: Arc<CheckpointStore>,
         rest_index: Option<Arc<RestIndexStore>>,
+        jsonrpc_index: Option<Arc<IndexStore>>,
         mut pruning_config: AuthorityStorePruningConfig,
         is_validator: bool,
         epoch_duration_ms: u64,
@@ -759,6 +803,7 @@ impl AuthorityStorePruner {
                 perpetual_db,
                 checkpoint_store,
                 rest_index,
+                jsonrpc_index,
                 pruner_db,
                 AuthorityStorePruningMetrics::new(registry),
                 archive_readers,