[objects pruner] live pruner support (MystenLabs#7923)

First iteration of live pruner.
**Implementation**: `AuthorityStorePruner` is subscribed to updates from
`CheckpointExecutor`
 
Notes:
* for now the same deletion method is used as in original pruner:
`delete_range`. Eventually we may want to experiment with point-deletes.
Starting with delete_range to address potential tombstone concern
separately

crates/sui-benchmark/Cargo.toml

@@ -55,6 +55,7 @@ sysinfo = "0.27.5"
 [target.'cfg(msim)'.dependencies]
 sui-macros = { path = "../sui-macros" }
 sui-simulator = { path = "../sui-simulator" }
+typed-store = { path = "../typed-store" }
 
 [features]
 benchmark = ["narwhal-node/benchmark"]

crates/sui-benchmark/tests/simtest.rs

@@ -4,6 +4,7 @@
 #[cfg(msim)]
 mod test {
 
+    use itertools::Itertools;
     use rand::{thread_rng, Rng};
     use std::str::FromStr;
     use std::sync::{Arc, Mutex};
@@ -20,13 +21,19 @@ mod test {
         workloads::make_combination_workload,
         LocalValidatorAggregatorProxy, ValidatorProxy,
     };
-    use sui_config::SUI_KEYSTORE_FILENAME;
+    use sui_config::{AUTHORITIES_DB_NAME, SUI_KEYSTORE_FILENAME};
     use sui_macros::{register_fail_points, sim_test};
     use sui_simulator::{configs::*, SimConfig};
-    use sui_types::object::Owner;
+    use sui_types::object::{Object, Owner};
+    use sui_types::storage::ObjectKey;
     use test_utils::messages::get_sui_gas_object_with_wallet_context;
     use test_utils::network::{TestCluster, TestClusterBuilder};
     use tracing::info;
+    use typed_store::rocks::ReadWriteOptions;
+    use typed_store::{
+        rocks::{DBMap, MetricConf},
+        traits::Map,
+    };
 
     fn test_config() -> SimConfig {
         env_config(
@@ -140,6 +147,35 @@ mod test {
         test_simulated_load(test_cluster, 120).await;
     }
 
+    #[sim_test(config = "test_config()")]
+    async fn test_simulated_load_pruning() {
+        let epoch_duration_ms = 1000;
+        let test_cluster = build_test_cluster(7, epoch_duration_ms).await;
+        test_simulated_load(test_cluster.clone(), 5).await;
+        // waiting enough time to get all transactions into checkpoints
+        tokio::time::sleep(Duration::from_millis(2 * epoch_duration_ms)).await;
+
+        let swarm_dir = test_cluster.swarm.dir().join(AUTHORITIES_DB_NAME);
+        let validator_path = std::fs::read_dir(swarm_dir).unwrap().next().unwrap();
+        let db_path = validator_path.unwrap().path().join("store");
+
+        let db = typed_store::rocks::open_cf(&db_path, None, MetricConf::default(), &["objects"]);
+        let objects = DBMap::<ObjectKey, Object>::reopen(
+            &db.unwrap(),
+            Some("objects"),
+            &ReadWriteOptions {
+                ignore_range_deletions: false,
+            },
+        )
+        .unwrap();
+
+        let iter = objects.iter().skip_to_last().reverse();
+        for (_, group) in &iter.group_by(|item| item.0 .0) {
+            // assure  only last version is kept
+            assert_eq!(group.count(), 1);
+        }
+    }
+
     async fn build_test_cluster(
         default_num_validators: usize,
         default_epoch_duration_ms: u64,

crates/sui-config/data/fullnode-template-with-path.yaml

@@ -17,6 +17,7 @@ authority-store-pruning-config:
   objects-pruning-initial-delay-secs: 3600
   num-latest-epoch-dbs-to-retain: 3
   epoch-db-pruning-period-secs: 3600
+  enable-live-pruner: true
 
 protocol-key-pair:
   path: "protocol.key"

crates/sui-config/data/fullnode-template.yaml

@@ -17,3 +17,4 @@ authority-store-pruning-config:
   objects-pruning-initial-delay-secs: 3600
   num-latest-epoch-dbs-to-retain: 3
   epoch-db-pruning-period-secs: 3600
+  enable-live-pruner: true

crates/sui-config/src/node.rs

@@ -277,16 +277,18 @@ pub struct AuthorityStorePruningConfig {
     pub objects_pruning_initial_delay_secs: u64,
     pub num_latest_epoch_dbs_to_retain: usize,
     pub epoch_db_pruning_period_secs: u64,
+    pub enable_live_pruner: bool,
 }
 
 impl Default for AuthorityStorePruningConfig {
     fn default() -> Self {
         Self {
             objects_num_latest_versions_to_retain: u64::MAX,
-            objects_pruning_period_secs: u64::MAX,
-            objects_pruning_initial_delay_secs: u64::MAX,
+            objects_pruning_period_secs: 24 * 60 * 60,
+            objects_pruning_initial_delay_secs: 60 * 60,
             num_latest_epoch_dbs_to_retain: usize::MAX,
             epoch_db_pruning_period_secs: u64::MAX,
+            enable_live_pruner: cfg!(test) || cfg!(msim),
         }
     }
 }
@@ -301,6 +303,7 @@ impl AuthorityStorePruningConfig {
             objects_pruning_initial_delay_secs: 60 * 60,
             num_latest_epoch_dbs_to_retain: 3,
             epoch_db_pruning_period_secs: 60 * 60,
+            enable_live_pruner: cfg!(test) || cfg!(msim),
         }
     }
     pub fn fullnode_config() -> Self {
@@ -310,6 +313,7 @@ impl AuthorityStorePruningConfig {
             objects_pruning_initial_delay_secs: 60 * 60,
             num_latest_epoch_dbs_to_retain: 3,
             epoch_db_pruning_period_secs: 60 * 60,
+            enable_live_pruner: cfg!(test) || cfg!(msim),
         }
     }
 }

crates/sui-config/src/swarm.rs

@@ -85,6 +85,7 @@ pub struct FullnodeConfigBuilder<'a> {
     rpc_port: Option<u16>,
     // port for admin interface
     admin_port: Option<u16>,
+    enable_pruner: bool,
 }
 
 impl<'a> FullnodeConfigBuilder<'a> {
@@ -98,6 +99,7 @@ impl<'a> FullnodeConfigBuilder<'a> {
             p2p_port: None,
             rpc_port: None,
             admin_port: None,
+            enable_pruner: true,
         }
     }
 
@@ -156,6 +158,11 @@ impl<'a> FullnodeConfigBuilder<'a> {
         self
     }
 
+    pub fn set_enable_pruner(mut self, status: bool) -> Self {
+        self.enable_pruner = status;
+        self
+    }
+
     pub fn build(self) -> Result<NodeConfig, anyhow::Error> {
         let protocol_key_pair = get_key_pair_from_rng::<AuthorityKeyPair, _>(&mut OsRng).1;
         let worker_key_pair = get_key_pair_from_rng::<NetworkKeyPair, _>(&mut OsRng).1;
@@ -216,6 +223,8 @@ impl<'a> FullnodeConfigBuilder<'a> {
         let rpc_port = self.rpc_port.unwrap_or_else(|| get_available_port(9000));
         let jsonrpc_server_url = format!("{}:{}", listen_ip, rpc_port);
         let json_rpc_address: SocketAddr = jsonrpc_server_url.parse().unwrap();
+        let mut authority_store_pruning_config = AuthorityStorePruningConfig::fullnode_config();
+        authority_store_pruning_config.enable_live_pruner = self.enable_pruner;
 
         Ok(NodeConfig {
             protocol_key_pair: AuthorityKeyPairWithPath::new(protocol_key_pair),
@@ -237,7 +246,7 @@ impl<'a> FullnodeConfigBuilder<'a> {
             grpc_load_shed: None,
             grpc_concurrency_limit: None,
             p2p_config,
-            authority_store_pruning_config: AuthorityStorePruningConfig::fullnode_config(),
+            authority_store_pruning_config,
             end_of_epoch_broadcast_channel_capacity:
                 default_end_of_epoch_broadcast_channel_capacity(),
             checkpoint_executor_config: Default::default(),

crates/sui-config/tests/snapshots/snapshot_tests__network_config_snapshot_matches.snap

@@ -67,6 +67,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -135,6 +136,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -203,6 +205,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -271,6 +274,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -339,6 +343,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -407,6 +412,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100
@@ -475,6 +481,7 @@ validator_configs:
       objects-pruning-initial-delay-secs: 3600
       num-latest-epoch-dbs-to-retain: 3
       epoch-db-pruning-period-secs: 3600
+      enable-live-pruner: false
     end-of-epoch-broadcast-channel-capacity: 128
     checkpoint-executor-config:
       checkpoint-execution-max-concurrency: 100

crates/sui-core/src/authority/authority_store.rs

@@ -4,6 +4,7 @@
 use super::authority_store_pruner::AuthorityStorePruner;
 use super::{authority_store_tables::AuthorityPerpetualTables, *};
 use crate::authority::authority_per_epoch_store::AuthorityPerEpochStore;
+use crate::checkpoints::checkpoint_executor::CheckpointExecutionMessage;
 use once_cell::sync::OnceCell;
 use rocksdb::Options;
 use serde::{Deserialize, Serialize};
@@ -18,7 +19,7 @@ use sui_types::object::Owner;
 use sui_types::object::PACKAGE_VERSION;
 use sui_types::storage::{ChildObjectResolver, ObjectKey};
 use sui_types::{base_types::SequenceNumber, fp_bail, fp_ensure, storage::ParentSync};
-use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
+use tokio::sync::{mpsc, RwLock, RwLockReadGuard, RwLockWriteGuard};
 use tracing::{debug, info, trace};
 use typed_store::rocks::DBBatch;
 use typed_store::traits::Map;
@@ -60,6 +61,7 @@ impl AuthorityStore {
         genesis: &Genesis,
         committee_store: &Arc<CommitteeStore>,
         pruning_config: &AuthorityStorePruningConfig,
+        checkpoint_stream: mpsc::Receiver<CheckpointExecutionMessage>,
     ) -> SuiResult<Self> {
         let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(path, db_options.clone()));
         if perpetual_tables.database_is_empty()? {
@@ -69,7 +71,14 @@ impl AuthorityStore {
         let committee = committee_store
             .get_committee(&cur_epoch)?
             .expect("Committee of the current epoch must exist");
-        Self::open_inner(genesis, perpetual_tables, committee, pruning_config).await
+        Self::open_inner(
+            genesis,
+            perpetual_tables,
+            committee,
+            pruning_config,
+            checkpoint_stream,
+        )
+        .await
     }
 
     pub async fn open_with_committee_for_testing(
@@ -83,18 +92,27 @@ impl AuthorityStore {
         // as the genesis committee.
         assert_eq!(committee.epoch, 0);
         let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(path, db_options.clone()));
-        Self::open_inner(genesis, perpetual_tables, committee.clone(), pruning_config).await
+        Self::open_inner(
+            genesis,
+            perpetual_tables,
+            committee.clone(),
+            pruning_config,
+            mpsc::channel(1).1,
+        )
+        .await
     }
 
     async fn open_inner(
         genesis: &Genesis,
         perpetual_tables: Arc<AuthorityPerpetualTables>,
         committee: Committee,
         pruning_config: &AuthorityStorePruningConfig,
+        checkpoint_stream: mpsc::Receiver<CheckpointExecutionMessage>,
     ) -> SuiResult<Self> {
         let epoch = committee.epoch;
 
-        let _store_pruner = AuthorityStorePruner::new(perpetual_tables.clone(), pruning_config);
+        let _store_pruner =
+            AuthorityStorePruner::new(perpetual_tables.clone(), pruning_config, checkpoint_stream);
 
         let store = Self {
             mutex_table: MutexTable::new(NUM_SHARDS, SHARD_SIZE),