fix: drop max_retries from DownloadCoordinator to prevent sync stall (#632)

* fix: drop `max_retries` from `DownloadCoordinator` to prevent sync stall

When a entry exceeded max retries it was permanently dropped from the
pipeline which lead to stalling sync. All pipeline retries now
continue indefinitely avoid giving up on required sync data.
Retry counts are kept for logging visibility.

* fix: clean up `retry_counts` entry on successful `receive`

Addresses CodeRabbit review comment on PR #632
#632 (comment)

* refactor: make all `handle_timeouts` fire-and-forget for consistency

`enqueue_retry` already logs each retry attempt, making the
manager-level timeout logging redundant. Aligns blocks and filters
pipelines with the fire-and-forget pattern already used by
filter_headers, masternodes, and block_headers.

Addresses CodeRabbit review comment on PR #632
#632 (comment)

Author: xdustinface

dash-spv/src/sync/block_headers/segment_state.rs

@@ -50,8 +50,7 @@ impl SegmentState {
             coordinator: DownloadCoordinator::new(
                 DownloadConfig::default()
                     .with_max_concurrent(1) // Only 1 request at a time (sequential getheaders)
-                    .with_timeout(HEADERS_TIMEOUT)
-                    .with_max_retries(3),
+                    .with_timeout(HEADERS_TIMEOUT),
             ),
             buffered_headers: Vec::new(),
             complete: false,

dash-spv/src/sync/blocks/pipeline.rs

@@ -19,9 +19,6 @@ const MAX_CONCURRENT_BLOCK_DOWNLOADS: usize = 20;
 /// Timeout for block downloads before retry.
 const BLOCK_TIMEOUT: Duration = Duration::from_secs(30);
 
-/// Maximum number of retries for block downloads.
-const BLOCK_MAX_RETRIES: u32 = 3;
-
 /// Maximum blocks per GetData request, kept a bit lower for better download distribution to multiple peers
 const BLOCKS_PER_REQUEST: usize = 8;
 
@@ -64,8 +61,7 @@ impl BlocksPipeline {
             coordinator: DownloadCoordinator::new(
                 DownloadConfig::default()
                     .with_max_concurrent(MAX_CONCURRENT_BLOCK_DOWNLOADS)
-                    .with_timeout(BLOCK_TIMEOUT)
-                    .with_max_retries(BLOCK_MAX_RETRIES),
+                    .with_timeout(BLOCK_TIMEOUT),
             ),
             pending_heights: BTreeSet::new(),
             downloaded: BTreeMap::new(),
@@ -173,16 +169,8 @@ impl BlocksPipeline {
     }
 
     /// Check for timed out downloads and re-queue them.
-    ///
-    /// Returns the list of timed out block hashes.
-    pub(super) fn handle_timeouts(&mut self) -> Vec<BlockHash> {
-        let timed_out = self.coordinator.check_and_retry_timeouts();
-
-        if !timed_out.is_empty() {
-            tracing::debug!("Re-queued {} timed out block downloads", timed_out.len());
-        }
-
-        timed_out
+    pub(super) fn handle_timeouts(&mut self) {
+        self.coordinator.check_and_retry_timeouts();
     }
 }
 
@@ -324,10 +312,8 @@ mod tests {
         // Wait for timeout
         std::thread::sleep(Duration::from_millis(20));
 
-        let timed_out = pipeline.handle_timeouts();
+        pipeline.handle_timeouts();
 
-        assert_eq!(timed_out.len(), 1);
-        assert_eq!(timed_out[0], hash);
         assert_eq!(pipeline.coordinator.active_count(), 0);
         assert_eq!(pipeline.coordinator.pending_count(), 1);
     }
@@ -430,53 +416,6 @@ mod tests {
         assert!(pipeline.is_complete());
     }
 
-    #[test]
-    fn test_handle_timeouts_with_multiple_retries() {
-        let mut pipeline = BlocksPipeline {
-            coordinator: DownloadCoordinator::new(
-                DownloadConfig::default()
-                    .with_max_concurrent(MAX_CONCURRENT_BLOCK_DOWNLOADS)
-                    .with_timeout(Duration::from_millis(1))
-                    .with_max_retries(2),
-            ),
-            pending_heights: BTreeSet::new(),
-            downloaded: BTreeMap::new(),
-            hash_to_height: HashMap::new(),
-        };
-
-        // Use coordinator to set up in-flight state
-        let hash = test_hash(1);
-        pipeline.coordinator.enqueue([hash]);
-        let hashes = pipeline.coordinator.take_pending(1);
-        pipeline.coordinator.mark_sent(&hashes);
-
-        // First timeout - returns item (it's re-queued)
-        std::thread::sleep(Duration::from_millis(5));
-        let timed_out = pipeline.handle_timeouts();
-        assert_eq!(timed_out.len(), 1);
-        assert_eq!(pipeline.coordinator.pending_count(), 1);
-
-        // Re-send the retry
-        let items = pipeline.coordinator.take_pending(1);
-        pipeline.coordinator.mark_sent(&items);
-
-        // Second timeout - still re-queued
-        std::thread::sleep(Duration::from_millis(5));
-        let timed_out = pipeline.handle_timeouts();
-        assert_eq!(timed_out.len(), 1);
-        assert_eq!(pipeline.coordinator.pending_count(), 1);
-
-        // Re-send
-        let items = pipeline.coordinator.take_pending(1);
-        pipeline.coordinator.mark_sent(&items);
-
-        // Third timeout - exceeds max retries, NOT re-queued
-        std::thread::sleep(Duration::from_millis(5));
-        let timed_out = pipeline.handle_timeouts();
-        assert_eq!(timed_out.len(), 0);
-        assert_eq!(pipeline.coordinator.pending_count(), 0);
-    }
-
     #[test]
     fn test_receive_block_duplicate() {
         let mut pipeline = BlocksPipeline::new();

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

@@ -183,10 +183,7 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface + 'static> SyncM
 
     async fn tick(&mut self, requests: &RequestSender) -> SyncResult<Vec<SyncEvent>> {
         // Handle timeouts
-        let timed_out = self.pipeline.handle_timeouts();
-        if !timed_out.is_empty() {
-            tracing::debug!("Re-queued {} timed out block downloads", timed_out.len());
-        }
+        self.pipeline.handle_timeouts();
 
         self.send_pending(requests).await?;
 

dash-spv/src/sync/download_coordinator.rs

@@ -17,16 +17,13 @@ pub struct DownloadConfig {
     max_concurrent: usize,
     /// Timeout duration for requests.
     timeout: Duration,
-    /// Maximum retry attempts before giving up.
-    max_retries: u32,
 }
 
 impl Default for DownloadConfig {
     fn default() -> Self {
         Self {
             max_concurrent: 10,
             timeout: Duration::from_secs(30),
-            max_retries: 3,
         }
     }
 }
@@ -43,12 +40,6 @@ impl DownloadConfig {
         self.timeout = timeout;
         self
     }
-
-    /// Create config with custom max retries.
-    pub(crate) fn with_max_retries(mut self, max: u32) -> Self {
-        self.max_retries = max;
-        self
-    }
 }
 
 /// Generic download coordinator.
@@ -109,17 +100,11 @@ impl<K: Hash + Eq + Clone> DownloadCoordinator<K> {
     }
 
     /// Queue an item for retry (goes to front of queue).
-    ///
-    /// Returns false if max retries exceeded.
-    pub(crate) fn enqueue_retry(&mut self, item: K) -> bool {
+    pub(crate) fn enqueue_retry(&mut self, item: K) {
         let count = self.retry_counts.entry(item.clone()).or_insert(0);
-        if *count >= self.config.max_retries {
-            tracing::warn!("Max retries ({}) exceeded, giving up", self.config.max_retries);
-            return false;
-        }
         *count += 1;
+        tracing::warn!("Retrying item (attempt {})", count);
         self.pending.push_front(item);
-        true
     }
 
     /// Get the number of items available to send (respecting concurrency limit).
@@ -155,6 +140,7 @@ impl<K: Hash + Eq + Clone> DownloadCoordinator<K> {
     /// Returns true if the item was being tracked, false if unexpected.
     pub(crate) fn receive(&mut self, key: &K) -> bool {
         if self.in_flight.remove(key).is_some() {
+            self.retry_counts.remove(key);
             self.last_progress = Instant::now();
             true
         } else {
@@ -194,11 +180,13 @@ impl<K: Hash + Eq + Clone> DownloadCoordinator<K> {
     /// Check for timed-out items and re-enqueue them for retry.
     ///
     /// Combines `check_timeouts()` and `enqueue_retry()` in one call.
-    /// Returns only items that were successfully re-queued. Items that
-    /// exceeded their max retry count are excluded from the result.
+    /// Returns all timed-out items that were re-queued.
     pub(crate) fn check_and_retry_timeouts(&mut self) -> Vec<K> {
         let timed_out = self.check_timeouts();
-        timed_out.into_iter().filter(|item| self.enqueue_retry(item.clone())).collect()
+        for item in &timed_out {
+            self.enqueue_retry(item.clone());
+        }
+        timed_out
     }
 
     /// Check if the coordinator has no work (empty pending and in-flight).
@@ -252,16 +240,6 @@ mod tests {
         assert_eq!(items, vec![99, 1, 2]);
     }
 
-    #[test]
-    fn test_max_retries() {
-        let mut coord: DownloadCoordinator<u32> =
-            DownloadCoordinator::new(DownloadConfig::default().with_max_retries(2));
-
-        assert!(coord.enqueue_retry(1));
-        assert!(coord.enqueue_retry(1));
-        assert!(!coord.enqueue_retry(1)); // Exceeds max
-    }
-
     #[test]
     fn test_take_pending() {
         let mut coord: DownloadCoordinator<u32> = DownloadCoordinator::default();
@@ -363,42 +341,11 @@ mod tests {
 
     #[test]
     fn test_config_builders() {
-        let config = DownloadConfig::default()
-            .with_max_concurrent(20)
-            .with_timeout(Duration::from_secs(60))
-            .with_max_retries(5);
+        let config =
+            DownloadConfig::default().with_max_concurrent(20).with_timeout(Duration::from_secs(60));
 
         assert_eq!(config.max_concurrent, 20);
         assert_eq!(config.timeout, Duration::from_secs(60));
-        assert_eq!(config.max_retries, 5);
-    }
-
-    #[test]
-    fn test_check_and_retry_timeouts_excludes_exceeded_retries() {
-        let mut coord: DownloadCoordinator<u32> = DownloadCoordinator::new(
-            DownloadConfig::default().with_timeout(Duration::from_millis(10)).with_max_retries(1),
-        );
-
-        // Send two items and let them time out
-        coord.mark_sent(&[1, 2]);
-        std::thread::sleep(Duration::from_millis(20));
-
-        // First round: both should be re-queued successfully
-        let requeued = coord.check_and_retry_timeouts();
-        assert_eq!(requeued.len(), 2);
-        assert!(requeued.contains(&1));
-        assert!(requeued.contains(&2));
-
-        // Drain pending and send again so they can time out a second time
-        let items = coord.take_pending(2);
-        coord.mark_sent(&items);
-        std::thread::sleep(Duration::from_millis(20));
-
-        // Second round: both have exceeded max_retries (1), so neither should be returned
-        let requeued = coord.check_and_retry_timeouts();
-        assert!(requeued.is_empty());
-        // Items should not have been re-added to pending
-        assert_eq!(coord.pending_count(), 0);
     }
 
     #[test]

dash-spv/src/sync/filter_headers/pipeline.rs

@@ -24,9 +24,6 @@ const MAX_CONCURRENT_CFHEADERS_REQUESTS: usize = 10;
 /// Timeout for CFHeaders requests. Single response but allow time for network latency.
 const FILTER_HEADERS_TIMEOUT: Duration = Duration::from_secs(20);
 
-/// Maximum number of retries for CFHeaders requests.
-const FILTER_HEADERS_MAX_RETRIES: u32 = 3;
-
 /// Pipeline for downloading compact block filter headers.
 ///
 /// Uses DownloadCoordinator<BlockHash> for batch-level tracking (keyed by stop_hash),
@@ -58,8 +55,7 @@ impl FilterHeadersPipeline {
             coordinator: DownloadCoordinator::new(
                 DownloadConfig::default()
                     .with_max_concurrent(MAX_CONCURRENT_CFHEADERS_REQUESTS)
-                    .with_timeout(FILTER_HEADERS_TIMEOUT)
-                    .with_max_retries(FILTER_HEADERS_MAX_RETRIES),
+                    .with_timeout(FILTER_HEADERS_TIMEOUT),
             ),
             batch_starts: HashMap::new(),
             buffered: HashMap::new(),
@@ -265,22 +261,10 @@ impl FilterHeadersPipeline {
     }
 
     /// Re-enqueue timed out requests for retry.
-    ///
-    /// Returns heights that exceeded max retries and were permanently dropped.
-    pub(super) fn handle_timeouts(&mut self) -> Vec<u32> {
-        let mut failed = Vec::new();
+    pub(super) fn handle_timeouts(&mut self) {
         for stop_hash in self.coordinator.check_timeouts() {
-            if !self.coordinator.enqueue_retry(stop_hash) {
-                if let Some(start_height) = self.batch_starts.remove(&stop_hash) {
-                    tracing::warn!(
-                        "CFHeaders batch at height {} exceeded max retries, dropping",
-                        start_height
-                    );
-                    failed.push(start_height);
-                }
-            }
+            self.coordinator.enqueue_retry(stop_hash);
         }
-        failed
     }
 }
 
@@ -402,9 +386,7 @@ mod tests {
 
         let mut pipeline = FilterHeadersPipeline {
             coordinator: DownloadCoordinator::new(
-                DownloadConfig::default()
-                    .with_timeout(Duration::from_millis(1))
-                    .with_max_retries(3),
+                DownloadConfig::default().with_timeout(Duration::from_millis(1)),
             ),
             batch_starts: HashMap::new(),
             buffered: HashMap::new(),
@@ -418,46 +400,10 @@ mod tests {
 
         std::thread::sleep(Duration::from_millis(5));
 
-        let failed = pipeline.handle_timeouts();
-        assert!(failed.is_empty()); // First retry succeeds
+        pipeline.handle_timeouts();
         assert_eq!(pipeline.coordinator.pending_count(), 1);
     }
 
-    #[test]
-    fn test_handle_timeouts_max_retries_exceeded() {
-        use std::time::Duration;
-
-        let mut pipeline = FilterHeadersPipeline {
-            coordinator: DownloadCoordinator::new(
-                DownloadConfig::default()
-                    .with_timeout(Duration::from_millis(1))
-                    .with_max_retries(1),
-            ),
-            batch_starts: HashMap::new(),
-            buffered: HashMap::new(),
-            next_expected: 100,
-            target_height: 2000,
-        };
-
-        let stop_hash = BlockHash::from_byte_array([0x01; 32]);
-        pipeline.batch_starts.insert(stop_hash, 100);
-
-        // First timeout + retry
-        pipeline.coordinator.mark_sent(&[stop_hash]);
-        std::thread::sleep(Duration::from_millis(5));
-        let failed = pipeline.handle_timeouts();
-        assert!(failed.is_empty());
-
-        // Re-send retry
-        let items = pipeline.coordinator.take_pending(1);
-        pipeline.coordinator.mark_sent(&items);
-
-        // Second timeout exceeds max
-        std::thread::sleep(Duration::from_millis(5));
-        let failed = pipeline.handle_timeouts();
-        assert_eq!(failed, vec![100]);
-    }
-
     #[test]
     fn test_send_pending_errors_on_missing_batch_starts() {
         let mut pipeline = FilterHeadersPipeline::new();
@@ -482,9 +428,7 @@ mod tests {
 
         let mut pipeline = FilterHeadersPipeline {
             coordinator: DownloadCoordinator::new(
-                DownloadConfig::default()
-                    .with_timeout(Duration::from_millis(1))
-                    .with_max_retries(0),
+                DownloadConfig::default().with_timeout(Duration::from_millis(1)),
             ),
             batch_starts: HashMap::new(),
             buffered: HashMap::new(),
@@ -501,10 +445,10 @@ mod tests {
 
         std::thread::sleep(Duration::from_millis(5));
 
-        let mut failed = pipeline.handle_timeouts();
-        failed.sort();
-        assert_eq!(failed.len(), 2);
-        assert!(failed.contains(&1));
-        assert!(failed.contains(&2001));
+        pipeline.handle_timeouts();
+        // Both batches re-queued
+        assert_eq!(pipeline.coordinator.pending_count(), 2);
+        assert!(pipeline.batch_starts.contains_key(&hash1));
+        assert!(pipeline.batch_starts.contains_key(&hash2));
     }
 }