Improving logs for transactions sync and disable re-broadcasting while syncing

sync/src/chain.rs

@@ -1447,7 +1447,7 @@ impl ChainSync {
 		}
 
 		let mut item_count = r.item_count();
-		trace!(target: "sync", "{} -> Transactions ({} entries)", peer_id, item_count);
+		trace!(target: "sync", "{:02} -> Transactions ({} entries)", peer_id, item_count);
 		item_count = min(item_count, MAX_TX_TO_IMPORT);
 		let mut transactions = Vec::with_capacity(item_count);
 		for i in 0 .. item_count {
@@ -1987,7 +1987,7 @@ impl ChainSync {
 							stats.propagated(*hash, id, block_number);
 						}
 						peer_info.last_sent_transactions = all_transactions_hashes.clone();
-						return Some((*peer_id, all_transactions_rlp.clone()));
+						return Some((*peer_id, all_transactions_hashes.len(), all_transactions_rlp.clone()));
 					}
 
 					// Get hashes of all transactions to send to this peer
@@ -2008,21 +2008,23 @@ impl ChainSync {
 					}
 
 					peer_info.last_sent_transactions = all_transactions_hashes.clone();
-					Some((*peer_id, packet.out()))
+					Some((*peer_id, to_send.len(), packet.out()))
 				})
 				.collect::<Vec<_>>()
 		};
 
 		// Send RLPs
-		let sent = lucky_peers.len();
-		if sent > 0 {
-			for (peer_id, rlp) in lucky_peers {
+		let peers = lucky_peers.len();
+		if peers > 0 {
+			let mut max_sent = 0;
+			for (peer_id, sent, rlp) in lucky_peers {
 				self.send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp);
+				trace!(target: "sync", "{:02} <- Transactions ({} entries)", peer_id, sent);
+				max_sent = max(max_sent, sent);
 			}
-
-			trace!(target: "sync", "Sent up to {} transactions to {} peers.", transactions.len(), sent);
+			debug!(target: "sync", "Sent up to {} transactions to {} peers.", max_sent, peers);
 		}
-		sent
+		peers
 	}
 
 	fn propagate_latest_blocks(&mut self, io: &mut SyncIo, sealed: &[H256]) {
@@ -2042,7 +2044,6 @@ impl ChainSync {
 				trace!(target: "sync", "Sent sealed block to all peers");
 			};
 		}
-		self.propagate_new_transactions(io);
 		self.last_sent_block_number = chain_info.best_block_number;
 	}
 
@@ -2070,7 +2071,9 @@ impl ChainSync {
 	/// called when block is imported to chain - propagates the blocks and updates transactions sent to peers
 	pub fn chain_new_blocks(&mut self, io: &mut SyncIo, _imported: &[H256], invalid: &[H256], enacted: &[H256], _retracted: &[H256], sealed: &[H256], proposed: &[Bytes]) {
 		let queue_info = io.chain().queue_info();
-		if !self.status().is_syncing(queue_info) || !sealed.is_empty() {
+		let is_syncing = self.status().is_syncing(queue_info);
+
+		if !is_syncing || !sealed.is_empty() {
 			trace!(target: "sync", "Propagating blocks, state={:?}", self.state);
 			self.propagate_latest_blocks(io, sealed);
 			self.propagate_proposed_blocks(io, proposed);
@@ -2080,7 +2083,7 @@ impl ChainSync {
 			self.restart(io);
 		}
 
-		if !enacted.is_empty() {
+		if !is_syncing && !enacted.is_empty() {
 			// Select random peers to re-broadcast transactions to.
 			let mut random = random::new();
 			let len = self.peers.len();
@@ -2531,7 +2534,7 @@ mod tests {
 	}
 
 	#[test]
-	fn propagates_new_transactions_after_new_block() {
+	fn does_not_propagate_new_transactions_after_new_block() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
 		client.insert_transaction_to_queue();
@@ -2541,16 +2544,16 @@ mod tests {
 		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		let peer_count = sync.propagate_new_transactions(&mut io);
 		io.chain.insert_transaction_to_queue();
-		// New block import should trigger propagation.
+		// New block import should not trigger propagation.
+		// (we only propagate on timeout)
 		sync.chain_new_blocks(&mut io, &[], &[], &[], &[], &[], &[]);
 
 		// 2 message should be send
-		assert_eq!(2, io.packets.len());
+		assert_eq!(1, io.packets.len());
 		// 1 peer should receive the message
 		assert_eq!(1, peer_count);
 		// TRANSACTIONS_PACKET
 		assert_eq!(0x02, io.packets[0].packet_id);
-		assert_eq!(0x02, io.packets[1].packet_id);
 	}
 
 	#[test]