diff --git a/node/bft/src/primary.rs b/node/bft/src/primary.rs index eaa428802f..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,10 +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: `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) { @@ -200,42 +218,53 @@ 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, 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 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 { + // 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 node has finished syncing to round {}", + self_.ledger.latest_round() + ); + self_ + .apply_proposal_cache( + latest_certificate_round, + proposed_batch, + signed_proposals, + 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(()); } - // Write the proposed batch. - *self.proposed_batch.write() = proposed_batch; - // Write the signed proposals. - *self.signed_proposals.write() = signed_proposals; - // Writ 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")), @@ -245,6 +274,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, @@ -293,12 +352,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); @@ -312,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. @@ -1328,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()); @@ -1362,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; } @@ -1381,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; } @@ -1408,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; } @@ -1430,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; } @@ -1452,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; } @@ -1489,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..3a3452b24f 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,20 @@ 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); + 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 +931,20 @@ 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); + 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(); - } + 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 +1151,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. @@ -1702,6 +1741,7 @@ mod tests { advance_with_sync_blocks_lock: Default::default(), metrics: Default::default(), prepare_requests_lock: Default::default(), + synced_notify: Default::default(), } } 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 } }