Merge branch 'v0.40-dev' into fix/syncProgress

Author: QuantumExplorer

.github/workflows/rust.yml

@@ -329,6 +329,7 @@ jobs:
 
   integrations_tests:
     name: Integration Tests
+    if: ${{ false }} # Temporarily disabled
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -356,4 +357,3 @@ jobs:
         uses: reviewdog/action-actionlint@v1
         with:
           fail_on_error: true
-

dash-network-ffi/src/lib.rs

@@ -1,5 +1,7 @@
 //! FFI bindings for dash-network library
 
+use std::fmt;
+
 use dash_network::Network as DashNetwork;
 
 // Initialize function
@@ -70,10 +72,6 @@ impl NetworkInfo {
         self.network.magic()
     }
 
-    pub fn to_string(&self) -> String {
-        self.network.to_string()
-    }
-
     pub fn is_core_v20_active(&self, block_height: u32) -> bool {
         self.network.core_v20_is_active_at(block_height)
     }
@@ -83,6 +81,12 @@ impl NetworkInfo {
     }
 }
 
+impl fmt::Display for NetworkInfo {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.network.fmt(f)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

dash-spv-ffi/src/bin/ffi_cli.rs

@@ -215,9 +215,6 @@ fn main() {
                     break;
                 }
                 dash_spv_ffi_sync_progress_destroy(prog_ptr);
-            } else {
-                // If progress is unavailable, assume sync finished or errored
-                break;
             }
             thread::sleep(Duration::from_millis(300));
         }

dash-spv/src/client/message_handler.rs

@@ -7,6 +7,7 @@ use crate::network::NetworkManager;
 use crate::storage::StorageManager;
 use crate::sync::sequential::SequentialSyncManager;
 use crate::types::{MempoolState, SpvEvent, SpvStats};
+// Removed local ad-hoc compact filter construction in favor of always processing full blocks
 use key_wallet_manager::wallet_interface::WalletInterface;
 use std::sync::Arc;
 use tokio::sync::RwLock;
@@ -238,7 +239,23 @@ impl<
                     block.txdata.len()
                 );
 
-                // Process new block (update state, check watched items)
+                // 1) Ensure header processing and chain tip update for this block
+                //    Route the header through the sequential sync manager as a Headers message
+                let headers_msg = NetworkMessage::Headers(vec![block.header]);
+                if let Err(e) = self
+                    .sync_manager
+                    .handle_message(headers_msg, &mut *self.network, &mut *self.storage)
+                    .await
+                {
+                    tracing::error!(
+                        "❌ Failed to process header for block {} via sync manager: {}",
+                        block_hash,
+                        e
+                    );
+                    return Err(SpvError::Sync(e));
+                }
+
+                // 2) Always process the full block (privacy and correctness)
                 if let Err(e) = self.process_new_block(block).await {
                     tracing::error!("❌ Failed to process new block {}: {}", block_hash, e);
                     return Err(e);
@@ -434,31 +451,16 @@ impl<
             self.network.send_message(getdata).await.map_err(SpvError::Network)?;
         }
 
