Implement Retention Period support

- new cli flag retention_period_days allows the user to retain more days of data

Author: coderofstuff

consensus/core/src/config/mod.rs

@@ -68,6 +68,9 @@ pub struct Config {
 
     /// A scale factor to apply to memory allocation bounds
     pub ram_scale: f64,
+
+    /// The number of days to keep data for
+    pub retention_period_days: f64,
 }
 
 impl Config {
@@ -95,6 +98,7 @@ impl Config {
             initial_utxo_set: Default::default(),
             disable_upnp: false,
             ram_scale: 1.0,
+            retention_period_days: 3.0,
         }
     }
 

consensus/src/model/stores/pruning.rs

@@ -42,6 +42,7 @@ pub trait PruningStoreReader {
     /// This is usually the pruning point, though it might lag a bit behind until data prune completes (and for archival
     /// nodes it will remain the initial syncing point or the last pruning point before turning to an archive)
     fn history_root(&self) -> StoreResult<Hash>;
+    fn retention_period_root(&self) -> StoreResult<Hash>;
 }
 
 pub trait PruningStore: PruningStoreReader {
@@ -54,14 +55,16 @@ pub struct DbPruningStore {
     db: Arc<DB>,
     access: CachedDbItem<PruningPointInfo>,
     history_root_access: CachedDbItem<Hash>,
+    retention_period_root_access: CachedDbItem<Hash>,
 }
 
 impl DbPruningStore {
     pub fn new(db: Arc<DB>) -> Self {
         Self {
             db: Arc::clone(&db),
             access: CachedDbItem::new(db.clone(), DatabaseStorePrefixes::PruningPoint.into()),
-            history_root_access: CachedDbItem::new(db, DatabaseStorePrefixes::HistoryRoot.into()),
+            history_root_access: CachedDbItem::new(db.clone(), DatabaseStorePrefixes::HistoryRoot.into()),
+            retention_period_root_access: CachedDbItem::new(db, DatabaseStorePrefixes::RetentionPeriodRoot.into()),
         }
     }
 
@@ -76,6 +79,10 @@ impl DbPruningStore {
     pub fn set_history_root(&mut self, batch: &mut WriteBatch, history_root: Hash) -> StoreResult<()> {
         self.history_root_access.write(BatchDbWriter::new(batch), &history_root)
     }
+
+    pub fn set_retention_period_root(&mut self, batch: &mut WriteBatch, retention_period_root: Hash) -> StoreResult<()> {
+        self.retention_period_root_access.write(BatchDbWriter::new(batch), &retention_period_root)
+    }
 }
 
 impl PruningStoreReader for DbPruningStore {
@@ -98,6 +105,10 @@ impl PruningStoreReader for DbPruningStore {
     fn history_root(&self) -> StoreResult<Hash> {
         self.history_root_access.read()
     }
+
+    fn retention_period_root(&self) -> StoreResult<Hash> {
+        self.retention_period_root_access.read()
+    }
 }
 
 impl PruningStore for DbPruningStore {

consensus/src/pipeline/pruning_processor/processor.rs

@@ -34,7 +34,7 @@ use kaspa_consensus_core::{
     BlockHashMap, BlockHashSet, BlockLevel,
 };
 use kaspa_consensusmanager::SessionLock;
-use kaspa_core::{debug, info, warn};
+use kaspa_core::{debug, info, time::unix_now, trace, warn};
 use kaspa_database::prelude::{BatchDbWriter, MemoryWriter, StoreResultExtensions, DB};
 use kaspa_hashes::Hash;
 use kaspa_muhash::MuHash;
@@ -134,6 +134,7 @@ impl PruningProcessor {
         let pruning_point_read = self.pruning_point_store.read();
         let pruning_point = pruning_point_read.pruning_point().unwrap();
         let history_root = pruning_point_read.history_root().unwrap_option();
+        let retention_period_root = pruning_point_read.retention_period_root().unwrap_or(pruning_point);
         let pruning_utxoset_position = self.pruning_utxoset_stores.read().utxoset_position().unwrap_option();
         drop(pruning_point_read);
 
@@ -153,11 +154,17 @@ impl PruningProcessor {
             }
         }
 
+        trace!(
+            "history_root: {:?} | retention_period_root: {} | pruning_point: {}",
+            history_root,
+            retention_period_root,
+            pruning_point
+        );
         if let Some(history_root) = history_root {
             // This indicates the node crashed or was forced to stop during a former data prune operation hence
             // we need to complete it
-            if history_root != pruning_point {
-                self.prune(pruning_point);
+            if history_root != retention_period_root {
+                self.prune(pruning_point, retention_period_root);
             }
         }
 
@@ -175,6 +182,15 @@ impl PruningProcessor {
         );
 
         if !new_pruning_points.is_empty() {
+            let retention_period_root = if let Ok(retention_period_root) = pruning_point_read.retention_period_root() {
+                retention_period_root
+            } else if let Ok(history_root) = pruning_point_read.history_root() {
+                history_root
+            } else {
+                // pruning point always exist
+                pruning_point_read.pruning_point().unwrap()
+            };
+
             // Update past pruning points and pruning point stores
             let mut batch = WriteBatch::default();
             let mut pruning_point_write = RwLockUpgradableReadGuard::upgrade(pruning_point_read);
@@ -183,10 +199,14 @@ impl PruningProcessor {
             }
             let new_pp_index = current_pruning_info.index + new_pruning_points.len() as u64;
             let new_pruning_point = *new_pruning_points.last().unwrap();
+            let adjusted_retention_period_root = self.advance_retention_period_root(retention_period_root, new_pruning_point);
             pruning_point_write.set_batch(&mut batch, new_pruning_point, new_candidate, new_pp_index).unwrap();
+            pruning_point_write.set_retention_period_root(&mut batch, adjusted_retention_period_root).unwrap();
             self.db.write(batch).unwrap();
             drop(pruning_point_write);
 
+            trace!("New Pruning Point: {} | New Retention Period Root: {}", new_pruning_point, adjusted_retention_period_root);
+
             // Inform the user
             info!("Periodic pruning point movement: advancing from {} to {}", current_pruning_info.pruning_point, new_pruning_point);
 
@@ -198,7 +218,7 @@ impl PruningProcessor {
             info!("Updated the pruning point UTXO set");
 
             // Finally, prune data in the new pruning point past
-            self.prune(new_pruning_point);
+            self.prune(new_pruning_point, adjusted_retention_period_root);
         } else if new_candidate != current_pruning_info.candidate {
             let mut pruning_point_write = RwLockUpgradableReadGuard::upgrade(pruning_point_read);
             pruning_point_write.set(current_pruning_info.pruning_point, new_candidate, current_pruning_info.index).unwrap();
@@ -238,7 +258,7 @@ impl PruningProcessor {
         info!("Pruning point UTXO commitment was verified correctly (sanity test)");
     }
 
-    fn prune(&self, new_pruning_point: Hash) {
+    fn prune(&self, new_pruning_point: Hash, retention_period_root: Hash) {
         if self.config.is_archival {
             warn!("The node is configured as an archival node -- avoiding data pruning. Note this might lead to heavy disk usage.");
             return;
@@ -384,7 +404,7 @@ impl PruningProcessor {
         let (mut counter, mut traversed) = (0, 0);
         info!("Header and Block pruning: starting traversal from: {} (genesis: {})", queue.iter().reusable_format(", "), genesis);
         while let Some(current) = queue.pop_front() {
-            if reachability_read.is_dag_ancestor_of_result(new_pruning_point, current).unwrap() {
+            if reachability_read.is_dag_ancestor_of_result(retention_period_root, current).unwrap() {
                 continue;
             }
             traversed += 1;
@@ -517,12 +537,47 @@ impl PruningProcessor {
             // Set the history root to the new pruning point only after we successfully pruned its past
             let mut pruning_point_write = self.pruning_point_store.write();
             let mut batch = WriteBatch::default();
-            pruning_point_write.set_history_root(&mut batch, new_pruning_point).unwrap();
+            pruning_point_write.set_history_root(&mut batch, retention_period_root).unwrap();
             self.db.write(batch).unwrap();
             drop(pruning_point_write);
         }
     }
 
+    /// Adjusts the retention period root forward until the maximum chain block is reached that covers the retention period.
+    /// This is the last chain block B such that B.timestamp < retention_period_days_ago. This may return the old hash if
+    /// the retention period cannot be covered yet with the node's current history.
+    ///
+    /// This function is expected to be called only when a new pruning point is determined and right before
+    /// doing any pruning.
+    ///
+    /// retention_period_root is guaranteed to be in the past(pruning_point)
+    fn advance_retention_period_root(&self, retention_period_root: Hash, pruning_point: Hash) -> Hash {
+        // The retention period in milliseconds we need to cover
+        let retention_period_ms = (self.config.retention_period_days * 86400.0 * 1000.0).ceil() as u64;
+
+        let retention_period_root_ts_target = unix_now().saturating_sub(retention_period_ms);
+        let mut new_retention_period_root = retention_period_root;
+
+        trace!(
+            "Adjusting the retention period root to cover the required retention period. Target timestamp: {}",
+            retention_period_root_ts_target,
+        );
+
+        for block in self.reachability_service.forward_chain_iterator(retention_period_root, pruning_point, true) {
+            let timestamp = self.headers_store.get_compact_header_data(block).unwrap().timestamp;
+            trace!("block | timestamp = {} | {}", block, timestamp);
+            if timestamp >= retention_period_root_ts_target {
+                trace!("block {} timestamp {} >= {}", block, timestamp, retention_period_root_ts_target);
+                // We are now at a chain block that is at or above our retention period target
+                // Break here so we can return the hash from the previous iteration
+                break;
+            }
+            new_retention_period_root = block;
+        }
+
+        new_retention_period_root
+    }
+
     fn past_pruning_points(&self) -> BlockHashSet {
         (0..self.pruning_point_store.read().get().unwrap().index)
             .map(|index| self.past_pruning_points_store.get(index).unwrap())

consensus/src/pipeline/virtual_processor/processor.rs

@@ -1081,6 +1081,7 @@ impl VirtualStateProcessor {
             self.past_pruning_points_store.insert_batch(&mut batch, 0, self.genesis.hash).unwrap_or_exists();
             pruning_point_write.set_batch(&mut batch, self.genesis.hash, self.genesis.hash, 0).unwrap();
             pruning_point_write.set_history_root(&mut batch, self.genesis.hash).unwrap();
+            pruning_point_write.set_retention_period_root(&mut batch, self.genesis.hash).unwrap();
             pruning_utxoset_write.set_utxoset_position(&mut batch, self.genesis.hash).unwrap();
             self.db.write(batch).unwrap();
             drop(pruning_point_write);

consensus/src/processes/pruning_proof/mod.rs

@@ -201,6 +201,7 @@ impl PruningProofManager {
         let mut batch = WriteBatch::default();
         pruning_point_write.set_batch(&mut batch, new_pruning_point, new_pruning_point, (pruning_points.len() - 1) as u64).unwrap();
         pruning_point_write.set_history_root(&mut batch, new_pruning_point).unwrap();
+        pruning_point_write.set_retention_period_root(&mut batch, new_pruning_point).unwrap();
         self.db.write(batch).unwrap();
         drop(pruning_point_write);
     }

database/src/registry.rs

@@ -45,6 +45,9 @@ pub enum DatabaseStorePrefixes {
     TempGhostdag = 40,
     TempGhostdagCompact = 41,
 
+    // ---- Retention Period Root ----
+    RetentionPeriodRoot = 50,
+
     // ---- Metadata ----
     MultiConsensusMetadata = 124,
     ConsensusEntries = 125,

kaspad/src/args.rs

@@ -90,6 +90,7 @@ pub struct Args {
     #[serde(rename = "nogrpc")]
     pub disable_grpc: bool,
     pub ram_scale: f64,
+    pub retention_period_days: f64,
 }
 
 impl Default for Args {
@@ -140,6 +141,7 @@ impl Default for Args {
             disable_dns_seeding: false,
             disable_grpc: false,
             ram_scale: 1.0,
+            retention_period_days: 3.0,
         }
     }
 }
@@ -159,6 +161,7 @@ impl Args {
         config.p2p_listen_address = self.listen.unwrap_or(ContextualNetAddress::unspecified());
         config.externalip = self.externalip.map(|v| v.normalize(config.default_p2p_port()));
         config.ram_scale = self.ram_scale;
+        config.retention_period_days = self.retention_period_days;
 
         #[cfg(feature = "devnet-prealloc")]
         if let Some(num_prealloc_utxos) = self.num_prealloc_utxos {
@@ -369,6 +372,13 @@ Setting to 0 prevents the preallocation and sets the maximum to {}, leading to 0
                 .help("Apply a scale factor to memory allocation bounds. Nodes with limited RAM (~4-8GB) should set this to ~0.3-0.5 respectively. Nodes with
 a large RAM (~64GB) can set this value to ~3.0-4.0 and gain superior performance especially for syncing peers faster"),
         )
+        .arg(
+            Arg::new("retention-period-days")
+                .long("retention-period-days")
+                .require_equals(true)
+                .value_parser(clap::value_parser!(f64))
+                .help("The number of total days of data to keep.")
+        )
         ;
 
     #[cfg(feature = "devnet-prealloc")]
@@ -448,6 +458,7 @@ impl Args {
             disable_dns_seeding: arg_match_unwrap_or::<bool>(&m, "nodnsseed", defaults.disable_dns_seeding),
             disable_grpc: arg_match_unwrap_or::<bool>(&m, "nogrpc", defaults.disable_grpc),
             ram_scale: arg_match_unwrap_or::<f64>(&m, "ram-scale", defaults.ram_scale),
+            retention_period_days: arg_match_unwrap_or::<f64>(&m, "retention-period-days", defaults.retention_period_days),
 
             #[cfg(feature = "devnet-prealloc")]
             num_prealloc_utxos: m.get_one::<u64>("num-prealloc-utxos").cloned(),