A fast-path for requesting AvailableData from backing validators (#2453)

* guide changes for a fast-path requesting from backing validators

* add backing group to availability recovery message

* add new phase to interaction

* typos

* add full data messages

* handle new network messages

* dispatch full data requests

* cleanup

* check chunk index

* test for invalid recovery

* tests

* Typos.

* fix some grumbles

* be more explicit about error handling and control flow

* fast-path param

* use with_chunks_only in Service

Co-authored-by: Robert Klotzner <robert.klotzner@gmx.at>

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

@@ -25,13 +25,13 @@ use std::pin::Pin;
 use futures::{channel::{oneshot, mpsc}, prelude::*, stream::FuturesUnordered};
 use futures_timer::Delay;
 use lru::LruCache;
-use rand::{seq::SliceRandom, thread_rng};
+use rand::seq::SliceRandom;
 use streamunordered::{StreamUnordered, StreamYield};
 
 use polkadot_primitives::v1::{
 	AuthorityDiscoveryId, AvailableData, CandidateReceipt, CandidateHash,
 	Hash, ErasureChunk, ValidatorId, ValidatorIndex,
-	SessionInfo, SessionIndex, BlakeTwo256, HashT,
+	SessionInfo, SessionIndex, BlakeTwo256, HashT, GroupIndex,
 };
 use polkadot_subsystem::{
 	SubsystemContext, SubsystemResult, SubsystemError, Subsystem, SpawnedSubsystem, FromOverseer,
@@ -59,6 +59,7 @@ const LOG_TARGET: &str = "availability_recovery";
 
 const COST_MERKLE_PROOF_INVALID: Rep = Rep::CostMinor("Merkle proof was invalid");
 const COST_UNEXPECTED_CHUNK: Rep = Rep::CostMinor("Peer has sent an unexpected chunk");
+const COST_INVALID_AVAILABLE_DATA: Rep = Rep::CostMinor("Peer provided invalid available data");
 
 // How many parallel requests interaction should have going at once.
 const N_PARALLEL: usize = 50;
@@ -67,18 +68,54 @@ const N_PARALLEL: usize = 50;
 const LRU_SIZE: usize = 16;
 
 // A timeout for a chunk request.
+#[cfg(not(test))]
 const CHUNK_REQUEST_TIMEOUT: Duration = Duration::from_secs(3);
 
-// A period to poll and clean AwaitedChunks.
-const AWAITED_CHUNKS_CLEANUP_INTERVAL: Duration = Duration::from_secs(1);
+#[cfg(test)]
+const CHUNK_REQUEST_TIMEOUT: Duration = Duration::from_millis(100);
+
+// A timeout for a full data request.
+#[cfg(not(test))]
+const FULL_DATA_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[cfg(test)]
+const FULL_DATA_REQUEST_TIMEOUT: Duration = Duration::from_millis(100);
+
+// A period to poll and clean awaited data.
+const AWAITED_CLEANUP_INTERVAL: Duration = Duration::from_secs(1);
 
 /// The Availability Recovery Subsystem.
-pub struct AvailabilityRecoverySubsystem;
+pub struct AvailabilityRecoverySubsystem {
+	fast_path: bool,
+}
 
-type ChunkResponse = Result<(PeerId, ErasureChunk), RecoveryError>;
+type DataResponse<T> = (PeerId, ValidatorIndex, T);
 
-/// Data we keep around for every chunk that we are awaiting.
-struct AwaitedChunk {
+/// Awaited data from the network.
+enum Awaited {
+	Chunk(AwaitedData<ErasureChunk>),
+	FullData(AwaitedData<AvailableData>),
+}
+
+impl Awaited {
+	fn is_canceled(&self) -> bool {
+		match *self {
+			Awaited::Chunk(ref c) => c.response.is_canceled(),
+			Awaited::FullData(ref fd) => fd.response.is_canceled(),
+		}
+	}
+
+	/// Token to cancel the connection request to the validator.
+	fn token(&self) -> usize {
+		match *self {
+			Awaited::Chunk(ref c) => c.token,
+			Awaited::FullData(ref fd) => fd.token,
+		}
+	}
+}
+
+/// Data we keep around for network data that we are awaiting.
+struct AwaitedData<T> {
 	/// Index of the validator we have requested this chunk from.
 	validator_index: ValidatorIndex,
 
@@ -89,7 +126,7 @@ struct AwaitedChunk {
 	token: usize,
 
 	/// Result sender.
-	response: oneshot::Sender<ChunkResponse>,
+	response: oneshot::Sender<DataResponse<T>>,
 }
 
 /// Accumulate all awaiting sides for some particular `AvailableData`.
@@ -104,11 +141,19 @@ enum FromInteraction {
 	Concluded(CandidateHash, Result<AvailableData, RecoveryError>),
 
 	/// Make a request of a particular chunk from a particular validator.
-	MakeRequest(
+	MakeChunkRequest(
 		AuthorityDiscoveryId,
 		CandidateHash,
 		ValidatorIndex,
-		oneshot::Sender<ChunkResponse>,
+		oneshot::Sender<DataResponse<ErasureChunk>>,
+	),
+
+	/// Make a request of the full data from a particular validator.
+	MakeFullDataRequest(
+		AuthorityDiscoveryId,
+		CandidateHash,
+		ValidatorIndex,
+		oneshot::Sender<DataResponse<AvailableData>>,
 	),
 
 	/// Report a peer.
@@ -118,21 +163,27 @@ enum FromInteraction {
 	),
 }
 
-/// A state of a single interaction reconstructing an available data.
-struct Interaction {
-	/// A communication channel with the `State`.
-	to_state: mpsc::Sender<FromInteraction>,
+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.
+	shuffled_backers: Vec<ValidatorIndex>,
+}
 
+struct RequestChunksPhase {
+	// a random shuffling of the validators which indicates the order in which we connect to the validators and
+	// request the chunk from them.
+	shuffling: Vec<ValidatorIndex>,
+	received_chunks: HashMap<ValidatorIndex, ErasureChunk>,
+	requesting_chunks: FuturesUnordered<Timeout<oneshot::Receiver<DataResponse<ErasureChunk>>>>,
+}
+
+struct InteractionParams {
 	/// Discovery ids of `validators`.
 	validator_authority_keys: Vec<AuthorityDiscoveryId>,
 
 	/// Validators relevant to this `Interaction`.
 	validators: Vec<ValidatorId>,
 
-	/// A random shuffling of the validators which indicates the order in which we connect
-	/// to the validators and request the chunk from them.
-	shuffling: Vec<ValidatorIndex>,
-
 	/// The number of pieces needed.
 	threshold: usize,
 
@@ -141,35 +192,118 @@ struct Interaction {
 
 	/// The root of the erasure encoding of the para block.
 	erasure_root: Hash,
-
-	/// The chunks that we have received from peers.
-	received_chunks: HashMap<PeerId, ErasureChunk>,
-
-	/// The chunk requests that are waiting to complete.
-	requesting_chunks: FuturesUnordered<Timeout<oneshot::Receiver<ChunkResponse>>>,
 }
 
-const fn is_unavailable(
-	received_chunks: usize,
-	requesting_chunks: usize,
-	n_validators: usize,
-	threshold: usize,
-) -> bool {
-	received_chunks + requesting_chunks + n_validators < threshold
+enum InteractionPhase {
+	RequestFromBackers(RequestFromBackersPhase),
+	RequestChunks(RequestChunksPhase),
 }
 
-impl Interaction {
-	async fn launch_parallel_requests(&mut self) -> error::Result<()> {
+/// A state of a single interaction reconstructing an available data.
+struct Interaction {
+	/// A communication channel with the `State`.
+	to_state: mpsc::Sender<FromInteraction>,
+
+	/// The parameters of the interaction.
+	params: InteractionParams,
+
+	/// The phase of the interaction.
+	phase: InteractionPhase,
+}
+
+impl RequestFromBackersPhase {
+	fn new(mut backers: Vec<ValidatorIndex>) -> Self {
+		backers.shuffle(&mut rand::thread_rng());
+
+		RequestFromBackersPhase {
+			shuffled_backers: backers,
+		}
+	}
+
+	// Run this phase to completion, returning `true` if data was successfully recovered and
+	// false otherwise.
+	async fn run(
+		&mut self,
+		params: &InteractionParams,
+		to_state: &mut mpsc::Sender<FromInteraction>
+	) -> Result<bool, mpsc::SendError> {
+		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),
+				Some(i) => i,
+			};
+
+			let (tx, rx) = oneshot::channel();
+
+			// Request data.
+			to_state.send(FromInteraction::MakeFullDataRequest(
+				params.validator_authority_keys[validator_index as usize].clone(),
+				params.candidate_hash.clone(),
+				validator_index,
+				tx,
+			)).await?;
+
+			match rx.timeout(FULL_DATA_REQUEST_TIMEOUT).await {
+				Some(Ok((peer_id, _validator_index, 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?;
+
+						return Ok(true);
+					} else {
+						to_state.send(FromInteraction::ReportPeer(
+							peer_id.clone(),
+							COST_INVALID_AVAILABLE_DATA,
+						)).await?;
+					}
+				}
+				Some(Err(e)) => {
+					tracing::debug!(
+						target: LOG_TARGET,
+						err = ?e,
+						"A response channel was cancelled while waiting for full data",
+					);
+				}
+				None => {
+					tracing::debug!(
+						target: LOG_TARGET,
+						"A full data request has timed out",
+					);
+				}
+			}
+		}
+	}
+}
+
+impl RequestChunksPhase {
+	fn new(n_validators: ValidatorIndex) -> Self {
+		let mut shuffling: Vec<_> = (0..n_validators).collect();
+		shuffling.shuffle(&mut rand::thread_rng());
+
+		RequestChunksPhase {
+			shuffling,
+			received_chunks: HashMap::new(),
+			requesting_chunks: FuturesUnordered::new(),
+		}
+	}
+
+	async fn launch_parallel_requests(
+		&mut self,
+		params: &InteractionParams,
+		to_state: &mut mpsc::Sender<FromInteraction>,
+	) -> Result<(), mpsc::SendError> {
 		while self.requesting_chunks.len() < N_PARALLEL {
 			if let Some(validator_index) = self.shuffling.pop() {
 				let (tx, rx) = oneshot::channel();
 
-				self.to_state.send(FromInteraction::MakeRequest(
-					self.validator_authority_keys[validator_index as usize].clone(),
-					self.candidate_hash.clone(),
+				to_state.send(FromInteraction::MakeChunkRequest(
+					params.validator_authority_keys[validator_index as usize].clone(),
+					params.candidate_hash.clone(),
 					validator_index,
 					tx,
-				)).await.map_err(error::Error::ClosedToState)?;
+				)).await?;
 
 				self.requesting_chunks.push(rx.timeout(CHUNK_REQUEST_TIMEOUT));
 			} else {
@@ -180,7 +314,11 @@ impl Interaction {
 		Ok(())
 	}
 
-	async fn wait_for_chunks(&mut self) -> error::Result<()> {
+	async fn wait_for_chunks(
+		&mut self,
+		params: &InteractionParams,
+		to_state: &mut mpsc::Sender<FromInteraction>,
+	) -> Result<(), mpsc::SendError> {
 		// Check if the requesting chunks is not empty not to poll to completion.
 		if self.requesting_chunks.is_empty() {
 			return Ok(());
@@ -189,43 +327,49 @@ impl Interaction {
 		// Poll for new updates from requesting_chunks.
 		while let Some(request_result) = self.requesting_chunks.next().await {
 			match request_result {
-				Some(Ok(Ok((peer_id, chunk)))) => {
+				Some(Ok((peer_id, validator_index, chunk))) => {
 					// Check merkle proofs of any received chunks, and any failures should
 					// lead to issuance of a FromInteraction::ReportPeer message.
+
+					// We need to check that the validator index matches the chunk index and
+					// not blindly trust the data from an untrusted peer.
+					if validator_index != chunk.index {
+						to_state.send(FromInteraction::ReportPeer(
+							peer_id.clone(),
+							COST_MERKLE_PROOF_INVALID,
+						)).await?;
+
+						continue;
+					}
+
+
 					if let Ok(anticipated_hash) = branch_hash(
-						&self.erasure_root,
+						&params.erasure_root,
 						&chunk.proof,
 						chunk.index as usize,
 					) {
 						let erasure_chunk_hash = BlakeTwo256::hash(&chunk.chunk);
 
 						if erasure_chunk_hash != anticipated_hash {
-							self.to_state.send(FromInteraction::ReportPeer(
-									peer_id.clone(),
-									COST_MERKLE_PROOF_INVALID,
-							)).await.map_err(error::Error::ClosedToState)?;
-						}
-					} else {
-						self.to_state.send(FromInteraction::ReportPeer(
+							to_state.send(FromInteraction::ReportPeer(
 								peer_id.clone(),
 								COST_MERKLE_PROOF_INVALID,
-						)).await.map_err(error::Error::ClosedToState)?;
+							)).await?;
+						} else {
+							self.received_chunks.insert(validator_index, chunk);
+						}
+					} else {
+						to_state.send(FromInteraction::ReportPeer(
+							peer_id.clone(),
+							COST_MERKLE_PROOF_INVALID,
+						)).await?;
 					}
-
-					self.received_chunks.insert(peer_id, chunk);
 				}
 				Some(Err(e)) => {
 					tracing::debug!(
 						target: LOG_TARGET,
 						err = ?e,
-						"A response channel was cacelled while waiting for a chunk",
-					);
-				}
-				Some(Ok(Err(e))) => {
-					tracing::debug!(
-						target: LOG_TARGET,
-						err = ?e,
-						"A chunk request ended with an error",
+						"A response channel was cancelled while waiting for a chunk",
 					);
 				}
 				None => {
@@ -233,8 +377,6 @@ impl Interaction {
 						target: LOG_TARGET,
 						"A chunk request has timed out",
 					);
-					// we break here to launch another request.
-					break;
 				}
 			}
 		}
@@ -242,58 +384,70 @@ impl Interaction {
 		Ok(())
 	}
 
-	async fn run(mut self) -> error::Result<()> {
+	async fn run(
+		&mut self,
+		params: &InteractionParams,
+		to_state: &mut mpsc::Sender<FromInteraction>,
+	) -> Result<(), mpsc::SendError> {
 		loop {
 			if is_unavailable(
 				self.received_chunks.len(),
 				self.requesting_chunks.len(),
 				self.shuffling.len(),
-				self.threshold,
+				params.threshold,
 			) {
-				self.to_state.send(FromInteraction::Concluded(
-					self.candidate_hash,
+				to_state.send(FromInteraction::Concluded(
+					params.candidate_hash,
 					Err(RecoveryError::Unavailable),
-				)).await.map_err(error::Error::ClosedToState)?;
+				)).await?;
 
 				return Ok(());
 			}
 
-			self.launch_parallel_requests().await?;
-
-			self.wait_for_chunks().await?;
+			self.launch_parallel_requests(params, to_state).await?;
+			self.wait_for_chunks(params, to_state).await?;
 
 			// If received_chunks has more than threshold entries, attempt to recover the data.
 			// If that fails, or a re-encoding of it doesn't match the expected erasure root,
 			// break and issue a FromInteraction::Concluded(RecoveryError::Invalid).
 			// Otherwise, issue a FromInteraction::Concluded(Ok(())).
-			if self.received_chunks.len() >= self.threshold {
+			if self.received_chunks.len() >= params.threshold {
 				let concluded = match polkadot_erasure_coding::reconstruct_v1(
-					self.validators.len(),
+					params.validators.len(),
 					self.received_chunks.values().map(|c| (&c.chunk[..], c.index as usize)),
 				) {
 					Ok(data) => {
-						if reconstructed_data_matches_root(self.validators.len(), &self.erasure_root, &data) {
-							FromInteraction::Concluded(self.candidate_hash.clone(), Ok(data))
+						if reconstructed_data_matches_root(params.validators.len(), &params.erasure_root, &data) {
+							FromInteraction::Concluded(params.candidate_hash.clone(), Ok(data))
 						} else {
 							FromInteraction::Concluded(
-								self.candidate_hash.clone(),
+								params.candidate_hash.clone(),
 								Err(RecoveryError::Invalid),
 							)
 						}
 					}
 					Err(_) => FromInteraction::Concluded(
-						self.candidate_hash.clone(),
+						params.candidate_hash.clone(),
 						Err(RecoveryError::Invalid),
 					),
 				};
 
-				self.to_state.send(concluded).await.map_err(error::Error::ClosedToState)?;
+				to_state.send(concluded).await?;
 				return Ok(());
 			}
 		}
 	}
 }
 
+const fn is_unavailable(
+	received_chunks: usize,
+	requesting_chunks: usize,
+	n_validators: usize,
+	threshold: usize,
+) -> bool {
+	received_chunks + requesting_chunks + n_validators < threshold
+}
+
 fn reconstructed_data_matches_root(
 	n_validators: usize,
 	expected_root: &Hash,
@@ -316,6 +470,32 @@ fn reconstructed_data_matches_root(
 	branches.root() == *expected_root
 }
 
+impl Interaction {
+	async fn run(mut self) -> error::Result<()> {
+		loop {
+			// These only fail if we cannot reach the underlying subsystem, which case there is nothing
+			// meaningful we can do.
+			match self.phase {
+				InteractionPhase::RequestFromBackers(ref mut from_backers) => {
+					if from_backers.run(&self.params, &mut self.to_state).await
+						.map_err(error::Error::ClosedToState)?
+					{
+						break Ok(())
+					} else {
+						self.phase = InteractionPhase::RequestChunks(
+							RequestChunksPhase::new(self.params.validators.len() as _)
+						);
+					}
+				}
+				InteractionPhase::RequestChunks(ref mut from_all) => {
+					break from_all.run(&self.params, &mut self.to_state).await
+						.map_err(error::Error::ClosedToState)
+				}
+			}
+		}
+	}
+}
+
 struct State {
 	/// Each interaction is implemented as its own async task,
 	/// and these handles are for communicating with them.
@@ -325,12 +505,12 @@ struct State {
 	live_block_hash: Hash,
 
 	/// We are waiting for these validators to connect and as soon as they
-	/// do to request the needed chunks we are awaitinf for.
-	discovering_validators: HashMap<AuthorityDiscoveryId, Vec<AwaitedChunk>>,
+	/// do, request the needed data we are waiting for.
+	discovering_validators: HashMap<AuthorityDiscoveryId, Vec<Awaited>>,
 
 	/// Requests that we have issued to the already connected validators
-	/// about the chunks we are interested in.
-	live_chunk_requests: HashMap<RequestId, (PeerId, AwaitedChunk)>,
+	/// about the data we are interested in.
+	live_requests: HashMap<RequestId, (PeerId, Awaited)>,
 
 	/// Derive request ids from this.
 	next_request_id: RequestId,
@@ -357,7 +537,7 @@ impl Default for State {
 			interactions: HashMap::new(),
 			live_block_hash: Hash::default(),
 			discovering_validators: HashMap::new(),
-			live_chunk_requests: HashMap::new(),
+			live_requests: HashMap::new(),
 			next_request_id: 0,
 			connecting_validators: StreamUnordered::new(),
 			availability_lru: LruCache::new(LRU_SIZE),
@@ -394,7 +574,7 @@ async fn handle_signal(
 
 			Ok(false)
 		}
-	    OverseerSignal::BlockFinalized(_, _) => Ok(false)
+		OverseerSignal::BlockFinalized(_, _) => Ok(false)
 	}
 }
 
@@ -415,16 +595,12 @@ async fn launch_interaction(
 	session_index: SessionIndex,
 	session_info: SessionInfo,
 	receipt: CandidateReceipt,
+	backing_group: Option<GroupIndex>,
 	response_sender: oneshot::Sender<Result<AvailableData, RecoveryError>>,
 ) -> error::Result<()> {
-	let threshold = recovery_threshold(session_info.validators.len())?;
 	let to_state = state.from_interaction_tx.clone();
-	let candidate_hash = receipt.hash();
-	let erasure_root = receipt.descriptor.erasure_root;
-	let validators = session_info.validators.clone();
-	let validator_authority_keys = session_info.discovery_keys.clone();
-	let mut shuffling: Vec<_> = (0..validators.len() as ValidatorIndex).collect();
 
+	let candidate_hash = receipt.hash();
 	state.interactions.insert(
 		candidate_hash.clone(),
 		InteractionHandle {
@@ -432,22 +608,27 @@ async fn launch_interaction(
 		}
 	);
 
-	{
-		// make borrow checker happy.
-		let mut rng = thread_rng();
-		shuffling.shuffle(&mut rng);
-	}
+	let params = InteractionParams {
+		validator_authority_keys: session_info.discovery_keys.clone(),
+		validators: session_info.validators.clone(),
+		threshold: recovery_threshold(session_info.validators.len())?,
+		candidate_hash,
+		erasure_root: receipt.descriptor.erasure_root,
+	};
+
+	let phase = backing_group
+		.and_then(|g| session_info.validator_groups.get(g.0 as usize))
+		.map(|group| InteractionPhase::RequestFromBackers(
+			RequestFromBackersPhase::new(group.clone())
+		))
+		.unwrap_or_else(|| InteractionPhase::RequestChunks(
+			RequestChunksPhase::new(params.validators.len() as _)
+		));
 
 	let interaction = Interaction {
 		to_state,
-		validator_authority_keys,
-		validators,
-		shuffling,
-		threshold,
-		candidate_hash,
-		erasure_root,
-		received_chunks: HashMap::new(),
-		requesting_chunks: FuturesUnordered::new(),
+		params,
+		phase,
 	};
 
 	let future = async move {
@@ -478,6 +659,7 @@ async fn handle_recover(
 	ctx: &mut impl SubsystemContext<Message = AvailabilityRecoveryMessage>,
 	receipt: CandidateReceipt,
 	session_index: SessionIndex,
+	backing_group: Option<GroupIndex>,
 	response_sender: oneshot::Sender<Result<AvailableData, RecoveryError>>,
 ) -> error::Result<()> {
 	let candidate_hash = receipt.hash();
@@ -512,6 +694,7 @@ async fn handle_recover(
 				session_index,
 				session_info,
 				receipt,
+				backing_group,
 				response_sender,
 			).await
 		}
@@ -543,6 +726,20 @@ async fn query_chunk(
 	Ok(rx.await.map_err(error::Error::CanceledQueryChunk)?)
 }
 
+/// Queries a chunk from av-store.
+#[tracing::instrument(level = "trace", skip(ctx), fields(subsystem = LOG_TARGET))]
+async fn query_full_data(
+	ctx: &mut impl SubsystemContext<Message = AvailabilityRecoveryMessage>,
+	candidate_hash: CandidateHash,
+) -> error::Result<Option<AvailableData>> {
+	let (tx, rx) = oneshot::channel();
+	ctx.send_message(AllMessages::AvailabilityStore(
+		AvailabilityStoreMessage::QueryAvailableData(candidate_hash, tx),
+	)).await;
+
+	Ok(rx.await.map_err(error::Error::CanceledQueryFullData)?)
+}
+
 /// Handles message from interaction.
 #[tracing::instrument(level = "trace", skip(ctx, state), fields(subsystem = LOG_TARGET))]
 async fn handle_from_interaction(
@@ -574,7 +771,7 @@ async fn handle_from_interaction(
 
 			state.availability_lru.put(candidate_hash, result);
 		}
-		FromInteraction::MakeRequest(id, candidate_hash, validator_index, response) => {
+		FromInteraction::MakeChunkRequest(id, candidate_hash, validator_index, response) => {
 			let (tx, rx) = mpsc::channel(2);
 
 			let message = NetworkBridgeMessage::ConnectToValidators {
@@ -587,12 +784,33 @@ async fn handle_from_interaction(
 
 			let token = state.connecting_validators.push(rx);
 
-			state.discovering_validators.entry(id).or_default().push(AwaitedChunk {
+			state.discovering_validators.entry(id).or_default().push(Awaited::Chunk(AwaitedData {
 				validator_index,
 				candidate_hash,
 				token,
 				response,
-			});
+			}));
+		}
+		FromInteraction::MakeFullDataRequest(id, candidate_hash, validator_index, response) => {
+			let (tx, rx) = mpsc::channel(2);
+
+			let message = NetworkBridgeMessage::ConnectToValidators {
+				validator_ids: vec![id.clone()],
+				peer_set: PeerSet::Validation,
+				connected: tx,
+			};
+
+			ctx.send_message(AllMessages::NetworkBridge(message)).await;
+
+			let token = state.connecting_validators.push(rx);
+
+			println!("pushing full data request");
+			state.discovering_validators.entry(id).or_default().push(Awaited::FullData(AwaitedData {
+				validator_index,
+				candidate_hash,
+				token,
+				response,
+			}));
 		}
 		FromInteraction::ReportPeer(peer_id, rep) => {
 			report_peer(ctx, peer_id, rep).await;
@@ -637,16 +855,18 @@ async fn handle_network_update(
 					)).await;
 				}
 				protocol_v1::AvailabilityRecoveryMessage::Chunk(request_id, chunk) => {
-					match state.live_chunk_requests.remove(&request_id) {
+					match state.live_requests.remove(&request_id) {
 						None => {
 							// If there doesn't exist one, report the peer and return.
 							report_peer(ctx, peer, COST_UNEXPECTED_CHUNK).await;
 						}
-						Some((peer_id, awaited_chunk)) if peer_id == peer => {
+						Some((peer_id, Awaited::Chunk(awaited_chunk))) if peer_id == peer => {
 							// If there exists an entry under r_id, remove it.
 							// Send the chunk response on the awaited_chunk for the interaction to handle.
 							if let Some(chunk) = chunk {
-								if awaited_chunk.response.send(Ok((peer_id, chunk))).is_err() {
+								if awaited_chunk.response.send(
+									(peer_id, awaited_chunk.validator_index, chunk)
+								).is_err() {
 									tracing::debug!(
 										target: LOG_TARGET,
 										"A sending side of the recovery request is closed",
@@ -657,14 +877,59 @@ async fn handle_network_update(
 						Some(a) => {
 							// If the peer in the entry doesn't match the sending peer,
 							// reinstate the entry, report the peer, and return
-							state.live_chunk_requests.insert(request_id, a);
+							state.live_requests.insert(request_id, a);
 							report_peer(ctx, peer, COST_UNEXPECTED_CHUNK).await;
 						}
 					}
 				}
-				protocol_v1::AvailabilityRecoveryMessage::RequestFullData(_, _) |
-				protocol_v1::AvailabilityRecoveryMessage::FullData(_, _) => {
-					// handled in https://github.com/paritytech/polkadot/pull/2453
+				protocol_v1::AvailabilityRecoveryMessage::RequestFullData(
+					request_id,
+					candidate_hash,
+				) => {
+					// Issue a
+					// AvailabilityStore::QueryAvailableData(candidate-hash, response)
+					// message.
+					let full_data = query_full_data(ctx, candidate_hash).await?;
+
+					// Whatever the result, issue an
+					// AvailabilityRecoveryV1Message::FullData(r_id, response) message.
+					let wire_message = protocol_v1::AvailabilityRecoveryMessage::FullData(
+						request_id,
+						full_data,
+					);
+
+					ctx.send_message(AllMessages::NetworkBridge(
+						NetworkBridgeMessage::SendValidationMessage(
+							vec![peer],
+							protocol_v1::ValidationProtocol::AvailabilityRecovery(wire_message),
+						),
+					)).await;
+				}
+				protocol_v1::AvailabilityRecoveryMessage::FullData(request_id, data) => {
+					match state.live_requests.remove(&request_id) {
+						None => {
+							// If there doesn't exist one, report the peer and return.
+							report_peer(ctx, peer, COST_UNEXPECTED_CHUNK).await;
+						}
+						Some((peer_id, Awaited::FullData(awaited))) if peer_id == peer => {
+							// If there exists an entry under r_id, remove it.
+							// Send the response on the awaited for the interaction to handle.
+							if let Some(data) = data {
+								if awaited.response.send((peer_id, awaited.validator_index, data)).is_err() {
+									tracing::debug!(
+										target: LOG_TARGET,
+										"A sending side of the recovery request is closed",
+									);
+								}
+							}
+						}
+						Some(a) => {
+							// If the peer in the entry doesn't match the sending peer,
+							// reinstate the entry, report the peer, and return
+							state.live_requests.insert(request_id, a);
+							report_peer(ctx, peer, COST_UNEXPECTED_CHUNK).await;
+						}
+					}
 				}
 			}
 		}
@@ -680,21 +945,27 @@ async fn handle_network_update(
 	Ok(())
 }
 
-/// Issues a chunk request to the validator we've been waiting for to connect to us.
-async fn issue_chunk_request(
+/// Issues a request to the validator we've been waiting for to connect to us.
+async fn issue_request(
 	state: &mut State,
 	ctx: &mut impl SubsystemContext<Message = AvailabilityRecoveryMessage>,
 	peer_id: PeerId,
-	awaited_chunk: AwaitedChunk,
+	awaited: Awaited,
 ) -> error::Result<()> {
 	let request_id = state.next_request_id;
 	state.next_request_id += 1;
 
-	let wire_message = protocol_v1::AvailabilityRecoveryMessage::RequestChunk(
-		request_id,
-		awaited_chunk.candidate_hash,
-		awaited_chunk.validator_index,
-	);
+	let wire_message = match awaited {
+		Awaited::Chunk(ref awaited_chunk) => protocol_v1::AvailabilityRecoveryMessage::RequestChunk(
+			request_id,
+			awaited_chunk.candidate_hash,
+			awaited_chunk.validator_index,
+		),
+		Awaited::FullData(ref awaited_data) => protocol_v1::AvailabilityRecoveryMessage::RequestFullData(
+			request_id,
+			awaited_data.candidate_hash,
+		),
+	};
 
 	ctx.send_message(AllMessages::NetworkBridge(
 		NetworkBridgeMessage::SendValidationMessage(
@@ -703,7 +974,7 @@ async fn issue_chunk_request(
 		),
 	)).await;
 
-	state.live_chunk_requests.insert(request_id, (peer_id, awaited_chunk));
+	state.live_requests.insert(request_id, (peer_id, awaited));
 
 	Ok(())
 }
@@ -716,23 +987,23 @@ async fn handle_validator_connected(
 	peer_id: PeerId,
 ) -> error::Result<()> {
 	if let Some(discovering) = state.discovering_validators.remove(&authority_id) {
-		for chunk in discovering {
-			issue_chunk_request(state, ctx, peer_id.clone(), chunk).await?;
+		for awaited in discovering {
+			issue_request(state, ctx, peer_id.clone(), awaited).await?;
 		}
 	}
 
 	Ok(())
 }
 
-/// Awaited chunks info that `State` holds has to be cleaned up
+/// Awaited info that `State` holds has to be cleaned up
 /// periodically since there is no way `Interaction` can communicate
 /// a timedout request.
-fn cleanup_awaited_chunks(state: &mut State) {
+fn cleanup_awaited(state: &mut State) {
 	let mut removed_tokens = Vec::new();
 
 	for (_, v) in state.discovering_validators.iter_mut() {
-		v.retain(|e| if e.response.is_canceled() {
-			removed_tokens.push(e.token);
+		v.retain(|e| if e.is_canceled() {
+			removed_tokens.push(e.token());
 			false
 		} else {
 			true
@@ -744,13 +1015,18 @@ fn cleanup_awaited_chunks(state: &mut State) {
 	}
 
 	state.discovering_validators.retain(|_, v| !v.is_empty());
-	state.live_chunk_requests.retain(|_, v| !v.1.response.is_canceled());
+	state.live_requests.retain(|_, v| !v.1.is_canceled());
 }
 
 impl AvailabilityRecoverySubsystem {
-	/// Create a new instance of `AvailabilityRecoverySubsystem`.
-	pub fn new() -> Self {
-		Self
+	/// Create a new instance of `AvailabilityRecoverySubsystem` which starts with a fast path to request data from backers.
+	pub fn with_fast_path() -> Self {
+		Self { fast_path: true }
+	}
+
+	/// Create a new instance of `AvailabilityRecoverySubsystem` which requests only chunks
+	pub fn with_chunks_only() -> Self {
+		Self { fast_path: false }
 	}
 
 	async fn run(
@@ -759,16 +1035,16 @@ impl AvailabilityRecoverySubsystem {
 	) -> SubsystemResult<()> {
 		let mut state = State::default();
 
-		let awaited_chunk_cleanup_interval = futures::stream::repeat(()).then(|_| async move {
-			Delay::new(AWAITED_CHUNKS_CLEANUP_INTERVAL).await;
+		let awaited_cleanup_interval = futures::stream::repeat(()).then(|_| async move {
+			Delay::new(AWAITED_CLEANUP_INTERVAL).await;
 		});
 
-		futures::pin_mut!(awaited_chunk_cleanup_interval);
+		futures::pin_mut!(awaited_cleanup_interval);
 
 		loop {
 			futures::select_biased! {
-				_v = awaited_chunk_cleanup_interval.next() => {
-					cleanup_awaited_chunks(&mut state);
+				_v = awaited_cleanup_interval.next() => {
+					cleanup_awaited(&mut state);
 				}
 				v = state.connecting_validators.next() => {
 					if let Some((v, token)) = v {
@@ -806,6 +1082,7 @@ impl AvailabilityRecoverySubsystem {
 								AvailabilityRecoveryMessage::RecoverAvailableData(
 									receipt,
 									session_index,
+									maybe_backing_group,
 									response_sender,
 								) => {
 									if let Err(e) = handle_recover(
@@ -813,6 +1090,7 @@ impl AvailabilityRecoverySubsystem {
 										&mut ctx,
 										receipt,
 										session_index,
+										maybe_backing_group.filter(|_| self.fast_path),
 										response_sender,
 									).await {
 										tracing::warn!(