Speeding up / less network for block sync (#1555)

* Fix init --num-shards doesn't work. Bump the parameters for catchup and chunk request retry periods

* Reduce chunk retry in tests and don't re-request block that is already missing chunks orphan

* Fix up some error propagation and return catchup_step_period to 100ms

chain/chain/src/chain.rs

@@ -684,7 +684,7 @@ impl Chain {
                             String::new()
                         },
                     );
-                    Err(ErrorKind::Orphan.into())
+                    Err(e)
                 }
                 ErrorKind::ChunksMissing(missing_chunks) => {
                     let block_hash = block.hash();
@@ -698,10 +698,11 @@ impl Chain {
                         "Process block: missing chunks. Block hash: {:?}. Missing chunks: {:?}",
                         block_hash, missing_chunks,
                     );
-                    Err(ErrorKind::ChunksMissing(missing_chunks).into())
+                    Err(e)
                 }
                 ErrorKind::EpochOutOfBounds => {
                     // Possibly block arrived before we finished processing all of the blocks for epoch before last.
+                    // Or someone is attacking with invalid chain.
                     debug!(target: "chain", "Received block {}/{} ignored, as epoch is unknown", block.header.inner.height, block.hash());
                     Ok(Some(prev_head))
                 }
@@ -715,7 +716,7 @@ impl Chain {
                     );
                     Err(ErrorKind::Unfit(msg.clone()).into())
                 }
-                e => Err(e.into()),
+                _ => Err(e),
             },
         }
     }
@@ -1469,6 +1470,12 @@ impl Chain {
     pub fn is_orphan(&self, hash: &CryptoHash) -> bool {
         self.orphans.contains(hash)
     }
+
+    /// Check if hash is for a known chunk orphan.
+    #[inline]
+    pub fn is_chunk_orphan(&self, hash: &CryptoHash) -> bool {
+        self.blocks_with_missing_chunks.contains(hash)
+    }
 }
 
 /// Chain update helper, contains information that is needed to process block

chain/client/src/client_actor.rs

@@ -716,18 +716,20 @@ impl ClientActor {
 
     fn receive_header(&mut self, header: BlockHeader, peer_info: PeerId) -> NetworkClientResponses {
         let hash = header.hash();
-        debug!(target: "client", "Received block header {} at {} from {}", hash, header.inner.height, peer_info);
+        debug!(target: "client", "{:?} Received block header {} at {} from {}", self.client.block_producer.as_ref().map(|bp| bp.account_id.clone()), hash, header.inner.height, peer_info);
 
         // Process block by chain, if it's valid header ask for the block.
         let result = self.client.process_block_header(&header);
 
         match result {
             Err(ref e) if e.kind() == near_chain::ErrorKind::EpochOutOfBounds => {
                 // Block header is either invalid or arrived too early. We ignore it.
+                debug!(target: "client", "Epoch out of bound for header {}", e);
                 return NetworkClientResponses::NoResponse;
             }
             Err(ref e) if e.is_bad_data() => {
-                return NetworkClientResponses::Ban { ban_reason: ReasonForBan::BadBlockHeader }
+                debug!(target: "client", "Error on receival of header: {}", e);
+                return NetworkClientResponses::Ban { ban_reason: ReasonForBan::BadBlockHeader };
             }
             // Some error that worth surfacing.
             Err(ref e) if e.is_error() => {

chain/client/src/sync.rs

@@ -327,7 +327,9 @@ impl BlockSync {
 
         let hashes_to_request = hashes
             .iter()
-            .filter(|x| !chain.get_block(x).is_ok() && !chain.is_orphan(x))
+            .filter(|x| {
+                !chain.get_block(x).is_ok() && !chain.is_orphan(x) && !chain.is_chunk_orphan(x)
+            })
             .take(block_count)
             .collect::<Vec<_>>();
         if hashes_to_request.len() > 0 {

chain/client/src/types.rs

@@ -167,7 +167,7 @@ impl ClientConfig {
             block_fetch_horizon: 50,
             state_fetch_horizon: 5,
             catchup_step_period: Duration::from_millis(block_prod_time / 2),
-            chunk_request_retry_period: Duration::from_millis(100),
+            chunk_request_retry_period: Duration::from_millis(block_prod_time / 5),
             block_header_fetch_horizon: 50,
             tracked_accounts: vec![],
             tracked_shards: vec![],

near/src/config.rs

@@ -272,7 +272,7 @@ impl NearConfig {
                 state_fetch_horizon: 5,
                 block_header_fetch_horizon: 50,
                 catchup_step_period: Duration::from_millis(100),
-                chunk_request_retry_period: Duration::from_millis(100),
+                chunk_request_retry_period: Duration::from_millis(200),
                 tracked_accounts: config.tracked_accounts,
                 tracked_shards: config.tracked_shards,
             },

near/src/main.rs

@@ -93,7 +93,7 @@ fn main() {
             let account_id = args.value_of("account-id");
             let test_seed = args.value_of("test-seed");
             let num_shards = args
-                .value_of("num_shards")
+                .value_of("num-shards")
                 .map(|s| s.parse().expect("Number of shards must be a number"))
                 .unwrap_or(1);
             let fast = args.is_present("fast");

near/tests/sync_nodes.rs

@@ -15,7 +15,7 @@ use near_client::{ClientActor, GetBlock};
 use near_crypto::{InMemorySigner, KeyType, Signer};
 use near_network::test_utils::{convert_boot_nodes, open_port, WaitOrTimeout};
 use near_network::{NetworkClientMessages, PeerInfo};
-use near_primitives::test_utils::{heavy_test, init_integration_logger};
+use near_primitives::test_utils::{heavy_test, init_integration_logger, init_test_logger};
 use near_primitives::transaction::SignedTransaction;
 use near_primitives::types::{BlockIndex, EpochId};
 use testlib::genesis_block;