Add attested candidate request retry timeouts

node/network/statement-distribution/Cargo.toml

@@ -7,6 +7,7 @@ edition.workspace = true
 
 [dependencies]
 futures = "0.3.21"
+futures-timer = "3"
 gum = { package = "tracing-gum", path = "../../gum" }
 polkadot-primitives = { path = "../../../primitives" }
 sp-staking = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }

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

@@ -103,6 +103,9 @@ enum MuxedMessage {
 	Responder(Option<vstaging::ResponderMessage>),
 	/// Messages from answered requests.
 	Response(vstaging::UnhandledResponse),
+	/// Message that a request is ready to be retried. This just acts as a signal that we should
+	/// dispatch all pending requests again.
+	RetryRequest(()),
 }
 
 #[overseer::contextbounds(StatementDistribution, prefix = self::overseer)]
@@ -114,26 +117,30 @@ impl MuxedMessage {
 		from_v1_responder: &mut mpsc::Receiver<V1ResponderMessage>,
 		from_responder: &mut mpsc::Receiver<vstaging::ResponderMessage>,
 	) -> MuxedMessage {
+		let (request_manager, response_manager) = state.request_and_response_managers();
 		// We are only fusing here to make `select` happy, in reality we will quit if one of those
 		// streams end:
 		let from_orchestra = ctx.recv().fuse();
 		let from_v1_requester = from_v1_requester.next();
 		let from_v1_responder = from_v1_responder.next();
 		let from_responder = from_responder.next();
-		let receive_response = vstaging::receive_response(state).fuse();
+		let receive_response = vstaging::receive_response(response_manager).fuse();
+		let retry_request = vstaging::next_retry(request_manager).fuse();
 		futures::pin_mut!(
 			from_orchestra,
 			from_v1_requester,
 			from_v1_responder,
 			from_responder,
-			receive_response
+			receive_response,
+			retry_request,
 		);
 		futures::select! {
 			msg = from_orchestra => MuxedMessage::Subsystem(msg.map_err(FatalError::SubsystemReceive)),
 			msg = from_v1_requester => MuxedMessage::V1Requester(msg),
 			msg = from_v1_responder => MuxedMessage::V1Responder(msg),
 			msg = from_responder => MuxedMessage::Responder(msg),
 			msg = receive_response => MuxedMessage::Response(msg),
+			msg = retry_request => MuxedMessage::RetryRequest(msg),
 		}
 	}
 }
