Beefy on-demand justifications as a custom RequestResponse protocol (p…

…aritytech#12124)

* client/beefy: create communication module and move gossip there

* client/beefy: move beefy_protocol_name module to communication

* client/beefy: move notification module under communication

* client/beefy: add incoming request_response protocol handler

* client/beefy: keep track of connected peers and their progress

* client/beefy: add logic for generating Justif requests

* client/beefy: cancel outdated on-demand justification requests

* try Andre's suggestion for JustificationEngine

* justif engine add justifs validation

* client/beefy: impl OnDemandJustificationsEngine async next()

* move beefy proto name test

* client/beefy: initialize OnDemandJustificationsEngine

* client/tests: allow for custom req-resp protocols

* client/beefy: on-demand-justif: implement simple peer selection strategy

* client/beefy: fix voter initialization

Fix corner case where voter gets a single burst of finality
notifications just when it starts.

The notification stream was consumed by "wait_for_pallet" logic,
then main loop would subscribe to finality notifications, but by that
time some notifications might've been lost.

Fix this by subscribing the main loop to notifications before waiting
for pallet to become available. Share the same stream with the main loop
so that notifications for blocks before pallet available are ignored,
while _all_ notifications after pallet available are processed.

Add regression test for this.

Signed-off-by: acatangiu <adrian@parity.io>

* client/beefy: make sure justif requests are always out for mandatory blocks

* client/beefy: add test for on-demand justifications sync

* client/beefy: tweak main loop event processing order

* client/beefy: run on-demand-justif-handler under same async task as voter

* client/beefy: add test for known-peers

* client/beefy: reorg request-response module

* client/beefy: add issue references for future work todos

* client/beefy: consolidate on-demand-justifications engine state machine

Signed-off-by: acatangiu <adrian@parity.io>

* client/beefy: fix for polkadot companion

* client/beefy: implement review suggestions

* cargo fmt and clippy

* fix merge damage

* fix rust-doc

* fix merge damage

* fix merge damage

* client/beefy: add test for justif proto name

Signed-off-by: acatangiu <adrian@parity.io>

client/beefy/rpc/src/lib.rs

@@ -35,7 +35,9 @@ use jsonrpsee::{
 };
 use log::warn;
 
-use beefy_gadget::notification::{BeefyBestBlockStream, BeefyVersionedFinalityProofStream};
+use beefy_gadget::communication::notification::{
+	BeefyBestBlockStream, BeefyVersionedFinalityProofStream,
+};
 
 mod notification;
 
@@ -165,8 +167,8 @@ mod tests {
 	use super::*;
 
 	use beefy_gadget::{
+		communication::notification::BeefyVersionedFinalityProofSender,
 		justification::BeefyVersionedFinalityProof,
-		notification::{BeefyBestBlockStream, BeefyVersionedFinalityProofSender},
 	};
 	use beefy_primitives::{known_payload_ids, Payload, SignedCommitment};
 	use codec::{Decode, Encode};

client/beefy/src/gossip.rs → client/beefy/src/communication/gossip.rs

@@ -16,7 +16,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use std::{collections::BTreeMap, time::Duration};
+use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
 use sc_network::PeerId;
 use sc_network_gossip::{MessageIntent, ValidationResult, Validator, ValidatorContext};
@@ -28,13 +28,12 @@ use log::{debug, trace};
 use parking_lot::{Mutex, RwLock};
 use wasm_timer::Instant;
 
+use crate::{communication::peers::KnownPeers, keystore::BeefyKeystore};
 use beefy_primitives::{
 	crypto::{Public, Signature},
 	VoteMessage,
 };
 
-use crate::keystore::BeefyKeystore;
-
 // Timeout for rebroadcasting messages.
 const REBROADCAST_AFTER: Duration = Duration::from_secs(60 * 5);
 
@@ -103,17 +102,19 @@ where
 	topic: B::Hash,
 	known_votes: RwLock<KnownVotes<B>>,
 	next_rebroadcast: Mutex<Instant>,
+	known_peers: Arc<Mutex<KnownPeers<B>>>,
 }
 
 impl<B> GossipValidator<B>
 where
 	B: Block,
 {
-	pub fn new() -> GossipValidator<B> {
+	pub fn new(known_peers: Arc<Mutex<KnownPeers<B>>>) -> GossipValidator<B> {
 		GossipValidator {
 			topic: topic::<B>(),
 			known_votes: RwLock::new(KnownVotes::new()),
 			next_rebroadcast: Mutex::new(Instant::now() + REBROADCAST_AFTER),
+			known_peers,
 		}
 	}
 
@@ -165,6 +166,7 @@ where
 
 			if BeefyKeystore::verify(&msg.id, &msg.signature, &msg.commitment.encode()) {
 				self.known_votes.write().add_known(&round, msg_hash);
+				self.known_peers.lock().note_vote_for(*sender, round);
 				return ValidationResult::ProcessAndKeep(self.topic)
 			} else {
 				// TODO: report peer
@@ -271,7 +273,7 @@ mod tests {
 
 	#[test]
 	fn note_and_drop_round_works() {
-		let gv = GossipValidator::<Block>::new();
+		let gv = GossipValidator::<Block>::new(Arc::new(Mutex::new(KnownPeers::new())));
 
 		gv.note_round(1u64);
 
@@ -298,7 +300,7 @@ mod tests {
 
 	#[test]
 	fn note_same_round_twice() {
-		let gv = GossipValidator::<Block>::new();
+		let gv = GossipValidator::<Block>::new(Arc::new(Mutex::new(KnownPeers::new())));
 
 		gv.note_round(3u64);
 		gv.note_round(7u64);
@@ -355,7 +357,7 @@ mod tests {
 
 	#[test]
 	fn should_avoid_verifying_signatures_twice() {
-		let gv = GossipValidator::<Block>::new();
+		let gv = GossipValidator::<Block>::new(Arc::new(Mutex::new(KnownPeers::new())));
 		let sender = sc_network::PeerId::random();
 		let mut context = TestContext;
 
@@ -391,7 +393,7 @@ mod tests {
 
 	#[test]
 	fn messages_allowed_and_expired() {
-		let gv = GossipValidator::<Block>::new();
+		let gv = GossipValidator::<Block>::new(Arc::new(Mutex::new(KnownPeers::new())));
 		let sender = sc_network::PeerId::random();
 		let topic = Default::default();
 		let intent = MessageIntent::Broadcast;
@@ -434,7 +436,7 @@ mod tests {
 
 	#[test]
 	fn messages_rebroadcast() {
-		let gv = GossipValidator::<Block>::new();
+		let gv = GossipValidator::<Block>::new(Arc::new(Mutex::new(KnownPeers::new())));
 		let sender = sc_network::PeerId::random();
 		let topic = Default::default();
 

client/beefy/src/communication/mod.rs

@@ -0,0 +1,118 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program 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.
+
+// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Communication streams for the BEEFY networking protocols.
+
+pub mod notification;
+pub mod request_response;
+
+pub(crate) mod gossip;
+pub(crate) mod peers;
+
+pub(crate) mod beefy_protocol_name {
+	use array_bytes::bytes2hex;
+	use sc_network::ProtocolName;
+
+	/// BEEFY votes gossip protocol name suffix.
+	const GOSSIP_NAME: &str = "/beefy/1";
+	/// BEEFY justifications protocol name suffix.
+	const JUSTIFICATIONS_NAME: &str = "/beefy/justifications/1";
+
+	/// Old names for the gossip protocol, used for backward compatibility.
+	pub(super) const LEGACY_NAMES: [&str; 1] = ["/paritytech/beefy/1"];
+
+	/// Name of the votes gossip protocol used by BEEFY.
+	///
+	/// Must be registered towards the networking in order for BEEFY voter to properly function.
+	pub fn gossip_protocol_name<Hash: AsRef<[u8]>>(
+		genesis_hash: Hash,
+		fork_id: Option<&str>,
+	) -> ProtocolName {
+		let genesis_hash = genesis_hash.as_ref();
+		if let Some(fork_id) = fork_id {
+			format!("/{}/{}{}", bytes2hex("", genesis_hash), fork_id, GOSSIP_NAME).into()
+		} else {
+			format!("/{}{}", bytes2hex("", genesis_hash), GOSSIP_NAME).into()
+		}
+	}
+
+	/// Name of the BEEFY justifications request-response protocol.
+	pub fn justifications_protocol_name<Hash: AsRef<[u8]>>(
+		genesis_hash: Hash,
+		fork_id: Option<&str>,
+	) -> ProtocolName {
+		let genesis_hash = genesis_hash.as_ref();
+		if let Some(fork_id) = fork_id {
+			format!("/{}/{}{}", bytes2hex("", genesis_hash), fork_id, JUSTIFICATIONS_NAME).into()
+		} else {
+			format!("/{}{}", bytes2hex("", genesis_hash), JUSTIFICATIONS_NAME).into()
+		}
+	}
+}
+
+/// Returns the configuration value to put in
+/// [`sc_network::config::NetworkConfiguration::extra_sets`].
+/// For standard protocol name see [`beefy_protocol_name::gossip_protocol_name`].
+pub fn beefy_peers_set_config(
+	gossip_protocol_name: sc_network::ProtocolName,
+) -> sc_network_common::config::NonDefaultSetConfig {
+	let mut cfg =
+		sc_network_common::config::NonDefaultSetConfig::new(gossip_protocol_name, 1024 * 1024);
+
+	cfg.allow_non_reserved(25, 25);
+	cfg.add_fallback_names(beefy_protocol_name::LEGACY_NAMES.iter().map(|&n| n.into()).collect());
+	cfg
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	use sp_core::H256;
+
+	#[test]
+	fn beefy_protocols_names() {
+		use beefy_protocol_name::{gossip_protocol_name, justifications_protocol_name};
+		// Create protocol name using random genesis hash.
+		let genesis_hash = H256::random();
+		let genesis_hex = array_bytes::bytes2hex("", genesis_hash.as_ref());
+
+		let expected_gossip_name = format!("/{}/beefy/1", genesis_hex);
+		let gossip_proto_name = gossip_protocol_name(&genesis_hash, None);
+		assert_eq!(gossip_proto_name.to_string(), expected_gossip_name);
+
+		let expected_justif_name = format!("/{}/beefy/justifications/1", genesis_hex);
+		let justif_proto_name = justifications_protocol_name(&genesis_hash, None);
+		assert_eq!(justif_proto_name.to_string(), expected_justif_name);
+
+		// Create protocol name using hardcoded genesis hash. Verify exact representation.
+		let genesis_hash = [
+			50, 4, 60, 123, 58, 106, 216, 246, 194, 188, 139, 193, 33, 212, 202, 171, 9, 55, 123,
+			94, 8, 43, 12, 251, 187, 57, 173, 19, 188, 74, 205, 147,
+		];
+		let genesis_hex = "32043c7b3a6ad8f6c2bc8bc121d4caab09377b5e082b0cfbbb39ad13bc4acd93";
+
+		let expected_gossip_name = format!("/{}/beefy/1", genesis_hex);
+		let gossip_proto_name = gossip_protocol_name(&genesis_hash, None);
+		assert_eq!(gossip_proto_name.to_string(), expected_gossip_name);
+
+		let expected_justif_name = format!("/{}/beefy/justifications/1", genesis_hex);
+		let justif_proto_name = justifications_protocol_name(&genesis_hash, None);
+		assert_eq!(justif_proto_name.to_string(), expected_justif_name);
+	}
+}

client/beefy/src/communication/peers.rs

@@ -0,0 +1,131 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program 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.
+
+// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Logic for keeping track of BEEFY peers.
+
+// TODO (issue #12296): replace this naive peer tracking with generic one that infers data
+// from multiple network protocols.
+
+use sc_network::PeerId;
+use sp_runtime::traits::{Block, NumberFor, Zero};
+use std::collections::{HashMap, VecDeque};
+
+struct PeerData<B: Block> {
+	last_voted_on: NumberFor<B>,
+}
+
+impl<B: Block> Default for PeerData<B> {
+	fn default() -> Self {
+		PeerData { last_voted_on: Zero::zero() }
+	}
+}
+
+/// Keep a simple map of connected peers
+/// and the most recent voting round they participated in.
+pub struct KnownPeers<B: Block> {
+	live: HashMap<PeerId, PeerData<B>>,
+}
+
+impl<B: Block> KnownPeers<B> {
+	pub fn new() -> Self {
+		Self { live: HashMap::new() }
+	}
+
+	/// Add new connected `peer`.
+	pub fn add_new(&mut self, peer: PeerId) {
+		self.live.entry(peer).or_default();
+	}
+
+	/// Note vote round number for `peer`.
+	pub fn note_vote_for(&mut self, peer: PeerId, round: NumberFor<B>) {
+		let data = self.live.entry(peer).or_default();
+		data.last_voted_on = round.max(data.last_voted_on);
+	}
+
+	/// Remove connected `peer`.
+	pub fn remove(&mut self, peer: &PeerId) {
+		self.live.remove(peer);
+	}
+
+	/// Return _filtered and cloned_ list of peers that have voted on `block` or higher.
+	pub fn at_least_at_block(&self, block: NumberFor<B>) -> VecDeque<PeerId> {
+		self.live
+			.iter()
+			.filter_map(|(k, v)| (v.last_voted_on >= block).then_some(k))
+			.cloned()
+			.collect()
+	}
+
+	/// Answer whether `peer` is part of `KnownPeers` set.
+	pub fn contains(&self, peer: &PeerId) -> bool {
+		self.live.contains_key(peer)
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn should_track_known_peers_progress() {
+		let (alice, bob, charlie) = (PeerId::random(), PeerId::random(), PeerId::random());
+		let mut peers = KnownPeers::<sc_network_test::Block>::new();
+		assert!(peers.live.is_empty());
+
+		// Alice and Bob new connected peers.
+		peers.add_new(alice);
+		peers.add_new(bob);
+		// 'Tracked' Bob seen voting for 5.
+		peers.note_vote_for(bob, 5);
+		// Previously unseen Charlie now seen voting for 10.
+		peers.note_vote_for(charlie, 10);
+
+		assert_eq!(peers.live.len(), 3);
+		assert!(peers.contains(&alice));
+		assert!(peers.contains(&bob));
+		assert!(peers.contains(&charlie));
+
+		// Get peers at block >= 5
+		let at_5 = peers.at_least_at_block(5);
+		// Should be Bob and Charlie
+		assert_eq!(at_5.len(), 2);
+		assert!(at_5.contains(&bob));
+		assert!(at_5.contains(&charlie));
+
+		// 'Tracked' Alice seen voting for 10.
+		peers.note_vote_for(alice, 10);
+
+		// Get peers at block >= 9
+		let at_9 = peers.at_least_at_block(9);
+		// Should be Charlie and Alice
+		assert_eq!(at_9.len(), 2);
+		assert!(at_9.contains(&charlie));
+		assert!(at_9.contains(&alice));
+
+		// Remove Alice
+		peers.remove(&alice);
+		assert_eq!(peers.live.len(), 2);
+		assert!(!peers.contains(&alice));
+
+		// Get peers at block >= 9
+		let at_9 = peers.at_least_at_block(9);
+		// Now should be just Charlie
+		assert_eq!(at_9.len(), 1);
+		assert!(at_9.contains(&charlie));
+	}
+}