Block import notifications

substrate/client/Cargo.toml

@@ -9,6 +9,7 @@ log = "0.3"
 parking_lot = "0.4"
 triehash = "0.1"
 hex-literal = "0.1"
+multiqueue = "0.3"
 ed25519 = { path = "../ed25519" }
 substrate-bft = { path = "../bft" }
 substrate-codec = { path = "../codec" }

substrate/client/src/client.rs

@@ -16,6 +16,8 @@
 
 //! Substrate Client
 
+use multiqueue;
+use parking_lot::Mutex;
 use primitives::{self, block, AuthorityId};
 use primitives::block::Id as BlockId;
 use primitives::storage::{StorageKey, StorageData};
@@ -28,10 +30,11 @@ use blockchain::{self, Info as ChainInfo, Backend as ChainBackend};
 use {error, in_mem, block_builder, runtime_io, bft};
 
 /// Polkadot Client
-#[derive(Debug)]
 pub struct Client<B, E> where B: backend::Backend {
 	backend: B,
 	executor: E,
+	import_notification_sink: Mutex<multiqueue::BroadcastFutSender<BlockImportNotification>>,
+	import_notification_stream: Mutex<multiqueue::BroadcastFutReceiver<BlockImportNotification>>,
 }
 
 /// Client info
@@ -82,6 +85,32 @@ pub enum BlockStatus {
 	Unknown,
 }
 
+/// Block data origin.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum BlockOrigin {
+	/// Genesis block built into the client.
+	Genesis,
+	/// Block is part of the initial sync with the network.
+	NetworkInitialSync,
+	/// Block was broadcasted on the network.
+	NetworkBroadcast,
+	/// Block that was collated by this node.
+	Own,
+	/// Block was imported from a file.
+	File,
+}
+
+/// Summary of an imported block
+#[derive(Clone, Debug)]
+pub struct BlockImportNotification {
+	/// Imported block origin.
+	pub origin: BlockOrigin,
+	/// Imported block header.
+	pub header: block::Header,
+	/// Is this the new best block.
+	pub is_new_best: bool,
+}
+
 /// A header paired with a justification which has already been checked.
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub struct JustifiedHeader {
@@ -122,6 +151,7 @@ impl<B, E> Client<B, E> where
 		where
 			F: FnOnce() -> (block::Header, Vec<(Vec<u8>, Vec<u8>)>)
 	{
+		let (sink, stream) = multiqueue::broadcast_fut_queue(64);
 		if backend.blockchain().header(BlockId::Number(0))?.is_none() {
 			trace!("Empty database, writing genesis block");
 			let (genesis_header, genesis_store) = build_genesis();
@@ -133,9 +163,16 @@ impl<B, E> Client<B, E> where
 		Ok(Client {
 			backend,
 			executor,
+			import_notification_sink: Mutex::new(sink),
+			import_notification_stream: Mutex::new(stream),
 		})
 	}
 
+	/// Get block import event stream.
+	pub fn import_notification_stream(&self) -> multiqueue::BroadcastFutReceiver<BlockImportNotification> {
+		self.import_notification_stream.lock().add_stream()
+	}
+
 	/// Get a reference to the state at a given block.
 	pub fn state_at(&self, block: &BlockId) -> error::Result<B::State> {
 		self.backend.state_at(*block)
@@ -236,6 +273,7 @@ impl<B, E> Client<B, E> where
 	/// Queue a block for import.
 	pub fn import_block(
 		&self,
+		origin: BlockOrigin,
 		header: JustifiedHeader,
 		body: Option<block::Body>,
 	) -> error::Result<ImportResult> {
@@ -261,9 +299,21 @@ impl<B, E> Client<B, E> where
 
 		let is_new_best = header.number == self.backend.blockchain().info()?.best_number + 1;
 		trace!("Imported {}, (#{}), best={}", block::HeaderHash::from(header.blake2_256()), header.number, is_new_best);
-		transaction.set_block_data(header, body, Some(justification.uncheck().into()), is_new_best)?;
+		transaction.set_block_data(header.clone(), body, Some(justification.uncheck().into()), is_new_best)?;
 		transaction.set_storage(overlay.drain())?;
 		self.backend.commit_operation(transaction)?;
+
+		if origin == BlockOrigin::NetworkBroadcast || origin == BlockOrigin::Own {
+			let notification = BlockImportNotification {
+				origin: origin,
+				header: header,
+				is_new_best: is_new_best,
+			};
+			if let Err(e) = self.import_notification_sink.lock().try_send(notification) {
+				warn!("Error queueing block import notification: {:?}", e);
+			}
+		}
+
 		Ok(ImportResult::Queued)
 	}
 
@@ -335,7 +385,7 @@ impl<B, E> bft::BlockImport for Client<B, E>
 			justification,
 		};
 
-		let _ = self.import_block(justified_header, Some(block.transactions));
+		let _ = self.import_block(BlockOrigin::Genesis, justified_header, Some(block.transactions));
 	}
 }
 

substrate/client/src/lib.rs

@@ -31,6 +31,7 @@ extern crate ed25519;
 
 extern crate triehash;
 extern crate parking_lot;