@@ -190,6 +197,7 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 		.map_err(FatalError::SpawnTask)?;
 
 		loop {
+			// Wait for the next message.
 			let message = MuxedMessage::receive(
 				&mut ctx,
 				&mut state,
@@ -198,6 +206,7 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 				&mut res_receiver,
 			)
 			.await;
+
 			match message {
 				MuxedMessage::Subsystem(result) => {
 					let result = self
@@ -244,6 +253,11 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 				MuxedMessage::Response(result) => {
 					vstaging::handle_response(&mut ctx, &mut state, result).await;
 				},
+				MuxedMessage::RetryRequest(()) => {
+					// A pending request is ready to retry. This is only a signal to call
+					// `dispatch_requests` again.
+					()
+				},
 			};
 
 			vstaging::dispatch_requests(&mut ctx, &mut state).await;

node/network/statement-distribution/src/vstaging/mod.rs

@@ -58,9 +58,12 @@ use futures::{
 	SinkExt, StreamExt,
 };
 
-use std::collections::{
-	hash_map::{Entry, HashMap},
-	HashSet,
+use std::{
+	collections::{
+		hash_map::{Entry, HashMap},
+		HashSet,
+	},
+	time::{Duration, Instant},
 };
 
 use crate::{
@@ -74,7 +77,7 @@ use groups::Groups;
 use requests::{CandidateIdentifier, RequestProperties};
 use statement_store::{StatementOrigin, StatementStore};
 
-pub use requests::{RequestManager, UnhandledResponse};
+pub use requests::{RequestManager, ResponseManager, UnhandledResponse};
 
 mod candidates;
 mod cluster;
@@ -121,6 +124,9 @@ const BENEFIT_VALID_STATEMENT: Rep = Rep::BenefitMajor("Peer provided a valid st
 const BENEFIT_VALID_STATEMENT_FIRST: Rep =
 	Rep::BenefitMajorFirst("Peer was the first to provide a given valid statement");
 
+/// The amount of time to wait before retrying when the node sends a request and it is dropped.
+pub(crate) const REQUEST_RETRY_DELAY: Duration = Duration::from_secs(1);
+
 struct PerRelayParentState {
 	local_validator: Option<LocalValidatorState>,
 	statement_store: StatementStore,
@@ -201,6 +207,7 @@ pub(crate) struct State {
 	keystore: SyncCryptoStorePtr,
 	authorities: HashMap<AuthorityDiscoveryId, PeerId>,
 	request_manager: RequestManager,
+	response_manager: ResponseManager,
 }
 
 impl State {
@@ -215,8 +222,15 @@ impl State {
 			keystore,
 			authorities: HashMap::new(),
 			request_manager: RequestManager::new(),
+			response_manager: ResponseManager::new(),
 		}
 	}
+
+	pub(crate) fn request_and_response_managers(
+		&mut self,
+	) -> (&mut RequestManager, &mut ResponseManager) {
+		(&mut self.request_manager, &mut self.response_manager)
+	}
 }
 
 // For the provided validator index, if there is a connected peer controlling the given authority
@@ -2313,6 +2327,10 @@ async fn apply_post_confirmation<Context>(
 /// Dispatch pending requests for candidate data & statements.
 #[overseer::contextbounds(StatementDistribution, prefix=self::overseer)]
 pub(crate) async fn dispatch_requests<Context>(ctx: &mut Context, state: &mut State) {
+	if !state.request_manager.has_pending_requests() {
+		return
+	}
+
 	let peers = &state.peers;
 	let peer_advertised = |identifier: &CandidateIdentifier, peer: &_| {
 		let peer_data = peers.get(peer)?;
@@ -2374,7 +2392,11 @@ pub(crate) async fn dispatch_requests<Context>(ctx: &mut Context, state: &mut St
 		})
 	};
 
-	while let Some(request) = state.request_manager.next_request(request_props, peer_advertised) {
+	while let Some(request) = state.request_manager.next_request(
+		&mut state.response_manager,
+		request_props,
+		peer_advertised,
+	) {
 		// Peer is supposedly connected.
 		ctx.send_message(NetworkBridgeTxMessage::SendRequests(
 			vec![Requests::AttestedCandidateVStaging(request)],
@@ -2387,13 +2409,24 @@ pub(crate) async fn dispatch_requests<Context>(ctx: &mut Context, state: &mut St
 /// Wait on the next incoming response. If there are no requests pending, this
 /// future never resolves. It is the responsibility of the user of this API
 /// to interrupt the future.
-pub(crate) async fn receive_response(state: &mut State) -> UnhandledResponse {
-	match state.request_manager.await_incoming().await {
+pub(crate) async fn receive_response(response_manager: &mut ResponseManager) -> UnhandledResponse {
+	match response_manager.incoming().await {
 		Some(r) => r,
 		None => futures::future::pending().await,
 	}
 }
 
+/// Wait on the next soonest retry on a pending request. If there are no retries pending, this
+/// future never resolves. Note that this only signals that a request is ready to retry; the user of
+/// this API must call `dispatch_requests`.
+pub(crate) async fn next_retry(request_manager: &mut RequestManager) {
+	match request_manager.next_retry_time() {
+		Some(instant) =>
+			futures_timer::Delay::new(instant.saturating_duration_since(Instant::now())).await,
+		None => futures::future::pending().await,
+	}
+}
+
 /// Handles an incoming response. This does the actual work of validating the response,
 /// importing statements, sending acknowledgements, etc.
 #[overseer::contextbounds(StatementDistribution, prefix=self::overseer)]
@@ -2615,15 +2648,15 @@ pub(crate) fn answer_request(state: &mut State, message: ResponderMessage) {
 }
 
 /// Messages coming from the background respond task.
-pub struct ResponderMessage {
+pub(crate) struct ResponderMessage {
 	request: IncomingRequest<AttestedCandidateRequest>,
 	sent_feedback: oneshot::Sender<()>,
 }
 
 /// A fetching task, taking care of fetching candidates via request/response.
 ///
 /// Runs in a background task and feeds request to [`answer_request`] through [`MuxedMessage`].
-pub async fn respond_task(
+pub(crate) async fn respond_task(
 	mut receiver: IncomingRequestReceiver<AttestedCandidateRequest>,
 	mut sender: mpsc::Sender<ResponderMessage>,
 ) {