Merge branch 'development' into st-monero-fixes

base_layer/core/src/base_node/state_machine_service/states/header_sync.rs

@@ -20,7 +20,7 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::{cmp::Ordering, time::Instant};
+use std::{cmp::Ordering, mem, time::Instant};
 
 use log::*;
 use tari_common_types::chain_metadata::ChainMetadata;
@@ -81,7 +81,7 @@ impl HeaderSyncState {
             shared.db.clone(),
             shared.consensus_rules.clone(),
             shared.connectivity.clone(),
-            &mut self.sync_peers,
+            mem::take(&mut self.sync_peers),
             shared.randomx_factory.clone(),
             &self.local_metadata,
         );

base_layer/core/src/base_node/state_machine_service/states/horizon_state_sync.rs

@@ -91,7 +91,7 @@ impl HorizonStateSync {
             db,
             connectivity,
             rules,
-            &sync_peers,
+            sync_peers,
             horizon_sync_height,
             prover,
             validator,

base_layer/core/src/base_node/sync/ban.rs

@@ -0,0 +1,65 @@
+//  Copyright 2023, The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use log::*;
+use tari_comms::{connectivity::ConnectivityRequester, peer_manager::NodeId};
+
+use crate::{base_node::BlockchainSyncConfig, common::BanReason};
+
+const LOG_TARGET: &str = "c::bn::sync";
+
+// Sync peers are banned if there exists a ban reason for the error and the peer is not on the allow list for sync.
+
+pub struct PeerBanManager {
+    config: BlockchainSyncConfig,
+    connectivity: ConnectivityRequester,
+}
+
+impl PeerBanManager {
+    pub fn new(config: BlockchainSyncConfig, connectivity: ConnectivityRequester) -> Self {
+        Self { config, connectivity }
+    }
+
+    pub async fn ban_peer_if_required(&mut self, node_id: &NodeId, ban_reason: &Option<BanReason>) {
+        if let Some(ban) = ban_reason {
+            if self.config.forced_sync_peers.contains(node_id) {
+                debug!(
+                    target: LOG_TARGET,
+                    "Not banning peer that is on the allow list for sync. Ban reason = {}", ban.reason()
+                );
+                return;
+            }
+            debug!(target: LOG_TARGET, "Sync peer {} removed from the sync peer list because {}", node_id, ban.reason());
+
+            match self
+                .connectivity
+                .ban_peer_until(node_id.clone(), ban.ban_duration, ban.reason().to_string())
+                .await
+            {
+                Ok(_) => {
+                    warn!(target: LOG_TARGET, "Banned sync peer {} for {:?} because {}", node_id, ban.ban_duration, ban.reason())
+                },
+                Err(err) => error!(target: LOG_TARGET, "Failed to ban sync peer {}: {}", node_id, err),
+            }
+        }
+    }
+}

base_layer/core/src/base_node/sync/block_sync/synchronizer.rs

@@ -37,12 +37,12 @@ use tracing;
 use super::error::BlockSyncError;
 use crate::{
     base_node::{
-        sync::{hooks::Hooks, rpc, SyncPeer},
+        sync::{ban::PeerBanManager, hooks::Hooks, rpc, SyncPeer},
         BlockchainSyncConfig,
     },
     blocks::{Block, ChainBlock},
     chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend},
-    common::{rolling_avg::RollingAverageTime, BanReason},
+    common::rolling_avg::RollingAverageTime,
     proto::base_node::SyncBlocksRequest,
     transactions::aggregated_body::AggregateBody,
     validation::{BlockBodyValidator, ValidationError},
@@ -57,6 +57,7 @@ pub struct BlockSynchronizer<B> {
     sync_peers: Vec<SyncPeer>,
     block_validator: Arc<dyn BlockBodyValidator<B>>,
     hooks: Hooks,
+    peer_ban_manager: PeerBanManager,
 }
 
 impl<B: BlockchainBackend + 'static> BlockSynchronizer<B> {
@@ -67,13 +68,15 @@ impl<B: BlockchainBackend + 'static> BlockSynchronizer<B> {
         sync_peers: Vec<SyncPeer>,
         block_validator: Arc<dyn BlockBodyValidator<B>>,
     ) -> Self {
+        let peer_ban_manager = PeerBanManager::new(config.clone(), connectivity.clone());
         Self {
             config,
             db,
             connectivity,
             sync_peers,
             block_validator,
             hooks: Default::default(),
+            peer_ban_manager,
         }
     }
 
@@ -175,27 +178,24 @@ impl<B: BlockchainBackend + 'static> BlockSynchronizer<B> {
             );
             match self.synchronize_blocks(sync_peer, client, max_latency).await {
                 Ok(_) => return Ok(()),
-
-                Err(err @ BlockSyncError::MaxLatencyExceeded { .. }) => {
-                    warn!(target: LOG_TARGET, "{}", err);
-                    latency_counter += 1;
-                    self.ban_peer_if_required(
-                        node_id,
-                        &BlockSyncError::get_ban_reason(&err, self.config.short_ban_period, self.config.ban_period),
-                    )
-                    .await;
-                    continue;
-                },
-
                 Err(err) => {
                     warn!(target: LOG_TARGET, "{}", err);
-                    self.ban_peer_if_required(
-                        node_id,
-                        &BlockSyncError::get_ban_reason(&err, self.config.short_ban_period, self.config.ban_period),
-                    )
-                    .await;
-                    self.remove_sync_peer(node_id);
-                    continue;
+                    let ban_reason =
+                        BlockSyncError::get_ban_reason(&err, self.config.short_ban_period, self.config.ban_period);
+                    if let Some(reason) = ban_reason {
+                        warn!(target: LOG_TARGET, "{}", err);
+                        self.peer_ban_manager
+                            .ban_peer_if_required(node_id, &Some(reason.clone()))
+                            .await;
+
+                        if reason.ban_duration > self.config.short_ban_period {
+                            self.remove_sync_peer(node_id);
+                        }
+                    }
+
+                    if let BlockSyncError::MaxLatencyExceeded { .. } = err {
+                        latency_counter += 1;
+                    }
                 },
             }
         }
