From d8012b7cf70f00f58aa32cf872deb77114842327 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:42:41 +0300 Subject: [PATCH] fix(dash-spv): promote finished header segments from the tick, not only on a message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A testnet wallet restore froze with the whole chain downloaded and none of the tail of it stored: Headers: Syncing 2520289/2520288 (100.0%) processed: 1274000, buffered: 1046289 Filter Headers: Syncing 1474000/2520288 (58.5%) Blocks: WaitForEvents last_relevant: 1472978 Peers stayed connected and chain locks kept arriving for the sixteen minutes the app was left running afterwards. Nothing advanced again. `processed`/`buffered` decode as 1,474,000 headers in storage and 1,046,289 more downloaded, validated and held in memory. The top line reads 100% because `current_height()` is `tip + buffered` — it counts what was downloaded, not what was kept. `take_ready_to_store` is the only thing that promotes a finished segment into storage, and its single caller was `handle_headers_pipeline` — reached only when a `Headers` message arrives. All 47 checkpoint segments had finished downloading by 22:23:12, so no further `Headers` would ever come and the promotion had nothing left to trigger it. Filter headers, filters and blocks then coasted to a stop over the next four minutes as they consumed the backlog they had been racing ahead on, which is what made the stall visible at 22:27:20. `tick` runs every 100ms and called only `handle_timeouts` and `send_pending`. It now takes, refills and stores in the same order `handle_headers_pipeline` uses — #950 established that ordering so a segment exposed by draining gets requested in the same pass — and finalizes if that was the last of the work. `store_ready_batches` and `finalize_sync_if_complete` are lifted out of `handle_headers_pipeline`, which still calls both in place. The drain itself stays at the call sites rather than moving into the helper, precisely so the take → refill → store order is preserved on both paths. No behaviour change on the message-driven path. **Not established:** why the first promotion opportunity — the message that completed segment 25 — stored nothing. No error is logged anywhere near it and no early return in the current code fits the evidence. This lets the pipeline recover from that miss; it does not explain the miss. Worth a `RUST_LOG=dash_spv::sync::block_headers=trace` reproduction. #950's `ACTIVE_SEGMENT_WINDOW` narrows the blast radius — at most eight segments' worth of headers strand instead of a million — but adds no trigger for promotion, so the stall survives it. The regression test covers both halves: a tick promotes buffered headers with no further message, and a tick over a complete pipeline announces the sync. It fails without the change at the promotion assertion. cargo test -p dash-spv --lib # 547 passed cargo test -p dash-spv --test header_dispatch_order # passed cargo clippy --all-targets + cargo fmt --check # clean --- dash-spv/src/sync/block_headers/manager.rs | 172 +++++++++++++++--- .../src/sync/block_headers/sync_manager.rs | 17 +- 2 files changed, 161 insertions(+), 28 deletions(-) diff --git a/dash-spv/src/sync/block_headers/manager.rs b/dash-spv/src/sync/block_headers/manager.rs index 7c76828af..6cefe07f2 100644 --- a/dash-spv/src/sync/block_headers/manager.rs +++ b/dash-spv/src/sync/block_headers/manager.rs @@ -158,6 +158,45 @@ impl BlockHeadersManager { } } + events.extend(self.store_ready_batches(ready_batches).await?); + + // After storing unsolicited post-sync headers, mark the tip complete so the next header goes through + // the clean reset path. Don't mark complete during active catch-up. + if !was_syncing && tip_was_complete && !events.is_empty() { + self.pipeline.mark_tip_complete(); + } + + if was_syncing { + events.extend(self.finalize_sync_if_complete(requests).await?); + } + + if matched.is_some() { + self.progress.bump_last_activity(); + } + Ok(events) + } + + /// Write segments that finished downloading into storage. + /// + /// Split out of [`Self::handle_headers_pipeline`] so `tick` can promote + /// headers too. Promotion used to be reachable only from there — i.e. only + /// when a `Headers` message arrived — which leaves a sync no way to finish + /// itself: once the last segment completes, no further `Headers` will ever + /// come, and the already-downloaded, already-validated tail simply stays in + /// memory. A testnet restore was found frozen exactly there, with 1,046,289 + /// headers buffered and the stored tip stuck at 1,474,000 while peers + /// stayed connected and chain locks kept arriving. + /// + /// Takes the batches rather than draining them itself, so callers keep + /// `take_ready_to_store` → `send_pending` → store ordering: draining can + /// expose a new segment at the end of the active window, and the refill + /// should pick it up in the same pass. + pub(super) async fn store_ready_batches( + &mut self, + ready_batches: Vec<(u32, Vec)>, + ) -> SyncResult> { + let mut events = Vec::new(); + for (_start_height, batch_headers) in ready_batches { if !batch_headers.is_empty() { // Validate chain continuity with current tip @@ -186,37 +225,42 @@ impl BlockHeadersManager { } } - // After storing unsolicited post-sync headers, mark the tip complete so the next header goes through - // the clean reset path. Don't mark complete during active catch-up. - if !was_syncing && tip_was_complete && !events.is_empty() { - self.pipeline.mark_tip_complete(); - } + Ok(events) + } - if was_syncing && self.pipeline.is_complete() { - // If blocks were announced during sync, request them before finalizing the sync - if !self.pending_announcements.is_empty() { - tracing::info!( - "Pipeline complete but {} blocks announced during sync, requesting headers", - self.pending_announcements.len() - ); - self.pipeline.reset_tip_segment(); - self.pipeline.send_pending(requests)?; - } else { - // Synced to the tip and no pending announcements, finalize and emit event - let tip = self.tip().await?; - self.progress.update_target_height(tip.height()); - self.progress.set_state(SyncState::Synced); - tracing::info!("Headers sync complete at height {}", tip.height()); - events.push(SyncEvent::BlockHeaderSyncComplete { - tip_height: tip.height(), - }); - } + /// Close out an initial sync whose pipeline has nothing left to download. + /// + /// Extracted alongside [`Self::store_ready_batches`] and for the same + /// reason: a promotion driven by `tick` has to be able to reach the + /// completion it just made possible, or the manager stores the last + /// segment and then sits in `Syncing` forever. + pub(super) async fn finalize_sync_if_complete( + &mut self, + requests: &RequestSender, + ) -> SyncResult> { + if !self.pipeline.is_complete() { + return Ok(Vec::new()); } - if matched.is_some() { - self.progress.bump_last_activity(); + // If blocks were announced during sync, request them before finalizing the sync + if !self.pending_announcements.is_empty() { + tracing::info!( + "Pipeline complete but {} blocks announced during sync, requesting headers", + self.pending_announcements.len() + ); + self.pipeline.reset_tip_segment(); + self.pipeline.send_pending(requests)?; + return Ok(Vec::new()); } - Ok(events) + + // Synced to the tip and no pending announcements, finalize and emit event + let tip = self.tip().await?; + self.progress.update_target_height(tip.height()); + self.progress.set_state(SyncState::Synced); + tracing::info!("Headers sync complete at height {}", tip.height()); + Ok(vec![SyncEvent::BlockHeaderSyncComplete { + tip_height: tip.height(), + }]) } /// Handle inventory announcements for new blocks. @@ -301,6 +345,80 @@ mod tests { manager } + /// Headers that finished downloading must reach storage even if no further + /// `Headers` message ever arrives. + /// + /// Promotion used to hang off `handle_headers_pipeline` alone, so the last + /// segment of a sync had nothing left to trigger it — the pipeline sat on + /// fully downloaded, fully validated headers forever. Seen in the field on + /// a testnet restore: 1,046,289 headers buffered, stored tip frozen at + /// 1,474,000, peers healthy, chain locks still arriving. + /// + /// Asserts the completion too, not just the store: a regression that keeps + /// the last segment but leaves the manager in `Syncing` is the same stall + /// wearing a different hat. + #[tokio::test] + async fn test_tick_promotes_buffered_headers_with_no_further_messages() { + let mut manager = create_test_manager().await; + let tip = manager.tip().await.unwrap(); + let start_height = tip.height(); + + manager.pipeline.init(start_height, *tip.hash(), start_height + 2); + manager.progress.set_state(SyncState::Syncing); + + let (sender, _rx) = create_test_request_sender(); + // Issue the request the segment expects, so the headers below are a + // legitimate answer to it rather than unsolicited. + manager.pipeline.send_pending(&sender).unwrap(); + + // The headers land in the pipeline without the manager's own + // message-driven promotion running — the state the field stall was in. + let headers: Vec = + Header::dummy_chain(2, *tip.hash()).iter().map(HashedBlockHeader::from).collect(); + manager.pipeline.receive_headers(&headers).unwrap(); + assert!( + manager.pipeline.total_buffered() > 0, + "precondition: headers are downloaded but not yet promoted" + ); + assert_eq!( + manager.tip().await.unwrap().height(), + start_height, + "precondition: storage has not advanced" + ); + + // No further `Headers` will arrive. The periodic tick is the only + // thing left that can finish the job. + let events = manager.tick(&sender).await.unwrap(); + + assert!( + manager.tip().await.unwrap().height() > start_height, + "tick must promote buffered headers into storage" + ); + assert!( + events.iter().any(|e| matches!(e, SyncEvent::BlockHeadersStored { .. })), + "the promotion must be reported, not done silently" + ); + + // Completing the pipeline must also be reachable from a tick: an empty + // response closes the tip segment, and the next tick has to announce + // the sync rather than leaving the manager in `Syncing` forever. + manager.pipeline.send_pending(&sender).unwrap(); + manager.pipeline.receive_headers(&[]).unwrap(); + assert!(manager.pipeline.is_complete(), "precondition: nothing left to download"); + + let events = manager.tick(&sender).await.unwrap(); + + assert_eq!( + manager.state(), + SyncState::Synced, + "a tick over a complete pipeline must finish the sync" + ); + assert!( + events.iter().any(|e| matches!(e, SyncEvent::BlockHeaderSyncComplete { .. })), + "completion must be announced, or downstream managers never 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/sync_manager.rs b/dash-spv/src/sync/block_headers/sync_manager.rs index bb4b56d3a..3acb85c8b 100644 --- a/dash-spv/src/sync/block_headers/sync_manager.rs +++ b/dash-spv/src/sync/block_headers/sync_manager.rs @@ -154,12 +154,27 @@ impl SyncManager for BlockHeadersMana // During initial sync, send more requests and log progress if self.state() == SyncState::Syncing { + // Take, refill, then store — the same order `handle_headers_pipeline` + // uses, since draining can expose a segment the refill should pick + // up in this pass rather than the next one. + let ready_batches = self.pipeline.take_ready_to_store(); + let sent = self.pipeline.send_pending(requests)?; if sent > 0 { tracing::debug!("Tick: pipeline sent {} more requests", sent); } - return Ok(vec![]); + // Promotion is otherwise reachable only from `handle_headers_pipeline`, + // i.e. only when a `Headers` message arrives — so a sync has no way to + // finish itself once the final segment completes, because no further + // `Headers` will come. That is not hypothetical: a testnet restore was + // found frozen with 1,046,289 headers buffered in memory and the stored + // tip stuck at 1,474,000, peers connected and chain locks still + // arriving, until the app was restarted. Retrying on the 100ms tick + // makes a missed promotion self-heal whatever caused it to be missed. + let mut events = self.store_ready_batches(ready_batches).await?; + events.extend(self.finalize_sync_if_complete(requests).await?); + return Ok(events); } // Post-sync: check for stale block announcements