-        // Process new blocks immediately when detected
+        // For blocks announced via inventory during tip sync, request full blocks for privacy
         if !blocks_to_request.is_empty() {
             tracing::info!(
-                "🔄 Processing {} new block announcements to stay synchronized",
+                "📥 Requesting {} new blocks announced via inventory",
                 blocks_to_request.len()
             );
 
-            // Extract block hashes
-            let block_hashes: Vec<dashcore::BlockHash> = blocks_to_request
-                .iter()
-                .filter_map(|inv| {
-                    if let Inventory::Block(hash) = inv {
-                        Some(*hash)
-                    } else {
-                        None
-                    }
-                })
-                .collect();
-
-            // Process each new block
-            for block_hash in block_hashes {
-                tracing::info!("📥 Requesting header for new block {}", block_hash);
-                if let Err(e) = self.process_new_block_hash(block_hash).await {
-                    tracing::error!("❌ Failed to process new block {}: {}", block_hash, e);
-                }
+            let getdata = NetworkMessage::GetData(blocks_to_request);
+            if let Err(e) = self.network.send_message(getdata).await {
+                tracing::error!("Failed to request announced blocks: {}", e);
             }
         }
 

dash-spv/src/client/mod.rs

@@ -935,14 +935,12 @@ impl<
 
                 // Emit detailed progress update
                 if last_rate_calc.elapsed() >= Duration::from_secs(1) {
-                    // Storage tip is the headers vector index (0-based).
-                    let current_storage_tip = {
+                    // Storage tip now represents the absolute blockchain height.
+                    let current_tip_height = {
                         let storage = self.storage.lock().await;
                         storage.get_tip_height().await.ok().flatten().unwrap_or(0)
                     };
-                    // Convert to absolute blockchain height: base + storage_tip
-                    let sync_base_height = { self.state.read().await.sync_base_height };
-                    let current_height = sync_base_height + current_storage_tip;
+                    let current_height = current_tip_height;
                     let peer_best = self
                         .network
                         .get_peer_best_height()
@@ -952,9 +950,9 @@ impl<
                         .unwrap_or(current_height);
 
                     // Calculate headers downloaded this second
-                    if current_storage_tip > last_height {
-                        headers_this_second = current_storage_tip - last_height;
-                        last_height = current_storage_tip;
+                    if current_tip_height > last_height {
+                        headers_this_second = current_tip_height - last_height;
+                        last_height = current_tip_height;
                     }
 
                     let headers_per_second = headers_this_second as f64;
@@ -1011,18 +1009,13 @@ impl<
                     last_rate_calc = Instant::now();
                 }
 
-                // Emit a detailed progress snapshot when filter/header heights change
-                let (_sync_base_height, abs_header_height, filter_header_height) = {
-                    let (storage_tip, filter_tip) = {
-                        let storage = self.storage.lock().await;
-                        let storage_tip =
-                            storage.get_tip_height().await.ok().flatten().unwrap_or(0);
-                        let filter_tip =
-                            storage.get_filter_tip_height().await.ok().flatten().unwrap_or(0);
-                        (storage_tip, filter_tip)
-                    };
-                    let base = { self.state.read().await.sync_base_height };
-                    (base, base + storage_tip, filter_tip)
+                // Emit filter headers progress only when heights change
+                let (abs_header_height, filter_header_height) = {
+                    let storage = self.storage.lock().await;
+                    let storage_tip = storage.get_tip_height().await.ok().flatten().unwrap_or(0);
+                    let filter_tip =
+                        storage.get_filter_tip_height().await.ok().flatten().unwrap_or(0);
+                    (storage_tip, filter_tip)
                 };
 
                 {
@@ -1879,8 +1872,13 @@ impl<
         let mut loaded_count = 0u32;
         let target_height = saved_state.chain_tip.height;
 
-        // Start from height 1 (genesis is already in ChainState)
-        let mut current_height = 1u32;
+        // Determine first height to load. Skip genesis (already present) unless we started from a checkpoint base.
+        let mut current_height =
+            if saved_state.synced_from_checkpoint && saved_state.sync_base_height > 0 {
+                saved_state.sync_base_height
+            } else {
+                1u32
+            };
 
         while current_height <= target_height {
             let end_height = (current_height + BATCH_SIZE - 1).min(target_height);
@@ -1895,12 +1893,12 @@ impl<
             };
 
             if headers.is_empty() {
-                tracing::error!(
-                    "Failed to load headers for range {}..{} - storage may be corrupted",
+                tracing::warn!(
+                    "No headers found for range {}..{} when restoring from state",
                     current_height,
                     end_height + 1
                 );
-                return Ok(false);
+                break;
             }
 
             // Validate headers before adding to chain state

dash-spv/src/client/status_display.rs

@@ -48,13 +48,13 @@ impl<'a, S: StorageManager + Send + Sync + 'static> StatusDisplay<'a, S> {
         // For checkpoint sync: height = checkpoint_height + storage_count
         let storage = self.storage.lock().await;
         if let Ok(Some(storage_tip)) = storage.get_tip_height().await {
-            let blockchain_height = state.sync_base_height + storage_tip;
+            let blockchain_height = storage_tip;
             if with_logging {
                 tracing::debug!(
-                    "Status display: storage_tip={}, sync_base={}, blockchain_height={}",
-                    storage_tip,
+                    "Status display: reported tip height={}, sync_base={}, raw_storage_tip={}",
+                    blockchain_height,
                     state.sync_base_height,
-                    blockchain_height
+                    storage_tip
                 );
             }
             blockchain_height

dash-spv/src/network/multi_peer.rs

@@ -3,7 +3,7 @@
 use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 use tokio::sync::{mpsc, Mutex};
@@ -70,6 +70,8 @@ pub struct MultiPeerNetworkManager {
     user_agent: Option<String>,
     /// Exclusive mode: restrict to configured peers only (no DNS or peer store)
     exclusive_mode: bool,
+    /// Cached count of currently connected peers for fast, non-blocking queries
+    connected_peer_count: Arc<AtomicUsize>,
 }
 
 impl MultiPeerNetworkManager {
@@ -123,6 +125,7 @@ impl MultiPeerNetworkManager {
             peers_sent_headers2: Arc::new(Mutex::new(HashSet::new())),
             user_agent: config.user_agent.clone(),
             exclusive_mode,
+            connected_peer_count: Arc::new(AtomicUsize::new(0)),
         })
     }
 
@@ -209,6 +212,7 @@ impl MultiPeerNetworkManager {
         let mempool_strategy = self.mempool_strategy;
         let read_timeout = self.read_timeout;
         let user_agent = self.user_agent.clone();
+        let connected_peer_count = self.connected_peer_count.clone();
 
         // Spawn connection task
         let mut tasks = self.tasks.lock().await;
@@ -235,6 +239,9 @@ impl MultiPeerNetworkManager {
                                 return;
                             }
 
+                            // Increment connected peer counter on successful add
+                            connected_peer_count.fetch_add(1, Ordering::Relaxed);
+
                             // Add to known addresses
                             addrv2_handler.add_known_address(addr, ServiceFlags::from(1)).await;
 
@@ -246,6 +253,7 @@ impl MultiPeerNetworkManager {
                                 addrv2_handler,
                                 shutdown,
                                 reputation_manager.clone(),
+                                connected_peer_count.clone(),
                             )
                             .await;
                         }
@@ -287,6 +295,7 @@ impl MultiPeerNetworkManager {
         addrv2_handler: Arc<AddrV2Handler>,
         shutdown: Arc<AtomicBool>,
         reputation_manager: Arc<PeerReputationManager>,
+        connected_peer_count: Arc<AtomicUsize>,
     ) {
         tokio::spawn(async move {
             log::debug!("Starting peer reader loop for {}", addr);
@@ -551,7 +560,11 @@ impl MultiPeerNetworkManager {
 
             // Remove from pool
             log::warn!("Disconnecting from {} (peer reader loop ended)", addr);
-            pool.remove_connection(&addr).await;
+            let removed = pool.remove_connection(&addr).await;
+            if removed.is_some() {
+                // Decrement connected peer counter when a connection is removed
+                connected_peer_count.fetch_sub(1, Ordering::Relaxed);
+            }
 
             // Give small positive reputation if peer maintained long connection
             let conn_duration = Duration::from_secs(60 * loop_iteration); // Rough estimate
@@ -576,6 +589,7 @@ impl MultiPeerNetworkManager {
         let peer_search_started = self.peer_search_started.clone();
         let initial_peers = self.initial_peers.clone();
         let data_dir = self.data_dir.clone();
+        let connected_peer_count = self.connected_peer_count.clone();
 
         // Check if we're in exclusive mode (explicit flag or peers configured)
         let exclusive_mode = self.exclusive_mode;
@@ -597,6 +611,8 @@ impl MultiPeerNetworkManager {
 
                 let count = pool.connection_count().await;
                 log::debug!("Connected peers: {}", count);
+                // Keep the cached counter in sync with actual pool count
+                connected_peer_count.store(count, Ordering::Relaxed);
                 if exclusive_mode {
                     // In exclusive mode, only reconnect to originally specified peers
                     for addr in initial_peers.iter() {
@@ -987,6 +1003,7 @@ impl Clone for MultiPeerNetworkManager {
             peers_sent_headers2: self.peers_sent_headers2.clone(),
             user_agent: self.user_agent.clone(),
             exclusive_mode: self.exclusive_mode,
+            connected_peer_count: self.connected_peer_count.clone(),
         }
     }
 }
@@ -1068,19 +1085,13 @@ impl NetworkManager for MultiPeerNetworkManager {
     }
 
     fn is_connected(&self) -> bool {
-        // We're "connected" if we have at least one peer
-        let pool = self.pool.clone();
-        let count = tokio::task::block_in_place(move || {
-            tokio::runtime::Handle::current().block_on(pool.connection_count())
-        });
-        count > 0
+        // Use cached counter to avoid blocking in async context
+        self.connected_peer_count.load(Ordering::Relaxed) > 0
     }
 
     fn peer_count(&self) -> usize {
-        let pool = self.pool.clone();
-        tokio::task::block_in_place(move || {
-            tokio::runtime::Handle::current().block_on(pool.connection_count())
-        })
+        // Use cached counter to avoid blocking in async context
+        self.connected_peer_count.load(Ordering::Relaxed)
     }
 
     fn peer_info(&self) -> Vec<PeerInfo> {