diff --git a/dash-spv/src/sync/block_headers/manager.rs b/dash-spv/src/sync/block_headers/manager.rs index 6cefe07f2..f9b898bab 100644 --- a/dash-spv/src/sync/block_headers/manager.rs +++ b/dash-spv/src/sync/block_headers/manager.rs @@ -263,6 +263,43 @@ impl BlockHeadersManager { }]) } + /// Fire one fallback `GetHeaders` for block announcements that have gone + /// unanswered past the wait timeout, then drop them. + /// + /// Announced-but-unobtainable hashes (a peer that vanished, an orphan that + /// never reconnected) otherwise pin the manager: `finalize_sync_if_complete` + /// refuses to finish while any announcement is outstanding, so it resets the + /// tip segment and re-requests every pass. That recovery loop only aged the + /// stale entries out on the `Synced` branch of `tick`, which a `Syncing` + /// manager never reaches — so during initial sync the loop ran forever, + /// `BlockHeaderSyncComplete` never fired, and every downstream manager + /// stalled behind it. Sweeping here, from every state, bounds the loop. (#960) + pub(super) fn prune_stale_announcements(&mut self, requests: &RequestSender) -> SyncResult<()> { + let now = Instant::now(); + let stale: Vec = self + .pending_announcements + .iter() + .filter(|(_, announced_at)| { + now.duration_since(**announced_at) + > super::sync_manager::UNSOLICITED_HEADERS_WAIT_TIMEOUT + }) + .map(|(hash, _)| *hash) + .collect(); + + if stale.is_empty() { + return Ok(()); + } + + tracing::info!("Sending fallback GetHeaders for {} stale announcements", stale.len()); + self.pipeline.reset_tip_segment(); + self.pipeline.send_pending(requests)?; + + for hash in stale { + self.pending_announcements.remove(&hash); + } + Ok(()) + } + /// Handle inventory announcements for new blocks. /// /// During initial sync, Dash Core sends inv (not header announcements) because @@ -419,6 +456,65 @@ mod tests { ); } + /// A block announced during initial sync that no peer will ever answer must + /// not pin the manager in `Syncing`. `finalize_sync_if_complete` refuses to + /// finish while any announcement is outstanding and re-requests it every + /// pass; the aging-out sweep that breaks that loop used to run only on the + /// `Synced` branch of `tick`, which a `Syncing` manager never reaches, so + /// the loop ran forever and `BlockHeaderSyncComplete` never fired. (#960) + #[tokio::test] + async fn test_syncing_prunes_unobtainable_announcement_and_finalizes() { + let mut manager = create_test_manager().await; + let tip = manager.tip().await.unwrap(); + manager.pipeline.init(tip.height(), *tip.hash(), tip.height()); + manager.progress.set_state(SyncState::Syncing); + + let (requests, mut rx) = create_test_request_sender(); + + // Drive the single tip segment complete: nothing left to download. + manager.pipeline.send_pending(&requests).unwrap(); + while rx.try_recv().is_ok() {} + manager.pipeline.receive_headers(&[]).unwrap(); + assert!(manager.pipeline.is_complete()); + + // A phantom announcement, already aged past the wait timeout. + let phantom = BlockHash::dummy(123); + manager.pending_announcements.insert( + phantom, + std::time::Instant::now() + .checked_sub(std::time::Duration::from_secs(30)) + .expect("monotonic clock has been running longer than 30s"), + ); + + // First tick: the sweep fires its fallback (which re-opens the tip + // segment) and drops the phantom so it can never block completion again. + let events = manager.tick(&requests).await.unwrap(); + assert!( + !manager.pending_announcements.contains_key(&phantom), + "the stale announcement must be pruned" + ); + assert!(!events.iter().any(|e| matches!(e, SyncEvent::BlockHeaderSyncComplete { .. }))); + assert_eq!(manager.state(), SyncState::Syncing); + + // The fallback GetHeaders is answered by the peer's tip: an empty + // response re-completes the tip segment. + while rx.try_recv().is_ok() {} + manager.pipeline.receive_headers(&[]).unwrap(); + assert!(manager.pipeline.is_complete()); + + // Second tick: nothing stale remains, so finalize runs and the sync ends. + let events = manager.tick(&requests).await.unwrap(); + assert_eq!( + manager.state(), + SyncState::Synced, + "with the announcement pruned, the tick must finish the sync" + ); + assert!( + events.iter().any(|e| matches!(e, SyncEvent::BlockHeaderSyncComplete { .. })), + "completion must be announced so downstream managers start" + ); + } + #[tokio::test] async fn test_block_headers_manager_new() { let manager = create_test_manager().await; diff --git a/dash-spv/src/sync/block_headers/pipeline.rs b/dash-spv/src/sync/block_headers/pipeline.rs index 286816ddf..18300b023 100644 --- a/dash-spv/src/sync/block_headers/pipeline.rs +++ b/dash-spv/src/sync/block_headers/pipeline.rs @@ -176,10 +176,19 @@ impl HeadersPipeline { continue; } // If tip segment was completed but receives new headers (post-sync), - // reset it so take_ready_to_store() can process the new headers + // reset it so take_ready_to_store() can process the new headers. + // + // Only ever move `next_to_store` back toward the tip, never + // forward: `send_pending` requests solely the window + // `[next_to_store, next_to_store + ACTIVE_SEGMENT_WINDOW)`, so + // jumping it forward past a still-downloading lower segment + // drops that segment out of the window for good and header sync + // hangs. When every lower segment is already stored + // `next_to_store` sits at (or past) this tip, so the min is a + // no-op there and only guards the out-of-order case. (#950) if segment.complete && segment.target_height.is_none() { segment.complete = false; - self.next_to_store = idx; + self.next_to_store = self.next_to_store.min(idx); // A headers announcement may contain multiple consecutive headers. The // coordinator tracks the response by its first previous hash, just like a // requested headers batch. @@ -320,8 +329,13 @@ impl HeadersPipeline { for (idx, segment) in self.segments.iter_mut().enumerate() { if segment.target_height.is_none() && segment.complete { segment.complete = false; - // Reset next_to_store so buffered headers can be processed - self.next_to_store = idx; + // Reset next_to_store so buffered headers can be processed, but + // only ever backward toward the tip. Moving it forward past a + // lower segment that has not finished downloading would strand + // that segment outside `send_pending`'s active window and hang + // header sync; when the lower segments are all stored, + // `next_to_store` already sits at or past this tip. (#950) + self.next_to_store = self.next_to_store.min(idx); tracing::debug!( "Reset tip segment {} at height {} for continued syncing", segment.segment_id, @@ -697,6 +711,80 @@ mod tests { assert!(pipeline.segments[2].can_send()); } + /// `reset_tip_segment` must not jump `next_to_store` forward past a lower + /// segment that is still downloading. `send_pending` only requests the + /// window `[next_to_store, next_to_store + ACTIVE_SEGMENT_WINDOW)`, so a + /// forward jump drops that segment out of the window for good and header + /// sync hangs. (#950) + #[test] + fn test_reset_tip_segment_does_not_skip_incomplete_lower_segment() { + let cm = create_test_checkpoint_manager(true); + let mut pipeline = HeadersPipeline::new(cm); + + // Segment 0: a checkpoint segment still downloading. + let mid = + SegmentState::new(0, 0, BlockHash::dummy(0), Some(100), Some(BlockHash::dummy(1))); + // Segment 1: the open-ended tip, completed early by an empty response. + let mut tip = SegmentState::new(1, 100, BlockHash::dummy(1), None, None); + tip.complete = true; + + pipeline.set_segments_for_test(vec![mid, tip]); + assert_eq!(pipeline.next_to_store, 0); + + assert!(pipeline.reset_tip_segment(), "the complete tip segment must reset"); + assert_eq!( + pipeline.next_to_store, 0, + "reset must not skip the still-downloading lower segment" + ); + + // The lower segment is still inside the window and gets requested. + let (sender, mut rx) = create_test_request_sender(); + assert!(pipeline.send_pending(&sender).unwrap() >= 1); + let mut requested = Vec::new(); + while let Ok(request) = rx.try_recv() { + if let NetworkRequest::SendMessage( + dashcore::network::message::NetworkMessage::GetHeaders(get), + ) = request + { + requested.push(get.locator_hashes[0]); + } + } + assert!( + requested.contains(&BlockHash::dummy(0)), + "the segment the reset skipped over must still be requested" + ); + } + + /// The same guard on the receive-path tip reset: an unsolicited post-sync + /// header for the completed tip must not advance `next_to_store` past a + /// lower segment that has not finished downloading. (#950) + #[test] + fn test_receive_tip_reset_does_not_skip_incomplete_lower_segment() { + let cm = create_test_checkpoint_manager(true); + let mut pipeline = HeadersPipeline::new(cm); + + let tip_hash = BlockHash::dummy(50); + let mid = + SegmentState::new(0, 0, BlockHash::dummy(0), Some(100), Some(BlockHash::dummy(1))); + let mut tip = SegmentState::new(1, 1000, tip_hash, None, None); + tip.complete = true; + + pipeline.set_segments_for_test(vec![mid, tip]); + assert_eq!(pipeline.next_to_store, 0); + + // A post-sync header extending the tip routes to the tip segment. + let mut header = Header::dummy(1); + header.prev_blockhash = tip_hash; + let matched = pipeline.receive_headers(&[header.into()]).unwrap(); + + assert_eq!(matched, Some(1), "the header must route to the tip segment"); + assert_eq!( + pipeline.next_to_store, 0, + "the tip reset must not skip the still-downloading lower segment" + ); + assert!(!pipeline.segments[1].complete, "the tip segment must be reset to non-complete"); + } + #[test] fn test_tip_complete_lifecycle() { let cm = create_test_checkpoint_manager(true); diff --git a/dash-spv/src/sync/block_headers/sync_manager.rs b/dash-spv/src/sync/block_headers/sync_manager.rs index 3acb85c8b..1a2c276ee 100644 --- a/dash-spv/src/sync/block_headers/sync_manager.rs +++ b/dash-spv/src/sync/block_headers/sync_manager.rs @@ -8,8 +8,7 @@ use crate::sync::{ }; use async_trait::async_trait; use dashcore::network::message::NetworkMessage; -use dashcore::BlockHash; -use std::time::{Duration, Instant}; +use std::time::Duration; /// Timeout waiting for unsolicited header messages after a block announcement. pub(super) const UNSOLICITED_HEADERS_WAIT_TIMEOUT: Duration = Duration::from_secs(3); @@ -152,6 +151,12 @@ impl SyncManager for BlockHeadersMana self.pipeline.handle_timeouts(); + // Age out announcements no peer will answer, in every state. This must + // run before the `Syncing` branch's `finalize_sync_if_complete`, which + // otherwise loops forever resetting the tip segment for a permanently + // unobtainable hash and never reaches `Synced`. (#960) + self.prune_stale_announcements(requests)?; + // During initial sync, send more requests and log progress if self.state() == SyncState::Syncing { // Take, refill, then store — the same order `handle_headers_pipeline` @@ -177,34 +182,6 @@ impl SyncManager for BlockHeadersMana return Ok(events); } - // Post-sync: check for stale block announcements - if self.state() == SyncState::Synced { - let now = Instant::now(); - let stale: Vec = self - .pending_announcements - .iter() - .filter(|(_, announced_at)| { - now.duration_since(**announced_at) > UNSOLICITED_HEADERS_WAIT_TIMEOUT - }) - .map(|(hash, _)| *hash) - .collect(); - - if !stale.is_empty() { - tracing::info!( - "Sending fallback GetHeaders for {} stale announcements", - stale.len() - ); - - // Reset tip segment and send requests via pipeline - self.pipeline.reset_tip_segment(); - self.pipeline.send_pending(requests)?; - - for hash in stale { - self.pending_announcements.remove(&hash); - } - } - } - Ok(vec![]) } diff --git a/dash-spv/src/sync/download_coordinator.rs b/dash-spv/src/sync/download_coordinator.rs index 6acb0f3ca..ec87d03ac 100644 --- a/dash-spv/src/sync/download_coordinator.rs +++ b/dash-spv/src/sync/download_coordinator.rs @@ -152,6 +152,24 @@ impl DownloadCoordinator { } } + /// Return items `take_pending` handed out that were never dispatched. + /// + /// `take_pending` removes items from the queue before the caller sends + /// them, and only `mark_sent` moves them into in-flight — so a caller that + /// bails out between the two leaves those items in neither tracker, and + /// nothing puts them back: `check_timeouts` and `requeue_in_flight` both + /// walk in-flight only. This restores them to the front of the queue in + /// their original order, so the next send reissues them lowest-first and a + /// sequential consumer keeps making progress. + /// + /// Retry counts are deliberately left alone: the request never reached a + /// peer, so no peer failed to answer it. + pub(crate) fn return_unsent(&mut self, items: Vec) { + for item in items.into_iter().rev() { + self.pending.push_front(item); + } + } + /// Handle a received item. /// /// Returns true if the item was being tracked, false if unexpected. @@ -362,6 +380,35 @@ mod tests { assert_eq!(requeued, vec![1, 2, 3]); } + /// `return_unsent` must put items `take_pending` handed out but that were + /// never dispatched back at the *front* of the queue, in their original + /// order, without charging them a retry. Nothing else covers the gap + /// between `take_pending` and `mark_sent` — `check_timeouts` and + /// `requeue_in_flight` both walk in-flight only — so an item dropped there + /// is never requested again. (#972) + #[test] + fn test_return_unsent_restores_order_without_charging_a_retry() { + let mut coord: DownloadCoordinator = DownloadCoordinator::default(); + coord.enqueue([1, 2, 3, 4]); + + let taken = coord.take_pending(3); + assert_eq!(taken, vec![1, 2, 3]); + assert_eq!(coord.pending_count(), 1); + + // The caller dispatched 1, then failed on 2 and gave back the rest. + coord.mark_sent(&[1]); + coord.return_unsent(taken[1..].to_vec()); + + assert!(coord.is_in_flight(&1), "the dispatched item stays in flight"); + assert_eq!(coord.pending_count(), 3); + assert_eq!( + coord.take_pending(3), + vec![2, 3, 4], + "returned items go back ahead of what was never taken, in order" + ); + assert!(coord.retry_counts.is_empty(), "a request that never left is not a retry"); + } + #[test] fn test_requeue_in_flight_preserves_retry_counts() { let mut coord: DownloadCoordinator = DownloadCoordinator::default(); diff --git a/dash-spv/src/sync/filter_headers/manager.rs b/dash-spv/src/sync/filter_headers/manager.rs index d2f1e5c90..c9d16623f 100644 --- a/dash-spv/src/sync/filter_headers/manager.rs +++ b/dash-spv/src/sync/filter_headers/manager.rs @@ -205,6 +205,31 @@ impl FilterHeadersManager &mut self, tip_height: u32, requests: &RequestSender, + ) -> SyncResult> { + // `block_header_tip_height` is the watermark the tick's storage-tip + // check (`tip > block_header_tip_height`) uses to decide whether to + // re-arm this path. The fallible pipeline work below must not be marked + // done before it is queued: if the watermark advanced first and the + // work then failed (a storage error out of `init`/`extend_target`, or a + // `send_pending`), the next tick would see `tip == block_header_tip`, + // skip the retry, and filter-header sync would wedge for good. So + // advance the watermark, but restore it on failure so the tick retries. + // (#960) + let prev_block_header_tip = self.progress.block_header_tip_height(); + let result = self.arm_pipeline_for_new_headers(tip_height, requests).await; + if result.is_err() { + self.progress.update_block_header_tip_height(prev_block_header_tip); + } + result + } + + /// The fallible body of [`Self::handle_new_headers`]: advance the watermark + /// and target, then init or extend the pipeline and send. Kept separate so + /// the caller can restore the watermark if any step here fails. + async fn arm_pipeline_for_new_headers( + &mut self, + tip_height: u32, + requests: &RequestSender, ) -> SyncResult> { self.progress.update_block_header_tip_height(tip_height); self.update_target_height(tip_height); @@ -484,6 +509,73 @@ mod tests { } } + /// A storage failure in the pipeline work must leave the block-header + /// watermark where it was, so the tick's `tip > block_header_tip_height` + /// check re-arms this path and retries. Advancing the watermark first (as + /// the code used to) marked the range done before it was queued: the next + /// tick saw no advance, never retried, and filter-header sync wedged. (#960) + #[tokio::test] + async fn test_handle_new_headers_keeps_watermark_when_pipeline_work_fails() { + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + let header_storage = storage.block_headers(); + let mut manager = + FilterHeadersManager::new(header_storage.clone(), storage.filter_headers()) + .await + .expect("Failed to create FilterHeadersManager"); + let (sender, mut rx) = create_test_request_sender(); + + // Mid-sync, last told the tip was 1000. Block-header storage is empty, + // so the pipeline init below cannot resolve any batch's stop hash. + manager.progress.update_current_height(0); + manager.progress.update_block_header_tip_height(1000); + manager.set_state(SyncState::Syncing); + + let err = manager.handle_new_headers(3000, &sender).await; + assert!(err.is_err(), "init over empty header storage must fail"); + assert_eq!( + manager.progress.block_header_tip_height(), + 1000, + "the watermark must be restored so the next tick retries" + ); + assert!(manager.pipeline.is_complete(), "a failed init must not leave a half-built queue"); + assert!(rx.try_recv().is_err(), "nothing should have been requested"); + + // The headers arrive; the retry now succeeds and advances the + // watermark, and a request for the range goes out. + let mut headers = Vec::new(); + let mut prev = BlockHash::from_byte_array([0u8; 32]); + for nonce in 0..3100u32 { + let header = HashedBlockHeader::from(BlockHeader { + version: Version::from_consensus(1), + prev_blockhash: prev, + merkle_root: dashcore::TxMerkleNode::all_zeros(), + time: 0, + bits: CompactTarget::from_consensus(0x2100ffff), + nonce, + }); + prev = *header.hash(); + headers.push(header); + } + header_storage.write().await.store_headers_at_height(&headers, 0).await.unwrap(); + + manager + .handle_new_headers(3000, &sender) + .await + .expect("retry must succeed once headers land"); + assert_eq!( + manager.progress.block_header_tip_height(), + 3000, + "a successful arm advances the watermark" + ); + assert!( + matches!( + rx.try_recv(), + Ok(NetworkRequest::SendMessage(NetworkMessage::GetCFHeaders(_))) + ), + "the retry must send CFHeaders requests for the range" + ); + } + #[tokio::test] async fn test_on_disconnect() { let mut manager = create_test_manager().await; diff --git a/dash-spv/src/sync/filter_headers/pipeline.rs b/dash-spv/src/sync/filter_headers/pipeline.rs index 97911cbad..10e637e56 100644 --- a/dash-spv/src/sync/filter_headers/pipeline.rs +++ b/dash-spv/src/sync/filter_headers/pipeline.rs @@ -64,6 +64,35 @@ impl FilterHeadersPipeline { } } + /// Resolve the stop hash and start height of every batch covering + /// `[start_height, target_height]`, without touching pipeline state. + /// + /// The stop-hash lookups are the only fallible step in `init` and + /// `extend_target`, and resolving them all up front lets those callers + /// commit their state changes in one infallible pass — so a missing header + /// leaves the pipeline exactly as it was and the caller can retry cleanly + /// instead of advancing `target_height` past batches that never got queued. + async fn resolve_batches( + storage: &impl BlockHeaderStorage, + start_height: u32, + target_height: u32, + ) -> SyncResult> { + let mut batches = Vec::new(); + let mut current = start_height; + while current <= target_height { + let batch_end = (current + FILTER_HEADERS_BATCH_SIZE - 1).min(target_height); + + let stop_hash = + storage.get_header(batch_end).await?.map(|h| *h.hash()).ok_or_else(|| { + SyncError::Storage(format!("Missing header at height {}", batch_end)) + })?; + + batches.push((stop_hash, current)); + current = batch_end + 1; + } + Ok(batches) + } + /// Extend the pipeline to a new target height. /// /// Queues additional batches from the current target to the new target. @@ -77,27 +106,18 @@ impl FilterHeadersPipeline { return Ok(()); } - self.target_height = new_target; - - // Queue batches from (old_target + 1) to new_target - let mut current = old_target + 1; - let mut added = 0; - - while current <= new_target { - let batch_end = (current + FILTER_HEADERS_BATCH_SIZE - 1).min(new_target); - - // Get stop hash for this batch - let stop_hash = - storage.get_header(batch_end).await?.map(|h| *h.hash()).ok_or_else(|| { - SyncError::Storage(format!("Missing header at height {}", batch_end)) - })?; + // Resolve every batch before mutating state: `target_height` must not + // advance past batches a failed lookup left unqueued, or a later + // `extend_target(new_target)` returns early (`new_target <= old_target`) + // and the gap is never filled. + let batches = Self::resolve_batches(storage, old_target + 1, new_target).await?; + let added = batches.len(); + for (stop_hash, start) in batches { self.coordinator.enqueue([stop_hash]); - self.batch_starts.insert(stop_hash, current); - added += 1; - - current = batch_end + 1; + self.batch_starts.insert(stop_hash, start); } + self.target_height = new_target; if added > 0 { tracing::info!( @@ -130,27 +150,20 @@ impl FilterHeadersPipeline { start_height: u32, target_height: u32, ) -> SyncResult<()> { + // Resolve every stop hash before clearing and rebuilding state, so a + // missing header leaves the pipeline untouched and the caller retries + // cleanly rather than resuming from a half-built queue. + let batches = Self::resolve_batches(storage, start_height, target_height).await?; + self.coordinator.clear(); self.batch_starts.clear(); self.buffered.clear(); self.next_expected = start_height; self.target_height = target_height; - // Build request queue - let mut current = start_height; - while current <= target_height { - let batch_end = (current + FILTER_HEADERS_BATCH_SIZE - 1).min(target_height); - - // Get stop hash for this batch - let stop_hash = - storage.get_header(batch_end).await?.map(|h| *h.hash()).ok_or_else(|| { - SyncError::Storage(format!("Missing header at height {}", batch_end)) - })?; - + for (stop_hash, start) in batches { self.coordinator.enqueue([stop_hash]); - self.batch_starts.insert(stop_hash, current); - - current = batch_end + 1; + self.batch_starts.insert(stop_hash, start); } tracing::info!( @@ -164,6 +177,18 @@ impl FilterHeadersPipeline { } /// Send pending requests using a RequestSender (synchronous). + /// + /// `take_pending` pulls the whole slice off the queue up front, but only a + /// dispatched request reaches `mark_sent`. Every early exit below therefore + /// hands the batch it failed on — and every batch still queued behind it — + /// back to the coordinator first. Otherwise those batches end up in neither + /// tracker and nothing ever asks for them again: `handle_timeouts` and + /// `requeue_in_flight` both walk in-flight only, and `extend_target` only + /// appends above `target_height`. `next_expected` then stays pinned to the + /// lowest lost batch, the batches above it pile up in `buffered` so + /// `is_complete` never turns true, and because `handle_new_headers` re-inits + /// the pipeline only when it is complete, filter-header sync wedges for + /// good — the same failure mode `requeue_failed` exists to prevent. (#972) pub(super) fn send_pending(&mut self, requests: &RequestSender) -> SyncResult { let count = self.coordinator.available_to_send(); if count == 0 { @@ -173,15 +198,19 @@ impl FilterHeadersPipeline { let stop_hashes = self.coordinator.take_pending(count); let mut sent = 0; - for stop_hash in stop_hashes { - let Some(&start_height) = self.batch_starts.get(&stop_hash) else { + for (idx, &stop_hash) in stop_hashes.iter().enumerate() { + let Some(start_height) = self.batch_starts.get(&stop_hash).copied() else { + self.coordinator.return_unsent(stop_hashes[idx..].to_vec()); return Err(SyncError::InvalidState(format!( "No batch_starts entry for pending stop_hash {}", stop_hash ))); }; - requests.request_filter_headers(start_height, stop_hash)?; + if let Err(e) = requests.request_filter_headers(start_height, stop_hash) { + self.coordinator.return_unsent(stop_hashes[idx..].to_vec()); + return Err(e.into()); + } self.coordinator.mark_sent(&[stop_hash]); @@ -240,6 +269,22 @@ impl FilterHeadersPipeline { } } + /// Re-track a batch whose processing failed so a later `send_pending` + /// reissues it. + /// + /// `receive` clears a batch from the coordinator and `batch_starts` the + /// moment the bytes arrive, before the caller runs the fallible + /// `process_cfheaders`. If that storage write fails the batch would + /// otherwise vanish from every tracker while `next_expected` stays pinned + /// to it, and `extend_target` only ever appends above `target_height` — so + /// the hole is never revisited and filter-header sync stalls. Putting the + /// batch back on the retry queue with its start height restored makes the + /// next tick request it again. + pub(super) fn requeue_failed(&mut self, start_height: u32, stop_hash: BlockHash) { + self.batch_starts.insert(stop_hash, start_height); + self.coordinator.enqueue_retry(stop_hash); + } + /// Advance to the next expected height after processing. /// /// Returns any buffered responses that are now ready. @@ -430,6 +475,94 @@ mod tests { assert!(matches!(err, SyncError::InvalidState(_))); } + /// Collect the `GetCFHeaders` requests a test sender received, in order. + fn drain_getcfheaders( + rx: &mut tokio::sync::mpsc::UnboundedReceiver, + ) -> Vec<(u32, BlockHash)> { + let mut requested = Vec::new(); + while let Ok(request) = rx.try_recv() { + match request { + NetworkRequest::SendMessage(NetworkMessage::GetCFHeaders(GetCFHeaders { + start_height, + stop_hash, + .. + })) => requested.push((start_height, stop_hash)), + other => panic!("Expected GetCFHeaders, got {:?}", other), + } + } + requested + } + + /// Queue three consecutive batches with their start heights. + fn pipeline_with_three_queued_batches() -> (FilterHeadersPipeline, [BlockHash; 3]) { + let mut pipeline = FilterHeadersPipeline::new(); + pipeline.next_expected = 1; + pipeline.target_height = 6000; + + let hashes = [ + BlockHash::from_byte_array([0x01; 32]), + BlockHash::from_byte_array([0x02; 32]), + BlockHash::from_byte_array([0x03; 32]), + ]; + for (i, hash) in hashes.iter().enumerate() { + pipeline.coordinator.enqueue([*hash]); + pipeline.batch_starts.insert(*hash, 1 + 2000 * i as u32); + } + (pipeline, hashes) + } + + /// A failed dispatch must not strand the batches `take_pending` already + /// pulled off the queue. They were removed from pending and never reached + /// `mark_sent`, so without the restore they are tracked nowhere: + /// `handle_timeouts` and `requeue_in_flight` walk in-flight only, and + /// `extend_target` only appends above `target_height`. `next_expected` + /// would stay pinned to the lowest of them forever. (#972) + #[test] + fn test_send_pending_returns_every_batch_when_dispatch_fails() { + let (mut pipeline, hashes) = pipeline_with_three_queued_batches(); + + // A dropped receiver fails every dispatch. + let (tx, rx) = unbounded_channel(); + drop(rx); + assert!(pipeline.send_pending(&RequestSender::new(tx)).is_err()); + + assert_eq!(pipeline.coordinator.active_count(), 0, "nothing was dispatched"); + assert_eq!(pipeline.coordinator.pending_count(), 3, "every batch is still queued"); + + // A working sender reissues all three, lowest first, with their + // original start heights. + let (tx, mut rx) = unbounded_channel(); + assert_eq!(pipeline.send_pending(&RequestSender::new(tx)).unwrap(), 3); + assert_eq!( + drain_getcfheaders(&mut rx), + vec![(1, hashes[0]), (2001, hashes[1]), (4001, hashes[2])] + ); + } + + /// The same guard on the `InvalidState` exit, mid-run: batch 1 was + /// dispatched and stays in flight, while the batch that tripped the error + /// and the one still queued behind it go back to the front of the queue + /// instead of disappearing. (#972) + #[test] + fn test_send_pending_returns_remaining_batches_on_state_error() { + let (mut pipeline, hashes) = pipeline_with_three_queued_batches(); + // Batch 2 loses its start height, so it trips the InvalidState exit. + pipeline.batch_starts.remove(&hashes[1]); + + let (tx, mut rx) = unbounded_channel(); + let err = pipeline.send_pending(&RequestSender::new(tx)).unwrap_err(); + assert!(matches!(err, SyncError::InvalidState(_))); + + // Only batch 1 went out, and it is accounted for as in flight. + assert_eq!(drain_getcfheaders(&mut rx), vec![(1, hashes[0])]); + assert!(pipeline.coordinator.is_in_flight(&hashes[0])); + assert_eq!(pipeline.coordinator.active_count(), 1); + + // The failed batch and the one behind it are queued, in order. + assert_eq!(pipeline.coordinator.pending_count(), 2); + assert_eq!(pipeline.coordinator.take_pending(2), vec![hashes[1], hashes[2]]); + } + /// A peer disconnect requeues in-flight batches without discarding what the /// pipeline has already made of the ones that came back, so the reissued /// requests carry their original start heights and nothing is re-downloaded. @@ -484,6 +617,132 @@ mod tests { assert_eq!(pipeline.target_height, 6000); } + /// A batch whose `process_cfheaders` fails after `receive` already dropped + /// it from the trackers must be retryable: `requeue_failed` puts it back so + /// the next `send_pending` reissues it, and nothing is stranded with + /// `next_expected` pinned to a batch nothing tracks. (#964) + #[test] + fn test_requeue_failed_makes_a_dropped_batch_retryable() { + let mut pipeline = FilterHeadersPipeline::new(); + pipeline.next_expected = 1; + pipeline.target_height = 2000; + + let stop_hash = BlockHash::from_byte_array([0x07; 32]); + pipeline.coordinator.mark_sent(&[stop_hash]); + pipeline.batch_starts.insert(stop_hash, 1); + + let cfheaders = CFHeaders { + filter_type: 0, + stop_hash, + previous_filter_header: FilterHeader::all_zeros(), + filter_hashes: vec![FilterHash::all_zeros()], + }; + + // In-order receive hands the batch to the caller and drops it from + // every tracker. + assert!(pipeline.receive(1, cfheaders).is_some()); + assert!(!pipeline.coordinator.is_in_flight(&stop_hash)); + assert!(!pipeline.batch_starts.contains_key(&stop_hash)); + + // Processing failed downstream: re-track the batch. + pipeline.requeue_failed(1, stop_hash); + + // The next send_pending reissues exactly that batch with its original + // start height, and progress is untouched. + let (tx, mut rx) = unbounded_channel(); + let requests = RequestSender::new(tx); + assert_eq!(pipeline.send_pending(&requests).unwrap(), 1); + + let mut reissued = Vec::new(); + while let Ok(request) = rx.try_recv() { + match request { + NetworkRequest::SendMessage(NetworkMessage::GetCFHeaders(GetCFHeaders { + start_height, + stop_hash, + .. + })) => reissued.push((start_height, stop_hash)), + other => panic!("Expected GetCFHeaders, got {:?}", other), + } + } + assert_eq!(reissued, vec![(1, stop_hash)]); + assert_eq!(pipeline.next_expected(), 1); + } + + /// A missing header mid-`extend_target` must leave `target_height` and the + /// queue untouched, or a later `extend_target` to the same target returns + /// early (`new_target <= old_target`) and the unqueued batches are lost — + /// filter-header sync stalls with the watermark parked above them. (#964) + #[tokio::test] + async fn test_extend_target_is_atomic_on_a_missing_header() { + use crate::storage::{BlockHeaderStorage, DiskStorageManager, StorageManager}; + use crate::types::HashedBlockHeader; + use dashcore::{block::Version, BlockHash, CompactTarget, Header as BlockHeader}; + + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + let header_storage = storage.block_headers(); + + // Store headers up to height 2000 only. + let mut headers = Vec::new(); + let mut prev = BlockHash::from_byte_array([0u8; 32]); + for nonce in 0..2001u32 { + let header = HashedBlockHeader::from(BlockHeader { + version: Version::from_consensus(1), + prev_blockhash: prev, + merkle_root: dashcore::TxMerkleNode::all_zeros(), + time: 0, + bits: CompactTarget::from_consensus(0x2100ffff), + nonce, + }); + prev = *header.hash(); + headers.push(header); + } + header_storage.write().await.store_headers_at_height(&headers, 0).await.unwrap(); + + let mut pipeline = FilterHeadersPipeline::new(); + { + let guard = header_storage.read().await; + pipeline.init(&*guard, 1, 2000).await.expect("init within stored range"); + } + assert_eq!(pipeline.target_height, 2000); + let pending_after_init = pipeline.coordinator.pending_count(); + + // Extending past the stored headers fails at the first missing stop hash. + { + let guard = header_storage.read().await; + assert!(pipeline.extend_target(&*guard, 5000).await.is_err()); + } + assert_eq!( + pipeline.target_height, 2000, + "a failed extend must not advance target_height past unqueued batches" + ); + assert_eq!( + pipeline.coordinator.pending_count(), + pending_after_init, + "a failed extend must not enqueue a partial set of batches" + ); + + // Once the headers exist, the same extend succeeds and advances. + let mut more = Vec::new(); + for nonce in 2001..5001u32 { + let header = HashedBlockHeader::from(BlockHeader { + version: Version::from_consensus(1), + prev_blockhash: prev, + merkle_root: dashcore::TxMerkleNode::all_zeros(), + time: 0, + bits: CompactTarget::from_consensus(0x2100ffff), + nonce, + }); + prev = *header.hash(); + more.push(header); + } + header_storage.write().await.store_headers_at_height(&more, 2001).await.unwrap(); + { + let guard = header_storage.read().await; + pipeline.extend_target(&*guard, 5000).await.expect("extend within stored range"); + } + assert_eq!(pipeline.target_height, 5000); + } + #[test] fn test_handle_timeouts_multiple_batches() { use std::time::Duration; diff --git a/dash-spv/src/sync/filter_headers/sync_manager.rs b/dash-spv/src/sync/filter_headers/sync_manager.rs index 353bf2b8a..08e907bd5 100644 --- a/dash-spv/src/sync/filter_headers/sync_manager.rs +++ b/dash-spv/src/sync/filter_headers/sync_manager.rs @@ -60,8 +60,18 @@ impl SyncManager for FilterHeade // Try to receive (may buffer if out of order) if let Some(data) = self.pipeline.receive(start_height, cfheaders) { - // In order - process immediately - let count = self.process_cfheaders(&data, start_height).await?; + // In order - process immediately. `receive` already dropped this + // batch from the pipeline's trackers, so if the fallible store + // fails, re-enqueue it before propagating or the hole at + // `next_expected` is never revisited (see `requeue_failed`). + let stop_hash = data.stop_hash; + let count = match self.process_cfheaders(&data, start_height).await { + Ok(count) => count, + Err(e) => { + self.pipeline.requeue_failed(start_height, stop_hash); + return Err(e); + } + }; if count == 0 { return Err(SyncError::Network("CFHeaders batch contained no headers".to_string())); } @@ -91,7 +101,17 @@ impl SyncManager for FilterHeade while !ready_batches.is_empty() { // Take ownership and process each batch for (height, data) in std::mem::take(&mut ready_batches) { - let count = self.process_cfheaders(&data, height).await?; + // A promoted buffered batch was likewise already dropped + // from the trackers by `receive`; re-enqueue on a failed + // store so the tick retries it. + let stop_hash = data.stop_hash; + let count = match self.process_cfheaders(&data, height).await { + Ok(count) => count, + Err(e) => { + self.pipeline.requeue_failed(height, stop_hash); + return Err(e); + } + }; if count == 0 { return Err(SyncError::Network( "CFHeaders batch contained no headers".to_string(), diff --git a/dash/src/sml/llmq_entry_verification.rs b/dash/src/sml/llmq_entry_verification.rs index bec73d58f..21c9a0fcd 100644 --- a/dash/src/sml/llmq_entry_verification.rs +++ b/dash/src/sml/llmq_entry_verification.rs @@ -138,6 +138,31 @@ impl From for LLMQEntryVerificationStatus { QuorumValidationError::CycleBaseHeightTooLow(height) => { Self::Skipped(LLMQEntryVerificationSkipStatus::MissedList(height)) } + // The quorum index is not signature-covered, so a peer can rewrite + // it, and an out-of-range index only means this engine cannot locate + // the quorum's cycle from what it holds — a fresh QRInfo from an + // honest peer resolves it. Classifying it `Skipped`, not `Invalid`, + // keeps a tampered index from aborting `feed_qr_info` and wedging + // masternode sync. This never marks the entry `Verified`, and the + // rotated-cycle stores (`rotated_quorums_per_cycle`) still retain + // only `Verified` entries. However, it does leave the entry + // reachable through `quorum_entry_for_hash_at_or_before_height` + // (masternode_list_engine/helpers.rs), which excludes only + // `Invalid` entries and is the lookup dash-spv-ffi's + // platform_integration uses to serve quorum public keys. That + // lookup already serves `Skipped` entries routinely — + // `Skipped(NotMarkedForVerification)` is the default status for + // quorums entering a stored list — so this reclassification + // neither creates that exposure nor widens it beyond entries + // already present in stored lists. Tightening that lookup to + // require `Verified` is a behavioral change deliberately left as + // a follow-up. (#934) + QuorumValidationError::InvalidQuorumIndex { + quorum_hash, + index, + } => Self::Skipped(LLMQEntryVerificationSkipStatus::OtherContext(format!( + "invalid quorum index {index} for {quorum_hash}" + ))), other => Self::Invalid(other), } } @@ -230,6 +255,24 @@ mod tests { 5, )), ), + // A wire-supplied quorum index is not signature-covered, so an + // out-of-range one degrades the single quorum (Skipped), never + // aborts the feed (Invalid). + { + let quorum_hash = crate::QuorumHash::from_byte_array([6; 32]); + ( + QuorumValidationError::InvalidQuorumIndex { + quorum_hash, + index: 42, + }, + LLMQEntryVerificationStatus::Skipped( + LLMQEntryVerificationSkipStatus::OtherContext(format!( + "invalid quorum index {} for {}", + 42, quorum_hash + )), + ), + ) + }, ( QuorumValidationError::InvalidQuorumPublicKey, LLMQEntryVerificationStatus::Invalid(QuorumValidationError::InvalidQuorumPublicKey), diff --git a/dash/src/sml/masternode_list_engine/mod.rs b/dash/src/sml/masternode_list_engine/mod.rs index 0be111c73..1de7ab15a 100644 --- a/dash/src/sml/masternode_list_engine/mod.rs +++ b/dash/src/sml/masternode_list_engine/mod.rs @@ -883,6 +883,17 @@ impl MasternodeListEngine { diffs.sort_by_key(|diff| self.block_container.get_height(&diff.block_hash)); let mut sigs_by_work_height = BTreeMap::new(); + // A work height maps to one work block, which has one ChainLock + // signature, so every genuine entry keying it carries the same + // signature. A second, DIFFERING signature for the same work height can + // only come from a crafted diff. This map is built from unvalidated wire + // data, and last-write-wins would let that crafted entry overwrite the + // genuine signature, reconstruct the wrong member set, and fail the + // aggregate check on honest data — a non-inferred `Invalid` that aborts + // the whole feed and wedges masternode sync. Record the conflict and + // drop the height entirely below, so the quorums resting on it degrade + // to a recoverable `Skipped` instead of a fatal `Invalid`. (#934) + let mut conflicting_work_heights = BTreeSet::new(); for diff in &diffs { for sig_obj in &diff.quorums_chainlock_signatures { for &index in &sig_obj.index_set { @@ -896,17 +907,33 @@ impl MasternodeListEngine { .rotated_quorum_cycle_base(quorum) .and_then(|cycle_base| cycle_base.checked_sub(WORK_DIFF_DEPTH)) { - sigs_by_work_height.insert(work_height, sig_obj.signature); + match sigs_by_work_height.get(&work_height) { + Some(existing) if *existing != sig_obj.signature => { + conflicting_work_heights.insert(work_height); + } + Some(_) => {} + None => { + sigs_by_work_height.insert(work_height, sig_obj.signature); + } + } } } } } + for work_height in &conflicting_work_heights { + sigs_by_work_height.remove(work_height); + } let mut inferred_work_heights = BTreeSet::new(); for diff in &diffs { let Some(newest_work_height) = self.diff_newest_work_height(diff) else { continue; }; + // A poisoned height must not be resurrected by the elimination pass + // either: leaving it unkeyed degrades its quorums to `Skipped`. + if conflicting_work_heights.contains(&newest_work_height) { + continue; + } if sigs_by_work_height.contains_key(&newest_work_height) { continue; } @@ -994,7 +1021,13 @@ impl MasternodeListEngine { /// of the active set and lands the base on a DKG interval boundary. Any /// other index shifts the base into a cycle the quorum does not belong /// to, and a base derived from that keys another cycle's work block. - #[cfg(feature = "quorum_validation")] + /// + /// Not feature-gated: the always-compiled rotated member reconstruction in + /// `find_rotated_masternodes_for_quorums` derives its cycle base here too, + /// so both paths agree on which indices resolve a cycle. `allow(dead_code)` + /// covers the build where `quorum_validation` is off and that reconstruction + /// is the only caller. + #[allow(dead_code)] fn rotated_quorum_cycle_base(&self, quorum: &QuorumEntry) -> Option { let height = self.block_container.get_height(&quorum.quorum_hash)?; let quorum_index = quorum.quorum_index?; @@ -2008,6 +2041,97 @@ mod tests { ); } + /// A work height maps to one work block, which has one ChainLock signature, + /// so every genuine entry keying it carries the same signature. A second, + /// DIFFERING signature can only come from a crafted diff. Last-write-wins + /// would let it replace the genuine signature, reconstruct the wrong member + /// set, and fail the aggregate check on honest data — a non-inferred + /// `Invalid` that aborts the feed and wedges masternode sync. The height is + /// dropped on conflict so its quorums degrade to a recoverable `Skipped` + /// instead of serving a forged signature. (#934) + #[cfg(feature = "quorum_validation")] + #[test] + fn rotation_cl_sigs_by_work_height_drops_a_work_block_with_conflicting_signatures() { + let isd_type = Network::Mainnet.isd_llmq_type(); + let interval = isd_type.params().dkg_params.interval; + let cycle_base = interval * 100; + let diff_end = cycle_base + interval - WORK_DIFF_DEPTH; + let work_height = cycle_base - WORK_DIFF_DEPTH; + + let end_hash = BlockHash::from_byte_array([1; 32]); + let quorum_a = QuorumHash::from_byte_array([3; 32]); + let quorum_b = QuorumHash::from_byte_array([4; 32]); + let genuine_sig = BLSSignature::from([7; 96]); + let forged_sig = BLSSignature::from([8; 96]); + + // Two commitments of the SAME cycle (indices 3 and 5 both key + // `work_height`) carrying DIFFERENT signatures — the second is forged. + let diff = make_cl_sig_diff( + end_hash, + vec![ + make_quorum_entry(isd_type, quorum_a, Some(3)), + make_quorum_entry(isd_type, quorum_b, Some(5)), + ], + vec![(genuine_sig, vec![0]), (forged_sig, vec![1])], + ); + + let engine = engine_knowing_blocks(&[ + (diff_end, end_hash), + (cycle_base + 3, quorum_a), + (cycle_base + 5, quorum_b), + ]); + + let (sigs, inferred) = engine.rotation_cl_sigs_by_work_height([&diff]); + assert!( + !sigs.contains_key(&work_height), + "a work height with conflicting signatures must be dropped, got {sigs:?}" + ); + assert!( + !inferred.contains(&work_height), + "a dropped height must not be resurrected as an inferred key" + ); + } + + /// A repeated, consistent signature for one work block is genuine and must + /// survive: it is the ChainLock the whole cycle shares. Only a conflicting + /// signature drops the height. + #[cfg(feature = "quorum_validation")] + #[test] + fn rotation_cl_sigs_by_work_height_keeps_a_work_block_with_a_repeated_signature() { + let isd_type = Network::Mainnet.isd_llmq_type(); + let interval = isd_type.params().dkg_params.interval; + let cycle_base = interval * 100; + let diff_end = cycle_base + interval - WORK_DIFF_DEPTH; + let work_height = cycle_base - WORK_DIFF_DEPTH; + + let end_hash = BlockHash::from_byte_array([1; 32]); + let quorum_a = QuorumHash::from_byte_array([3; 32]); + let quorum_b = QuorumHash::from_byte_array([4; 32]); + let sig = BLSSignature::from([7; 96]); + + let diff = make_cl_sig_diff( + end_hash, + vec![ + make_quorum_entry(isd_type, quorum_a, Some(3)), + make_quorum_entry(isd_type, quorum_b, Some(5)), + ], + vec![(sig, vec![0]), (sig, vec![1])], + ); + + let engine = engine_knowing_blocks(&[ + (diff_end, end_hash), + (cycle_base + 3, quorum_a), + (cycle_base + 5, quorum_b), + ]); + + let (sigs, _) = engine.rotation_cl_sigs_by_work_height([&diff]); + assert_eq!( + sigs.get(&work_height), + Some(&sig), + "a consistent repeated signature must be kept" + ); + } + /// Which rotation type an active set is keyed by decides which quorum /// indices exist at all, so a wrong one rejects genuine commitments. It /// comes from configuration wherever the deployment fixes it, and only a @@ -2571,6 +2695,54 @@ mod tests { } } + /// A rewritten, non-signature-covered quorum index in an active set must + /// not abort the whole feed. The unhardened cycle-base derivation used to + /// accept an out-of-range index, reconstruct against the wrong cycle, and + /// index the member set raw with it — reaching a `CorruptedCodeExecution` + /// (or, once hardened, an `InvalidQuorumIndex`) that a non-inferred `Invalid` + /// turns into a feed abort, wedging masternode sync. The index now degrades + /// the one quorum to `Skipped` and the feed still succeeds. (#934) + #[cfg(feature = "quorum_validation")] + #[test] + fn feed_qr_info_degrades_a_tampered_quorum_index_instead_of_aborting() { + let (mut engine, mut qr_info) = load_qrinfo_2240504_fixture(); + + let target = qr_info + .last_commitment_per_index + .first() + .expect("fixture must carry rotation commitments") + .quorum_hash; + let llmq_type = qr_info.last_commitment_per_index[0].llmq_type; + let active_count = llmq_type.active_quorum_count(); + assert!(engine.block_container.get_height(&target).is_some()); + + // Rewrite index 0 to one well outside the active set (its own true + // index plus a whole cycle, the shape the audit observed). + qr_info.last_commitment_per_index[0].quorum_index = + Some(active_count as i16 + llmq_type.params().dkg_params.interval as i16); + + let feed_result = engine + .feed_qr_info(qr_info, false, true) + .expect("a tampered quorum index must not abort the feed") + .expect("expected a feed result"); + + // The tampered entry degrades to Skipped and is never stored verified; + // a cycle holding a skipped entry is not stored. + let status = engine + .quorum_statuses + .get(&llmq_type) + .and_then(|m| m.get(&target)) + .map(|(_, _, status)| status.clone()); + assert!( + matches!(status, Some(LLMQEntryVerificationStatus::Skipped(_))), + "a tampered index must settle as Skipped, got {status:?}" + ); + assert!( + feed_result.stored_cycle_height.is_none(), + "a cycle holding a skipped entry must not be stored" + ); + } + #[cfg(feature = "quorum_validation")] #[test] fn feed_qr_info_rejects_post_v20_with_missing_chainlock_signatures() { diff --git a/dash/src/sml/masternode_list_engine/rotated_quorum_construction.rs b/dash/src/sml/masternode_list_engine/rotated_quorum_construction.rs index 81ec89662..dea47581f 100644 --- a/dash/src/sml/masternode_list_engine/rotated_quorum_construction.rs +++ b/dash/src/sml/masternode_list_engine/rotated_quorum_construction.rs @@ -6,9 +6,7 @@ use crate::prelude::CoreBlockHeight; use crate::sml::llmq_type::LLMQType; use crate::sml::llmq_type::rotation::{LLMQQuarterReconstructionType, LLMQQuarterUsageType}; use crate::sml::masternode_list::MasternodeList; -use crate::sml::masternode_list_engine::{ - MasternodeListEngine, cycle_quarter_work_heights, rotated_cycle_base_height, -}; +use crate::sml::masternode_list_engine::{MasternodeListEngine, cycle_quarter_work_heights}; use crate::sml::masternode_list_entry::qualified_masternode_list_entry::QualifiedMasternodeListEntry; use crate::sml::quorum_entry::qualified_quorum_entry::{ QualifiedQuorumEntry, VerifyingChainLockSignaturesType, @@ -64,7 +62,10 @@ impl MasternodeListEngine { BTreeMap::new(); for quorum in quorums { let quorum_hash = quorum.quorum_entry.quorum_hash; - let Some(quorum_block_height) = self.block_container.get_height(&quorum_hash) else { + // Keep the missing-height case distinct (it maps to `Skipped` and + // tells the caller to refetch that block); the hardened resolver + // below re-derives the height itself. + if self.block_container.get_height(&quorum_hash).is_none() { members_by_quorum_hash.insert( quorum_hash, Err(QuorumValidationError::RequiredBlockNotPresent( @@ -73,7 +74,7 @@ impl MasternodeListEngine { )), ); continue; - }; + } let llmq_type = quorum.quorum_entry.llmq_type; let Some(quorum_index) = quorum.quorum_entry.quorum_index else { members_by_quorum_hash.insert( @@ -82,8 +83,16 @@ impl MasternodeListEngine { ); continue; }; - let Some(cycle_base_height) = - rotated_cycle_base_height(quorum_block_height, quorum_index) + // Derive the cycle base through the hardened resolver, which rejects + // an index outside the active set or one that lands the base off a + // DKG interval boundary. `quorum_index` is not signature-covered, so + // a peer can rewrite it; the unhardened `rotated_cycle_base_height` + // used to accept any index that merely did not underflow, then index + // `members_by_index` raw with it below — a rewritten index reached a + // `CorruptedCodeExecution` that aborted the whole `feed_qr_info` and + // wedged masternode sync. Failing this one entry with a benign + // `InvalidQuorumIndex` (classified `Skipped`) degrades it instead. (#934) + let Some(cycle_base_height) = self.rotated_quorum_cycle_base(&quorum.quorum_entry) else { members_by_quorum_hash.insert( quorum_hash, @@ -126,11 +135,19 @@ impl MasternodeListEngine { } } }; + // Bounds-check the wire-supplied index before indexing the + // reconstructed set. The hardened cycle base above already confines + // the index to the active set, but the reconstruction can yield + // fewer indexed groups than that (a short final quarter), so a raw + // `get` can still miss. Report it as a benign, per-quorum + // `InvalidQuorumIndex` (classified `Skipped`) instead of the old + // `CorruptedCodeExecution`, which was treated as proof of corruption + // and aborted the whole feed. (#934) let result = members_by_index.get(quorum_index as usize).cloned().ok_or( - QuorumValidationError::CorruptedCodeExecution(format!( - "expected masternode list entry members for {}", - quorum_index - )), + QuorumValidationError::InvalidQuorumIndex { + quorum_hash, + index: quorum_index, + }, ); members_by_quorum_hash.insert(quorum_hash, result); } @@ -631,6 +648,33 @@ mod tests { } } + /// An index outside the active set — but not underflowing the quorum height + /// — must be rejected before any reconstruction. The unhardened cycle-base + /// derivation accepted it, reconstructed against the wrong cycle, then + /// indexed the member set raw with the out-of-range value; the hardened + /// resolver rejects it as a clean per-quorum `InvalidQuorumIndex`. (#934) + #[test] + fn an_index_outside_the_active_set_is_rejected_before_reconstruction() { + let mut engine = MasternodeListEngine::default_for_network(Network::Mainnet); + let quorum_hash = QuorumHash::from_byte_array([7; 32]); + engine.feed_block_height(10_000, quorum_hash); + + // Llmqtype60_75 has 32 active quorums, so 33 names no slot, yet it does + // not underflow height 10_000 — exactly the case the old derivation let + // through. + let active_count = LLMQType::Llmqtype60_75.active_quorum_count(); + assert_eq!(active_count, 32, "test assumes the 60_75 active quorum count"); + let quorum = rotated_quorum(quorum_hash, active_count as i16 + 1); + + let error = engine + .find_rotated_masternodes_for_quorum(&quorum) + .expect_err("an index outside the active set must be rejected"); + assert!( + matches!(error, QuorumValidationError::InvalidQuorumIndex { index, .. } if index == active_count as i16 + 1), + "expected InvalidQuorumIndex, got {error}" + ); + } + /// Every quorum of a cycle resolves the member sets from its own quarter /// signatures, and one that carries none cannot stand in for the cycle. /// Caching its failure would leave every sibling unverifiable on a diff --git a/dash/src/sml/quorum_validation_error.rs b/dash/src/sml/quorum_validation_error.rs index b8f895d71..b97ddb301 100644 --- a/dash/src/sml/quorum_validation_error.rs +++ b/dash/src/sml/quorum_validation_error.rs @@ -93,9 +93,6 @@ pub enum QuorumValidationError { index: i16, }, - #[error("Cycle base height {0} is below the history a rotated quorum reconstruction needs")] - CycleBaseHeightTooLow(CoreBlockHeight), - #[error("Corrupted code execution: {0}")] CorruptedCodeExecution(String), #[error("Expected only rotated quorums, but got quorum {0} of type {1}")] @@ -104,6 +101,15 @@ pub enum QuorumValidationError { /// Error indicating that a required feature is not turned on. #[error("Feature not turned on: {0}")] FeatureNotTurnedOn(String), + + // Keep new variants appended at the end: with `bincode` the discriminant is + // the variant's ordinal, so inserting mid-enum shifts every later variant's + // encoding and makes a persisted engine blob decode as the wrong error on + // upgrade. This one was previously inserted between `InvalidQuorumIndex` and + // `CorruptedCodeExecution`, silently re-numbering the three variants after + // it; it now sits last so those keep their original discriminants. (#934) + #[error("Cycle base height {0} is below the history a rotated quorum reconstruction needs")] + CycleBaseHeightTooLow(CoreBlockHeight), } impl From for QuorumValidationError { @@ -111,3 +117,46 @@ impl From for QuorumValidationError { QuorumValidationError::SMLError(value) } } + +#[cfg(all(test, feature = "bincode"))] +mod tests { + use super::*; + use bincode::{config, decode_from_slice, encode_to_vec}; + + /// `bincode` encodes an enum variant by its ordinal, so a persisted + /// `QuorumValidationError` blob is keyed to the variant order at write time. + /// `CycleBaseHeightTooLow` was once inserted between `InvalidQuorumIndex` and + /// `CorruptedCodeExecution`, shifting the three variants after it by one; + /// moving it to the end restores their discriminants so an engine blob + /// written before the shift still decodes as the variant it was saved as. + /// (#934) + #[test] + fn later_variants_keep_their_legacy_discriminants() { + // Legacy layout (before `CycleBaseHeightTooLow` existed) put + // `CorruptedCodeExecution` at ordinal 22, encoded as `[22, len, bytes]` + // under the standard config's varint tag. + let legacy = [22u8, 1, b'x']; + let (decoded, _): (QuorumValidationError, usize) = + decode_from_slice(&legacy, config::standard()).expect("legacy blob must decode"); + assert_eq!( + decoded, + QuorumValidationError::CorruptedCodeExecution("x".to_string()), + "ordinal 22 must still decode as CorruptedCodeExecution, not the appended variant" + ); + + // The current encoder still writes that same discriminant. + let encoded = encode_to_vec( + QuorumValidationError::CorruptedCodeExecution("x".to_string()), + config::standard(), + ) + .expect("encode"); + assert_eq!(encoded, legacy, "CorruptedCodeExecution must encode at its legacy ordinal 22"); + + // The relocated variant sits last now and still round-trips. + let relocated = QuorumValidationError::CycleBaseHeightTooLow(5); + let bytes = encode_to_vec(&relocated, config::standard()).expect("encode"); + let (back, _): (QuorumValidationError, usize) = + decode_from_slice(&bytes, config::standard()).expect("decode"); + assert_eq!(back, relocated); + } +}