Update disputes prioritisation in dispute-coordinator (#6130)

* Scraper processes CandidateBacked events

* Change definition of best-effort

* Fix `dispute-coordinator` tests

* Unit test for dispute filtering

* Clarification comment

* Add tests

* Fix logic

If a dispute is not backed, not included and not confirmed we
don't participate but we do import votes.

* Add metrics for refrained participations

* Revert "Add tests"

This reverts commit 7b8391a.

* Revert "Unit test for dispute filtering"

This reverts commit 92ba5fe.

* fix dispute-coordinator tests

* Fix scraping

* new tests

* Small fixes in guide

* Apply suggestions from code review

Co-authored-by: Andrei Sandu <[email protected]>

* Fix some comments and remove a pointless test

* Code review feedback

* Clarification comment in tests

* Some tests

* Reference counted `CandidateHash` in scraper

* Proper handling for Backed and Included candidates in scraper

Backed candidates which are not included should be kept for a
predetermined window of finalized blocks. E.g. if a candidate is backed
but not included in block 2, and the window size is 2, the same
candidate should be cleaned after block 4 is finalized.

Add reference counting for candidates in scraper. A candidate can be
added on multiple block heights so we have to make sure we don't clean
it prematurely from the scraper.

Add tests.

* Update comments in tests

* Guide update

* Fix cleanup logic for `backed_candidates_by_block_number`

* Simplify cleanup

* Make spellcheck happy

* Update tests

* Extract candidate backing logic in separate struct

* Code review feedback

* Treat  backed and included candidates in the same fashion

* Update some comments

* Small improvements in test

* spell check

* Fix some more comments

* clean -> prune

* Code review feedback

* Reword comment

* spelling

Co-authored-by: Andrei Sandu <[email protected]>

Author: tdimitrov

node/core/dispute-coordinator/src/initialized.rs

@@ -869,12 +869,21 @@ impl Initialized {
 			}
 		}
 
+		let has_own_vote = new_state.has_own_vote();
+		let is_disputed = new_state.is_disputed();
+		let has_controlled_indices = !env.controlled_indices().is_empty();
+		let is_backed = self.scraper.is_candidate_backed(&candidate_hash);
+		let is_confirmed = new_state.is_confirmed();
+		// We participate only in disputes which are included, backed or confirmed
+		let allow_participation = is_included || is_backed || is_confirmed;
+
 		// Participate in dispute if we did not cast a vote before and actually have keys to cast a
