Statement Distribution Per Peer Rate Limit

.gitlab/pipeline/zombienet/polkadot.yml

@@ -166,6 +166,14 @@ zombienet-polkadot-functional-0012-elastic-scaling-mvp:
       --local-dir="${LOCAL_DIR}/functional"
       --test="0012-elastic-scaling-mvp.zndsl"
 
+zombienet-polkadot-functional-0013-spam-statement-distribution-requests:
+  extends:
+    - .zombienet-polkadot-common
+  script:
+    - /home/nonroot/zombie-net/scripts/ci/run-test-local-env-manager.sh
+      --local-dir="${LOCAL_DIR}/functional"
+      --test="0013-spam-statement-distribution-requests.zndsl"
+
 zombienet-polkadot-smoke-0001-parachains-smoke-test:
   extends:
     - .zombienet-polkadot-common

polkadot/node/malus/Cargo.toml

@@ -37,6 +37,7 @@ polkadot-node-core-dispute-coordinator = { path = "../core/dispute-coordinator"
 polkadot-node-core-candidate-validation = { path = "../core/candidate-validation" }
 polkadot-node-core-backing = { path = "../core/backing" }
 polkadot-node-primitives = { path = "../primitives" }
+polkadot-node-network-protocol = { path = "../network/protocol" }
 polkadot-primitives = { path = "../../primitives" }
 color-eyre = { version = "0.6.1", default-features = false }
 assert_matches = "1.5"

polkadot/node/malus/src/malus.rs

@@ -40,6 +40,8 @@ enum NemesisVariant {
 	DisputeAncestor(DisputeAncestorOptions),
 	/// Delayed disputing of finalized candidates.
 	DisputeFinalizedCandidates(DisputeFinalizedCandidatesOptions),
+	/// Spam many request statements instead of sending a single one.
+	SpamStatementRequests(SpamStatementRequestsOptions),
 }
 
 #[derive(Debug, Parser)]
@@ -98,6 +100,11 @@ impl MalusCli {
 					finality_delay,
 				)?
 			},
+			NemesisVariant::SpamStatementRequests(opts) => {
+				let SpamStatementRequestsOptions { spam_factor, cli } = opts;
+
+				polkadot_cli::run_node(cli, SpamStatementRequests { spam_factor }, finality_delay)?
+			},
 		}
 		Ok(())
 	}

polkadot/node/malus/src/variants/mod.rs

@@ -20,13 +20,15 @@ mod back_garbage_candidate;
 mod common;
 mod dispute_finalized_candidates;
 mod dispute_valid_candidates;
+mod spam_statement_requests;
 mod suggest_garbage_candidate;
 mod support_disabled;
 
 pub(crate) use self::{
 	back_garbage_candidate::{BackGarbageCandidateOptions, BackGarbageCandidates},
 	dispute_finalized_candidates::{DisputeFinalizedCandidates, DisputeFinalizedCandidatesOptions},
 	dispute_valid_candidates::{DisputeAncestorOptions, DisputeValidCandidates},
+	spam_statement_requests::{SpamStatementRequests, SpamStatementRequestsOptions},
 	suggest_garbage_candidate::{SuggestGarbageCandidateOptions, SuggestGarbageCandidates},
 	support_disabled::{SupportDisabled, SupportDisabledOptions},
 };

polkadot/node/malus/src/variants/spam_statement_requests.rs

