Crescendo-related RAM optimizations & miscellaneous (#650)

* move genesis log to debug

* disable p2p server if connect is active

* log inbound/outbound counts

* use local cache for pruning proof build for saving ram (the proof is held in memory)

* do the same (local cache) for pruning proof received via p2p

* prettify

* logs (and 2x cache capacity)

* fix rare edge case

* fix comment

* bump version to 0.17.1

Author: michaelsutton

Cargo.lock

@@ -63,7 +63,7 @@ members = [
 
 [workspace.package]
 rust-version = "1.82.0"
-version = "0.17.0"
+version = "0.17.1"
 authors = ["Kaspa developers"]
 license = "ISC"
 repository = "https://github.com/kaspanet/rusty-kaspa"
@@ -80,61 +80,61 @@ include = [
 ]
 
 [workspace.dependencies]
-# kaspa-testing-integration = { version = "0.17.0", path = "testing/integration" }
-kaspa-addresses = { version = "0.17.0", path = "crypto/addresses" }
-kaspa-addressmanager = { version = "0.17.0", path = "components/addressmanager" }
-kaspa-bip32 = { version = "0.17.0", path = "wallet/bip32" }
-kaspa-cli = { version = "0.17.0", path = "cli" }
-kaspa-connectionmanager = { version = "0.17.0", path = "components/connectionmanager" }
-kaspa-consensus = { version = "0.17.0", path = "consensus" }
-kaspa-consensus-core = { version = "0.17.0", path = "consensus/core" }
-kaspa-consensus-client = { version = "0.17.0", path = "consensus/client" }
-kaspa-consensus-notify = { version = "0.17.0", path = "consensus/notify" }
-kaspa-consensus-wasm = { version = "0.17.0", path = "consensus/wasm" }
-kaspa-consensusmanager = { version = "0.17.0", path = "components/consensusmanager" }
-kaspa-core = { version = "0.17.0", path = "core" }
-kaspa-daemon = { version = "0.17.0", path = "daemon" }
-kaspa-database = { version = "0.17.0", path = "database" }
-kaspa-grpc-client = { version = "0.17.0", path = "rpc/grpc/client" }
-kaspa-grpc-core = { version = "0.17.0", path = "rpc/grpc/core" }
-kaspa-grpc-server = { version = "0.17.0", path = "rpc/grpc/server" }
-kaspa-hashes = { version = "0.17.0", path = "crypto/hashes" }
-kaspa-index-core = { version = "0.17.0", path = "indexes/core" }
-kaspa-index-processor = { version = "0.17.0", path = "indexes/processor" }
-kaspa-math = { version = "0.17.0", path = "math" }
-kaspa-merkle = { version = "0.17.0", path = "crypto/merkle" }
-kaspa-metrics-core = { version = "0.17.0", path = "metrics/core" }
-kaspa-mining = { version = "0.17.0", path = "mining" }
-kaspa-mining-errors = { version = "0.17.0", path = "mining/errors" }
-kaspa-muhash = { version = "0.17.0", path = "crypto/muhash" }
-kaspa-notify = { version = "0.17.0", path = "notify" }
-kaspa-p2p-flows = { version = "0.17.0", path = "protocol/flows" }
-kaspa-p2p-lib = { version = "0.17.0", path = "protocol/p2p" }
-kaspa-perf-monitor = { version = "0.17.0", path = "metrics/perf_monitor" }
-kaspa-pow = { version = "0.17.0", path = "consensus/pow" }
-kaspa-rpc-core = { version = "0.17.0", path = "rpc/core" }
-kaspa-rpc-macros = { version = "0.17.0", path = "rpc/macros" }
-kaspa-rpc-service = { version = "0.17.0", path = "rpc/service" }
-kaspa-txscript = { version = "0.17.0", path = "crypto/txscript" }
-kaspa-txscript-errors = { version = "0.17.0", path = "crypto/txscript/errors" }
-kaspa-utils = { version = "0.17.0", path = "utils" }
-kaspa-utils-tower = { version = "0.17.0", path = "utils/tower" }
-kaspa-utxoindex = { version = "0.17.0", path = "indexes/utxoindex" }
-kaspa-wallet = { version = "0.17.0", path = "wallet/native" }
-kaspa-wallet-cli-wasm = { version = "0.17.0", path = "wallet/wasm" }
-kaspa-wallet-keys = { version = "0.17.0", path = "wallet/keys" }
-kaspa-wallet-pskt = { version = "0.17.0", path = "wallet/pskt" }
-kaspa-wallet-core = { version = "0.17.0", path = "wallet/core" }
-kaspa-wallet-macros = { version = "0.17.0", path = "wallet/macros" }
-kaspa-wasm = { version = "0.17.0", path = "wasm" }
-kaspa-wasm-core = { version = "0.17.0", path = "wasm/core" }
-kaspa-wrpc-client = { version = "0.17.0", path = "rpc/wrpc/client" }
-kaspa-wrpc-proxy = { version = "0.17.0", path = "rpc/wrpc/proxy" }
-kaspa-wrpc-server = { version = "0.17.0", path = "rpc/wrpc/server" }
-kaspa-wrpc-wasm = { version = "0.17.0", path = "rpc/wrpc/wasm" }
-kaspa-wrpc-example-subscriber = { version = "0.17.0", path = "rpc/wrpc/examples/subscriber" }
-kaspad = { version = "0.17.0", path = "kaspad" }
-kaspa-alloc = { version = "0.17.0", path = "utils/alloc" }
+# kaspa-testing-integration = { version = "0.17.1", path = "testing/integration" }
+kaspa-addresses = { version = "0.17.1", path = "crypto/addresses" }
+kaspa-addressmanager = { version = "0.17.1", path = "components/addressmanager" }
+kaspa-bip32 = { version = "0.17.1", path = "wallet/bip32" }
+kaspa-cli = { version = "0.17.1", path = "cli" }
+kaspa-connectionmanager = { version = "0.17.1", path = "components/connectionmanager" }
+kaspa-consensus = { version = "0.17.1", path = "consensus" }
+kaspa-consensus-core = { version = "0.17.1", path = "consensus/core" }
+kaspa-consensus-client = { version = "0.17.1", path = "consensus/client" }
+kaspa-consensus-notify = { version = "0.17.1", path = "consensus/notify" }
+kaspa-consensus-wasm = { version = "0.17.1", path = "consensus/wasm" }
+kaspa-consensusmanager = { version = "0.17.1", path = "components/consensusmanager" }
+kaspa-core = { version = "0.17.1", path = "core" }
+kaspa-daemon = { version = "0.17.1", path = "daemon" }
+kaspa-database = { version = "0.17.1", path = "database" }
+kaspa-grpc-client = { version = "0.17.1", path = "rpc/grpc/client" }
+kaspa-grpc-core = { version = "0.17.1", path = "rpc/grpc/core" }
+kaspa-grpc-server = { version = "0.17.1", path = "rpc/grpc/server" }
+kaspa-hashes = { version = "0.17.1", path = "crypto/hashes" }
+kaspa-index-core = { version = "0.17.1", path = "indexes/core" }
+kaspa-index-processor = { version = "0.17.1", path = "indexes/processor" }
+kaspa-math = { version = "0.17.1", path = "math" }
+kaspa-merkle = { version = "0.17.1", path = "crypto/merkle" }
+kaspa-metrics-core = { version = "0.17.1", path = "metrics/core" }
+kaspa-mining = { version = "0.17.1", path = "mining" }
+kaspa-mining-errors = { version = "0.17.1", path = "mining/errors" }
+kaspa-muhash = { version = "0.17.1", path = "crypto/muhash" }
+kaspa-notify = { version = "0.17.1", path = "notify" }
+kaspa-p2p-flows = { version = "0.17.1", path = "protocol/flows" }
+kaspa-p2p-lib = { version = "0.17.1", path = "protocol/p2p" }
+kaspa-perf-monitor = { version = "0.17.1", path = "metrics/perf_monitor" }
+kaspa-pow = { version = "0.17.1", path = "consensus/pow" }
+kaspa-rpc-core = { version = "0.17.1", path = "rpc/core" }
+kaspa-rpc-macros = { version = "0.17.1", path = "rpc/macros" }
+kaspa-rpc-service = { version = "0.17.1", path = "rpc/service" }
+kaspa-txscript = { version = "0.17.1", path = "crypto/txscript" }
+kaspa-txscript-errors = { version = "0.17.1", path = "crypto/txscript/errors" }
+kaspa-utils = { version = "0.17.1", path = "utils" }
+kaspa-utils-tower = { version = "0.17.1", path = "utils/tower" }
+kaspa-utxoindex = { version = "0.17.1", path = "indexes/utxoindex" }
+kaspa-wallet = { version = "0.17.1", path = "wallet/native" }
+kaspa-wallet-cli-wasm = { version = "0.17.1", path = "wallet/wasm" }
+kaspa-wallet-keys = { version = "0.17.1", path = "wallet/keys" }
+kaspa-wallet-pskt = { version = "0.17.1", path = "wallet/pskt" }
+kaspa-wallet-core = { version = "0.17.1", path = "wallet/core" }
+kaspa-wallet-macros = { version = "0.17.1", path = "wallet/macros" }
+kaspa-wasm = { version = "0.17.1", path = "wasm" }
+kaspa-wasm-core = { version = "0.17.1", path = "wasm/core" }
+kaspa-wrpc-client = { version = "0.17.1", path = "rpc/wrpc/client" }
+kaspa-wrpc-proxy = { version = "0.17.1", path = "rpc/wrpc/proxy" }
+kaspa-wrpc-server = { version = "0.17.1", path = "rpc/wrpc/server" }
+kaspa-wrpc-wasm = { version = "0.17.1", path = "rpc/wrpc/wasm" }
+kaspa-wrpc-example-subscriber = { version = "0.17.1", path = "rpc/wrpc/examples/subscriber" }
+kaspad = { version = "0.17.1", path = "kaspad" }
+kaspa-alloc = { version = "0.17.1", path = "utils/alloc" }
 
 # external
 aes = "0.8.3"

Cargo.toml

@@ -5,7 +5,7 @@ use kaspa_consensus_core::{
     blockhash::{BlockHashExtensions, BlockHashes},
     header::Header,
     pruning::PruningPointProof,
-    BlockHashSet, BlockLevel, HashMapCustomHasher, KType,
+    BlockHashMap, BlockHashSet, BlockLevel, HashMapCustomHasher, KType,
 };
 use kaspa_core::debug;
 use kaspa_database::prelude::{CachePolicy, ConnBuilder, StoreError, StoreResult, StoreResultEmptyTuple, StoreResultExtensions, DB};
@@ -71,6 +71,11 @@ impl PruningProofManager {
         let pp_header = self.headers_store.get_header_with_block_level(pp).unwrap();
         let (ghostdag_stores, selected_tip_by_level, roots_by_level) = self.calc_gd_for_all_levels(&pp_header, temp_db);
 
+        // The pruning proof can contain many duplicate headers (across levels), so we use a local cache in order
+        // to make sure we hold a single Arc per header
+        let mut cache: BlockHashMap<Arc<Header>> = BlockHashMap::with_capacity(4 * self.pruning_proof_m as usize);
+        let mut get_header = |hash| cache.entry(hash).or_insert_with_key(|&hash| self.headers_store.get_header(hash).unwrap()).clone();
+
         (0..=self.max_block_level)
             .map(|level| {
                 let level = level as usize;
@@ -114,7 +119,7 @@ impl PruningProofManager {
                 let mut headers = Vec::with_capacity(2 * self.pruning_proof_m as usize);
                 let mut queue = BinaryHeap::<Reverse<SortableBlock>>::new();
                 let mut visited = BlockHashSet::new();
-                queue.push(Reverse(SortableBlock::new(root, self.headers_store.get_header(root).unwrap().blue_work)));
+                queue.push(Reverse(SortableBlock::new(root, get_header(root).blue_work)));
                 while let Some(current) = queue.pop() {
                     let current = current.0.hash;
                     if !visited.insert(current) {
@@ -130,9 +135,9 @@ impl PruningProofManager {
                         continue;
                     }
 
-                    headers.push(self.headers_store.get_header(current).unwrap());
+                    headers.push(get_header(current));
                     for child in self.relations_stores.read()[level].get_children(current).unwrap().read().iter().copied() {
-                        queue.push(Reverse(SortableBlock::new(child, self.headers_store.get_header(child).unwrap().blue_work)));
+                        queue.push(Reverse(SortableBlock::new(child, get_header(child).blue_work)));
                     }
                 }
 
@@ -232,14 +237,14 @@ impl PruningProofManager {
 
         // We only have the headers store (which has level 0 blue_scores) to assemble the proof data from.
         // We need to look deeper at higher levels (2x deeper every level) to find 2M (plus margin) blocks at that level
-        // TODO: uncomment when the full fix to minimize proof sizes come.
+        // TODO: uncomment when the full fix to minimize proof sizes comes.
         // let mut required_base_level_depth = self.estimated_blue_depth_at_level_0(
         //     level,
         //     required_level_depth + 100, // We take a safety margin
         //     current_dag_level,
         // );
-        // NOTE: Starting from required_level_depth (a much lower starting point than normal) will typically require N iterations
-        // for every N level higher than current dag level. This is fine since the steps per iteration are still exponential
+        // NOTE: Starting from required_level_depth (a much lower starting point than normal) will typically require O(N) iterations
+        // for level L + N where L is the current dag level. This is fine since the steps per iteration are still exponential
         // and so we will complete each level in not much more than N iterations per level.
         // We start here anyway so we can try to minimize the proof size when the current dag level goes down significantly.
         let mut required_base_level_depth = required_level_depth + 100;

consensus/src/processes/pruning_proof/build.rs

@@ -383,6 +383,11 @@ impl PruningProofManager {
             }
         }
         let proof = Arc::new(self.build_pruning_point_proof(pp));
+        info!(
+            "Built headers proof with overall {} headers ({} unique)",
+            proof.iter().map(|l| l.len()).sum::<usize>(),
+            proof.iter().flatten().unique_by(|h| h.hash).count()
+        );
         cache_lock.replace(CachedPruningPointData { pruning_point: pp, data: proof.clone() });
         proof
     }

consensus/src/processes/pruning_proof/mod.rs

@@ -110,12 +110,15 @@ impl PruningProofManager {
             ) {
                 let proof_level_blue_work_diff = proof_selected_tip_gd.blue_work.saturating_sub(proof_common_ancestor_gd.blue_work);
                 for parent in self.parents_manager.parents_at_level(&current_pp_header, level).iter().copied() {
-                    let parent_blue_work = current_consensus_ghostdag_stores[level_idx].get_blue_work(parent).unwrap();
-                    let parent_blue_work_diff = parent_blue_work.saturating_sub(common_ancestor_gd.blue_work);
-                    if parent_blue_work_diff.saturating_add(pruning_period_work)
-                        >= proof_level_blue_work_diff.saturating_add(prover_claimed_pruning_period_work)
+                    // Not all parents by level are guaranteed to be GD populated, but at least one of them will (the proof level selected tip)
+                    if let Some(parent_blue_work) = current_consensus_ghostdag_stores[level_idx].get_blue_work(parent).unwrap_option()
                     {
-                        return Err(PruningImportError::PruningProofInsufficientBlueWork);
+                        let parent_blue_work_diff = parent_blue_work.saturating_sub(common_ancestor_gd.blue_work);
+                        if parent_blue_work_diff.saturating_add(pruning_period_work)
+                            >= proof_level_blue_work_diff.saturating_add(prover_claimed_pruning_period_work)
+                        {
+                            return Err(PruningImportError::PruningProofInsufficientBlueWork);
+                        }
                     }
                 }
 

consensus/src/processes/pruning_proof/validate.rs

@@ -300,16 +300,16 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
                 let headers_store = DbHeadersStore::new(consensus_db, CachePolicy::Empty, CachePolicy::Empty);
 
                 if headers_store.has(config.genesis.hash).unwrap() {
-                    info!("Genesis is found in active consensus DB. No action needed.");
+                    debug!("Genesis is found in active consensus DB. No action needed.");
                 } else {
-                    let msg = "Genesis not found in active consensus DB. This happens when Testnet 11 is restarted and your database needs to be fully deleted. Do you confirm the delete? (y/n)";
+                    let msg = "Genesis not found in active consensus DB. This happens when Testnets are restarted and your database needs to be fully deleted. Do you confirm the delete? (y/n)";
                     get_user_approval_or_exit(msg, args.yes);
 
                     is_db_reset_needed = true;
                 }
             }
             None => {
-                info!("Consensus not initialized yet. Skipping genesis check.");
+                debug!("Consensus not initialized yet. Skipping genesis check.");
             }
         }
     }
@@ -454,8 +454,9 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
     let connect_peers = args.connect_peers.iter().map(|x| x.normalize(config.default_p2p_port())).collect::<Vec<_>>();
     let add_peers = args.add_peers.iter().map(|x| x.normalize(config.default_p2p_port())).collect();
     let p2p_server_addr = args.listen.unwrap_or(ContextualNetAddress::unspecified()).normalize(config.default_p2p_port());
-    // connect_peers means no DNS seeding and no outbound peers
+    // connect_peers means no DNS seeding and no outbound/inbound peers
     let outbound_target = if connect_peers.is_empty() { args.outbound_target } else { 0 };
+    let inbound_limit = if connect_peers.is_empty() { args.inbound_limit } else { 0 };
     let dns_seeders = if connect_peers.is_empty() && !args.disable_dns_seeding { config.dns_seeders } else { &[] };
 
     let grpc_server_addr = args.rpclisten.unwrap_or(ContextualNetAddress::loopback()).normalize(config.default_rpc_port());
@@ -556,7 +557,7 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
         add_peers,
         p2p_server_addr,
         outbound_target,
-        args.inbound_limit,
+        inbound_limit,
         dns_seeders,
         config.default_p2p_port(),
         p2p_tower_counters.clone(),

kaspad/src/daemon.rs

@@ -65,9 +65,12 @@ impl AsyncService for P2pService {
         // Prepare a shutdown signal receiver
         let shutdown_signal = self.shutdown.listener.clone();
 
-        let p2p_adaptor =
+        let p2p_adaptor = if self.inbound_limit == 0 {
+            Adaptor::client_only(self.flow_context.hub().clone(), self.flow_context.clone(), self.counters.clone())
+        } else {
             Adaptor::bidirectional(self.listen, self.flow_context.hub().clone(), self.flow_context.clone(), self.counters.clone())
-                .unwrap();
+                .unwrap()
+        };
         let connection_manager = ConnectionManager::new(
             p2p_adaptor.clone(),
             self.outbound_target,

protocol/flows/src/service.rs

@@ -6,6 +6,7 @@ use crate::{
     },
 };
 use futures::future::{join_all, select, try_join_all, Either};
+use itertools::Itertools;
 use kaspa_consensus_core::{
     api::BlockValidationFuture,
     block::Block,
@@ -245,7 +246,11 @@ impl IbdFlow {
         // Pruning proof generation and communication might take several minutes, so we allow a long 10 minute timeout
         let msg = dequeue_with_timeout!(self.incoming_route, Payload::PruningPointProof, Duration::from_secs(600))?;
         let proof: PruningPointProof = msg.try_into()?;
-        debug!("received proof with overall {} headers", proof.iter().map(|l| l.len()).sum::<usize>());
+        info!(
+            "Received headers proof with overall {} headers ({} unique)",
+            proof.iter().map(|l| l.len()).sum::<usize>(),
+            proof.iter().flatten().unique_by(|h| h.hash).count()
+        );
 
         let proof_metadata = PruningProofMetadata::new(relay_block.header.blue_work);
 

protocol/flows/src/v5/ibd/flow.rs

@@ -15,7 +15,7 @@ use kaspa_consensus_core::{
 use kaspa_hashes::Hash;
 use kaspa_utils::networking::{IpAddress, PeerId};
 
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 // ----------------------------------------------------------------------------
 // consensus_core to protowire
@@ -85,7 +85,23 @@ impl TryFrom<protowire::RequestIbdChainBlockLocatorMessage> for (Option<Hash>, O
 impl TryFrom<protowire::PruningPointProofMessage> for PruningPointProof {
     type Error = ConversionError;
     fn try_from(msg: protowire::PruningPointProofMessage) -> Result<Self, Self::Error> {
-        msg.headers.into_iter().map(|v| v.try_into()).collect()
+        // The pruning proof can contain many duplicate headers (across levels), so we use a local cache in order
+        // to make sure we hold a single Arc per header
+        let mut cache: HashMap<Hash, Arc<Header>> = HashMap::with_capacity(4000);
+        msg.headers
+            .into_iter()
+            .map(|level| {
+                level
+                    .headers
+                    .into_iter()
+                    .map(|x| {
+                        let header: Header = x.try_into()?;
+                        // Clone the existing Arc if found
+                        Ok(cache.entry(header.hash).or_insert_with(|| Arc::new(header)).clone())
+                    })
+                    .collect()
+            })
+            .collect()
     }
 }