Retain index data up to retention period root (kaspanet#657)

* test: add test for getting populated tx

- Includes retention_period_days support in simpa

* Retain index up to retention root

- instead of pruning point

* Use get_daa_score instead of compact_header

* Use sink timestamp as "now" in all cases for retention

Author: coderofstuff

consensus/src/pipeline/pruning_processor/processor.rs

@@ -35,7 +35,7 @@ use kaspa_consensus_core::{
     BlockHashMap, BlockHashSet, BlockLevel,
 };
 use kaspa_consensusmanager::SessionLock;
-use kaspa_core::{debug, info, time::unix_now, trace, warn};
+use kaspa_core::{debug, info, trace, warn};
 use kaspa_database::prelude::{BatchDbWriter, MemoryWriter, StoreResultExtensions, DB};
 use kaspa_hashes::Hash;
 use kaspa_muhash::MuHash;
@@ -383,7 +383,7 @@ impl PruningProcessor {
 
             // Prune the selected chain index below the pruning point
             let mut selected_chain_write = self.selected_chain_store.write();
-            selected_chain_write.prune_below_pruning_point(BatchDbWriter::new(&mut batch), new_pruning_point).unwrap();
+            selected_chain_write.prune_below_pruning_point(BatchDbWriter::new(&mut batch), retention_period_root).unwrap();
 
             // Flush the batch to the DB
             self.db.write(batch).unwrap();
@@ -563,7 +563,9 @@ impl PruningProcessor {
                 let retention_period_ms = (retention_period_days * 86400.0 * 1000.0).ceil() as u64;
 
                 // The target timestamp we would like to find a point below
-                let retention_period_root_ts_target = unix_now().saturating_sub(retention_period_ms);
+                let sink_timestamp_as_current_time =
+                    self.headers_store.get_timestamp(self.lkg_virtual_state.load().ghostdag_data.selected_parent).unwrap();
+                let retention_period_root_ts_target = sink_timestamp_as_current_time.saturating_sub(retention_period_ms);
 
                 // Iterate from the new pruning point to the prev retention root and search for the first point with enough days above it.
                 // Note that prev retention root is always a past pruning point, so we can iterate via pruning samples until we reach it.

consensus/src/pipeline/virtual_processor/utxo_inquirer.rs

@@ -28,8 +28,7 @@ impl VirtualStateProcessor {
     ) -> Result<SignableTransaction, UtxoInquirerError> {
         let retention_period_root_daa_score = self
             .headers_store
-            .get_compact_header_data(retention_period_root_hash)
-            .map(|compact_header| compact_header.daa_score)
+            .get_daa_score(retention_period_root_hash)
             .map_err(|_| UtxoInquirerError::MissingCompactHeaderForBlockHash(retention_period_root_hash))?;
 
         if accepting_block_daa_score < retention_period_root_daa_score {
@@ -94,11 +93,8 @@ impl VirtualStateProcessor {
             .get_by_hash(retention_period_root_hash)
             .map_err(|_| UtxoInquirerError::MissingIndexForHash(retention_period_root_hash))?;
         let (tip_index, tip_hash) = sc_read.get_tip().map_err(|_| UtxoInquirerError::MissingTipData)?;
-        let tip_daa_score = self
-            .headers_store
-            .get_compact_header_data(tip_hash)
-            .map(|tip| tip.daa_score)
-            .map_err(|_| UtxoInquirerError::MissingCompactHeaderForBlockHash(tip_hash))?;
+        let tip_daa_score =
+            self.headers_store.get_daa_score(tip_hash).map_err(|_| UtxoInquirerError::MissingCompactHeaderForBlockHash(tip_hash))?;
 
         // For a chain segment it holds that len(segment) <= daa_score(segment end) - daa_score(segment start). This is true
         // because each chain block increases the daa score by at least one. Hence we can lower bound our search by high index
@@ -117,14 +113,14 @@ impl VirtualStateProcessor {
                 UtxoInquirerError::MissingHashAtIndex(mid)
             })?;
 
-            // 2. Get the compact header so we have access to the daa_score. Error if we cannot find the header
-            let compact_header = self.headers_store.get_compact_header_data(hash).map_err(|_| {
-                trace!("Did not find a compact header with hash {}", hash);
+            // 2. Get the daa_score. Error if the header is not found
+            let daa_score = self.headers_store.get_daa_score(hash).map_err(|_| {
+                trace!("Did not find a header with hash {}", hash);
                 UtxoInquirerError::MissingCompactHeaderForBlockHash(hash)
             })?;
 
             // 3. Compare block daa score to our target
-            match compact_header.daa_score.cmp(&target_daa_score) {
+            match daa_score.cmp(&target_daa_score) {
                 cmp::Ordering::Equal => {
                     // We found the chain block we need
                     break hash;

simpa/src/main.rs

@@ -124,6 +124,8 @@ struct Args {
     rocksdb_mem_budget: Option<usize>,
     #[arg(long, default_value_t = false)]
     long_payload: bool,
+    #[arg(long)]
+    retention_period_days: Option<f64>,
 }
 
 #[cfg(feature = "heap")]
@@ -198,7 +200,10 @@ fn main_impl(mut args: Args) {
         .apply_args(|config| apply_args_to_consensus_params(&args, &mut config.params))
         .apply_args(|config| apply_args_to_perf_params(&args, &mut config.perf))
         .adjust_perf_params_to_consensus_params()
-        .apply_args(|config| config.ram_scale = args.ram_scale)
+        .apply_args(|config| {
+            config.ram_scale = args.ram_scale;
+            config.retention_period_days = args.retention_period_days;
+        })
         .skip_proof_of_work()
         .enable_sanity_checks();
     if !args.test_pruning {
@@ -260,7 +265,36 @@ fn main_impl(mut args: Args) {
         let num_blocks = hashes.len();
         let num_txs = print_stats(&consensus, &hashes, args.delay, args.bps, config.ghostdag_k().before());
         info!("There are {num_blocks} blocks with {num_txs} transactions overall above the current pruning point");
+
+        if args.retention_period_days.is_some() {
+            let hashes_retention = topologically_ordered_hashes(&consensus, consensus.get_retention_period_root());
+            info!("There are {} blocks above the retention period root", hashes_retention.len());
+        }
+
         consensus.validate_pruning_points(consensus.get_sink()).unwrap();
+
+        // Test whether we can still retrieve a populated transaction given a txid and the accepting block daa score.
+        for hash in hashes.iter().cloned() {
+            if !consensus.is_chain_block(hash).unwrap() {
+                // only chain blocks are worth checking the acceptance data of
+                continue;
+            }
+
+            if let Ok(block_acceptance_data) = consensus.get_block_acceptance_data(hash) {
+                block_acceptance_data.iter().for_each(|cbad| {
+                    let block = consensus.get_block(hash).unwrap();
+                    cbad.accepted_transactions.iter().for_each(|ate| {
+                        assert!(
+                            consensus.get_populated_transaction(ate.transaction_id, block.header.daa_score).is_ok(),
+                            "Expected to find find tx {} at accepted daa {} via get_populated_transaction",
+                            ate.transaction_id,
+                            block.header.daa_score
+                        );
+                    });
+                });
+            }
+        }
+
         drop(consensus);
         return;
     }