Merge pull request #4179 from ProvableHQ/fix/mainnet-syncing-when-synced

vicsn · web-flow · commit 3ec1b5112e01 · 2026-03-23T14:12:05.000+01:00
[Fix] Do not attempt to sync when only one block behind
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -788,7 +788,7 @@ workflows:
           filters:
             branches:
               only:
-                - release-mainnet-4.5.4
+                - fix/mainnet-syncing-when-synced
                 - canary
                 - testnet
                 - mainnet
@@ -798,7 +798,7 @@ workflows:
           filters:
             branches:
               only:
-                - release-mainnet-4.5.4
+                - fix/mainnet-syncing-when-synced
                 - canary
                 - testnet
                 - mainnet
@@ -810,7 +810,7 @@ workflows:
           filters:
             branches:
               only:
-                - release-mainnet-4.5.4
+                - fix/mainnet-syncing-when-synced
                 - canary
                 - testnet
                 - mainnet
diff --git a/node/bft/src/gateway.rs b/node/bft/src/gateway.rs
@@ -51,7 +51,7 @@ use snarkos_node_network::{
     get_repo_commit_hash,
     log_repo_sha_comparison,
 };
-use snarkos_node_sync::{InsertBlockResponseError, MAX_BLOCKS_BEHIND, communication_service::CommunicationService};
+use snarkos_node_sync::{MAX_BLOCKS_BEHIND, communication_service::CommunicationService};
 use snarkos_node_tcp::{
     Config,
     Connection,
@@ -68,6 +68,7 @@ use snarkvm::{
         narwhal::{BatchHeader, Data},
     },
     prelude::{Address, Field},
+    utilities::flatten_error,
 };
 
 use colored::Colorize;
