Fix batch replacement bug causing lost gap-limit addresses after reconnect

When start_download() replaces an active batch after disconnect/reconnect,
account for existing blocks_remaining entries by setting initial
pending_blocks count on the replacement batch. Also preserve collected
addresses from the old batch. Without this, the replacement batch commits
prematurely when its own pending_blocks reaches zero, even though old
blocks referencing the same batch_start are still being processed and
may generate gap-limit addresses that need rescanning.

Author: xdustinface

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

@@ -257,6 +257,35 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         if stored_filters_tip >= batch_end {
             batch.mark_verified();
         }
+
+        // When replacing an existing batch (e.g. after disconnect/reconnect),
+        // preserve collected addresses and account for blocks that are still
+        // pending in blocks_remaining. Without this, the replacement batch
+        // would commit prematurely when its own pending_blocks reaches 0,
+        // even though old blocks referencing this batch_start are still being
+        // processed and may generate gap-limit addresses.
+        if let Some(mut old_batch) = self.active_batches.remove(&scan_start) {
+            let old_addresses = old_batch.take_collected_addresses();
+            if !old_addresses.is_empty() {
+                batch.add_addresses(old_addresses.iter().cloned());
+                tracing::debug!(
+                    "Preserved {} collected addresses from replaced batch {}",
+                    old_addresses.len(),
+                    scan_start,
+                );
+            }
+        }
+        let existing_remaining =
+            self.blocks_remaining.values().filter(|(_, bs)| *bs == scan_start).count() as u32;
+        if existing_remaining > 0 {
+            batch.set_pending_blocks(existing_remaining);
+            tracing::debug!(
+                "Batch {} has {} blocks still pending from previous scan",
+                scan_start,
+                existing_remaining,
+            );
+        }
+
         self.active_batches.insert(scan_start, batch);
         self.committed_height = scan_start.saturating_sub(1);
 
@@ -776,12 +805,13 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::network::MessageType;
+    use crate::network::{MessageType, NetworkManager};
     use crate::storage::{
         DiskStorageManager, PersistentBlockHeaderStorage, PersistentFilterHeaderStorage,
         PersistentFilterStorage, StorageManager,
     };
     use crate::sync::{ManagerIdentifier, SyncManagerProgress};
+    use crate::test_utils::MockNetworkManager;
     use key_wallet_manager::test_utils::MockWallet;
 
     type TestFiltersManager = FiltersManager<
@@ -949,4 +979,118 @@ mod tests {
         // After take, should be empty
         assert!(batch.take_collected_addresses().is_empty());
     }
+
+    #[tokio::test]
+    async fn test_replacement_batch_accounts_for_existing_blocks_remaining() {
+        // Verifies that when a batch has pending_blocks correctly accounting for
+        // existing blocks_remaining entries (as set up by start_download after
+        // reconnect), the batch does not commit until all blocks are processed.
+        use dashcore::Network;
+
+        let mut manager = create_test_manager().await;
+        manager.set_state(SyncState::Syncing);
+
+        let hash1 = dashcore::block::Header::dummy(0).block_hash();
+        let hash2 = dashcore::block::Header::dummy(1).block_hash();
+
+        // Simulate post-replacement state: batch with pending_blocks=2
+        // matching the 2 entries in blocks_remaining
+        let mut batch = FiltersBatch::new(0, 4999, HashMap::new());
+        batch.mark_scanned();
+        batch.set_pending_blocks(2);
+        manager.active_batches.insert(0, batch);
+        manager.blocks_remaining.insert(hash1, (100, 0));
+        manager.blocks_remaining.insert(hash2, (200, 0));
+
+        let network = MockNetworkManager::new();
+        let requests = network.request_sender();
+
+        // Process first block with new addresses (gap limit discovery)
+        let addr = dashcore::Address::dummy(Network::Testnet, 1);
+        let event = SyncEvent::BlockProcessed {
+            block_hash: hash1,
+            height: 100,
+            new_addresses: vec![addr],
+        };
+        let manager_ref: &mut dyn SyncManager = &mut manager;
+        manager_ref.handle_sync_event(&event, &requests).await.unwrap();
+
+        // Batch should still be active with 1 block remaining
+        assert!(
+            manager.active_batches.contains_key(&0),
+            "batch must not commit with 1 block still remaining"
+        );
+        assert_eq!(manager.active_batches.get(&0).unwrap().pending_blocks(), 1);
+
+        // Process second block
+        let event = SyncEvent::BlockProcessed {
+            block_hash: hash2,
+            height: 200,
+            new_addresses: vec![],
+        };
+        let manager_ref: &mut dyn SyncManager = &mut manager;
+        manager_ref.handle_sync_event(&event, &requests).await.unwrap();
+
+        // Batch should have committed (pending_blocks=0, rescan ran with empty filters)
+        assert!(
+            !manager.active_batches.contains_key(&0),
+            "batch should commit after all blocks processed"
+        );
+        assert_eq!(manager.committed_height, 4999);
+    }
+
+    #[tokio::test]
+    async fn test_zero_pending_blocks_causes_premature_batch_commit() {
+        // Documents the bug scenario: if pending_blocks doesn't account for
+        // existing blocks_remaining (the pre-fix behavior), the batch commits
+        // after the first BlockProcessed event, and addresses from subsequent
+        // blocks are silently lost.
+        use dashcore::Network;
+
+        let mut manager = create_test_manager().await;
+        manager.set_state(SyncState::Syncing);
+
+        let hash1 = dashcore::block::Header::dummy(0).block_hash();
+        let hash2 = dashcore::block::Header::dummy(1).block_hash();
+
+        // Bug scenario: batch has pending_blocks=0 (not accounting for existing blocks)
+        let mut batch = FiltersBatch::new(0, 4999, HashMap::new());
+        batch.mark_scanned();
+        batch.set_pending_blocks(0);
+        manager.active_batches.insert(0, batch);
+        manager.blocks_remaining.insert(hash1, (100, 0));
+        manager.blocks_remaining.insert(hash2, (200, 0));
+
+        let network = MockNetworkManager::new();
+        let requests = network.request_sender();
+
+        // Process first block with new addresses
+        let addr = dashcore::Address::dummy(Network::Testnet, 1);
+        let event = SyncEvent::BlockProcessed {
+            block_hash: hash1,
+            height: 100,
+            new_addresses: vec![addr],
+        };
+        let manager_ref: &mut dyn SyncManager = &mut manager;
+        manager_ref.handle_sync_event(&event, &requests).await.unwrap();
+
+        // Bug: batch committed prematurely because pending_blocks was 0
+        assert!(
+            !manager.active_batches.contains_key(&0),
+            "with pending_blocks=0, batch commits after first block"
+        );
+
+        // Process second block - batch is gone, addresses are silently lost
+        let addr2 = dashcore::Address::dummy(Network::Testnet, 2);
+        let event = SyncEvent::BlockProcessed {
+            block_hash: hash2,
+            height: 200,
+            new_addresses: vec![addr2],
+        };
+        let manager_ref: &mut dyn SyncManager = &mut manager;
+        let events = manager_ref.handle_sync_event(&event, &requests).await.unwrap();
+
+        // No events emitted because batch was already committed and removed
+        assert!(events.is_empty(), "addresses from hash2 are lost");
+    }
 }

