Batch vote import in dispute-distribution

node/network/dispute-distribution/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 
 [dependencies]
 futures = "0.3.21"
+futures-timer = "3.0.2"
 gum = { package = "tracing-gum", path = "../../gum" }
 derive_more = "0.99.17"
 parity-scale-codec = { version = "3.1.5", features = ["std"] }
@@ -21,6 +22,7 @@ sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "maste
 thiserror = "1.0.31"
 fatality = "0.0.6"
 lru = "0.7.7"
+indexmap = "1.9.1"
 
 [dev-dependencies]
 async-trait = "0.1.53"

node/network/dispute-distribution/src/error.rs

@@ -57,6 +57,8 @@ pub type Result<T> = std::result::Result<T, Error>;
 
 pub type FatalResult<T> = std::result::Result<T, FatalError>;
 
+pub type JfyiResult<T> = std::result::Result<T, JfyiError>;
+
 /// Utility for eating top level errors and log them.
 ///
 /// We basically always want to try and continue on error. This utility function is meant to

node/network/dispute-distribution/src/lib.rs

@@ -24,6 +24,8 @@
 //! The sender is responsible for getting our vote out, see [`sender`]. The receiver handles
 //! incoming [`DisputeRequest`]s and offers spam protection, see [`receiver`].
 
+use std::time::Duration;
+
 use futures::{channel::mpsc, FutureExt, StreamExt, TryFutureExt};
 
 use polkadot_node_network_protocol::authority_discovery::AuthorityDiscovery;
@@ -57,6 +59,8 @@ use self::sender::{DisputeSender, TaskFinish};
 
 ///	## The receiver [`DisputesReceiver`]
 ///
+///	TODO: Obsolete:
+///
 ///	The receiving side is implemented as `DisputesReceiver` and is run as a separate long running task within
 ///	this subsystem ([`DisputesReceiver::run`]).
 ///
@@ -93,6 +97,20 @@ pub use metrics::Metrics;
 
 const LOG_TARGET: &'static str = "parachain::dispute-distribution";
 
