Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.

core/finality-grandpa: Gossip synced blocks via neighbor packets #3755

Closed
mxinden wants to merge 5 commits into from
Closed

core/finality-grandpa: Gossip synced blocks via neighbor packets #3755

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9ed26a2
*: Add documentation comments and questions
mxinden Oct 2, 2019
b356765
core/finality-grandpa: Add known_vote_blocks to NeighborPacket
mxinden Oct 2, 2019
87ba689
core/finality-grandpa: Pass client down to NetworkBridge
mxinden Oct 3, 2019
b6fb3f2
core/finality-grandpa: Pass client down to GossipValidator
mxinden Oct 3, 2019
dd2a75e
core/finality-grandpa: Make NeighborPacket aware of a Block for its hash
mxinden Oct 3, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions core/finality-grandpa/primitives/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,9 +53,11 @@ pub type AuthorityWeight = u64;
pub type AuthorityIndex = u64;

/// The identifier of a GRANDPA set.
// TODO: Explain what a set is. A set of notes participating in the grandpa process?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A set is a given stable set of authorities, in this case it is identified by a monotonic id.

pub type SetId = u64;

/// The round indicator.
// TODO: Explain what a round is and how it relates to a set. A round in which each participant within the set votes?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A round is a voting round of the GRANDPA protocol, rounds exist in the context of a given set, i.e. they validate with the authorities assigned to the given set id.

pub type RoundNumber = u64;

/// A scheduled change of authority set.
Expand Down
133 changes: 91 additions & 42 deletions core/finality-grandpa/src/communication/gossip.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,9 @@
//! 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.
//! If a peer is at round r, it 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.
//!
Expand Down Expand Up @@ -70,8 +70,8 @@
//!
//! The logic for issuing and tracking pending catch up requests is implemented
//! in the `GossipValidator`. A catch up request is issued anytime we see a
//! neighbor packet from a peer at a round `CATCH_UP_THRESHOLD` higher than at
//! we are.
//! neighbor packet from a peer at a round `CATCH_UP_THRESHOLD` higher than the
//! one we are at.
//!
//! ## Expiration
//!
Expand All @@ -80,7 +80,7 @@
//!
//! ## Message Validation
//!
//! We only send polite messages to peers,
//! We only send polite messages to peers.

use sr_primitives::traits::{NumberFor, Block as BlockT, Zero};
use network::consensus_gossip::{self as network_gossip, MessageIntent, ValidatorContext};
Expand All @@ -93,7 +93,7 @@ use log::{trace, debug, warn};
use futures::prelude::*;
use futures::sync::mpsc;

use crate::{environment, CatchUp, CompactCommit, SignedMessage};
use crate::{environment, CatchUp, CompactCommit, SignedMessage, BlockStatus};
use super::{cost, benefit, Round, SetId};