tasks/todo.md

@@ -1,26 +1,36 @@
-# Fix block re-processing stalling sync on restart
+# Fix flaky test_sync_with_peer_disconnection
 
 ## Problem
-On restart, `BlocksManager` loads block 68 from storage and re-processes it through the wallet.
-The wallet returns 0 new transactions and 0 new addresses (complete no-op), but `BlockProcessed`
-is still emitted. The test restarts on this event before any network downloads complete, creating
-an infinite loop stuck at block 68.
+After disconnect/reconnect, `FiltersManager.start_download()` creates a new batch at `scan_start`
+and inserts it into `active_batches`, replacing any existing batch at the same key. The new batch's
+`pending_blocks` only counts newly discovered blocks (those not in `filters_matched`), but
+`blocks_remaining` still has entries from the original scan pointing to the same `batch_start`.
 
-We can't simply skip re-processing because gap limit rescanning legitimately needs to re-process
-blocks with newly discovered addresses. Height-based checks would break that.
+When the first old block is processed, it decrements `pending_blocks` on the replacement batch
+(where it was never counted), driving it to 0 prematurely. The batch commits and is removed from
+`active_batches`. All subsequent `BlockProcessed` events (many with gap-limit `new_addresses`)
+can't find their batch, so addresses are silently lost, rescans never happen, and transactions at
+those addresses are never found.
 
-## Approach
-Add `new_tx_count` to `BlockProcessed` event to distinguish productive processing from no-op
-re-processing. The wallet's `process_block` already returns this info (`BlockProcessingResult`)
-but it wasn't exposed in the event. The `FiltersManager` handles the event identically regardless.
-The test (and other consumers) can use this to distinguish real progress from redundant processing.
+## Root Cause
+`start_download()` line 260: `self.active_batches.insert(scan_start, batch)` replaces the old batch.
+The new batch starts with `pending_blocks = 0` and `scan_batch` only adds the count of blocks not
+already in `filters_matched`. But `blocks_remaining` has entries from the old scan that will still
+trigger `decrement_pending_blocks` on this new batch.
+
+## Fix
+In `start_download()`, when creating a replacement batch at an existing `scan_start`:
+1. Preserve `collected_addresses` from the old batch (already-discovered gap-limit addresses)
+2. Count existing `blocks_remaining` entries for this batch_start
+3. Set initial `pending_blocks` on the new batch to that count before `scan_batch` adds more
+
+This ensures the batch won't commit until all blocks (old and new) are processed.
 
 ## Tasks
-- [ ] Add `new_tx_count: usize` field to `SyncEvent::BlockProcessed`
-- [ ] Update construction in `process_buffered_blocks` to include `new_tx_count`
-- [ ] Update pattern matches in FiltersManager, events description, and tests
-- [ ] Update test `is_restart_event` to only trigger on productive `BlockProcessed` events
-- [ ] Verify compilation and tests pass
+- [ ] In `start_download()`, before inserting new batch: extract old batch state and count blocks_remaining
+- [ ] Set initial `pending_blocks` and transfer collected_addresses to new batch
+- [ ] Add unit test for batch replacement with existing blocks_remaining
+- [ ] Run `test_sync_with_peer_disconnection` multiple times to verify fix
 - [ ] Commit
 
 ## Review