Fix pubsub new_blocks notifications to include all blocks (#9987)

Fix: new blocks notifications sometimes missing in pubsub RPC

Implement new struct to pass to `new_blocks()` with extra parameter - `has_more_blocks_to_import`, which was previously used to determine whether the notification should be sent. Now it's up to each implementation to decide what to do.

Updated all implementations to behave as before, except `eth_pubsub`, which will send notification even when the queue is not empty.

Update tests.

ethcore/private-tx/src/lib.rs

@@ -69,7 +69,6 @@ pub use error::{Error, ErrorKind};
 
 use std::sync::{Arc, Weak};
 use std::collections::{HashMap, HashSet, BTreeMap};
-use std::time::Duration;
 use ethereum_types::{H128, H256, U256, Address};
 use hash::keccak;
 use rlp::*;
@@ -82,7 +81,7 @@ use ethcore::executed::{Executed};
 use transaction::{SignedTransaction, Transaction, Action, UnverifiedTransaction};
 use ethcore::{contract_address as ethcore_contract_address};
 use ethcore::client::{
-	Client, ChainNotify, ChainRoute, ChainMessageType, ClientIoMessage, BlockId,
+	Client, ChainNotify, NewBlocks, ChainMessageType, ClientIoMessage, BlockId,
 	CallContract, Call, BlockInfo
 };
 use ethcore::account_provider::AccountProvider;
@@ -733,12 +732,11 @@ fn find_account_password(passwords: &Vec<Password>, account_provider: &AccountPr
 }
 
 impl ChainNotify for Provider {
-	fn new_blocks(&self, imported: Vec<H256>, _invalid: Vec<H256>, _route: ChainRoute, _sealed: Vec<H256>, _proposed: Vec<Bytes>, _duration: Duration) {
-		if !imported.is_empty() {
-			trace!(target: "privatetx", "New blocks imported, try to prune the queue");
-			if let Err(err) = self.process_verification_queue() {
-				warn!(target: "privatetx", "Cannot prune private transactions queue. error: {:?}", err);
-			}
+	fn new_blocks(&self, new_blocks: NewBlocks) {
+		if new_blocks.imported.is_empty() || new_blocks.has_more_blocks_to_import { return }
+		trace!(target: "privatetx", "New blocks imported, try to prune the queue");
+		if let Err(err) = self.process_verification_queue() {
+			warn!(target: "privatetx", "Cannot prune private transactions queue. error: {:?}", err);
 		}
 	}
 }

ethcore/src/client/chain_notify.rs

@@ -114,19 +114,51 @@ impl ChainRoute {
 	}
 }
 
+/// Used by `ChainNotify` `new_blocks()`
+pub struct NewBlocks {
+	/// Imported blocks
+	pub imported: Vec<H256>,
+	/// Invalid blocks
+	pub invalid: Vec<H256>,
+	/// Route
+	pub route: ChainRoute,
+	/// Sealed
+	pub sealed: Vec<H256>,
+	/// Block bytes.
+	pub proposed: Vec<Bytes>,
+	/// Duration
+	pub duration: Duration,
+	/// Has more blocks to import
+	pub has_more_blocks_to_import: bool,
+}
+
+impl NewBlocks {
+	/// Constructor
+	pub fn new (
+		imported: Vec<H256>,
+		invalid: Vec<H256>,
+		route: ChainRoute,
+		sealed: Vec<H256>,
+		proposed: Vec<Bytes>,
+		duration: Duration,
+		has_more_blocks_to_import: bool,
+	) -> NewBlocks {
+		NewBlocks {
+			imported,
+			invalid,
+			route,
+			sealed,
+			proposed,
+			duration,
+			has_more_blocks_to_import,
+		}
+	}
+}
+
 /// Represents what has to be handled by actor listening to chain events
 pub trait ChainNotify : Send + Sync {
 	/// fires when chain has new blocks.
-	fn new_blocks(
-		&self,
-		_imported: Vec<H256>,
-		_invalid: Vec<H256>,
-		_route: ChainRoute,
-		_sealed: Vec<H256>,
-		// Block bytes.
-		_proposed: Vec<Bytes>,
-		_duration: Duration,
-	) {
+	fn new_blocks( &self, _new_blocks: NewBlocks) {
 		// does nothing by default
 	}
 

ethcore/src/client/client.rs

@@ -44,7 +44,7 @@ use client::{
 use client::{
 	BlockId, TransactionId, UncleId, TraceId, ClientConfig, BlockChainClient,
 	TraceFilter, CallAnalytics, Mode,
-	ChainNotify, ChainRoute, PruningInfo, ProvingBlockChainClient, EngineInfo, ChainMessageType,
+	ChainNotify, NewBlocks, ChainRoute, PruningInfo, ProvingBlockChainClient, EngineInfo, ChainMessageType,
 	IoClient, BadBlocks,
 };
 use client::bad_blocks;
@@ -268,7 +268,7 @@ impl Importer {
 		}
 
 		let max_blocks_to_import = client.config.max_round_blocks_to_import;
-		let (imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration, is_empty) = {
+		let (imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration, has_more_blocks_to_import) = {
 			let mut imported_blocks = Vec::with_capacity(max_blocks_to_import);
 			let mut invalid_blocks = HashSet::new();
 			let mut proposed_blocks = Vec::with_capacity(max_blocks_to_import);
@@ -322,26 +322,29 @@ impl Importer {
 			if !invalid_blocks.is_empty() {
 				self.block_queue.mark_as_bad(&invalid_blocks);
 			}
-			let is_empty = self.block_queue.mark_as_good(&imported_blocks);
-			(imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, start.elapsed(), is_empty)
+			let has_more_blocks_to_import = !self.block_queue.mark_as_good(&imported_blocks);
+			(imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, start.elapsed(), has_more_blocks_to_import)
 		};
 
 		{
-			if !imported_blocks.is_empty() && is_empty {
+			if !imported_blocks.is_empty() {
 				let route = ChainRoute::from(import_results.as_ref());
 
-				if is_empty {
+				if !has_more_blocks_to_import {
 					self.miner.chain_new_blocks(client, &imported_blocks, &invalid_blocks, route.enacted(), route.retracted(), false);
 				}
 
 				client.notify(|notify| {
 					notify.new_blocks(
-						imported_blocks.clone(),
-						invalid_blocks.clone(),
-						route.clone(),
-						Vec::new(),
-						proposed_blocks.clone(),
-						duration,
+						NewBlocks::new(
+							imported_blocks.clone(),
+							invalid_blocks.clone(),
+							route.clone(),
+							Vec::new(),
+							proposed_blocks.clone(),
+							duration,
+							has_more_blocks_to_import,
+						)
 					);
 				});
 			}
@@ -2342,12 +2345,15 @@ impl ImportSealedBlock for Client {
 		);
 		self.notify(|notify| {
 			notify.new_blocks(
-				vec![hash],
-				vec![],
-				route.clone(),
-				vec![hash],
-				vec![],
-				start.elapsed(),
+				NewBlocks::new(
+					vec![hash],
+					vec![],
+					route.clone(),
+					vec![hash],
+					vec![],
+					start.elapsed(),
+					false
+				)
 			);
 		});
 		self.db.read().key_value().flush().expect("DB flush failed.");
@@ -2360,12 +2366,15 @@ impl BroadcastProposalBlock for Client {
 		const DURATION_ZERO: Duration = Duration::from_millis(0);
 		self.notify(|notify| {
 			notify.new_blocks(
-				vec![],
-				vec![],
-				ChainRoute::default(),
-				vec![],
-				vec![block.rlp_bytes()],
-				DURATION_ZERO,
+				NewBlocks::new(
+					vec![],
+					vec![],
+					ChainRoute::default(),
+					vec![],
+					vec![block.rlp_bytes()],
+					DURATION_ZERO,
+					false
+				)
 			);
 		});
 	}

ethcore/src/client/mod.rs

@@ -34,7 +34,7 @@ pub use self::evm_test_client::{EvmTestClient, EvmTestError, TransactResult};
 pub use self::io_message::ClientIoMessage;
 #[cfg(any(test, feature = "test-helpers"))]
 pub use self::test_client::{TestBlockChainClient, EachBlockWith};
-pub use self::chain_notify::{ChainNotify, ChainRoute, ChainRouteType, ChainMessageType};
+pub use self::chain_notify::{ChainNotify, NewBlocks, ChainRoute, ChainRouteType, ChainMessageType};
 pub use self::traits::{
     Nonce, Balance, ChainInfo, BlockInfo, ReopenBlock, PrepareOpenBlock, CallContract, TransactionInfo, RegistryInfo, ScheduleInfo, ImportSealedBlock, BroadcastProposalBlock, ImportBlock,
     StateOrBlock, StateClient, Call, EngineInfo, AccountData, BlockChain, BlockProducer, SealedBlockImporter, BadBlocks,

ethcore/src/snapshot/watcher.rs

@@ -17,14 +17,13 @@
 //! Watcher for snapshot-related chain events.
 
 use parking_lot::Mutex;
-use client::{BlockInfo, Client, ChainNotify, ChainRoute, ClientIoMessage};
+use client::{BlockInfo, Client, ChainNotify, NewBlocks, ClientIoMessage};
 use ids::BlockId;
 
 use io::IoChannel;
 use ethereum_types::H256;
-use bytes::Bytes;
 
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 // helper trait for transforming hashes to numbers and checking if syncing.
 trait Oracle: Send + Sync {
@@ -99,20 +98,12 @@ impl Watcher {
 }
 
 impl ChainNotify for Watcher {
-	fn new_blocks(
-		&self,
-		imported: Vec<H256>,
-		_: Vec<H256>,
-		_: ChainRoute,
-		_: Vec<H256>,
-		_: Vec<Bytes>,
-		_duration: Duration)
-	{
-		if self.oracle.is_major_importing() { return }
+	fn new_blocks(&self, new_blocks: NewBlocks) {
+		if self.oracle.is_major_importing() || new_blocks.has_more_blocks_to_import { return }
 
-		trace!(target: "snapshot_watcher", "{} imported", imported.len());
+		trace!(target: "snapshot_watcher", "{} imported", new_blocks.imported.len());
 
-		let highest = imported.into_iter()
+		let highest = new_blocks.imported.into_iter()
 			.filter_map(|h| self.oracle.to_number(h))
 			.filter(|&num| num >= self.period + self.history)
 			.map(|num| num - self.history)
@@ -130,7 +121,7 @@ impl ChainNotify for Watcher {
 mod tests {
 	use super::{Broadcast, Oracle, Watcher};
 
-	use client::{ChainNotify, ChainRoute};
+	use client::{ChainNotify, NewBlocks, ChainRoute};
 
 	use ethereum_types::{H256, U256};
 
@@ -170,14 +161,15 @@ mod tests {
 			history: history,
 		};
 
-		watcher.new_blocks(
+		watcher.new_blocks(NewBlocks::new(
 			hashes,
 			vec![],
 			ChainRoute::default(),
 			vec![],
 			vec![],
 			DURATION_ZERO,
-		);
+			false
+		));
 	}
 
 	// helper

ethcore/sync/src/api.rs

@@ -29,7 +29,7 @@ use types::pruning_info::PruningInfo;
 use ethereum_types::{H256, H512, U256};
 use io::{TimerToken};
 use ethcore::ethstore::ethkey::Secret;
-use ethcore::client::{BlockChainClient, ChainNotify, ChainRoute, ChainMessageType};
+use ethcore::client::{BlockChainClient, ChainNotify, NewBlocks, ChainMessageType};
 use ethcore::snapshot::SnapshotService;
 use ethcore::header::BlockNumber;
 use sync_io::NetSyncIo;
@@ -498,27 +498,22 @@ impl ChainNotify for EthSync {
 		}
 	}
 
-	fn new_blocks(&self,
-		imported: Vec<H256>,
-		invalid: Vec<H256>,
-		route: ChainRoute,
-		sealed: Vec<H256>,
-		proposed: Vec<Bytes>,
-		_duration: Duration)
+	fn new_blocks(&self, new_blocks: NewBlocks)
 	{
+		if new_blocks.has_more_blocks_to_import { return }
 		use light::net::Announcement;
 
 		self.network.with_context(self.subprotocol_name, |context| {
 			let mut sync_io = NetSyncIo::new(context, &*self.eth_handler.chain, &*self.eth_handler.snapshot_service,
 				&self.eth_handler.overlay);
 			self.eth_handler.sync.write().chain_new_blocks(
 				&mut sync_io,
-				&imported,
-				&invalid,
-				route.enacted(),
-				route.retracted(),
-				&sealed,
-				&proposed);
+				&new_blocks.imported,
+				&new_blocks.invalid,
+				new_blocks.route.enacted(),
+				new_blocks.route.retracted(),
+				&new_blocks.sealed,
+				&new_blocks.proposed);
 		});
 
 		self.network.with_context(self.light_subprotocol_name, |context| {

ethcore/sync/src/tests/helpers.rs

@@ -16,14 +16,13 @@
 
 use std::collections::{VecDeque, HashSet, HashMap};
 use std::sync::Arc;
-use std::time::Duration;
 use ethereum_types::H256;
 use parking_lot::{RwLock, Mutex};
 use bytes::Bytes;
 use network::{self, PeerId, ProtocolId, PacketId, SessionInfo};
 use tests::snapshot::*;
 use ethcore::client::{TestBlockChainClient, BlockChainClient, Client as EthcoreClient,
-	ClientConfig, ChainNotify, ChainRoute, ChainMessageType, ClientIoMessage};
+	ClientConfig, ChainNotify, NewBlocks, ChainMessageType, ClientIoMessage};
 use ethcore::header::BlockNumber;
 use ethcore::snapshot::SnapshotService;
 use ethcore::spec::Spec;
@@ -535,23 +534,18 @@ impl IoHandler<ClientIoMessage> for TestIoHandler {
 }
 
 impl ChainNotify for EthPeer<EthcoreClient> {
-	fn new_blocks(&self,
-		imported: Vec<H256>,
-		invalid: Vec<H256>,
-		route: ChainRoute,
-		sealed: Vec<H256>,
-		proposed: Vec<Bytes>,
-		_duration: Duration)
+	fn new_blocks(&self, new_blocks: NewBlocks)
 	{
-		let (enacted, retracted) = route.into_enacted_retracted();
+		if new_blocks.has_more_blocks_to_import { return }
+		let (enacted, retracted) = new_blocks.route.into_enacted_retracted();
 
 		self.new_blocks_queue.write().push_back(NewBlockMessage {
-			imported,
-			invalid,
+			imported: new_blocks.imported,
+			invalid: new_blocks.invalid,
 			enacted,
 			retracted,
-			sealed,
-			proposed,
+			sealed: new_blocks.sealed,
+			proposed: new_blocks.proposed,
 		});
 	}