@@ -0,0 +1,161 @@
+// Copyright (C) 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/>.
+
+//! A malicious node variant that attempts spam statement requests.
+//!
+//! This malus variant behaves honestly in everything except when propagating statement distribution
+//! requests through the network bridge subsystem. Instead of sending a single request when it needs
+//! something it attempts to spam the peer with multiple requests.
+//!
+//! Attention: For usage with `zombienet` only!
+
+#![allow(missing_docs)]
+
+use polkadot_cli::{
+	service::{
+		AuxStore, Error, ExtendedOverseerGenArgs,
+		Overseer, OverseerConnector, OverseerGen, OverseerGenArgs, OverseerHandle,
+	},
+	validator_overseer_builder, Cli,
+};
+use polkadot_node_network_protocol::request_response::{outgoing::Requests, OutgoingRequest};
+use polkadot_node_subsystem::{messages::NetworkBridgeTxMessage, SpawnGlue};
+use polkadot_node_subsystem_types::{ChainApiBackend, RuntimeApiSubsystemClient};
+use sp_core::traits::SpawnNamed;
+
+// Filter wrapping related types.
+use crate::{interceptor::*, shared::MALUS};
+
+use std::sync::Arc;
+
+/// Wraps around network bridge and replaces it.
+#[derive(Clone)]
+struct RequestSpammer<Spawner> {
+	spawner: Spawner, //stores the actual network bridge subsystem spawner
+	spam_factor: u32, // How many statement distribution requests to send.
+}
+
+impl<Sender, Spawner> MessageInterceptor<Sender> for RequestSpammer<Spawner>
+where
+	Sender: overseer::NetworkBridgeTxSenderTrait + Clone + Send + 'static,
+	Spawner: overseer::gen::Spawner + Clone + 'static,
+{
+	type Message = NetworkBridgeTxMessage;
+
+	/// Intercept NetworkBridgeTxMessage::SendRequests with Requests::AttestedCandidateV2 inside and
+	/// duplicate that request
+	fn intercept_incoming(
+		&self,
+		_subsystem_sender: &mut Sender,
+		msg: FromOrchestra<Self::Message>,
+	) -> Option<FromOrchestra<Self::Message>> {
+		match msg {
+			FromOrchestra::Communication {
+				msg: NetworkBridgeTxMessage::SendRequests(requests, if_disconnected),
+			} => {
+				let mut new_requests = Vec::new();
+
+				for request in requests {
+					match request {
+						Requests::AttestedCandidateV2(ref req) => {
+							// Temporarily store peer and payload for duplication
+							let peer_to_duplicate = req.peer.clone();
+							let payload_to_duplicate = req.payload.clone();
+							// Push the original request
+							new_requests.push(request);
+
+							// Duplicate for spam purposes
+							gum::info!(
+								target: MALUS,
+								"😈 Duplicating AttestedCandidateV2 request extra {:?} times to peer: {:?}.", self.spam_factor, peer_to_duplicate,
+							);
+							new_requests.extend((0..self.spam_factor - 1).map(|_| {
+								let (new_outgoing_request, _) = OutgoingRequest::new(
+									peer_to_duplicate.clone(),
+									payload_to_duplicate.clone(),
+								);
+								Requests::AttestedCandidateV2(new_outgoing_request)
+							}));
+
+						}
+						_ => {
+							new_requests.push(request);
+						}
+					}
+				}
+
+				// Passthrough the message with a potentially modified number of requests
+				Some(FromOrchestra::Communication {
+					msg: NetworkBridgeTxMessage::SendRequests(new_requests, if_disconnected),
+				})
+			},
+			FromOrchestra::Communication { msg } => Some(FromOrchestra::Communication { msg }),
+			FromOrchestra::Signal(signal) => Some(FromOrchestra::Signal(signal)),
+		}
+	}
+}
+
+//----------------------------------------------------------------------------------
+
+#[derive(Debug, clap::Parser)]
+#[clap(rename_all = "kebab-case")]
+#[allow(missing_docs)]
+pub struct SpamStatementRequestsOptions {
+	/// How many statement distribution requests to send.
+	#[clap(long, ignore_case = true, default_value_t = 1000, value_parser = clap::value_parser!(u32).range(0..=10000000))]
+	pub spam_factor: u32,
+
+	#[clap(flatten)]
+	pub cli: Cli,
+}
+
+/// SpamStatementRequests implementation wrapper which implements `OverseerGen` glue.
+pub(crate) struct SpamStatementRequests {
+	/// How many statement distribution requests to send.
+	pub spam_factor: u32,
+}
+
+impl OverseerGen for SpamStatementRequests {
+	fn generate<Spawner, RuntimeClient>(
+		&self,
+		connector: OverseerConnector,
+		args: OverseerGenArgs<'_, Spawner, RuntimeClient>,
+		ext_args: Option<ExtendedOverseerGenArgs>,
+	) -> Result<(Overseer<SpawnGlue<Spawner>, Arc<RuntimeClient>>, OverseerHandle), Error>
+	where
+		RuntimeClient: RuntimeApiSubsystemClient + ChainApiBackend + AuxStore + 'static,
+		Spawner: 'static + SpawnNamed + Clone + Unpin,
+	{
+		gum::info!(
+			target: MALUS,
+			"😈 Started Malus node that duplicates each statement distribution request spam_factor = {:?} times.",
+			&self.spam_factor,
+		);
+
+		let request_spammer = RequestSpammer {
+			spawner: SpawnGlue(args.spawner.clone()),
+			spam_factor: self.spam_factor,
+		};
+
+		validator_overseer_builder(
+			args,
+			ext_args.expect("Extended arguments required to build validator overseer are provided"),
+		)?
+		.replace_network_bridge_tx(move |cb| InterceptedSubsystem::new(cb, request_spammer))
+		.build_with_connector(connector)
+		.map_err(|e| e.into())
+	}
+}