+/// Rate limit on the `receiver` side.
+///
+/// If messages from one peer come in at a higher rate than every `RECEIVE_RATE_LIMIT` on average, we
+/// start dropping messages from that peer to enforce that limit.
+pub const RECEIVE_RATE_LIMIT: Duration = Duration::from_millis(100);
+
+/// Rate limit on the `sender` side.
+///
+/// In order to not hit the `RECEIVE_RATE_LIMIT` on the receiving side, we limit out sending rate as
+/// well.
+///
+/// We add 50ms extra, just to have some save margin to the `RECEIVE_RATE_LIMIT`.
+pub const SEND_RATE_LIMIT: Duration = RECEIVE_RATE_LIMIT.saturating_add(Duration::from_millis(50));
+
 /// The dispute distribution subsystem.
 pub struct DisputeDistributionSubsystem<AD> {
 	/// Easy and efficient runtime access for this subsystem.
@@ -172,6 +190,12 @@ where
 		ctx.spawn("disputes-receiver", receiver.run().boxed())
 			.map_err(FatalError::SpawnTask)?;
 
+		// Process messages for sending side.
+		//
+		// Note: We want the sender to be rate limited and we are currently taking advantage of the
+		// fact that the root task of this subsystem is only concerned with sending: Functions of
+		// `DisputeSender` might back pressure if the rate limit is hit, which will slow down this
+		// loop. If this fact ever changes, we will likely need another task.
 		loop {
 			let message = MuxedMessage::receive(&mut ctx, &mut self.sender_rx).await;
 			match message {
@@ -247,9 +271,10 @@ impl MuxedMessage {
 		// ends.
 		let from_overseer = ctx.recv().fuse();
 		futures::pin_mut!(from_overseer, from_sender);
-		futures::select!(
-			msg = from_overseer => MuxedMessage::Subsystem(msg.map_err(FatalError::SubsystemReceive)),
+		// We select biased to make sure we finish up loose ends, before starting new work.
+		futures::select_biased!(
 			msg = from_sender.next() => MuxedMessage::Sender(msg),
+			msg = from_overseer => MuxedMessage::Subsystem(msg.map_err(FatalError::SubsystemReceive)),
 		)
 	}
 }

node/network/dispute-distribution/src/receiver/batch.rs

@@ -0,0 +1,202 @@
+// Copyright 2022 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::{
+	cmp::{Ord, Ordering},
+	collections::{hash_map, BTreeMap, BinaryHeap, HashMap, HashSet},
+	time::Instant,
+};
+
+use futures_timer::Delay;
+use polkadot_node_network_protocol::request_response::{
+	incoming::OutgoingResponseSender, v1::DisputeRequest,
+};
+use polkadot_node_primitives::SignedDisputeStatement;
+use polkadot_primitives::v2::{CandidateHash, CandidateReceipt, ValidatorIndex};
+
+use super::BATCH_COLLECTING_INTERVAL;
+
+/// TODO: Limit number of batches
+
+// - Batches can be added very rate limit timeout.
+// - They have to be checked every BATCH_COLLECTING_INTERVAL.
+// - We can get the earliest next wakeup - keep ordered list of wakeups! Then we always know when
+// the next one comes - only needs to get updated on insert. - Tada!
+struct Batches {
+	batches: HashMap<CandidateHash, Batch>,
+	check_waker: Option<Delay>,
+	pending_wakes: BinaryHeap<PendingWake>,
+}
+
+/// Represents some batch waiting for its next tick to happen at `next_tick`.
+///
+/// This is an internal type meant to be used in the `pending_wakes` `BinaryHeap` field of
+/// `Batches`. It provides an `Ord` instance, that sorts descending with regard to `Instant` (so we
+/// get a `min-heap` with the earliest `Instant` at the top.
+#[derive(Eq, PartialEq)]
+struct PendingWake {
+	candidate_hash: CandidateHash,
+	next_tick: Instant,
+}
+
+/// A found batch is either really found or got created so it can be found.
+enum FoundBatch<'a> {
+	/// Batch just got created.
+	Created(&'a mut Batch),
+	/// Batch already existed.
+	Found(&'a mut Batch),
+}
+
+impl Batches {
+	/// Create new empty `Batches`.
+	pub fn new() -> Self {
+		Self { batches: HashMap::new(), check_waker: None, pending_wakes: BinaryHeap::new() }
+	}
+
+	/// Find a particular batch.
+	///
+	/// That is either find it, or we create it as reflected by the result `FoundBatch`.
+	pub fn find_batch(
+		&mut self,
+		candidate_hash: CandidateHash,
+		candidate_receipt: CandidateReceipt,
+	) -> FoundBatch {
+		debug_assert!(candidate_hash == candidate_receipt.hash());
+		match self.batches.entry(candidate_hash) {
+			hash_map::Entry::Vacant(vacant) =>
+				FoundBatch::Created(vacant.insert(Batch::new(candidate_receipt))),
+			hash_map::Entry::Occupied(occupied) => FoundBatch::Found(occupied.get_mut()),
+		}
+	}
+	// Next steps:
+	//
+	// - Make sure binary heap above stays current.
+	// - Use head of binary heap to schedule next wakeup.
+	// - Provide funtion that provides imports delayed by the wakeup future (similar to rate
+	// limiting).
+	// - Important: Direct updating of last_tick of `Batch` has to be forbidden as this would break
+	// our binary heap. Instead all updates have to go through `Batches`
+}
+
+/// A batch of votes to be imported into the `dispute-coordinator`.
+///
+/// Vote imports are way more efficient when performed in batches, hence we batch together incoming
+/// votes until the rate of incoming votes falls below a threshold, then we import into the dispute
+/// coordinator.
+///
+/// A `Batch` keeps track of the votes to be imported and the current incoming rate, on rate update
+/// it will "flush" in case the incoming rate dropped too low, preparing the import.
+pub struct Batch {
+	/// The actual candidate this batch is concerned with.
+	candidate_receipt: CandidateReceipt,
+
+	/// Cache `CandidateHash` to do efficient sanity checks.
+	candidate_hash: CandidateHash,
+
+	/// All valid votes received in this batch so far.
+	///
+	/// We differentiate between valid and invalid votes, so we can detect (and drop) duplicates,
+	/// while still allowing validators to equivocate.
+	///
+	/// Detecting and rejecting duplicats is crucial in order to effectively enforce
+	/// `MIN_KEEP_BATCH_ALIVE_VOTES` per `BATCH_COLLECTING_INTERVAL`. If we would count duplicates
+	/// here, the mechanism would be broken.
+	valid_votes: HashMap<ValidatorIndex, SignedDisputeStatement>,
+
+	/// All invalid votes received in this batch so far.
+	invalid_votes: HashMap<ValidatorIndex, SignedDisputeStatement>,
+
+	/// How many votes have been batched in the last `BATCH_COLLECTING_INTERVAL`?
+	votes_batched_since_last_tick: u32,
+
+	/// Timestamp of creation or last time we checked incoming rate.
+	last_tick: Instant,
+
+	/// Requesters waiting for a response.
+	pending_responses: Vec<OutgoingResponseSender<DisputeRequest>>,
+}
+
+impl Batch {
+	/// Create a new empty batch based on the given `CandidateReceipt`.
+	///
+	/// To create a `Batch` use Batches::find_batch`.
+	fn new(candidate_receipt: CandidateReceipt) -> Self {
+		Self {
+			candidate_hash: candidate_receipt.hash(),
+			candidate_receipt,
+			valid_votes: HashMap::new(),
+			invalid_votes: HashMap::new(),
+			votes_batched_since_last_tick: 0,
+			last_tick: Instant::now(),
+			pending_responses: Vec::new(),
+		}
+	}
+
+	/// Add votes from a validator into the batch.
+	///
+	/// The statements are supposed to be the valid and invalid statements received in a
+	/// `DisputeRequest`.
+	///
+	/// The given `pending_response` is the corresponding response sender. If at least one of the
+	/// votes is new as far as this batch is concerned we record the pending_response, for later
+	/// use. In case both votes are known already, we return the response sender as an `Err` value.
+	pub fn add_votes(
+		&mut self,
+		valid_vote: (SignedDisputeStatement, ValidatorIndex),
+		invalid_vote: (SignedDisputeStatement, ValidatorIndex),
+		pending_response: OutgoingResponseSender<DisputeRequest>,
+	) -> Result<(), OutgoingResponseSender<DisputeRequest>> {
+		debug_assert!(valid_vote.0.candidate_hash() == invalid_vote.0.candidate_hash());
+		debug_assert!(valid_vote.0.candidate_hash() == &self.candidate_hash);
+
+		let mut duplicate = true;
+
+		if self.valid_votes.insert(valid_vote.1, valid_vote.0).is_none() {
+			self.votes_batched_since_last_tick += 1;
+			duplicate = false;
+		}
+		if self.invalid_votes.insert(invalid_vote.1, invalid_vote.0).is_none() {
+			self.votes_batched_since_last_tick += 1;
+			duplicate = false;
+		}
+
+		if duplicate {
+			Err(pending_response)
+		} else {
+			Ok(())
+		}
+	}
+
+	/// When the next "tick" is supposed to happen.
+	fn time_next_tick(&self) -> Instant {
+		self.last_tick + BATCH_COLLECTING_INTERVAL
+	}
+}
+
+impl PartialOrd<PendingWake> for PendingWake {
+	fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+		Some(self.cmp(other))
+	}
+}
+impl Ord for PendingWake {
+	fn cmp(&self, other: &Self) -> Ordering {
+		// Reverse order for min-heap:
+		match other.next_tick.cmp(&self.next_tick) {
+			Ordering::Equal => other.candidate_hash.cmp(&self.candidate_hash),
+			o => o,
+		}
+	}
+}

node/network/dispute-distribution/src/receiver/error.rs

@@ -21,6 +21,7 @@ use fatality::Nested;
 
 use polkadot_node_network_protocol::{request_response::incoming, PeerId};
 use polkadot_node_subsystem_util::runtime;
+use polkadot_primitives::v2::AuthorityDiscoveryId;
 
 use crate::LOG_TARGET;
 
@@ -44,16 +45,22 @@ pub enum Error {
 	#[error("Dispute request with invalid signatures, from peer {0}.")]
 	InvalidSignature(PeerId),
 
+	#[error("Received votes from peer {0} have been completely redundant.")]
+	RedundantMessage(PeerId),
+
 	#[error("Import of dispute got canceled for peer {0} - import failed for some reason.")]
 	ImportCanceled(PeerId),
 
 	#[error("Peer {0} attempted to participate in dispute and is not a validator.")]
 	NotAValidator(PeerId),
+
+	#[error("Authority {0} sent messages at a too high rate.")]
+	AuthorityFlooding(AuthorityDiscoveryId),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
 
-pub type JfyiErrorResult<T> = std::result::Result<T, JfyiError>;
+pub type JfyiResult<T> = std::result::Result<T, JfyiError>;
 
 /// Utility for eating top level errors and log them.
 ///