Prepare for network protocol version upgrades (#5084)

* explicitly tag network requests with version

* fmt

* make PeerSet more aware of versioning

* some generalization of the network bridge to support upgrades

* walk back some renaming

* walk back some version stuff

* extract version from fallback

* remove V1 from NetworkBridgeUpdate

* add accidentally-removed timer

* implement focusing for versioned messages

* fmt

* fix up network bridge & tests

* remove inaccurate version check in bridge

* remove some TODO [now]s

* fix fallout in statement distribution

* fmt

* fallout in gossip-support

* fix fallout in collator-protocol

* fix fallout in bitfield-distribution

* fix fallout in approval-distribution

* fmt

* use never!

* fmt

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

@@ -22,7 +22,8 @@
 
 use futures::{channel::oneshot, FutureExt as _};
 use polkadot_node_network_protocol::{
-	v1 as protocol_v1, PeerId, UnifiedReputationChange as Rep, View,
+	self as net_protocol, v1 as protocol_v1, PeerId, UnifiedReputationChange as Rep, Versioned,
+	View,
 };
 use polkadot_node_primitives::approval::{
 	AssignmentCert, BlockApprovalMeta, IndirectAssignmentCert, IndirectSignedApprovalVote,
@@ -457,11 +458,11 @@ impl State {
 		ctx: &mut (impl SubsystemContext<Message = ApprovalDistributionMessage>
 		          + overseer::SubsystemContext<Message = ApprovalDistributionMessage>),
 		metrics: &Metrics,
-		event: NetworkBridgeEvent<protocol_v1::ApprovalDistributionMessage>,
+		event: NetworkBridgeEvent<net_protocol::ApprovalDistributionMessage>,
 		rng: &mut (impl CryptoRng + Rng),
 	) {
 		match event {
-			NetworkBridgeEvent::PeerConnected(peer_id, role, _) => {
+			NetworkBridgeEvent::PeerConnected(peer_id, role, _, _) => {
 				// insert a blank view if none already present
 				gum::trace!(target: LOG_TARGET, ?peer_id, ?role, "Peer connected");
 				self.peer_views.entry(peer_id).or_default();
@@ -501,7 +502,7 @@ impl State {
 					live
 				});
 			},
-			NetworkBridgeEvent::PeerMessage(peer_id, msg) => {
+			NetworkBridgeEvent::PeerMessage(peer_id, Versioned::V1(msg)) => {
 				self.process_incoming_peer_message(ctx, metrics, peer_id, msg, rng).await;
 			},
 		}
@@ -1068,9 +1069,9 @@ impl State {
 
 			ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
 				peers,
-				protocol_v1::ValidationProtocol::ApprovalDistribution(
+				Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 					protocol_v1::ApprovalDistributionMessage::Assignments(assignments),
-				),
+				)),
 			))
 			.await;
 		}
@@ -1330,9 +1331,9 @@ impl State {
 
 			ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
 				peers,
-				protocol_v1::ValidationProtocol::ApprovalDistribution(
+				Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 					protocol_v1::ApprovalDistributionMessage::Approvals(approvals),
-				),
+				)),
 			))
 			.await;
 		}
@@ -1458,9 +1459,9 @@ impl State {
 
 			ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
 				vec![peer_id.clone()],
-				protocol_v1::ValidationProtocol::ApprovalDistribution(
+				Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 					protocol_v1::ApprovalDistributionMessage::Assignments(assignments_to_send),
-				),
+				)),
 			))
 			.await;
 		}
@@ -1474,10 +1475,10 @@ impl State {
 			);
 
 			ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
-				vec![peer_id.clone()],
-				protocol_v1::ValidationProtocol::ApprovalDistribution(
+				vec![peer_id],
+				Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 					protocol_v1::ApprovalDistributionMessage::Approvals(approvals_to_send),
-				),
+				)),
 			))
 			.await;
 		}
@@ -1676,19 +1677,19 @@ async fn adjust_required_routing_and_propagate(
 	for (peer, assignments_packet) in peer_assignments {
 		ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
 			vec![peer],
-			protocol_v1::ValidationProtocol::ApprovalDistribution(
+			Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 				protocol_v1::ApprovalDistributionMessage::Assignments(assignments_packet),
-			),
+			)),
 		))
 		.await;
 	}
 
 	for (peer, approvals_packet) in peer_approvals {
 		ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
 			vec![peer],
-			protocol_v1::ValidationProtocol::ApprovalDistribution(
+			Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
 				protocol_v1::ApprovalDistributionMessage::Approvals(approvals_packet),
-			),
+			)),
 		))
 		.await;
 	}
@@ -1779,7 +1780,7 @@ impl ApprovalDistribution {
 		Context: overseer::SubsystemContext<Message = ApprovalDistributionMessage>,
 	{
 		match msg {
-			ApprovalDistributionMessage::NetworkBridgeUpdateV1(event) => {
+			ApprovalDistributionMessage::NetworkBridgeUpdate(event) => {
 				state.handle_network_msg(ctx, metrics, event, rng).await;
 			},
 			ApprovalDistributionMessage::NewBlocks(metas) => {