Retry availability until the receiver of the request is dropped (#2763)

* guide updates

* keep interactions alive until receivers drop

* retry indefinitely

* cancel approval tasks on finality

* use swap_remove instead of remove

node/core/approval-voting/src/lib.rs

@@ -53,6 +53,7 @@ use sp_application_crypto::Pair;
 use kvdb::KeyValueDB;
 
 use futures::prelude::*;
+use futures::future::RemoteHandle;
 use futures::channel::{mpsc, oneshot};
 
 use std::collections::{BTreeMap, HashMap};
@@ -444,6 +445,7 @@ enum Action {
 	WriteCandidateEntry(CandidateHash, CandidateEntry),
 	LaunchApproval {
 		indirect_cert: IndirectAssignmentCert,
+		relay_block_number: BlockNumber,
 		candidate_index: CandidateIndex,
 		session: SessionIndex,
 		candidate: CandidateReceipt,
@@ -452,6 +454,8 @@ enum Action {
 	Conclude,
 }
 
+type BackgroundTaskMap = BTreeMap<BlockNumber, Vec<RemoteHandle<()>>>;
+
 async fn run<C>(
 	mut ctx: C,
 	subsystem: ApprovalVotingSubsystem,
@@ -472,6 +476,9 @@ async fn run<C>(
 
 	let mut wakeups = Wakeups::default();
 
+	// map block numbers to background work.
+	let mut background_tasks = BTreeMap::new();
+
 	let mut last_finalized_height: Option<BlockNumber> = None;
 	let mut background_rx = background_rx.fuse();
 
@@ -489,7 +496,7 @@ async fn run<C>(
 				)?
 			}
 			next_msg = ctx.recv().fuse() => {
-				handle_from_overseer(
+				let actions = handle_from_overseer(
 					&mut ctx,
 					&mut state,
 					&subsystem.metrics,
@@ -497,7 +504,13 @@ async fn run<C>(
 					next_msg?,
 					&mut last_finalized_height,
 					&wakeups,
-				).await?
+				).await?;
+
+				if let Some(finalized_height) = last_finalized_height {
+					cleanup_background_tasks(finalized_height, &mut background_tasks);
+				}
+
+				actions
 			}
 			background_request = background_rx.next().fuse() => {
 				if let Some(req) = background_request {
@@ -519,6 +532,7 @@ async fn run<C>(
 			&mut wakeups,
 			db_writer,
 			&background_tx,
+			&mut background_tasks,
 			actions,
 		).await? {
 			break;
@@ -535,6 +549,7 @@ async fn handle_actions(
 	wakeups: &mut Wakeups,
 	db: &dyn KeyValueDB,
 	background_tx: &mpsc::Sender<BackgroundRequest>,
+	background_tasks: &mut BackgroundTaskMap,
 	actions: impl IntoIterator<Item = Action>,
 ) -> SubsystemResult<bool> {
 	let mut transaction = approval_db::v1::Transaction::default();
@@ -555,6 +570,7 @@ async fn handle_actions(
 			}
 			Action::LaunchApproval {
 				indirect_cert,
+				relay_block_number,
 				candidate_index,
 				session,
 				candidate,
@@ -569,7 +585,7 @@ async fn handle_actions(
 					candidate_index,
 				).into());
 
-				launch_approval(
+				let handle = launch_approval(
 					ctx,
 					background_tx.clone(),
 					session,
@@ -578,7 +594,11 @@ async fn handle_actions(
 					block_hash,
 					candidate_index as _,
 					backing_group,
-				).await?
+				).await?;
+
+				if let Some(handle) = handle {
+					background_tasks.entry(relay_block_number).or_default().push(handle);
+				}
 			}
 			Action::Conclude => { conclude = true; }
 		}
@@ -594,6 +614,19 @@ async fn handle_actions(
 	Ok(conclude)
 }
 
+// Clean up all background tasks which are no longer needed as they correspond to a
+// finalized block.
+fn cleanup_background_tasks(
+	current_finalized_block: BlockNumber,
+	tasks: &mut BackgroundTaskMap,
+) {
+	let after = tasks.split_off(&(current_finalized_block + 1));
+	*tasks = after;
+
+	// tasks up to the finalized block are dropped, and `RemoteHandle` cancels
+	// the task on drop.
+}
+
 // Handle an incoming signal from the overseer. Returns true if execution should conclude.
 async fn handle_from_overseer(
 	ctx: &mut impl SubsystemContext,
@@ -1533,6 +1566,7 @@ fn process_wakeup(
 			// sanity: should always be present.
 			actions.push(Action::LaunchApproval {
 				indirect_cert,
+				relay_block_number: block_entry.block_number(),
 				candidate_index: i as _,
 				session: block_entry.session(),
 				candidate: candidate_entry.candidate_receipt().clone(),
@@ -1566,6 +1600,9 @@ fn process_wakeup(
 	Ok(actions)
 }
 
+// Launch approval work, returning an `AbortHandle` which corresponds to the background task
+// spawned. When the background work is no longer needed, the `AbortHandle` should be dropped
+// to cancel the background work and any requests it has spawned.
 async fn launch_approval(
 	ctx: &mut impl SubsystemContext,
 	mut background_tx: mpsc::Sender<BackgroundRequest>,
@@ -1575,7 +1612,7 @@ async fn launch_approval(
 	block_hash: Hash,
 	candidate_index: usize,
 	backing_group: GroupIndex,
-) -> SubsystemResult<()> {
+) -> SubsystemResult<Option<RemoteHandle<()>>> {
 	let (a_tx, a_rx) = oneshot::channel();
 	let (code_tx, code_rx) = oneshot::channel();
 	let (context_num_tx, context_num_rx) = oneshot::channel();
@@ -1610,7 +1647,7 @@ async fn launch_approval(
 				candidate.descriptor.relay_parent,
 			);
 
-			return Ok(());
+			return Ok(None);
 		}
 	};
 
@@ -1719,7 +1756,10 @@ async fn launch_approval(
 		}
 	};
 
-	ctx.spawn("approval-checks", Box::pin(background)).await
+	let (background, remote_handle) = background.remote_handle();
+	ctx.spawn("approval-checks", Box::pin(background))
+		.await
+		.map(move |()| Some(remote_handle))
 }
 
 // Issue and import a local approval vote. Should only be invoked after approval checks

node/core/approval-voting/src/persisted_entries.rs

@@ -397,6 +397,11 @@ impl BlockEntry {
 	pub fn candidates(&self) -> &[(CoreIndex, CandidateHash)] {
 		&self.candidates
 	}
+
+	/// Access the block number of the block entry.
+	pub fn block_number(&self) -> BlockNumber {
+		self.block_number
+	}
 }
 
 impl From<crate::approval_db::v1::BlockEntry> for BlockEntry {

node/network/availability-recovery/src/error.rs

@@ -16,7 +16,7 @@
 
 //! The `Error` and `Result` types used by the subsystem.
 
-use futures::channel::{mpsc, oneshot};
+use futures::channel::oneshot;
 use thiserror::Error;
 
 /// Error type used by the Availability Recovery subsystem.
@@ -34,9 +34,6 @@ pub enum Error {
 	#[error("failed to send response")]
 	CanceledResponseSender,
 
-	#[error("to_state channel is closed")]
-	ClosedToState(#[source] mpsc::SendError),
-
 	#[error(transparent)]
 	Runtime(#[from] polkadot_subsystem::errors::RuntimeApiError),
 

node/network/availability-recovery/src/lib.rs

@@ -19,9 +19,11 @@
 #![warn(missing_docs)]
 
 use std::collections::HashMap;
+use std::pin::Pin;
 
-use futures::{channel::{oneshot, mpsc}, prelude::*, stream::FuturesUnordered};
-use futures::future::BoxFuture;
+use futures::{channel::oneshot, prelude::*, stream::FuturesUnordered};
+use futures::future::{BoxFuture, RemoteHandle, FutureExt};
+use futures::task::{Context, Poll};
 use lru::LruCache;
 use rand::seq::SliceRandom;
 
@@ -33,7 +35,7 @@ use polkadot_primitives::v1::{
 use polkadot_node_primitives::{ErasureChunk, AvailableData};
 use polkadot_subsystem::{
 	SubsystemContext, SubsystemResult, SubsystemError, Subsystem, SpawnedSubsystem, FromOverseer,
-	OverseerSignal, ActiveLeavesUpdate,
+	OverseerSignal, ActiveLeavesUpdate, SubsystemSender,
 	errors::RecoveryError,
 	jaeger,
 	messages::{
@@ -67,21 +69,6 @@ pub struct AvailabilityRecoverySubsystem {
 	fast_path: bool,
 }
 
-/// Accumulate all awaiting sides for some particular `AvailableData`.
-struct InteractionHandle {
-	awaiting: Vec<oneshot::Sender<Result<AvailableData, RecoveryError>>>,
-}
-
-/// A message received by main code from an async `Interaction` task.
-#[derive(Debug)]
-enum FromInteraction {
-	/// An interaction concluded.
-	Concluded(CandidateHash, Result<AvailableData, RecoveryError>),
-
-	/// Send a request on the network service.
-	NetworkRequest(Requests),
-}
-
 struct RequestFromBackersPhase {
 	// a random shuffling of the validators from the backing group which indicates the order
 	// in which we connect to them and request the chunk.
@@ -95,7 +82,7 @@ struct RequestChunksPhase {
 	received_chunks: HashMap<ValidatorIndex, ErasureChunk>,
 	requesting_chunks: FuturesUnordered<BoxFuture<
 		'static,
-		Result<Option<ErasureChunk>, RequestError>>,
+		Result<Option<ErasureChunk>, (ValidatorIndex, RequestError)>>,
 	>,
 }
 
@@ -122,9 +109,8 @@ enum InteractionPhase {
 }
 
 /// A state of a single interaction reconstructing an available data.
-struct Interaction {
-	/// A communication channel with the `State`.
-	to_state: mpsc::Sender<FromInteraction>,
+struct Interaction<S> {
+	sender: S,
 
 	/// The parameters of the interaction.
 	params: InteractionParams,
@@ -142,13 +128,12 @@ impl RequestFromBackersPhase {
 		}
 	}
 
-	// Run this phase to completion, returning `true` if data was successfully recovered and
-	// false otherwise.
+	// Run this phase to completion.
 	async fn run(
 		&mut self,
 		params: &InteractionParams,
-		to_state: &mut mpsc::Sender<FromInteraction>
-	) -> Result<bool, mpsc::SendError> {
+		sender: &mut impl SubsystemSender,
+	) -> Result<AvailableData, RecoveryError> {
 		tracing::trace!(
 			target: LOG_TARGET,
 			candidate_hash = ?params.candidate_hash,
@@ -158,7 +143,7 @@ impl RequestFromBackersPhase {
 		loop {
 			// Pop the next backer, and proceed to next phase if we're out.
 			let validator_index = match self.shuffled_backers.pop() {
-				None => return Ok(false),
+				None => return Err(RecoveryError::Unavailable),
 				Some(i) => i,
 			};
 
@@ -168,21 +153,21 @@ impl RequestFromBackersPhase {
 				req_res::v1::AvailableDataFetchingRequest { candidate_hash: params.candidate_hash },
 			);
 
-			to_state.send(FromInteraction::NetworkRequest(Requests::AvailableDataFetching(req))).await?;
+			sender.send_message(NetworkBridgeMessage::SendRequests(
+				vec![Requests::AvailableDataFetching(req)],
+				IfDisconnected::TryConnect,
+			).into()).await;
 
 			match res.await {
 				Ok(req_res::v1::AvailableDataFetchingResponse::AvailableData(data)) => {
 					if reconstructed_data_matches_root(params.validators.len(), &params.erasure_root, &data) {
-						to_state.send(
-							FromInteraction::Concluded(params.candidate_hash.clone(), Ok(data))
-						).await?;
-
 						tracing::trace!(
 							target: LOG_TARGET,
 							candidate_hash = ?params.candidate_hash,
 							"Received full data",
 						);
-						return Ok(true);
+
+						return Ok(data);
 					} else {
 						tracing::debug!(
 							target: LOG_TARGET,
@@ -222,8 +207,8 @@ impl RequestChunksPhase {
 	async fn launch_parallel_requests(
 		&mut self,
 		params: &InteractionParams,
-		to_state: &mut mpsc::Sender<FromInteraction>,
-	) -> Result<(), mpsc::SendError> {
+		sender: &mut impl SubsystemSender,
+	) {
 		let max_requests = std::cmp::min(N_PARALLEL, params.threshold);
 		while self.requesting_chunks.len() < max_requests {
 			if let Some(validator_index) = self.shuffling.pop() {
@@ -247,39 +232,36 @@ impl RequestChunksPhase {
 					raw_request.clone(),
 				);
 
-				to_state.send(FromInteraction::NetworkRequest(Requests::ChunkFetching(req))).await?;
+				sender.send_message(NetworkBridgeMessage::SendRequests(
+					vec![Requests::ChunkFetching(req)],
+					IfDisconnected::TryConnect,
+				).into()).await;
 
 				self.requesting_chunks.push(Box::pin(async move {
 					match res.await {
 						Ok(req_res::v1::ChunkFetchingResponse::Chunk(chunk))
 							=> Ok(Some(chunk.recombine_into_chunk(&raw_request))),
 						Ok(req_res::v1::ChunkFetchingResponse::NoSuchChunk) => Ok(None),
-						Err(e) => Err(e),
+						Err(e) => Err((validator_index, e)),
 					}
 				}));
 			} else {
 				break;
 			}
 		}
-
-		Ok(())
 	}
 
 	async fn wait_for_chunks(
 		&mut self,
 		params: &InteractionParams,
-	) -> Result<(), mpsc::SendError> {
-		// Check if the requesting chunks is not empty not to poll to completion.
-		if self.requesting_chunks.is_empty() {
-			return Ok(());
-		}
-
+	) {
 		// Poll for new updates from requesting_chunks.
-		while let Some(request_result) = self.requesting_chunks.next().await {
+		while let Poll::Ready(Some(request_result))
+			= futures::poll!(self.requesting_chunks.next())
+		{
 			match request_result {
 				Ok(Some(chunk)) => {
-					// Check merkle proofs of any received chunks, and any failures should
-					// lead to issuance of a FromInteraction::ReportPeer message.
+					// Check merkle proofs of any received chunks.
 
 					let validator_index = chunk.index;
 
@@ -313,24 +295,30 @@ impl RequestChunksPhase {
 					}
 				}
 				Ok(None) => {}
-				Err(e) => {
+				Err((validator_index, e)) => {
 					tracing::debug!(
 						target: LOG_TARGET,
 						err = ?e,
+						?validator_index,
 						"Failure requesting chunk",
 					);
+
+					match e {
+						RequestError::InvalidResponse(_) => {}
+						RequestError::NetworkError(_) | RequestError::Canceled(_) => {
+							self.shuffling.push(validator_index);
+						}
+					}
 				}
 			}
 		}
-
-		Ok(())
 	}
 
 	async fn run(
 		&mut self,
 		params: &InteractionParams,
-		to_state: &mut mpsc::Sender<FromInteraction>,
-	) -> Result<(), mpsc::SendError> {
+		sender: &mut impl SubsystemSender,
+	) -> Result<AvailableData, RecoveryError> {
 		loop {
 			if is_unavailable(
 				self.received_chunks.len(),
@@ -347,23 +335,18 @@ impl RequestChunksPhase {
 					n_validators = %params.validators.len(),
 					"Data recovery is not possible",
 				);
-				to_state.send(FromInteraction::Concluded(
-					params.candidate_hash,
-					Err(RecoveryError::Unavailable),
-				)).await?;
 
-				return Ok(());
+				return Err(RecoveryError::Unavailable);
 			}