Extract legacy statement distribution to its own module (#6026)

* add compatibility type to v2 statement distribution message

* warning cleanup

* handle compatibility layer for v2

* clean up an unimplemented!() block

* circulate statements based on version

* extract legacy v1 code into separate module

* remove unimplemented

* clean up naming of from_requester/responder

* remove TODOs

* have backing share seconded statements with PVD

* fmt

* fix warning

* Quick fix unused warning for not yet implemented/used staging messages.

* Fix network bridge test

* Fix wrong merge.

We now have 23 subsystems (network bridge split + prospective
parachains)

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

node/core/backing/src/lib.rs

@@ -1606,10 +1606,7 @@ async fn sign_import_and_distribute_statement<Context>(
 	if let Some(signed_statement) = sign_statement(&*rp_state, statement, keystore, metrics).await {
 		import_statement(ctx, rp_state, per_candidate, &signed_statement).await?;
 
-		let smsg = StatementDistributionMessage::Share(
-			rp_state.parent,
-			StatementWithPVD::drop_pvd_from_signed(signed_statement.clone()),
-		);
+		let smsg = StatementDistributionMessage::Share(rp_state.parent, signed_statement.clone());
 		ctx.send_unbounded_message(smsg);
 
 		Ok(Some(signed_statement))

node/core/backing/src/tests/mod.rs

@@ -928,7 +928,7 @@ fn backing_misbehavior_works() {
 					signed_statement,
 				)
 			) if relay_parent == test_state.relay_parent => {
-				assert_eq!(*signed_statement.payload(), Statement::Valid(candidate_a_hash));
+				assert_eq!(*signed_statement.payload(), StatementWithPVD::Valid(candidate_a_hash));
 			}
 		);
 
@@ -1092,14 +1092,14 @@ fn backing_dont_second_invalid() {
 			virtual_overseer.recv().await,
 			AllMessages::CandidateValidation(
 				CandidateValidationMessage::ValidateFromExhaustive(
-					_pvd,
+					pvd,
 					_validation_code,
 					candidate_receipt,
 					_pov,
 					timeout,
 					tx,
 				),
-			) if _pvd == pvd_b &&
+			) if pvd == pvd_b &&
 				_validation_code == validation_code_b &&
 				*_pov == pov_block_b && &candidate_receipt.descriptor == candidate_b.descriptor() &&
 				timeout == BACKING_EXECUTION_TIMEOUT &&
@@ -1135,7 +1135,7 @@ fn backing_dont_second_invalid() {
 					signed_statement,
 				)
 			) if parent_hash == test_state.relay_parent => {
-				assert_eq!(*signed_statement.payload(), Statement::Seconded(candidate_b));
+				assert_eq!(*signed_statement.payload(), StatementWithPVD::Seconded(candidate_b, pvd_b.clone()));
 			}
 		);
 

node/core/backing/src/tests/prospective_parachains.rs