+extern crate multiqueue;
 #[cfg(test)] #[macro_use] extern crate hex_literal;
 #[macro_use] extern crate error_chain;
 #[macro_use] extern crate log;
@@ -43,5 +44,5 @@ pub mod genesis;
 pub mod block_builder;
 mod client;
 
-pub use client::{Client, ClientInfo, CallResult, ImportResult, BlockStatus, new_in_mem};
+pub use client::{Client, ClientInfo, CallResult, ImportResult, BlockStatus, BlockOrigin, new_in_mem};
 pub use blockchain::Info as ChainInfo;

substrate/network/src/chain.rs

@@ -16,15 +16,15 @@
 
 //! Blockchain access trait
 
-use client::{self, Client as PolkadotClient, ImportResult, ClientInfo, BlockStatus};
+use client::{self, Client as PolkadotClient, ImportResult, ClientInfo, BlockStatus, BlockOrigin};
 use client::error::Error;
 use state_machine;
 use primitives::block::{self, Id as BlockId};
 use primitives::bft::Justification;
 
 pub trait Client: Send + Sync {
-	/// Given a hash return a header
-	fn import(&self, header: block::Header, justification: Justification, body: Option<block::Body>) -> Result<ImportResult, Error>;
+	/// Import a new block. Parent is supposed to be existing in the blockchain.
+	fn import(&self, is_best: bool, header: block::Header, justification: Justification, body: Option<block::Body>) -> Result<ImportResult, Error>;
 
 	/// Get blockchain info.
 	fn info(&self) -> Result<ClientInfo, Error>;
@@ -50,10 +50,11 @@ impl<B, E> Client for PolkadotClient<B, E> where
 	E: state_machine::CodeExecutor + Send + Sync + 'static,
 	Error: From<<<B as client::backend::Backend>::State as state_machine::backend::Backend>::Error>, {
 
-	fn import(&self, header: block::Header, justification: Justification, body: Option<block::Body>) -> Result<ImportResult, Error> {
+	fn import(&self, is_best: bool, header: block::Header, justification: Justification, body: Option<block::Body>) -> Result<ImportResult, Error> {
 		// TODO: defer justification check.
 		let justified_header = self.check_justification(header, justification.into())?;
-		(self as &PolkadotClient<B, E>).import_block(justified_header, body)
+		let origin = if is_best { BlockOrigin::NetworkBroadcast } else { BlockOrigin::NetworkInitialSync };
+		(self as &PolkadotClient<B, E>).import_block(origin, justified_header, body)
 	}
 
 	fn info(&self) -> Result<ClientInfo, Error> {

substrate/network/src/sync.rs

@@ -84,9 +84,13 @@ impl ChainSync {
 		}
 	}
 
+	fn best_seen_block(&self) -> Option<BlockNumber> {
+		self.peers.values().max_by_key(|p| p.best_number).map(|p| p.best_number)
+	}
+
 	/// Returns sync status
 	pub fn status(&self) -> Status {
-		let best_seen = self.peers.values().max_by_key(|p| p.best_number).map(|p| p.best_number);
+		let best_seen = self.best_seen_block();
 		let state = match &best_seen {
 			&Some(n) if n > self.best_queued_number && n - self.best_queued_number > 5 => SyncState::Downloading,
 			_ => SyncState::Idle,
@@ -97,6 +101,7 @@ impl ChainSync {
 		}
 	}
 
+	/// Handle new connected peer.
 	pub fn new_peer(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId) {
 		if let Some(info) = protocol.peer_info(peer_id) {
 			match (protocol.chain().block_status(&BlockId::Hash(info.best_hash)), info.best_number) {
@@ -211,6 +216,7 @@ impl ChainSync {
 			vec![]
 		};
 
+		let best_seen = self.best_seen_block();
 		// Blocks in the response/drain should be in ascending order.
 		for block in new_blocks {
 			let origin = block.origin;
@@ -220,7 +226,9 @@ impl ChainSync {
 					let number = header.number;
 					let hash = header_hash(&header);
 					let parent = header.parent_hash;
+					let is_best = best_seen.as_ref().map_or(false, |n| number >= *n);
 					let result = protocol.chain().import(
+						is_best,
 						header,
 						justification,
 						block.body

substrate/network/src/test/mod.rs

@@ -19,7 +19,7 @@ mod sync;
 use std::collections::{VecDeque, HashSet, HashMap};
 use std::sync::Arc;
 use parking_lot::RwLock;
-use client::{self, genesis};
+use client::{self, genesis, BlockOrigin};
 use client::block_builder::BlockBuilder;
 use primitives::block::Id as BlockId;
 use primitives;
@@ -185,7 +185,7 @@ impl Peer {
 			trace!("Generating {}, (#{})", primitives::block::HeaderHash::from(block.header.blake2_256()), block.header.number);
 			let justification = Self::justify(&block.header);
 			let justified = self.client.check_justification(block.header, justification).unwrap();
-			self.client.import_block(justified, Some(block.transactions)).unwrap();
+			self.client.import_block(BlockOrigin::File, justified, Some(block.transactions)).unwrap();
 		}
 	}