Generic request/response infrastructure for Polkadot (#2352)

* Move NetworkBridgeEvent to subsystem::messages.

It is not protocol related at all, it is in fact only part of the
subsystem communication as it gets wrapped into messages of each
subsystem.

* Request/response infrastructure is taking shape.

WIP: Does not compile.

* Multiplexer variant not supported by Rusts type system.

* request_response::request type checks.

* Cleanup.

* Minor fixes for request_response.

* Implement request sending + move multiplexer.

Request multiplexer is moved to bridge, as there the implementation is
more straight forward as we can specialize on `AllMessages` for the
multiplexing target.

Sending of requests is mostly complete, apart from a few `From`
instances. Receiving is also almost done, initializtion needs to be
fixed and the multiplexer needs to be invoked.

* Remove obsolete multiplexer.

* Initialize bridge with multiplexer.

* Finish generic request sending/receiving.

Subsystems are now able to receive and send requests and responses via
the overseer.

* Doc update.

* Fixes.

* Link issue for not yet implemented code.

* Fixes suggested by @ordian - thanks!

- start encoding at 0
- don't crash on zero protocols
- don't panic on not yet implemented request handling

* Update node/network/protocol/src/request_response/v1.rs

Use index 0 instead of 1.

Co-authored-by: Andronik Ordian <write@reusable.software>

* Update node/network/protocol/src/request_response.rs

Co-authored-by: Andronik Ordian <write@reusable.software>

* Fix existing tests.

* Better avoidance of division by zoro errors.

* Doc fixes.

* send_request -> start_request.

* Fix missing renamings.

* Update substrate.

* Pass TryConnect instead of true.

* Actually import `IfDisconnected`.

* Fix wrong import.

* Update node/network/bridge/src/lib.rs

typo

Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update node/network/bridge/src/multiplexer.rs

Remove redundant import.

Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>

* Stop doing tracing from within `From` instance.

Thanks for the catch @tomaka!

* Get rid of redundant import.

* Formatting cleanup.

* Fix tests.

* Add link to issue.

* Clarify comments some more.

* Fix tests.

* Formatting fix.

* tabs

* Fix link

Co-authored-by: Bernhard Schuster <bernhard@ahoi.io>

* Use map_err.

Co-authored-by: Bernhard Schuster <bernhard@ahoi.io>

* Improvements inspired by suggestions by @drahnr.

- Channel size is now determined by function.
- Explicitely scope NetworkService::start_request.

Co-authored-by: Andronik Ordian <write@reusable.software>
Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>
Co-authored-by: Bernhard Schuster <bernhard@ahoi.io>

polkadot/node/network/approval-distribution/src/lib.rs

@@ -35,13 +35,13 @@ use polkadot_node_primitives::{
 use polkadot_node_subsystem::{
 	messages::{
 		AllMessages, ApprovalDistributionMessage, ApprovalVotingMessage, NetworkBridgeMessage,
-		AssignmentCheckResult, ApprovalCheckResult,
+		AssignmentCheckResult, ApprovalCheckResult, NetworkBridgeEvent,
 	},
 	ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem, SubsystemContext,
 };
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
 use polkadot_node_network_protocol::{
-	PeerId, View, NetworkBridgeEvent, v1 as protocol_v1, ReputationChange as Rep,
+	PeerId, View, v1 as protocol_v1, ReputationChange as Rep,
 };
 
 const LOG_TARGET: &str = "approval_distribution";

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

@@ -32,7 +32,7 @@ use sp_keystore::{CryptoStore, SyncCryptoStorePtr};
 
 use polkadot_erasure_coding::branch_hash;
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, NetworkBridgeEvent, PeerId, ReputationChange as Rep, View, OurView,
+	v1 as protocol_v1, PeerId, ReputationChange as Rep, View, OurView,
 };
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
 use polkadot_primitives::v1::{
@@ -42,7 +42,7 @@ use polkadot_primitives::v1::{
 };
 use polkadot_subsystem::messages::{
 	AllMessages, AvailabilityDistributionMessage, AvailabilityStoreMessage, ChainApiMessage,
-	NetworkBridgeMessage, RuntimeApiMessage, RuntimeApiRequest,
+	NetworkBridgeMessage, RuntimeApiMessage, RuntimeApiRequest, NetworkBridgeEvent
 };
 use polkadot_subsystem::{
 	jaeger, errors::{ChainApiError, RuntimeApiError}, PerLeafSpan,
@@ -843,6 +843,15 @@ impl AvailabilityDistributionSubsystem {
 						);
 					}
 				}
+				FromOverseer::Communication {
+					msg: AvailabilityDistributionMessage::AvailabilityFetchingRequest(_),
+				} => { 
+					// TODO: Implement issue 2306:
+					tracing::warn!(
+						target: LOG_TARGET,
+						"To be implemented, see: https://github.com/paritytech/polkadot/issues/2306 !",
+					);
+				}
 				FromOverseer::Signal(OverseerSignal::ActiveLeaves(ActiveLeavesUpdate {
 					activated: _,
 					deactivated: _,

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

@@ -39,10 +39,11 @@ use polkadot_subsystem::{
 	errors::RecoveryError,
 	messages::{
 		AvailabilityStoreMessage, AvailabilityRecoveryMessage, AllMessages, NetworkBridgeMessage,
+		NetworkBridgeEvent,
 	},
 };
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, NetworkBridgeEvent, PeerId, ReputationChange as Rep, RequestId,
+	v1 as protocol_v1, PeerId, ReputationChange as Rep, RequestId,
 };
 use polkadot_node_subsystem_util::{
 	Timeout, TimeoutExt,

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

@@ -30,7 +30,7 @@ use polkadot_primitives::v1::{
 use polkadot_erasure_coding::{branches, obtain_chunks_v1 as obtain_chunks};
 use polkadot_node_subsystem_util::TimeoutExt;
 use polkadot_subsystem_testhelpers as test_helpers;
-use polkadot_subsystem::{messages::{RuntimeApiMessage, RuntimeApiRequest}, JaegerSpan};
+use polkadot_subsystem::{messages::{RuntimeApiMessage, RuntimeApiRequest, NetworkBridgeEvent}, JaegerSpan};
 
 type VirtualOverseer = test_helpers::TestSubsystemContextHandle<AvailabilityRecoveryMessage>;
 

polkadot/node/network/bitfield-distribution/src/lib.rs

@@ -27,12 +27,12 @@ use futures::{channel::oneshot, FutureExt};
 
 use polkadot_subsystem::messages::*;
 use polkadot_subsystem::{
-	PerLeafSpan, ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem, SubsystemContext,
-	SubsystemResult,
+	PerLeafSpan, ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem,
+	SubsystemContext, SubsystemResult,
 };
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
 use polkadot_primitives::v1::{Hash, SignedAvailabilityBitfield, SigningContext, ValidatorId};
-use polkadot_node_network_protocol::{v1 as protocol_v1, PeerId, NetworkBridgeEvent, View, ReputationChange, OurView};
+use polkadot_node_network_protocol::{v1 as protocol_v1, PeerId, View, ReputationChange, OurView};
 use std::collections::{HashMap, HashSet};
 
 const COST_SIGNATURE_INVALID: ReputationChange =

polkadot/node/network/bridge/Cargo.toml

@@ -15,6 +15,7 @@ sc-authority-discovery = { git = "https://github.com/paritytech/substrate", bran
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
 polkadot-node-network-protocol = { path = "../protocol" }
+strum = "0.20.0"
 
 [dev-dependencies]
 assert_matches = "1.4.0"

polkadot/node/network/bridge/src/action.rs

@@ -22,13 +22,14 @@ use polkadot_node_network_protocol::{
 	peer_set::PeerSet, v1 as protocol_v1, PeerId, ReputationChange,
 };
 use polkadot_primitives::v1::{AuthorityDiscoveryId, BlockNumber};
-use polkadot_subsystem::messages::NetworkBridgeMessage;
+use polkadot_subsystem::messages::{AllMessages, NetworkBridgeMessage};
 use polkadot_subsystem::{ActiveLeavesUpdate, FromOverseer, OverseerSignal};
 use sc_network::Event as NetworkEvent;
 
-use polkadot_node_network_protocol::ObservedRole;
+use polkadot_node_network_protocol::{request_response::Requests, ObservedRole};
 
-use super::{WireMessage, LOG_TARGET, MALFORMED_MESSAGE_COST};
+use super::multiplexer::RequestMultiplexError;
+use super::{WireMessage, MALFORMED_MESSAGE_COST};
 
 /// Internal type combining all actions a `NetworkBridge` might perform.
 ///
@@ -43,6 +44,9 @@ pub(crate) enum Action {
 	/// Ask network to send a collation message.
 	SendCollationMessages(Vec<(Vec<PeerId>, protocol_v1::CollationProtocol)>),
 
+	/// Ask network to send requests.
+	SendRequests(Vec<Requests>),
+
 	/// Ask network to connect to validators.
 	ConnectToValidators {
 		validator_ids: Vec<AuthorityDiscoveryId>,
@@ -76,13 +80,32 @@ pub(crate) enum Action {
 		Vec<WireMessage<protocol_v1::CollationProtocol>>,
 	),
 
-	Abort,
+	/// Send a message to another subsystem or the overseer.
+	///
+	/// Used for handling incoming requests.
+	SendMessage(AllMessages),
+
+	/// Abort with reason.
+	Abort(AbortReason),
 	Nop,
 }
 
+#[derive(Debug)]
+pub(crate) enum AbortReason {
+	/// Received error from overseer:
+	SubsystemError(polkadot_subsystem::SubsystemError),
+	/// The stream of incoming events concluded.
+	EventStreamConcluded,
+	/// The stream of incoming requests concluded.
+	RequestStreamConcluded,
+	/// We received OverseerSignal::Conclude
+	OverseerConcluded,
+}
+
 impl From<polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>>> for Action {
-	#[tracing::instrument(level = "trace", fields(subsystem = LOG_TARGET))]
-	fn from(res: polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>>) -> Self {
+	fn from(
+		res: polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>>,
+	) -> Self {
 		match res {
 			Ok(FromOverseer::Signal(OverseerSignal::ActiveLeaves(active_leaves))) => {
 				Action::ActiveLeaves(active_leaves)
@@ -90,7 +113,9 @@ impl From<polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>
 			Ok(FromOverseer::Signal(OverseerSignal::BlockFinalized(_hash, number))) => {
 				Action::BlockFinalized(number)
 			}
-			Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => Action::Abort,
+			Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => {
+				Action::Abort(AbortReason::OverseerConcluded)
+			}
 			Ok(FromOverseer::Communication { msg }) => match msg {
 				NetworkBridgeMessage::ReportPeer(peer, rep) => Action::ReportPeer(peer, rep),
 				NetworkBridgeMessage::SendValidationMessage(peers, msg) => {
@@ -99,6 +124,7 @@ impl From<polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>
 				NetworkBridgeMessage::SendCollationMessage(peers, msg) => {
 					Action::SendCollationMessages(vec![(peers, msg)])
 				}
+				NetworkBridgeMessage::SendRequests(reqs) => Action::SendRequests(reqs),
 				NetworkBridgeMessage::SendValidationMessages(msgs) => {
 					Action::SendValidationMessages(msgs)
 				}
@@ -113,25 +139,15 @@ impl From<polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>
 					connected,
 				},
 			},
-			Err(e) => {
-				tracing::warn!(target: LOG_TARGET, err = ?e, "Shutting down Network Bridge due to error");
-				Action::Abort
-			}
+			Err(e) => Action::Abort(AbortReason::SubsystemError(e)),
 		}
 	}
 }
 
 impl From<Option<NetworkEvent>> for Action {
-	#[tracing::instrument(level = "trace", fields(subsystem = LOG_TARGET))]
 	fn from(event: Option<NetworkEvent>) -> Action {
 		match event {
-			None => {
-				tracing::info!(
-					target: LOG_TARGET,
-					"Shutting down Network Bridge: underlying event stream concluded"
-				);
-				Action::Abort
-			}
+			None => Action::Abort(AbortReason::EventStreamConcluded),
 			Some(NetworkEvent::Dht(_))
 			| Some(NetworkEvent::SyncConnected { .. })
 			| Some(NetworkEvent::SyncDisconnected { .. }) => Action::Nop,
@@ -153,7 +169,9 @@ impl From<Option<NetworkEvent>> for Action {
 			Some(NetworkEvent::NotificationsReceived { remote, messages }) => {
 				let v_messages: Result<Vec<_>, _> = messages
 					.iter()
-					.filter(|(protocol, _)| protocol == &PeerSet::Validation.into_protocol_name())
+					.filter(|(protocol, _)| {
+						protocol == &PeerSet::Validation.into_protocol_name()
+					})
 					.map(|(_, msg_bytes)| WireMessage::decode(&mut msg_bytes.as_ref()))
 					.collect();
 
@@ -164,7 +182,9 @@ impl From<Option<NetworkEvent>> for Action {
 
 				let c_messages: Result<Vec<_>, _> = messages
 					.iter()
-					.filter(|(protocol, _)| protocol == &PeerSet::Collation.into_protocol_name())
+					.filter(|(protocol, _)| {
+						protocol == &PeerSet::Collation.into_protocol_name()
+					})
 					.map(|(_, msg_bytes)| WireMessage::decode(&mut msg_bytes.as_ref()))
 					.collect();
 
@@ -182,3 +202,13 @@ impl From<Option<NetworkEvent>> for Action {
 		}
 	}
 }
+
+impl From<Option<Result<AllMessages, RequestMultiplexError>>> for Action {
+	fn from(event: Option<Result<AllMessages, RequestMultiplexError>>) -> Self {
+		match event {
+			None => Action::Abort(AbortReason::RequestStreamConcluded),
+			Some(Err(err)) => Action::ReportPeer(err.peer, MALFORMED_MESSAGE_COST),
+			Some(Ok(msg)) => Action::SendMessage(msg),
+		}
+	}
+}

polkadot/node/network/bridge/src/lib.rs

@@ -30,11 +30,11 @@ use polkadot_subsystem::{
 use polkadot_subsystem::messages::{
 	NetworkBridgeMessage, AllMessages, AvailabilityDistributionMessage,
 	BitfieldDistributionMessage, PoVDistributionMessage, StatementDistributionMessage,
-	CollatorProtocolMessage, ApprovalDistributionMessage,
+	CollatorProtocolMessage, ApprovalDistributionMessage, NetworkBridgeEvent,
 };
 use polkadot_primitives::v1::{Hash, BlockNumber};
 use polkadot_node_network_protocol::{
-	ReputationChange, PeerId, peer_set::PeerSet, View, NetworkBridgeEvent, v1 as protocol_v1, OurView,
+	ReputationChange, PeerId, peer_set::PeerSet, View, v1 as protocol_v1, OurView,
 };
 
 /// Peer set infos for network initialization.
@@ -53,7 +53,7 @@ mod validator_discovery;
 /// All requested `NetworkBridgeMessage` user actions  and `NetworkEvent` network messages are
 /// translated to `Action` before being processed by `run_network`.
 mod action;
-use action::Action;
+use action::{Action, AbortReason};
 
 /// Actual interfacing to the network based on the `Network` trait.
 ///
@@ -61,6 +61,10 @@ use action::Action;
 mod network;
 use network::{Network, send_message};
 
+/// Request multiplexer for combining the multiple request sources into a single `Stream` of `AllMessages`.
+mod multiplexer;
+pub use multiplexer::RequestMultiplexer;
+
 
 /// The maximum amount of heads a peer is allowed to have in their view at any time.
 ///
@@ -95,6 +99,7 @@ pub struct NetworkBridge<N, AD> {
 	/// `Network` trait implementing type.
 	network_service: N,
 	authority_discovery_service: AD,
+	request_multiplexer: RequestMultiplexer,
 }
 
 impl<N, AD> NetworkBridge<N, AD> {
@@ -102,10 +107,11 @@ impl<N, AD> NetworkBridge<N, AD> {
 	///
 	/// This assumes that the network service has had the notifications protocol for the network
 	/// bridge already registered. See [`peers_sets_info`](peers_sets_info).
-	pub fn new(network_service: N, authority_discovery_service: AD) -> Self {
+	pub fn new(network_service: N, authority_discovery_service: AD, request_multiplexer: RequestMultiplexer) -> Self {
 		NetworkBridge {
 			network_service,
 			authority_discovery_service,
+			request_multiplexer,
 		}
 	}
 }
@@ -119,12 +125,7 @@ impl<Net, AD, Context> Subsystem<Context> for NetworkBridge<Net, AD>
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
 		// Swallow error because failure is fatal to the node and we log with more precision
 		// within `run_network`.
-		let Self { network_service, authority_discovery_service } = self;
-		let future = run_network(
-				network_service,
-				authority_discovery_service,
-				ctx,
-			)
+		let future = run_network(self, ctx)
 			.map_err(|e| {
 				SubsystemError::with_origin("network-bridge", e)
 			})
@@ -142,17 +143,16 @@ struct PeerData {
 }
 
 /// Main driver, processing network events and messages from other subsystems.
-#[tracing::instrument(skip(network_service, authority_discovery_service, ctx), fields(subsystem = LOG_TARGET))]
+#[tracing::instrument(skip(bridge, ctx), fields(subsystem = LOG_TARGET))]
 async fn run_network<N, AD>(
-	mut network_service: N,
-	mut authority_discovery_service: AD,
+	mut bridge: NetworkBridge<N, AD>,
 	mut ctx: impl SubsystemContext<Message=NetworkBridgeMessage>,
 ) -> SubsystemResult<()>
 where
 	N: Network + validator_discovery::Network,
 	AD: validator_discovery::AuthorityDiscovery,
 {
-	let mut event_stream = network_service.event_stream().fuse();
+	let mut event_stream = bridge.network_service.event_stream().fuse();
 
 	// Most recent heads are at the back.
 	let mut live_heads: Vec<(Hash, Arc<JaegerSpan>)> = Vec::with_capacity(MAX_VIEW_HEADS);
@@ -169,22 +169,55 @@ where
 		let action = {
 			let subsystem_next = ctx.recv().fuse();
 			let mut net_event_next = event_stream.next().fuse();
+			let mut req_res_event_next = bridge.request_multiplexer.next().fuse();
 			futures::pin_mut!(subsystem_next);
 
 			futures::select! {
 				subsystem_msg = subsystem_next => Action::from(subsystem_msg),
 				net_event = net_event_next => Action::from(net_event),
+				req_res_event = req_res_event_next  => Action::from(req_res_event),
 			}
 		};
 
 		match action {
 			Action::Nop => {}
-			Action::Abort => return Ok(()),
+			Action::Abort(reason) => match reason {
+				AbortReason::SubsystemError(err) => {
+					tracing::warn!(
+						target: LOG_TARGET,
+						err = ?err,
+						"Shutting down Network Bridge due to error"
+					);
+					return Err(SubsystemError::Context(format!(
+						"Received SubsystemError from overseer: {:?}",
+						err
+					)));
+				}
+				AbortReason::EventStreamConcluded => {
+					tracing::info!(
+						target: LOG_TARGET,
+						"Shutting down Network Bridge: underlying request stream concluded"
+					);
+					return Err(SubsystemError::Context(
+						"Incoming network event stream concluded.".to_string(),
+					));
+				}
+				AbortReason::RequestStreamConcluded => {
+					tracing::info!(
+						target: LOG_TARGET,
+						"Shutting down Network Bridge: underlying request stream concluded"
+					);
+					return Err(SubsystemError::Context(
+						"Incoming network request stream concluded".to_string(),
+					));
+				}
+				AbortReason::OverseerConcluded => return Ok(()),
+			}
 
 			Action::SendValidationMessages(msgs) => {
 				for (peers, msg) in msgs {
 					send_message(
-							&mut network_service,
+							&mut bridge.network_service,
 							peers,
 							PeerSet::Validation,
 							WireMessage::ProtocolMessage(msg),
@@ -195,7 +228,7 @@ where
 			Action::SendCollationMessages(msgs) => {
 				for (peers, msg) in msgs {
 					send_message(
-							&mut network_service,
+							&mut bridge.network_service,
 							peers,
 							PeerSet::Collation,
 							WireMessage::ProtocolMessage(msg),
@@ -203,6 +236,12 @@ where
 				}
 			}
 
+			Action::SendRequests(reqs) => {
+				for req in reqs {
+					bridge.network_service.start_request(req);
+				}
+			},
+
 			Action::ConnectToValidators {
 				validator_ids,
 				connected,
@@ -210,21 +249,28 @@ where
 				let (ns, ads) = validator_discovery.on_request(
 					validator_ids,
 					connected,
-					network_service,
-					authority_discovery_service,
+					bridge.network_service,
+					bridge.authority_discovery_service,
 				).await;
-				network_service = ns;
-				authority_discovery_service = ads;
+				bridge.network_service = ns;
+				bridge.authority_discovery_service = ads;
 			},
 
-			Action::ReportPeer(peer, rep) => network_service.report_peer(peer, rep).await?,
+			Action::ReportPeer(peer, rep) => {
+				tracing::debug!(
+					target: LOG_TARGET,
+					peer = ?peer,
+					"Peer sent us an invalid request",
+				);
+				bridge.network_service.report_peer(peer, rep).await?
+			}
 
 			Action::ActiveLeaves(ActiveLeavesUpdate { activated, deactivated }) => {
 				live_heads.extend(activated);
 				live_heads.retain(|h| !deactivated.contains(&h.0));
 
 				update_our_view(
-					&mut network_service,
+					&mut bridge.network_service,
 					&mut ctx,
 					&live_heads,
 					&mut local_view,
@@ -250,7 +296,7 @@ where
 					PeerSet::Collation => &mut collation_peers,
 				};
 
-				validator_discovery.on_peer_connected(&peer, &mut authority_discovery_service).await;
+				validator_discovery.on_peer_connected(&peer, &mut bridge.authority_discovery_service).await;
 
 				match peer_map.entry(peer.clone()) {
 					hash_map::Entry::Occupied(_) => continue,
@@ -311,7 +357,7 @@ where
 						peer.clone(),
 						&mut validation_peers,
 						v_messages,
-						&mut network_service,
+						&mut bridge.network_service,
 					).await?;
 
 					dispatch_validation_events_to_all(events, &mut ctx).await;
@@ -322,12 +368,13 @@ where
 						peer.clone(),
 						&mut collation_peers,
 						c_messages,
-						&mut network_service,
+						&mut bridge.network_service,
 					).await?;
 
 					dispatch_collation_events_to_all(events, &mut ctx).await;
 				}
 			},
+			Action::SendMessage(msg) => ctx.send_message(msg).await,
 		}
 	}
 }
@@ -562,9 +609,10 @@ mod tests {
 	use polkadot_node_subsystem_util::metered;
 	use polkadot_node_network_protocol::view;
 	use sc_network::Multiaddr;
+	use sc_network::config::RequestResponseConfig;
 	use sp_keyring::Sr25519Keyring;
 	use polkadot_primitives::v1::AuthorityDiscoveryId;
-	use polkadot_node_network_protocol::ObservedRole;
+	use polkadot_node_network_protocol::{ObservedRole, request_response::request::Requests};
 
 	use crate::network::{Network, NetworkAction};
 
@@ -572,6 +620,7 @@ mod tests {
 	struct TestNetwork {
 		net_events: Arc<Mutex<Option<SingleItemStream<NetworkEvent>>>>,
 		action_tx: metered::UnboundedMeteredSender<NetworkAction>,
+		_req_configs: Vec<RequestResponseConfig>,
 	}
 
 	struct TestAuthorityDiscovery;
@@ -583,7 +632,7 @@ mod tests {
 		net_tx: SingleItemSink<NetworkEvent>,
 	}
 
-	fn new_test_network() -> (
+	fn new_test_network(req_configs: Vec<RequestResponseConfig>) -> (
 		TestNetwork,
 		TestNetworkHandle,
 		TestAuthorityDiscovery,
@@ -595,6 +644,7 @@ mod tests {
 			TestNetwork {
 				net_events: Arc::new(Mutex::new(Some(net_rx))),
 				action_tx,
+				_req_configs: req_configs,
 			},
 			TestNetworkHandle {
 				action_rx,
@@ -617,6 +667,9 @@ mod tests {
 		{
 			Box::pin((&mut self.action_tx).sink_map_err(Into::into))
 		}
+
+		fn start_request(&self, _: Requests) {
+		}
 	}
 
 	#[async_trait]
@@ -698,12 +751,18 @@ mod tests {
 
 	fn test_harness<T: Future<Output=()>>(test: impl FnOnce(TestHarness) -> T) {
 		let pool = sp_core::testing::TaskExecutor::new();
-		let (network, network_handle, discovery) = new_test_network();
+		let (request_multiplexer, req_configs) = RequestMultiplexer::new();
+		let (network, network_handle, discovery) = new_test_network(req_configs);
 		let (context, virtual_overseer) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
 
+		let bridge = NetworkBridge {
+			network_service: network,
+			authority_discovery_service: discovery,
+			request_multiplexer,
+		};
+
 		let network_bridge = run_network(
-			network,
-			discovery,
+			bridge,
 			context,
 		)
 			.map_err(|_| panic!("subsystem execution failed"))

polkadot/node/network/bridge/src/multiplexer.rs

@@ -0,0 +1,132 @@
+// Copyright 2021 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/>.
+
+use std::pin::Pin;
+
+use futures::channel::mpsc;
+use futures::stream::Stream;
+use futures::task::{Context, Poll};
+use strum::IntoEnumIterator;
+
+use parity_scale_codec::{Decode, Error as DecodingError};
+
+use sc_network::config as network;
+use sc_network::PeerId;
+
+use polkadot_node_network_protocol::request_response::{
+	request::IncomingRequest, v1, Protocol, RequestResponseConfig,
+};
+use polkadot_subsystem::messages::AllMessages;
+
+/// Multiplex incoming network requests.
+///
+/// This multiplexer consumes all request streams and makes them a `Stream` of a single message
+/// type, useful for the network bridge to send them via the `Overseer` to other subsystems.
+pub struct RequestMultiplexer {
+	receivers: Vec<(Protocol, mpsc::Receiver<network::IncomingRequest>)>,
+	next_poll: usize,
+}
+
+/// Multiplexing can fail in case of invalid messages.
+pub struct RequestMultiplexError {
+	/// The peer that sent the invalid message.
+	pub peer: PeerId,
+	/// The error that occurred.
+	pub error: DecodingError,
+}
+
+impl RequestMultiplexer {
+	/// Create a new `RequestMultiplexer`.
+	///
+	/// This function uses `Protocol::get_config` for each available protocol and creates a
+	/// `RequestMultiplexer` from it. The returned `RequestResponseConfig`s must be passed to the
+	/// network implementation.
+	pub fn new() -> (Self, Vec<RequestResponseConfig>) {
+		let (receivers, cfgs): (Vec<_>, Vec<_>) = Protocol::iter()
+			.map(|p| {
+				let (rx, cfg) = p.get_config();
+				((p, rx), cfg)
+			})
+			.unzip();
+
+		(
+			Self {
+				receivers,
+				next_poll: 0,
+			},
+			cfgs,
+		)
+	}
+}
+
+impl Stream for RequestMultiplexer {
+	type Item = Result<AllMessages, RequestMultiplexError>;
+
+	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+		let len = self.receivers.len();
+		let mut count = len;
+		let mut i = self.next_poll;
+		let mut result = Poll::Ready(None);
+		// Poll streams in round robin fashion:
+		while count > 0 {
+			// % safe, because count initialized to len, loop would not be entered if 0, also
+			// length of receivers is fixed.
+			let (p, rx): &mut (_, _) = &mut self.receivers[i % len];
+			i += 1;
+			count -= 1;
+			match Pin::new(rx).poll_next(cx) {
+				// If at least one stream is pending, then we are not done yet (No
+				// Ready(None)).
+				Poll::Pending => result = Poll::Pending,
+				// Receiver is a fused stream, which allows for this simple handling of 
+				// exhausted ones.
+				Poll::Ready(None) => {}
+				Poll::Ready(Some(v)) => {
+					result = Poll::Ready(Some(multiplex_single(*p, v)));
+					break;
+				}
+			}
+		}
+		self.next_poll = i;
+		result
+	}
+}
+
+/// Convert a single raw incoming request into a `MultiplexMessage`.
+fn multiplex_single(
+	p: Protocol,
+	network::IncomingRequest {
+		payload,
+		peer,
+		pending_response,
+	}: network::IncomingRequest,
+) -> Result<AllMessages, RequestMultiplexError> {
+	let r = match p {
+		Protocol::AvailabilityFetching => From::from(IncomingRequest::new(
+			peer,
+			decode_with_peer::<v1::AvailabilityFetchingRequest>(peer, payload)?,
+			pending_response,
+		)),
+	};
+	Ok(r)
+}
+
+fn decode_with_peer<Req: Decode>(
+	peer: PeerId,
+	payload: Vec<u8>,
+) -> Result<Req, RequestMultiplexError> {
+	Req::decode(&mut payload.as_ref()).map_err(|error| RequestMultiplexError { peer, error })
+}

polkadot/node/network/bridge/src/network.rs

@@ -24,12 +24,18 @@ use futures::stream::BoxStream;
 use parity_scale_codec::Encode;
 
 use sc_network::Event as NetworkEvent;
+use sc_network::{NetworkService, IfDisconnected};
 
-use super::LOG_TARGET;
-use polkadot_node_network_protocol::{peer_set::PeerSet, PeerId, ReputationChange};
+use polkadot_node_network_protocol::{
+	peer_set::PeerSet,
+	request_response::{OutgoingRequest, Requests},
+	PeerId, ReputationChange,
+};
 use polkadot_primitives::v1::{Block, Hash};
 use polkadot_subsystem::{SubsystemError, SubsystemResult};
 
+use super::LOG_TARGET;
+
 /// Send a message to the network.
 ///
 /// This function is only used internally by the network-bridge, which is responsible to only send
@@ -86,7 +92,6 @@ pub enum NetworkAction {
 }
 
 /// An abstraction over networking for the purposes of this subsystem.
-///
 pub trait Network: Send + 'static {
 	/// Get a stream of all events occurring on the network. This may include events unrelated
 	/// to the Polkadot protocol - the user of this function should filter only for events related
@@ -99,6 +104,9 @@ pub trait Network: Send + 'static {
 		&'a mut self,
 	) -> Pin<Box<dyn Sink<NetworkAction, Error = SubsystemError> + Send + 'a>>;
 
+	/// Send a request to a remote peer.
+	fn start_request(&self, req: Requests);
+
 	/// Report a given peer as either beneficial (+) or costly (-) according to the given scalar.
 	fn report_peer(
 		&mut self,
@@ -129,9 +137,9 @@ pub trait Network: Send + 'static {
 	}
 }
 
-impl Network for Arc<sc_network::NetworkService<Block, Hash>> {
+impl Network for Arc<NetworkService<Block, Hash>> {
 	fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
-		sc_network::NetworkService::event_stream(self, "polkadot-network-bridge").boxed()
+		NetworkService::event_stream(self, "polkadot-network-bridge").boxed()
 	}
 
 	#[tracing::instrument(level = "trace", skip(self), fields(subsystem = LOG_TARGET))]
@@ -141,7 +149,7 @@ impl Network for Arc<sc_network::NetworkService<Block, Hash>> {
 		use futures::task::{Context, Poll};
 
 		// wrapper around a NetworkService to make it act like a sink.
-		struct ActionSink<'b>(&'b sc_network::NetworkService<Block, Hash>);
+		struct ActionSink<'b>(&'b NetworkService<Block, Hash>);
 
 		impl<'b> Sink<NetworkAction> for ActionSink<'b> {
 			type Error = SubsystemError;
@@ -180,4 +188,23 @@ impl Network for Arc<sc_network::NetworkService<Block, Hash>> {
 
 		Box::pin(ActionSink(&**self))
 	}
+
+	fn start_request(&self, req: Requests) {
+		let (
+			protocol,
+			OutgoingRequest {
+				peer,
+				payload,
+				pending_response,
+			},
+		) = req.encode_request();
+
+		NetworkService::start_request(&*self,
+			peer,
+			protocol.into_protocol_name(),
+			payload,
+			pending_response,
+			IfDisconnected::TryConnect,
+		);
+	}
 }

polkadot/node/network/collator-protocol/src/collator_side.rs

@@ -26,9 +26,9 @@ use polkadot_primitives::v1::{
 use polkadot_subsystem::{
 	jaeger, PerLeafSpan,
 	FromOverseer, OverseerSignal, SubsystemContext,
-	messages::{AllMessages, CollatorProtocolMessage, NetworkBridgeMessage},
+	messages::{AllMessages, CollatorProtocolMessage, NetworkBridgeMessage, NetworkBridgeEvent},
 };
-use polkadot_node_network_protocol::{v1 as protocol_v1, View, PeerId, NetworkBridgeEvent, RequestId, OurView};
+use polkadot_node_network_protocol::{v1 as protocol_v1, View, PeerId, RequestId, OurView};
 use polkadot_node_subsystem_util::{
 	validator_discovery,
 	request_validators_ctx,

polkadot/node/network/collator-protocol/src/validator_side.rs

@@ -32,10 +32,11 @@ use polkadot_subsystem::{
 	FromOverseer, OverseerSignal, SubsystemContext,
 	messages::{
 		AllMessages, CandidateSelectionMessage, CollatorProtocolMessage, NetworkBridgeMessage,
+		NetworkBridgeEvent,
 	},
 };
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, View, OurView, PeerId, ReputationChange as Rep, RequestId, NetworkBridgeEvent,
+	v1 as protocol_v1, View, OurView, PeerId, ReputationChange as Rep, RequestId,
 };
 use polkadot_node_subsystem_util::{TimeoutExt as _, metrics::{self, prometheus}};
 

polkadot/node/network/pov-distribution/src/lib.rs

@@ -29,7 +29,7 @@ use polkadot_subsystem::{
 	ActiveLeavesUpdate, OverseerSignal, SubsystemContext, SubsystemResult, SubsystemError, Subsystem,
 	FromOverseer, SpawnedSubsystem,
 	messages::{
-		PoVDistributionMessage, AllMessages, NetworkBridgeMessage,
+		PoVDistributionMessage, AllMessages, NetworkBridgeMessage, NetworkBridgeEvent,
 	},
 };
 use polkadot_node_subsystem_util::{
@@ -40,7 +40,7 @@ use polkadot_node_subsystem_util::{
 	metrics::{self, prometheus},
 };
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, ReputationChange as Rep, NetworkBridgeEvent, PeerId, OurView,
+	v1 as protocol_v1, ReputationChange as Rep, PeerId, OurView,
 };
 
 use futures::prelude::*;

polkadot/node/network/protocol/Cargo.toml

@@ -13,6 +13,7 @@ parity-scale-codec = { version = "2.0.0", default-features = false, features = [
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 strum = { version = "0.20", features = ["derive"] }
 thiserror = "1.0.23"
+futures = "0.3.12"
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
 zstd = "0.5.0"

polkadot/node/network/protocol/src/lib.rs

@@ -21,7 +21,7 @@
 
 use polkadot_primitives::v1::{Hash, BlockNumber};
 use parity_scale_codec::{Encode, Decode};
-use std::{convert::TryFrom, fmt, collections::HashMap};
+use std::{fmt, collections::HashMap};
 
 pub use sc_network::{ReputationChange, PeerId};
 #[doc(hidden)]
@@ -33,6 +33,9 @@ pub use std::sync::Arc;
 /// Peer-sets and protocols used for parachains.
 pub mod peer_set;
 
+/// Request/response protocols used in Polkadot.
+pub mod request_response;
+
 /// A unique identifier of a request.
 pub type RequestId = u64;
 
@@ -85,25 +88,6 @@ impl Into<sc_network::ObservedRole> for ObservedRole {
 	}
 }
 
-/// Events from network.
-#[derive(Debug, Clone, PartialEq)]
-pub enum NetworkBridgeEvent<M> {
-	/// A peer has connected.
-	PeerConnected(PeerId, ObservedRole),
-
-	/// A peer has disconnected.
-	PeerDisconnected(PeerId),
-
-	/// Peer has sent a message.
-	PeerMessage(PeerId, M),
-
-	/// Peer's `View` has changed.
-	PeerViewChange(PeerId, View),
-
-	/// Our view has changed.
-	OurViewChange(OurView),
-}
-
 macro_rules! impl_try_from {
 	($m_ty:ident, $variant:ident, $out:ty) => {
 		impl TryFrom<$m_ty> for $out {
@@ -132,29 +116,6 @@ macro_rules! impl_try_from {
 	}
 }
 
-impl<M> NetworkBridgeEvent<M> {
-	/// Focus an overarching network-bridge event into some more specific variant.
-	///
-	/// This acts as a call to `clone`, except in the case where the event is a message event,
-	/// in which case the clone can be expensive and it only clones if the message type can
-	/// be focused.
-	pub fn focus<'a, T>(&'a self) -> Result<NetworkBridgeEvent<T>, WrongVariant>
-		where T: 'a + Clone, &'a T: TryFrom<&'a M, Error = WrongVariant>
-	{
-		Ok(match *self {
-			NetworkBridgeEvent::PeerConnected(ref peer, ref role)
-				=> NetworkBridgeEvent::PeerConnected(peer.clone(), role.clone()),
-			NetworkBridgeEvent::PeerDisconnected(ref peer)
-				=> NetworkBridgeEvent::PeerDisconnected(peer.clone()),
-			NetworkBridgeEvent::PeerMessage(ref peer, ref msg)
-				=> NetworkBridgeEvent::PeerMessage(peer.clone(), <&'a T>::try_from(msg)?.clone()),
-			NetworkBridgeEvent::PeerViewChange(ref peer, ref view)
-				=> NetworkBridgeEvent::PeerViewChange(peer.clone(), view.clone()),
-			NetworkBridgeEvent::OurViewChange(ref view)
-				=> NetworkBridgeEvent::OurViewChange(view.clone()),
-		})
-	}
-}
 
 /// Specialized wrapper around [`View`].
 ///

polkadot/node/network/protocol/src/peer_set.rs

@@ -23,7 +23,8 @@ use strum::{EnumIter, IntoEnumIterator};
 /// The peer-sets and thus the protocols which are used for the network.
 #[derive(Debug, Clone, Copy, PartialEq, EnumIter)]
 pub enum PeerSet {
-	/// The validation peer-set is responsible for all messages related to candidate validation and communication among validators.
+	/// The validation peer-set is responsible for all messages related to candidate validation and
+	/// communication among validators.
 	Validation,
 	/// The collation peer-set is used for validator<>collator communication.
 	Collation,

polkadot/node/network/protocol/src/request_response.rs

@@ -0,0 +1,119 @@
+// Copyright 2021 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/>.
+
+//! Overview over request/responses as used in `Polkadot`.
+//!
+//! enum  Protocol .... List of all supported protocols.
+//!
+//! enum  Requests  .... List of all supported requests, each entry matches one in protocols, but
+//! has the actual request as payload.
+//!
+//! struct IncomingRequest .... wrapper for incoming requests, containing a sender for sending
+//! responses.
+//!
+//! struct OutgoingRequest .... wrapper for outgoing requests, containing a sender used by the
+//! networking code for delivering responses/delivery errors.
+//!
+//! trait `IsRequest` .... A trait describing a particular request. It is used for gathering meta
+//! data, like what is the corresponding response type.
+//!
+//!  Versioned (v1 module): The actual requests and responses as sent over the network.
+
+use std::borrow::Cow;
+use std::time::Duration;
+
+use futures::channel::mpsc;
+use strum::EnumIter;
+
+pub use sc_network::config as network;
+pub use sc_network::config::RequestResponseConfig;
+
+/// All requests that can be sent to the network bridge.
+pub mod request;
+pub use request::{IncomingRequest, OutgoingRequest, Requests};
+
+///// Multiplexer for incoming requests.
+// pub mod multiplexer;
+
+/// Actual versioned requests and responses, that are sent over the wire.
+pub mod v1;
+
+/// A protocol per subsystem seems to make the most sense, this way we don't need any dispatching
+/// within protocols.
+#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq, EnumIter)]
+pub enum Protocol {
+	/// Protocol for availability fetching, used by availability distribution.
+	AvailabilityFetching,
+}
+
+/// Default request timeout in seconds.
+const DEFAULT_REQUEST_TIMEOUT: u64 = 8; 
+
+impl Protocol {
+	/// Get a configuration for a given Request response protocol.
+	///
+	/// Returns a receiver for messages received on this protocol and the requested
+	/// `ProtocolConfig`.
+	///
+	/// See also `dispatcher::RequestDispatcher`,  which makes use of this function and provides a more
+	/// high-level interface.
+	pub fn get_config(
+		self,
+	) -> (
+		mpsc::Receiver<network::IncomingRequest>,
+		RequestResponseConfig,
+	) {
+		let p_name = self.into_protocol_name();
+		let (tx, rx) = mpsc::channel(self.get_channel_size());
+		let cfg = match self {
+			Protocol::AvailabilityFetching => RequestResponseConfig {
+				name: p_name,
+				// Arbitrary very conservative numbers:
+				// TODO: Get better numbers, see https://github.com/paritytech/polkadot/issues/2370
+				max_request_size: 10_000,
+				max_response_size: 1_000_000,
+				// Also just some relative conservative guess:
+				request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT),
+				inbound_queue: Some(tx),
+			},
+		};
+		(rx, cfg)
+	}
+
+	// Channel sizes for the supported protocols.
+	fn get_channel_size(self) -> usize {
+		match self {
+			// Hundreds of validators will start requesting their chunks once they see a candidate
+			// awaiting availability on chain. Given that they will see that block at different
+			// times (due to network delays), 100 seems big enough to accomodate for "bursts",
+			// assuming we can service requests relatively quickly, which would need to be measured
+			// as well.
+			Protocol::AvailabilityFetching => 100,
+		}
+	}
+
+	/// Get the protocol name of this protocol, as understood by substrate networking.
+	pub fn into_protocol_name(self) -> Cow<'static, str> {
+		self.get_protocol_name_static().into()
+	}
+
+	/// Get the protocol name associated with each peer set as static str.
+	pub const fn get_protocol_name_static(self) -> &'static str {
+		match self {
+			Protocol::AvailabilityFetching => "/polkadot/req_availability/1",
+		}
+	}
+}

polkadot/node/network/protocol/src/request_response/request.rs

@@ -0,0 +1,209 @@
+// Copyright 2021 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/>.
+
+use futures::channel::oneshot;
+use futures::prelude::Future;
+
+use parity_scale_codec::{Decode, Encode, Error as DecodingError};
+use sc_network as network;
+use sc_network::config as netconfig;
+use sc_network::PeerId;
+
+use super::{v1, Protocol};
+
+/// Common properties of any `Request`.
+pub trait IsRequest {
+	/// Each request has a corresponding `Response`.
+	type Response;
+
+	/// What protocol this `Request` implements.
+	const PROTOCOL: Protocol;
+}
+
+/// All requests that can be sent to the network bridge via `NetworkBridgeMessage::SendRequest`.
+#[derive(Debug)]
+pub enum Requests {
+	/// Request an availability chunk from a node.
+	AvailabilityFetching(OutgoingRequest<v1::AvailabilityFetchingRequest>),
+}
+
+impl Requests {
+	/// Get the protocol this request conforms to.
+	pub fn get_protocol(&self) -> Protocol {
+		match self {
+			Self::AvailabilityFetching(_) => Protocol::AvailabilityFetching,
+		}
+	}
+
+	/// Encode the request.
+	///
+	/// The corresponding protocol is returned as well, as we are now leaving typed territory.
+	///
+	/// Note: `Requests` is just an enum collecting all supported requests supported by network
+	/// bridge, it is never sent over the wire. This function just encodes the individual requests
+	/// contained in the enum.
+	pub fn encode_request(self) -> (Protocol, OutgoingRequest<Vec<u8>>) {
+		match self {
+			Self::AvailabilityFetching(r) => r.encode_request(),
+		}
+	}
+}
+
+/// A request to be sent to the network bridge, including a sender for sending responses/failures.
+///
+/// The network implementation will make use of that sender for informing the requesting subsystem
+/// about responses/errors.
+#[derive(Debug)]
+pub struct OutgoingRequest<Req> {
+	/// Intendent recipient of this request.
+	pub peer: PeerId,
+	/// The actual request to send over the wire.
+	pub payload: Req,
+	/// Sender which is used by networking to get us back a response.
+	pub pending_response: oneshot::Sender<Result<Vec<u8>, network::RequestFailure>>,
+}
+
+/// Any error that can occur when sending a request.
+pub enum RequestError {
+	/// Response could not be decoded.
+	InvalidResponse(DecodingError),
+
+	/// Some error in substrate/libp2p happened.
+	NetworkError(network::RequestFailure),
+
+	/// Response got canceled by networking.
+	Canceled(oneshot::Canceled),
+}
+
+impl<Req> OutgoingRequest<Req>
+where
+	Req: IsRequest + Encode,
+	Req::Response: Decode,
+{
+	/// Create a new `OutgoingRequest`.
+	///
+	/// It will contain a sender that is used by the networking for sending back responses. The
+	/// connected receiver is returned as the second element in the returned tuple.
+	pub fn new(
+		peer: PeerId,
+		payload: Req,
+	) -> (
+		Self,
+		impl Future<Output = Result<Req::Response, RequestError>>,
+	) {
+		let (tx, rx) = oneshot::channel();
+		let r = Self {
+			peer,
+			payload,
+			pending_response: tx,
+		};
+		(r, receive_response::<Req>(rx))
+	}
+
+	/// Encode a request into a `Vec<u8>`.
+	///
+	/// As this throws away type information, we also return the `Protocol` this encoded request
+	/// adheres to.
+	pub fn encode_request(self) -> (Protocol, OutgoingRequest<Vec<u8>>) {
+		let OutgoingRequest {
+			peer,
+			payload,
+			pending_response,
+		} = self;
+		let encoded = OutgoingRequest {
+			peer,
+			payload: payload.encode(),
+			pending_response,
+		};
+		(Req::PROTOCOL, encoded)
+	}
+}
+
+impl From<DecodingError> for RequestError {
+	fn from(err: DecodingError) -> Self {
+		Self::InvalidResponse(err)
+	}
+}
+
+impl From<network::RequestFailure> for RequestError {
+	fn from(err: network::RequestFailure) -> Self {
+		Self::NetworkError(err)
+	}
+}
+
+impl From<oneshot::Canceled> for RequestError {
+	fn from(err: oneshot::Canceled) -> Self {
+		Self::Canceled(err)
+	}
+}
+
+/// A request coming in, including a sender for sending responses.
+///
+/// `IncomingRequest`s are produced by `RequestMultiplexer` on behalf of the network bridge.
+#[derive(Debug)]
+pub struct IncomingRequest<Req> {
+	/// PeerId of sending peer.
+	pub peer: PeerId,
+	/// The sent request.
+	pub payload: Req,
+	pending_response: oneshot::Sender<netconfig::OutgoingResponse>,
+}
+
+impl<Req> IncomingRequest<Req>
+where
+	Req: IsRequest,
+	Req::Response: Encode,
+{
+	/// Create new `IncomingRequest`.
+	pub fn new(
+		peer: PeerId,
+		payload: Req,
+		pending_response: oneshot::Sender<netconfig::OutgoingResponse>,
+	) -> Self {
+		Self {
+			peer,
+			payload,
+			pending_response,
+		}
+	}
+
+	/// Send the response back.
+	///
+	/// On success we return Ok(()), on error we return the not sent `Response`.
+	///
+	/// netconfig::OutgoingResponse exposes a way of modifying the peer's reputation. If needed we
+	/// can change this function to expose this feature as well.
+	pub fn send_response(self, resp: Req::Response) -> Result<(), Req::Response> {
+		self.pending_response
+			.send(netconfig::OutgoingResponse {
+				result: Ok(resp.encode()),
+				reputation_changes: Vec::new(),
+			})
+			.map_err(|_| resp)
+	}
+}
+
+/// Future for actually receiving a typed response for an OutgoingRequest.
+async fn receive_response<Req>(
+	rec: oneshot::Receiver<Result<Vec<u8>, network::RequestFailure>>,
+) -> Result<Req::Response, RequestError>
+where
+	Req: IsRequest,
+	Req::Response: Decode,
+{
+	let raw = rec.await??;
+	Ok(Decode::decode(&mut raw.as_ref())?)
+}

polkadot/node/network/protocol/src/request_response/v1.rs

@@ -0,0 +1,44 @@
+// Copyright 2021 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/>.
+
+//! Requests and responses as sent over the wire for the individual protocols.
+
+use parity_scale_codec::{Decode, Encode};
+
+use polkadot_primitives::v1::{CandidateHash, ErasureChunk, ValidatorIndex};
+
+use super::request::IsRequest;
+use super::Protocol;
+
+/// Request an availability chunk.
+#[derive(Debug, Clone, Encode, Decode)]
+pub struct AvailabilityFetchingRequest {
+	candidate_hash: CandidateHash,
+	index: ValidatorIndex,
+}
+
+/// Receive a rqeuested erasure chunk.
+#[derive(Debug, Clone, Encode, Decode)]
+pub enum AvailabilityFetchingResponse {
+	/// The requested chunk.
+	#[codec(index = 0)]
+	Chunk(ErasureChunk),
+}
+
+impl IsRequest for AvailabilityFetchingRequest {
+	type Response = AvailabilityFetchingResponse;
+	const PROTOCOL: Protocol = Protocol::AvailabilityFetching;
+}

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

@@ -27,7 +27,7 @@ use polkadot_subsystem::{
 	ActiveLeavesUpdate, FromOverseer, OverseerSignal, PerLeafSpan,
 	messages::{
 		AllMessages, NetworkBridgeMessage, StatementDistributionMessage, CandidateBackingMessage,
-		RuntimeApiMessage, RuntimeApiRequest,
+		RuntimeApiMessage, RuntimeApiRequest, NetworkBridgeEvent,
 	},
 };
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
@@ -36,7 +36,7 @@ use polkadot_primitives::v1::{
 	Hash, CompactStatement, ValidatorIndex, ValidatorId, SigningContext, ValidatorSignature, CandidateHash,
 };
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, View, PeerId, ReputationChange as Rep, NetworkBridgeEvent, OurView,
+	v1 as protocol_v1, View, PeerId, ReputationChange as Rep, OurView,
 };
 
 use futures::prelude::*;

polkadot/node/overseer/src/lib.rs

@@ -1961,9 +1961,9 @@ mod tests {
 	use futures::{executor, pin_mut, select, FutureExt, pending};
 
 	use polkadot_primitives::v1::{BlockData, CollatorPair, PoV, CandidateHash};
-	use polkadot_subsystem::{messages::RuntimeApiRequest, JaegerSpan};
+	use polkadot_subsystem::{messages::RuntimeApiRequest, messages::NetworkBridgeEvent, JaegerSpan};
 	use polkadot_node_primitives::{Collation, CollationGenerationConfig};
-	use polkadot_node_network_protocol::{PeerId, ReputationChange, NetworkBridgeEvent};
+	use polkadot_node_network_protocol::{PeerId, ReputationChange};
 	use polkadot_node_subsystem_util::metered;
 
 	use sp_core::crypto::Pair as _;

polkadot/node/service/src/lib.rs

@@ -39,6 +39,8 @@ use {
 	sp_trie::PrefixedMemoryDB,
 	sc_client_api::ExecutorProvider,
 };
+#[cfg(feature = "real-overseer")]
+use polkadot_network_bridge::RequestMultiplexer;
 
 use sp_core::traits::SpawnNamed;
 
@@ -347,6 +349,7 @@ fn real_overseer<Spawner, RuntimeClient>(
 	_: AvailabilityConfig,
 	_: Arc<sc_network::NetworkService<Block, Hash>>,
 	_: AuthorityDiscoveryService,
+	request_multiplexer: (),
 	registry: Option<&Registry>,
 	spawner: Spawner,
 	_: IsCollator,
@@ -373,6 +376,7 @@ fn real_overseer<Spawner, RuntimeClient>(
 	availability_config: AvailabilityConfig,
 	network_service: Arc<sc_network::NetworkService<Block, Hash>>,
 	authority_discovery: AuthorityDiscoveryService,
+	request_multiplexer: RequestMultiplexer,
 	registry: Option<&Registry>,
 	spawner: Spawner,
 	is_collator: IsCollator,
@@ -456,6 +460,7 @@ where
 		network_bridge: NetworkBridgeSubsystem::new(
 			network_service,
 			authority_discovery,
+			request_multiplexer,
 		),
 		pov_distribution: PoVDistributionSubsystem::new(
 			Metrics::register(registry)?,
@@ -597,6 +602,15 @@ pub fn new_full<RuntimeApi, Executor>(
 	config.network.request_response_protocols.push(sc_finality_grandpa_warp_sync::request_response_config_for_chain(
 		&config, task_manager.spawn_handle(), backend.clone(),
 	));
+	#[cfg(feature = "real-overseer")]
+	fn register_request_response(config: &mut sc_network::config::NetworkConfiguration) -> RequestMultiplexer {
+		let (multiplexer, configs) = RequestMultiplexer::new();
+		config.request_response_protocols.extend(configs);
+		multiplexer
+	}
+	#[cfg(not(feature = "real-overseer"))]
+	fn register_request_response(_: &mut sc_network::config::NetworkConfiguration) {}
+	let request_multiplexer = register_request_response(&mut config.network);
 
 	let (network, network_status_sinks, system_rpc_tx, network_starter) =
 		service::build_network(service::BuildNetworkParams {
@@ -693,6 +707,7 @@ pub fn new_full<RuntimeApi, Executor>(
 			availability_config,
 			network.clone(),
 			authority_discovery_service,
+			request_multiplexer,
 			prometheus_registry.as_ref(),
 			spawner,
 			is_collator,

polkadot/node/subsystem/src/messages.rs

@@ -25,7 +25,8 @@
 use futures::channel::{mpsc, oneshot};
 use thiserror::Error;
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, NetworkBridgeEvent, ReputationChange, PeerId,
+	v1 as protocol_v1, ReputationChange, PeerId,
+	request_response::{Requests, request::IncomingRequest, v1 as req_res_v1},
 };
 use polkadot_node_primitives::{
 	CollationGenerationConfig, SignedFullStatement, ValidationResult,
@@ -44,6 +45,11 @@ use polkadot_primitives::v1::{
 use polkadot_statement_table::v1::Misbehavior;
 use std::{sync::Arc, collections::btree_map::BTreeMap};
 
+
+/// Network events as transmitted to other subsystems, wrapped in their message types.
+pub mod network_bridge_event;
+pub use network_bridge_event::NetworkBridgeEvent;
+
 /// Subsystem messages where each message is always bound to a relay parent.
 pub trait BoundToRelayParent {
 	/// Returns the relay parent this message is bound to.
@@ -212,6 +218,9 @@ pub enum NetworkBridgeMessage {
 	/// Send a batch of collation messages.
 	SendCollationMessages(Vec<(Vec<PeerId>, protocol_v1::CollationProtocol)>),
 
+	/// Send requests via substrate request/response.
+	SendRequests(Vec<Requests>),
+
 	/// Connect to peers who represent the given `validator_ids`.
 	///
 	/// Also ask the network to stay connected to these peers at least
@@ -237,6 +246,7 @@ impl NetworkBridgeMessage {
 			Self::SendValidationMessages(_) => None,
 			Self::SendCollationMessages(_) => None,
 			Self::ConnectToValidators { .. } => None,
+			Self::SendRequests { .. } => None,
 		}
 	}
 }
@@ -246,6 +256,8 @@ impl NetworkBridgeMessage {
 pub enum AvailabilityDistributionMessage {
 	/// Event from the network bridge.
 	NetworkBridgeUpdateV1(NetworkBridgeEvent<protocol_v1::AvailabilityDistributionMessage>),
+	/// Incoming request for an availability chunk.
+	AvailabilityFetchingRequest(IncomingRequest<req_res_v1::AvailabilityFetchingRequest>)
 }
 
 /// Availability Recovery Message.
@@ -266,6 +278,7 @@ impl AvailabilityDistributionMessage {
 	pub fn relay_parent(&self) -> Option<Hash> {
 		match self {
 			Self::NetworkBridgeUpdateV1(_) => None,
+			Self::AvailabilityFetchingRequest(_) => None,
 		}
 	}
 }
@@ -693,3 +706,9 @@ pub enum AllMessages {
 	/// Message for the Approval Distribution subsystem.
 	ApprovalDistribution(ApprovalDistributionMessage),
 }
+
+impl From<IncomingRequest<req_res_v1::AvailabilityFetchingRequest>> for AllMessages {
+	fn from(req: IncomingRequest<req_res_v1::AvailabilityFetchingRequest>) -> Self {
+		From::<AvailabilityDistributionMessage>::from(From::from(req))
+	}
+}

polkadot/node/subsystem/src/messages/network_bridge_event.rs

@@ -0,0 +1,73 @@
+// Copyright 2017-2021 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/>.
+
+use std::convert::TryFrom;
+
+pub use sc_network::{ReputationChange, PeerId};
+use polkadot_node_network_protocol::{WrongVariant, ObservedRole, OurView, View};
+
+/// Events from network.
+#[derive(Debug, Clone, PartialEq)]
+pub enum NetworkBridgeEvent<M> {
+	/// A peer has connected.
+	PeerConnected(PeerId, ObservedRole),
+
+	/// A peer has disconnected.
+	PeerDisconnected(PeerId),
+
+	/// Peer has sent a message.
+	PeerMessage(PeerId, M),
+
+	/// Peer's `View` has changed.
+	PeerViewChange(PeerId, View),
+
+	/// Our view has changed.
+	OurViewChange(OurView),
+}
+
+impl<M> NetworkBridgeEvent<M> {
+	/// Focus an overarching network-bridge event into some more specific variant.
+	///
+	/// This tries to transform M in `PeerMessage` to a message type specific to a subsystem.
+	/// It is used to dispatch events coming from a peer set to the various subsystems that are
+	/// handled within that peer set. More concretly a `ValidationProtocol` will be transformed
+	/// for example into a `BitfieldDistributionMessage` in case of the `BitfieldDistribution`
+	/// constructor.
+	///
+	/// Therefore a NetworkBridgeEvent<ValidationProtocol> will become for example a
+	/// NetworkBridgeEvent<BitfieldDistributionMessage>, with the more specific message type
+	/// `BitfieldDistributionMessage`.
+	///
+	/// This acts as a call to `clone`, except in the case where the event is a message event,
+	/// in which case the clone can be expensive and it only clones if the message type can
+	/// be focused.
+	pub fn focus<'a, T>(&'a self) -> Result<NetworkBridgeEvent<T>, WrongVariant>
+		where T: 'a + Clone, &'a T: TryFrom<&'a M, Error = WrongVariant>
+	{
+		Ok(match *self {
+			NetworkBridgeEvent::PeerConnected(ref peer, ref role)
+				=> NetworkBridgeEvent::PeerConnected(peer.clone(), role.clone()),
+			NetworkBridgeEvent::PeerDisconnected(ref peer)
+				=> NetworkBridgeEvent::PeerDisconnected(peer.clone()),
+			NetworkBridgeEvent::PeerMessage(ref peer, ref msg)
+				=> NetworkBridgeEvent::PeerMessage(peer.clone(), <&'a T>::try_from(msg)?.clone()),
+			NetworkBridgeEvent::PeerViewChange(ref peer, ref view)
+				=> NetworkBridgeEvent::PeerViewChange(peer.clone(), view.clone()),
+			NetworkBridgeEvent::OurViewChange(ref view)
+				=> NetworkBridgeEvent::OurViewChange(view.clone()),
+		})
+	}
+}