@@ -413,32 +413,7 @@ impl<B: BlockchainBackend + 'static> BlockSynchronizer<B> {
         Ok(())
     }
 
-    // Sync peers are banned if there exists a ban reason for the error and the peer is not on the allow list for sync.
     // Sync peers are also removed from the list of sync peers if the ban duration is longer than the short ban period.
-    async fn ban_peer_if_required(&mut self, node_id: &NodeId, ban_reason: &Option<BanReason>) {
-        if let Some(ban) = ban_reason {
-            if self.config.forced_sync_peers.contains(node_id) {
-                debug!(
-                    target: LOG_TARGET,
-                    "Not banning peer that is on the allow list for sync. Ban reason = {}", ban.reason()
-                );
-                return;
-            }
-            debug!(target: LOG_TARGET, "Sync peer {} removed from the sync peer list because {}", node_id, ban.reason());
-
-            match self
-                .connectivity
-                .ban_peer_until(node_id.clone(), ban.ban_duration, ban.reason().to_string())
-                .await
-            {
-                Ok(_) => {
-                    warn!(target: LOG_TARGET, "Banned sync peer {} for {:?} because {}", node_id, ban.ban_duration, ban.reason())
-                },
-                Err(err) => error!(target: LOG_TARGET, "Failed to ban sync peer {}: {}", node_id, err),
-            }
-        }
-    }
-
     fn remove_sync_peer(&mut self, node_id: &NodeId) {
         if let Some(pos) = self.sync_peers.iter().position(|p| p.node_id() == node_id) {
             self.sync_peers.remove(pos);

base_layer/core/src/base_node/sync/header_sync/error.rs

@@ -28,10 +28,12 @@ use tari_comms::{
     protocol::rpc::{RpcError, RpcStatus},
 };
 
-use crate::{blocks::BlockError, chain_storage::ChainStorageError, validation::ValidationError};
+use crate::{blocks::BlockError, chain_storage::ChainStorageError, common::BanReason, validation::ValidationError};
 
 #[derive(Debug, thiserror::Error)]
 pub enum BlockHeaderSyncError {
+    #[error("No more sync peers available: {0}")]
+    NoMoreSyncPeers(String),
     #[error("RPC error: {0}")]
     RpcError(#[from] RpcError),
     #[error("RPC request failed: {0}")]
@@ -77,6 +79,8 @@ pub enum BlockHeaderSyncError {
         actual: Option<u128>,
         local: u128,
     },
+    #[error("This peer sent too many headers ({0}) in response to a chain split request")]
+    PeerSentTooManyHeaders(usize),
     #[error("Peer {peer} exceeded maximum permitted sync latency. latency: {latency:.2?}s, max: {max_latency:.2?}s")]
     MaxLatencyExceeded {
         peer: NodeId,
@@ -86,3 +90,42 @@ pub enum BlockHeaderSyncError {
     #[error("All sync peers exceeded max allowed latency")]
     AllSyncPeersExceedLatency,
 }
+
+impl BlockHeaderSyncError {
+    pub fn get_ban_reason(&self, short_ban: Duration, long_ban: Duration) -> Option<BanReason> {
+        match self {
+            // no ban
+            BlockHeaderSyncError::NoMoreSyncPeers(_) |
+            BlockHeaderSyncError::RpcError(_) |
+            BlockHeaderSyncError::RpcRequestError(_) |
+            BlockHeaderSyncError::SyncFailedAllPeers |
+            BlockHeaderSyncError::FailedToBan(_) |
+            BlockHeaderSyncError::AllSyncPeersExceedLatency |
+            BlockHeaderSyncError::ConnectivityError(_) |
+            BlockHeaderSyncError::NotInSync |
+            BlockHeaderSyncError::ChainStorageError(_) => None,
+
+            // short ban
+            err @ BlockHeaderSyncError::MaxLatencyExceeded { .. } => Some(BanReason {
+                reason: format!("{}", err),
+                ban_duration: short_ban,
+            }),
+
+            // long ban
+            err @ BlockHeaderSyncError::ReceivedInvalidHeader(_) |
+            err @ BlockHeaderSyncError::ValidationFailed(_) |
+            err @ BlockHeaderSyncError::FoundHashIndexOutOfRange(_, _) |
+            err @ BlockHeaderSyncError::StartHashNotFound(_) |
+            err @ BlockHeaderSyncError::InvalidBlockHeight { .. } |
+            err @ BlockHeaderSyncError::ChainSplitNotFound(_) |
+            err @ BlockHeaderSyncError::InvalidProtocolResponse(_) |
+            err @ BlockHeaderSyncError::ChainLinkBroken { .. } |
+            err @ BlockHeaderSyncError::BlockError(_) |
+            err @ BlockHeaderSyncError::PeerSentInaccurateChainMetadata { .. } |
+            err @ BlockHeaderSyncError::PeerSentTooManyHeaders(_) => Some(BanReason {
+                reason: format!("{}", err),
+                ban_duration: long_ban,
+            }),
+        }
+    }
+}