[Storage][Pruner] Use ledger pruner config for state kv pruner. (#7374)

config/src/config/storage_config.rs

@@ -117,11 +117,6 @@ pub const NO_OP_STORAGE_PRUNER_CONFIG: PrunerConfig = PrunerConfig {
         prune_window: 0,
         batch_size: 0,
     },
-    state_kv_pruner_config: StateKvPrunerConfig {
-        enable: false,
-        prune_window: 0,
-        batch_size: 0,
-    },
 };
 
 #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
@@ -165,19 +160,6 @@ pub struct EpochSnapshotPrunerConfig {
     pub batch_size: usize,
 }
 
-#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct StateKvPrunerConfig {
-    /// Boolean to enable/disable the state kv pruner. The state pruner is responsible for
-    /// pruning state tree nodes.
-    pub enable: bool,
-    /// Window size in versions.
-    pub prune_window: u64,
-    /// Similar to the variable above but for state kv pruner. It means the number of versions to
-    /// prune a time.
-    pub batch_size: usize,
-}
-
 // Config for the epoch ending state pruner is actually in the same format as the state merkle
 // pruner, but it has it's own type hence separate default values. This converts it to the same
 // type, to use the same pruner implementation (but parameterized on the stale node index DB schema).
@@ -197,7 +179,6 @@ pub struct PrunerConfig {
     pub ledger_pruner_config: LedgerPrunerConfig,
     pub state_merkle_pruner_config: StateMerklePrunerConfig,
     pub epoch_snapshot_pruner_config: EpochSnapshotPrunerConfig,
-    pub state_kv_pruner_config: StateKvPrunerConfig,
 }
 
 impl Default for LedgerPrunerConfig {
@@ -248,17 +229,6 @@ impl Default for EpochSnapshotPrunerConfig {
     }
 }
 
