Bug fixes: ungraceful test crash fixes

.github/workflows/ci.yaml

@@ -113,6 +113,9 @@ jobs:
             target/
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
 
+      - name: Run cargo tests non--release
+        run: cargo nextest run --no-fail-fast
+
       - name: Run cargo build with devnet-prealloc feature
         run: cargo build --release --features devnet-prealloc --workspace --all --tests --benches
 

components/addressmanager/src/lib.rs

@@ -546,7 +546,7 @@ mod address_store_with_cache {
             // Assert that initial distribution is skewed, and hence not uniform from the outset.
             assert!(bucket_reduction_ratio >= 1.25);
 
-            let db = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+            let db = create_temp_db!(ConnBuilder::default().with_files_limit(10)).expect("Failed to create temp db");
             let config = Config::new(SIMNET_PARAMS);
             let (am, _) = AddressManager::new(Arc::new(config), db.1, Arc::new(TickService::default()));
 

consensus/src/consensus/test_consensus.rs

@@ -68,7 +68,7 @@ impl TestConsensus {
 
     /// Creates a test consensus instance based on `config` with a temp DB and the provided `notification_sender`
     pub fn with_notifier(config: &Config, notification_sender: Sender<Notification>, context: SubscriptionContext) -> Self {
-        let (db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let notification_root = Arc::new(ConsensusNotificationRoot::with_context(notification_sender, context));
         let counters = Default::default();
         let tx_script_cache_counters = Default::default();
@@ -89,7 +89,7 @@ impl TestConsensus {
 
     /// Creates a test consensus instance based on `config` with a temp DB and no notifier
     pub fn new(config: &Config) -> Self {
-        let (db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let (dummy_notification_sender, _) = async_channel::unbounded();
         let notification_root = Arc::new(ConsensusNotificationRoot::new(dummy_notification_sender));
         let counters = Default::default();

consensus/src/model/stores/relations.rs

@@ -419,7 +419,7 @@ mod tests {
 
     #[test]
     fn test_db_relations_store() {
-        let (lt, db) = create_temp_db!(kaspa_database::prelude::ConnBuilder::default().with_files_limit(10));
+        let (lt, db) = create_temp_db!(kaspa_database::prelude::ConnBuilder::default().with_files_limit(10)).unwrap();
         test_relations_store(DbRelationsStore::new(
             db,
             0,

consensus/src/model/stores/tips.rs

@@ -103,7 +103,7 @@ mod tests {
 
     #[test]
     fn test_update_tips() {
-        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let mut store = DbTipsStore::new(db.clone());
         store.add_tip(1.into(), &[]).unwrap();
         store.add_tip(3.into(), &[]).unwrap();

consensus/src/processes/pruning_proof/build.rs

@@ -67,7 +67,7 @@ impl PruningProofManager {
             return vec![];
         }
 
-        let (_db_lifetime, temp_db) = kaspa_database::create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_db_lifetime, temp_db) = kaspa_database::create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         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);
 

consensus/src/processes/pruning_proof/validate.rs

@@ -182,7 +182,7 @@ impl PruningProofManager {
 
         let headers_estimate = self.estimate_proof_unique_size(proof);
 
-        let (db_lifetime, db) = kaspa_database::create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (db_lifetime, db) = kaspa_database::create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let cache_policy = CachePolicy::Count(2 * self.pruning_proof_m as usize);
         let headers_store =
             Arc::new(DbHeadersStore::new(db.clone(), CachePolicy::Count(headers_estimate), CachePolicy::Count(headers_estimate)));

consensus/src/processes/reachability/inquirer.rs

@@ -386,7 +386,7 @@ mod tests {
     /// Runs a DAG test-case with full verification using the staging store mechanism.
     /// Note: runtime is quadratic in the number of blocks so should be used with mildly small DAGs (~50)
     fn run_dag_test_case_with_staging(test: &DagTestCase) {
-        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let cache_policy = CachePolicy::Count(test.blocks.len() / 3);
         let reachability = RwLock::new(DbReachabilityStore::new(db.clone(), cache_policy, cache_policy));
         let mut relations = DbRelationsStore::with_prefix(db.clone(), &[], CachePolicy::Empty, CachePolicy::Empty);
@@ -520,7 +520,7 @@ mod tests {
             run_dag_test_case(&mut relations, &mut reachability, &test);
 
             // Run with direct DB stores
-            let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+            let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
             let cache_policy = CachePolicy::Count(test.blocks.len() / 3);
             let mut reachability = DbReachabilityStore::new(db.clone(), cache_policy, cache_policy);
             let mut relations = DbRelationsStore::new(db, 0, cache_policy, cache_policy);

consensus/src/processes/relations.rs

@@ -159,7 +159,7 @@ mod tests {
 
     #[test]
     fn test_delete_level_relations_zero_cache() {
-        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let mut relations = DbRelationsStore::new(db.clone(), 0, CachePolicy::Empty, CachePolicy::Empty);
         relations.insert(ORIGIN, Default::default()).unwrap();
         relations.insert(1.into(), Arc::new(vec![ORIGIN])).unwrap();

database/src/access.rs

@@ -229,7 +229,7 @@ mod tests {
 
     #[test]
     fn test_delete_all() {
-        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let access = CachedDbAccess::<Hash, u64>::new(db.clone(), CachePolicy::Count(2), vec![1, 2]);
 
         access.write_many(DirectDbWriter::new(&db), &mut (0..16).map(|i| (i.into(), 2))).unwrap();

database/src/set_access.rs

@@ -205,7 +205,7 @@ mod tests {
 
     #[test]
     fn test_delete_bucket() {
-        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).expect("Failed to create temp db");
         let access = DbSetAccess::<Hash, u64>::new(db.clone(), vec![1, 2]);
 
         for i in 0..16 {

database/src/utils.rs

@@ -41,23 +41,38 @@ impl Drop for DbLifetime {
     }
 }
 
-pub fn get_kaspa_tempdir() -> TempDir {
+pub fn get_kaspa_tempdir() -> Result<TempDir, std::io::Error> {
     let global_tempdir = std::env::temp_dir();
     let kaspa_tempdir = global_tempdir.join("rusty-kaspa");
-    std::fs::create_dir_all(kaspa_tempdir.as_path()).unwrap();
-    let db_tempdir = tempfile::tempdir_in(kaspa_tempdir.as_path()).unwrap();
-    db_tempdir
+    std::fs::create_dir_all(&kaspa_tempdir).map_err(|err| {
+        std::io::Error::new(err.kind(), format!("Failed to create kaspa directory '{}': {}", kaspa_tempdir.display(), err))
+    })?;
+    tempfile::tempdir_in(&kaspa_tempdir).map_err(|err| {
+        std::io::Error::new(err.kind(), format!("Failed to create db tempdir in '{}': {}", kaspa_tempdir.display(), err))
+    })
 }
 
 /// Creates a DB within a temp directory under `<OS SPECIFIC TEMP DIR>/kaspa-rust`
 /// Callers must keep the `TempDbLifetime` guard for as long as they wish the DB to exist.
 #[macro_export]
 macro_rules! create_temp_db {
     ($conn_builder: expr) => {{
-        let db_tempdir = $crate::utils::get_kaspa_tempdir();
-        let db_path = db_tempdir.path().to_owned();
-        let db = $conn_builder.with_db_path(db_path).build().unwrap();
-        ($crate::utils::DbLifetime::new(db_tempdir, std::sync::Arc::downgrade(&db)), db)
+        // Create the temporary directory.
+        let db_tempdir = $crate::utils::get_kaspa_tempdir().unwrap();
+        // Extract and clone the DB path for later use (for error messages).
+        let db_tempdir_path = db_tempdir.path().to_owned();
+        // Build the database.
+        $conn_builder
+            .with_db_path(db_tempdir_path.clone())
+            .build()
+            .map(|db| {
+                // On success, move `db_tempdir` into the DbLifetime.
+                ($crate::utils::DbLifetime::new(db_tempdir, std::sync::Arc::downgrade(&db)), db)
+            })
+            .map_err(|e| {
+                // Use the cloned path for the error message.
+                format!("Failed to create temp db at {}: {}", db_tempdir_path.display(), e)
+            })
     }};
 }
 

indexes/processor/src/processor.rs

@@ -154,7 +154,7 @@ mod tests {
     impl NotifyPipeline {
         fn new() -> Self {
             let (consensus_sender, consensus_receiver) = unbounded();
-            let (utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+            let (utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
             let config = Arc::new(Config::new(DEVNET_PARAMS));
             let tc = TestConsensus::new(&config);
             tc.init();

indexes/utxoindex/src/index.rs

@@ -253,7 +253,7 @@ mod tests {
 
         // Initialize all components, and virtual change emulator proxy.
         let mut virtual_change_emulator = VirtualChangeEmulator::new();
-        let (_utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+        let (_utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
         let config = Config::new(DEVNET_PARAMS);
         let tc = Arc::new(TestConsensus::new(&config));
         let consensus_manager = Arc::new(ConsensusManager::from_consensus(tc.consensus_clone()));

notify/src/subscription/single.rs

@@ -383,11 +383,17 @@ impl Display for UtxosChangedSubscription {
 
 impl Drop for UtxosChangedSubscription {
     fn drop(&mut self) {
-        trace!(
-            "UtxosChangedSubscription: {} in total (drop {})",
-            UTXOS_CHANGED_SUBSCRIPTIONS.fetch_sub(1, Ordering::SeqCst) - 1,
-            self
-        );
+        // Prevent underflow; [ERROR] thread 'tokio-runtime-worker' panicked at notify/src/subscription/single.rs:388:13: attempt to subtract with overflow
+        let _ = UTXOS_CHANGED_SUBSCRIPTIONS.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |count| {
+            if count > 0 {
+                Some(count - 1)
+            } else {
+                log::warn!("Tried subtracting UtxosChangedSubscription when it's already zero!");
+                None
+            }
+        });
+
+        trace!("UtxosChangedSubscription: {} in total (drop {})", UTXOS_CHANGED_SUBSCRIPTIONS.load(Ordering::SeqCst), self);
     }
 }
 

protocol/p2p/src/core/connection_handler.rs

@@ -5,7 +5,7 @@ use crate::pb::{
 };
 use crate::{ConnectionInitializer, Router};
 use futures::FutureExt;
-use kaspa_core::{debug, info};
+use kaspa_core::{debug, error, info};
 use kaspa_utils::networking::NetAddress;
 use kaspa_utils_tower::{
     counters::TowerConnectionCounters,
@@ -85,9 +85,11 @@ impl ConnectionHandler {
                 .serve_with_shutdown(serve_address.into(), termination_receiver.map(drop))
                 .await;
 
+            // log:error! was panic! before, but crashes testing ungracefully (error! not implemented)
+            // Now error is propagated
             match serve_result {
                 Ok(_) => info!("P2P Server stopped: {}", serve_address),
-                Err(err) => panic!("P2P, Server {serve_address} stopped with error: {err:?}"),
+                Err(err) => error!("P2P, Server {serve_address} stopped with error: {err:?}"),
             }
         });
         Ok(termination_sender)

simpa/src/main.rs

@@ -301,7 +301,8 @@ fn main_impl(mut args: Args) {
     }
 
     // Benchmark the DAG validation time
-    let (_lifetime2, db2) = create_temp_db!(ConnBuilder::default().with_parallelism(num_cpus::get()).with_files_limit(default_fd));
+    let (_lifetime2, db2) =
+        create_temp_db!(ConnBuilder::default().with_parallelism(num_cpus::get()).with_files_limit(default_fd)).unwrap();
     let (dummy_notification_sender, _) = unbounded();
     let notification_root = Arc::new(ConsensusNotificationRoot::new(dummy_notification_sender));
     let consensus2 = Arc::new(Consensus::new(

simpa/src/simulator/network.rs

@@ -71,10 +71,10 @@ impl KaspaNetworkSimulator {
                 (true, Some(dir), false, _) => create_permanent_db!(dir, builder),
 
                 (_, _, true, Some(rocksdb_stats_period_sec)) => {
-                    create_temp_db!(builder.enable_stats().with_stats_period(rocksdb_stats_period_sec))
+                    create_temp_db!(builder.enable_stats().with_stats_period(rocksdb_stats_period_sec)).unwrap()
                 }
-                (_, _, true, None) => create_temp_db!(builder.enable_stats()),
-                (_, _, false, _) => create_temp_db!(builder),
+                (_, _, true, None) => create_temp_db!(builder.enable_stats()).unwrap(),
+                (_, _, false, _) => create_temp_db!(builder).unwrap(),
             };
 
             let (dummy_notification_sender, _) = unbounded();

testing/integration/src/common/daemon.rs

@@ -135,7 +135,7 @@ impl Daemon {
     }
 
     pub fn with_manager(client_manager: Arc<ClientManager>, fd_total_budget: i32) -> Daemon {
-        let appdir_tempdir = get_kaspa_tempdir();
+        let appdir_tempdir = get_kaspa_tempdir().unwrap();
         client_manager.args.write().appdir = Some(appdir_tempdir.path().to_str().unwrap().to_owned());
         let (core, _) = create_core_with_runtime(&Default::default(), &client_manager.args.read(), fd_total_budget);
         let async_service = &Arc::downcast::<AsyncRuntime>(core.find(AsyncRuntime::IDENT).unwrap().arc_any()).unwrap();

testing/integration/src/consensus_integration_tests.rs

@@ -128,7 +128,7 @@ fn reachability_stretch_test(use_attack_json: bool) {
     map.get_mut(&blocks[0]).unwrap().parents.push(root);
 
     // Act
-    let (_temp_db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+    let (_temp_db_lifetime, db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
     let mut store = DbReachabilityStore::new(db.clone(), CachePolicy::Count(50_000), CachePolicy::Count(50_000));
     let mut relations = DbRelationsStore::new(db, 0, CachePolicy::Count(100_000), CachePolicy::Count(100_000)); // TODO: remove level
     let mut builder = DagBuilder::new(&mut store, &mut relations);
@@ -955,9 +955,9 @@ async fn json_test(file_path: &str, concurrency: bool) {
     let notify_service = Arc::new(NotifyService::new(tc.notification_root(), notification_recv, subscription_context.clone()));
 
     // External storage for storing block bodies. This allows separating header and body processing phases
-    let (_external_db_lifetime, external_storage) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+    let (_external_db_lifetime, external_storage) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
     let external_block_store = DbBlockTransactionsStore::new(external_storage, CachePolicy::Count(config.perf.block_data_cache_size));
-    let (_utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10));
+    let (_utxoindex_db_lifetime, utxoindex_db) = create_temp_db!(ConnBuilder::default().with_files_limit(10)).unwrap();
     let consensus_manager = Arc::new(ConsensusManager::new(Arc::new(TestConsensusFactory::new(tc.clone()))));
     let utxoindex = UtxoIndex::new(consensus_manager.clone(), utxoindex_db).unwrap();
     let index_service = Arc::new(IndexService::new(
@@ -1739,7 +1739,7 @@ async fn staging_consensus_test() {
     init_allocator_with_default_settings();
     let config = ConfigBuilder::new(MAINNET_PARAMS).build();
 
-    let db_tempdir = get_kaspa_tempdir();
+    let db_tempdir = get_kaspa_tempdir().unwrap();
     let db_path = db_tempdir.path().to_owned();
     let consensus_db_dir = db_path.join("consensus");
     let meta_db_dir = db_path.join("meta");
@@ -1773,6 +1773,7 @@ async fn staging_consensus_test() {
 
     core.shutdown();
     core.join(joins);
+    drop(consensus_manager);
 }
 
 /// Tests the KIP-10 transaction introspection opcode activation by verifying that:
@@ -1826,7 +1827,7 @@ async fn run_kip10_activation_test() {
     let mut genesis_multiset = MuHash::new();
     consensus.append_imported_pruning_point_utxos(&initial_utxo_collection, &mut genesis_multiset);
     consensus.import_pruning_point_utxo_set(config.genesis.hash, genesis_multiset).unwrap();
-    consensus.init();
+    let wait_handles = consensus.init();
 
     // Build blockchain up to one block before activation
     let mut index = 0;
@@ -1885,6 +1886,8 @@ async fn run_kip10_activation_test() {
     let status = consensus.add_utxo_valid_block_with_parents((index + 1).into(), vec![index.into()], vec![tx.clone()]).await;
     assert!(matches!(status, Ok(BlockStatus::StatusUTXOValid)));
     assert!(consensus.lkg_virtual_state.load().accepted_tx_ids.contains(&tx_id));
+    consensus.shutdown(wait_handles);
+    drop(consensus);
 }
 
 #[tokio::test]
@@ -1981,7 +1984,7 @@ async fn payload_activation_test() {
     let mut genesis_multiset = MuHash::new();
     consensus.append_imported_pruning_point_utxos(&initial_utxo_collection, &mut genesis_multiset);
     consensus.import_pruning_point_utxo_set(config.genesis.hash, genesis_multiset).unwrap();
-    consensus.init();
+    let wait_handles = consensus.init();
 
     // Build blockchain up to one block before activation
     let mut index = 0;
@@ -2048,6 +2051,8 @@ async fn payload_activation_test() {
 
     assert!(matches!(status, Ok(BlockStatus::StatusUTXOValid)));
     assert!(consensus.lkg_virtual_state.load().accepted_tx_ids.contains(&tx_id));
+    consensus.shutdown(wait_handles);
+    drop(consensus);
 }
 
 #[tokio::test]
@@ -2114,7 +2119,7 @@ async fn runtime_sig_op_counting_test() {
     let mut genesis_multiset = MuHash::new();
     consensus.append_imported_pruning_point_utxos(&initial_utxo_collection, &mut genesis_multiset);
     consensus.import_pruning_point_utxo_set(config.genesis.hash, genesis_multiset).unwrap();
-    consensus.init();
+    let wait_handles = consensus.init();
 
     // Build blockchain up to one block before activation
     let mut index = 0;
@@ -2184,4 +2189,6 @@ async fn runtime_sig_op_counting_test() {
     // Test 2: After activation, tx should be accepted as runtime counting only sees 1 executed sig op
     let status = consensus.add_utxo_valid_block_with_parents((index + 1).into(), vec![index.into()], vec![tx]).await;
     assert!(matches!(status, Ok(BlockStatus::StatusUTXOValid)));
+    consensus.shutdown(wait_handles);
+    drop(consensus);
 }

wallet/core/src/tx/generator/test.rs

@@ -721,7 +721,9 @@ fn test_generator_inputs_32k_outputs_2_fees_exclude() -> Result<()> {
 #[test]
 fn test_generator_inputs_250k_outputs_2_sweep() -> Result<()> {
     let f = 130.0;
-    let generator = make_generator(test_network_id(), &[f; 250_000], &[], Fees::None, change_address, PaymentDestination::Change);
+    let head: Vec<f64> = vec![f; 250_000]; // <-- NOTE: this is allocated in heap unlike before, to
+                                           // prevent stackoverflow
+    let generator = make_generator(test_network_id(), &head, &[], Fees::None, change_address, PaymentDestination::Change);
     generator.unwrap().harness().accumulate(2875).finalize();
     Ok(())
 }