backing: introduce CanSecond request for advertisements filtering (#…

…6225)

* Drop BoundToRelayParent

* draft changes

* fix backing tests

* Fix genesis ancestry

* Fix validator side tests

* more tests

node/core/backing/src/lib.rs

@@ -85,10 +85,10 @@ use polkadot_node_primitives::{
 };
 use polkadot_node_subsystem::{
 	messages::{
-		AvailabilityDistributionMessage, AvailabilityStoreMessage, CandidateBackingMessage,
-		CandidateValidationMessage, CollatorProtocolMessage, HypotheticalDepthRequest,
-		ProspectiveParachainsMessage, ProvisionableData, ProvisionerMessage, RuntimeApiMessage,
-		RuntimeApiRequest, StatementDistributionMessage,
+		AvailabilityDistributionMessage, AvailabilityStoreMessage, CanSecondRequest,
+		CandidateBackingMessage, CandidateValidationMessage, CollatorProtocolMessage,
+		HypotheticalDepthRequest, ProspectiveParachainsMessage, ProvisionableData,
+		ProvisionerMessage, RuntimeApiMessage, RuntimeApiRequest, StatementDistributionMessage,
 	},
 	overseer, ActiveLeavesUpdate, FromOrchestra, OverseerSignal, SpawnedSubsystem, SubsystemError,
 };
@@ -758,6 +758,8 @@ async fn handle_communication<Context>(
 					"Received `GetBackedCandidates` for an unknown relay parent."
 				);
 			},
+		CandidateBackingMessage::CanSecond(request, tx) =>
+			handle_can_second_request(ctx, state, request, tx).await,
 	}
 
 	Ok(())