@@ -1276,7 +1276,7 @@ fn concurrent_dependent_candidates() {
 					let payload = statement.payload();
 					assert_matches!(
 						payload.clone(),
-						Statement::Valid(hash)
+						StatementWithPVD::Valid(hash)
 							if hash == candidate_a_hash || hash == candidate_b_hash =>
 						{
 							assert!(valid_statements.insert(hash));

node/network/bridge/src/rx/tests.rs

@@ -1414,14 +1414,16 @@ fn network_protocol_versioning_subsystem_msg() {
 			}
 		);
 
-		let metadata = protocol_vstaging::StatementMetadata {
+		let metadata = protocol_v1::StatementMetadata {
 			relay_parent: Hash::zero(),
 			candidate_hash: CandidateHash::default(),
 			signed_by: ValidatorIndex(0),
 			signature: sp_core::crypto::UncheckedFrom::unchecked_from([1u8; 64]),
 		};
 		let statement_distribution_message =
-			protocol_vstaging::StatementDistributionMessage::LargeStatement(metadata);
+			protocol_vstaging::StatementDistributionMessage::V1Compatibility(
+				protocol_v1::StatementDistributionMessage::LargeStatement(metadata),
+			);
 		let msg = protocol_vstaging::ValidationProtocol::StatementDistribution(
 			statement_distribution_message.clone(),
 		);

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

@@ -946,7 +946,7 @@ async fn handle_network_msg<Context>(
 		PeerMessage(remote, Versioned::V1(msg)) => {
 			handle_incoming_peer_message(ctx, runtime, state, remote, msg).await?;
 		},
-		PeerMessage(_, Versioned::VStaging(msg)) => {},
+		PeerMessage(_, Versioned::VStaging(_msg)) => {},
 		NewGossipTopology { .. } => {
 			// impossible!
 		},

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

@@ -1084,7 +1084,7 @@ async fn handle_network_msg<Context>(
 		PeerMessage(remote, Versioned::V1(msg)) => {
 			process_incoming_peer_message(ctx, state, remote, msg).await;
 		},
-		PeerMessage(_, Versioned::VStaging(msg)) => {},
+		PeerMessage(_, Versioned::VStaging(_msg)) => {},
 	}
 
 	Ok(())

node/network/protocol/src/lib.rs

@@ -592,8 +592,8 @@ pub mod vstaging {
 	use parity_scale_codec::{Decode, Encode};
 
 	use polkadot_primitives::vstaging::{
-		CandidateHash, CandidateIndex, CollatorId, CollatorSignature, CompactStatement, Hash,
-		Id as ParaId, UncheckedSignedAvailabilityBitfield, ValidatorIndex, ValidatorSignature,
+		CandidateIndex, CollatorId, CollatorSignature, Hash, Id as ParaId,
+		UncheckedSignedAvailabilityBitfield,
 	};
 
 	use polkadot_node_primitives::{
@@ -612,67 +612,16 @@ pub mod vstaging {
 	/// Network messages used by the statement distribution subsystem.
 	#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
 	pub enum StatementDistributionMessage {
-		/// A signed full statement under a given relay-parent.
-		#[codec(index = 0)]
-		Statement(Hash, UncheckedSignedFullStatement),
-		/// Seconded statement with large payload (e.g. containing a runtime upgrade).
+		// TODO [now]: notifications for v2
+		/// All messages for V1 for compatibility with the statement distribution
+		/// protocol, for relay-parents that don't support asynchronous backing.
 		///
-		/// We only gossip the hash in that case, actual payloads can be fetched from sending node
-		/// via request/response.
-		#[codec(index = 1)]
-		LargeStatement(StatementMetadata),
-	}
-
-	/// Data that makes a statement unique.
-	#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq, Hash)]
-	pub struct StatementMetadata {
-		/// Relay parent this statement is relevant under.
-		pub relay_parent: Hash,
-		/// Hash of the candidate that got validated.
-		pub candidate_hash: CandidateHash,
-		/// Validator that attested the validity.
-		pub signed_by: ValidatorIndex,
-		/// Signature of seconding validator.
-		pub signature: ValidatorSignature,
-	}
-
-	impl StatementDistributionMessage {
-		/// Get fingerprint describing the contained statement uniquely.
-		pub fn get_fingerprint(&self) -> (CompactStatement, ValidatorIndex) {
-			match self {
-				Self::Statement(_, statement) => (
-					statement.unchecked_payload().to_compact(),
-					statement.unchecked_validator_index(),
-				),
-				Self::LargeStatement(meta) =>
-					(CompactStatement::Seconded(meta.candidate_hash), meta.signed_by),
-			}
-		}
-
-		/// Get the signature from the statement.
-		pub fn get_signature(&self) -> ValidatorSignature {
-			match self {
-				Self::Statement(_, statement) => statement.unchecked_signature().clone(),
-				Self::LargeStatement(metadata) => metadata.signature.clone(),
-			}
-		}
-
-		/// Get contained relay parent.
-		pub fn get_relay_parent(&self) -> Hash {
-			match self {
-				Self::Statement(r, _) => *r,
-				Self::LargeStatement(meta) => meta.relay_parent,
-			}
-		}
-
-		/// Whether this message contains a large statement.
-		pub fn is_large_statement(&self) -> bool {
-			if let Self::LargeStatement(_) = self {
-				true
-			} else {
-				false
-			}
-		}
+		/// These are illegal to send to V1 peers, and illegal to send concerning relay-parents
+		/// which support asynchronous backing. This backwards compatibility should be
+		/// considered immediately deprecated and can be removed once the node software
+		/// is not required to support asynchronous backing anymore.
+		#[codec(index = 255)]
+		V1Compatibility(crate::v1::StatementDistributionMessage),
 	}
 
 	/// Network messages used by the approval distribution subsystem.