statement-distribution: implement validator disabling

Author: ordian

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

@@ -75,6 +75,9 @@ pub enum Error {
 	#[error("Fetching availability cores failed {0:?}")]
 	FetchAvailabilityCores(RuntimeApiError),
 
+	#[error("Fetching disabled validators failed {0:?}")]
+	FetchDisabledValidators(RuntimeApiError),
+
 	#[error("Fetching validator groups failed {0:?}")]
 	FetchValidatorGroups(RuntimeApiError),
 

polkadot/node/network/statement-distribution/src/v2/grid.rs

@@ -253,7 +253,9 @@ impl GridTracker {
 	/// This checks whether the peer is allowed to send us manifests
 	/// about this group at this relay-parent. This also does sanity
 	/// checks on the format of the manifest and the amount of votes
-	/// it contains. It has effects on the stored state only when successful.
+	/// it contains. It assumes that the votes from disabled validators
+	/// are already filtered out.
+	/// It has effects on the stored state only when successful.
 	///
 	/// This returns a `bool` on success, which if true indicates that an acknowledgement is
 	/// to be sent in response to the received manifest. This only occurs when the

polkadot/node/network/statement-distribution/src/v2/mod.rs

@@ -95,6 +95,7 @@ const COST_UNEXPECTED_STATEMENT: Rep = Rep::CostMinor("Unexpected Statement");
 const COST_UNEXPECTED_STATEMENT_MISSING_KNOWLEDGE: Rep =
 	Rep::CostMinor("Unexpected Statement, missing knowledge for relay parent");
 const COST_EXCESSIVE_SECONDED: Rep = Rep::CostMinor("Sent Excessive `Seconded` Statements");
+const COST_DISABLED_VALIDATOR: Rep = Rep::CostMajor("Sent a statement from a disabled validator");
 
 const COST_UNEXPECTED_MANIFEST_MISSING_KNOWLEDGE: Rep =
 	Rep::CostMinor("Unexpected Manifest, missing knowlege for relay parent");
@@ -473,6 +474,18 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 		.map_err(JfyiError::RuntimeApiUnavailable)?
 		.map_err(JfyiError::FetchAvailabilityCores)?;
 
+		let disabled_validators = polkadot_node_subsystem_util::request_disabled_validators(
+			new_relay_parent,
+			ctx.sender(),
+		)
+		.await
+		.await
+		.map_err(JfyiError::RuntimeApiUnavailable)?
+		.map_err(JfyiError::FetchDisabledValidators)?;
+
+		// deduplicate and order
+		let disabled_validators = disabled_validators.into_iter().collect();
+
 		let group_rotation_info =
 			polkadot_node_subsystem_util::request_validator_groups(new_relay_parent, ctx.sender())
 				.await
@@ -533,7 +546,7 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 			new_relay_parent,
 			PerRelayParentState {
 				local_validator,
-				statement_store: StatementStore::new(&per_session.groups),
+				statement_store: StatementStore::new(&per_session.groups, &disabled_validators),
 				availability_cores,
 				group_rotation_info,
 				seconding_limit,
@@ -1300,6 +1313,20 @@ async fn handle_incoming_statement<Context>(
 			},
 		};
 
+	if per_relay_parent
+		.statement_store
+		.is_disabled(&statement.unchecked_validator_index())
+	{
+		gum::debug!(
+			target: LOG_TARGET,
+			?relay_parent,
+			validator_index = ?statement.unchecked_validator_index(),
+			"Ignoring a statement from disabled validator."
+		);
+		modify_reputation(reputation, ctx.sender(), peer, COST_DISABLED_VALIDATOR).await;
+		return
+	}
+
 	let cluster_sender_index = {
 		// This block of code only returns `Some` when both the originator and
 		// the sending peer are in the cluster.
@@ -1412,7 +1439,7 @@ async fn handle_incoming_statement<Context>(
 		checked_statement.clone(),
 		StatementOrigin::Remote,
 	) {
-		Err(statement_store::ValidatorUnknown) => {
+		Err(statement_store::Error::ValidatorUnknown) => {
 			// sanity: should never happen.
 			gum::warn!(
 				target: LOG_TARGET,
@@ -1423,6 +1450,17 @@ async fn handle_incoming_statement<Context>(
 
 			return
 		},
+		Err(statement_store::Error::ValidatorDisabled) => {
+			// sanity: should never happen, checked above.
+			gum::warn!(
+				target: LOG_TARGET,
+				?relay_parent,
+				validator_index = ?originator_index,
+				"Error - accepted message from disabled validator."
+			);
+
+			return
+		},
 		Ok(known) => known,
 	};
 
@@ -1906,7 +1944,7 @@ async fn handle_incoming_manifest_common<'a, Context>(
 	candidate_hash: CandidateHash,
 	relay_parent: Hash,
 	para_id: ParaId,
-	manifest_summary: grid::ManifestSummary,
+	mut manifest_summary: grid::ManifestSummary,
 	manifest_kind: grid::ManifestKind,
 	reputation: &mut ReputationAggregator,
 ) -> Option<ManifestImportSuccess<'a>> {
@@ -1984,6 +2022,16 @@ async fn handle_incoming_manifest_common<'a, Context>(
 	// 2. sanity checks: peer is validator, bitvec size, import into grid tracker
 	let group_index = manifest_summary.claimed_group_index;
 	let claimed_parent_hash = manifest_summary.claimed_parent_hash;
+
+	// Ignore votes from disabled validators when counting towards the threshold.
+	let disabled_mask = per_session
+		.groups
+		.get(group_index)
+		.map(|group| relay_parent_state.statement_store.disabled_bitmask(group))
+		.unwrap_or_default();
+	manifest_summary.statement_knowledge.mask_seconded(&disabled_mask);
+	manifest_summary.statement_knowledge.mask_valid(&disabled_mask);
+
 	let acknowledge = match local_validator.grid_tracker.import_manifest(
 		grid_topology,
 		&per_session.groups,
@@ -2743,11 +2791,20 @@ pub(crate) fn answer_request(state: &mut State, message: ResponderMessage) {
 
 	// Transform mask with 'OR' semantics into one with 'AND' semantics for the API used
 	// below.
-	let and_mask = StatementFilter {
+	let mut and_mask = StatementFilter {
 		seconded_in_group: !mask.seconded_in_group.clone(),
 		validated_in_group: !mask.validated_in_group.clone(),
 	};
 
+	// Ignore disabled validators when sending the response.
+	let disabled_mask = per_session
+		.groups
+		.get(confirmed.group_index())
+		.map(|group| relay_parent_state.statement_store.disabled_bitmask(group))
+		.unwrap_or_default();
+	and_mask.mask_seconded(&disabled_mask);
+	and_mask.mask_valid(&disabled_mask);
+
 	let response = AttestedCandidateResponse {
 		candidate_receipt: (&**confirmed.candidate_receipt()).clone(),
 		persisted_validation_data: confirmed.persisted_validation_data().clone(),

polkadot/node/network/statement-distribution/src/v2/statement_store.rs

@@ -28,7 +28,10 @@ use polkadot_node_network_protocol::v2::StatementFilter;
 use polkadot_primitives::{
 	CandidateHash, CompactStatement, GroupIndex, SignedStatement, ValidatorIndex,
 };
-use std::collections::hash_map::{Entry as HEntry, HashMap};
+use std::collections::{
+	hash_map::{Entry as HEntry, HashMap},
+	BTreeSet,
+};
 
 use super::groups::Groups;
 
@@ -68,7 +71,7 @@ pub struct StatementStore {
 
 impl StatementStore {
 	/// Create a new [`StatementStore`]
-	pub fn new(groups: &Groups) -> Self {
+	pub fn new(groups: &Groups, disabled: &BTreeSet<ValidatorIndex>) -> Self {
 		let mut validator_meta = HashMap::new();
 		for (g, group) in groups.all().iter().enumerate() {
 			for (i, v) in group.iter().enumerate() {
@@ -78,6 +81,7 @@ impl StatementStore {
 						seconded_count: 0,
 						within_group_index: i,
 						group: GroupIndex(g as _),
+						is_disabled: disabled.contains(v),
 					},
 				);
 			}
@@ -91,18 +95,21 @@ impl StatementStore {
 	}
 
 	/// Insert a statement. Returns `true` if was not known already, `false` if it was.
-	/// Ignores statements by unknown validators and returns an error.
+	/// Ignores statements by unknown and disabled validators and returns an error.
 	pub fn insert(
 		&mut self,
 		groups: &Groups,
 		statement: SignedStatement,
 		origin: StatementOrigin,
-	) -> Result<bool, ValidatorUnknown> {
+	) -> Result<bool, Error> {
 		let validator_index = statement.validator_index();
 		let validator_meta = match self.validator_meta.get_mut(&validator_index) {
-			None => return Err(ValidatorUnknown),
+			None => return Err(Error::ValidatorUnknown),
 			Some(m) => m,
 		};
+		if validator_meta.is_disabled {
+			return Err(Error::ValidatorDisabled)
+		}
 
 		let compact = statement.payload().clone();
 		let fingerprint = (validator_index, compact.clone());
@@ -134,7 +141,7 @@ impl StatementStore {
 						"groups passed into `insert` differ from those used at store creation"
 					);
 
-					return Err(ValidatorUnknown)
+					return Err(Error::ValidatorUnknown)
 				},
 			};
 
@@ -154,6 +161,25 @@ impl StatementStore {
 		Ok(true)
 	}
 
+	/// Returns true if the validator is disabled as of current relay parent.
+	pub fn is_disabled(&self, index: &ValidatorIndex) -> bool {
+		self.validator_meta.get(&index).map(|m| m.is_disabled).unwrap_or(false)
+	}
+
+	/// A convenience funtion to generate a disabled bitmask for the given backing group.
+	/// The output bits are set to `true` for validators that are disabled.
+	/// The group size should match the size of the backing group.
+	pub fn disabled_bitmask(&self, group: &[ValidatorIndex]) -> BitVec<u8, BitOrderLsb0> {
+		let group_size = group.len();
+		let mut mask = BitVec::repeat(false, group_size);
+		for i in group {
+			if self.is_disabled(i) {
+				mask.set(i.0 as usize, true);
+			}
+		}
+		mask
+	}
+
 	/// Fill a `StatementFilter` to be used in the grid topology with all statements
 	/// we are already aware of.
 	pub fn fill_statement_filter(
@@ -250,14 +276,20 @@ impl StatementStore {
 }
 
 /// Error indicating that the validator was unknown.
-pub struct ValidatorUnknown;
+pub enum Error {
+	/// The validator was unknown.
+	ValidatorUnknown,
+	/// A statement from a disabled validator should be rejected.
+	ValidatorDisabled,
+}
 
 type Fingerprint = (ValidatorIndex, CompactStatement);
 
 struct ValidatorMeta {
 	group: GroupIndex,
 	within_group_index: usize,
 	seconded_count: usize,
+	is_disabled: bool,
 }
 
 struct GroupStatements {

polkadot/node/network/statement-distribution/src/v2/tests/mod.rs

@@ -169,12 +169,21 @@ impl TestState {
 					collator: None,
 				})
 			}),
+			disabled_validators: Default::default(),
 			para_data: (0..self.session_info.validator_groups.len())
 				.map(|i| (ParaId::from(i as u32), PerParaData::new(1, vec![1, 2, 3].into())))
 				.collect(),
 		}
 	}
 
+	fn make_dummy_leaf_with_disabled_validators(
+		&self,
+		relay_parent: Hash,
+		disabled_validators: Vec<ValidatorIndex>,
+	) -> TestLeaf {
+		TestLeaf { disabled_validators, ..self.make_dummy_leaf(relay_parent) }
+	}
+
 	fn make_availability_cores(&self, f: impl Fn(usize) -> CoreState) -> Vec<CoreState> {
 		(0..self.session_info.validator_groups.len()).map(f).collect()
 	}
@@ -340,6 +349,7 @@ struct TestLeaf {
 	parent_hash: Hash,
 	session: SessionIndex,
 	availability_cores: Vec<CoreState>,
+	disabled_validators: Vec<ValidatorIndex>,
 	para_data: Vec<(ParaId, PerParaData)>,
 }
 
@@ -375,7 +385,15 @@ async fn handle_leaf_activation(
 	test_state: &TestState,
 	is_new_session: bool,
 ) {
-	let TestLeaf { number, hash, parent_hash, para_data, session, availability_cores } = leaf;
+	let TestLeaf {
+		number,
+		hash,
+		parent_hash,
+		para_data,
+		session,
+		availability_cores,
+		disabled_validators,
+	} = leaf;
 
 	assert_matches!(
 		virtual_overseer.recv().await,
@@ -431,6 +449,14 @@ async fn handle_leaf_activation(
 		}
 	);
 
+	assert_matches!(
+		virtual_overseer.recv().await,
+		AllMessages::RuntimeApi(
+			RuntimeApiMessage::Request(parent, RuntimeApiRequest::DisabledValidators(tx))) if parent == *hash => {
+			tx.send(Ok(disabled_validators.clone())).unwrap();
+		}
+	);
+
 	let validator_groups = test_state.session_info.validator_groups.to_vec();
 	let group_rotation_info =
 		GroupRotationInfo { session_start_block: 1, group_rotation_frequency: 12, now: 1 };