fix: reset FilterManager in flight state when all peers disconnect

dash-spv/src/sync/filters/manager.rs

@@ -102,6 +102,16 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         }
     }
 
+    /// Clear all in-flight processing state. Called on peer disconnect when
+    /// in-flight filter batches and block tracking become invalid.
+    pub(super) fn clear_in_flight_state(&mut self) {
+        self.active_batches.clear();
+        self.blocks_remaining.clear();
+        self.filters_matched.clear();
+        self.pending_batches.clear();
+        self.filter_pipeline = FiltersPipeline::new();
+    }
+
     async fn load_filters(
         &self,
         start_height: u32,

dash-spv/src/sync/filters/sync_manager.rs

@@ -36,6 +36,11 @@ impl<
         &[MessageType::CFilter]
     }
 
+    fn stop_sync(&mut self) {
+        self.set_state(SyncState::WaitingForConnections);
+        self.clear_in_flight_state();
+    }
+
     async fn initialize(&mut self) -> SyncResult<()> {
         let wallet = self.wallet.read().await;
         let synced_height = wallet.synced_height();
@@ -80,6 +85,9 @@ impl<
         // Already at or beyond stored filters tip - check if fully synced
         if stored_filters_tip > 0 && stored_filters_tip == self.progress.current_height() {
             self.progress.update_filter_header_tip_height(stored_filters_tip);
+            // Initialize the pipeline at the current tip. On full disconnect in-flight state gets
+            // reset, so we need to initialize the pipeline otherwise it would re-queue from height 1.
+            self.filter_pipeline.init(stored_filters_tip + 1, stored_filters_tip);
             // Only emit SyncComplete if we've also reached the chain tip
             if self.progress.current_height() >= self.progress.target_height() {
                 self.set_state(SyncState::Synced);