feat: enhance filter synchronization with flow control and request tracking

Author: PastaPastaPasta

dash-spv/src/client/config.rs

@@ -56,6 +56,15 @@ pub struct ClientConfig {
 
     /// Log level for tracing.
     pub log_level: String,
+    
+    /// Maximum concurrent filter requests (default: 8).
+    pub max_concurrent_filter_requests: usize,
+    
+    /// Enable flow control for filter requests (default: true).
+    pub enable_filter_flow_control: bool,
+    
+    /// Delay between filter requests in milliseconds (default: 50).
+    pub filter_request_delay_ms: u64,
 }
 
 impl Default for ClientConfig {
@@ -76,6 +85,9 @@ impl Default for ClientConfig {
             max_peers: 8,
             enable_persistence: true,
             log_level: "info".to_string(),
+            max_concurrent_filter_requests: 16,
+            enable_filter_flow_control: true,
+            filter_request_delay_ms: 0,
         }
     }
 }
@@ -159,6 +171,24 @@ impl ClientConfig {
         self
     }
 
+    /// Set maximum concurrent filter requests.
+    pub fn with_max_concurrent_filter_requests(mut self, max_requests: usize) -> Self {
+        self.max_concurrent_filter_requests = max_requests;
+        self
+    }
+    
+    /// Enable or disable filter flow control.
+    pub fn with_filter_flow_control(mut self, enabled: bool) -> Self {
+        self.enable_filter_flow_control = enabled;
+        self
+    }
+    
+    /// Set delay between filter requests.
+    pub fn with_filter_request_delay(mut self, delay_ms: u64) -> Self {
+        self.filter_request_delay_ms = delay_ms;
+        self
+    }
+    
     /// Validate the configuration.
     pub fn validate(&self) -> Result<(), String> {
         if self.peers.is_empty() {
@@ -177,6 +207,10 @@ impl ClientConfig {
             return Err("max_peers must be > 0".to_string());
         }
 
+        if self.max_concurrent_filter_requests == 0 {
+            return Err("max_concurrent_filter_requests must be > 0".to_string());
+        }
+        
         Ok(())
     }
 

dash-spv/src/client/filter_sync.rs

@@ -55,8 +55,9 @@ impl<'a> FilterSyncCoordinator<'a> {
         }
         drop(running);
 
-        // Get current tip height to determine range
-        let tip_height = self.storage.get_tip_height().await
+        // Get current filter tip height to determine range (use filter headers, not block headers)
+        // This ensures consistency between range calculation and progress tracking
+        let tip_height = self.storage.get_filter_tip_height().await
             .map_err(|e| SpvError::Storage(e))?
             .unwrap_or(0);
 
@@ -79,7 +80,7 @@ impl<'a> FilterSyncCoordinator<'a> {
         let start_height = earliest_height.min(default_start); // Go back to the earliest required height
         let actual_count = tip_height - start_height + 1; // Actual number of blocks available
 
-        tracing::info!("Requesting filters from height {} to {} ({} blocks)", 
+        tracing::info!("Requesting filters from height {} to {} ({} blocks based on filter tip height)", 
                       start_height, tip_height, actual_count);
         tracing::info!("Filter processing and matching will happen automatically in background thread as CFilter messages arrive");
 
@@ -91,45 +92,30 @@ impl<'a> FilterSyncCoordinator<'a> {
         Ok(Vec::new())
     }
 
-    /// Sync filters in coordination with the monitoring loop using simplified processing
+    /// Sync filters in coordination with the monitoring loop using flow control processing
     async fn sync_filters_coordinated(&mut self, start_height: u32, count: u32) -> Result<()> {
-        let end_height = start_height + count - 1;
-        
-        tracing::info!("Starting coordinated filter sync from height {} to {} ({} filters expected)", 
-                      start_height, end_height, count);
+        tracing::info!("Starting coordinated filter sync with flow control from height {} to {} ({} filters expected)", 
+                      start_height, start_height + count - 1, count);
 
         // Start tracking filter sync progress
         crate::sync::filters::FilterSyncManager::start_filter_sync_tracking(
             self.stats, 
             count as u64
         ).await;
 
-        // Use batch processing to send filter requests
-        let batch_size = 100;
-        let mut current_height = start_height;
-        let mut batches_sent = 0;
-        
-        // Send all filter requests in batches
-        while current_height <= end_height {
-            let batch_end = (current_height + batch_size - 1).min(end_height);
-            
-            tracing::debug!("Sending batch {}: heights {} to {}", batches_sent + 1, current_height, batch_end);
-            
-            // Get stop hash for this batch
-            let stop_hash = self.storage.get_header(batch_end).await
-                .map_err(|e| SpvError::Storage(e))?
-                .ok_or_else(|| SpvError::Config("Stop header not found".to_string()))?
-                .block_hash();
-            
-            // Send the request - monitoring loop will handle the responses via filter processor
-            self.sync_manager.filter_sync_mut().request_filters(&mut *self.network, current_height, stop_hash).await
-                .map_err(|e| SpvError::Sync(e))?;
-            
-            current_height = batch_end + 1;
-            batches_sent += 1;
-        }
+        // Use the new flow control method
+        self.sync_manager.filter_sync_mut()
+            .sync_filters_with_flow_control(
+                &mut *self.network,
+                &mut *self.storage,
+                Some(start_height),
+                Some(count)
+            ).await
+            .map_err(|e| SpvError::Sync(e))?;
 
-        tracing::info!("✅ All filter requests sent ({} batches), processing via filter processor thread", batches_sent);
+        let (pending_count, active_count, flow_enabled) = self.sync_manager.filter_sync().get_flow_control_status();
+        tracing::info!("✅ Filter sync with flow control initiated (flow control enabled: {}, {} requests queued, {} active)", 
+                      flow_enabled, pending_count, active_count);
 
         Ok(())
     }

dash-spv/src/client/message_handler.rs

@@ -177,8 +177,19 @@ impl<'a> MessageHandler<'a> {
             NetworkMessage::CFilter(cfilter) => {
                 tracing::debug!("Received CFilter for block {}", cfilter.block_hash);
 
-                // Let the sync manager handle sync coordination (just tracking, not the full filter)
-                if let Err(e) = self.sync_manager.handle_cfilter_message(cfilter.block_hash, &mut *self.storage).await {
+                // Record the height of this received filter for gap tracking
+                crate::sync::filters::FilterSyncManager::record_filter_received_at_height(
+                    self.stats, 
+                    &*self.storage, 
+                    &cfilter.block_hash
+                ).await;
+                
+                // Enhanced sync coordination with flow control
+                if let Err(e) = self.sync_manager.handle_cfilter_message(
+                    cfilter.block_hash, 
+                    &mut *self.storage, 
+                    &mut *self.network
+                ).await {
                     tracing::error!("Failed to handle CFilter in sync manager: {}", e);
                 }
 

dash-spv/src/client/mod.rs

@@ -221,8 +221,8 @@ impl DashSpvClient {
         // Create shared data structures
         let watch_items = Arc::new(RwLock::new(HashSet::new()));
 
-        // Create sync manager
-        let sync_manager = SyncManager::new(&config);
+        // Create sync manager with shared filter heights
+        let sync_manager = SyncManager::new(&config, stats.read().await.received_filter_heights.clone());
 
         // Create validation manager
         let validation = ValidationManager::new(config.validation_mode);
@@ -449,6 +449,10 @@ impl DashSpvClient {
         let mut last_consistency_check = Instant::now();
         let consistency_check_interval = std::time::Duration::from_secs(300); // Every 5 minutes
 
+        // Timer for filter gap checking
+        let mut last_filter_gap_check = Instant::now();
+        let filter_gap_check_interval = std::time::Duration::from_secs(10);
+
         loop {
             // Check if we should stop
             let running = self.running.read().await;
@@ -506,13 +510,40 @@ impl DashSpvClient {
             if last_status_update.elapsed() >= status_update_interval {
                 self.update_status_display().await;
 
-                // Report filter sync progress if active
-                let (filters_requested, filters_received, progress, timeout) = 
-                    crate::sync::filters::FilterSyncManager::get_filter_sync_status(&self.stats).await;
+                // Report enhanced filter sync progress if active
+                let (filters_requested, filters_received, basic_progress, timeout, total_missing, actual_coverage, missing_ranges) = 
+                    crate::sync::filters::FilterSyncManager::get_filter_sync_status_with_gaps(&self.stats, self.sync_manager.filter_sync()).await;
 
                 if filters_requested > 0 {
-                    tracing::info!("📊 Filter sync progress: {:.1}% ({}/{} filters received)", 
-                                  progress, filters_received, filters_requested);
+                    // Check if sync is truly complete: both basic progress AND gap analysis must indicate completion
+                    // This fixes a bug where "Complete!" was shown when only gap analysis returned 0 missing filters
+                    // but basic progress (filters_received < filters_requested) indicated incomplete sync.
+                    let is_complete = filters_received >= filters_requested && total_missing == 0;
+                    
+                    // Debug logging for completion detection
+                    if filters_received >= filters_requested && total_missing > 0 {
+                        tracing::debug!("🔍 Completion discrepancy detected: basic progress complete ({}/{}) but {} missing filters detected", 
+                                       filters_received, filters_requested, total_missing);
+                    }
+                    
+                    if !is_complete {
+                        tracing::info!("📊 Filter sync: Basic {:.1}% ({}/{}), Actual coverage {:.1}%, Missing: {} filters in {} ranges", 
+                                      basic_progress, filters_received, filters_requested, actual_coverage, total_missing, missing_ranges.len());
+                        
+                        // Show first few missing ranges for debugging
+                        if missing_ranges.len() > 0 {
+                            let show_count = missing_ranges.len().min(3);
+                            for (i, (start, end)) in missing_ranges.iter().enumerate().take(show_count) {
+                                tracing::warn!("  Gap {}: range {}-{} ({} filters)", i + 1, start, end, end - start + 1);
+                            }
+                            if missing_ranges.len() > show_count {
+                                tracing::warn!("  ... and {} more gaps", missing_ranges.len() - show_count);
+                            }
+                        }
+                    } else {
+                        tracing::info!("📊 Filter sync progress: {:.1}% ({}/{} filters received) - Complete!", 
+                                      basic_progress, filters_received, filters_requested);
+                    }
 
                     if timeout {
                         tracing::warn!("⚠️  Filter sync timeout: no filters received in 30+ seconds");
@@ -549,6 +580,17 @@ impl DashSpvClient {
                 last_consistency_check = Instant::now();
             }
 
+            // Check for missing filters and retry periodically
+            if last_filter_gap_check.elapsed() >= filter_gap_check_interval {
+                if self.config.enable_filters {
+                    if let Err(e) = self.sync_manager.filter_sync_mut()
+                        .check_and_retry_missing_filters(&mut *self.network, &*self.storage).await {
+                        tracing::warn!("Failed to check and retry missing filters: {}", e);
+                    }
+                }
+                last_filter_gap_check = Instant::now();
+            }
+
             // Handle network messages
             match self.network.receive_message().await {
                 Ok(Some(message)) => {