do not enter sync mode if peer is at an invalid hash (#8071)

Unfortunately, it is very hard to add test for this because the syncing logic is in ClientActor. I already tested this in localnet and Robin tested it on mocknet and we verified that it behaves as expected.

My plan is to merge this PR as it is, but I'll have a follow up PR where I refactor the ClientActor code to move the logic into client and also add tests.

chain/chain-primitives/src/error.rs

@@ -344,4 +344,6 @@ pub enum BlockKnownError {
     KnownInStore,
     #[error("already known in blocks in processing")]
     KnownInProcessing,
+    #[error("already known in invalid blocks")]
+    KnownAsInvalid,
 }

chain/chain/src/chain.rs

@@ -75,6 +75,7 @@ use crate::{metrics, DoomslugThresholdMode};
 use actix::Message;
 use crossbeam_channel::{unbounded, Receiver, Sender};
 use delay_detector::DelayDetector;
+use lru::LruCache;
 use near_client_primitives::types::StateSplitApplyingStatus;
 use near_primitives::shard_layout::{
     account_id_to_shard_id, account_id_to_shard_uid, ShardLayout, ShardUId,
@@ -97,6 +98,9 @@ const MAX_ORPHAN_AGE_SECS: u64 = 300;
 // its NUM_ORPHAN_ANCESTORS_CHECK'th ancestor has been accepted
 pub const NUM_ORPHAN_ANCESTORS_CHECK: u64 = 3;
 
+/// The size of the invalid_blocks in-memory pool
+pub const INVALID_CHUNKS_POOL_SIZE: usize = 5000;
+
 // Maximum number of orphans that we can request missing chunks
 // Note that if there are no forks, the maximum number of orphans we would
 // request missing chunks will not exceed NUM_ORPHAN_ANCESTORS_CHECK,
@@ -407,6 +411,9 @@ pub fn check_known(
     if chain.blocks_with_missing_chunks.contains(block_hash) {
         return Ok(Err(BlockKnownError::KnownInMissingChunks));
     }
+    if chain.is_block_invalid(block_hash) {
+        return Ok(Err(BlockKnownError::KnownAsInvalid));
+    }
     check_known_store(chain, block_hash)
 }
 
@@ -438,6 +445,8 @@ pub struct Chain {
     /// Time when head was updated most recently.
     last_time_head_updated: Instant,
 
+    invalid_blocks: LruCache<CryptoHash, ()>,
+
     /// Support for sandbox's patch_state requests.
     ///
     /// Sandbox needs ability to arbitrary modify the state. Blockchains
@@ -512,6 +521,7 @@ impl Chain {
             apply_chunks_sender: sc,
             apply_chunks_receiver: rc,
             last_time_head_updated: Clock::instant(),
+            invalid_blocks: LruCache::new(INVALID_CHUNKS_POOL_SIZE),
             pending_state_patch: Default::default(),
         })
     }
@@ -685,6 +695,7 @@ impl Chain {
             orphans: OrphanBlockPool::new(),
             blocks_with_missing_chunks: MissingChunksPool::new(),
             blocks_in_processing: BlocksInProcessing::new(),
+            invalid_blocks: LruCache::new(INVALID_CHUNKS_POOL_SIZE),
             genesis: genesis.clone(),
             transaction_validity_period: chain_genesis.transaction_validity_period,
             epoch_length: chain_genesis.epoch_length,
@@ -2110,6 +2121,10 @@ impl Chain {
         let new_head =
             match self.postprocess_block_only(me, &block, block_preprocess_info, apply_results) {
                 Err(err) => {
+                    if err.is_bad_data() {
+                        self.invalid_blocks.put(*block.hash(), ());
+                        metrics::NUM_INVALID_BLOCKS.inc();
+                    }
                     self.blocks_delay_tracker.mark_block_errored(&block_hash, err.to_string());
                     return Err(err);
                 }
@@ -3644,12 +3659,9 @@ impl Chain {
                                     &transaction.transaction.block_hash,
                                     transaction_validity_period,
                                 )
-                                .map_err(|_|
-
-                                    Error::from(Error::InvalidTransactions))?;
-                                }
+                                .map_err(|_| Error::from(Error::InvalidTransactions))?;
                         }
-                    ;
+                    };
 
                     let chunk_inner = chunk.cloned_header().take_inner();
                     let gas_limit = chunk_inner.gas_limit();
@@ -3700,7 +3712,6 @@ impl Chain {
                             true,
                             is_first_block_with_chunk_of_version,
                             state_patch,
-
                         ) {
                             Ok(apply_result) => {
                                 let apply_split_result_or_state_changes =
@@ -3761,7 +3772,6 @@ impl Chain {
                             false,
                             false,
                             state_patch,
-
                         ) {
                             Ok(apply_result) => {
                                 let apply_split_result_or_state_changes =
@@ -4281,6 +4291,11 @@ impl Chain {
         self.blocks_in_processing.contains(hash)
     }
 
+    #[inline]
+    pub fn is_block_invalid(&self, hash: &CryptoHash) -> bool {
+        self.invalid_blocks.contains(hash)
+    }
+
     /// Check if can sync with sync_hash
     pub fn check_sync_hash_validity(&self, sync_hash: &CryptoHash) -> Result<bool, Error> {
         let head = self.head()?;

chain/chain/src/metrics.rs

@@ -105,3 +105,6 @@ pub static BLOCKS_DROPPED_BECAUSE_POOL_IS_FULL: Lazy<IntGauge> = Lazy::new(|| {
     )
     .unwrap()
 });
+pub static NUM_INVALID_BLOCKS: Lazy<IntGauge> = Lazy::new(|| {
+    try_create_int_gauge("near_num_invalid_blocks", "Number of invalid blocks").unwrap()
+});

chain/client/src/client_actor.rs

@@ -1466,9 +1466,16 @@ impl ClientActor {
         let head = self.client.chain.head()?;
         let mut is_syncing = self.client.sync_status.is_syncing();
 
-        let peer_info = if let Some(peer_info) =
-            self.network_info.highest_height_peers.choose(&mut thread_rng())
-        {
+        // Only consider peers whose latest block is not invalid blocks
+        let eligible_peers: Vec<_> = self
+            .network_info
+            .highest_height_peers
+            .iter()
+            .filter(|p| !self.client.chain.is_block_invalid(&p.highest_block_hash))
+            .collect();
+        metrics::PEERS_WITH_INVALID_HASH
+            .set(self.network_info.highest_height_peers.len() as i64 - eligible_peers.len() as i64);
+        let peer_info = if let Some(peer_info) = eligible_peers.choose(&mut thread_rng()) {
             peer_info
         } else {
             if !self.client.config.skip_sync_wait {

chain/client/src/metrics.rs

@@ -142,6 +142,11 @@ pub(crate) static PROTOCOL_UPGRADE_BLOCK_HEIGHT: Lazy<IntGauge> = Lazy::new(|| {
     .unwrap()
 });
 
+pub(crate) static PEERS_WITH_INVALID_HASH: Lazy<IntGauge> = Lazy::new(|| {
+    try_create_int_gauge("near_peers_with_invalid_hash", "Number of peers that are on invalid hash")
+        .unwrap()
+});
+
 pub(crate) static CHUNK_SKIPPED_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
     try_create_int_counter_vec(
         "near_chunk_skipped_total",