-		// local vote:
-		if !new_state.has_own_vote() &&
-			new_state.is_disputed() &&
-			!env.controlled_indices().is_empty()
-		{
+		// local vote. Disputes should fall in one of the categories below, otherwise we will refrain
+		// from participation:
+		// - `is_included` lands in prioritised queue
+		// - `is_confirmed` | `is_backed` lands in best effort queue
+		// We don't participate in disputes on finalized candidates.
+		if !has_own_vote && is_disputed && has_controlled_indices && allow_participation {
 			let priority = ParticipationPriority::with_priority_if(is_included);
 			gum::trace!(
 				target: LOG_TARGET,
@@ -896,6 +905,23 @@ impl Initialized {
 				)
 				.await;
 			log_error(r)?;
+		} else {
+			gum::trace!(
+				target: LOG_TARGET,
+				?candidate_hash,
+				?is_confirmed,
+				?has_own_vote,
+				?is_disputed,
+				?has_controlled_indices,
+				?allow_participation,
+				?is_included,
+				?is_backed,
+				"Will not queue participation for candidate"
+			);
+
+			if !allow_participation {
+				self.metrics.on_refrained_participation();
+			}
 		}
 
 		// Also send any already existing approval vote on new disputes:

node/core/dispute-coordinator/src/lib.rs

@@ -286,7 +286,7 @@ impl DisputeCoordinatorSubsystem {
 
 		let mut participation_requests = Vec::new();
 		let mut unconfirmed_disputes: UnconfirmedDisputes = UnconfirmedDisputes::new();
-		let (mut scraper, votes) = ChainScraper::new(ctx.sender(), initial_head).await?;
+		let (scraper, votes) = ChainScraper::new(ctx.sender(), initial_head).await?;
 		for ((session, ref candidate_hash), status) in active_disputes {
 			let votes: CandidateVotes =
 				match overlay_db.load_candidate_votes(session, candidate_hash) {

node/core/dispute-coordinator/src/metrics.rs

@@ -30,6 +30,8 @@ struct MetricsInner {
 	queued_participations: prometheus::CounterVec<prometheus::U64>,
 	/// How long vote cleanup batches take.
 	vote_cleanup_time: prometheus::Histogram,
+	/// Number of refrained participations.
+	refrained_participations: prometheus::Counter<prometheus::U64>,
 }
 
 /// Candidate validation metrics.
@@ -88,6 +90,12 @@ impl Metrics {
 	pub(crate) fn time_vote_cleanup(&self) -> Option<prometheus::prometheus::HistogramTimer> {
 		self.0.as_ref().map(|metrics| metrics.vote_cleanup_time.start_timer())
 	}
+
+	pub(crate) fn on_refrained_participation(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.refrained_participations.inc();
+		}
+	}
 }
 
 impl metrics::Metrics for Metrics {
@@ -147,6 +155,14 @@ impl metrics::Metrics for Metrics {
 				)?,
 				registry,
 			)?,
+			refrained_participations: prometheus::register(
+			prometheus::Counter::with_opts(
+				prometheus::Opts::new(
+					"polkadot_parachain_dispute_refrained_participations",
+					"Number of refrained participations. We refrain from participation if all of the following conditions are met: disputed candidate is not included, not backed and not confirmed.",
+				))?,
+				registry,
+			)?,
 		};
 		Ok(Metrics(Some(metrics)))
 	}

node/core/dispute-coordinator/src/scraping/candidates.rs

@@ -0,0 +1,148 @@
+use polkadot_primitives::v2::{BlockNumber, CandidateHash};
+use std::collections::{BTreeMap, HashMap, HashSet};
+
+/// Keeps `CandidateHash` in reference counted way.
+/// Each `insert` saves a value with `reference count == 1` or increases the reference
+/// count if the value already exists.
+/// Each `remove` decreases the reference count for the corresponding `CandidateHash`.
+/// If the reference count reaches 0 - the value is removed.
+struct RefCountedCandidates {
+	candidates: HashMap<CandidateHash, usize>,
+}
+
+impl RefCountedCandidates {
+	pub fn new() -> Self {
+		Self { candidates: HashMap::new() }
+	}
+	// If `CandidateHash` doesn't exist in the `HashMap` it is created and its reference
+	// count is set to 1.
+	// If `CandidateHash` already exists in the `HashMap` its reference count is increased.
+	pub fn insert(&mut self, candidate: CandidateHash) {
+		*self.candidates.entry(candidate).or_default() += 1;
+	}
+
+	// If a `CandidateHash` with reference count equals to 1 is about to be removed - the
+	// candidate is dropped from the container too.
+	// If a `CandidateHash` with reference count biger than 1 is about to be removed - the
+	// reference count is decreased and the candidate remains in the container.
+	pub fn remove(&mut self, candidate: &CandidateHash) {
+		match self.candidates.get_mut(candidate) {
+			Some(v) if *v > 1 => *v -= 1,
+			Some(v) => {
+				assert!(*v == 1);
+				self.candidates.remove(candidate);
+			},
+			None => {},
+		}
+	}
+
+	pub fn contains(&self, candidate: &CandidateHash) -> bool {
+		self.candidates.contains_key(&candidate)
+	}
+}
+
+#[cfg(test)]
+mod ref_counted_candidates_tests {
+	use super::*;
+	use polkadot_primitives::v2::{BlakeTwo256, HashT};
+
+	#[test]
+	fn element_is_removed_when_refcount_reaches_zero() {
+		let mut container = RefCountedCandidates::new();
+
+		let zero = CandidateHash(BlakeTwo256::hash(&vec![0]));
+		let one = CandidateHash(BlakeTwo256::hash(&vec![1]));
+		// add two separate candidates
+		container.insert(zero); // refcount == 1
+		container.insert(one);
+
+		// and increase the reference count for the first
+		container.insert(zero); // refcount == 2
+
+		assert!(container.contains(&zero));
+		assert!(container.contains(&one));
+
+		// remove once -> refcount == 1
+		container.remove(&zero);
+		assert!(container.contains(&zero));
+		assert!(container.contains(&one));
+
+		// remove once -> refcount == 0
+		container.remove(&zero);
+		assert!(!container.contains(&zero));
+		assert!(container.contains(&one));
+
+		// remove the other element
+		container.remove(&one);
+		assert!(!container.contains(&zero));
+		assert!(!container.contains(&one));
+	}
+}
+
+/// Keeps track of scraped candidates. Supports `insert`, `remove_up_to_height` and `contains`
+/// operations.
+pub struct ScrapedCandidates {
+	/// Main data structure which keeps the candidates we know about. `contains` does lookups only here.
+	candidates: RefCountedCandidates,
+	/// Keeps track at which block number a candidate was inserted. Used in `remove_up_to_height`.
+	/// Without this tracking we won't be able to remove all candidates before block X.
+	candidates_by_block_number: BTreeMap<BlockNumber, HashSet<CandidateHash>>,
+}
+
+impl ScrapedCandidates {
+	pub fn new() -> Self {
+		Self {
+			candidates: RefCountedCandidates::new(),
+			candidates_by_block_number: BTreeMap::new(),
+		}
+	}
+
+	pub fn contains(&self, candidate_hash: &CandidateHash) -> bool {
+		self.candidates.contains(candidate_hash)
+	}
+
+	// Removes all candidates up to a given height. The candidates at the block height are NOT removed.
+	pub fn remove_up_to_height(&mut self, height: &BlockNumber) {
+		let not_stale = self.candidates_by_block_number.split_off(&height);
+		let stale = std::mem::take(&mut self.candidates_by_block_number);
+		self.candidates_by_block_number = not_stale;
+		for candidates in stale.values() {
+			for c in candidates {
+				self.candidates.remove(c);
+			}
+		}
+	}
+
+	pub fn insert(&mut self, block_number: BlockNumber, candidate_hash: CandidateHash) {
+		self.candidates.insert(candidate_hash);
+		self.candidates_by_block_number
+			.entry(block_number)
+			.or_default()
+			.insert(candidate_hash);
+	}
+
+	// Used only for tests to verify the pruning doesn't leak data.
+	#[cfg(test)]
+	pub fn candidates_by_block_number_is_empty(&self) -> bool {
+		self.candidates_by_block_number.is_empty()
+	}
+}
+
+#[cfg(test)]
+mod scraped_candidates_tests {
+	use super::*;
+	use polkadot_primitives::v2::{BlakeTwo256, HashT};
+
+	#[test]
+	fn stale_candidates_are_removed() {
+		let mut candidates = ScrapedCandidates::new();
+		let target = CandidateHash(BlakeTwo256::hash(&vec![1, 2, 3]));
+		candidates.insert(1, target);
+
+		assert!(candidates.contains(&target));
+
+		candidates.remove_up_to_height(&2);
+		assert!(!candidates.contains(&target));
+		assert!(candidates.candidates_by_block_number_is_empty());
+	}
+}