use crate::{
consensus::{
services::{ConsensusServices, DbGhostdagManager, DbParentsManager, DbPruningPointManager},
storage::ConsensusStorage,
},
model::{
services::reachability::{MTReachabilityService, ReachabilityService},
stores::{
ghostdag::{CompactGhostdagData, GhostdagStoreReader},
headers::HeaderStoreReader,
past_pruning_points::PastPruningPointsStoreReader,
pruning::{PruningStore, PruningStoreReader},
reachability::{DbReachabilityStore, ReachabilityStoreReader, StagingReachabilityStore},
relations::StagingRelationsStore,
selected_chain::SelectedChainStore,
statuses::StatusesStoreReader,
tips::{TipsStore, TipsStoreReader},
utxo_diffs::UtxoDiffsStoreReader,
},
},
processes::{pruning_proof::PruningProofManager, reachability::inquirer as reachability, relations},
};
use crossbeam_channel::Receiver as CrossbeamReceiver;
use itertools::Itertools;
use kaspa_consensus_core::{
blockhash::ORIGIN,
blockstatus::BlockStatus::StatusHeaderOnly,
config::Config,
muhash::MuHashExtensions,
pruning::{PruningPointProof, PruningPointTrustedData},
trusted::ExternalGhostdagData,
BlockHashMap, BlockHashSet, BlockLevel,
};
use kaspa_consensusmanager::SessionLock;
use kaspa_core::{debug, info, warn};
use kaspa_database::prelude::{BatchDbWriter, MemoryWriter, StoreResultExtensions, DB};
use kaspa_hashes::Hash;
use kaspa_muhash::MuHash;
use kaspa_utils::iter::IterExtensions;
use parking_lot::RwLockUpgradableReadGuard;
use rocksdb::WriteBatch;
use std::{
collections::{hash_map::Entry::Vacant, VecDeque},
ops::Deref,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
time::{Duration, Instant},
};
pub enum PruningProcessingMessage {
Exit,
Process { sink_ghostdag_data: CompactGhostdagData },
}
pub struct PruningProcessor {
receiver: CrossbeamReceiver<PruningProcessingMessage>,
db: Arc<DB>,
storage: Arc<ConsensusStorage>,
reachability_service: MTReachabilityService<DbReachabilityStore>,
ghostdag_managers: Arc<Vec<DbGhostdagManager>>,
pruning_point_manager: DbPruningPointManager,
pruning_proof_manager: Arc<PruningProofManager>,
parents_manager: DbParentsManager,
pruning_lock: SessionLock,
config: Arc<Config>,
is_consensus_exiting: Arc<AtomicBool>,
}
impl Deref for PruningProcessor {
type Target = ConsensusStorage;
fn deref(&self) -> &Self::Target {
&self.storage
}
}
impl PruningProcessor {
pub fn new(
receiver: CrossbeamReceiver<PruningProcessingMessage>,
db: Arc<DB>,
storage: &Arc<ConsensusStorage>,
services: &Arc<ConsensusServices>,
pruning_lock: SessionLock,
config: Arc<Config>,
is_consensus_exiting: Arc<AtomicBool>,
) -> Self {
Self {
receiver,
db,
storage: storage.clone(),
reachability_service: services.reachability_service.clone(),
ghostdag_managers: services.ghostdag_managers.clone(),
pruning_point_manager: services.pruning_point_manager.clone(),
pruning_proof_manager: services.pruning_proof_manager.clone(),
parents_manager: services.parents_manager.clone(),
pruning_lock,
config,
is_consensus_exiting,
}
}
pub fn worker(self: &Arc<Self>) {
let Ok(PruningProcessingMessage::Process { sink_ghostdag_data }) = self.receiver.recv() else {
return;
};
self.recover_pruning_workflows_if_needed();
self.advance_pruning_point_and_candidate_if_possible(sink_ghostdag_data);
while let Ok(PruningProcessingMessage::Process { sink_ghostdag_data }) = self.receiver.recv() {
self.advance_pruning_point_and_candidate_if_possible(sink_ghostdag_data);
}
}
fn recover_pruning_workflows_if_needed(&self) {
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 pruning_utxoset_position = self.pruning_utxoset_stores.read().utxoset_position().unwrap_option();
drop(pruning_point_read);
debug!(
"[PRUNING PROCESSOR] recovery check: current pruning point: {}, history root: {:?}, pruning utxoset position: {:?}",
pruning_point, history_root, pruning_utxoset_position
);
if let Some(pruning_utxoset_position) = pruning_utxoset_position {
if pruning_utxoset_position != pruning_point {
info!("Recovering pruning utxo-set from {} to the pruning point {}", pruning_utxoset_position, pruning_point);
if !self.advance_pruning_utxoset(pruning_utxoset_position, pruning_point) {
info!("Interrupted while advancing the pruning point UTXO set: Process is exiting");
return;
}
}
}
if let Some(history_root) = history_root {
if history_root != pruning_point {
self.prune(pruning_point);
}
}
}
fn advance_pruning_point_and_candidate_if_possible(&self, sink_ghostdag_data: CompactGhostdagData) {
let pruning_point_read = self.pruning_point_store.upgradable_read();
let current_pruning_info = pruning_point_read.get().unwrap();
let (new_pruning_points, new_candidate) = self.pruning_point_manager.next_pruning_points_and_candidate_by_ghostdag_data(
sink_ghostdag_data,
None,
current_pruning_info.candidate,
current_pruning_info.pruning_point,
);
if !new_pruning_points.is_empty() {
let mut batch = WriteBatch::default();
let mut pruning_point_write = RwLockUpgradableReadGuard::upgrade(pruning_point_read);
for (i, past_pp) in new_pruning_points.iter().copied().enumerate() {
self.past_pruning_points_store.insert_batch(&mut batch, current_pruning_info.index + i as u64 + 1, past_pp).unwrap();
}
let new_pp_index = current_pruning_info.index + new_pruning_points.len() as u64;
let new_pruning_point = *new_pruning_points.last().unwrap();
pruning_point_write.set_batch(&mut batch, new_pruning_point, new_candidate, new_pp_index).unwrap();
self.db.write(batch).unwrap();
drop(pruning_point_write);
info!("Periodic pruning point movement: advancing from {} to {}", current_pruning_info.pruning_point, new_pruning_point);
if !self.advance_pruning_utxoset(current_pruning_info.pruning_point, new_pruning_point) {
info!("Interrupted while advancing the pruning point UTXO set: Process is exiting");
return;
}
info!("Updated the pruning point UTXO set");
self.prune(new_pruning_point);
} 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();
}
}
fn advance_pruning_utxoset(&self, utxoset_position: Hash, new_pruning_point: Hash) -> bool {
let mut pruning_utxoset_write = self.pruning_utxoset_stores.write();
for chain_block in self.reachability_service.forward_chain_iterator(utxoset_position, new_pruning_point, true).skip(1) {
if self.is_consensus_exiting.load(Ordering::Relaxed) {
return false;
}
let utxo_diff = self.utxo_diffs_store.get(chain_block).expect("chain blocks have utxo state");
let mut batch = WriteBatch::default();
pruning_utxoset_write.utxo_set.write_diff_batch(&mut batch, utxo_diff.as_ref()).unwrap();
pruning_utxoset_write.set_utxoset_position(&mut batch, chain_block).unwrap();
self.db.write(batch).unwrap();
}
drop(pruning_utxoset_write);
if self.config.enable_sanity_checks {
info!("Performing a sanity check that the new UTXO set has the expected UTXO commitment");
self.assert_utxo_commitment(new_pruning_point);
}
true
}
fn assert_utxo_commitment(&self, pruning_point: Hash) {
info!("Verifying the new pruning point UTXO commitment (sanity test)");
let commitment = self.headers_store.get_header(pruning_point).unwrap().utxo_commitment;
let mut multiset = MuHash::new();
let pruning_utxoset_read = self.pruning_utxoset_stores.read();
for (outpoint, entry) in pruning_utxoset_read.utxo_set.iterator().map(|r| r.unwrap()) {
multiset.add_utxo(&outpoint, &entry);
}
assert_eq!(multiset.finalize(), commitment, "Updated pruning point utxo set does not match the header utxo commitment");
info!("Pruning point UTXO commitment was verified correctly (sanity test)");
}
fn prune(&self, new_pruning_point: 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;
}
info!("Header and Block pruning: preparing proof and anticone data...");
let proof = self.pruning_proof_manager.get_pruning_point_proof();
let data = self
.pruning_proof_manager
.get_pruning_point_anticone_and_trusted_data()
.expect("insufficient depth error is unexpected here");
let genesis = self.past_pruning_points_store.get(0).unwrap();
assert_eq!(new_pruning_point, proof[0].last().unwrap().hash);
assert_eq!(new_pruning_point, data.anticone[0]);
assert_eq!(genesis, self.config.genesis.hash);
assert_eq!(genesis, proof.last().unwrap().last().unwrap().hash);
let keep_blocks: BlockHashSet = data.anticone.iter().copied().collect();
let mut keep_relations: BlockHashMap<BlockLevel> = std::iter::empty()
.chain(data.anticone.iter().copied())
.chain(data.daa_window_blocks.iter().map(|th| th.header.hash))
.chain(data.ghostdag_blocks.iter().map(|gd| gd.hash))
.chain(proof[0].iter().map(|h| h.hash))
.map(|h| (h, 0)) .collect();
let keep_headers: BlockHashSet = self.past_pruning_points();
info!("Header and Block pruning: waiting for consensus write permissions...");
let mut prune_guard = self.pruning_lock.blocking_write();
info!("Starting Header and Block pruning...");
{
let mut counter = 0;
let mut batch = WriteBatch::default();
for kept in keep_relations.keys().copied() {
let Some(ghostdag) = self.ghostdag_primary_store.get_data(kept).unwrap_option() else {
continue;
};
if ghostdag.unordered_mergeset().any(|h| !keep_relations.contains_key(&h)) {
let mut mutable_ghostdag: ExternalGhostdagData = ghostdag.as_ref().into();
mutable_ghostdag.mergeset_blues.retain(|h| keep_relations.contains_key(h));
mutable_ghostdag.mergeset_reds.retain(|h| keep_relations.contains_key(h));
mutable_ghostdag.blues_anticone_sizes.retain(|k, _| keep_relations.contains_key(k));
if !keep_relations.contains_key(&mutable_ghostdag.selected_parent) {
mutable_ghostdag.selected_parent = ORIGIN;
}
counter += 1;
self.ghostdag_primary_store.update_batch(&mut batch, kept, &Arc::new(mutable_ghostdag.into())).unwrap();
}
}
self.db.write(batch).unwrap();
info!("Header and Block pruning: updated ghostdag data for {} blocks", counter);
}
drop(prune_guard);
for (level, level_proof) in proof.iter().enumerate().skip(1) {
let level = level as BlockLevel;
let roots_parents_at_level = data
.anticone
.iter()
.copied()
.map(|hash| self.headers_store.get_header_with_block_level(hash).expect("pruning point anticone is not pruned"))
.filter(|root| level > root.block_level) .flat_map(|root| self.parents_manager.parents_at_level(&root.header, level).iter().copied().collect_vec());
for hash in level_proof.iter().map(|header| header.hash).chain(roots_parents_at_level) {
if let Vacant(e) = keep_relations.entry(hash) {
e.insert(level);
}
}
}
prune_guard = self.pruning_lock.blocking_write();
let mut lock_acquire_time = Instant::now();
let mut reachability_read = self.reachability_store.upgradable_read();
{
let mut batch = WriteBatch::default();
let mut tips_write = self.body_tips_store.write();
let pruned_tips = tips_write
.get()
.unwrap()
.read()
.iter()
.copied()
.filter(|&h| !reachability_read.is_dag_ancestor_of_result(new_pruning_point, h).unwrap())
.collect_vec();
tips_write.prune_tips_with_writer(BatchDbWriter::new(&mut batch), &pruned_tips).unwrap();
if !pruned_tips.is_empty() {
info!(
"Header and Block pruning: pruned {} tips: {}...{}",
pruned_tips.len(),
pruned_tips.iter().take(5.min((pruned_tips.len() + 1) / 2)).reusable_format(", "),
pruned_tips.iter().rev().take(5.min(pruned_tips.len() / 2)).reusable_format(", ")
)
}
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();
self.db.write(batch).unwrap();
drop(selected_chain_write);
drop(tips_write);
}
let mut queue = VecDeque::<Hash>::from_iter(reachability_read.get_children(ORIGIN).unwrap().iter().copied());
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() {
continue;
}
traversed += 1;
queue.extend(reachability_read.get_children(current).unwrap().iter());
if lock_acquire_time.elapsed() > Duration::from_millis(5) {
drop(reachability_read);
if self.is_consensus_exiting.load(Ordering::Relaxed) {
drop(prune_guard);
info!("Header and Block pruning interrupted: Process is exiting");
return;
}
prune_guard.blocking_yield();
lock_acquire_time = Instant::now();
reachability_read = self.reachability_store.upgradable_read();
}
if traversed % 1000 == 0 {
info!("Header and Block pruning: traversed: {}, pruned {}...", traversed, counter);
}
self.block_window_cache_for_difficulty.remove(¤t);
self.block_window_cache_for_past_median_time.remove(¤t);
if !keep_blocks.contains(¤t) {
let mut batch = WriteBatch::default();
let mut level_relations_write = self.relations_stores.write();
let mut reachability_relations_write = self.reachability_relations_store.write();
let mut staging_relations = StagingRelationsStore::new(&mut reachability_relations_write);
let mut staging_reachability = StagingReachabilityStore::new(reachability_read);
let mut statuses_write = self.statuses_store.write();
self.utxo_multisets_store.delete_batch(&mut batch, current).unwrap();
self.utxo_diffs_store.delete_batch(&mut batch, current).unwrap();
self.acceptance_data_store.delete_batch(&mut batch, current).unwrap();
self.block_transactions_store.delete_batch(&mut batch, current).unwrap();
if let Some(&affiliated_proof_level) = keep_relations.get(¤t) {
if statuses_write.get(current).unwrap_option().is_some_and(|s| s.is_valid()) {
statuses_write.set_batch(&mut batch, current, StatusHeaderOnly).unwrap();
}
for lower_level in 0..affiliated_proof_level as usize {
let mut staging_level_relations = StagingRelationsStore::new(&mut level_relations_write[lower_level]);
relations::delete_level_relations(MemoryWriter, &mut staging_level_relations, current).unwrap_option();
staging_level_relations.commit(&mut batch).unwrap();
self.ghostdag_stores[lower_level].delete_batch(&mut batch, current).unwrap_option();
}
} else {
counter += 1;
let mergeset = relations::delete_reachability_relations(
MemoryWriter, &mut staging_relations,
&staging_reachability,
current,
);
reachability::delete_block(&mut staging_reachability, current, &mut mergeset.iter().copied()).unwrap();
let block_level = self.headers_store.get_header_with_block_level(current).unwrap().block_level;
(0..=block_level as usize).for_each(|level| {
let mut staging_level_relations = StagingRelationsStore::new(&mut level_relations_write[level]);
relations::delete_level_relations(MemoryWriter, &mut staging_level_relations, current).unwrap_option();
staging_level_relations.commit(&mut batch).unwrap();
self.ghostdag_stores[level].delete_batch(&mut batch, current).unwrap_option();
});
self.daa_excluded_store.delete_batch(&mut batch, current).unwrap();
self.depth_store.delete_batch(&mut batch, current).unwrap();
statuses_write.delete_batch(&mut batch, current).unwrap();
if !keep_headers.contains(¤t) {
self.headers_store.delete_batch(&mut batch, current).unwrap();
}
}
let reachability_write = staging_reachability.commit(&mut batch).unwrap();
staging_relations.commit(&mut batch).unwrap();
self.db.write(batch).unwrap();
drop(reachability_write);
drop(statuses_write);
drop(reachability_relations_write);
drop(level_relations_write);
reachability_read = self.reachability_store.upgradable_read();
}
}
drop(reachability_read);
drop(prune_guard);
info!("Header and Block pruning completed: traversed: {}, pruned {}", traversed, counter);
info!(
"Header and Block pruning stats: proof size: {}, pruning point and anticone: {}, unique headers in proof and windows: {}, pruning points in history: {}",
proof.iter().map(|l| l.len()).sum::<usize>(),
keep_blocks.len(),
keep_relations.len(),
keep_headers.len()
);
if self.config.enable_sanity_checks {
self.assert_proof_rebuilding(proof, new_pruning_point);
self.assert_data_rebuilding(data, new_pruning_point);
}
{
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();
self.db.write(batch).unwrap();
drop(pruning_point_write);
}
}
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())
.collect()
}
fn assert_proof_rebuilding(&self, ref_proof: Arc<PruningPointProof>, new_pruning_point: Hash) {
info!("Rebuilding the pruning proof after pruning data (sanity test)");
let proof_hashes = ref_proof.iter().flatten().map(|h| h.hash).collect::<Vec<_>>();
let built_proof = self.pruning_proof_manager.build_pruning_point_proof(new_pruning_point);
let built_proof_hashes = built_proof.iter().flatten().map(|h| h.hash).collect::<Vec<_>>();
assert_eq!(proof_hashes.len(), built_proof_hashes.len(), "Rebuilt proof does not match the expected reference");
for (i, (a, b)) in proof_hashes.into_iter().zip(built_proof_hashes).enumerate() {
if a != b {
panic!("Proof built following pruning does not match the previous proof: built[{}]={}, prev[{}]={}", i, b, i, a);
}
}
info!("Proof was rebuilt successfully following pruning");
}
fn assert_data_rebuilding(&self, ref_data: Arc<PruningPointTrustedData>, new_pruning_point: Hash) {
info!("Rebuilding pruning point trusted data (sanity test)");
let virtual_state = self.lkg_virtual_state.load();
let built_data = self
.pruning_proof_manager
.calculate_pruning_point_anticone_and_trusted_data(new_pruning_point, virtual_state.parents.iter().copied());
assert_eq!(
ref_data.anticone.iter().copied().collect::<BlockHashSet>(),
built_data.anticone.iter().copied().collect::<BlockHashSet>()
);
assert_eq!(
ref_data.daa_window_blocks.iter().map(|th| th.header.hash).collect::<BlockHashSet>(),
built_data.daa_window_blocks.iter().map(|th| th.header.hash).collect::<BlockHashSet>()
);
assert_eq!(
ref_data.ghostdag_blocks.iter().map(|gd| gd.hash).collect::<BlockHashSet>(),
built_data.ghostdag_blocks.iter().map(|gd| gd.hash).collect::<BlockHashSet>()
);
info!("Trusted data was rebuilt successfully following pruning");
}
}