Implement Retention Period Support (#592)

* Implement Retention Period support

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

* Replace get_source with get_retention_root

* Set retention period default to 0

* Rename references to source to retention root

* Update retention root advancement to use pruning point samples

* Default retention period point to pruning point only and not including history root

* Move new root check outside

* Rename history root to retention checkpoint

* Use Option for retention_period_days

* Fix get timestamp; change end condition of retention root advance loop

* Show logs for retention period disk usage and require min 3 days if set

* Minor comment fixes

* Consistent naming to retention period root

* consistent comments

* simplify retention root search loop

* important: handle archival upgrade correctly

* daemon check: fix constants and msg

* ergonomic rust style

* keep retention root in place for archives

* update tn10 doc to the upcoming version

---------

Co-authored-by: Michael Sutton <msutton@cs.huji.ac.il>

Author: coderofstuff

components/consensusmanager/src/session.rs

@@ -254,9 +254,9 @@ impl ConsensusSessionOwned {
         self.clone().spawn_blocking(move |c| c.get_current_block_color(hash)).await
     }
 
-    /// source refers to the earliest block from which the current node has full header & block data  
-    pub async fn async_get_source(&self) -> Hash {
-        self.clone().spawn_blocking(|c| c.get_source()).await
+    /// retention period root refers to the earliest block from which the current node has full header & block data  
+    pub async fn async_get_retention_period_root(&self) -> Hash {
+        self.clone().spawn_blocking(|c| c.get_retention_period_root()).await
     }
 
     pub async fn async_estimate_block_count(&self) -> BlockCount {

consensus/core/src/api/mod.rs

@@ -143,8 +143,8 @@ pub trait ConsensusApi: Send + Sync {
         unimplemented!()
     }
 
-    /// source refers to the earliest block from which the current node has full header & block data  
-    fn get_source(&self) -> Hash {
+    /// retention period root refers to the earliest block from which the current node has full header & block data  
+    fn get_retention_period_root(&self) -> Hash {
         unimplemented!()
     }
 

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: Option<f64>,
 }
 
 impl Config {
@@ -95,6 +98,7 @@ impl Config {
             initial_utxo_set: Default::default(),
             disable_upnp: false,
             ram_scale: 1.0,
+            retention_period_days: None,
         }
     }
 

consensus/src/consensus/mod.rs

@@ -307,13 +307,39 @@ impl Consensus {
             is_consensus_exiting,
         };
 
-        // TODO (post HF): remove the upgrade
-        // Database upgrade to include pruning samples
-        this.pruning_samples_database_upgrade();
+        // Run database upgrades if any
+        this.run_database_upgrades();
 
         this
     }
 
+    /// A procedure for calling database upgrades which are self-contained (i.e., do not require knowing the DB version)
+    fn run_database_upgrades(&self) {
+        // Upgrade to initialize the new retention root field correctly
+        self.retention_root_database_upgrade();
+
+        // TODO (post HF): remove this upgrade
+        // Database upgrade to include pruning samples
+        self.pruning_samples_database_upgrade();
+    }
+
+    fn retention_root_database_upgrade(&self) {
+        let mut pruning_point_store = self.pruning_point_store.write();
+        if pruning_point_store.retention_period_root().unwrap_option().is_none() {
+            let mut batch = rocksdb::WriteBatch::default();
+            if self.config.is_archival {
+                // The retention checkpoint is what was previously known as history root
+                let retention_checkpoint = pruning_point_store.retention_checkpoint().unwrap();
+                pruning_point_store.set_retention_period_root(&mut batch, retention_checkpoint).unwrap();
+            } else {
+                // For non-archival nodes the retention root was the pruning point
+                let pruning_point = pruning_point_store.get().unwrap().pruning_point;
+                pruning_point_store.set_retention_period_root(&mut batch, pruning_point).unwrap();
+            }
+            self.db.write(batch).unwrap();
+        }
+    }
+
     fn pruning_samples_database_upgrade(&self) {
         //
         // For the first time this version runs, make sure we populate pruning samples
@@ -565,8 +591,8 @@ impl ConsensusApi for Consensus {
         // Verify the block exists and can be assumed to have relations and reachability data
         self.validate_block_exists(hash).ok()?;
 
-        // Verify that the block is in future(source), where Ghostdag data is complete
-        self.services.reachability_service.is_dag_ancestor_of(self.get_source(), hash).then_some(())?;
+        // Verify that the block is in future(retention root), where Ghostdag data is complete
+        self.services.reachability_service.is_dag_ancestor_of(self.get_retention_period_root(), hash).then_some(())?;
 
         let sink = self.get_sink();
 
@@ -621,31 +647,26 @@ impl ConsensusApi for Consensus {
         self.lkg_virtual_state.load().to_virtual_state_approx_id()
     }
 
-    fn get_source(&self) -> Hash {
-        if self.config.is_archival {
-            // we use the history root in archival cases.
-            return self.pruning_point_store.read().history_root().unwrap();
-        }
-        self.pruning_point_store.read().pruning_point().unwrap()
+    fn get_retention_period_root(&self) -> Hash {
+        self.pruning_point_store.read().retention_period_root().unwrap()
     }
 
-    /// Estimates number of blocks and headers stored in the node
+    /// Estimates the number of blocks and headers stored in the node database.
     ///
-    /// This is an estimation based on the daa score difference between the node's `source` and `sink`'s daa score,
+    /// This is an estimation based on the DAA score difference between the node's `retention root` and `virtual`'s DAA score,
     /// as such, it does not include non-daa blocks, and does not include headers stored as part of the pruning proof.  
     fn estimate_block_count(&self) -> BlockCount {
-        // PRUNE SAFETY: node is either archival or source is the pruning point which its header is kept permanently
-        let source_score = self.headers_store.get_compact_header_data(self.get_source()).unwrap().daa_score;
+        // PRUNE SAFETY: retention root is always a current or past pruning point which its header is kept permanently
+        let retention_period_root_score = self.headers_store.get_daa_score(self.get_retention_period_root()).unwrap();
         let virtual_score = self.get_virtual_daa_score();
         let header_count = self
             .headers_store
-            .get_compact_header_data(self.get_headers_selected_tip())
+            .get_daa_score(self.get_headers_selected_tip())
             .unwrap_option()
-            .map(|h| h.daa_score)
             .unwrap_or(virtual_score)
             .max(virtual_score)
-            - source_score;
-        let block_count = virtual_score - source_score;
+            - retention_period_root_score;
+        let block_count = virtual_score - retention_period_root_score;
         BlockCount { header_count, block_count }
     }
 
@@ -668,12 +689,12 @@ impl ConsensusApi for Consensus {
         // Verify that the block exists
         self.validate_block_exists(low)?;
 
-        // Verify that source is on chain(block)
+        // Verify that retention root is on chain(block)
         self.services
             .reachability_service
-            .is_chain_ancestor_of(self.get_source(), low)
+            .is_chain_ancestor_of(self.get_retention_period_root(), low)
             .then_some(())
-            .ok_or(ConsensusError::General("the queried hash does not have source on its chain"))?;
+            .ok_or(ConsensusError::General("the queried hash does not have retention root on its chain"))?;
 
         Ok(self.services.dag_traversal_manager.calculate_chain_path(low, self.get_sink(), chain_path_added_limit))
     }
@@ -748,7 +769,7 @@ impl ConsensusApi for Consensus {
     fn get_populated_transaction(&self, txid: Hash, accepting_block_daa_score: u64) -> Result<SignableTransaction, UtxoInquirerError> {
         // We need consistency between the pruning_point_store, utxo_diffs_store, block_transactions_store, selected chain and headers store reads
         let _guard = self.pruning_lock.blocking_read();
-        self.virtual_processor.get_populated_transaction(txid, accepting_block_daa_score, self.get_source())
+        self.virtual_processor.get_populated_transaction(txid, accepting_block_daa_score, self.get_retention_period_root())
     }
 
     fn get_virtual_parents(&self) -> BlockHashSet {

consensus/src/model/stores/pruning.rs

@@ -39,9 +39,16 @@ pub trait PruningStoreReader {
     fn get(&self) -> StoreResult<PruningPointInfo>;
 
     /// Represent the point after which data is fully held (i.e., history is consecutive from this point and up to virtual).
-    /// 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>;
+    /// This is usually a pruning point that is at or below the retention period requirement (and for archival
+    /// nodes it will remain the initial syncing point or the last pruning point before turning to an archive).
+    /// At every pruning point movement, this is adjusted to the next pruning point sample that satisfies the required
+    /// retention period.
+    fn retention_period_root(&self) -> StoreResult<Hash>;
+
+    // During pruning, this is a reference to the retention root before the pruning point move.
+    // After pruning, this is updated to point to the retention period root.
+    // This checkpoint is used to determine if pruning has successfully completed.
+    fn retention_checkpoint(&self) -> StoreResult<Hash>;
 }
 
 pub trait PruningStore: PruningStoreReader {
@@ -53,15 +60,17 @@ pub trait PruningStore: PruningStoreReader {
 pub struct DbPruningStore {
     db: Arc<DB>,
     access: CachedDbItem<PruningPointInfo>,
-    history_root_access: CachedDbItem<Hash>,
+    retention_checkpoint_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()),
+            retention_checkpoint_access: CachedDbItem::new(db.clone(), DatabaseStorePrefixes::RetentionCheckpoint.into()),
+            retention_period_root_access: CachedDbItem::new(db, DatabaseStorePrefixes::RetentionPeriodRoot.into()),
         }
     }
 
@@ -73,8 +82,12 @@ impl DbPruningStore {
         self.access.write(BatchDbWriter::new(batch), &PruningPointInfo { pruning_point, candidate, index })
     }
 
-    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_checkpoint(&mut self, batch: &mut WriteBatch, retention_checkpoint: Hash) -> StoreResult<()> {
+        self.retention_checkpoint_access.write(BatchDbWriter::new(batch), &retention_checkpoint)
+    }
+
+    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)
     }
 }
 
@@ -95,8 +108,12 @@ impl PruningStoreReader for DbPruningStore {
         self.access.read()
     }
 
-    fn history_root(&self) -> StoreResult<Hash> {
-        self.history_root_access.read()
+    fn retention_checkpoint(&self) -> StoreResult<Hash> {
+        self.retention_checkpoint_access.read()
+    }
+
+    fn retention_period_root(&self) -> StoreResult<Hash> {
+        self.retention_period_root_access.read()
     }
 }