-impl Default for StateKvPrunerConfig {
-    fn default() -> Self {
-        Self {
-            // TODO(grao): Keep it the same as ledger pruner config for now, will revisit later.
-            enable: true,
-            prune_window: 150_000_000,
-            batch_size: 500,
-        }
-    }
-}
-
 impl Default for StorageConfig {
     fn default() -> StorageConfig {
         StorageConfig {

execution/executor-benchmark/src/main.rs

@@ -3,8 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use aptos_config::config::{
-    EpochSnapshotPrunerConfig, LedgerPrunerConfig, PrunerConfig, StateKvPrunerConfig,
-    StateMerklePrunerConfig,
+    EpochSnapshotPrunerConfig, LedgerPrunerConfig, PrunerConfig, StateMerklePrunerConfig,
 };
 use aptos_executor::block_executor::TransactionBlockExecutor;
 use aptos_executor_benchmark::{
@@ -40,9 +39,6 @@ struct PrunerOpt {
     #[structopt(long)]
     enable_ledger_pruner: bool,
 
-    #[structopt(long)]
-    enable_state_kv_pruner: bool,
-
     #[structopt(long, default_value = "100000")]
     state_prune_window: u64,
 
@@ -52,9 +48,6 @@ struct PrunerOpt {
     #[structopt(long, default_value = "100000")]
     ledger_prune_window: u64,
 
-    #[structopt(long, default_value = "100000")]
-    state_kv_prune_window: u64,
-
     #[structopt(long, default_value = "500")]
     ledger_pruning_batch_size: usize,
 
@@ -63,9 +56,6 @@ struct PrunerOpt {
 
     #[structopt(long, default_value = "500")]
     epoch_snapshot_pruning_batch_size: usize,
-
-    #[structopt(long, default_value = "500")]
-    state_kv_pruning_batch_size: usize,
 }
 
 impl PrunerOpt {
@@ -87,11 +77,6 @@ impl PrunerOpt {
                 batch_size: self.ledger_pruning_batch_size,
                 user_pruning_window_offset: 0,
             },
-            state_kv_pruner_config: StateKvPrunerConfig {
-                enable: self.enable_state_kv_pruner,
-                prune_window: self.state_kv_prune_window,
-                batch_size: self.state_kv_pruning_batch_size,
-            },
         }
     }
 }

storage/aptosdb/src/aptosdb_test.rs

@@ -13,7 +13,7 @@ use crate::{
 };
 use aptos_config::config::{
     EpochSnapshotPrunerConfig, LedgerPrunerConfig, PrunerConfig, RocksdbConfigs,
-    StateKvPrunerConfig, StateMerklePrunerConfig, BUFFERED_STATE_TARGET_ITEMS,
+    StateMerklePrunerConfig, BUFFERED_STATE_TARGET_ITEMS,
     DEFAULT_MAX_NUM_NODES_PER_LRU_CACHE_SHARD,
 };
 use aptos_crypto::{hash::CryptoHash, HashValue};
@@ -200,11 +200,6 @@ pub fn test_state_merkle_pruning_impl(
                 prune_window: 10,
                 batch_size: 1,
             },
-            state_kv_pruner_config: StateKvPrunerConfig {
-                enable: true,
-                prune_window: 10,
-                batch_size: 1,
-            },
         },
         RocksdbConfigs::default(),
         false, /* enable_indexer */

storage/aptosdb/src/lib.rs

@@ -305,10 +305,8 @@ impl AptosDB {
             Arc::clone(&arc_state_merkle_rocksdb),
             pruner_config.epoch_snapshot_pruner_config.into(),
         );
-        let state_kv_pruner = StateKvPrunerManager::new(
-            Arc::clone(&state_kv_db),
-            pruner_config.state_kv_pruner_config,
-        );
+        let state_kv_pruner =
+            StateKvPrunerManager::new(Arc::clone(&state_kv_db), pruner_config.ledger_pruner_config);
         let state_store = Arc::new(StateStore::new(
             Arc::clone(&ledger_rocksdb),
             Arc::clone(&arc_state_merkle_rocksdb),

storage/aptosdb/src/pruner/state_kv_pruner_manager.rs

@@ -10,7 +10,7 @@ use crate::{
     pruner_utils,
     state_kv_db::StateKvDb,
 };
-use aptos_config::config::StateKvPrunerConfig;
+use aptos_config::config::LedgerPrunerConfig;
 use aptos_infallible::Mutex;
 use aptos_types::transaction::Version;
 use std::{sync::Arc, thread::JoinHandle};
@@ -85,7 +85,7 @@ impl PrunerManager for StateKvPrunerManager {
 
 impl StateKvPrunerManager {
     /// Creates a worker thread that waits on a channel for pruning commands.
-    pub fn new(state_kv_db: Arc<StateKvDb>, state_kv_pruner_config: StateKvPrunerConfig) -> Self {
+    pub fn new(state_kv_db: Arc<StateKvDb>, state_kv_pruner_config: LedgerPrunerConfig) -> Self {
         let state_kv_pruner = pruner_utils::create_state_kv_pruner(state_kv_db);
 
         if state_kv_pruner_config.enable {

storage/aptosdb/src/pruner/state_kv_pruner_worker.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::pruner::{db_pruner::DBPruner, state_kv_pruner::StateKvPruner};
-use aptos_config::config::StateKvPrunerConfig;
+use aptos_config::config::LedgerPrunerConfig;
 use aptos_logger::{
     error,
     prelude::{sample, SampleRate},
@@ -34,7 +34,7 @@ pub struct StateKvPrunerWorker {
 impl StateKvPrunerWorker {
     pub(crate) fn new(
         state_kv_pruner: Arc<StateKvPruner>,
-        state_kv_pruner_config: StateKvPrunerConfig,
+        state_kv_pruner_config: LedgerPrunerConfig,
     ) -> Self {
         Self {
             pruning_time_interval_in_ms: if cfg!(test) { 100 } else { 1 },

storage/aptosdb/src/pruner/state_store/test.rs

@@ -10,7 +10,7 @@ use crate::{
     test_helper::{arb_state_kv_sets, update_store},
     AptosDB, PrunerManager, StateKvPrunerManager, StateMerklePrunerManager,
 };
-use aptos_config::config::{StateKvPrunerConfig, StateMerklePrunerConfig};
+use aptos_config::config::{LedgerPrunerConfig, StateMerklePrunerConfig};
 use aptos_crypto::HashValue;
 use aptos_schemadb::{ReadOptions, SchemaBatch, DB};
 use aptos_storage_interface::{jmt_update_refs, jmt_updates, DbReader};
@@ -379,10 +379,11 @@ fn verify_state_value_pruner(inputs: Vec<Vec<(StateKey, Option<StateValue>)>>) {
 
     let mut version = 0;
     let mut current_state_values = HashMap::new();
-    let pruner = StateKvPrunerManager::new(Arc::clone(&db.state_kv_db), StateKvPrunerConfig {
+    let pruner = StateKvPrunerManager::new(Arc::clone(&db.state_kv_db), LedgerPrunerConfig {
         enable: true,
         prune_window: 0,
         batch_size: 1,
+        user_pruning_window_offset: 0,
     });
     for batch in inputs {
         update_store(store, batch.clone().into_iter(), version);

storage/aptosdb/src/state_store/mod.rs

@@ -412,7 +412,7 @@ impl StateStore {
         let state_merkle_db = Arc::new(StateMerkleDb::new(arc_state_merkle_rocksdb, 0));
         let state_kv_pruner = StateKvPrunerManager::new(
             Arc::clone(&state_kv_db),
-            NO_OP_STORAGE_PRUNER_CONFIG.state_kv_pruner_config,
+            NO_OP_STORAGE_PRUNER_CONFIG.ledger_pruner_config,
         );
         let state_db = Arc::new(StateDb {
             ledger_db,