use std::collections::{HashMap, VecDeque};
Expand Down Expand Up @@ -184,6 +184,7 @@ impl<N: Ord> View<N> {
const KEEP_RECENT_ROUNDS: usize = 3;

/// Tracks topics we keep messages for.
// TODO: What is a topic? What is a KeepTopic? Maybe more of a TopicsToKeep?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A topic is an identifier for gossip messages, you subscribe to a given topic and you get messages for it. At the beginning of each round a new topic is created named something like "{SET_ID}-{ROUND_ID}". And there's also a global topic (per set) for commit and catch up messages.

IIRC this data structure is used to keep track of what topics we are currently interested in. It is used in this context to validate messages we might get from other peers and also to inform them about our current status (neighbor packet).

struct KeepTopics<B: BlockT> {
current_set: SetId,
rounds: VecDeque<(Round, SetId)>,
Expand Down Expand Up @@ -251,21 +252,23 @@ pub(super) enum GossipMessage<Block: BlockT> {
/// Grandpa commit message with round and set info.
Commit(FullCommitMessage<Block>),
/// A neighbor packet. Not repropagated.
Neighbor(VersionedNeighborPacket<NumberFor<Block>>),
Neighbor(VersionedNeighborPacket<Block>),
/// Grandpa catch up request message with round and set info. Not repropagated.
CatchUpRequest(CatchUpRequestMessage),
/// Grandpa catch up message with round and set info. Not repropagated.
CatchUp(FullCatchUpMessage<Block>),
}

impl<Block: BlockT> From<NeighborPacket<NumberFor<Block>>> for GossipMessage<Block> {
fn from(neighbor: NeighborPacket<NumberFor<Block>>) -> Self {
impl<Block: BlockT> From<NeighborPacket<Block>> for GossipMessage<Block> {
fn from(neighbor: NeighborPacket<Block>) -> Self {
GossipMessage::Neighbor(VersionedNeighborPacket::V1(neighbor))
}
}

/// Network level message with topic information.
#[derive(Debug, Encode, Decode)]
// TODO: Are both *prevote* and *precommit* *vote* messages? If so, this would need to be called
// PrevoteOrPrecommitMessage, right?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this can be renamed to PrevoteOrPrecommitMessage to be more explicit.

pub(super) struct VoteOrPrecommitMessage<Block: BlockT> {
/// The round this message is from.
pub(super) round: Round,
Expand All @@ -289,24 +292,35 @@ pub(super) struct FullCommitMessage<Block: BlockT> {
/// 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<N> {
pub(super) struct NeighborPacket<Block: BlockT> where
NumberFor<Block>: Ord,
{
/// The round the node is currently at.
pub(super) round: Round,
/// The set ID the node is currently at.
pub(super) set_id: SetId,
/// The highest finalizing commit observed.
pub(super) commit_finalized_height: N,
pub(super) commit_finalized_height: NumberFor<Block>,
/// Hashes of the blocks which:
///
/// - the node has synced
///
/// - have been mentioned by nodes in vote (Prevote, Precommit) messages
///
/// - within the advertised Round set in `round` field.
pub(super) known_vote_blocks: Vec<Block::Hash>,
}

/// A versioned neighbor packet.
#[derive(Debug, Encode, Decode)]
pub(super) enum VersionedNeighborPacket<N> {
pub(super) enum VersionedNeighborPacket<Block: BlockT> {
// TODO: Should we be updating this version given that this effort is a breaking change?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ideally we should add a new variant here V2 wrapping the new neighbor packet type. In principle I think it should be possible to make this backwards-compatible (but I didn't think it through tbqh.)

#[codec(index = "1")]
V1(NeighborPacket<N>),
V1(NeighborPacket<Block>),
}

impl<N> VersionedNeighborPacket<N> {
fn into_neighbor_packet(self) -> NeighborPacket<N> {
impl<Block: BlockT> VersionedNeighborPacket<Block> {
fn into_neighbor_packet(self) -> NeighborPacket<Block> {
match self {
VersionedNeighborPacket::V1(p) => p,
}
Expand Down Expand Up @@ -337,9 +351,9 @@ pub(super) struct FullCatchUpMessage<Block: BlockT> {
/// peer.
#[derive(Clone, Copy, Debug, PartialEq)]
pub(super) enum Misbehavior {
// invalid neighbor message, considering the last one.
// Invalid neighbor message, considering the last one.
InvalidViewChange,
// could not decode neighbor message. bytes-length of the packet.
// Could not decode neighbor message. Bytes-length of the packet.
UndecodablePacket(i32),
// Bad catch up message (invalid signatures).
BadCatchUpMessage {
Expand Down Expand Up @@ -398,18 +412,21 @@ impl<N> PeerInfo<N> {
}
}

/// The peers we're connected do in gossip.
struct Peers<N> {
inner: HashMap<PeerId, PeerInfo<N>>,
/// The peers we're connected to in gossip.
struct Peers<Block: BlockT> {
inner: HashMap<PeerId, PeerInfo<NumberFor<Block>>>,
}

impl<N> Default for Peers<N> {
impl<Block: BlockT> Default for Peers<Block> {
fn default() -> Self {
Peers { inner: HashMap::new() }
}
}

impl<N: Ord> Peers<N> {
impl<Block> Peers<Block> where
Block: BlockT,
NumberFor<Block>: Ord,
{
fn new_peer(&mut self, who: PeerId, roles: Roles) {
self.inner.insert(who, PeerInfo::new(roles));
}
Expand All @@ -419,8 +436,8 @@ impl<N: Ord> Peers<N> {
}

// returns a reference to the new view, if the peer is known.
fn update_peer_state(&mut self, who: &PeerId, update: NeighborPacket<N>)
-> Result<Option<&View<N>>, Misbehavior>
fn update_peer_state(&mut self, who: &PeerId, update: NeighborPacket<Block>)
-> Result<Option<&View<NumberFor<Block>>>, Misbehavior>
{
let peer = match self.inner.get_mut(who) {
None => return Ok(None),
Expand All @@ -447,7 +464,7 @@ impl<N: Ord> Peers<N> {
Ok(Some(&peer.view))
}

fn update_commit_height(&mut self, who: &PeerId, new_height: N) -> Result<(), Misbehavior> {
fn update_commit_height(&mut self, who: &PeerId, new_height: NumberFor<Block>) -> Result<(), Misbehavior> {
let peer = match self.inner.get_mut(who) {
None => return Ok(()),
Some(p) => p,
Expand All @@ -465,12 +482,13 @@ impl<N: Ord> Peers<N> {
Ok(())
}

fn peer<'a>(&'a self, who: &PeerId) -> Option<&'a PeerInfo<N>> {
fn peer<'a>(&'a self, who: &PeerId) -> Option<&'a PeerInfo<NumberFor<Block>>> {
self.inner.get(who)
}
}

#[derive(Debug, PartialEq)]
// TODO: Why not use the new-type pattern for i32 (cost)?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the cost is defined externally as i32 (probably in the peerset). There's no specific why we can't wrap it in our own type anyway.

pub(super) enum Action<H> {
// repropagate under given topic, to the given peers, applying cost/benefit to originator.
Keep(H, i32),
Expand All @@ -497,21 +515,29 @@ enum PendingCatchUp {
},
}

struct Inner<Block: BlockT> {
/// The inner lock-protected struct of a GossipValidator, enabling GossipValidator itself to be std::marker::Sync.
// TODO: Why does GossipValidator need to be Sync? Does Grandpa logic need to be parallel? I guess in this case related
// to the fact that the GossipValidator also acts as a network Validator (trait) thus is shared by two substrate
// components (network & finality-grandpa).
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Precisely. The network layer uses the validator to validate messages, while grandpa code needs to keep it up-to-date with the current view of the GRANDPA protocol (which set we're in, round, etc.)

struct Inner<Block: BlockT, Client> {
local_view: Option<View<NumberFor<Block>>>,
peers: Peers<NumberFor<Block>>,
peers: Peers<Block>,
live_topics: KeepTopics<Block>,
authorities: Vec<AuthorityId>,
config: crate::Config,
next_rebroadcast: Instant,
pending_catch_up: PendingCatchUp,
catch_up_enabled: bool,
client: Client,
// TODO: Add doc comment
/// Blocks we have seen mentioned in votes. One need to check ad-hoc whether they are available locally.
known_vote_blocks: Vec<Block::Hash>,
}

type MaybeMessage<Block> = Option<(Vec<PeerId>, NeighborPacket<NumberFor<Block>>)>;
type MaybeMessage<Block> = Option<(Vec<PeerId>, NeighborPacket<Block>)>;

impl<Block: BlockT> Inner<Block> {
fn new(config: crate::Config, catch_up_enabled: bool) -> Self {
impl<Block: BlockT, Client: BlockStatus<Block>> Inner<Block, Client> {
fn new(config: crate::Config, catch_up_enabled: bool, client: Client) -> Self {
Inner {
local_view: None,
peers: Peers::default(),
Expand All @@ -521,12 +547,17 @@ impl<Block: BlockT> Inner<Block> {
pending_catch_up: PendingCatchUp::None,
catch_up_enabled,
config,
client,
known_vote_blocks: vec![],
}
}

/// Note a round in the current set has started.
fn note_round(&mut self, round: Round) -> MaybeMessage<Block> {
{
// Given that we start a new round, reset the known vote blocks tracking blocks of the last round.
self.known_vote_blocks = vec![];

let local_view = match self.local_view {
None => return None,
Some(ref mut v) => if v.round == round {
Expand Down Expand Up @@ -632,6 +663,13 @@ impl<Block: BlockT> Inner<Block> {
return Action::Discard(cost::BAD_SIGNATURE);
}

// TODO: Adding a block in a function called 'validate' seems missleading.
//
// TODO: Is it save to give these vote and commit messages any attention at this point? Do they need to be
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At this point we have already done checked that the message is well-formed and the signature is valid. Any further validation will only happen once we have the full ancestry of the block imported now.

// further validated before we add their block hash to the list?
//
// self.known_vote_blocks.push(/* hash of the block*/);

let topic = super::round_topic::<Block>(full.round.0, full.set_id.0);
Action::Keep(topic, benefit::ROUND_MESSAGE)
}
Expand Down Expand Up @@ -850,7 +888,7 @@ impl<Block: BlockT> Inner<Block> {
(catch_up, report)
}

fn import_neighbor_message(&mut self, who: &PeerId, update: NeighborPacket<NumberFor<Block>>)
fn import_neighbor_message(&mut self, who: &PeerId, update: NeighborPacket<Block>)
-> (Vec<Block::Hash>, Action<Block::Hash>, Option<GossipMessage<Block>>, Option<Report>)
{
let update_res = self.peers.update_peer_state(who, update);
Expand All @@ -877,10 +915,16 @@ impl<Block: BlockT> Inner<Block> {

fn multicast_neighbor_packet(&self) -> MaybeMessage<Block> {
self.local_view.as_ref().map(|local_view| {
let known_vote_blocks = self.known_vote_blocks.iter().filter(|h| {
// TODO: Remove unwrap.
self.client.block_number(**h).unwrap().map(|_| true).unwrap_or(false)
}).map(|h| h.clone()).collect();

let packet = NeighborPacket {
round: local_view.round,
set_id: local_view.set_id,
commit_finalized_height: local_view.last_commit.unwrap_or(Zero::zero()),
known_vote_blocks: known_vote_blocks,
};

let peers = self.peers.inner.keys().cloned().collect();
Expand Down Expand Up @@ -921,24 +965,27 @@ impl<Block: BlockT> Inner<Block> {
}

/// A validator for GRANDPA gossip messages.
pub(super) struct GossipValidator<Block: BlockT> {
inner: parking_lot::RwLock<Inner<Block>>,
// TODO: The name `GossipValidator` seems confusing, as it does more than validating (e.g. constructing new messages).
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it has a bunch of side-effects which are not entirely obvious.

pub(super) struct GossipValidator<Block: BlockT, Client> {
// TODO: Why does all of core/grandpa need to be Sync, hence the RwLock?
Copy link
Contributor

@andresilva andresilva Oct 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's multiple components of substrate that will need to interact with GRANDPA. The network gets a GossipValidator, the import pipeline gets a GrandpaBlockImport, and there's also the GrandpaVoter which is running in some future.

inner: parking_lot::RwLock<Inner<Block, Client>>,
set_state: environment::SharedVoterSetState<Block>,
report_sender: mpsc::UnboundedSender<PeerReport>,
}

impl<Block: BlockT> GossipValidator<Block> {
impl<Block: BlockT, Client: BlockStatus<Block>> GossipValidator<Block, Client> {
/// Create a new gossip-validator. The current set is initialized to 0. If
/// `catch_up_enabled` is set to false then the validator will not issue any
/// catch up requests (useful e.g. when running just the GRANDPA observer).
pub(super) fn new(
config: crate::Config,
set_state: environment::SharedVoterSetState<Block>,
catch_up_enabled: bool,
) -> (GossipValidator<Block>, ReportStream) {
client: Client,
) -> (GossipValidator<Block, Client>, ReportStream) {
let (tx, rx) = mpsc::unbounded();
let val = GossipValidator {
inner: parking_lot::RwLock::new(Inner::new(config, catch_up_enabled)),
inner: parking_lot::RwLock::new(Inner::new(config, catch_up_enabled, client)),
set_state,
report_sender: tx,
};
Expand All @@ -948,7 +995,7 @@ impl<Block: BlockT> GossipValidator<Block> {

/// Note a round in the current set has started.
pub(super) fn note_round<F>(&self, round: Round, send_neighbor: F)
where F: FnOnce(Vec<PeerId>, NeighborPacket<NumberFor<Block>>)
where F: FnOnce(Vec<PeerId>, NeighborPacket<Block>)
{
let maybe_msg = self.inner.write().note_round(round);
if let Some((to, msg)) = maybe_msg {
Expand All @@ -959,7 +1006,7 @@ impl<Block: BlockT> GossipValidator<Block> {
/// Note that a voter set with given ID has started. Updates the current set to given
/// value and initializes the round to 0.
pub(super) fn note_set<F>(&self, set_id: SetId, authorities: Vec<AuthorityId>, send_neighbor: F)
where F: FnOnce(Vec<PeerId>, NeighborPacket<NumberFor<Block>>)
where F: FnOnce(Vec<PeerId>, NeighborPacket<Block>)
{
let maybe_msg = self.inner.write().note_set(set_id, authorities);
if let Some((to, msg)) = maybe_msg {
Expand All @@ -969,7 +1016,7 @@ impl<Block: BlockT> GossipValidator<Block> {

/// Note that we've imported a commit finalizing a given block.
pub(super) fn note_commit_finalized<F>(&self, finalized: NumberFor<Block>, send_neighbor: F)
where F: FnOnce(Vec<PeerId>, NeighborPacket<NumberFor<Block>>)
where F: FnOnce(Vec<PeerId>, NeighborPacket<Block>)
{
let maybe_msg = self.inner.write().note_commit_finalized(finalized);
if let Some((to, msg)) = maybe_msg {
Expand All @@ -986,6 +1033,7 @@ impl<Block: BlockT> GossipValidator<Block> {
let _ = self.report_sender.unbounded_send(PeerReport { who, cost_benefit });
}

// TODO: @Max this is the place where we could track voted block hashes for the current round.
pub(super) fn do_validate(&self, who: &PeerId, mut data: &[u8])
-> (Action<Block::Hash>, Vec<Block::Hash>, Option<GossipMessage<Block>>)
{
Expand Down Expand Up @@ -1037,7 +1085,7 @@ impl<Block: BlockT> GossipValidator<Block> {
}
}

impl<Block: BlockT> network_gossip::Validator<Block> for GossipValidator<Block> {
impl<Block: BlockT, Client: BlockStatus<Block> + Send + Sync> network_gossip::Validator<Block> for GossipValidator<Block, Client> {
fn new_peer(&self, context: &mut dyn ValidatorContext<Block>, who: &PeerId, roles: Roles) {
let packet = {
let mut inner = self.inner.write();
Expand All @@ -1048,6 +1096,7 @@ impl<Block: BlockT> network_gossip::Validator<Block> for GossipValidator<Block>
round: v.round,
set_id: v.set_id,
commit_finalized_height: v.last_commit.unwrap_or(Zero::zero()),
known_vote_blocks: vec![],
}
})
};
Expand Down
Loading