TemporaryFix: Do not demote for empty chunk response. (#1431)

* Do not demote for empty chunk response. And fix some logs. * More. * nit. * more. * nit. * nit. * More log change. * resolve comments.
Conflux-Chain · May 14, 2020 · 80e0909 · 80e0909
1 parent beb293a
commit 80e0909
Show file tree

Hide file tree

Showing 9 changed files with 23 additions and 15 deletions.
diff --git a/core/src/consensus/consensus_inner/consensus_executor.rs b/core/src/consensus/consensus_inner/consensus_executor.rs
@@ -1083,7 +1083,7 @@ impl ConsensusExecutionHandler {
 
                         env.accumulated_gas_used += executed.gas_used;
                         gas_fee = executed.fee;
-                        warn!(
+                        debug!(
                             "tx execution error: transaction={:?}, err={:?}",
                             transaction, error
                         );

diff --git a/core/src/consensus/mod.rs b/core/src/consensus/mod.rs
@@ -904,7 +904,7 @@ impl ConsensusGraph {
         let state = State::new(
             state_db,
             Default::default(), /* vm */
-            0, /* block_number */
+            0,                  /* block_number */
         );
         let gas_cost = gas_limit.full_mul(gas_price);
         let mut gas_sponsored = false;

diff --git a/core/src/sync/error.rs b/core/src/sync/error.rs
@@ -86,6 +86,13 @@ error_chain! {
             display("invalid snapshot chunk: {}", reason),
         }
 
+        // FIXME: This works as a compatible fix when the snapshot provider cannot serve the chunk.
+        // We should add another reply like `UnsupportedSnapshot` and remove this.
+        EmptySnapshotChunk {
+            description("empty snapshot chunk")
+            display("Receive an empty snapshot chunk response, retry later")
+        }
+
         AlreadyThrottled(msg_name: &'static str) {
             description("packet already throttled"),
             display("packet already throttled: {:?}", msg_name),

diff --git a/core/src/sync/message/get_blocks_response.rs b/core/src/sync/message/get_blocks_response.rs
@@ -40,7 +40,7 @@ impl Handleable for GetBlocksResponse {
 
         // TODO Check block size in advance to avoid attacks causing OOM.
         if ctx.manager.is_block_queue_full() {
-            warn!("recover_public_queue is full, discard GetBlocksResponse");
+            debug!("recover_public_queue is full, discard GetBlocksResponse");
             return Ok(());
         }
 

diff --git a/core/src/sync/state/snapshot_chunk_response.rs b/core/src/sync/state/snapshot_chunk_response.rs
@@ -7,7 +7,7 @@ use crate::{
     sync::{
         message::{msgid, Context, Handleable},
         state::{storage::Chunk, SnapshotChunkRequest},
-        Error, ErrorKind,
+        Error,
     },
 };
 use rlp::Encodable;
@@ -43,7 +43,7 @@ impl Handleable for SnapshotChunkResponse {
             ctx.manager
                 .request_manager
                 .resend_request_to_another_peer(ctx.io, &message);
-            bail!(ErrorKind::InvalidSnapshotChunk(e.description().into()));
+            return Err(e);
         }
 
         ctx.manager.state_sync.handle_snapshot_chunk_response(

diff --git a/core/src/sync/state/snapshot_chunk_sync.rs b/core/src/sync/state/snapshot_chunk_sync.rs
@@ -521,7 +521,7 @@ impl SnapshotChunkSync {
         // 1. received a out-of-date snapshot chunk, e.g. new era started.
         // 2. Duplicated chunks received, and it has been processed.
         if inner.downloading_chunks.remove(&chunk_key).is_none() {
-            info!("Snapshot chunk received, but not in downloading queue");
+            info!("Snapshot chunk received, but not in downloading queue, progess is {:?}", *inner);
             // FIXME Handle out-of-date chunks
             inner
                 .sync_candidate_manager
@@ -551,7 +551,7 @@ impl SnapshotChunkSync {
             )?;
             inner.status = Status::Completed;
         }
-        debug!("sync state progress: {:?}", *inner);
+        info!("sync state progress: {:?}", *inner);
         Ok(())
     }
 

diff --git a/core/src/sync/state/storage.rs b/core/src/sync/state/storage.rs
@@ -352,9 +352,7 @@ impl Chunk {
             // maximal number of snapshots and cannot give a
             // response temporarily, we should differentiate this
             // from dishonest behaviors.
-            return Err(
-                ErrorKind::InvalidSnapshotChunk("empty chunk".into()).into()
-            );
+            return Err(ErrorKind::EmptySnapshotChunk.into());
         }
         if self.keys.len() != self.values.len() {
             return Err(ErrorKind::InvalidSnapshotChunk(

diff --git a/core/src/sync/synchronization_protocol_handler.rs b/core/src/sync/synchronization_protocol_handler.rs
@@ -10,7 +10,7 @@ use crate::{
     block_data_manager::BlockStatus,
     light_protocol::Provider as LightProvider,
     message::{decode_msg, Message, MsgId},
-    parameters::sync::*,
+    parameters::{block::MAX_BLOCK_SIZE_IN_BYTES, sync::*},
     rand::Rng,
     sync::{
         message::{
@@ -114,7 +114,9 @@ impl<T: TaskSize> AsyncTaskQueue<T> {
             inner: RwLock::new(AsyncTaskQueueInner {
                 tasks: VecDeque::new(),
                 size: 0,
-                moving_average: 1000.0, // Set to 1KB as initial size.
+                // Set to max value at start to avoid sending too many requests
+                // at the start.
+                moving_average: MAX_BLOCK_SIZE_IN_BYTES as f64,
             }),
             work_type: work_type as HandlerWorkType,
             max_capacity,
@@ -488,7 +490,7 @@ impl SynchronizationProtocolHandler {
                 if !self.syn.handshaking_peers.read().contains_key(peer)
                     || msg_id != msgid::STATUS
                 {
-                    warn!("Message from unknown peer {:?}", msg_id);
+                    debug!("Message from unknown peer {:?}", msg_id);
                     return Ok(());
                 }
             } else {
@@ -561,6 +563,7 @@ impl SynchronizationProtocolHandler {
             ErrorKind::InvalidSnapshotChunk(_) => {
                 op = Some(UpdateNodeOperation::Demotion)
             }
+            ErrorKind::EmptySnapshotChunk => disconnect = false,
             ErrorKind::AlreadyThrottled(_) => {
                 op = Some(UpdateNodeOperation::Remove)
             }

diff --git a/network/src/service.rs b/network/src/service.rs
@@ -1192,7 +1192,7 @@ impl NetworkServiceInner {
                 }
                 deregister = remote || sess.done();
                 failure_id = sess.id().cloned();
-                debug!(
+                info!(
                     "kill connection, deregister = {}, reason = {:?}, session = {:?}, op = {:?}",
                     deregister, reason, *sess, op
                 );
@@ -1267,7 +1267,7 @@ impl NetworkServiceInner {
             deregister = remote || sess.done();
             token = sess.token();
             assert_eq!(sess.id().unwrap().clone(), node_id.clone());
-            debug!(
+            info!(
                 "kill connection, deregister = {}, reason = {:?}, session = {:?}, op = {:?}",
                 deregister, reason, *sess, op
             );