polkadot/node/network/statement-distribution/src/v2/mod.rs

@@ -58,6 +58,8 @@ use sp_keystore::KeystorePtr;
 use fatality::Nested;
 use futures::{
 	channel::{mpsc, oneshot},
+	future::FutureExt,
+	select,
 	stream::FuturesUnordered,
 	SinkExt, StreamExt,
 };
@@ -3350,33 +3352,56 @@ pub(crate) async fn respond_task(
 	mut sender: mpsc::Sender<ResponderMessage>,
 ) {
 	let mut pending_out = FuturesUnordered::new();
+	let mut active_peers = HashSet::new();
+
 	loop {
-		// Ensure we are not handling too many requests in parallel.
-		if pending_out.len() >= MAX_PARALLEL_ATTESTED_CANDIDATE_REQUESTS as usize {
-			// Wait for one to finish:
-			pending_out.next().await;
-		}
+		select! {
+			// New request
+			request_result = receiver.recv(|| vec![COST_INVALID_REQUEST]).fuse() => {
+				let request = match request_result.into_nested() {
+					Ok(Ok(v)) => v,
+					Err(fatal) => {
+						gum::debug!(target: LOG_TARGET, error = ?fatal, "Shutting down request responder");
+						return
+					},
+					Ok(Err(jfyi)) => {
+						gum::debug!(target: LOG_TARGET, error = ?jfyi, "Decoding request failed");
+						continue
+					},
+				};
 
-		let req = match receiver.recv(|| vec![COST_INVALID_REQUEST]).await.into_nested() {
-			Ok(Ok(v)) => v,
-			Err(fatal) => {
-				gum::debug!(target: LOG_TARGET, error = ?fatal, "Shutting down request responder");
-				return
+				// If peer currently being served drop request
+				if active_peers.contains(&request.peer) {
+					gum::debug!(target: LOG_TARGET, "Peer already being served, dropping request");
+					continue
+				}
+
+				// If we are over parallel limit wait for one to finish
+				if pending_out.len() >= MAX_PARALLEL_ATTESTED_CANDIDATE_REQUESTS as usize {
+					gum::debug!(target: LOG_TARGET, "Over max parallel requests, waiting for one to finish");
+					let (_, peer) = pending_out.select_next_some().await;
+					active_peers.remove(&peer);
+				}
+
+				// Start serving the request
+				let (pending_sent_tx, pending_sent_rx) = oneshot::channel();
+				let peer = request.peer;
+				if let Err(err) = sender
+					.feed(ResponderMessage { request, sent_feedback: pending_sent_tx })
+					.await
+				{
+					gum::debug!(target: LOG_TARGET, ?err, "Shutting down responder");
+					return
+				}
+				let future_with_peer = pending_sent_rx.map(move |result| (result, peer));
+				pending_out.push(future_with_peer);
+				active_peers.insert(peer);
 			},
-			Ok(Err(jfyi)) => {
-				gum::debug!(target: LOG_TARGET, error = ?jfyi, "Decoding request failed");
-				continue
+			// Request served/finished
+			result = pending_out.select_next_some() => {
+				let (_, peer) = result;
+				active_peers.remove(&peer);
 			},
-		};
-
-		let (pending_sent_tx, pending_sent_rx) = oneshot::channel();
-		if let Err(err) = sender
-			.feed(ResponderMessage { request: req, sent_feedback: pending_sent_tx })
-			.await
-		{
-			gum::debug!(target: LOG_TARGET, ?err, "Shutting down responder");
-			return
 		}
-		pending_out.push(pending_sent_rx);
 	}
 }