@@ -651,15 +652,27 @@ impl<N: Network> Gateway<N> {
                     // Send the blocks to the sync module.
                     match sync_sender.insert_block_response(peer_ip, blocks.0, latest_consensus_version).await {
                         Ok(_) => Ok(true),
-                        Err(err @ InsertBlockResponseError::EmptyBlockResponse)
-                        | Err(err @ InsertBlockResponseError::NoConsensusVersion)
-                        | Err(err @ InsertBlockResponseError::ConsensusVersionMismatch { .. }) => {
-                            error!("Peer '{peer_ip}' sent an invalid block response - {err}");
-                            self.ip_ban_peer(peer_ip, Some(&err.to_string()));
-                            Err(err.into())
+                        Err(err) if err.is_benign() => {
+                            let err: anyhow::Error = err.into();
+                            let err = err.context(format!("Ignoring block response from peer '{peer_ip}'"));
+                            debug!("{}", flatten_error(err));
+                            Ok(true)
+                        }
+                        Err(err) if err.is_invalid_consensus_version() => {
+                            let err: anyhow::Error = err.into();
+                            let err = err.context(format!("Peer sent an invalid block response '{peer_ip}'"));
+
+                            let msg = flatten_error(&err);
+                            error!("{msg}");
+                            self.ip_ban_peer(peer_ip, Some(&msg));
+                            Err(err)
                         }
                         Err(err) => {
-                            warn!("Unable to process block response from '{peer_ip}' - {err}");
+                            let err: anyhow::Error = err.into();
+                            let err = err.context(format!("Peer '{peer_ip}' sent an invalid block response"));
+                            warn!("{}", flatten_error(err));
+
+                            // TODO(kaimast): This needs more testing to ensure disconnect is the correct action.
                             Ok(true)
                         }
                     }
diff --git a/node/bft/src/helpers/channels.rs b/node/bft/src/helpers/channels.rs
@@ -263,7 +263,7 @@ pub struct SyncSender<N: Network> {
         SocketAddr,
         Vec<Block<N>>,
         Option<ConsensusVersion>,
-        oneshot::Sender<Result<(), InsertBlockResponseError>>,
+        oneshot::Sender<Result<(), InsertBlockResponseError<N>>>,
     )>,
     pub tx_block_sync_remove_peer: mpsc::Sender<SocketAddr>,
     pub tx_block_sync_update_peer_locators: mpsc::Sender<(SocketAddr, BlockLocators<N>, oneshot::Sender<Result<()>>)>,
@@ -291,7 +291,7 @@ impl<N: Network> SyncSender<N> {
         peer_ip: SocketAddr,
         blocks: Vec<Block<N>>,
         latest_consensus_version: Option<ConsensusVersion>,
-    ) -> Result<(), InsertBlockResponseError> {
+    ) -> Result<(), InsertBlockResponseError<N>> {
         // Initialize a callback sender and receiver.
         let (callback_sender, callback_receiver) = oneshot::channel();
         // Send the request to advance with sync blocks.
@@ -320,7 +320,7 @@ pub struct SyncReceiver<N: Network> {
         SocketAddr,
         Vec<Block<N>>,
         Option<ConsensusVersion>,
-        oneshot::Sender<Result<(), InsertBlockResponseError>>,
+        oneshot::Sender<Result<(), InsertBlockResponseError<N>>>,
     )>,
     pub rx_block_sync_remove_peer: mpsc::Receiver<SocketAddr>,
     pub rx_block_sync_update_peer_locators: mpsc::Receiver<(SocketAddr, BlockLocators<N>, oneshot::Sender<Result<()>>)>,
diff --git a/node/bft/src/sync/mod.rs b/node/bft/src/sync/mod.rs
@@ -347,9 +347,8 @@ impl<N: Network> Sync<N> {
             }
         }
 
-        // Do not attempt to sync if there are no blocks to sync.
-        // This prevents redundant log messages and performing unnecessary computation.
-        if !self.block_sync.can_block_sync() {
+        // Do not attempt to sync if there are no blocks to sync, or we are too close to the tip.
+        if self.is_synced() {
             return;
         }
 
@@ -386,7 +385,7 @@ impl<N: Network> Sync<N> {
         peer_ip: SocketAddr,
         blocks: Vec<Block<N>>,
         latest_consensus_version: Option<ConsensusVersion>,
-    ) -> Result<(), InsertBlockResponseError> {
+    ) -> Result<(), InsertBlockResponseError<N>> {
         self.block_sync.insert_block_responses(peer_ip, blocks, latest_consensus_version)
 
         // No need to advance block sync here, as the new response will
diff --git a/node/src/client/router.rs b/node/src/client/router.rs
@@ -30,7 +30,6 @@ use snarkos_node_router::{
         UnconfirmedTransaction,
     },
 };
-use snarkos_node_sync::InsertBlockResponseError;
 use snarkos_node_tcp::{Connection, ConnectionSide, Tcp};
 use snarkvm::{
     console::network::{ConsensusVersion, Network},
@@ -220,21 +219,32 @@ impl<N: Network, C: ConsensusStorage<N>> Inbound<N> for Client<N, C> {
         latest_consensus_version: Option<ConsensusVersion>,
     ) -> bool {
         // We do not need to explicitly sync here because insert_block_response, will wake up the sync task.
-        if let Err(err) = self.sync.insert_block_responses(peer_ip, blocks, latest_consensus_version) {
-            warn!("Failed to insert block response from '{peer_ip}' - {err}");
-
-            // If the error indicates the peer missed an upgrade and forked, ban it.
-            if matches!(
-                err,
-                InsertBlockResponseError::ConsensusVersionMismatch { .. }
-                    | InsertBlockResponseError::NoConsensusVersion
-            ) {
+        match self.sync.insert_block_responses(peer_ip, blocks, latest_consensus_version) {
+            Ok(_) => true,
+            Err(err) if err.is_benign() => {
+                let err: anyhow::Error = err.into();
+                debug!("{}", flatten_error(err.context(format!("Ignoring block response from peer '{peer_ip}'"))));
+                true
+            }
+            Err(err) if err.is_invalid_consensus_version() => {
+                // If the error indicates the peer missed an upgrade and forked, ban it.
+                let err: anyhow::Error = err.into();
+                let err = err.context(format!("Peer sent an invalid block response '{peer_ip}'"));
+
+                let msg = flatten_error(&err);
+                error!("{msg}");
                 self.router().ip_ban_peer(peer_ip, Some(&err.to_string()));
+
+                false
             }
+            Err(err) => {
+                let err: anyhow::Error = err.into();
+                let err = err.context(format!("Failed to insert block response from '{peer_ip}'"));
+                warn!("{}", flatten_error(err));
 
-            false
-        } else {
-            true
+                // TODO(kaimast): This needs more testing to ensure disconnect is the correct action.
+                true
+            }
         }
     }
 
diff --git a/node/sync/src/block_sync.rs b/node/sync/src/block_sync.rs
@@ -118,7 +118,7 @@ pub struct BlockRequestsSummary {
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum InsertBlockResponseError {
+pub enum InsertBlockResponseError<N: Network> {
     #[error("Empty block response")]
     EmptyBlockResponse,
     #[error("The peer did not send a consensus version")]
@@ -127,10 +127,36 @@ pub enum InsertBlockResponseError {
         "The peer's consensus version for height {last_height} does not match ours: expected {expected_version}, got {peer_version}"
     )]
     ConsensusVersionMismatch { peer_version: ConsensusVersion, expected_version: ConsensusVersion, last_height: u32 },
+    #[error("Block Sync already advanced to block {height}")]
+    BlockSyncAlreadyAdvanced { height: u32 },
+    #[error("No such request for height {height}")]
+    NoSuchRequest { height: u32 },
+    #[error("Invalid block hash for height {height} from '{peer_ip}'")]
+    InvalidBlockHash { height: u32, peer_ip: SocketAddr },
+    #[error(
+        "The previous block hash in candidate block {height} from '{peer_ip}' is incorrect: expected {expected}, but got {actual}"
+    )]
+    InvalidPreviousBlockHash { height: u32, peer_ip: SocketAddr, expected: N::BlockHash, actual: N::BlockHash },
+    #[error("Candidate block {height} from '{peer_ip}' is malformed")]
+    MalformedBlock { height: u32, peer_ip: SocketAddr },
+    #[error("The sync pool did not request block {height} from '{peer_ip}'")]
+    WrongSyncPeer { height: u32, peer_ip: SocketAddr },
     #[error("{}", flatten_error(.0))]
     Other(#[from] anyhow::Error),
 }
 
+impl<N: Network> InsertBlockResponseError<N> {
+    /// Returns `true` if the error does not indicate malicious or faulty behavior.
+    pub fn is_benign(&self) -> bool {
+        matches!(self, Self::NoSuchRequest { .. } | Self::BlockSyncAlreadyAdvanced { .. })
+    }
+
+    // Returns true if the error is about an invalid consensus version.
+    pub fn is_invalid_consensus_version(&self) -> bool {
+        matches!(self, Self::ConsensusVersionMismatch { .. } | Self::NoConsensusVersion)
+    }
+}
+
 impl<N: Network> OutstandingRequest<N> {
     /// Get a reference to the IPs of peers that have not responded to the request (yet).
     fn sync_ips(&self) -> &IndexSet<SocketAddr> {
@@ -473,7 +499,7 @@ impl<N: Network> BlockSync<N> {
         peer_ip: SocketAddr,
         blocks: Vec<Block<N>>,
         latest_consensus_version: Option<ConsensusVersion>,
-    ) -> Result<(), InsertBlockResponseError> {
+    ) -> Result<(), InsertBlockResponseError<N>> {
         // Attempt to insert the block responses, and break if we encounter an error.
         let result = 'outer: {
             let Some(last_height) = blocks.as_slice().last().map(|b| b.height()) else {
@@ -500,9 +526,7 @@ impl<N: Network> BlockSync<N> {
 
             // Insert the candidate blocks into the sync pool.
             for block in blocks {
-                if let Err(error) = self.insert_block_response(peer_ip, block) {
-                    break 'outer Err(error.into());
-                }
+                self.insert_block_response(peer_ip, block)?;
             }
 
             Ok(())
@@ -916,35 +940,42 @@ impl<N: Network> BlockSync<N> {
 
     /// Inserts the given block response, after checking that the request exists and the response is well-formed.
     /// On success, this function removes the peer IP from the request sync peers and inserts the response.
-    fn insert_block_response(&self, peer_ip: SocketAddr, block: Block<N>) -> Result<()> {
+    fn insert_block_response(&self, peer_ip: SocketAddr, block: Block<N>) -> Result<(), InsertBlockResponseError<N>> {
         // Retrieve the block height.
         let height = block.height();
         let mut requests = self.requests.write();
 
         if self.ledger.contains_block_height(height) {
-            bail!("The sync request was removed because we already advanced");
+            return Err(InsertBlockResponseError::BlockSyncAlreadyAdvanced { height });
         }
 
-        let Some(entry) = requests.get_mut(&height) else { bail!("The sync pool did not request block {height}") };
+        let Some(entry) = requests.get_mut(&height) else {
+            return Err(InsertBlockResponseError::NoSuchRequest { height });
+        };
 
         // Retrieve the request entry for the candidate block.
         let (expected_hash, expected_previous_hash, sync_ips) = &entry.request;
 
         // Ensure the candidate block hash matches the expected hash.
         if let Some(expected_hash) = expected_hash {
             if block.hash() != *expected_hash {
-                bail!("The block hash for candidate block {height} from '{peer_ip}' is incorrect")
+                return Err(InsertBlockResponseError::InvalidBlockHash { height, peer_ip });
             }
         }
         // Ensure the previous block hash matches if it exists.
         if let Some(expected_previous_hash) = expected_previous_hash {
             if block.previous_hash() != *expected_previous_hash {
-                bail!("The previous block hash in candidate block {height} from '{peer_ip}' is incorrect")
+                return Err(InsertBlockResponseError::InvalidPreviousBlockHash {
+                    height,
+                    peer_ip,
+                    expected: *expected_previous_hash,
+                    actual: block.previous_hash(),
+                });
             }
         }
         // Ensure the sync pool requested this block from the given peer.
         if !sync_ips.contains(&peer_ip) {
-            bail!("The sync pool did not request block {height} from '{peer_ip}'")
+            return Err(InsertBlockResponseError::WrongSyncPeer { height, peer_ip });
         }
 
         // Remove the peer IP from the request entry.
@@ -953,7 +984,7 @@ impl<N: Network> BlockSync<N> {
         if let Some(existing_block) = &entry.response {
             // If the candidate block was already present, ensure it is the same block.
             if block != *existing_block {
-                bail!("Candidate block {height} from '{peer_ip}' is malformed");
+                return Err(InsertBlockResponseError::MalformedBlock { height, peer_ip });
             }
         } else {
             entry.response = Some(block.clone());

Original file line number	Diff line number	Diff line change
`@@ -347,9 +347,8 @@ impl<N: Network> Sync<N> {`
`347`	`347`	`}`
`348`	`348`	`}`
`349`	`349`
`350`		`- // Do not attempt to sync if there are no blocks to sync.`
`351`		`- // This prevents redundant log messages and performing unnecessary computation.`
`352`		`- if !self.block_sync.can_block_sync() {`
	`350`	`+ // Do not attempt to sync if there are no blocks to sync, or we are too close to the tip.`
	`351`	`+ if self.is_synced() {`
`353`	`352`	`return;`
`354`	`353`	`}`
`355`	`354`
`@@ -386,7 +385,7 @@ impl<N: Network> Sync<N> {`
`386`	`385`	`peer_ip: SocketAddr,`
`387`	`386`	`blocks: Vec<Block<N>>,`
`388`	`387`	`latest_consensus_version: Option<ConsensusVersion>,`
`389`		`- ) -> Result<(), InsertBlockResponseError> {`
	`388`	`+ ) -> Result<(), InsertBlockResponseError<N>> {`
`390`	`389`	`self.block_sync.insert_block_responses(peer_ip, blocks, latest_consensus_version)`
`391`	`390`
`392`	`391`	`// No need to advance block sync here, as the new response will`