Batch vote import in dispute-distribution

Cargo.toml

@@ -128,9 +128,9 @@ maintenance = { status = "actively-developed" }
 #
 # This list is ordered alphabetically.
 [profile.dev.package]
-blake2b_simd = { opt-level = 3 }
 blake2 = { opt-level = 3 }
 blake2-rfc = { opt-level = 3 }
+blake2b_simd = { opt-level = 3 }
 chacha20poly1305 = { opt-level = 3 }
 cranelift-codegen = { opt-level = 3 }
 cranelift-wasm = { opt-level = 3 }
@@ -141,8 +141,8 @@ curve25519-dalek = { opt-level = 3 }
 ed25519-dalek = { opt-level = 3 }
 flate2 = { opt-level = 3 }
 futures-channel = { opt-level = 3 }
-hashbrown = { opt-level = 3 }
 hash-db = { opt-level = 3 }
+hashbrown = { opt-level = 3 }
 hmac = { opt-level = 3 }
 httparse = { opt-level = 3 }
 integer-sqrt = { opt-level = 3 }
@@ -154,8 +154,8 @@ libz-sys = { opt-level = 3 }
 mio = { opt-level = 3 }
 nalgebra = { opt-level = 3 }
 num-bigint = { opt-level = 3 }
-parking_lot_core = { opt-level = 3 }
 parking_lot = { opt-level = 3 }
+parking_lot_core = { opt-level = 3 }
 percent-encoding = { opt-level = 3 }
 primitive-types = { opt-level = 3 }
 reed-solomon-novelpoly = { opt-level = 3 }
@@ -165,6 +165,7 @@ sha2 = { opt-level = 3 }
 sha3 = { opt-level = 3 }
 smallvec = { opt-level = 3 }
 snow = { opt-level = 3 }
+substrate-bip39 = {opt-level = 3}
 twox-hash = { opt-level = 3 }
 uint = { opt-level = 3 }
 wasmi = { opt-level = 3 }

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/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;
@@ -64,16 +66,19 @@ use self::sender::{DisputeSender, TaskFinish};
 ///	via a dedicated channel and forwarding them to the dispute coordinator via
 ///	`DisputeCoordinatorMessage::ImportStatements`. Being the interface to the network and untrusted
 ///	nodes, the reality is not that simple of course. Before importing statements the receiver will
-///	make sure as good as it can to filter out malicious/unwanted/spammy requests. For this it does
-///	the following:
+///	batch up imports as well as possible for efficient imports while maintaining timely dispute
+///	resolution and handling of spamming validators:
 ///
 ///	- Drop all messages from non validator nodes, for this it requires the [`AuthorityDiscovery`]
 ///	service.
-///	- Drop messages from a node, if we are already importing a message from that node (flood).
-///	- Drop messages from nodes, that provided us messages where the statement import failed.
+///	- Drop messages from a node, if it sends at a too high rate.
+///	- Filter out duplicate messages (over some period of time).
 ///	- Drop any obviously invalid votes (invalid signatures for example).
 ///	- Ban peers whose votes were deemed invalid.
 ///
+///	In general dispute-distribution works on limiting the work the dispute-coordinator will have to
+///	do, while at the same time making it aware of new disputes as fast as possible.
+///
 /// For successfully imported votes, we will confirm the receipt of the message back to the sender.
 /// This way a received confirmation guarantees, that the vote has been stored to disk by the
 /// receiver.
@@ -93,6 +98,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 +191,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 +272,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/metrics.rs

@@ -72,9 +72,12 @@ impl Metrics {
 	}
 
 	/// Statements have been imported.
-	pub fn on_imported(&self, label: &'static str) {
+	pub fn on_imported(&self, label: &'static str, num_requests: usize) {
 		if let Some(metrics) = &self.0 {
-			metrics.imported_requests.with_label_values(&[label]).inc()
+			metrics
+				.imported_requests
+				.with_label_values(&[label])
+				.inc_by(num_requests as u64)
 		}
 	}