From 5ec60a8d2a8e2404f14a63b95711a984840cea75 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 29 Apr 2026 13:58:08 +0000 Subject: [PATCH 1/5] Initial plan From 958a65ef5488de1719c2b5e92b62df5c2dc90c1c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 29 Apr 2026 14:29:48 +0000 Subject: [PATCH 2/5] fix: defer proposal cache load when ledger is too far behind cache round Instead of failing hard when the proposal cache round is more than MAX_GC_ROUNDS ahead of the current ledger, spawn a background task that polls the ledger every 10 seconds until it has synced within GC range of the cached proposal, then applies the cached state. Also move the load_proposal_cache() call in Primary::run to after sync.run() and gateway.run() so that the sync module and gateway are already running when the deferred background task eventually fires. Agent-Logs-Url: https://github.com/ProvableHQ/snarkOS/sessions/9204919c-ba7e-430e-a2c2-b57afdc851c9 Co-authored-by: vicsn <24724627+vicsn@users.noreply.github.com> --- node/bft/src/primary.rs | 71 ++++++++++++++++++++++++++++++++++------- 1 file changed, 60 insertions(+), 11 deletions(-) diff --git a/node/bft/src/primary.rs b/node/bft/src/primary.rs index eaa428802f..d6663b02b7 100644 --- a/node/bft/src/primary.rs +++ b/node/bft/src/primary.rs @@ -190,6 +190,11 @@ impl Primary { } /// Load the proposal cache file and update the Primary state with the stored data. + /// + /// If the proposal cache is more than `MAX_GC_ROUNDS` ahead of the current ledger, loading is + /// deferred: a background task is spawned that polls the ledger until it has synced within GC + /// range of the cached proposal, at which point the cached state is applied. This allows the + /// node to start normally after restoring an older ledger snapshot instead of failing hard. async fn load_proposal_cache(&self) -> Result<()> { // Fetch the signed proposals from the file system if it exists. match ProposalCache::::exists(&self.node_data_dir) { @@ -200,25 +205,65 @@ impl Primary { let (latest_certificate_round, proposed_batch, signed_proposals, pending_certificates) = proposal_cache.into(); - // Verify that the proposal cache is not too far ahead of the ledger. - // If the cache round exceeds the ledger round by more than MAX_GC_ROUNDS, the ledger - // snapshot is too old to recover from the cached state. The operator must restore a - // more recent ledger snapshot before restarting the node. let ledger_round = self.ledger.latest_round(); let max_gc_rounds = BatchHeader::::MAX_GC_ROUNDS as u64; + + // Check whether the proposal cache is too far ahead of the current ledger. if latest_certificate_round > ledger_round.saturating_add(max_gc_rounds) { - bail!( - "The proposal cache (round {latest_certificate_round}) is more than {max_gc_rounds} \ - rounds ahead of the ledger (round {ledger_round}). \ - Please restore a more recent ledger snapshot before restarting the node." + // The ledger snapshot is older than MAX_GC_ROUNDS relative to the cached + // proposal. Rather than failing hard, defer loading the cache until the + // node has synced within GC range of the cached proposal round. + info!( + "The proposal cache (round {latest_certificate_round}) is more than \ + {max_gc_rounds} GC rounds ahead of the ledger (round {ledger_round}). \ + Deferring proposal cache load until the ledger syncs within GC range." ); + let self_ = self.clone(); + self.spawn(async move { + // Poll until the ledger has synced within GC rounds of the cached proposal. + loop { + let current_ledger_round = self_.ledger.latest_round(); + if latest_certificate_round <= current_ledger_round.saturating_add(max_gc_rounds) { + break; + } + tokio::time::sleep(Duration::from_secs(10)).await; + } + info!( + "Loading the deferred proposal cache (round {latest_certificate_round}) \ + now that the ledger has synced to round {}", + self_.ledger.latest_round() + ); + // Write the proposed batch. + *self_.proposed_batch.write() = proposed_batch; + // Write the signed proposals. + *self_.signed_proposals.write() = signed_proposals; + // Write the propose lock. + *self_.propose_lock.lock().await = latest_certificate_round; + // Update the storage with the pending certificates. + for certificate in pending_certificates { + let batch_id = certificate.batch_id(); + // We use a dummy IP because the node should not need to request from any peers. + // The storage should have stored all the transmissions. If not, we simply + // skip the certificate. + if let Err(err) = + self_.sync_with_certificate_from_peer::(DUMMY_SELF_IP, certificate).await + { + let err = err.context(format!( + "Failed to load stored certificate {} from proposal cache", + fmt_id(batch_id) + )); + warn!("{}", &flatten_error(err)); + } + } + }); + return Ok(()); } // Write the proposed batch. *self.proposed_batch.write() = proposed_batch; // Write the signed proposals. *self.signed_proposals.write() = signed_proposals; - // Writ the propose lock. + // Write the propose lock. *self.propose_lock.lock().await = latest_certificate_round; // Update the storage with the pending certificates. @@ -293,12 +338,16 @@ impl Primary { let (sync_sender, sync_receiver) = init_sync_channels(); // Next, initialize the sync module and sync the storage from ledger. self.sync.initialize(sync_callback)?; - // Next, load and process the proposal cache before running the sync module. - self.load_proposal_cache().await?; // Next, run the sync module. self.sync.run(ping, sync_receiver).await?; // Next, initialize the gateway. self.gateway.run(primary_sender, worker_senders, Some(sync_sender)).await; + // Next, load and process the proposal cache. + // Note: This is done after starting the sync module and gateway so that, if the proposal + // cache is more than MAX_GC_ROUNDS ahead of the current ledger, a background task can be + // spawned to defer loading until the ledger has synced within GC range of the cached + // proposal round. + self.load_proposal_cache().await?; // Lastly, start the primary handlers. // Note: This ensures the primary does not start communicating before syncing is complete. self.start_handlers(primary_receiver); From 53eec1cbd2e49e3f1e2d8d703fd56255ed05dd40 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 29 Apr 2026 14:34:07 +0000 Subject: [PATCH 3/5] refactor: extract apply_proposal_cache helper, add periodic log in deferred polling loop Agent-Logs-Url: https://github.com/ProvableHQ/snarkOS/sessions/9204919c-ba7e-430e-a2c2-b57afdc851c9 Co-authored-by: vicsn <24724627+vicsn@users.noreply.github.com> --- node/bft/src/primary.rs | 101 +++++++++++++++++++++++----------------- 1 file changed, 57 insertions(+), 44 deletions(-) diff --git a/node/bft/src/primary.rs b/node/bft/src/primary.rs index d6663b02b7..54f748b9c7 100644 --- a/node/bft/src/primary.rs +++ b/node/bft/src/primary.rs @@ -221,11 +221,22 @@ impl Primary { let self_ = self.clone(); self.spawn(async move { // Poll until the ledger has synced within GC rounds of the cached proposal. + // Log periodically so operators can see that the node is still waiting. + let mut iters: u32 = 0; loop { let current_ledger_round = self_.ledger.latest_round(); if latest_certificate_round <= current_ledger_round.saturating_add(max_gc_rounds) { break; } + // Log a reminder every ~5 minutes (every 30 × 10-second iterations). + if iters % 30 == 0 { + info!( + "Waiting to load the proposal cache (round {latest_certificate_round}): \ + ledger is at round {current_ledger_round}, need to reach at least round {}.", + latest_certificate_round.saturating_sub(max_gc_rounds) + ); + } + iters = iters.saturating_add(1); tokio::time::sleep(Duration::from_secs(10)).await; } info!( @@ -233,54 +244,26 @@ impl Primary { now that the ledger has synced to round {}", self_.ledger.latest_round() ); - // Write the proposed batch. - *self_.proposed_batch.write() = proposed_batch; - // Write the signed proposals. - *self_.signed_proposals.write() = signed_proposals; - // Write the propose lock. - *self_.propose_lock.lock().await = latest_certificate_round; - // Update the storage with the pending certificates. - for certificate in pending_certificates { - let batch_id = certificate.batch_id(); - // We use a dummy IP because the node should not need to request from any peers. - // The storage should have stored all the transmissions. If not, we simply - // skip the certificate. - if let Err(err) = - self_.sync_with_certificate_from_peer::(DUMMY_SELF_IP, certificate).await - { - let err = err.context(format!( - "Failed to load stored certificate {} from proposal cache", - fmt_id(batch_id) - )); - warn!("{}", &flatten_error(err)); - } - } + self_ + .apply_proposal_cache( + latest_certificate_round, + proposed_batch, + signed_proposals, + pending_certificates, + ) + .await; }); return Ok(()); } - // Write the proposed batch. - *self.proposed_batch.write() = proposed_batch; - // Write the signed proposals. - *self.signed_proposals.write() = signed_proposals; - // Write the propose lock. - *self.propose_lock.lock().await = latest_certificate_round; - - // Update the storage with the pending certificates. - for certificate in pending_certificates { - let batch_id = certificate.batch_id(); - // We use a dummy IP because the node should not need to request from any peers. - // The storage should have stored all the transmissions. If not, we simply - // skip the certificate. - if let Err(err) = self.sync_with_certificate_from_peer::(DUMMY_SELF_IP, certificate).await - { - let err = err.context(format!( - "Failed to load stored certificate {} from proposal cache", - fmt_id(batch_id) - )); - warn!("{}", &flatten_error(err)); - } - } + // The cache is within GC range; apply it immediately. + self.apply_proposal_cache( + latest_certificate_round, + proposed_batch, + signed_proposals, + pending_certificates, + ) + .await; Ok(()) } Err(err) => Err(err.context("Failed to read the signed proposals from the file system")), @@ -290,6 +273,36 @@ impl Primary { } } + /// Applies the proposal cache state to the primary: writes the proposed batch, signed + /// proposals, propose lock, and loads any pending certificates into storage. + async fn apply_proposal_cache( + &self, + latest_certificate_round: u64, + proposed_batch: Option>, + signed_proposals: SignedProposals, + pending_certificates: IndexSet>, + ) { + // Write the proposed batch. + *self.proposed_batch.write() = proposed_batch; + // Write the signed proposals. + *self.signed_proposals.write() = signed_proposals; + // Write the propose lock. + *self.propose_lock.lock().await = latest_certificate_round; + + // Update the storage with the pending certificates. + for certificate in pending_certificates { + let batch_id = certificate.batch_id(); + // We use a dummy IP because the node should not need to request from any peers. + // The storage should have stored all the transmissions. If not, we simply + // skip the certificate. + if let Err(err) = self.sync_with_certificate_from_peer::(DUMMY_SELF_IP, certificate).await { + let err = + err.context(format!("Failed to load stored certificate {} from proposal cache", fmt_id(batch_id))); + warn!("{}", &flatten_error(err)); + } + } + } + /// Run the primary instance. pub async fn run( &self, From 8963b97354ebfd455b47c07565702aa49d2cca91 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 30 Apr 2026 14:28:01 +0000 Subject: [PATCH 4/5] fix: replace polling with Notify-based mechanism for deferred proposal cache loading - Add `synced_notify: Notify` to `BlockSync` that fires whenever the block sync state transitions to `SyncStatus::Synced` - Change `SyncState::update_is_block_synced` (and the three callers) to return `bool` indicating whether the state just transitioned to Synced - Add `BlockSync::wait_until_block_synced()` and `Sync::wait_until_synced()` that suspend until the Synced notification is received - Add `proposal_cache_loaded: Arc` to `Primary`; gate `Primary::is_synced()` on this flag in addition to the block-sync state - In `load_proposal_cache`, mark the flag false before spawning the deferred task; the task awaits `sync.wait_until_synced()` then applies the cache and restores the flag to true - Update all `self_.sync.is_synced()` calls in Primary handlers to go through `self_.is_synced()` so the proposal_cache_loaded gate is respected Agent-Logs-Url: https://github.com/ProvableHQ/snarkOS/sessions/fda630fe-85eb-4dfa-be54-97cbb824bca6 Co-authored-by: vicsn <24724627+vicsn@users.noreply.github.com> --- node/bft/src/primary.rs | 74 ++++++++++++++------------ node/bft/src/sync/mod.rs | 8 +++ node/sync/src/block_sync.rs | 65 +++++++++++++++++----- node/sync/src/block_sync/sync_state.rs | 31 ++++++----- 4 files changed, 117 insertions(+), 61 deletions(-) diff --git a/node/bft/src/primary.rs b/node/bft/src/primary.rs index 54f748b9c7..5877182e3c 100644 --- a/node/bft/src/primary.rs +++ b/node/bft/src/primary.rs @@ -84,7 +84,11 @@ use std::{ collections::{HashMap, HashSet}, future::Future, net::SocketAddr, - sync::{Arc, OnceLock}, + sync::{ + Arc, + OnceLock, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; #[cfg(not(feature = "locktick"))] @@ -137,6 +141,11 @@ pub struct Primary { propose_lock: Arc>, /// The node configuration directory. node_data_dir: NodeDataDir, + /// Whether the proposal cache has been loaded (or there is no cache to load). + /// + /// This is used to prevent the primary from being considered synced before the proposal cache + /// has been applied, avoiding proposing stale batches after restoring an old ledger snapshot. + proposal_cache_loaded: Arc, } impl Primary { @@ -186,15 +195,19 @@ impl Primary { handles: Default::default(), propose_lock: Default::default(), node_data_dir, + // Default to `true` (loaded / not pending): the proposal cache is not pending + // unless `load_proposal_cache` explicitly defers it. + proposal_cache_loaded: Arc::new(AtomicBool::new(true)), }) } /// Load the proposal cache file and update the Primary state with the stored data. /// /// If the proposal cache is more than `MAX_GC_ROUNDS` ahead of the current ledger, loading is - /// deferred: a background task is spawned that polls the ledger until it has synced within GC - /// range of the cached proposal, at which point the cached state is applied. This allows the - /// node to start normally after restoring an older ledger snapshot instead of failing hard. + /// deferred: `proposal_cache_loaded` is set to `false` and a background task is spawned that + /// awaits the first time the node becomes block-synced, then applies the cached state and + /// restores the flag. While the cache is pending, [`Self::is_synced`] returns `false` so + /// the primary will not propose or sign batches until the cache has been loaded. async fn load_proposal_cache(&self) -> Result<()> { // Fetch the signed proposals from the file system if it exists. match ProposalCache::::exists(&self.node_data_dir) { @@ -211,37 +224,23 @@ impl Primary { // Check whether the proposal cache is too far ahead of the current ledger. if latest_certificate_round > ledger_round.saturating_add(max_gc_rounds) { // The ledger snapshot is older than MAX_GC_ROUNDS relative to the cached - // proposal. Rather than failing hard, defer loading the cache until the - // node has synced within GC range of the cached proposal round. + // proposal. Rather than failing hard, mark the cache as not yet loaded and + // defer applying it until the block sync reaches the required height. info!( "The proposal cache (round {latest_certificate_round}) is more than \ {max_gc_rounds} GC rounds ahead of the ledger (round {ledger_round}). \ - Deferring proposal cache load until the ledger syncs within GC range." + Deferring proposal cache load until the node has finished syncing." ); + // Mark the cache as pending: `is_synced()` will return `false` until + // this flag is cleared after the cache is applied. + self.proposal_cache_loaded.store(false, Ordering::Release); let self_ = self.clone(); self.spawn(async move { - // Poll until the ledger has synced within GC rounds of the cached proposal. - // Log periodically so operators can see that the node is still waiting. - let mut iters: u32 = 0; - loop { - let current_ledger_round = self_.ledger.latest_round(); - if latest_certificate_round <= current_ledger_round.saturating_add(max_gc_rounds) { - break; - } - // Log a reminder every ~5 minutes (every 30 × 10-second iterations). - if iters % 30 == 0 { - info!( - "Waiting to load the proposal cache (round {latest_certificate_round}): \ - ledger is at round {current_ledger_round}, need to reach at least round {}.", - latest_certificate_round.saturating_sub(max_gc_rounds) - ); - } - iters = iters.saturating_add(1); - tokio::time::sleep(Duration::from_secs(10)).await; - } + // Wait for the block sync to reach the Synced state for the first time. + self_.sync.wait_until_synced().await; info!( "Loading the deferred proposal cache (round {latest_certificate_round}) \ - now that the ledger has synced to round {}", + now that the node has finished syncing to round {}", self_.ledger.latest_round() ); self_ @@ -252,6 +251,8 @@ impl Primary { pending_certificates, ) .await; + // Mark the proposal cache as loaded so `is_synced()` can return `true`. + self_.proposal_cache_loaded.store(true, Ordering::Release); }); return Ok(()); } @@ -374,8 +375,11 @@ impl Primary { } /// Returns `true` if the primary is synced. + /// + /// Returns `false` when the block sync is not yet complete, or when there is a pending + /// proposal cache that has not yet been applied (see [`Self::load_proposal_cache`]). pub fn is_synced(&self) -> bool { - self.sync.is_synced() + self.proposal_cache_loaded.load(Ordering::Acquire) && self.sync.is_synced() } /// Returns the gateway. @@ -1390,7 +1394,7 @@ impl Primary { self.spawn(async move { while let Some((peer_ip, primary_certificate)) = rx_primary_ping.recv().await { // If the primary is not synced, then do not process the primary ping. - if self_.sync.is_synced() { + if self_.is_synced() { trace!("Processing new primary ping from '{peer_ip}'"); } else { trace!("Skipping a primary ping from '{peer_ip}' {}", "(node is syncing)".dimmed()); @@ -1424,7 +1428,7 @@ impl Primary { loop { tokio::time::sleep(Duration::from_millis(WORKER_PING_IN_MS)).await; // If the primary is not synced, then do not broadcast the worker ping(s). - if !self_.sync.is_synced() { + if !self_.is_synced() { trace!("Skipping worker ping(s) {}", "(node is syncing)".dimmed()); continue; } @@ -1443,7 +1447,7 @@ impl Primary { tokio::time::sleep(Duration::from_millis(MAX_BATCH_DELAY_IN_MS)).await; let current_round = self_.current_round(); // If the primary is not synced, then do not propose a batch. - if !self_.sync.is_synced() { + if !self_.is_synced() { debug!("Skipping batch proposal for round {current_round} {}", "(node is syncing)".dimmed()); continue; } @@ -1470,7 +1474,7 @@ impl Primary { self.spawn(async move { while let Some((peer_ip, batch_propose)) = rx_batch_propose.recv().await { // If the primary is not synced, then do not sign the batch. - if !self_.sync.is_synced() { + if !self_.is_synced() { trace!("Skipping a batch proposal from '{peer_ip}' {}", "(node is syncing)".dimmed()); continue; } @@ -1492,7 +1496,7 @@ impl Primary { self.spawn(async move { while let Some((peer_ip, batch_signature)) = rx_batch_signature.recv().await { // If the primary is not synced, then do not store the signature. - if !self_.sync.is_synced() { + if !self_.is_synced() { trace!("Skipping a batch signature from '{peer_ip}' {}", "(node is syncing)".dimmed()); continue; } @@ -1514,7 +1518,7 @@ impl Primary { self.spawn(async move { while let Some((peer_ip, batch_certificate)) = rx_batch_certified.recv().await { // If the primary is not synced, then do not store the certificate. - if !self_.sync.is_synced() { + if !self_.is_synced() { trace!("Skipping a certified batch from '{peer_ip}' {}", "(node is syncing)".dimmed()); continue; } @@ -1551,7 +1555,7 @@ impl Primary { // Sleep briefly. tokio::time::sleep(Duration::from_millis(MAX_BATCH_DELAY_IN_MS)).await; // If the primary is not synced, then do not increment to the next round. - if !self_.sync.is_synced() { + if !self_.is_synced() { trace!("Skipping round increment {}", "(node is syncing)".dimmed()); continue; } diff --git a/node/bft/src/sync/mod.rs b/node/bft/src/sync/mod.rs index bf8bfab8cc..8680c39737 100644 --- a/node/bft/src/sync/mod.rs +++ b/node/bft/src/sync/mod.rs @@ -993,6 +993,14 @@ impl Sync { self.block_sync.is_block_synced() } + /// Waits asynchronously until the node is fully synced (block-level sync is complete). + /// + /// Returns immediately if the node is already synced; otherwise suspends until the + /// block sync state transitions to [`snarkos_node_sync::BftSyncMode`] synced. + pub async fn wait_until_synced(&self) { + self.block_sync.wait_until_block_synced().await; + } + /// Returns the number of blocks the node is behind the greatest peer height. pub fn num_blocks_behind(&self) -> Option { self.block_sync.num_blocks_behind() diff --git a/node/sync/src/block_sync.rs b/node/sync/src/block_sync.rs index fd76998ccb..00483c0791 100644 --- a/node/sync/src/block_sync.rs +++ b/node/sync/src/block_sync.rs @@ -224,6 +224,12 @@ pub struct BlockSync { /// Gets notified when we received a new block response. response_notify: Notify, + /// Gets notified when the block sync state transitions to [`SyncStatus::Synced`]. + /// + /// This is fired each time the status transitions to Synced. Waiters that subscribe via + /// [`Self::wait_until_block_synced`] will be woken as soon as the node is done syncing. + synced_notify: Notify, + /// Tracks sync speed metrics: BlockSyncMetrics, @@ -246,6 +252,7 @@ impl BlockSync { sync_state: RwLock::new(sync_state), peer_notify: Default::default(), response_notify: Default::default(), + synced_notify: Default::default(), locators: Default::default(), requests: Default::default(), common_ancestors: Default::default(), @@ -271,6 +278,22 @@ impl BlockSync { self.response_notify.notified().await } + /// Waits asynchronously until the block sync state reaches [`SyncStatus::Synced`]. + /// + /// Returns immediately if the node is already synced. + /// Otherwise blocks until [`Self::synced_notify`] fires. + pub async fn wait_until_block_synced(&self) { + loop { + // Register the waiter BEFORE checking the condition to avoid a race where the + // notification fires between the check and the await. + let notified = self.synced_notify.notified(); + if self.is_block_synced() { + return; + } + notified.await; + } + } + /// Returns `true` if the node is synced up to the latest block (within the given tolerance). #[inline] pub fn is_block_synced(&self) -> bool { @@ -873,14 +896,19 @@ impl BlockSync { } // -- Finally, update sync state and notify the sync loop about the change. -- - if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { - self.sync_state.write().set_greatest_peer_height(greatest_peer_height); - } else { - error!("Got new block locators but greatest peer height is zero."); - } + let just_became_synced = + if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { + self.sync_state.write().set_greatest_peer_height(greatest_peer_height) + } else { + error!("Got new block locators but greatest peer height is zero."); + false + }; // Even if the greatest peer height did not change, we still received new block locators // that the sync loop might need to proceed. self.peer_notify.notify_one(); + if just_became_synced { + self.synced_notify.notify_one(); + } Ok(()) } @@ -902,15 +930,19 @@ impl BlockSync { self.remove_block_requests_to_peer(peer_ip); // Update sync state, because the greatest peer height may have decreased. - if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { - self.sync_state.write().set_greatest_peer_height(greatest_peer_height); - } else { - // There are no more peers left. - self.sync_state.write().clear_greatest_peer_height(); - } + let just_became_synced = + if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { + self.sync_state.write().set_greatest_peer_height(greatest_peer_height) + } else { + // There are no more peers left. + self.sync_state.write().clear_greatest_peer_height() + }; // Notify the sync loop that something changed. self.peer_notify.notify_one(); + if just_became_synced { + self.synced_notify.notify_one(); + } } } @@ -1117,15 +1149,20 @@ impl BlockSync { /// This is a no-op if `new_height` is equal or less to the current sync height. pub fn set_sync_height(&self, new_height: u32) { // Scope state lock to avoid locking state and metrics at the same time. - let fully_synced = { + let (just_became_synced, fully_synced) = { let mut state = self.sync_state.write(); - state.set_sync_height(new_height); - !state.can_issue_new_block_requests() + let just_became_synced = state.set_sync_height(new_height); + (just_became_synced, !state.can_issue_new_block_requests()) }; if fully_synced { self.metrics.mark_fully_synced(); } + + // Notify any tasks waiting for the node to become synced. + if just_became_synced { + self.synced_notify.notify_one(); + } } /// Inserts a block request for the given height. diff --git a/node/sync/src/block_sync/sync_state.rs b/node/sync/src/block_sync/sync_state.rs index 13834ebad0..02031ea692 100644 --- a/node/sync/src/block_sync/sync_state.rs +++ b/node/sync/src/block_sync/sync_state.rs @@ -128,43 +128,47 @@ impl SyncState { /// Update the height we are synced to. /// If the value is lower than the current height, the sync height remains unchanged. - pub fn set_sync_height(&mut self, sync_height: u32) { + /// Returns `true` if this call caused the state to transition to [`SyncStatus::Synced`]. + pub fn set_sync_height(&mut self, sync_height: u32) -> bool { if sync_height <= self.sync_height { - return; + return false; } trace!("Sync height increased from {old_height} to {sync_height}", old_height = self.sync_height); self.sync_height = sync_height; - self.update_is_block_synced(); + self.update_is_block_synced() } /// Update the greatest known height of a connected peer. - pub fn set_greatest_peer_height(&mut self, peer_height: u32) { + /// Returns `true` if this call caused the state to transition to [`SyncStatus::Synced`]. + pub fn set_greatest_peer_height(&mut self, peer_height: u32) -> bool { if let Some(old_height) = self.greatest_peer_height { match old_height.cmp(&peer_height) { - Ordering::Equal => return, + Ordering::Equal => return false, Ordering::Greater => warn!("Greatest peer height reduced from {old_height} to {peer_height}"), Ordering::Less => trace!("Greatest peer height increased from {old_height} to {peer_height}"), } } self.greatest_peer_height = Some(peer_height); - self.update_is_block_synced(); + self.update_is_block_synced() } /// Remove the greatest peer height (used when all peers disconnect). - pub fn clear_greatest_peer_height(&mut self) { + /// Returns `true` if this call caused the state to transition to [`SyncStatus::Synced`]. + pub fn clear_greatest_peer_height(&mut self) -> bool { // No-op if there is no change. if self.greatest_peer_height.is_none() { - return; + return false; } self.greatest_peer_height = None; - self.update_is_block_synced(); + self.update_is_block_synced() } /// Updates the state of `is_block_synced` for the sync module. - fn update_is_block_synced(&mut self) { + /// Returns `true` if the state JUST transitioned to [`SyncStatus::Synced`]. + fn update_is_block_synced(&mut self) -> bool { trace!( "Updating is_block_synced: greatest_peer_height={greatest_peer:?}, current_height={current}, status={status:?}", greatest_peer = self.greatest_peer_height, @@ -183,9 +187,9 @@ impl SyncState { None => SyncStatus::Unsynced, }; - // Return early if the state is unchanged + // Return early if the state is unchanged. if new_status == old_status { - return; + return false; } // Measure how long sync took. @@ -223,5 +227,8 @@ impl SyncState { // Update the `IS_SYNCED` metric. #[cfg(feature = "metrics")] metrics::gauge(metrics::bft::IS_SYNCED, self.status == SyncStatus::Synced); + + // Return whether the state JUST transitioned to Synced. + self.status == SyncStatus::Synced } } From 65504fbffd148baa273dc1ad165e1bbb1a833710 Mon Sep 17 00:00:00 2001 From: Kai Mast Date: Thu, 30 Apr 2026 12:41:27 -0700 Subject: [PATCH 5/5] fix(node/sync): ensure tests compile --- node/sync/src/block_sync.rs | 31 +++++++++++++++++-------------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/node/sync/src/block_sync.rs b/node/sync/src/block_sync.rs index 00483c0791..3a3452b24f 100644 --- a/node/sync/src/block_sync.rs +++ b/node/sync/src/block_sync.rs @@ -896,13 +896,14 @@ impl BlockSync { } // -- Finally, update sync state and notify the sync loop about the change. -- - let just_became_synced = - if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { - self.sync_state.write().set_greatest_peer_height(greatest_peer_height) - } else { - error!("Got new block locators but greatest peer height is zero."); - false - }; + let just_became_synced = if let Some(greatest_peer_height) = + self.locators.read().values().map(|l| l.latest_locator_height()).max() + { + self.sync_state.write().set_greatest_peer_height(greatest_peer_height) + } else { + error!("Got new block locators but greatest peer height is zero."); + false + }; // Even if the greatest peer height did not change, we still received new block locators // that the sync loop might need to proceed. self.peer_notify.notify_one(); @@ -930,13 +931,14 @@ impl BlockSync { self.remove_block_requests_to_peer(peer_ip); // Update sync state, because the greatest peer height may have decreased. - let just_became_synced = - if let Some(greatest_peer_height) = self.locators.read().values().map(|l| l.latest_locator_height()).max() { - self.sync_state.write().set_greatest_peer_height(greatest_peer_height) - } else { - // There are no more peers left. - self.sync_state.write().clear_greatest_peer_height() - }; + let just_became_synced = if let Some(greatest_peer_height) = + self.locators.read().values().map(|l| l.latest_locator_height()).max() + { + self.sync_state.write().set_greatest_peer_height(greatest_peer_height) + } else { + // There are no more peers left. + self.sync_state.write().clear_greatest_peer_height() + }; // Notify the sync loop that something changed. self.peer_notify.notify_one(); @@ -1739,6 +1741,7 @@ mod tests { advance_with_sync_blocks_lock: Default::default(), metrics: Default::default(), prepare_requests_lock: Default::default(), + synced_notify: Default::default(), } }