@@ -1093,22 +1095,8 @@ async fn seconding_sanity_check<Context>(
 	candidate_hash: CandidateHash,
 	candidate_para: ParaId,
 	parent_head_data_hash: Hash,
-	head_data_hash: Hash,
 	candidate_relay_parent: Hash,
 ) -> SecondingAllowed {
-	// Note that `GetHypotheticalDepths` doesn't account for recursion,
-	// i.e. candidates can appear at multiple depths in the tree and in fact
-	// at all depths, and we don't know what depths a candidate will ultimately occupy
-	// because that's dependent on other candidates we haven't yet received.
-	//
-	// The only way to effectively rule this out is to have candidate receipts
-	// directly commit to the parachain block number or some other incrementing
-	// counter. That requires a major primitives format upgrade, so for now
-	// we just rule out trivial cycles.
-	if parent_head_data_hash == head_data_hash {
-		return SecondingAllowed::No
-	}
-
 	let mut membership = Vec::new();
 	let mut responses = FuturesOrdered::<BoxFuture<'_, Result<_, oneshot::Canceled>>>::new();
 
@@ -1193,6 +1181,46 @@ async fn seconding_sanity_check<Context>(
 	SecondingAllowed::Yes(membership)
 }
 
+/// Performs seconding sanity check for an advertisement.
+#[overseer::contextbounds(CandidateBacking, prefix = self::overseer)]
+async fn handle_can_second_request<Context>(
+	ctx: &mut Context,
+	state: &State,
+	request: CanSecondRequest,
+	tx: oneshot::Sender<bool>,
+) {
+	let relay_parent = request.candidate_relay_parent;
+	let response = if state
+		.per_relay_parent
+		.get(&relay_parent)
+		.map_or(false, |pr_state| pr_state.prospective_parachains_mode.is_enabled())
+	{
+		let result = seconding_sanity_check(
+			ctx,
+			&state.per_leaf,
+			&state.implicit_view,
+			request.candidate_hash,
+			request.candidate_para_id,
+			request.parent_head_data_hash,
+			relay_parent,
+		)
+		.await;
+
+		match result {
+			SecondingAllowed::No => false,
+			SecondingAllowed::Yes(membership) => {
+				// Candidate should be recognized by at least some fragment tree.
+				membership.iter().any(|(_, m)| !m.is_empty())
+			},
+		}
+	} else {
+		// Relay parent is unknown or async backing is disabled.
+		false
+	};
+
+	let _ = tx.send(response);
+}
+
 #[overseer::contextbounds(CandidateBacking, prefix = self::overseer)]
 async fn handle_validated_candidate_command<Context>(
 	ctx: &mut Context,
@@ -1219,6 +1247,19 @@ async fn handle_validated_candidate_command<Context>(
 							return Ok(())
 						}
 
+						let parent_head_data_hash = persisted_validation_data.parent_head.hash();
+						// Note that `GetHypotheticalDepths` doesn't account for recursion,
+						// i.e. candidates can appear at multiple depths in the tree and in fact
+						// at all depths, and we don't know what depths a candidate will ultimately occupy
+						// because that's dependent on other candidates we haven't yet received.
+						//
+						// The only way to effectively rule this out is to have candidate receipts
+						// directly commit to the parachain block number or some other incrementing
+						// counter. That requires a major primitives format upgrade, so for now
+						// we just rule out trivial cycles.
+						if parent_head_data_hash == commitments.head_data.hash() {
+							return Ok(())
+						}
 						// sanity check that we're allowed to second the candidate
 						// and that it doesn't conflict with other candidates we've
 						// seconded.
@@ -1229,7 +1270,6 @@ async fn handle_validated_candidate_command<Context>(
 							candidate_hash,
 							candidate.descriptor().para_id,
 							persisted_validation_data.parent_head.hash(),
-							commitments.head_data.hash(),
 							candidate.descriptor().relay_parent,
 						)
 						.await
@@ -1517,14 +1557,21 @@ async fn import_statement<Context>(
 					"Candidate backed",
 				);
 
-				// Inform the prospective parachains subsystem
-				// that the candidate is now backed.
 				if rp_state.prospective_parachains_mode.is_enabled() {
+					// Inform the prospective parachains subsystem
+					// that the candidate is now backed.
 					ctx.send_message(ProspectiveParachainsMessage::CandidateBacked(
 						para_id,
 						candidate_hash,
 					))
 					.await;
+					// Backed candidate potentially unblocks new advertisements,
+					// notify collator protocol.
+					ctx.send_message(CollatorProtocolMessage::Backed {
+						para_id,
+						para_head: backed.candidate.descriptor.para_head,
+					})
+					.await;
 				} else {
 					// The provisioner waits on candidate-backing, which means
 					// that we need to send unbounded messages to avoid cycles.

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

@@ -212,7 +212,7 @@ impl TestCandidateBuilder {
 				erasure_root: self.erasure_root,
 				collator: dummy_collator(),
 				signature: dummy_collator_signature(),
-				para_head: dummy_hash(),
+				para_head: self.head_data.hash(),
 				validation_code_hash: ValidationCode(self.validation_code).hash(),
 				persisted_validation_data_hash: self.persisted_validation_data_hash,
 			},

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

@@ -959,6 +959,7 @@ fn backing_works() {
 		.build();
 
 		let candidate_a_hash = candidate_a.hash();
+		let candidate_a_para_head = candidate_a.descriptor().para_head;
 
 		let public1 = CryptoStore::sr25519_generate_new(
 			&*test_state.keystore,
@@ -1034,7 +1035,7 @@ fn backing_works() {
 		)
 		.await;
 
-		// Prospective parachains are notified about candidate backed.
+		// Prospective parachains and collator protocol are notified about candidate backed.
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::ProspectiveParachains(
@@ -1043,6 +1044,13 @@ fn backing_works() {
 				),
 			) if candidate_a_hash == candidate_hash && candidate_para_id == para_id
 		);
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::CollatorProtocol(CollatorProtocolMessage::Backed {
+				para_id: _para_id,
+				para_head,
+			}) if para_id == _para_id && candidate_a_para_head == para_head
+		);
 
 		assert_matches!(
 			virtual_overseer.recv().await,
@@ -1255,6 +1263,7 @@ fn concurrent_dependent_candidates() {
 				AllMessages::ProspectiveParachains(
 					ProspectiveParachainsMessage::CandidateBacked(..),
 				) => {},
+				AllMessages::CollatorProtocol(CollatorProtocolMessage::Backed { .. }) => {},
 				AllMessages::StatementDistribution(StatementDistributionMessage::Share(
 					_,
 					statement,

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

@@ -789,12 +789,6 @@ async fn process_msg<Context>(
 				},
 			}
 		},
-		ReportCollator(_) => {
-			gum::warn!(
-				target: LOG_TARGET,
-				"ReportCollator message is not expected on the collator side of the protocol",
-			);
-		},
 		NetworkBridgeUpdate(event) => {
 			// We should count only this shoulder in the histogram, as other shoulders are just introducing noise
 			let _ = state.metrics.time_process_msg();
@@ -807,7 +801,13 @@ async fn process_msg<Context>(
 				);
 			}
 		},
-		_ => {},
+		msg @ (ReportCollator(..) | Invalid(..) | Seconded(..) | Backed { .. }) => {
+			gum::warn!(
+				target: LOG_TARGET,
+				"{:?} message is not expected on the collator side of the protocol",
+				msg,
+			);
+		},
 	}
 
 	Ok(())

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

@@ -112,6 +112,21 @@ impl PendingCollation {
 	}
 }
 
+/// vstaging advertisement that was rejected by the backing
+/// subsystem. Validator may fetch it later if its fragment
+/// membership gets recognized before relay parent goes out of view.
+#[derive(Debug, Clone)]
+pub struct BlockedAdvertisement {
+	/// Peer that advertised the collation.
+	pub peer_id: PeerId,
+	/// Collator id.
+	pub collator_id: CollatorId,
+	/// The relay-parent of the candidate.
+	pub candidate_relay_parent: Hash,
+	/// Hash of the candidate.
+	pub candidate_hash: CandidateHash,
+}
+
 /// Performs a sanity check between advertised and fetched collations.
 ///
 /// Since the persisted validation data is constructed using the advertised