Ensure all known BFT messages are imported when starting consensus (#147)

* a little more BFT tracing

* import cached BFT messages into the produced stream

substrate/substrate/bft/src/generic/accumulator.rs

@@ -339,6 +339,7 @@ impl<Candidate, Digest, AuthorityId, Signature> Accumulator<Candidate, Digest, A
 				count.committed += 1;
 
 				if count.committed >= self.threshold {
+					trace!(target: "bft", "observed threshold-commit for round {} with {} commits", self.round_number, count.committed);
 					Some(digest)
 				} else {
 					None

substrate/substrate/bft/src/lib.rs

@@ -348,6 +348,7 @@ impl<P, I> BftService<P, I>
 
 		let n = authorities.len();
 		let max_faulty = max_faulty_of(n);
+		trace!(target: "bft", "max_faulty_of({})={}", n, max_faulty);
 
 		let local_id = self.key.public().0;
 

substrate/substrate/network/src/consensus.rs

@@ -46,7 +46,7 @@ pub struct Consensus {
 	peers: HashMap<PeerId, PeerConsensus>,
 	our_candidate: Option<(Hash, Vec<u8>)>,
 	statement_sink: Option<mpsc::UnboundedSender<message::Statement>>,
-	bft_message_sink: Option<mpsc::UnboundedSender<message::LocalizedBftMessage>>,
+	bft_message_sink: Option<(mpsc::UnboundedSender<message::LocalizedBftMessage>, Hash)>,
 	messages: HashMap<Hash, (Instant, message::Message)>,
 }
 
@@ -143,26 +143,41 @@ impl Consensus {
 		if let Some(ref mut peer) = self.peers.get_mut(&peer_id) {
 			peer.known_messages.insert(hash);
 			// TODO: validate signature?
-			if let Some(sink) = self.bft_message_sink.take() {
-				if let Err(e) = sink.unbounded_send(message.clone()) {
-					trace!(target:"sync", "Error broadcasting BFT message notification: {:?}", e);
-				} else {
-					self.bft_message_sink = Some(sink);
+			if let Some((sink, parent_hash)) = self.bft_message_sink.take() {
+				if message.parent_hash == parent_hash {
+					if let Err(e) = sink.unbounded_send(message.clone()) {
+						trace!(target:"sync", "Error broadcasting BFT message notification: {:?}", e);
+					} else {
+						self.bft_message_sink = Some((sink, parent_hash));
+					}
 				}
 			}
 		} else {
 			trace!(target:"sync", "Ignored BFT statement from unregistered peer {}", peer_id);
 			return;
 		}
+
 		let message = Message::BftMessage(message);
 		self.register_message(hash.clone(), message.clone());
 		// Propagate to other peers.
 		self.propagate(io, protocol, message, hash);
 	}
 
-	pub fn bft_messages(&mut self) -> mpsc::UnboundedReceiver<message::LocalizedBftMessage>{
+	pub fn bft_messages(&mut self, parent_hash: Hash) -> mpsc::UnboundedReceiver<message::LocalizedBftMessage>{
 		let (sink, stream) = mpsc::unbounded();
-		self.bft_message_sink = Some(sink);
+
+		for (_, message) in self.messages.iter() {
+			let bft_message = match *message {
+				(_, Message::BftMessage(ref msg)) => msg,
+				_ => continue,
+			};
+
+			if bft_message.parent_hash == parent_hash {
+				sink.unbounded_send(bft_message.clone()).expect("receiving end known to be open; qed");
+			}
+		}
+
+		self.bft_message_sink = Some((sink, parent_hash));
 		stream
 	}
 

substrate/substrate/network/src/protocol.rs

@@ -317,8 +317,8 @@ impl Protocol {
 	}
 
 	/// See `ConsensusService` trait.
-	pub fn bft_messages(&self) -> BftMessageStream {
-		self.consensus.lock().bft_messages()
+	pub fn bft_messages(&self, parent_hash: Hash) -> BftMessageStream {
+		self.consensus.lock().bft_messages(parent_hash)
 	}
 
 	/// See `ConsensusService` trait.

substrate/substrate/network/src/service.rs

@@ -91,8 +91,9 @@ pub trait ConsensusService: Send + Sync {
 	/// Pass `None` to clear the candidate.
 	fn set_local_candidate(&self, candidate: Option<(Hash, Vec<u8>)>);
 
-	/// Get BFT message stream.
-	fn bft_messages(&self) -> BftMessageStream;
+	/// Get BFT message stream for messages corresponding to consensus on given
+	/// parent hash.
+	fn bft_messages(&self, parent_hash: Hash) -> BftMessageStream;
 	/// Send out a BFT message.
 	fn send_bft_message(&self, message: LocalizedBftMessage);
 }
@@ -254,8 +255,8 @@ impl ConsensusService for Service {
 		self.handler.protocol.set_local_candidate(candidate)
 	}
 
-	fn bft_messages(&self) -> BftMessageStream {
-		self.handler.protocol.bft_messages()
+	fn bft_messages(&self, parent_hash: Hash) -> BftMessageStream {
+		self.handler.protocol.bft_messages(parent_hash)
 	}
 
 	fn send_bft_message(&self, message: LocalizedBftMessage) {

substrate/substrate/network/src/test/consensus.rs

@@ -0,0 +1,49 @@
+// Copyright 2017 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use super::*;
+use message::*;
+use futures::Stream;
+
+#[test]
+fn bft_messages_include_those_sent_before_asking_for_stream() {
+	let mut config = ::config::ProtocolConfig::default();
+	config.roles = ::service::Role::VALIDATOR | ::service::Role::FULL;
+
+	let mut net = TestNet::new_with_config(2, config);
+	net.sync(); // necessary for handshaking
+
+	let peer = net.peer(0);
+	let mut io = TestIo::new(&peer.queue, None);
+	let bft_message = BftMessage::Consensus(SignedConsensusMessage::Vote(SignedConsensusVote {
+		vote: ConsensusVote::AdvanceRound(0),
+		sender: [0; 32],
+		signature: Default::default(),
+	}));
+
+	let localized = LocalizedBftMessage {
+		message: bft_message,
+		parent_hash: [1; 32].into(),
+	};
+
+
+	let as_bytes = ::serde_json::to_vec(&Message::BftMessage(localized.clone())).unwrap();
+	peer.sync.handle_packet(&mut io, 1, &as_bytes[..]);
+
+	let stream = peer.sync.bft_messages([1; 32].into());
+
+	assert_eq!(stream.wait().next(), Some(Ok(localized)));
+}

substrate/substrate/network/src/test/mod.rs

@@ -14,6 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
+mod consensus;
 mod sync;
 
 use std::collections::{VecDeque, HashSet, HashMap};