diff --git a/core/finality-grandpa/src/communication.rs b/core/finality-grandpa/src/communication.rs
deleted file mode 100644
index f498b51460ac..000000000000
--- a/core/finality-grandpa/src/communication.rs
+++ /dev/null
@@ -1,543 +0,0 @@
-// Copyright 2017-2019 Parity Technologies (UK) Ltd.
-// This file is part of Substrate.
-
-// Substrate 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.
-
-// Substrate 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 Substrate. If not, see .
-
-//! Incoming message streams that verify signatures, and outgoing message streams
-//! that sign or re-shape.
-
-use std::collections::HashMap;
-use std::sync::Arc;
-
-use grandpa::VoterSet;
-use grandpa::Message::{Prevote, Precommit};
-use futures::prelude::*;
-use futures::sync::mpsc;
-use log::{debug, trace};
-use parity_codec::{Encode, Decode};
-use substrate_primitives::{ed25519, Pair};
-use substrate_telemetry::{telemetry, CONSENSUS_INFO};
-use runtime_primitives::traits::Block as BlockT;
-use tokio::timer::Interval;
-use crate::{Error, Network, Message, SignedMessage, Commit,
- CompactCommit, GossipMessage, FullCommitMessage, VoteOrPrecommitMessage};
-use ed25519::{Public as AuthorityId, Signature as AuthoritySignature};
-
-fn localized_payload(round: u64, set_id: u64, message: &E) -> Vec {
- (message, round, set_id).encode()
-}
-
-#[derive(Clone, Copy, Eq, PartialEq, PartialOrd, Ord)]
-struct Round(u64);
-#[derive(Clone, Copy, Eq, PartialEq, PartialOrd, Ord)]
-struct SetId(u64);
-
-enum Broadcast {
- // round, set id, encoded commit.
- Commit(Round, SetId, Vec),
- // round, set id, encoded signed message.
- Message(Round, SetId, Vec),
- // round, set id, announcement of block hash that should be downloaded
- Announcement(Round, SetId, Block::Hash),
- // round, set id being dropped.
- DropRound(Round, SetId),
- // set_id being dropped.
- DropSet(SetId),
-}
-
-impl Broadcast {
- fn set_id(&self) -> SetId {
- match *self {
- Broadcast::Commit(_, s, _) => s,
- Broadcast::Message(_, s, _) => s,
- Broadcast::Announcement(_, s, _) => s,
- Broadcast::DropRound(_, s) => s,
- Broadcast::DropSet(s) => s,
- }
- }
-}
-
-/// Produces a future that should be run in the background and proxies
-/// and rebroadcasts messages.
-pub(crate) fn rebroadcasting_network>(network: N) -> (BroadcastWorker, BroadcastHandle) {
- use std::time::Duration;
- const REBROADCAST_PERIOD: Duration = Duration::from_secs(60);
-
- let (tx, rx) = mpsc::unbounded();
-
- (
- BroadcastWorker {
- interval: Interval::new_interval(REBROADCAST_PERIOD),
- set_id: SetId(0), // will be overwritten on first item to broadcast.
- last_commit: None,
- round_messages: (Round(0), Vec::new()),
- announcements: HashMap::new(),
- network: network.clone(),
- incoming_broadcast: rx,
- },
- BroadcastHandle {
- relay: tx,
- network,
- },
- )
-}
-
-// A worker which broadcasts messages to the background, potentially
-// rebroadcasting.
-#[must_use = "network rebroadcast future must be driven to completion"]
-pub(crate) struct BroadcastWorker> {
- interval: Interval,
- set_id: SetId,
- last_commit: Option<(Round, Vec)>,
- round_messages: (Round, Vec>),
- announcements: HashMap,
- network: N,
- incoming_broadcast: mpsc::UnboundedReceiver>,
-}
-
-/// A handle used by communication work to broadcast to network.
-#[derive(Clone)]
-pub(crate) struct BroadcastHandle {
- relay: mpsc::UnboundedSender>,
- network: N,
-}
-
-impl> Future for BroadcastWorker {
- type Item = ();
- type Error = Error;
-
- fn poll(&mut self) -> Poll<(), Error> {
- {
- let mut rebroadcast = false;
- loop {
- match self.interval.poll().map_err(Error::Timer)? {
- Async::NotReady => break,
- Async::Ready(_) => { rebroadcast = true; }
- }
- }
-
- if rebroadcast {
- let SetId(set_id) = self.set_id;
- if let Some((Round(c_round), ref c_commit)) = self.last_commit {
- self.network.send_commit(c_round, set_id, c_commit.clone(), true);
- }
-
- let Round(round) = self.round_messages.0;
- for message in self.round_messages.1.iter().cloned() {
- self.network.send_message(round, set_id, message, true);
- }
-
- for (&announce_hash, &Round(round)) in &self.announcements {
- self.network.announce(round, set_id, announce_hash);
- }
- }
- }
-
- loop {
- match self.incoming_broadcast.poll().expect("UnboundedReceiver does not yield errors; qed") {
- Async::NotReady => return Ok(Async::NotReady),
- Async::Ready(None) => return Err(Error::Network(
- "all broadcast handles dropped, connection to network severed".into()
- )),
- Async::Ready(Some(item)) => {
- if item.set_id() > self.set_id {
- self.set_id = item.set_id();
- self.last_commit = None;
- self.round_messages = (Round(0), Vec::new());
- self.announcements.clear();
- }
-
- match item {
- Broadcast::Commit(round, set_id, commit) => {
- if self.set_id == set_id {
- if round >= self.last_commit.as_ref()
- .map_or(Round(0), |&(r, _)| r)
- {
- self.last_commit = Some((round, commit.clone()));
- }
- }
-
- // always send out to network.
- self.network.send_commit(round.0, self.set_id.0, commit, false);
- }
- Broadcast::Message(round, set_id, message) => {
- if self.set_id == set_id {
- if round > self.round_messages.0 {
- self.round_messages = (round, vec![message.clone()]);
- } else if round == self.round_messages.0 {
- self.round_messages.1.push(message.clone());
- };
-
- // ignore messages from earlier rounds.
- }
-
- // always send out to network.
- self.network.send_message(round.0, set_id.0, message, false);
- }
- Broadcast::Announcement(round, set_id, hash) => {
- if self.set_id == set_id {
- self.announcements.insert(hash, round);
- }
-
- // always send out.
- self.network.announce(round.0, set_id.0, hash);
- }
- Broadcast::DropRound(round, set_id) => {
- // stop making announcements for any dead rounds.
- self.announcements.retain(|_, &mut r| r > round);
- self.network.drop_round_messages(round.0, set_id.0);
- }
- Broadcast::DropSet(set_id) => {
- // stop making announcements for any dead rounds.
- self.network.drop_set_messages(set_id.0);
- }
- }
- }
- }
- }
- }
-}
-
-impl> Network for BroadcastHandle {
- type In = N::In;
-
- fn messages_for(&self, round: u64, set_id: u64) -> Self::In {
- self.network.messages_for(round, set_id)
- }
-
- fn send_message(&self, round: u64, set_id: u64, message: Vec, _force: bool) {
- let _ = self.relay.unbounded_send(Broadcast::Message(Round(round), SetId(set_id), message));
- }
-
- fn drop_round_messages(&self, round: u64, set_id: u64) {
- let _ = self.relay.unbounded_send(Broadcast::DropRound(Round(round), SetId(set_id)));
- }
-
- fn drop_set_messages(&self, set_id: u64) {
- let _ = self.relay.unbounded_send(Broadcast::DropSet(SetId(set_id)));
- }
-
- fn commit_messages(&self, set_id: u64) -> Self::In {
- self.network.commit_messages(set_id)
- }
-
- fn send_commit(&self, round: u64, set_id: u64, message: Vec, _force: bool) {
- let _ = self.relay.unbounded_send(Broadcast::Commit(Round(round), SetId(set_id), message));
- }
-
- fn announce(&self, round: u64, set_id: u64, block: B::Hash) {
- let _ = self.relay.unbounded_send(
- Broadcast::Announcement(Round(round), SetId(set_id), block)
- );
- }
-}
-
-// check a message.
-pub(crate) fn check_message_sig(
- message: &Message,
- id: &AuthorityId,
- signature: &AuthoritySignature,
- round: u64,
- set_id: u64,
-) -> Result<(), ()> {
- let as_public = AuthorityId::from_raw(id.0);
- let encoded_raw = localized_payload(round, set_id, message);
- if ed25519::Pair::verify(signature, &encoded_raw, as_public) {
- Ok(())
- } else {
- debug!(target: "afg", "Bad signature on message from {:?}", id);
- Err(())
- }
-}
-
-/// converts a message stream into a stream of signed messages.
-/// the output stream checks signatures also.
-pub(crate) fn checked_message_stream(
- inner: S,
- voters: Arc>,
-)
- -> impl Stream,Error=Error> where
- S: Stream,Error=()>
-{
- inner
- .filter_map(|raw| {
- let decoded = GossipMessage::::decode(&mut &raw[..]);
- if decoded.is_none() {
- debug!(target: "afg", "Skipping malformed message {:?}", raw);
- }
- decoded
- })
- .and_then(move |msg| {
- match msg {
- GossipMessage::VoteOrPrecommit(msg) => {
- // check signature.
- if !voters.contains_key(&msg.message.id) {
- debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
- return Ok(None);
- }
-
- match &msg.message.message {
- Prevote(prevote) => {
- telemetry!(CONSENSUS_INFO; "afg.received_prevote";
- "voter" => ?format!("{}", msg.message.id),
- "target_number" => ?prevote.target_number,
- "target_hash" => ?prevote.target_hash,
- );
- },
- Precommit(precommit) => {
- telemetry!(CONSENSUS_INFO; "afg.received_precommit";
- "voter" => ?format!("{}", msg.message.id),
- "target_number" => ?precommit.target_number,
- "target_hash" => ?precommit.target_hash,
- );
- },
- };
-
- Ok(Some(msg.message))
- }
- _ => {
- debug!(target: "afg", "Skipping unknown message type");
- return Ok(None);
- }
- }
- })
- .filter_map(|x| x)
- .map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
-}
-
-pub(crate) struct OutgoingMessages> {
- round: u64,
- set_id: u64,
- locals: Option<(Arc, AuthorityId)>,
- sender: mpsc::UnboundedSender>,
- network: N,
-}
-
-impl> Sink for OutgoingMessages
-{
- type SinkItem = Message;
- type SinkError = Error;
-
- fn start_send(&mut self, msg: Message) -> StartSend, Error> {
- // when locals exist, sign messages on import
- if let Some((ref pair, ref local_id)) = self.locals {
- let encoded = localized_payload(self.round, self.set_id, &msg);
- let signature = pair.sign(&encoded[..]);
-
- let target_hash = msg.target().0.clone();
- let signed = SignedMessage:: {
- message: msg,
- signature,
- id: local_id.clone(),
- };
-
- let message = GossipMessage::VoteOrPrecommit(VoteOrPrecommitMessage:: {
- message: signed.clone(),
- round: self.round,
- set_id: self.set_id,
- });
-
- // announce our block hash to peers and propagate the
- // message.
- self.network.announce(self.round, self.set_id, target_hash);
- self.network.send_message(self.round, self.set_id, message.encode(), false);
-
- // forward the message to the inner sender.
- let _ = self.sender.unbounded_send(signed);
- }
-
- Ok(AsyncSink::Ready)
- }
-
- fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
-
- fn close(&mut self) -> Poll<(), Error> {
- // ignore errors since we allow this inner sender to be closed already.
- self.sender.close().or_else(|_| Ok(Async::Ready(())))
- }
-}
-
-impl> Drop for OutgoingMessages {
- fn drop(&mut self) {
- self.network.drop_round_messages(self.round, self.set_id);
- }
-}
-
-/// A sink for outgoing messages. This signs the messages with the key,
-/// if we are an authority. A stream for the signed messages is also returned.
-///
-/// A future can push unsigned messages into the sink. They will be automatically
-/// broadcast to the network. The returned stream should be combined with other input.
-pub(crate) fn outgoing_messages>(
- round: u64,
- set_id: u64,
- local_key: Option>,
- voters: Arc>,
- network: N,
-) -> (
- impl Stream,Error=Error>,
- OutgoingMessages,
-) {
- let locals = local_key.and_then(|pair| {
- let public = pair.public();
- let id = AuthorityId(public.0);
- if voters.contains_key(&id) {
- Some((pair, id))
- } else {
- None
- }
- });
-
- let (tx, rx) = mpsc::unbounded();
- let outgoing = OutgoingMessages:: {
- round,
- set_id,
- network,
- locals,
- sender: tx,
- };
-
- let rx = rx.map_err(move |()| Error::Network(
- format!("Failed to receive on unbounded receiver for round {}", round)
- ));
-
- (rx, outgoing)
-}
-
-fn check_compact_commit(
- msg: CompactCommit,
- voters: &VoterSet,
-) -> Option> {
- if msg.precommits.len() != msg.auth_data.len() || msg.precommits.is_empty() {
- debug!(target: "afg", "Skipping malformed compact commit");
- return None;
- }
-
- // check signatures on all contained precommits.
- for (_, ref id) in &msg.auth_data {
- if !voters.contains_key(id) {
- debug!(target: "afg", "Skipping commit containing unknown voter {}", id);
- return None;
- }
- }
-
- Some(msg)
-}
-
-/// A stream for incoming commit messages. This checks all the signatures on the
-/// messages.
-pub(crate) fn checked_commit_stream(
- inner: S,
- voters: Arc>,
-)
- -> impl Stream),Error=Error> where
- S: Stream,Error=()>
-{
- inner
- .filter_map(|raw| {
- // this could be optimized by decoding piecewise.
- let decoded = GossipMessage::::decode(&mut &raw[..]);
- if decoded.is_none() {
- trace!(target: "afg", "Skipping malformed commit message {:?}", raw);
- }
- decoded
- })
- .filter_map(move |msg| {
- match msg {
- GossipMessage::Commit(msg) => {
- let round = msg.round;
- let precommits_signed_by: Vec =
- msg.message.auth_data.iter().map(move |(_, a)| {
- format!("{}", a)
- }).collect();
- telemetry!(CONSENSUS_INFO; "afg.received_commit";
- "contains_precommits_signed_by" => ?precommits_signed_by,
- "target_number" => ?msg.message.target_number,
- "target_hash" => ?msg.message.target_hash,
- );
- check_compact_commit::(msg.message, &*voters).map(move |c| (round, c))
- },
- _ => {
- debug!(target: "afg", "Skipping unknown message type");
- return None;
- }
- }
- })
- .map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")))
-}
-
-/// An output sink for commit messages.
-pub(crate) struct CommitsOut> {
- network: N,
- set_id: u64,
- _marker: ::std::marker::PhantomData,
- is_voter: bool,
-}
-
-impl> CommitsOut {
- /// Create a new commit output stream.
- pub(crate) fn new(network: N, set_id: u64, is_voter: bool) -> Self {
- CommitsOut {
- network,
- set_id,
- is_voter,
- _marker: Default::default(),
- }
- }
-}
-
-impl> Sink for CommitsOut {
- type SinkItem = (u64, Commit);
- type SinkError = Error;
-
- fn start_send(&mut self, input: (u64, Commit)) -> StartSend {
- if !self.is_voter {
- return Ok(AsyncSink::Ready);
- }
-
- let (round, commit) = input;
- telemetry!(CONSENSUS_INFO; "afg.commit_issued";
- "target_number" => ?commit.target_number, "target_hash" => ?commit.target_hash,
- );
- let (precommits, auth_data) = commit.precommits.into_iter()
- .map(|signed| (signed.precommit, (signed.signature, signed.id)))
- .unzip();
-
- let compact_commit = CompactCommit:: {
- target_hash: commit.target_hash,
- target_number: commit.target_number,
- precommits,
- auth_data
- };
-
- let message = GossipMessage::Commit(FullCommitMessage:: {
- round: round,
- set_id: self.set_id,
- message: compact_commit,
- });
-
- self.network.send_commit(round, self.set_id, Encode::encode(&message), false);
-
- Ok(AsyncSink::Ready)
- }
-
- fn close(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
- fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
-}
-
-impl> Drop for CommitsOut {
- fn drop(&mut self) {
- self.network.drop_set_messages(self.set_id);
- }
-}
diff --git a/core/finality-grandpa/src/communication/gossip.rs b/core/finality-grandpa/src/communication/gossip.rs
new file mode 100644
index 000000000000..720c083db792
--- /dev/null
+++ b/core/finality-grandpa/src/communication/gossip.rs
@@ -0,0 +1,1005 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate. If not, see .
+
+//! Gossip and politeness for polite-grandpa.
+//!
+//! This module implements the following message types:
+//! #### Neighbor Packet
+//!
+//! The neighbor packet is sent to only our neighbors. It contains this information
+//!
+//! - Current Round
+//! - Current voter set ID
+//! - Last finalized hash from commit messages.
+//!
+//! If a peer is at a given voter set, it is impolite to send messages from
+//! an earlier voter set. It is extremely impolite to send messages
+//! from a future voter set. "future-set" messages can be dropped and ignored.
+//!
+//! If a peer is at round r, is impolite to send messages about r-2 or earlier and extremely
+//! impolite to send messages about r+1 or later. "future-round" messages can
+//! be dropped and ignored.
+//!
+//! It is impolite to send a neighbor packet which moves backwards in protocol state.
+//!
+//! This is beneficial if it conveys some progress in the protocol state of the peer.
+//!
+//! #### Prevote / Precommit
+//!
+//! These are votes within a round. Noting that we receive these messages
+//! from our peers who are not necessarily voters, we have to account the benefit
+//! based on what they might have seen.
+//!
+//! #### Propose
+//!
+//! This is a broadcast by a known voter of the last-round estimate.
+
+//! #### Commit
+//!
+//! These are used to announce past agreement of finality.
+//!
+//! It is impolite to send commits which are earlier than the last commit
+//! sent. It is especially impolite to send commits which are invalid, or from
+//! a different Set ID than the receiving peer has indicated.
+//!
+//! Sending a commit is polite when it may finalize something that the receiving peer
+//! was not aware of.
+//!
+//! ## Expiration
+//!
+//! We keep some amount of recent rounds' messages, but do not accept new ones from rounds
+//! older than our current_round - 1.
+//!
+//! ## Message Validation
+//!
+//! We only send polite messages to peers,
+
+use runtime_primitives::traits::{NumberFor, Block as BlockT, Zero};
+use network::consensus_gossip::{self as network_gossip, MessageIntent, ValidatorContext};
+use network::{config::Roles, PeerId};
+use parity_codec::{Encode, Decode};
+
+use substrate_telemetry::{telemetry, CONSENSUS_DEBUG};
+use log::{trace, debug};
+
+use crate::{CompactCommit, SignedMessage};
+use super::{Round, SetId, Network};
+
+use std::collections::{HashMap, VecDeque};
+use std::time::{Duration, Instant};
+
+const REBROADCAST_AFTER: Duration = Duration::from_secs(60 * 5);
+
+/// An outcome of examining a message.
+#[derive(Debug, PartialEq, Clone, Copy)]
+enum Consider {
+ /// Accept the message.
+ Accept,
+ /// Message is too early. Reject.
+ RejectPast,
+ /// Message is from the future. Reject.
+ RejectFuture,
+}
+
+/// A view of protocol state.
+#[derive(Debug)]
+struct View {
+ round: Round, // the current round we are at.
+ set_id: SetId, // the current voter set id.
+ last_commit: Option, // commit-finalized block height, if any.
+}
+
+impl Default for View {
+ fn default() -> Self {
+ View {
+ round: Round(0),
+ set_id: SetId(0),
+ last_commit: None,
+ }
+ }
+}
+
+impl View {
+ /// Update the set ID. implies a reset to round 0.
+ fn update_set(&mut self, set_id: SetId) {
+ if set_id != self.set_id {
+ self.set_id = set_id;
+ self.round = Round(0);
+ }
+ }
+
+ /// Consider a round and set ID combination under a current view.
+ fn consider_vote(&self, round: Round, set_id: SetId) -> Consider {
+ // only from current set
+ if set_id < self.set_id { return Consider::RejectPast }
+ if set_id > self.set_id { return Consider::RejectFuture }
+
+ // only r-1 ... r+1
+ if round.0 > self.round.0.saturating_add(1) { return Consider::RejectFuture }
+ if round.0 < self.round.0.saturating_sub(1) { return Consider::RejectPast }
+
+ Consider::Accept
+ }
+
+ /// Consider a set-id global message. Rounds are not taken into account, but are implicitly
+ /// because we gate on finalization of a further block than a previous commit.
+ fn consider_global(&self, set_id: SetId, number: N) -> Consider {
+ // only from current set
+ if set_id < self.set_id { return Consider::RejectPast }
+ if set_id > self.set_id { return Consider::RejectFuture }
+
+ // only commits which claim to prove a higher block number than
+ // the one we're aware of.
+ match self.last_commit {
+ None => Consider::Accept,
+ Some(ref num) => if num < &number {
+ Consider::Accept
+ } else {
+ Consider::RejectPast
+ }
+ }
+ }
+}
+
+const KEEP_RECENT_ROUNDS: usize = 3;
+
+/// Tracks topics we keep messages for.
+struct KeepTopics {
+ current_set: SetId,
+ rounds: VecDeque<(Round, SetId)>,
+ reverse_map: HashMap, SetId)>
+}
+
+impl KeepTopics {
+ fn new() -> Self {
+ KeepTopics {
+ current_set: SetId(0),
+ rounds: VecDeque::with_capacity(KEEP_RECENT_ROUNDS + 1),
+ reverse_map: HashMap::new(),
+ }
+ }
+
+ fn push(&mut self, round: Round, set_id: SetId) {
+ self.current_set = std::cmp::max(self.current_set, set_id);
+ self.rounds.push_back((round, set_id));
+
+ // the 1 is for the current round.
+ while self.rounds.len() > KEEP_RECENT_ROUNDS + 1 {
+ let _ = self.rounds.pop_front();
+ }
+
+ let mut map = HashMap::with_capacity(KEEP_RECENT_ROUNDS + 2);
+ map.insert(super::global_topic::(self.current_set.0), (None, self.current_set));
+
+ for &(round, set) in &self.rounds {
+ map.insert(
+ super::round_topic::(round.0, set.0),
+ (Some(round), set)
+ );
+ }
+
+ self.reverse_map = map;
+ }
+
+ fn topic_info(&self, topic: &B::Hash) -> Option<(Option, SetId)> {
+ self.reverse_map.get(topic).cloned()
+ }
+}
+
+// topics to send to a neighbor based on their view.
+fn neighbor_topics(view: &View>) -> Vec {
+ let s = view.set_id;
+ let mut topics = vec![
+ super::global_topic::(s.0),
+ super::round_topic::(view.round.0, s.0),
+ ];
+
+ if view.round.0 != 0 {
+ let r = Round(view.round.0 - 1);
+ topics.push(super::round_topic::(r.0, s.0))
+ }
+
+ topics
+}
+
+/// Grandpa gossip message type.
+/// This is the root type that gets encoded and sent on the network.
+#[derive(Debug, Encode, Decode)]
+pub(super) enum GossipMessage {
+ /// Grandpa message with round and set info.
+ VoteOrPrecommit(VoteOrPrecommitMessage),
+ /// Grandpa commit message with round and set info.
+ Commit(FullCommitMessage),
+ /// A neighbor packet. Not repropagated.
+ Neighbor(VersionedNeighborPacket>),
+}
+
+/// Network level message with topic information.
+#[derive(Debug, Encode, Decode)]
+pub(super) struct VoteOrPrecommitMessage {
+ /// The round this message is from.
+ pub(super) round: Round,
+ /// The voter set ID this message is from.
+ pub(super) set_id: SetId,
+ /// The message itself.
+ pub(super) message: SignedMessage,
+}
+
+/// Network level commit message with topic information.
+#[derive(Debug, Encode, Decode)]
+pub(super) struct FullCommitMessage {
+ /// The round this message is from.
+ pub(super) round: Round,
+ /// The voter set ID this message is from.
+ pub(super) set_id: SetId,
+ /// The compact commit message.
+ pub(super) message: CompactCommit,
+}
+
+/// V1 neighbor packet. Neighbor packets are sent from nodes to their peers
+/// and are not repropagated. These contain information about the node's state.
+#[derive(Debug, Encode, Decode, Clone)]
+pub(super) struct NeighborPacket {
+ round: Round,
+ set_id: SetId,
+ commit_finalized_height: N,
+}
+
+/// A versioned neighbor packet.
+#[derive(Debug, Encode, Decode)]
+pub(super) enum VersionedNeighborPacket {
+ #[codec(index = "1")]
+ V1(NeighborPacket),
+}
+
+impl VersionedNeighborPacket {
+ fn into_neighbor_packet(self) -> NeighborPacket {
+ match self {
+ VersionedNeighborPacket::V1(p) => p,
+ }
+ }
+}
+
+// cost scalars for reporting peers.
+mod cost {
+ pub(super) const PAST_REJECTION: i32 = -50;
+ pub(super) const BAD_SIGNATURE: i32 = -100;
+ pub(super) const MALFORMED_COMMIT: i32 = -1000;
+ pub(super) const FUTURE_MESSAGE: i32 = -500;
+
+ pub(super) const INVALID_VIEW_CHANGE: i32 = -500;
+ pub(super) const PER_UNDECODABLE_BYTE: i32 = -5;
+ pub(super) const PER_SIGNATURE_CHECKED: i32 = -25;
+ pub(super) const PER_BLOCK_LOADED: i32 = -10;
+}
+
+// benefit scalars for reporting peers.
+mod benefit {
+ pub(super) const ROUND_MESSAGE: i32 = 100;
+ pub(super) const BASIC_VALIDATED_COMMIT: i32 = 100;
+ pub(super) const PER_EQUIVOCATION: i32 = 10;
+}
+
+/// Misbehavior that peers can perform.
+///
+/// `cost` gives a cost that can be used to perform cost/benefit analysis of a
+/// peer.
+#[derive(Clone, Copy, Debug, PartialEq)]
+enum Misbehavior {
+ // invalid neighbor message, considering the last one.
+ InvalidViewChange,
+ // could not decode neighbor message. bytes-length of the packet.
+ UndecodablePacket(i32),
+ // Bad commit message
+ BadCommitMessage {
+ signatures_checked: i32,
+ blocks_loaded: i32,
+ equivocations_caught: i32,
+ },
+ // A message received that's from the future relative to our view.
+ // always misbehavior.
+ FutureMessage,
+}
+
+impl Misbehavior {
+ fn cost(&self) -> i32 {
+ use Misbehavior::*;
+
+ match *self {
+ InvalidViewChange => cost::INVALID_VIEW_CHANGE,
+ UndecodablePacket(bytes) => bytes.saturating_mul(cost::PER_UNDECODABLE_BYTE),
+ BadCommitMessage { signatures_checked, blocks_loaded, equivocations_caught } => {
+ let cost = cost::PER_SIGNATURE_CHECKED
+ .saturating_mul(signatures_checked)
+ .saturating_add(cost::PER_BLOCK_LOADED.saturating_mul(blocks_loaded));
+
+ let benefit = equivocations_caught.saturating_mul(benefit::PER_EQUIVOCATION);
+
+ (benefit as i32).saturating_add(cost as i32)
+ },
+ FutureMessage => cost::FUTURE_MESSAGE,
+ }
+ }
+}
+
+struct PeerInfo {
+ view: View,
+}
+
+impl PeerInfo {
+ fn new() -> Self {
+ PeerInfo {
+ view: View::default(),
+ }
+ }
+}
+
+/// The peers we're connected do in gossip.
+struct Peers {
+ inner: HashMap>,
+}
+
+impl Default for Peers {
+ fn default() -> Self {
+ Peers { inner: HashMap::new() }
+ }
+}
+
+impl Peers {
+ fn new_peer(&mut self, who: PeerId) {
+ self.inner.insert(who, PeerInfo::new());
+ }
+
+ fn peer_disconnected(&mut self, who: &PeerId) {
+ self.inner.remove(who);
+ }
+
+ // returns a reference to the new view, if the peer is known.
+ fn update_peer_state(&mut self, who: &PeerId, update: NeighborPacket)
+ -> Result>, Misbehavior>
+ {
+ let peer = match self.inner.get_mut(who) {
+ None => return Ok(None),
+ Some(p) => p,
+ };
+
+ let invalid_change = peer.view.set_id > update.set_id
+ || peer.view.round > update.round && peer.view.set_id == update.set_id
+ || peer.view.last_commit.as_ref() > Some(&update.commit_finalized_height);
+
+ if invalid_change {
+ return Err(Misbehavior::InvalidViewChange);
+ }
+
+ peer.view = View {
+ round: update.round,
+ set_id: update.set_id,
+ last_commit: Some(update.commit_finalized_height),
+ };
+
+ trace!(target: "afg", "Peer {} updated view. Now at {:?}, {:?}",
+ who, peer.view.round, peer.view.set_id);
+
+ Ok(Some(&peer.view))
+ }
+
+ fn update_commit_height(&mut self, who: &PeerId, new_height: N) -> Result<(), Misbehavior> {
+ let peer = match self.inner.get_mut(who) {
+ None => return Ok(()),
+ Some(p) => p,
+ };
+
+ if peer.view.last_commit.as_ref() >= Some(&new_height) {
+ return Err(Misbehavior::InvalidViewChange);
+ }
+ peer.view.last_commit = Some(new_height);
+
+ Ok(())
+ }
+
+ fn peer<'a>(&'a self, who: &PeerId) -> Option<&'a PeerInfo> {
+ self.inner.get(who)
+ }
+}
+
+#[derive(Debug)]
+enum Action {
+ // repropagate under given topic, to the given peers, applying cost/benefit to originator.
+ Keep(H, i32),
+ // discard and process.
+ ProcessAndDiscard(H, i32),
+ // discard, applying cost/benefit to originator.
+ Discard(i32),
+}
+
+struct Inner {
+ local_view: View>,
+ peers: Peers>,
+ live_topics: KeepTopics,
+ config: crate::Config,
+ next_rebroadcast: Instant,
+}
+
+impl Inner {
+ fn new(config: crate::Config) -> Self {
+ Inner {
+ local_view: View::default(),
+ peers: Peers::default(),
+ live_topics: KeepTopics::new(),
+ next_rebroadcast: Instant::now() + REBROADCAST_AFTER,
+ config,
+ }
+ }
+
+ /// Note a round in a set has started.
+ fn note_round>(&mut self, round: Round, set_id: SetId, net: &N) {
+ if self.local_view.round == round && self.local_view.set_id == set_id {
+ return
+ }
+
+ debug!(target: "afg", "Voter {} noting beginning of round {:?} to network.",
+ self.config.name(), (round, set_id));
+
+ self.local_view.round = round;
+ self.local_view.set_id = set_id;
+
+ self.live_topics.push(round, set_id);
+ self.multicast_neighbor_packet(net);
+ }
+
+ /// Note that a voter set with given ID has started. Does nothing if the last
+ /// call to the function was with the same `set_id`.
+ fn note_set>(&mut self, set_id: SetId, net: &N) {
+ if self.local_view.set_id == set_id { return }
+
+ self.local_view.update_set(set_id);
+ self.live_topics.push(Round(0), set_id);
+ self.multicast_neighbor_packet(net);
+ }
+
+ /// Note that we've imported a commit finalizing a given block.
+ fn note_commit_finalized>(&mut self, finalized: NumberFor, net: &N) {
+ if self.local_view.last_commit.as_ref() < Some(&finalized) {
+ self.local_view.last_commit = Some(finalized);
+ self.multicast_neighbor_packet(net)
+ }
+ }
+
+ fn consider_vote(&self, round: Round, set_id: SetId) -> Consider {
+ self.local_view.consider_vote(round, set_id)
+ }
+
+ fn consider_global(&self, set_id: SetId, number: NumberFor) -> Consider {
+ self.local_view.consider_global(set_id, number)
+ }
+
+ fn cost_past_rejection(&self, _who: &PeerId, _round: Round, _set_id: SetId) -> i32 {
+ // hardcoded for now.
+ cost::PAST_REJECTION
+ }
+
+ fn validate_round_message(&self, who: &PeerId, full: &VoteOrPrecommitMessage)
+ -> Action
+ {
+ match self.consider_vote(full.round, full.set_id) {
+ Consider::RejectFuture => return Action::Discard(Misbehavior::FutureMessage.cost()),
+ Consider::RejectPast =>
+ return Action::Discard(self.cost_past_rejection(who, full.round, full.set_id)),
+ Consider::Accept => {},
+ }
+
+ if let Err(()) = super::check_message_sig::(
+ &full.message.message,
+ &full.message.id,
+ &full.message.signature,
+ full.round.0,
+ full.set_id.0,
+ ) {
+ debug!(target: "afg", "Bad message signature {}", full.message.id);
+ telemetry!(CONSENSUS_DEBUG; "afg.bad_msg_signature"; "signature" => ?full.message.id);
+ return Action::Discard(cost::BAD_SIGNATURE);
+ }
+
+ let topic = super::round_topic::(full.round.0, full.set_id.0);
+ Action::Keep(topic, benefit::ROUND_MESSAGE)
+ }
+
+ fn validate_commit_message(&mut self, who: &PeerId, full: &FullCommitMessage)
+ -> Action
+ {
+ use grandpa::Message as GrandpaMessage;
+
+ if let Err(misbehavior) = self.peers.update_commit_height(who, full.message.target_number) {
+ return Action::Discard(misbehavior.cost());
+ }
+
+ match self.consider_global(full.set_id, full.message.target_number) {
+ Consider::RejectFuture => return Action::Discard(Misbehavior::FutureMessage.cost()),
+ Consider::RejectPast =>
+ return Action::Discard(self.cost_past_rejection(who, full.round, full.set_id)),
+ Consider::Accept => {},
+ }
+
+ if full.message.precommits.len() != full.message.auth_data.len() || full.message.precommits.is_empty() {
+ debug!(target: "afg", "Malformed compact commit");
+ telemetry!(CONSENSUS_DEBUG; "afg.malformed_compact_commit";
+ "precommits_len" => ?full.message.precommits.len(),
+ "auth_data_len" => ?full.message.auth_data.len(),
+ "precommits_is_empty" => ?full.message.precommits.is_empty(),
+ );
+ return Action::Discard(cost::MALFORMED_COMMIT);
+ }
+
+ // check signatures on all contained precommits.
+ for (i, (precommit, &(ref sig, ref id))) in full.message.precommits.iter()
+ .zip(&full.message.auth_data)
+ .enumerate()
+ {
+ if let Err(()) = super::check_message_sig::(
+ &GrandpaMessage::Precommit(precommit.clone()),
+ id,
+ sig,
+ full.round.0,
+ full.set_id.0,
+ ) {
+ debug!(target: "afg", "Bad commit message signature {}", id);
+ telemetry!(CONSENSUS_DEBUG; "afg.bad_commit_msg_signature"; "id" => ?id);
+ return Action::Discard(Misbehavior::BadCommitMessage {
+ signatures_checked: i as i32,
+ blocks_loaded: 0,
+ equivocations_caught: 0,
+ }.cost());
+ }
+ }
+
+ // always discard commits initially and rebroadcast after doing full
+ // checking.
+ let topic = super::global_topic::(full.set_id.0);
+ Action::ProcessAndDiscard(topic, benefit::BASIC_VALIDATED_COMMIT)
+ }
+
+ fn import_neighbor_message(&mut self, who: &PeerId, update: NeighborPacket>)
+ -> (Vec, Action)
+ {
+ let (cb, topics) = match self.peers.update_peer_state(who, update) {
+ Ok(view) => (100i32, view.map(|view| neighbor_topics::(view))),
+ Err(misbehavior) => (misbehavior.cost(), None)
+ };
+
+ let neighbor_topics = topics.unwrap_or_default();
+
+ // always discard, it's valid for one hop.
+ (neighbor_topics, Action::Discard(cb))
+ }
+
+ fn construct_neighbor_packet(&self) -> GossipMessage {
+ let packet = NeighborPacket {
+ round: self.local_view.round,
+ set_id: self.local_view.set_id,
+ commit_finalized_height: self.local_view.last_commit.unwrap_or(Zero::zero()),
+ };
+
+ GossipMessage::Neighbor(VersionedNeighborPacket::V1(packet))
+ }
+
+ fn multicast_neighbor_packet>(&self, net: &N) {
+ let packet = self.construct_neighbor_packet();
+ let peers = self.peers.inner.keys().cloned().collect();
+ net.send_message(peers, packet.encode());
+ }
+}
+
+/// A validator for GRANDPA gossip messages.
+pub(super) struct GossipValidator {
+ inner: parking_lot::RwLock>,
+}
+
+impl GossipValidator {
+ /// Create a new gossip-validator.
+ pub(super) fn new(config: crate::Config) -> GossipValidator {
+ GossipValidator { inner: parking_lot::RwLock::new(Inner::new(config)) }
+ }
+
+ /// Note a round in a set has started.
+ pub(super) fn note_round>(&self, round: Round, set_id: SetId, net: &N) {
+ self.inner.write().note_round(round, set_id, net);
+ }
+
+ /// Note that a voter set with given ID has started.
+ pub(super) fn note_set>(&self, set_id: SetId, net: &N) {
+ self.inner.write().note_set(set_id, net);
+ }
+
+ /// Note that we've imported a commit finalizing a given block.
+ pub(super) fn note_commit_finalized>(&self, finalized: NumberFor, net: &N) {
+ self.inner.write().note_commit_finalized(finalized, net);
+ }
+
+ fn report(&self, _who: &PeerId, _cost_benefit: i32) {
+ // report
+ }
+
+ fn do_validate(&self, who: &PeerId, mut data: &[u8])
+ -> (Action, Vec)
+ {
+ let mut broadcast_topics = Vec::new();
+ let action = {
+ let mut inner = self.inner.write();
+ match GossipMessage::::decode(&mut data) {
+ Some(GossipMessage::VoteOrPrecommit(ref message))
+ => inner.validate_round_message(who, message),
+ Some(GossipMessage::Commit(ref message)) => inner.validate_commit_message(who, message),
+ Some(GossipMessage::Neighbor(update)) => {
+ let (topics, action) = inner.import_neighbor_message(
+ who,
+ update.into_neighbor_packet(),
+ );
+
+ broadcast_topics = topics;
+ action
+ }
+ None => {
+ debug!(target: "afg", "Error decoding message");
+ telemetry!(CONSENSUS_DEBUG; "afg.err_decoding_msg"; "" => "");
+
+ let len = std::cmp::min(i32::max_value() as usize, data.len()) as i32;
+ Action::Discard(Misbehavior::UndecodablePacket(len).cost())
+ }
+ }
+ };
+
+ (action, broadcast_topics)
+ }
+}
+
+impl network_gossip::Validator for GossipValidator {
+ fn new_peer(&self, context: &mut ValidatorContext, who: &PeerId, _roles: Roles) {
+ let packet_data = {
+ let mut inner = self.inner.write();
+ inner.peers.new_peer(who.clone());
+ inner.construct_neighbor_packet().encode()
+ };
+ context.send_message(who, packet_data);
+ }
+
+ fn peer_disconnected(&self, _context: &mut ValidatorContext, who: &PeerId) {
+ self.inner.write().peers.peer_disconnected(who);
+ }
+
+ fn validate(&self, context: &mut ValidatorContext, who: &PeerId, data: &[u8])
+ -> network_gossip::ValidationResult
+ {
+ let (action, broadcast_topics) = self.do_validate(who, data);
+
+ // not with lock held!
+ for topic in broadcast_topics {
+ context.send_topic(who, topic, false);
+ }
+
+ match action {
+ Action::Keep(topic, cb) => {
+ self.report(who, cb);
+ network_gossip::ValidationResult::ProcessAndKeep(topic)
+ }
+ Action::ProcessAndDiscard(topic, cb) => {
+ self.report(who, cb);
+ network_gossip::ValidationResult::ProcessAndDiscard(topic)
+ }
+ Action::Discard(cb) => {
+ self.report(who, cb);
+ network_gossip::ValidationResult::Discard
+ }
+ }
+ }
+
+ fn message_allowed<'a>(&'a self)
+ -> Box bool + 'a>
+ {
+ let (inner, do_rebroadcast) = {
+ use parking_lot::RwLockWriteGuard;
+
+ let mut inner = self.inner.write();
+ let now = Instant::now();
+ let do_rebroadcast = if now >= inner.next_rebroadcast {
+ inner.next_rebroadcast = now + REBROADCAST_AFTER;
+ true
+ } else {
+ false
+ };
+
+ // downgrade to read-lock.
+ (RwLockWriteGuard::downgrade(inner), do_rebroadcast)
+ };
+
+ Box::new(move |who, intent, topic, mut data| {
+ if let MessageIntent::PeriodicRebroadcast = intent {
+ return do_rebroadcast;
+ }
+
+ let peer = match inner.peers.peer(who) {
+ None => return false,
+ Some(x) => x,
+ };
+
+ // if the topic is not something we're keeping at the moment,
+ // do not send.
+ let (maybe_round, set_id) = match inner.live_topics.topic_info(&topic) {
+ None => return false,
+ Some(x) => x,
+ };
+
+ // if the topic is not something the peer accepts, discard.
+ if let Some(round) = maybe_round {
+ return peer.view.consider_vote(round, set_id) == Consider::Accept
+ }
+
+ // global message.
+ let our_best_commit = inner.local_view.last_commit;
+ let peer_best_commit = peer.view.last_commit;
+
+ match GossipMessage::::decode(&mut data) {
+ None => false,
+ Some(GossipMessage::Commit(full)) => {
+ // we only broadcast our best commit and only if it's
+ // better than last received by peer.
+ Some(full.message.target_number) == our_best_commit
+ && Some(full.message.target_number) > peer_best_commit
+ }
+ Some(GossipMessage::Neighbor(_)) => false,
+ Some(GossipMessage::VoteOrPrecommit(_)) => false, // should not be the case.
+ }
+ })
+ }
+
+ fn message_expired<'a>(&'a self) -> Box bool + 'a> {
+ let inner = self.inner.read();
+ Box::new(move |topic, mut data| {
+ // if the topic is not one of the ones that we are keeping at the moment,
+ // it is expired.
+ match inner.live_topics.topic_info(&topic) {
+ None => return true,
+ Some((Some(_), _)) => return false, // round messages don't require further checking.
+ Some((None, _)) => {},
+ };
+
+ // global messages -- only keep the best commit.
+ let best_commit = inner.local_view.last_commit;
+
+ match GossipMessage::::decode(&mut data) {
+ None => true,
+ Some(GossipMessage::Commit(full))
+ => Some(full.message.target_number) != best_commit,
+ Some(_) => true,
+ }
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use network_gossip::Validator as GossipValidatorT;
+ use network::test::Block;
+
+ // some random config (not really needed)
+ fn config() -> crate::Config {
+ crate::Config {
+ gossip_duration: Duration::from_millis(10),
+ justification_period: 256,
+ local_key: None,
+ name: None,
+ }
+ }
+
+ #[derive(Clone)]
+ struct StubNetwork;
+
+ impl super::Network for StubNetwork {
+ type In = futures::stream::Empty;
+
+ fn messages_for(&self, _topic: Block::Hash) -> Self::In {
+ futures::stream::empty()
+ }
+
+ fn register_validator(
+ &self,
+ _validator: std::sync::Arc>,
+ ) {
+
+ }
+
+ fn gossip_message(&self, _topic: Block::Hash, _data: Vec, _force: bool) {
+
+ }
+
+ fn send_message(&self, _who: Vec, _data: Vec) {
+
+ }
+
+ fn announce(&self, _block: Block::Hash) {
+
+ }
+ }
+
+ #[test]
+ fn view_vote_rules() {
+ let view = View { round: Round(100), set_id: SetId(1), last_commit: Some(1000u64) };
+
+ assert_eq!(view.consider_vote(Round(98), SetId(1)), Consider::RejectPast);
+ assert_eq!(view.consider_vote(Round(1), SetId(0)), Consider::RejectPast);
+ assert_eq!(view.consider_vote(Round(1000), SetId(0)), Consider::RejectPast);
+
+ assert_eq!(view.consider_vote(Round(99), SetId(1)), Consider::Accept);
+ assert_eq!(view.consider_vote(Round(100), SetId(1)), Consider::Accept);
+ assert_eq!(view.consider_vote(Round(101), SetId(1)), Consider::Accept);
+
+ assert_eq!(view.consider_vote(Round(102), SetId(1)), Consider::RejectFuture);
+ assert_eq!(view.consider_vote(Round(1), SetId(2)), Consider::RejectFuture);
+ assert_eq!(view.consider_vote(Round(1000), SetId(2)), Consider::RejectFuture);
+ }
+
+ #[test]
+ fn view_global_message_rules() {
+ let view = View { round: Round(100), set_id: SetId(2), last_commit: Some(1000u64) };
+
+ assert_eq!(view.consider_global(SetId(3), 1), Consider::RejectFuture);
+ assert_eq!(view.consider_global(SetId(3), 1000), Consider::RejectFuture);
+ assert_eq!(view.consider_global(SetId(3), 10000), Consider::RejectFuture);
+
+ assert_eq!(view.consider_global(SetId(1), 1), Consider::RejectPast);
+ assert_eq!(view.consider_global(SetId(1), 1000), Consider::RejectPast);
+ assert_eq!(view.consider_global(SetId(1), 10000), Consider::RejectPast);
+
+ assert_eq!(view.consider_global(SetId(2), 1), Consider::RejectPast);
+ assert_eq!(view.consider_global(SetId(2), 1000), Consider::RejectPast);
+ assert_eq!(view.consider_global(SetId(2), 1001), Consider::Accept);
+ assert_eq!(view.consider_global(SetId(2), 10000), Consider::Accept);
+ }
+
+ #[test]
+ fn unknown_peer_cannot_be_updated() {
+ let mut peers = Peers::default();
+ let id = PeerId::random();
+
+ let update = NeighborPacket {
+ round: Round(5),
+ set_id: SetId(10),
+ commit_finalized_height: 50,
+ };
+
+ let res = peers.update_peer_state(&id, update.clone());
+ assert!(res.unwrap().is_none());
+
+ // connect & disconnect.
+ peers.new_peer(id.clone());
+ peers.peer_disconnected(&id);
+
+ let res = peers.update_peer_state(&id, update.clone());
+ assert!(res.unwrap().is_none());
+ }
+
+ #[test]
+ fn update_peer_state() {
+ let update1 = NeighborPacket {
+ round: Round(5),
+ set_id: SetId(10),
+ commit_finalized_height: 50u32,
+ };
+
+ let update2 = NeighborPacket {
+ round: Round(6),
+ set_id: SetId(10),
+ commit_finalized_height: 60,
+ };
+
+ let update3 = NeighborPacket {
+ round: Round(2),
+ set_id: SetId(11),
+ commit_finalized_height: 61,
+ };
+
+ let update4 = NeighborPacket {
+ round: Round(3),
+ set_id: SetId(11),
+ commit_finalized_height: 80,
+ };
+
+ let mut peers = Peers::default();
+ let id = PeerId::random();
+
+ peers.new_peer(id.clone());
+
+ let mut check_update = move |update: NeighborPacket<_>| {
+ let view = peers.update_peer_state(&id, update.clone()).unwrap().unwrap();
+ assert_eq!(view.round, update.round);
+ assert_eq!(view.set_id, update.set_id);
+ assert_eq!(view.last_commit, Some(update.commit_finalized_height));
+ };
+
+ check_update(update1);
+ check_update(update2);
+ check_update(update3);
+ check_update(update4);
+ }
+
+ #[test]
+ fn invalid_view_change() {
+ let mut peers = Peers::default();
+
+ let id = PeerId::random();
+ peers.new_peer(id.clone());
+
+ peers.update_peer_state(&id, NeighborPacket {
+ round: Round(10),
+ set_id: SetId(10),
+ commit_finalized_height: 10,
+ }).unwrap().unwrap();
+
+ let mut check_update = move |update: NeighborPacket<_>| {
+ let err = peers.update_peer_state(&id, update.clone()).unwrap_err();
+ assert_eq!(err, Misbehavior::InvalidViewChange);
+ };
+
+ // round moves backwards.
+ check_update(NeighborPacket {
+ round: Round(9),
+ set_id: SetId(10),
+ commit_finalized_height: 10,
+ });
+ // commit finalized height moves backwards.
+ check_update(NeighborPacket {
+ round: Round(10),
+ set_id: SetId(10),
+ commit_finalized_height: 9,
+ });
+ // set ID moves backwards.
+ check_update(NeighborPacket {
+ round: Round(10),
+ set_id: SetId(9),
+ commit_finalized_height: 10,
+ });
+ }
+
+ #[test]
+ fn messages_not_expired_immediately() {
+ let val = GossipValidator::::new(config());
+
+ let set_id = 1;
+
+ for round_num in 1u64..10 {
+ val.note_round(Round(round_num), SetId(set_id), &StubNetwork);
+ }
+
+ {
+ let mut is_expired = val.message_expired();
+ let last_kept_round = 10u64 - KEEP_RECENT_ROUNDS as u64 - 1;
+
+ // messages from old rounds are expired.
+ for round_num in 1u64..last_kept_round {
+ println!("{} should be expired?", round_num);
+ let topic = crate::communication::round_topic::(round_num, 1);
+ assert!(is_expired(topic, &[1, 2, 3]));
+ }
+
+ // messages from not-too-old rounds are not expired.
+ for round_num in last_kept_round..10 {
+ println!("{} should not be expired?", round_num);
+ let topic = crate::communication::round_topic::(round_num, 1);
+ assert!(!is_expired(topic, &[1, 2, 3]));
+ }
+ }
+ }
+}
diff --git a/core/finality-grandpa/src/communication/mod.rs b/core/finality-grandpa/src/communication/mod.rs
new file mode 100644
index 000000000000..6d006039f3dc
--- /dev/null
+++ b/core/finality-grandpa/src/communication/mod.rs
@@ -0,0 +1,561 @@
+// Copyright 2017-2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate. If not, see .
+
+//! Communication streams for the polite-grandpa networking protocol.
+//!
+//! GRANDPA nodes communicate over a gossip network, where messages are not sent to
+//! peers until they have reached a given round.
+//!
+//! Rather than expressing protocol rules,
+//! polite-grandpa just carries a notion of impoliteness. Nodes which pass some arbitrary
+//! threshold of impoliteness are removed. Messages are either costly, or beneficial.
+//!
+//! For instance, it is _impolite_ to send the same message more than once.
+//! In the future, there will be a fallback for allowing sending the same message
+//! under certain conditions that are used to un-stick the protocol.
+
+use std::sync::Arc;
+
+use grandpa::VoterSet;
+use grandpa::Message::{Prevote, Precommit};
+use futures::prelude::*;
+use futures::sync::{oneshot, mpsc};
+use log::{debug, trace};
+use parity_codec::{Encode, Decode};
+use substrate_primitives::{ed25519, Pair};
+use substrate_telemetry::{telemetry, CONSENSUS_DEBUG, CONSENSUS_INFO};
+use runtime_primitives::ConsensusEngineId;
+use runtime_primitives::traits::{Block as BlockT, Hash as HashT, Header as HeaderT, NumberFor};
+use network::{consensus_gossip as network_gossip, Service as NetworkService,};
+use network_gossip::ConsensusMessage;
+
+use crate::{Error, Message, SignedMessage, Commit, CompactCommit};
+use gossip::{
+ GossipMessage, FullCommitMessage, VoteOrPrecommitMessage, GossipValidator
+};
+use substrate_primitives::ed25519::{Public as AuthorityId, Signature as AuthoritySignature};
+
+pub mod gossip;
+
+/// The consensus engine ID of GRANDPA.
+pub const GRANDPA_ENGINE_ID: ConsensusEngineId = [b'a', b'f', b'g', b'1'];
+
+/// A handle to the network. This is generally implemented by providing some
+/// handle to a gossip service or similar.
+///
+/// Intended to be a lightweight handle such as an `Arc`.
+pub trait Network: Clone {
+ /// A stream of input messages for a topic.
+ type In: Stream;
+
+ /// Get a stream of messages for a specific gossip topic.
+ fn messages_for(&self, topic: Block::Hash) -> Self::In;
+
+ /// Register a gossip validator.
+ fn register_validator(&self, validator: Arc>);
+
+ /// Gossip a message out to all connected peers.
+ ///
+ /// Force causes it to be sent to all peers, even if they've seen it already.
+ /// Only should be used in case of consensus stall.
+ fn gossip_message(&self, topic: Block::Hash, data: Vec, force: bool);
+
+ /// Send a message to a bunch of specific peers, even if they've seen it already.
+ fn send_message(&self, who: Vec, data: Vec);
+
+ /// Inform peers that a block with given hash should be downloaded.
+ fn announce(&self, block: Block::Hash);
+}
+
+/// Create a unique topic for a round and set-id combo.
+pub(crate) fn round_topic(round: u64, set_id: u64) -> B::Hash {
+ <::Hashing as HashT>::hash(format!("{}-{}", set_id, round).as_bytes())
+}
+
+/// Create a unique topic for global messages on a set ID.
+pub(crate) fn global_topic(set_id: u64) -> B::Hash {
+ <::Hashing as HashT>::hash(format!("{}-GLOBAL", set_id).as_bytes())
+}
+
+impl Network for Arc> where
+ B: BlockT,
+ S: network::specialization::NetworkSpecialization,
+{
+ type In = NetworkStream;
+
+ fn messages_for(&self, topic: B::Hash) -> Self::In {
+ let (tx, rx) = oneshot::channel();
+ self.with_gossip(move |gossip, _| {
+ let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, topic);
+ let _ = tx.send(inner_rx);
+ });
+ NetworkStream { outer: rx, inner: None }
+ }
+
+ fn register_validator(&self, validator: Arc>) {
+ self.with_gossip(
+ move |gossip, context| gossip.register_validator(context, GRANDPA_ENGINE_ID, validator)
+ )
+ }
+
+ fn gossip_message(&self, topic: B::Hash, data: Vec, force: bool) {
+ let msg = ConsensusMessage {
+ engine_id: GRANDPA_ENGINE_ID,
+ data,
+ };
+ self.with_gossip(
+ move |gossip, ctx| gossip.multicast(ctx, topic, msg, force)
+ )
+ }
+
+ fn send_message(&self, who: Vec, data: Vec) {
+ let msg = ConsensusMessage {
+ engine_id: GRANDPA_ENGINE_ID,
+ data,
+ };
+
+ self.with_gossip(move |gossip, ctx| for who in &who {
+ gossip.send_message(ctx, who, msg.clone())
+ })
+ }
+
+ fn announce(&self, block: B::Hash) {
+ self.announce_block(block)
+ }
+}
+
+/// A stream used by NetworkBridge in its implementation of Network.
+pub struct NetworkStream {
+ inner: Option>,
+ outer: oneshot::Receiver>
+}
+
+impl Stream for NetworkStream {
+ type Item = network_gossip::TopicNotification;
+ type Error = ();
+
+ fn poll(&mut self) -> Poll, Self::Error> {
+ if let Some(ref mut inner) = self.inner {
+ return inner.poll();
+ }
+ match self.outer.poll() {
+ Ok(futures::Async::Ready(mut inner)) => {
+ let poll_result = inner.poll();
+ self.inner = Some(inner);
+ poll_result
+ },
+ Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
+ Err(_) => Err(())
+ }
+ }
+}
+
+/// Bridge between the underlying network service, gossiping consensus messages and Grandpa
+pub(crate) struct NetworkBridge> {
+ service: N,
+ validator: Arc>,
+}
+
+impl> NetworkBridge {
+ /// Create a new NetworkBridge to the given NetworkService
+ pub(crate) fn new(service: N, config: crate::Config) -> Self {
+ let validator = Arc::new(GossipValidator::new(config));
+ service.register_validator(validator.clone());
+ NetworkBridge { service, validator: validator }
+ }
+
+ /// Get the round messages for a round in a given set ID. These are signature-checked.
+ pub(crate) fn round_communication(
+ &self,
+ round: Round,
+ set_id: SetId,
+ voters: Arc>,
+ local_key: Option>,
+ has_voted: HasVoted,
+ ) -> (
+ impl Stream,Error=Error>,
+ impl Sink,SinkError=Error>,
+ ) {
+ self.validator.note_round(round, set_id, &self.service);
+
+ let locals = local_key.and_then(|pair| {
+ let public = pair.public();
+ let id = AuthorityId(public.0);
+ if voters.contains_key(&id) {
+ Some((pair, id))
+ } else {
+ None
+ }
+ });
+
+ let topic = round_topic::(round.0, set_id.0);
+ let incoming = self.service.messages_for(topic)
+ .filter_map(|notification| {
+ let decoded = GossipMessage::::decode(&mut ¬ification.message[..]);
+ if decoded.is_none() {
+ debug!(target: "afg", "Skipping malformed message {:?}", notification);
+ }
+ decoded
+ })
+ .and_then(move |msg| {
+ match msg {
+ GossipMessage::VoteOrPrecommit(msg) => {
+ // check signature.
+ if !voters.contains_key(&msg.message.id) {
+ debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
+ return Ok(None);
+ }
+
+ match &msg.message.message {
+ Prevote(prevote) => {
+ telemetry!(CONSENSUS_INFO; "afg.received_prevote";
+ "voter" => ?format!("{}", msg.message.id),
+ "target_number" => ?prevote.target_number,
+ "target_hash" => ?prevote.target_hash,
+ );
+ },
+ Precommit(precommit) => {
+ telemetry!(CONSENSUS_INFO; "afg.received_precommit";
+ "voter" => ?format!("{}", msg.message.id),
+ "target_number" => ?precommit.target_number,
+ "target_hash" => ?precommit.target_hash,
+ );
+ },
+ };
+
+ Ok(Some(msg.message))
+ }
+ _ => {
+ debug!(target: "afg", "Skipping unknown message type");
+ return Ok(None);
+ }
+ }
+ })
+ .filter_map(|x| x)
+ .map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")));
+
+ let (tx, out_rx) = mpsc::unbounded();
+ let outgoing = OutgoingMessages:: {
+ round: round.0,
+ set_id: set_id.0,
+ network: self.service.clone(),
+ locals,
+ sender: tx,
+ has_voted,
+ };
+
+ let out_rx = out_rx.map_err(move |()| Error::Network(
+ format!("Failed to receive on unbounded receiver for round {}", round.0)
+ ));
+
+ let incoming = incoming.select(out_rx);
+
+ (incoming, outgoing)
+ }
+
+ /// Set up the global communication streams.
+ pub(crate) fn global_communication(&self,
+ set_id: SetId,
+ voters: Arc>,
+ is_voter: bool
+ ) -> (
+ impl Stream- ), Error = Error>,
+ impl Sink), SinkError = Error>,
+ ) {
+ self.validator.note_set(set_id, &self.service);
+
+ let topic = global_topic::(set_id.0);
+ let incoming = self.service.messages_for(topic)
+ .filter_map(|notification| {
+ // this could be optimized by decoding piecewise.
+ let decoded = GossipMessage::::decode(&mut ¬ification.message[..]);
+ if decoded.is_none() {
+ trace!(target: "afg", "Skipping malformed commit message {:?}", notification);
+ }
+ decoded
+ })
+ .filter_map(move |msg| {
+ match msg {
+ GossipMessage::Commit(msg) => {
+ let round = msg.round;
+ let precommits_signed_by: Vec =
+ msg.message.auth_data.iter().map(move |(_, a)| {
+ format!("{}", a)
+ }).collect();
+ telemetry!(CONSENSUS_INFO; "afg.received_commit";
+ "contains_precommits_signed_by" => ?precommits_signed_by,
+ "target_number" => ?msg.message.target_number,
+ "target_hash" => ?msg.message.target_hash,
+ );
+ check_compact_commit::(msg.message, &*voters).map(move |c| (round.0, c))
+ },
+ _ => {
+ debug!(target: "afg", "Skipping unknown message type");
+ return None;
+ }
+ }
+ })
+ .map_err(|()| Error::Network(format!("Failed to receive message on unbounded stream")));
+
+ let outgoing = CommitsOut::::new(
+ self.service.clone(),
+ set_id.0,
+ is_voter,
+ );
+
+ (incoming, outgoing)
+ }
+
+ pub(crate) fn note_commit_finalized(&self, number: NumberFor) {
+ self.validator.note_commit_finalized(number, &self.service);
+ }
+}
+
+impl> Clone for NetworkBridge {
+ fn clone(&self) -> Self {
+ NetworkBridge {
+ service: self.service.clone(),
+ validator: Arc::clone(&self.validator),
+ }
+ }
+}
+
+fn localized_payload(round: u64, set_id: u64, message: &E) -> Vec {
+ (message, round, set_id).encode()
+}
+
+/// Type-safe wrapper around u64 when indicating that it's a round number.
+#[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Ord, Encode, Decode)]
+pub struct Round(pub u64);
+
+/// Type-safe wrapper around u64 when indicating that it's a set ID.
+#[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Ord, Encode, Decode)]
+pub struct SetId(pub u64);
+
+// check a message.
+pub(crate) fn check_message_sig(
+ message: &Message,
+ id: &AuthorityId,
+ signature: &AuthoritySignature,
+ round: u64,
+ set_id: u64,
+) -> Result<(), ()> {
+ let as_public = AuthorityId::from_raw(id.0);
+ let encoded_raw = localized_payload(round, set_id, message);
+ if ed25519::Pair::verify(signature, &encoded_raw, as_public) {
+ Ok(())
+ } else {
+ debug!(target: "afg", "Bad signature on message from {:?}", id);
+ Err(())
+ }
+}
+
+/// Whether we've voted already during a prior run of the program.
+#[derive(Decode, Encode)]
+pub(crate) enum HasVoted {
+ /// Has not voted already in this round.
+ #[codec(index = "0")]
+ No,
+ /// Has cast a proposal.
+ #[codec(index = "1")]
+ Proposed,
+ /// Has cast a prevote.
+ #[codec(index = "2")]
+ Prevoted,
+ /// Has cast a precommit (implies prevote.)
+ #[codec(index = "3")]
+ Precommitted,
+}
+
+impl HasVoted {
+ #[allow(unused)]
+ fn can_propose(&self) -> bool {
+ match *self {
+ HasVoted::No => true,
+ HasVoted::Proposed | HasVoted::Prevoted | HasVoted::Precommitted => false,
+ }
+ }
+
+ fn can_prevote(&self) -> bool {
+ match *self {
+ HasVoted::No | HasVoted::Proposed => true,
+ HasVoted::Prevoted | HasVoted::Precommitted => false,
+ }
+ }
+
+ fn can_precommit(&self) -> bool {
+ match *self {
+ HasVoted::No | HasVoted::Proposed | HasVoted::Prevoted => true,
+ HasVoted::Precommitted => false,
+ }
+ }
+}
+
+/// A sink for outgoing messages to the network.
+struct OutgoingMessages> {
+ round: u64,
+ set_id: u64,
+ locals: Option<(Arc, AuthorityId)>,
+ sender: mpsc::UnboundedSender>,
+ network: N,
+ has_voted: HasVoted,
+}
+
+impl> Sink for OutgoingMessages
+{
+ type SinkItem = Message;
+ type SinkError = Error;
+
+ fn start_send(&mut self, msg: Message) -> StartSend, Error> {
+ // only sign if we haven't voted in this round already.
+ let should_sign = match msg {
+ grandpa::Message::Prevote(_) => self.has_voted.can_prevote(),
+ grandpa::Message::Precommit(_) => self.has_voted.can_precommit(),
+ };
+
+ // when locals exist, sign messages on import
+ if let (true, &Some((ref pair, ref local_id))) = (should_sign, &self.locals) {
+ let encoded = localized_payload(self.round, self.set_id, &msg);
+ let signature = pair.sign(&encoded[..]);
+
+ let target_hash = msg.target().0.clone();
+ let signed = SignedMessage:: {
+ message: msg,
+ signature,
+ id: local_id.clone(),
+ };
+
+ let message = GossipMessage::VoteOrPrecommit(VoteOrPrecommitMessage:: {
+ message: signed.clone(),
+ round: Round(self.round),
+ set_id: SetId(self.set_id),
+ });
+
+ debug!(
+ target: "afg",
+ "Announcing block {} to peers which we voted on in round {} in set {}",
+ target_hash,
+ self.round,
+ self.set_id,
+ );
+
+ telemetry!(
+ CONSENSUS_DEBUG; "afg.announcing_blocks_to_voted_peers";
+ "block" => ?target_hash, "round" => ?self.round, "set_id" => ?self.set_id,
+ );
+
+ // announce our block hash to peers and propagate the
+ // message.
+ self.network.announce(target_hash);
+
+ let topic = round_topic::(self.round, self.set_id);
+ self.network.gossip_message(topic, message.encode(), false);
+
+ // forward the message to the inner sender.
+ let _ = self.sender.unbounded_send(signed);
+ }
+
+ Ok(AsyncSink::Ready)
+ }
+
+ fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
+
+ fn close(&mut self) -> Poll<(), Error> {
+ // ignore errors since we allow this inner sender to be closed already.
+ self.sender.close().or_else(|_| Ok(Async::Ready(())))
+ }
+}
+
+fn check_compact_commit(
+ msg: CompactCommit,
+ voters: &VoterSet,
+) -> Option> {
+ if msg.precommits.len() != msg.auth_data.len() || msg.precommits.is_empty() {
+ debug!(target: "afg", "Skipping malformed compact commit");
+ return None;
+ }
+
+ // check signatures on all contained precommits.
+ for (_, ref id) in &msg.auth_data {
+ if !voters.contains_key(id) {
+ debug!(target: "afg", "Skipping commit containing unknown voter {}", id);
+ return None;
+ }
+ }
+
+ Some(msg)
+}
+/// An output sink for commit messages.
+struct CommitsOut> {
+ network: N,
+ set_id: SetId,
+ is_voter: bool,
+ _marker: ::std::marker::PhantomData,
+}
+
+impl> CommitsOut {
+ /// Create a new commit output stream.
+ pub(crate) fn new(network: N, set_id: u64, is_voter: bool) -> Self {
+ CommitsOut {
+ network,
+ set_id: SetId(set_id),
+ is_voter,
+ _marker: Default::default(),
+ }
+ }
+}
+
+impl> Sink for CommitsOut {
+ type SinkItem = (u64, Commit);
+ type SinkError = Error;
+
+ fn start_send(&mut self, input: (u64, Commit)) -> StartSend {
+ if !self.is_voter {
+ return Ok(AsyncSink::Ready);
+ }
+
+ let (round, commit) = input;
+ let round = Round(round);
+
+ telemetry!(CONSENSUS_INFO; "afg.commit_issued";
+ "target_number" => ?commit.target_number, "target_hash" => ?commit.target_hash,
+ );
+ let (precommits, auth_data) = commit.precommits.into_iter()
+ .map(|signed| (signed.precommit, (signed.signature, signed.id)))
+ .unzip();
+
+ let compact_commit = CompactCommit:: {
+ target_hash: commit.target_hash,
+ target_number: commit.target_number,
+ precommits,
+ auth_data
+ };
+
+ let message = GossipMessage::Commit(FullCommitMessage:: {
+ round: round,
+ set_id: self.set_id,
+ message: compact_commit,
+ });
+
+ let topic = global_topic::(self.set_id.0);
+ self.network.gossip_message(topic, message.encode(), false);
+
+ Ok(AsyncSink::Ready)
+ }
+
+ fn close(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
+ fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
+}
diff --git a/core/finality-grandpa/src/environment.rs b/core/finality-grandpa/src/environment.rs
index 587762b608e0..556b4aead76e 100644
--- a/core/finality-grandpa/src/environment.rs
+++ b/core/finality-grandpa/src/environment.rs
@@ -82,7 +82,7 @@ pub(crate) struct Environment, RA> {
pub(crate) config: Config,
pub(crate) authority_set: SharedAuthoritySet