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/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;
 
@@ -49,11 +50,14 @@ pub enum Error {
 
 	#[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.
 ///