add fallback request for req-response protocols (#2771)

Previously, it was only possible to retry the same request on a
different protocol name that had the exact same binary payloads.

Introduce a way of trying a different request on a different protocol if
the first one fails with Unsupported protocol.

This helps with adding new req-response versions in polkadot while
preserving compatibility with unupgraded nodes.

The way req-response protocols were bumped previously was that they were
bundled with some other notifications protocol upgrade, like for async
backing (but that is more complicated, especially if the feature does
not require any changes to a notifications protocol). Will be needed for
implementing https://github.com/polkadot-fellows/RFCs/pull/47

TODO:
- [x]  add tests
- [x] add guidance docs in polkadot about req-response protocol
versioning

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

@@ -22,13 +22,14 @@ use futures_timer::Delay;
 
 use parity_scale_codec::Encode;
 use polkadot_node_network_protocol::request_response::{
-	self as req_res, IncomingRequest, Recipient, ReqProtocolNames, Requests,
+	self as req_res, v1::AvailableDataFetchingRequest, IncomingRequest, Protocol, Recipient,
+	ReqProtocolNames, Requests,
 };
 use polkadot_node_subsystem_test_helpers::derive_erasure_chunks_with_proofs_and_root;
 
 use super::*;
 
-use sc_network::{config::RequestResponseConfig, IfDisconnected, OutboundFailure, RequestFailure};
+use sc_network::{IfDisconnected, OutboundFailure, ProtocolName, RequestFailure};
 
 use polkadot_node_primitives::{BlockData, PoV, Proof};
 use polkadot_node_subsystem::messages::{
@@ -48,8 +49,18 @@ type VirtualOverseer = TestSubsystemContextHandle<AvailabilityRecoveryMessage>;
 // Deterministic genesis hash for protocol names
 const GENESIS_HASH: Hash = Hash::repeat_byte(0xff);
 
-fn test_harness_fast_path<T: Future<Output = (VirtualOverseer, RequestResponseConfig)>>(
-	test: impl FnOnce(VirtualOverseer, RequestResponseConfig) -> T,
+fn request_receiver(
+	req_protocol_names: &ReqProtocolNames,
+) -> IncomingRequestReceiver<AvailableDataFetchingRequest> {
+	let receiver = IncomingRequest::get_config_receiver(req_protocol_names);
+	// Don't close the sending end of the request protocol. Otherwise, the subsystem will terminate.
+	std::mem::forget(receiver.1.inbound_queue);
+	receiver.0
+}
+
+fn test_harness<T: Future<Output = VirtualOverseer>>(
+	subsystem: AvailabilityRecoverySubsystem,
+	test: impl FnOnce(VirtualOverseer) -> T,
 ) {
 	let _ = env_logger::builder()
 		.is_test(true)
@@ -60,22 +71,18 @@ fn test_harness_fast_path<T: Future<Output = (VirtualOverseer, RequestResponseCo
 
 	let (context, virtual_overseer) = make_subsystem_context(pool.clone());
 
-	let (collation_req_receiver, req_cfg) =
-		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(&GENESIS_HASH, None));
-	let subsystem =
-		AvailabilityRecoverySubsystem::with_fast_path(collation_req_receiver, Metrics::new_dummy());
 	let subsystem = async {
 		subsystem.run(context).await.unwrap();
 	};
 
-	let test_fut = test(virtual_overseer, req_cfg);
+	let test_fut = test(virtual_overseer);
 
 	futures::pin_mut!(test_fut);
 	futures::pin_mut!(subsystem);
 
 	executor::block_on(future::join(
 		async move {
-			let (mut overseer, _req_cfg) = test_fut.await;
+			let mut overseer = test_fut.await;
 			overseer_signal(&mut overseer, OverseerSignal::Conclude).await;
 		},
 		subsystem,
@@ -83,80 +90,6 @@ fn test_harness_fast_path<T: Future<Output = (VirtualOverseer, RequestResponseCo
 	.1
 }
 
-fn test_harness_chunks_only<T: Future<Output = (VirtualOverseer, RequestResponseConfig)>>(
-	test: impl FnOnce(VirtualOverseer, RequestResponseConfig) -> T,
-) {
-	let _ = env_logger::builder()
-		.is_test(true)
-		.filter(Some("polkadot_availability_recovery"), log::LevelFilter::Trace)
-		.try_init();
-
-	let pool = sp_core::testing::TaskExecutor::new();
-
-	let (context, virtual_overseer) = make_subsystem_context(pool.clone());
-
-	let (collation_req_receiver, req_cfg) =
-		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(&GENESIS_HASH, None));
-	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
-		collation_req_receiver,
-		Metrics::new_dummy(),
-	);
-	let subsystem = subsystem.run(context);
-
-	let test_fut = test(virtual_overseer, req_cfg);
-
-	futures::pin_mut!(test_fut);
-	futures::pin_mut!(subsystem);
-
-	executor::block_on(future::join(
-		async move {
-			let (mut overseer, _req_cfg) = test_fut.await;
-			overseer_signal(&mut overseer, OverseerSignal::Conclude).await;
-		},
-		subsystem,
-	))
-	.1
-	.unwrap();
-}
-
-fn test_harness_chunks_if_pov_large<
-	T: Future<Output = (VirtualOverseer, RequestResponseConfig)>,
->(
-	test: impl FnOnce(VirtualOverseer, RequestResponseConfig) -> T,
-) {
-	let _ = env_logger::builder()
-		.is_test(true)
-		.filter(Some("polkadot_availability_recovery"), log::LevelFilter::Trace)
-		.try_init();
-
-	let pool = sp_core::testing::TaskExecutor::new();
-
-	let (context, virtual_overseer) = make_subsystem_context(pool.clone());
-
-	let (collation_req_receiver, req_cfg) =
-		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(&GENESIS_HASH, None));
-	let subsystem = AvailabilityRecoverySubsystem::with_chunks_if_pov_large(
-		collation_req_receiver,
-		Metrics::new_dummy(),
-	);
-	let subsystem = subsystem.run(context);
-
-	let test_fut = test(virtual_overseer, req_cfg);
-
-	futures::pin_mut!(test_fut);
-	futures::pin_mut!(subsystem);
-
-	executor::block_on(future::join(
-		async move {
-			let (mut overseer, _req_cfg) = test_fut.await;
-			overseer_signal(&mut overseer, OverseerSignal::Conclude).await;
-		},
-		subsystem,
-	))
-	.1
-	.unwrap();
-}
-
 const TIMEOUT: Duration = Duration::from_millis(300);
 
 macro_rules! delay {
@@ -342,11 +275,12 @@ impl TestState {
 
 	async fn test_chunk_requests(
 		&self,
+		req_protocol_names: &ReqProtocolNames,
 		candidate_hash: CandidateHash,
 		virtual_overseer: &mut VirtualOverseer,
 		n: usize,
 		who_has: impl Fn(usize) -> Has,
-	) -> Vec<oneshot::Sender<std::result::Result<Vec<u8>, RequestFailure>>> {
+	) -> Vec<oneshot::Sender<std::result::Result<(Vec<u8>, ProtocolName), RequestFailure>>> {
 		// arbitrary order.
 		let mut i = 0;
 		let mut senders = Vec::new();
@@ -380,7 +314,7 @@ impl TestState {
 
 								let _ = req.pending_response.send(
 									available_data.map(|r|
-										req_res::v1::ChunkFetchingResponse::from(r).encode()
+										(req_res::v1::ChunkFetchingResponse::from(r).encode(), req_protocol_names.get_name(Protocol::ChunkFetchingV1))
 									)
 								);
 							}
@@ -394,10 +328,11 @@ impl TestState {
 
 	async fn test_full_data_requests(
 		&self,
+		req_protocol_names: &ReqProtocolNames,
 		candidate_hash: CandidateHash,
 		virtual_overseer: &mut VirtualOverseer,
 		who_has: impl Fn(usize) -> Has,
-	) -> Vec<oneshot::Sender<std::result::Result<Vec<u8>, RequestFailure>>> {
+	) -> Vec<oneshot::Sender<std::result::Result<(Vec<u8>, ProtocolName), RequestFailure>>> {
 		let mut senders = Vec::new();
 		for _ in 0..self.validators.len() {
 			// Receive a request for a chunk.
@@ -433,9 +368,10 @@ impl TestState {
 							let done = available_data.as_ref().ok().map_or(false, |x| x.is_some());
 
 							let _ = req.pending_response.send(
-								available_data.map(|r|
-									req_res::v1::AvailableDataFetchingResponse::from(r).encode()
-								)
+								available_data.map(|r|(
+									req_res::v1::AvailableDataFetchingResponse::from(r).encode(),
+									req_protocol_names.get_name(Protocol::AvailableDataFetchingV1)
+								))
 							);
 
 							if done { break }
@@ -532,8 +468,13 @@ impl Default for TestState {
 #[test]
 fn availability_is_recovered_from_chunks_if_no_group_provided() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -565,6 +506,7 @@ fn availability_is_recovered_from_chunks_if_no_group_provided() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -600,6 +542,7 @@ fn availability_is_recovered_from_chunks_if_no_group_provided() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				new_candidate.hash(),
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -609,15 +552,20 @@ fn availability_is_recovered_from_chunks_if_no_group_provided() {
 
 		// A request times out with `Unavailable` error.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunks_only() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -649,6 +597,7 @@ fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunk
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -684,6 +633,7 @@ fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunk
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				new_candidate.hash(),
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -693,15 +643,20 @@ fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunk
 
 		// A request times out with `Unavailable` error.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn bad_merkle_path_leads_to_recovery_error() {
 	let mut test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -740,6 +695,7 @@ fn bad_merkle_path_leads_to_recovery_error() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -749,15 +705,20 @@ fn bad_merkle_path_leads_to_recovery_error() {
 
 		// A request times out with `Unavailable` error.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn wrong_chunk_index_leads_to_recovery_error() {
 	let mut test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -796,6 +757,7 @@ fn wrong_chunk_index_leads_to_recovery_error() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -805,15 +767,20 @@ fn wrong_chunk_index_leads_to_recovery_error() {
 
 		// A request times out with `Unavailable` error as there are no good peers.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn invalid_erasure_coding_leads_to_invalid_error() {
 	let mut test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		let pov = PoV { block_data: BlockData(vec![69; 64]) };
 
 		let (bad_chunks, bad_erasure_root) = derive_erasure_chunks_with_proofs_and_root(
@@ -859,6 +826,7 @@ fn invalid_erasure_coding_leads_to_invalid_error() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -868,15 +836,20 @@ fn invalid_erasure_coding_leads_to_invalid_error() {
 
 		// f+1 'valid' chunks can't produce correct data.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Invalid);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn fast_path_backing_group_recovers() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -911,20 +884,30 @@ fn fast_path_backing_group_recovers() {
 		test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
 
 		test_state
-			.test_full_data_requests(candidate_hash, &mut virtual_overseer, who_has)
+			.test_full_data_requests(
+				&req_protocol_names,
+				candidate_hash,
+				&mut virtual_overseer,
+				who_has,
+			)
 			.await;
 
 		// Recovered data should match the original one.
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn recovers_from_only_chunks_if_pov_large() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_if_pov_large(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_if_pov_large(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -965,6 +948,7 @@ fn recovers_from_only_chunks_if_pov_large() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -1009,6 +993,7 @@ fn recovers_from_only_chunks_if_pov_large() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				new_candidate.hash(),
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -1018,15 +1003,20 @@ fn recovers_from_only_chunks_if_pov_large() {
 
 		// A request times out with `Unavailable` error.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn fast_path_backing_group_recovers_if_pov_small() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_if_pov_large(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_if_pov_large(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1070,20 +1060,30 @@ fn fast_path_backing_group_recovers_if_pov_small() {
 		test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
 
 		test_state
-			.test_full_data_requests(candidate_hash, &mut virtual_overseer, who_has)
+			.test_full_data_requests(
+				&req_protocol_names,
+				candidate_hash,
+				&mut virtual_overseer,
+				who_has,
+			)
 			.await;
 
 		// Recovered data should match the original one.
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn no_answers_in_fast_path_causes_chunk_requests() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_fast_path(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_fast_path(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1119,13 +1119,19 @@ fn no_answers_in_fast_path_causes_chunk_requests() {
 		test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
 
 		test_state
-			.test_full_data_requests(candidate_hash, &mut virtual_overseer, who_has)
+			.test_full_data_requests(
+				&req_protocol_names,
+				candidate_hash,
+				&mut virtual_overseer,
+				who_has,
+			)
 			.await;
 
 		test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -1135,15 +1141,20 @@ fn no_answers_in_fast_path_causes_chunk_requests() {
 
 		// Recovered data should match the original one.
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn task_canceled_when_receivers_dropped() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1170,7 +1181,7 @@ fn task_canceled_when_receivers_dropped() {
 
 		for _ in 0..test_state.validators.len() {
 			match virtual_overseer.recv().timeout(TIMEOUT).await {
-				None => return (virtual_overseer, req_cfg),
+				None => return virtual_overseer,
 				Some(_) => continue,
 			}
 		}
@@ -1182,8 +1193,13 @@ fn task_canceled_when_receivers_dropped() {
 #[test]
 fn chunks_retry_until_all_nodes_respond() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1215,6 +1231,7 @@ fn chunks_retry_until_all_nodes_respond() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.validators.len() - test_state.threshold(),
@@ -1225,6 +1242,7 @@ fn chunks_retry_until_all_nodes_respond() {
 		// we get to go another round!
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.impossibility_threshold(),
@@ -1234,15 +1252,20 @@ fn chunks_retry_until_all_nodes_respond() {
 
 		// Recovered data should match the original one.
 		assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn not_returning_requests_wont_stall_retrieval() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1277,13 +1300,18 @@ fn not_returning_requests_wont_stall_retrieval() {
 
 		// Not returning senders won't cause the retrieval to stall:
 		let _senders = test_state
-			.test_chunk_requests(candidate_hash, &mut virtual_overseer, not_returning_count, |_| {
-				Has::DoesNotReturn
-			})
+			.test_chunk_requests(
+				&req_protocol_names,
+				candidate_hash,
+				&mut virtual_overseer,
+				not_returning_count,
+				|_| Has::DoesNotReturn,
+			)
 			.await;
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				// Should start over:
@@ -1295,6 +1323,7 @@ fn not_returning_requests_wont_stall_retrieval() {
 		// we get to go another round!
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -1304,15 +1333,20 @@ fn not_returning_requests_wont_stall_retrieval() {
 
 		// Recovered data should match the original one:
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn all_not_returning_requests_still_recovers_on_return() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1344,6 +1378,7 @@ fn all_not_returning_requests_still_recovers_on_return() {
 
 		let senders = test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.validators.len(),
@@ -1358,6 +1393,7 @@ fn all_not_returning_requests_still_recovers_on_return() {
 				std::mem::drop(senders);
 			},
 			test_state.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				// Should start over:
@@ -1370,6 +1406,7 @@ fn all_not_returning_requests_still_recovers_on_return() {
 		// we get to go another round!
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold(),
@@ -1379,15 +1416,20 @@ fn all_not_returning_requests_still_recovers_on_return() {
 
 		// Recovered data should match the original one:
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn returns_early_if_we_have_the_data() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1414,15 +1456,20 @@ fn returns_early_if_we_have_the_data() {
 		test_state.respond_to_available_data_query(&mut virtual_overseer, true).await;
 
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn does_not_query_local_validator() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1453,6 +1500,7 @@ fn does_not_query_local_validator() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.validators.len(),
@@ -1463,6 +1511,7 @@ fn does_not_query_local_validator() {
 		// second round, make sure it uses the local chunk.
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold() - 1,
@@ -1471,15 +1520,20 @@ fn does_not_query_local_validator() {
 			.await;
 
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }
 
 #[test]
 fn invalid_local_chunk_is_ignored() {
 	let test_state = TestState::default();
+	let req_protocol_names = ReqProtocolNames::new(&GENESIS_HASH, None);
+	let subsystem = AvailabilityRecoverySubsystem::with_chunks_only(
+		request_receiver(&req_protocol_names),
+		Metrics::new_dummy(),
+	);
 
-	test_harness_chunks_only(|mut virtual_overseer, req_cfg| async move {
+	test_harness(subsystem, |mut virtual_overseer| async move {
 		overseer_signal(
 			&mut virtual_overseer,
 			OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(new_leaf(
@@ -1512,6 +1566,7 @@ fn invalid_local_chunk_is_ignored() {
 
 		test_state
 			.test_chunk_requests(
+				&req_protocol_names,
 				candidate_hash,
 				&mut virtual_overseer,
 				test_state.threshold() - 1,
@@ -1520,6 +1575,6 @@ fn invalid_local_chunk_is_ignored() {
 			.await;
 
 		assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
-		(virtual_overseer, req_cfg)
+		virtual_overseer
 	});
 }