From 7b4410e0bb0729a1d40518fab37d825f0466c4fd Mon Sep 17 00:00:00 2001 From: Lukasz Rubaszewski <117115317+lrubasze@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:26:49 +0000 Subject: [PATCH 1/2] sync_service: bound AwaitingWarp wait with warp progress detection A warp sync that stays alive but never completes kept queued subscribe_all requests waiting indefinitely (livelock re-arming the deadline, or a starve/re-trigger cycle between Deciding and AwaitingWarp). Track warp's verified finalized height across deadline firings; after 3 windows without progress, commit AllForksOnly so subscribers get the checkpoint header. Warp stays resumable, so a later completion still reaches subscribers via Stop + resubscribe. Resolves TODO from #3268 (discussion_r3319656011). --- .../src/sync_service/substrate_compat.rs | 93 ++++++++++++++++--- 1 file changed, 78 insertions(+), 15 deletions(-) diff --git a/light-base/src/sync_service/substrate_compat.rs b/light-base/src/sync_service/substrate_compat.rs index 2aeed61201..20a9c7e420 100644 --- a/light-base/src/sync_service/substrate_compat.rs +++ b/light-base/src/sync_service/substrate_compat.rs @@ -54,6 +54,11 @@ const MODE_DECISION_TIMEOUT: Duration = Duration::from_secs(30); /// first peer locking us into the slow path when our local finalized is a stale checkpoint. const MODE_DECISION_MIN_PACKETS: usize = 2; +/// Deadline firings (`MODE_DECISION_TIMEOUT` apart) without warp progress before committing +/// AllForksOnly, so queued `SubscribeAll` requests always receive a response (the checkpoint +/// header). +const MODE_DECISION_MAX_WARP_STALLS: usize = 3; + /// Starts a sync service background task to synchronize a chain (relay chain or not) that is /// built with Substrate. pub(super) async fn start_substrate_compatible_chain( @@ -119,6 +124,8 @@ pub(super) async fn start_substrate_compatible_chain( mode: ModeState::Deciding, bootstrap_complete: false, deciding_packets_seen: 0, + warp_stalls: 0, + last_warp_progress: 0, mode_decision_deadline: future::Either::Left(Box::pin( platform.sleep(MODE_DECISION_TIMEOUT), )) @@ -144,6 +151,8 @@ pub(super) async fn start_substrate_compatible_chain( .as_mut() .unwrap_or_else(|| unreachable!()) .set_warp_completion_suppressed(true); + task.last_warp_progress = + warp_sync_progress(task.sync.as_ref().unwrap_or_else(|| unreachable!())); // Main loop of the syncing logic. // @@ -887,10 +896,7 @@ pub(super) async fn start_substrate_compatible_chain( target_finalized: finalized_block_height, }; // Keep the deadline armed as a warp-stall fallback. - task.mode_decision_deadline = future::Either::Left(Box::pin( - task.platform.sleep(MODE_DECISION_TIMEOUT), - )) - .fuse(); + arm_mode_decision_deadline(&mut task); // Allow warp completion to rebuild all_forks. task.sync .as_mut() @@ -1552,33 +1558,45 @@ pub(super) async fn start_substrate_compatible_chain( commit_all_forks_only(&mut task); } ModeState::AwaitingWarp { .. } => { - // TODO: warp never reaching `is_finished=true` keeps subscribe_all queued. - // https://github.com/paritytech/smoldot/pull/3268#discussion_r3319656011 - if warp_sync_can_proceed(task.sync.as_ref().unwrap_or_else(|| unreachable!())) { - task.mode_decision_deadline = future::Either::Left(Box::pin( - task.platform.sleep(MODE_DECISION_TIMEOUT), - )) - .fuse(); + let sync = task.sync.as_ref().unwrap_or_else(|| unreachable!()); + let progress = warp_sync_progress(sync); + let advanced = + progress > task.last_warp_progress || sync.has_pending_warp_completion(); + let can_proceed = warp_sync_can_proceed(sync); + task.last_warp_progress = progress; + task.warp_stalls = if advanced { 0 } else { task.warp_stalls + 1 }; + + if task.warp_stalls >= MODE_DECISION_MAX_WARP_STALLS { + log!( + &task.platform, + Debug, + &task.log_target, + "mode-decision; committed=AllForksOnly (warp stalled)", + warp_finalized = progress, + ); + commit_all_forks_only(&mut task); + } else if advanced || can_proceed { + arm_mode_decision_deadline(&mut task); log!( &task.platform, Debug, &task.log_target, "mode-decision; awaiting-warp deadline re-armed", + warp_finalized = progress, + stalls = task.warp_stalls, ); } else { // Warp starved: drop back to Deciding so a future warp-eligible peer // can re-trigger CommitWarpAhead instead of locking in AllForksOnly. // Re-suppress: no mode chosen yet, no subscribers drained. + // `warp_stalls` is kept so the cycle stays bounded. task.sync .as_mut() .unwrap_or_else(|| unreachable!()) .set_warp_completion_suppressed(true); task.mode = ModeState::Deciding; task.deciding_packets_seen = 0; - task.mode_decision_deadline = future::Either::Left(Box::pin( - task.platform.sleep(MODE_DECISION_TIMEOUT), - )) - .fuse(); + arm_mode_decision_deadline(&mut task); log!( &task.platform, Debug, @@ -1641,6 +1659,13 @@ struct Task { /// Below-gap packets observed while [`ModeState::Deciding`]; gates AllForksOnly commit. deciding_packets_seen: usize, + /// Deadline firings without warp progress; reset when warp advances. Survives the + /// `AwaitingWarp` → `Deciding` fallback so the starve/re-trigger cycle stays bounded. + warp_stalls: usize, + + /// Warp's verified finalized height at the previous deadline firing; progress detector. + last_warp_progress: u64, + /// Replaced with `pending` on mode commit so it never fires again. mode_decision_deadline: future::Fuse>, future::Pending<()>>>, @@ -1809,6 +1834,30 @@ fn warp_sync_can_proceed( } } +/// (Re-)arms the mode-decision deadline with a fresh `MODE_DECISION_TIMEOUT` window. +fn arm_mode_decision_deadline(task: &mut Task) { + task.mode_decision_deadline = + future::Either::Left(Box::pin(task.platform.sleep(MODE_DECISION_TIMEOUT))).fuse(); +} + +/// Warp sync's verified finalized height; an unchanged reading across a deadline window +/// means warp is alive but not progressing. +fn warp_sync_progress( + sync: &all::AllSync, +) -> u64 { + match sync.status() { + all::Status::WarpSyncFragments { + finalized_block_number, + .. + } + | all::Status::WarpSyncChainInformation { + finalized_block_number, + .. + } => finalized_block_number, + all::Status::Sync => sync.finalized_block_number(), + } +} + /// Responds to every queued `SubscribeAll` request. Each response allocates a fresh /// notification channel and pushes its sender into `task.all_notifications`. fn drain_pending_subscriptions(task: &mut Task) { @@ -1998,6 +2047,20 @@ mod tests { assert!(warp_sync_can_proceed(&sync)); } + // Advancing the height requires crypto-correct fragments (see module TODO). + #[test] + fn warp_progress_reports_warp_verified_finalized() { + let mut sync = fresh_sync(); + assert!(matches!( + sync.status(), + Status::WarpSyncFragments { .. } | Status::WarpSyncChainInformation { .. } + )); + assert_eq!(warp_sync_progress(&sync), 0); + let src = add_peer(&mut sync, 100); + dispatch_warp(&mut sync, src); + assert_eq!(warp_sync_progress(&sync), 0); + } + #[test] fn neighbor_packet_ignored_outside_deciding() { let mut sync = fresh_sync(); From 9bbd681ad4ceb1e2d1a6406d64d50593097386fe Mon Sep 17 00:00:00 2001 From: Lukasz Rubaszewski <117115317+lrubasze@users.noreply.github.com> Date: Mon, 13 Jul 2026 11:29:34 +0000 Subject: [PATCH 2/2] sync_service: commit AllForksOnly after a single stalled warp window A 30s window without warp progress now commits AllForksOnly directly, dropping the 3-window stall counter and the AwaitingWarp -> Deciding starve fallback. Committing early is cheap: warp stays unsuppressed and a later completion still resets subscriptions at the warped head. - keep the pending-completion check so a suppressed warp completion is never discarded by the commit - baseline last_warp_progress on AwaitingWarp entry so the window measures warp activity after entry, not progress left over from Deciding --- .../src/sync_service/substrate_compat.rs | 55 ++++++------------- 1 file changed, 17 insertions(+), 38 deletions(-) diff --git a/light-base/src/sync_service/substrate_compat.rs b/light-base/src/sync_service/substrate_compat.rs index 20a9c7e420..0def46bc36 100644 --- a/light-base/src/sync_service/substrate_compat.rs +++ b/light-base/src/sync_service/substrate_compat.rs @@ -54,11 +54,6 @@ const MODE_DECISION_TIMEOUT: Duration = Duration::from_secs(30); /// first peer locking us into the slow path when our local finalized is a stale checkpoint. const MODE_DECISION_MIN_PACKETS: usize = 2; -/// Deadline firings (`MODE_DECISION_TIMEOUT` apart) without warp progress before committing -/// AllForksOnly, so queued `SubscribeAll` requests always receive a response (the checkpoint -/// header). -const MODE_DECISION_MAX_WARP_STALLS: usize = 3; - /// Starts a sync service background task to synchronize a chain (relay chain or not) that is /// built with Substrate. pub(super) async fn start_substrate_compatible_chain( @@ -124,7 +119,6 @@ pub(super) async fn start_substrate_compatible_chain( mode: ModeState::Deciding, bootstrap_complete: false, deciding_packets_seen: 0, - warp_stalls: 0, last_warp_progress: 0, mode_decision_deadline: future::Either::Left(Box::pin( platform.sleep(MODE_DECISION_TIMEOUT), @@ -151,8 +145,6 @@ pub(super) async fn start_substrate_compatible_chain( .as_mut() .unwrap_or_else(|| unreachable!()) .set_warp_completion_suppressed(true); - task.last_warp_progress = - warp_sync_progress(task.sync.as_ref().unwrap_or_else(|| unreachable!())); // Main loop of the syncing logic. // @@ -895,6 +887,10 @@ pub(super) async fn start_substrate_compatible_chain( task.mode = ModeState::AwaitingWarp { target_finalized: finalized_block_height, }; + // Baseline for the stall check at the next deadline firing. + task.last_warp_progress = warp_sync_progress( + task.sync.as_ref().unwrap_or_else(|| unreachable!()), + ); // Keep the deadline armed as a warp-stall fallback. arm_mode_decision_deadline(&mut task); // Allow warp completion to rebuild all_forks. @@ -1560,22 +1556,13 @@ pub(super) async fn start_substrate_compatible_chain( ModeState::AwaitingWarp { .. } => { let sync = task.sync.as_ref().unwrap_or_else(|| unreachable!()); let progress = warp_sync_progress(sync); + // A pending completion counts as progress: committing here would + // discard it via `commit_all_forks_only`. let advanced = progress > task.last_warp_progress || sync.has_pending_warp_completion(); - let can_proceed = warp_sync_can_proceed(sync); task.last_warp_progress = progress; - task.warp_stalls = if advanced { 0 } else { task.warp_stalls + 1 }; - if task.warp_stalls >= MODE_DECISION_MAX_WARP_STALLS { - log!( - &task.platform, - Debug, - &task.log_target, - "mode-decision; committed=AllForksOnly (warp stalled)", - warp_finalized = progress, - ); - commit_all_forks_only(&mut task); - } else if advanced || can_proceed { + if advanced { arm_mode_decision_deadline(&mut task); log!( &task.platform, @@ -1583,26 +1570,20 @@ pub(super) async fn start_substrate_compatible_chain( &task.log_target, "mode-decision; awaiting-warp deadline re-armed", warp_finalized = progress, - stalls = task.warp_stalls, ); } else { - // Warp starved: drop back to Deciding so a future warp-eligible peer - // can re-trigger CommitWarpAhead instead of locking in AllForksOnly. - // Re-suppress: no mode chosen yet, no subscribers drained. - // `warp_stalls` is kept so the cycle stays bounded. - task.sync - .as_mut() - .unwrap_or_else(|| unreachable!()) - .set_warp_completion_suppressed(true); - task.mode = ModeState::Deciding; - task.deciding_packets_seen = 0; - arm_mode_decision_deadline(&mut task); + // Warp stalled for a full window: commit so queued `SubscribeAll` + // requests receive a response (the checkpoint header). Warp stays + // unsuppressed; a later completion still resets subscriptions at + // the warped head. log!( &task.platform, Debug, &task.log_target, - "mode-decision; warp starved, back to Deciding", + "mode-decision; committed=AllForksOnly (warp stalled)", + warp_finalized = progress, ); + commit_all_forks_only(&mut task); } } ModeState::Ready => { @@ -1659,11 +1640,9 @@ struct Task { /// Below-gap packets observed while [`ModeState::Deciding`]; gates AllForksOnly commit. deciding_packets_seen: usize, - /// Deadline firings without warp progress; reset when warp advances. Survives the - /// `AwaitingWarp` → `Deciding` fallback so the starve/re-trigger cycle stays bounded. - warp_stalls: usize, - - /// Warp's verified finalized height at the previous deadline firing; progress detector. + /// Warp's verified finalized height at `AwaitingWarp` entry or the previous deadline + /// firing; an unchanged reading across a `MODE_DECISION_TIMEOUT` window commits + /// AllForksOnly. last_warp_progress: u64, /// Replaced with `pending` on mode commit so it never fires again.