feat: rescan filters when a wallet falls behind committed height

When `wallet.synced_height()` drops below `FiltersManager`'s `progress.committed_height()`, a wallet was added or moved behind the current scan position and needs catch-up coverage. Add a check at the top of `FiltersManager::tick()` that detects this regression, clears in-flight pipeline state, lowers `committed_height` to the new aggregate min, and re-enters `start_download()`. The check runs in `Syncing`, `Synced`, and `WaitForEvents` states so idle additions are caught on the next 100ms tick.

Add `test_wallet_added_at_runtime_catches_up` in `dash-spv/tests/dashd_sync/tests_basic.rs`. After initial sync with `W1`, mine a block funding `W2`'s address and add `W2` at runtime with `birth_height` before that block. Assert the rescan picks up `W2`'s funding transaction and `W1`'s state is unchanged. Then add `W3` with `birth_height` beyond the tip and assert no spurious rescan or regression in either existing wallet.

Author: xdustinface

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

@@ -1312,4 +1312,115 @@ mod tests {
         assert_eq!(manager.state(), SyncState::Synced);
         assert!(events.is_empty());
     }
+
+    /// A wallet whose `synced_height` sits below the manager's `committed_height`
+    /// must trigger a rescan from the wallet's height. This simulates a wallet
+    /// being added at runtime behind current scan progress.
+    #[tokio::test]
+    async fn test_tick_rescans_when_wallet_falls_behind_committed() {
+        let mut manager = create_test_manager().await;
+
+        // Block headers required by start_download to resolve heights.
+        let headers = dashcore::block::Header::dummy_batch(0..201);
+        manager.header_storage.write().await.store_headers(&headers).await.unwrap();
+
+        // Manager believes filters are committed up to 100, with filter headers at 200.
+        manager.set_state(SyncState::Synced);
+        manager.progress.update_committed_height(100);
+        manager.progress.update_stored_height(100);
+        manager.progress.update_filter_header_tip_height(200);
+        manager.progress.update_target_height(200);
+
+        // Pre-populate in-flight state so we can verify clear_in_flight_state runs.
+        manager.active_batches.insert(101, FiltersBatch::new(101, 200, HashMap::new()));
+        manager.filters_matched.insert(dashcore::block::Header::dummy(0).block_hash());
+        manager.filter_pipeline.init(101, 200);
+
+        // MockWallet defaults to synced_height=0, so wallets_behind(100) = {MOCK_WALLET_ID}.
+        assert_eq!(manager.wallet.read().await.synced_height(), 0);
+
+        let (tx, _rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        let _events = manager.tick(&requests).await.unwrap();
+
+        // Old in-flight state was cleared and a fresh batch was created at scan_start=0.
+        assert!(!manager.active_batches.contains_key(&101));
+        assert!(manager.active_batches.contains_key(&0));
+        assert!(manager.filters_matched.is_empty());
+
+        // committed_height was lowered and start_download set it to scan_start - 1 = 0.
+        assert_eq!(manager.progress.committed_height(), 0);
+        assert_eq!(manager.state(), SyncState::Syncing);
+    }
+
+    /// When every managed wallet is at or beyond `committed_height`, the rescan
+    /// trigger must not fire even though the aggregate `synced_height` could
+    /// otherwise look stale.
+    #[tokio::test]
+    async fn test_tick_does_not_rescan_when_no_wallets_behind() {
+        let mut manager = create_test_manager().await;
+
+        // Wallet at synced_height=200, manager committed at 100 → no wallets behind.
+        manager.wallet.write().await.update_wallet_synced_height(&MOCK_WALLET_ID, 200);
+
+        manager.set_state(SyncState::Synced);
+        manager.progress.update_committed_height(100);
+        manager.progress.update_stored_height(100);
+        manager.progress.update_filter_header_tip_height(200);
+        manager.progress.update_target_height(200);
+
+        let (tx, _rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        let events = manager.tick(&requests).await.unwrap();
+
+        assert!(events.is_empty());
+        assert_eq!(manager.progress.committed_height(), 100);
+        assert_eq!(manager.state(), SyncState::Synced);
+        assert!(manager.active_batches.is_empty());
+    }
+
+    /// `committed_height = 0` on a fresh manager must not falsely trip the
+    /// rescan trigger. `wallets_behind(0)` returns an empty set since heights
+    /// are unsigned, so no wallet can be strictly less than 0.
+    #[tokio::test]
+    async fn test_tick_does_not_rescan_at_genesis_committed() {
+        let mut manager = create_test_manager().await;
+        // Default state: committed_height=0, wallet synced_height=0, state=WaitForEvents.
+        assert_eq!(manager.progress.committed_height(), 0);
+        assert_eq!(manager.state(), SyncState::WaitForEvents);
+
+        let (tx, _rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        let events = manager.tick(&requests).await.unwrap();
+
+        assert!(events.is_empty());
+        assert!(manager.is_idle());
+        assert_eq!(manager.state(), SyncState::WaitForEvents);
+    }
+
+    /// The rescan trigger only fires in `Syncing | Synced | WaitForEvents`.
+    /// `WaitingForConnections` must be skipped since we're not actively syncing.
+    #[tokio::test]
+    async fn test_tick_does_not_rescan_in_waiting_for_connections() {
+        let mut manager = create_test_manager().await;
+        manager.set_state(SyncState::WaitingForConnections);
+        manager.progress.update_committed_height(100);
+
+        // Wallet behind committed — would normally trip the trigger.
+        assert!(!manager.wallet.read().await.wallets_behind(100).is_empty());
+
+        let (tx, _rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        let events = manager.tick(&requests).await.unwrap();
+
+        assert!(events.is_empty());
+        // committed_height not lowered, no batches created.
+        assert_eq!(manager.progress.committed_height(), 100);
+        assert_eq!(manager.state(), SyncState::WaitingForConnections);
+        assert!(manager.active_batches.is_empty());
+    }
 }

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

@@ -193,6 +193,27 @@ impl<
     }
 
     async fn tick(&mut self, requests: &RequestSender) -> SyncResult<Vec<SyncEvent>> {
+        // Detect a wallet that was added behind our scan progress and rescan
+        // from its `synced_height`.
+        if matches!(self.state(), SyncState::Syncing | SyncState::Synced | SyncState::WaitForEvents)
+        {
+            let committed = self.progress.committed_height();
+            let wallet_read = self.wallet.read().await;
+            let behind = wallet_read.wallets_behind(committed);
+            let wallet_min_synced = wallet_read.synced_height();
+            drop(wallet_read);
+            if !behind.is_empty() {
+                tracing::info!(
+                    "Wallet synced_height {} fell below filter committed_height {}, restarting scan",
+                    wallet_min_synced,
+                    committed
+                );
+                self.clear_in_flight_state();
+                self.progress.update_committed_height(wallet_min_synced);
+                return self.start_download(requests).await;
+            }
+        }
+
         // TODO: Get rid of the send pending in here? Or decouple it from the header storage?
         // Run tick when Syncing OR when Synced with pending work (new blocks arriving)
         let has_pending_work = !self.active_batches.is_empty();