Generic request/response infrastructure for 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";

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: _,

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,

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>;
 

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 =

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"

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,21 +80,42 @@ 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)
 			}
 			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),
+		}
+	}
+}