diff --git a/Cargo.lock b/Cargo.lock index 1aa1be64918b..4968b0131c85 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -219,6 +219,35 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7deb0a829ca7bcfaf5da70b073a8d128619259a7be8216a355e23f00763059e5" +[[package]] +name = "async-channel" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee81ba99bee79f3c8ae114ae4baa7eaa326f63447cf2ec65e4393618b63f8770" +dependencies = [ + "concurrent-queue", + "event-listener", + "futures-core", +] + +[[package]] +name = "async-io" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a0fc2017a5cca12763bb5636092a7786b52789c23c5838a392db2eb99963fd3" +dependencies = [ + "cfg-if", + "concurrent-queue", + "futures-lite", + "libc", + "once_cell", + "parking", + "socket2", + "vec-arena", + "wepoll-sys-stjepang", + "winapi 0.3.9", +] + [[package]] name = "async-std" version = "1.6.2" @@ -239,7 +268,7 @@ dependencies = [ "pin-project-lite", "pin-utils", "slab", - "smol", + "smol 0.1.18", "wasm-bindgen-futures", ] @@ -272,6 +301,12 @@ dependencies = [ "syn 1.0.33", ] +[[package]] +name = "atomic-waker" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "065374052e7df7ee4047b1160cca5e1467a12351a40b3da123c870ba0b8eda2a" + [[package]] name = "atty" version = "0.2.14" @@ -469,12 +504,13 @@ dependencies = [ [[package]] name = "blocking" -version = "0.4.6" +version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d17efb70ce4421e351d61aafd90c16a20fb5bfe339fcdc32a86816280e62ce0" +checksum = "d2468ff7bf85066b4a3678fede6fe66db31846d753ff0adfbfab2c6a6e81612b" dependencies = [ - "futures-channel", - "futures-util", + "async-channel", + "atomic-waker", + "futures-lite", "once_cell", "parking", "waker-fn", @@ -1172,6 +1208,12 @@ dependencies = [ "serde_json", ] +[[package]] +name = "event-listener" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "699d84875f1b72b4da017e6b0f77dfa88c0137f089958a88974d15938cbc2976" + [[package]] name = "exit-future" version = "0.2.0" @@ -1625,6 +1667,19 @@ version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de27142b013a8e869c14957e6d2edeef89e97c289e69d042ee3a49acd8b51789" +[[package]] +name = "futures-lite" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af0bbcb0ec905ef6ee23fab499119b5da2362b8697d66e08d1ef01a8c0d438e2" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "memchr", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.5" @@ -3033,6 +3088,12 @@ dependencies = [ "libc", ] +[[package]] +name = "maplit" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d" + [[package]] name = "matches" version = "0.1.8" @@ -3243,6 +3304,17 @@ dependencies = [ "unsigned-varint 0.4.0", ] +[[package]] +name = "multitask" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c09c35271e7dcdb5f709779111f2c8e8ab8e06c1b587c1c6a9e179d865aaa5b4" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand", +] + [[package]] name = "nalgebra" version = "0.18.1" @@ -4145,9 +4217,9 @@ checksum = "ddfc878dac00da22f8f61e7af3157988424567ab01d9920b962ef7dcbd7cd865" [[package]] name = "parking" -version = "1.0.4" +version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1efcee3c6d23b94012e240525f131c6abaa9e5eeb8f211002d93beec3b7be350" +checksum = "50d4a6da31f8144a32532fe38fe8fb439a6842e0ec633f0037f0144c14e7f907" [[package]] name = "parking_lot" @@ -4326,6 +4398,31 @@ dependencies = [ "tempfile", ] +[[package]] +name = "polkadot-availability-bitfield-distribution" +version = "0.1.0" +dependencies = [ + "assert_matches", + "bitvec", + "env_logger", + "futures 0.3.5", + "futures-timer 3.0.2", + "log 0.4.8", + "maplit", + "parity-scale-codec", + "parking_lot 0.10.2", + "polkadot-network", + "polkadot-network-bridge", + "polkadot-node-primitives", + "polkadot-node-subsystem", + "polkadot-primitives", + "sc-network", + "smol 0.2.0", + "smol-timeout", + "sp-core", + "streamunordered", +] + [[package]] name = "polkadot-availability-store" version = "0.8.19" @@ -7196,6 +7293,29 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "smol" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "346a94824d48ed7c5fc7247f3cbbf0317bdfe15fc39d08f9262609cccce61254" +dependencies = [ + "async-io", + "blocking", + "multitask", + "num_cpus", + "once_cell", +] + +[[package]] +name = "smol-timeout" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "024818c1f00b80e8171ddcfcee33860134293aa3aced60c9cbd7a5a2d41db392" +dependencies = [ + "pin-project", + "smol 0.1.18", +] + [[package]] name = "snow" version = "0.7.1" @@ -9053,6 +9173,12 @@ version = "0.2.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6454029bf181f092ad1b853286f23e2c507d8e8194d01d92da4a55c274a5508c" +[[package]] +name = "vec-arena" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17dfb54bf57c9043f4616cb03dab30eff012cc26631b797d8354b916708db919" + [[package]] name = "vec_map" version = "0.8.2" diff --git a/Cargo.toml b/Cargo.toml index 4d6b40feb704..28ad17ac4d77 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,6 +47,7 @@ members = [ "node/network/bridge", "node/network/pov-distribution", "node/network/statement-distribution", + "node/network/bitfield-distribution", "node/overseer", "node/primitives", "node/service", diff --git a/node/network/bitfield-distribution/Cargo.toml b/node/network/bitfield-distribution/Cargo.toml new file mode 100644 index 000000000000..a6dab0307127 --- /dev/null +++ b/node/network/bitfield-distribution/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "polkadot-availability-bitfield-distribution" +version = "0.1.0" +authors = ["Parity Technologies "] +edition = "2018" + +[dependencies] +futures = "0.3.5" +futures-timer = "3.0.2" +log = "0.4.8" +streamunordered = "0.5.1" +codec = { package="parity-scale-codec", version = "1.3.0" } +node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" } +polkadot-primitives = { path = "../../../primitives" } +polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" } +polkadot-network-bridge = { path = "../../network/bridge" } +polkadot-network = { path = "../../../network" } +sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" } + +[dev-dependencies] +polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = [ "test-helpers" ] } +bitvec = { version = "0.17.4", default-features = false, features = ["alloc"] } +sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" } +parking_lot = "0.10.0" +maplit = "1.0.2" +smol = "0.2.0" +smol-timeout = "0.1.0" +env_logger = "0.7.1" +assert_matches = "1.3.0" \ No newline at end of file diff --git a/node/network/bitfield-distribution/src/lib.rs b/node/network/bitfield-distribution/src/lib.rs new file mode 100644 index 000000000000..bc7c6690c614 --- /dev/null +++ b/node/network/bitfield-distribution/src/lib.rs @@ -0,0 +1,1095 @@ +// Copyright 2020 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 . + +//! The bitfield distribution +//! +//! In case this node is a validator, gossips its own signed availability bitfield +//! for a particular relay parent. +//! Independently of that, gossips on received messages from peers to other interested peers. + +use codec::{Decode, Encode}; +use futures::{channel::oneshot, FutureExt}; + +use node_primitives::{ProtocolId, View}; + +use log::{trace, warn}; +use polkadot_subsystem::messages::*; +use polkadot_subsystem::{ + FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem, SubsystemContext, SubsystemResult, +}; +use polkadot_primitives::v1::{Hash, SignedAvailabilityBitfield, SigningContext, ValidatorId}; +use sc_network::ReputationChange; +use std::collections::{HashMap, HashSet}; + +const COST_SIGNATURE_INVALID: ReputationChange = + ReputationChange::new(-100, "Bitfield signature invalid"); +const COST_VALIDATOR_INDEX_INVALID: ReputationChange = + ReputationChange::new(-100, "Bitfield validator index invalid"); +const COST_MISSING_PEER_SESSION_KEY: ReputationChange = + ReputationChange::new(-133, "Missing peer session key"); +const COST_NOT_IN_VIEW: ReputationChange = + ReputationChange::new(-51, "Not interested in that parent hash"); +const COST_MESSAGE_NOT_DECODABLE: ReputationChange = + ReputationChange::new(-100, "Not interested in that parent hash"); +const COST_PEER_DUPLICATE_MESSAGE: ReputationChange = + ReputationChange::new(-500, "Peer sent the same message multiple times"); +const GAIN_VALID_MESSAGE_FIRST: ReputationChange = + ReputationChange::new(15, "Valid message with new information"); +const GAIN_VALID_MESSAGE: ReputationChange = + ReputationChange::new(10, "Valid message"); + +/// Checked signed availability bitfield that is distributed +/// to other peers. +#[derive(Encode, Decode, Debug, Clone, PartialEq, Eq)] +pub struct BitfieldGossipMessage { + /// The relay parent this message is relative to. + pub relay_parent: Hash, + /// The actual signed availability bitfield. + pub signed_availability: SignedAvailabilityBitfield, +} + +/// Data used to track information of peers and relay parents the +/// overseer ordered us to work on. +#[derive(Default, Clone)] +struct ProtocolState { + /// track all active peers and their views + /// to determine what is relevant to them. + peer_views: HashMap, + + /// Our current view. + view: View, + + /// Additional data particular to a relay parent. + per_relay_parent: HashMap, +} + +/// Data for a particular relay parent. +#[derive(Debug, Clone, Default)] +struct PerRelayParentData { + /// Signing context for a particular relay parent. + signing_context: SigningContext, + + /// Set of validators for a particular relay parent. + validator_set: Vec, + + /// Set of validators for a particular relay parent for which we + /// received a valid `BitfieldGossipMessage`. + /// Also serves as the list of known messages for peers connecting + /// after bitfield gossips were already received. + one_per_validator: HashMap, + + /// Avoid duplicate message transmission to our peers. + message_sent_to_peer: HashMap>, + + /// Track messages that were already received by a peer + /// to prevent flooding. + message_received_from_peer: HashMap>, +} + +impl PerRelayParentData { + /// Determines if that particular message signed by a validator is needed by the given peer. + fn message_from_validator_needed_by_peer( + &self, + peer: &PeerId, + validator: &ValidatorId, + ) -> bool { + if let Some(set) = self.message_sent_to_peer.get(peer) { + !set.contains(validator) + } else { + false + } + } +} + +fn network_update_message(n: NetworkBridgeEvent) -> AllMessages { + AllMessages::BitfieldDistribution(BitfieldDistributionMessage::NetworkBridgeUpdate(n)) +} + +/// The bitfield distribution subsystem. +pub struct BitfieldDistribution; + +impl BitfieldDistribution { + /// The protocol identifier for bitfield distribution. + const PROTOCOL_ID: ProtocolId = *b"bitd"; + + /// Start processing work as passed on from the Overseer. + async fn run(mut ctx: Context) -> SubsystemResult<()> + where + Context: SubsystemContext, + { + // startup: register the network protocol with the bridge. + ctx.send_message(AllMessages::NetworkBridge( + NetworkBridgeMessage::RegisterEventProducer(Self::PROTOCOL_ID, network_update_message), + )) + .await?; + + // work: process incoming messages from the overseer and process accordingly. + let mut state = ProtocolState::default(); + loop { + let message = ctx.recv().await?; + match message { + FromOverseer::Communication { + msg: BitfieldDistributionMessage::DistributeBitfield(hash, signed_availability), + } => { + trace!(target: "bitd", "Processing DistributeBitfield"); + handle_bitfield_distribution(&mut ctx, &mut state, hash, signed_availability) + .await?; + } + FromOverseer::Communication { + msg: BitfieldDistributionMessage::NetworkBridgeUpdate(event), + } => { + trace!(target: "bitd", "Processing NetworkMessage"); + // a network message was received + if let Err(e) = handle_network_msg(&mut ctx, &mut state, event).await { + warn!(target: "bitd", "Failed to handle incomming network messages: {:?}", e); + } + } + FromOverseer::Signal(OverseerSignal::StartWork(relay_parent)) => { + trace!(target: "bitd", "Start {:?}", relay_parent); + // query basic system parameters once + let (validator_set, signing_context) = + query_basics(&mut ctx, relay_parent).await?; + + let _ = state.per_relay_parent.insert( + relay_parent, + PerRelayParentData { + signing_context, + validator_set, + ..Default::default() + }, + ); + } + FromOverseer::Signal(OverseerSignal::StopWork(relay_parent)) => { + trace!(target: "bitd", "Stop {:?}", relay_parent); + // defer the cleanup to the view change + } + FromOverseer::Signal(OverseerSignal::Conclude) => { + trace!(target: "bitd", "Conclude"); + return Ok(()); + } + } + } + } +} + +/// Modify the reputation of a peer based on its behaviour. +async fn modify_reputation( + ctx: &mut Context, + peer: PeerId, + rep: ReputationChange, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + trace!(target: "bitd", "Reputation change of {:?} for peer {:?}", rep, peer); + ctx.send_message(AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep), + )) + .await +} + +/// Distribute a given valid and signature checked bitfield message. +/// +/// For this variant the source is this node. +async fn handle_bitfield_distribution( + ctx: &mut Context, + state: &mut ProtocolState, + relay_parent: Hash, + signed_availability: SignedAvailabilityBitfield, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + // Ignore anything the overseer did not tell this subsystem to work on + let mut job_data = state.per_relay_parent.get_mut(&relay_parent); + let job_data: &mut _ = if let Some(ref mut job_data) = job_data { + job_data + } else { + trace!( + target: "bitd", + "Not supposed to work on relay parent {} related data", + relay_parent + ); + + return Ok(()); + }; + let validator_set = &job_data.validator_set; + if validator_set.is_empty() { + trace!(target: "bitd", "Validator set for {:?} is empty", relay_parent); + return Ok(()); + } + + let validator_index = signed_availability.validator_index() as usize; + let validator = if let Some(validator) = validator_set.get(validator_index) { + validator.clone() + } else { + trace!(target: "bitd", "Could not find a validator for index {}", validator_index); + return Ok(()); + }; + + let peer_views = &mut state.peer_views; + let msg = BitfieldGossipMessage { + relay_parent, + signed_availability, + }; + + relay_message(ctx, job_data, peer_views, validator, msg).await?; + + Ok(()) +} + +/// Distribute a given valid and signature checked bitfield message. +/// +/// Can be originated by another subsystem or received via network from another peer. +async fn relay_message( + ctx: &mut Context, + job_data: &mut PerRelayParentData, + peer_views: &mut HashMap, + validator: ValidatorId, + message: BitfieldGossipMessage, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + // notify the overseer about a new and valid signed bitfield + ctx.send_message(AllMessages::Provisioner( + ProvisionerMessage::ProvisionableData(ProvisionableData::Bitfield( + message.relay_parent.clone(), + message.signed_availability.clone(), + )), + )) + .await?; + + let message_sent_to_peer = &mut (job_data.message_sent_to_peer); + + // pass on the bitfield distribution to all interested peers + let interested_peers = peer_views + .iter() + .filter_map(|(peer, view)| { + // check interest in the peer in this message's relay parent + if view.contains(&message.relay_parent) { + // track the message as sent for this peer + message_sent_to_peer + .entry(peer.clone()) + .or_default() + .insert(validator.clone()); + + Some(peer.clone()) + } else { + None + } + }) + .collect::>(); + + if interested_peers.is_empty() { + trace!( + target: "bitd", + "No peers are interested in gossip for relay parent {:?}", + message.relay_parent + ); + } else { + ctx.send_message(AllMessages::NetworkBridge( + NetworkBridgeMessage::SendMessage( + interested_peers, + BitfieldDistribution::PROTOCOL_ID, + message.encode(), + ), + )) + .await?; + } + Ok(()) +} + +/// Handle an incoming message from a peer. +async fn process_incoming_peer_message( + ctx: &mut Context, + state: &mut ProtocolState, + origin: PeerId, + message: BitfieldGossipMessage, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + // we don't care about this, not part of our view + if !state.view.contains(&message.relay_parent) { + return modify_reputation(ctx, origin, COST_NOT_IN_VIEW).await; + } + + // Ignore anything the overseer did not tell this subsystem to work on + let mut job_data = state.per_relay_parent.get_mut(&message.relay_parent); + let job_data: &mut _ = if let Some(ref mut job_data) = job_data { + job_data + } else { + return modify_reputation(ctx, origin, COST_NOT_IN_VIEW).await; + }; + + let validator_set = &job_data.validator_set; + if validator_set.is_empty() { + trace!( + target: "bitd", + "Validator set for relay parent {:?} is empty", + &message.relay_parent + ); + return modify_reputation(ctx, origin, COST_MISSING_PEER_SESSION_KEY).await; + } + + // Use the (untrusted) validator index provided by the signed payload + // and see if that one actually signed the availability bitset. + let signing_context = job_data.signing_context.clone(); + let validator_index = message.signed_availability.validator_index() as usize; + let validator = if let Some(validator) = validator_set.get(validator_index) { + validator.clone() + } else { + return modify_reputation(ctx, origin, COST_VALIDATOR_INDEX_INVALID).await; + }; + + // Check if the peer already sent us a message for the validator denoted in the message earlier. + // Must be done after validator index verification, in order to avoid storing an unbounded + // number of set entries. + let received_set = job_data + .message_received_from_peer + .entry(origin.clone()) + .or_default(); + + if !received_set.contains(&validator) { + received_set.insert(validator.clone()); + } else { + return modify_reputation(ctx, origin, COST_PEER_DUPLICATE_MESSAGE).await; + }; + + if message + .signed_availability + .check_signature(&signing_context, &validator) + .is_ok() + { + let one_per_validator = &mut (job_data.one_per_validator); + + // only relay_message a message of a validator once + if one_per_validator.get(&validator).is_some() { + trace!( + target: "bitd", + "Already received a message for validator at index {}", + validator_index + ); + modify_reputation(ctx, origin, GAIN_VALID_MESSAGE).await?; + return Ok(()); + } + one_per_validator.insert(validator.clone(), message.clone()); + + relay_message(ctx, job_data, &mut state.peer_views, validator, message).await?; + + modify_reputation(ctx, origin, GAIN_VALID_MESSAGE_FIRST).await + } else { + modify_reputation(ctx, origin, COST_SIGNATURE_INVALID).await + } +} +/// Deal with network bridge updates and track what needs to be tracked +/// which depends on the message type received. +async fn handle_network_msg( + ctx: &mut Context, + state: &mut ProtocolState, + bridge_message: NetworkBridgeEvent, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + match bridge_message { + NetworkBridgeEvent::PeerConnected(peerid, _role) => { + // insert if none already present + state.peer_views.entry(peerid).or_default(); + } + NetworkBridgeEvent::PeerDisconnected(peerid) => { + // get rid of superfluous data + state.peer_views.remove(&peerid); + } + NetworkBridgeEvent::PeerViewChange(peerid, view) => { + handle_peer_view_change(ctx, state, peerid, view).await?; + } + NetworkBridgeEvent::OurViewChange(view) => { + handle_our_view_change(state, view)?; + } + NetworkBridgeEvent::PeerMessage(remote, bytes) => { + if let Ok(gossiped_bitfield) = BitfieldGossipMessage::decode(&mut (bytes.as_slice())) { + trace!(target: "bitd", "Received bitfield gossip from peer {:?}", &remote); + process_incoming_peer_message(ctx, state, remote, gossiped_bitfield).await?; + } else { + modify_reputation(ctx, remote, COST_MESSAGE_NOT_DECODABLE).await?; + } + } + } + Ok(()) +} + +/// Handle the changes necassary when our view changes. +fn handle_our_view_change(state: &mut ProtocolState, view: View) -> SubsystemResult<()> { + let old_view = std::mem::replace(&mut (state.view), view); + + for added in state.view.difference(&old_view) { + if !state.per_relay_parent.contains_key(&added) { + warn!( + target: "bitd", + "Our view contains {} but the overseer never told use we should work on this", + &added + ); + } + } + for removed in old_view.difference(&state.view) { + // cleanup relay parents we are not interested in any more + let _ = state.per_relay_parent.remove(&removed); + } + Ok(()) +} + + +// Send the difference between two views which were not sent +// to that particular peer. +async fn handle_peer_view_change( + ctx: &mut Context, + state: &mut ProtocolState, + origin: PeerId, + view: View, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + let current = state.peer_views.entry(origin.clone()).or_default(); + + let delta_vec: Vec = (*current).difference(&view).cloned().collect(); + + *current = view; + + // Send all messages we've seen before and the peer is now interested + // in to that peer. + + let delta_set: Vec<(ValidatorId, BitfieldGossipMessage)> = delta_vec + .into_iter() + .filter_map(|new_relay_parent_interest| { + if let Some(job_data) = (&*state).per_relay_parent.get(&new_relay_parent_interest) { + // Send all jointly known messages for a validator (given the current relay parent) + // to the peer `origin`... + let one_per_validator = job_data.one_per_validator.clone(); + let origin = origin.clone(); + Some( + one_per_validator + .into_iter() + .filter(move |(validator, _message)| { + // ..except for the ones the peer already has + job_data.message_from_validator_needed_by_peer(&origin, validator) + }), + ) + } else { + // A relay parent is in the peers view, which is not in ours, ignore those. + None + } + }) + .flatten() + .collect(); + + for (validator, message) in delta_set.into_iter() { + send_tracked_gossip_message(ctx, state, origin.clone(), validator, message).await?; + } + + Ok(()) +} + +/// Send a gossip message and track it in the per relay parent data. +async fn send_tracked_gossip_message( + ctx: &mut Context, + state: &mut ProtocolState, + dest: PeerId, + validator: ValidatorId, + message: BitfieldGossipMessage, +) -> SubsystemResult<()> +where + Context: SubsystemContext, +{ + let job_data = if let Some(job_data) = state.per_relay_parent.get_mut(&message.relay_parent) { + job_data + } else { + return Ok(()); + }; + + let message_sent_to_peer = &mut (job_data.message_sent_to_peer); + message_sent_to_peer + .entry(dest.clone()) + .or_default() + .insert(validator.clone()); + + ctx.send_message(AllMessages::NetworkBridge( + NetworkBridgeMessage::SendMessage( + vec![dest], + BitfieldDistribution::PROTOCOL_ID, + message.encode(), + ), + )) + .await?; + + Ok(()) +} + +impl Subsystem for BitfieldDistribution +where + C: SubsystemContext + Sync + Send, +{ + fn start(self, ctx: C) -> SpawnedSubsystem { + SpawnedSubsystem { + name: "bitfield-distribution", + future: Box::pin(async move { Self::run(ctx) }.map(|_| ())), + } + } +} + +/// Query our validator set and signing context for a particular relay parent. +async fn query_basics( + ctx: &mut Context, + relay_parent: Hash, +) -> SubsystemResult<(Vec, SigningContext)> +where + Context: SubsystemContext, +{ + let (validators_tx, validators_rx) = oneshot::channel(); + let (signing_tx, signing_rx) = oneshot::channel(); + + let query_validators = AllMessages::RuntimeApi(RuntimeApiMessage::Request( + relay_parent.clone(), + RuntimeApiRequest::Validators(validators_tx), + )); + + let query_signing = AllMessages::RuntimeApi(RuntimeApiMessage::Request( + relay_parent.clone(), + RuntimeApiRequest::SigningContext(signing_tx), + )); + + ctx.send_messages(std::iter::once(query_validators).chain(std::iter::once(query_signing))) + .await?; + + Ok((validators_rx.await?, signing_rx.await?)) +} + +#[cfg(test)] +mod test { + use super::*; + use bitvec::bitvec; + use futures::executor; + use maplit::hashmap; + use polkadot_primitives::v1::{Signed, ValidatorPair, AvailabilityBitfield}; + use polkadot_subsystem::test_helpers::make_subsystem_context; + use smol_timeout::TimeoutExt; + use sp_core::crypto::Pair; + use std::time::Duration; + use assert_matches::assert_matches; + + macro_rules! view { + ( $( $hash:expr ),* $(,)? ) => [ + View(vec![ $( $hash.clone() ),* ]) + ]; + } + + macro_rules! peers { + ( $( $peer:expr ),* $(,)? ) => [ + vec![ $( $peer.clone() ),* ] + ]; + } + + macro_rules! launch { + ($fut:expr) => { + $fut + .timeout(Duration::from_millis(10)) + .await + .expect("10ms is more than enough for sending messages.") + .expect("Error values should really never occur.") + }; + } + + /// A very limited state, only interested in the relay parent of the + /// given message, which must be signed by `validator` and a set of peers + /// which are also only interested in that relay parent. + fn prewarmed_state( + validator: ValidatorId, + signing_context: SigningContext, + known_message: BitfieldGossipMessage, + peers: Vec, + ) -> ProtocolState { + let relay_parent = known_message.relay_parent.clone(); + ProtocolState { + per_relay_parent: hashmap! { + relay_parent.clone() => + PerRelayParentData { + signing_context, + validator_set: vec![validator.clone()], + one_per_validator: hashmap! { + validator.clone() => known_message.clone(), + }, + message_received_from_peer: hashmap!{}, + message_sent_to_peer: hashmap!{}, + }, + }, + peer_views: peers + .into_iter() + .map(|peer| (peer, view!(relay_parent))) + .collect(), + view: view!(relay_parent), + } + } + + fn state_with_view(view: View, relay_parent: Hash) -> (ProtocolState, SigningContext, ValidatorPair) { + let mut state = ProtocolState::default(); + + let (validator_pair, _seed) = ValidatorPair::generate(); + let validator = validator_pair.public(); + + let signing_context = SigningContext { + session_index: 1, + parent_hash: relay_parent.clone(), + }; + + state.per_relay_parent = view.0.iter().map(|relay_parent| {( + relay_parent.clone(), + PerRelayParentData { + signing_context: signing_context.clone(), + validator_set: vec![validator.clone()], + one_per_validator: hashmap!{}, + message_received_from_peer: hashmap!{}, + message_sent_to_peer: hashmap!{}, + }) + }).collect(); + + state.view = view; + + (state, signing_context, validator_pair) + } + + #[test] + fn receive_invalid_signature() { + let _ = env_logger::builder() + .filter(None, log::LevelFilter::Trace) + .is_test(true) + .try_init(); + + let hash_a: Hash = [0; 32].into(); + + let peer_a = PeerId::random(); + let peer_b = PeerId::random(); + assert_ne!(peer_a, peer_b); + + let signing_context = SigningContext { + session_index: 1, + parent_hash: hash_a.clone(), + }; + + // validator 0 key pair + let (validator_pair, _seed) = ValidatorPair::generate(); + let validator = validator_pair.public(); + + // another validator not part of the validatorset + let (mallicious, _seed) = ValidatorPair::generate(); + + let payload = AvailabilityBitfield(bitvec![bitvec::order::Lsb0, u8; 1u8; 32]); + let signed = + Signed::::sign(payload, &signing_context, 0, &mallicious); + + let msg = BitfieldGossipMessage { + relay_parent: hash_a.clone(), + signed_availability: signed.clone(), + }; + + let pool = sp_core::testing::SpawnBlockingExecutor::new(); + let (mut ctx, mut handle) = + make_subsystem_context::(pool); + + let mut state = prewarmed_state( + validator.clone(), + signing_context.clone(), + msg.clone(), + vec![peer_b.clone()], + ); + + executor::block_on(async move { + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + // reputation change due to invalid validator index + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, COST_SIGNATURE_INVALID) + } + ); + }); + } + + #[test] + fn receive_invalid_validator_index() { + let _ = env_logger::builder() + .filter(None, log::LevelFilter::Trace) + .is_test(true) + .try_init(); + + let hash_a: Hash = [0; 32].into(); + let hash_b: Hash = [1; 32].into(); // other + + let peer_a = PeerId::random(); + let peer_b = PeerId::random(); + assert_ne!(peer_a, peer_b); + + // validator 0 key pair + let (mut state, signing_context, validator_pair) = + state_with_view(view![hash_a, hash_b], hash_a.clone()); + + state.peer_views.insert(peer_b.clone(), view![hash_a]); + + let payload = AvailabilityBitfield(bitvec![bitvec::order::Lsb0, u8; 1u8; 32]); + let signed = + Signed::::sign(payload, &signing_context, 42, &validator_pair); + + let msg = BitfieldGossipMessage { + relay_parent: hash_a.clone(), + signed_availability: signed.clone(), + }; + + let pool = sp_core::testing::SpawnBlockingExecutor::new(); + let (mut ctx, mut handle) = + make_subsystem_context::(pool); + + executor::block_on(async move { + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + // reputation change due to invalid validator index + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, COST_VALIDATOR_INDEX_INVALID) + } + ); + }); + } + + #[test] + fn receive_duplicate_messages() { + let _ = env_logger::builder() + .filter(None, log::LevelFilter::Trace) + .is_test(true) + .try_init(); + + let hash_a: Hash = [0; 32].into(); + let hash_b: Hash = [1; 32].into(); + + let peer_a = PeerId::random(); + let peer_b = PeerId::random(); + assert_ne!(peer_a, peer_b); + + // validator 0 key pair + let (mut state, signing_context, validator_pair) = + state_with_view(view![hash_a, hash_b], hash_a.clone()); + + // create a signed message by validator 0 + let payload = AvailabilityBitfield(bitvec![bitvec::order::Lsb0, u8; 1u8; 32]); + let signed_bitfield = + Signed::::sign(payload, &signing_context, 0, &validator_pair); + + let msg = BitfieldGossipMessage { + relay_parent: hash_a.clone(), + signed_availability: signed_bitfield.clone(), + }; + + let pool = sp_core::testing::SpawnBlockingExecutor::new(); + let (mut ctx, mut handle) = + make_subsystem_context::(pool); + + executor::block_on(async move { + // send a first message + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + // none of our peers has any interest in any messages + // so we do not receive a network send type message here + // but only the one for the next subsystem + assert_matches!( + handle.recv().await, + AllMessages::Provisioner(ProvisionerMessage::ProvisionableData( + ProvisionableData::Bitfield(hash, signed) + )) => { + assert_eq!(hash, hash_a); + assert_eq!(signed, signed_bitfield) + } + ); + + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, GAIN_VALID_MESSAGE_FIRST) + } + ); + + // let peer A send the same message again + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_a.clone(), msg.encode()), + )); + + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_a); + assert_eq!(rep, GAIN_VALID_MESSAGE) + } + ); + + // let peer B send the initial message again + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, COST_PEER_DUPLICATE_MESSAGE) + } + ); + }); + } + #[test] + fn changing_view() { + let _ = env_logger::builder() + .filter(None, log::LevelFilter::Trace) + .is_test(true) + .try_init(); + + let hash_a: Hash = [0; 32].into(); + let hash_b: Hash = [1; 32].into(); + + let peer_a = PeerId::random(); + let peer_b = PeerId::random(); + assert_ne!(peer_a, peer_b); + + // validator 0 key pair + let (mut state, signing_context, validator_pair) = state_with_view(view![hash_a, hash_b], hash_a.clone()); + + // create a signed message by validator 0 + let payload = AvailabilityBitfield(bitvec![bitvec::order::Lsb0, u8; 1u8; 32]); + let signed_bitfield = + Signed::::sign(payload, &signing_context, 0, &validator_pair); + + let msg = BitfieldGossipMessage { + relay_parent: hash_a.clone(), + signed_availability: signed_bitfield.clone(), + }; + + let pool = sp_core::testing::SpawnBlockingExecutor::new(); + let (mut ctx, mut handle) = + make_subsystem_context::(pool); + + executor::block_on(async move { + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerConnected(peer_b.clone(), ObservedRole::Full), + )); + + // make peer b interested + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerViewChange(peer_b.clone(), view![hash_a, hash_b]), + )); + + assert!(state.peer_views.contains_key(&peer_b)); + + // recv a first message from the network + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + // gossip to the overseer + assert_matches!( + handle.recv().await, + AllMessages::Provisioner(ProvisionerMessage::ProvisionableData( + ProvisionableData::Bitfield(hash, signed) + )) => { + assert_eq!(hash, hash_a); + assert_eq!(signed, signed_bitfield) + } + ); + + // gossip to the network + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge(NetworkBridgeMessage::SendMessage ( + peers, proto, bytes + )) => { + assert_eq!(peers, peers![peer_b]); + assert_eq!(proto, BitfieldDistribution::PROTOCOL_ID); + assert_eq!(bytes, msg.encode()); + } + ); + + // reputation change for peer B + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, GAIN_VALID_MESSAGE_FIRST) + } + ); + + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerViewChange(peer_b.clone(), view![]), + )); + + assert!(state.peer_views.contains_key(&peer_b)); + assert_eq!( + state.peer_views.get(&peer_b).expect("Must contain value for peer B"), + &view![] + ); + + // on rx of the same message, since we are not interested, + // should give penalty + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_b.clone(), msg.encode()), + )); + + // reputation change for peer B + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_b); + assert_eq!(rep, COST_PEER_DUPLICATE_MESSAGE) + } + ); + + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerDisconnected(peer_b.clone()), + )); + + // we are not interested in any peers at all anymore + state.view = view![]; + + // on rx of the same message, since we are not interested, + // should give penalty + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_a.clone(), msg.encode()), + )); + + // reputation change for peer B + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_a); + assert_eq!(rep, COST_NOT_IN_VIEW) + } + ); + + }); + } + + + #[test] + fn invalid_peer_message() { + let _ = env_logger::builder() + .filter(None, log::LevelFilter::Trace) + .is_test(true) + .try_init(); + + let hash_a: Hash = [0; 32].into(); + let peer_a = PeerId::random(); + + // validator 0 key pair + let (mut state, _signing_context, _validator_pair) = state_with_view(view![], hash_a.clone()); + + let pool = sp_core::testing::SpawnBlockingExecutor::new(); + let (mut ctx, mut handle) = + make_subsystem_context::(pool); + + executor::block_on(async move { + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerConnected(peer_a.clone(), ObservedRole::Full), + )); + + // make peer b interested + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerViewChange(peer_a.clone(), view![hash_a]), + )); + + assert!(state.peer_views.contains_key(&peer_a)); + + // recv a first message from the network + launch!(handle_network_msg( + &mut ctx, + &mut state, + NetworkBridgeEvent::PeerMessage(peer_a.clone(), b"00AaBbCcDdEeFf".to_vec()), + )); + + // reputation change for peer A + assert_matches!( + handle.recv().await, + AllMessages::NetworkBridge( + NetworkBridgeMessage::ReportPeer(peer, rep) + ) => { + assert_eq!(peer, peer_a); + assert_eq!(rep, COST_MESSAGE_NOT_DECODABLE); + } + ); + + }); + } +} diff --git a/node/subsystem/src/messages.rs b/node/subsystem/src/messages.rs index 2040b413488d..a6b7859c38ef 100644 --- a/node/subsystem/src/messages.rs +++ b/node/subsystem/src/messages.rs @@ -34,7 +34,6 @@ use polkadot_primitives::v1::{ use polkadot_node_primitives::{ MisbehaviorReport, SignedFullStatement, View, ProtocolId, ValidationResult, }; - use std::sync::Arc; pub use sc_network::{ObservedRole, ReputationChange, PeerId}; diff --git a/roadmap/implementers-guide/src/node/availability/bitfield-distribution.md b/roadmap/implementers-guide/src/node/availability/bitfield-distribution.md index 97a5c14be3da..528b5f9d1d74 100644 --- a/roadmap/implementers-guide/src/node/availability/bitfield-distribution.md +++ b/roadmap/implementers-guide/src/node/availability/bitfield-distribution.md @@ -6,16 +6,30 @@ Validators vote on the availability of a backed candidate by issuing signed bitf `ProtocolId`: `b"bitd"` -Input: [`BitfieldDistributionMessage`](../../types/overseer-protocol.md#bitfield-distribution-message) +Input: +[`BitfieldDistributionMessage`](../../types/overseer-protocol.md#bitfield-distribution-message) which are gossiped to all peers, no matter if validator or not. + Output: -- `NetworkBridge::RegisterEventProducer(ProtocolId)` -- `NetworkBridge::SendMessage([PeerId], ProtocolId, Bytes)` -- `NetworkBridge::ReportPeer(PeerId, cost_or_benefit)` -- `BlockAuthorshipProvisioning::Bitfield(relay_parent, SignedAvailabilityBitfield)` +- `NetworkBridge::RegisterEventProducer(ProtocolId)` in order to register ourself as an event provider for the protocol. +- `NetworkBridge::SendMessage([PeerId], ProtocolId, Bytes)` gossip a verified incoming bitfield on to interested subsystems within this validator node. +- `NetworkBridge::ReportPeer(PeerId, cost_or_benefit)` improve or penalize the reputation of peers based on the messages that are received relative to the current view. +- `ProvisionerMessage::ProvisionableData(ProvisionableData::Bitfield(relay_parent, SignedAvailabilityBitfield))` pass + on the bitfield to the other submodules via the overseer. ## Functionality -This is implemented as a gossip system. Register a [network bridge](../utility/network-bridge.md) event producer on startup and track peer connection, view change, and disconnection events. Only accept bitfields relevant to our current view and only distribute bitfields to other peers when relevant to their most recent view. Check bitfield signatures in this subsystem and accept and distribute only one bitfield per validator. +This is implemented as a gossip system. Register a [network bridge](../utility/network-bridge.md) event producer on startup. + +It is necessary to track peer connection, view change, and disconnection events, in order to maintain an index of which peers are interested in which relay parent bitfields. + + +Before gossiping incoming bitfields, they must be checked to be signed by one of the validators +of the validator set relevant to the current relay parent. +Only accept bitfields relevant to our current view and only distribute bitfields to other peers when relevant to their most recent view. +Accept and distribute only one bitfield per validator. + When receiving a bitfield either from the network or from a `DistributeBitfield` message, forward it along to the block authorship (provisioning) subsystem for potential inclusion in a block. + +Peers connecting after a set of valid bitfield gossip messages was received, those messages must be cached and sent upon connection of new peers or re-connecting peers.