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 7b8391a087922ced942cde9cd2b50ff3f633efc0.

* Revert "Unit test for dispute filtering"

This reverts commit 92ba5fe678214ab360306313a33c781338e600a0.

* fix dispute-coordinator tests

* Fix scraping

* new tests

* Small fixes in guide

* Apply suggestions from code review

Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>

* 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 <54316454+sandreim@users.noreply.github.com>

polkadot/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:

polkadot/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) {

polkadot/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)))
 	}

polkadot/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());
+	}
+}

polkadot/node/core/dispute-coordinator/src/scraping/mod.rs

@@ -14,10 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::{
-	collections::{BTreeMap, HashSet},
-	num::NonZeroUsize,
-};
+use std::num::NonZeroUsize;
 
 use futures::channel::oneshot;
 use lru::LruCache;
@@ -40,6 +37,8 @@ use crate::{
 #[cfg(test)]
 mod tests;
 
+mod candidates;
+
 /// Number of hashes to keep in the LRU.
 ///
 ///
@@ -58,18 +57,21 @@ const LRU_OBSERVED_BLOCKS_CAPACITY: NonZeroUsize = match NonZeroUsize::new(20) {
 ///
 /// Concretely:
 ///
-/// - Monitors for inclusion events to keep track of candidates that have been included on chains.
+/// - Monitors for `CandidateIncluded` events to keep track of candidates that have been
+///   included on chains.
+/// - Monitors for `CandidateBacked` events to keep track of all backed candidates.
 /// - Calls `FetchOnChainVotes` for each block to gather potentially missed votes from chain.
 ///
 /// With this information it provides a `CandidateComparator` and as a return value of
 /// `process_active_leaves_update` any scraped votes.
+///
+/// Scraped candidates are available `CANDIDATE_LIFETIME_AFTER_FINALIZATION` more blocks
+/// after finalization as a precaution not to prune them prematurely.
 pub struct ChainScraper {
 	/// All candidates we have seen included, which not yet have been finalized.
-	included_candidates: HashSet<CandidateHash>,
-	/// including block -> `CandidateHash`
-	///
-	/// We need this to clean up `included_candidates` on finalization.
-	candidates_by_block_number: BTreeMap<BlockNumber, HashSet<CandidateHash>>,
+	included_candidates: candidates::ScrapedCandidates,
+	/// All candidates we have seen backed
+	backed_candidates: candidates::ScrapedCandidates,
 	/// Latest relay blocks observed by the provider.
 	///
 	/// We assume that ancestors of cached blocks are already processed, i.e. we have saved
@@ -85,6 +87,16 @@ impl ChainScraper {
 	/// As long as we have `MAX_FINALITY_LAG` this makes sense as a value.
 	pub(crate) const ANCESTRY_SIZE_LIMIT: u32 = MAX_FINALITY_LAG;
 
+	/// How many blocks after finalization a backed/included candidate should be kept.
+	/// We don't want to remove scraped candidates on finalization because we want to
+	/// be sure that disputes will conclude on abandoned forks.
+	/// Removing the candidate on finalization creates a possibility for an attacker to
+	/// avoid slashing. If a bad fork is abandoned too quickly because in the same another
+	/// better one gets finalized the entries for the bad fork will be pruned and we
+	/// will never participate in a dispute for it. We want such disputes to conclude
+	/// in a timely manner so that the offenders are slashed.
+	pub(crate) const CANDIDATE_LIFETIME_AFTER_FINALIZATION: BlockNumber = 2;
+
 	/// Create a properly initialized `OrderingProvider`.
 	///
 	/// Returns: `Self` and any scraped votes.
@@ -96,8 +108,8 @@ impl ChainScraper {
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
 		let mut s = Self {
-			included_candidates: HashSet::new(),
-			candidates_by_block_number: BTreeMap::new(),
+			included_candidates: candidates::ScrapedCandidates::new(),
+			backed_candidates: candidates::ScrapedCandidates::new(),
 			last_observed_blocks: LruCache::new(LRU_OBSERVED_BLOCKS_CAPACITY),
 		};
 		let update =
@@ -107,10 +119,15 @@ impl ChainScraper {
 	}
 
 	/// Check whether we have seen a candidate included on any chain.
-	pub fn is_candidate_included(&mut self, candidate_hash: &CandidateHash) -> bool {
+	pub fn is_candidate_included(&self, candidate_hash: &CandidateHash) -> bool {
 		self.included_candidates.contains(candidate_hash)
 	}
 
+	/// Check whether the candidate is backed
+	pub fn is_candidate_backed(&self, candidate_hash: &CandidateHash) -> bool {
+		self.backed_candidates.contains(candidate_hash)
+	}
+
 	/// Query active leaves for any candidate `CandidateEvent::CandidateIncluded` events.
 	///
 	/// and updates current heads, so we can query candidates for all non finalized blocks.
@@ -120,7 +137,7 @@ impl ChainScraper {
 		&mut self,
 		sender: &mut Sender,
 		update: &ActiveLeavesUpdate,
-	) -> crate::error::Result<Vec<ScrapedOnChainVotes>>
+	) -> Result<Vec<ScrapedOnChainVotes>>
 	where
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
@@ -158,21 +175,27 @@ impl ChainScraper {
 
 	/// Prune finalized candidates.
 	///
-	/// Once a candidate lives in a relay chain block that's behind the finalized chain/got
-	/// finalized, we can treat it as low priority.
-	pub fn process_finalized_block(&mut self, finalized: &BlockNumber) {
-		let not_finalized = self.candidates_by_block_number.split_off(finalized);
-		let finalized = std::mem::take(&mut self.candidates_by_block_number);
-		self.candidates_by_block_number = not_finalized;
-		// Clean up finalized:
-		for finalized_candidate in finalized.into_values().flatten() {
-			self.included_candidates.remove(&finalized_candidate);
+	/// We keep each candidate for `CANDIDATE_LIFETIME_AFTER_FINALIZATION` blocks after finalization.
+	/// After that we treat it as low priority.
+	pub fn process_finalized_block(&mut self, finalized_block_number: &BlockNumber) {
+		// `CANDIDATE_LIFETIME_AFTER_FINALIZATION - 1` because `finalized_block_number`counts to the
+		// candidate lifetime.
+		match finalized_block_number.checked_sub(Self::CANDIDATE_LIFETIME_AFTER_FINALIZATION - 1) {
+			Some(key_to_prune) => {
+				self.backed_candidates.remove_up_to_height(&key_to_prune);
+				self.included_candidates.remove_up_to_height(&key_to_prune);
+			},
+			None => {
+				// Nothing to prune. We are still in the beginning of the chain and there are not
+				// enough finalized blocks yet.
+			},
 		}
+		{}
 	}
 
 	/// Process candidate events of a block.
 	///
-	/// Keep track of all included candidates.
+	/// Keep track of all included and backed candidates.
 	async fn process_candidate_events<Sender>(
 		&mut self,
 		sender: &mut Sender,
@@ -182,28 +205,33 @@ impl ChainScraper {
 	where
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
-		// Get included events:
-		let included =
-			get_candidate_events(sender, block_hash)
-				.await?
-				.into_iter()
-				.filter_map(|ev| match ev {
-					CandidateEvent::CandidateIncluded(receipt, _, _, _) => Some(receipt),
-					_ => None,
-				});
-		for receipt in included {
-			let candidate_hash = receipt.hash();
-			gum::trace!(
-				target: LOG_TARGET,
-				?candidate_hash,
-				?block_number,
-				"Processing included event"
-			);
-			self.included_candidates.insert(candidate_hash);
-			self.candidates_by_block_number
-				.entry(block_number)
-				.or_default()
-				.insert(candidate_hash);
+		// Get included and backed events:
+		for ev in get_candidate_events(sender, block_hash).await? {
+			match ev {
+				CandidateEvent::CandidateIncluded(receipt, _, _, _) => {
+					let candidate_hash = receipt.hash();
+					gum::trace!(
+						target: LOG_TARGET,
+						?candidate_hash,
+						?block_number,
+						"Processing included event"
+					);
+					self.included_candidates.insert(block_number, candidate_hash);
+				},
+				CandidateEvent::CandidateBacked(receipt, _, _, _) => {
+					let candidate_hash = receipt.hash();
+					gum::trace!(
+						target: LOG_TARGET,
+						?candidate_hash,
+						?block_number,
+						"Processing backed event"
+					);
+					self.backed_candidates.insert(block_number, candidate_hash);
+				},
+				_ => {
+					// skip the rest
+				},
+			}
 		}
 		Ok(())
 	}

polkadot/node/core/dispute-coordinator/src/scraping/tests.rs

@@ -73,7 +73,12 @@ impl TestState {
 			assert_finalized_block_number_request(&mut ctx_handle, finalized_block_number).await;
 			gum::trace!(target: LOG_TARGET, "After assert_finalized_block_number");
 			// No ancestors requests, as list would be empty.
-			assert_candidate_events_request(&mut ctx_handle, &chain).await;
+			assert_candidate_events_request(
+				&mut ctx_handle,
+				&chain,
+				get_backed_and_included_candidate_events,
+			)
+			.await;
 			assert_chain_vote_request(&mut ctx_handle, &chain).await;
 		};
 
@@ -112,6 +117,10 @@ async fn process_active_leaves_update(
 		.unwrap();
 }
 
+fn process_finalized_block(scraper: &mut ChainScraper, finalized: &BlockNumber) {
+	scraper.process_finalized_block(&finalized)
+}
+
 fn make_candidate_receipt(relay_parent: Hash) -> CandidateReceipt {
 	let zeros = dummy_hash();
 	let descriptor = CandidateDescriptor {
@@ -145,16 +154,66 @@ fn get_block_number_hash(n: BlockNumber) -> Hash {
 }
 
 /// Get a dummy event that corresponds to candidate inclusion for the given block number.
-fn get_candidate_included_events(block_number: BlockNumber) -> Vec<CandidateEvent> {
-	vec![CandidateEvent::CandidateIncluded(
-		make_candidate_receipt(get_block_number_hash(block_number)),
+fn get_backed_and_included_candidate_events(block_number: BlockNumber) -> Vec<CandidateEvent> {
+	let candidate_receipt = make_candidate_receipt(get_block_number_hash(block_number));
+	vec![
+		CandidateEvent::CandidateIncluded(
+			candidate_receipt.clone(),
+			HeadData::default(),
+			CoreIndex::from(0),
+			GroupIndex::from(0),
+		),
+		CandidateEvent::CandidateBacked(
+			candidate_receipt,
+			HeadData::default(),
+			CoreIndex::from(0),
+			GroupIndex::from(0),
+		),
+	]
+}
+
+fn get_backed_candidate_event(block_number: BlockNumber) -> Vec<CandidateEvent> {
+	let candidate_receipt = make_candidate_receipt(get_block_number_hash(block_number));
+	vec![CandidateEvent::CandidateBacked(
+		candidate_receipt,
 		HeadData::default(),
 		CoreIndex::from(0),
 		GroupIndex::from(0),
 	)]
 }
+/// Hash for a 'magic' candidate. This is meant to be a special candidate used to verify special cases.
+fn get_magic_candidate_hash() -> Hash {
+	BlakeTwo256::hash(&"abc".encode())
+}
+/// Get a dummy event that corresponds to candidate inclusion for a hardcoded block number.
+/// Used to simulate candidates included multiple times at different block heights.
+fn get_backed_and_included_magic_candidate_events(
+	_block_number: BlockNumber,
+) -> Vec<CandidateEvent> {
+	let candidate_receipt = make_candidate_receipt(get_magic_candidate_hash());
+	vec![
+		CandidateEvent::CandidateIncluded(
+			candidate_receipt.clone(),
+			HeadData::default(),
+			CoreIndex::from(0),
+			GroupIndex::from(0),
+		),
+		CandidateEvent::CandidateBacked(
+			candidate_receipt,
+			HeadData::default(),
+			CoreIndex::from(0),
+			GroupIndex::from(0),
+		),
+	]
+}
 
-async fn assert_candidate_events_request(virtual_overseer: &mut VirtualOverseer, chain: &[Hash]) {
+async fn assert_candidate_events_request<F>(
+	virtual_overseer: &mut VirtualOverseer,
+	chain: &[Hash],
+	event_generator: F,
+) where
+	F: Fn(u32) -> Vec<CandidateEvent>,
+{
 	assert_matches!(
 		overseer_recv(virtual_overseer).await,
 		AllMessages::RuntimeApi(RuntimeApiMessage::Request(
@@ -163,7 +222,7 @@ async fn assert_candidate_events_request(virtual_overseer: &mut VirtualOverseer,
 		)) => {
 			let maybe_block_number = chain.iter().position(|h| *h == hash);
 			let response = maybe_block_number
-				.map(|num| get_candidate_included_events(num as u32))
+				.map(|num| event_generator(num as u32))
 				.unwrap_or_default();
 			tx.send(Ok(response)).unwrap();
 		}
@@ -207,12 +266,15 @@ async fn assert_block_ancestors_request(virtual_overseer: &mut VirtualOverseer,
 	);
 }
 
-async fn overseer_process_active_leaves_update(
+async fn overseer_process_active_leaves_update<F>(
 	virtual_overseer: &mut VirtualOverseer,
 	chain: &[Hash],
 	finalized_block: BlockNumber,
 	expected_ancestry_len: usize,
-) {
+	event_generator: F,
+) where
+	F: Fn(u32) -> Vec<CandidateEvent> + Clone,
+{
 	// Before walking through ancestors provider requests latest finalized block number.
 	assert_finalized_block_number_request(virtual_overseer, finalized_block).await;
 	// Expect block ancestors requests with respect to the ancestry step.
@@ -221,7 +283,7 @@ async fn overseer_process_active_leaves_update(
 	}
 	// For each ancestry and the head return corresponding candidates inclusions.
 	for _ in 0..expected_ancestry_len {
-		assert_candidate_events_request(virtual_overseer, chain).await;
+		assert_candidate_events_request(virtual_overseer, chain, event_generator.clone()).await;
 		assert_chain_vote_request(virtual_overseer, chain).await;
 	}
 }
@@ -236,7 +298,9 @@ fn scraper_provides_included_state_when_initialized() {
 		let TestState { mut chain, mut scraper, mut ctx } = state;
 
 		assert!(!scraper.is_candidate_included(&candidate_2.hash()));
+		assert!(!scraper.is_candidate_backed(&candidate_2.hash()));
 		assert!(scraper.is_candidate_included(&candidate_1.hash()));
+		assert!(scraper.is_candidate_backed(&candidate_1.hash()));
 
 		// After next active leaves update we should see the candidate included.
 		let next_update = next_leaf(&mut chain);
@@ -248,11 +312,13 @@ fn scraper_provides_included_state_when_initialized() {
 			&chain,
 			finalized_block_number,
 			expected_ancestry_len,
+			get_backed_and_included_candidate_events,
 		);
 		join(process_active_leaves_update(ctx.sender(), &mut scraper, next_update), overseer_fut)
 			.await;
 
 		assert!(scraper.is_candidate_included(&candidate_2.hash()));
+		assert!(scraper.is_candidate_backed(&candidate_2.hash()));
 	});
 }
 
@@ -274,6 +340,7 @@ fn scraper_requests_candidates_of_leaf_ancestors() {
 			&chain,
 			finalized_block_number,
 			BLOCKS_TO_SKIP,
+			get_backed_and_included_candidate_events,
 		);
 		join(process_active_leaves_update(ctx.sender(), &mut scraper, next_update), overseer_fut)
 			.await;
@@ -282,6 +349,7 @@ fn scraper_requests_candidates_of_leaf_ancestors() {
 		for block_number in 1..next_block_number {
 			let candidate = make_candidate_receipt(get_block_number_hash(block_number));
 			assert!(scraper.is_candidate_included(&candidate.hash()));
+			assert!(scraper.is_candidate_backed(&candidate.hash()));
 		}
 	});
 }
@@ -304,6 +372,7 @@ fn scraper_requests_candidates_of_non_cached_ancestors() {
 			&chain,
 			finalized_block_number,
 			BLOCKS_TO_SKIP[0],
+			get_backed_and_included_candidate_events,
 		);
 		join(process_active_leaves_update(ctx.sender(), &mut ordering, next_update), overseer_fut)
 			.await;
@@ -315,6 +384,7 @@ fn scraper_requests_candidates_of_non_cached_ancestors() {
 			&chain,
 			finalized_block_number,
 			BLOCKS_TO_SKIP[1],
+			get_backed_and_included_candidate_events,
 		);
 		join(process_active_leaves_update(ctx.sender(), &mut ordering, next_update), overseer_fut)
 			.await;
@@ -340,8 +410,170 @@ fn scraper_requests_candidates_of_non_finalized_ancestors() {
 			&chain,
 			finalized_block_number,
 			BLOCKS_TO_SKIP - finalized_block_number as usize, // Expect the provider not to go past finalized block.
+			get_backed_and_included_candidate_events,
 		);
 		join(process_active_leaves_update(ctx.sender(), &mut ordering, next_update), overseer_fut)
 			.await;
 	});
 }
+
+#[test]
+fn scraper_prunes_finalized_candidates() {
+	const TEST_TARGET_BLOCK_NUMBER: BlockNumber = 2;
+
+	// How many blocks should we skip before sending a leaf update.
+	const BLOCKS_TO_SKIP: usize = 3;
+
+	futures::executor::block_on(async {
+		let (state, mut virtual_overseer) = TestState::new().await;
+
+		let TestState { mut chain, mut scraper, mut ctx } = state;
+
+		// 1 because `TestState` starts at leaf 1.
+		let next_update = (1..BLOCKS_TO_SKIP).map(|_| next_leaf(&mut chain)).last().unwrap();
+
+		let mut finalized_block_number = 1;
+		let expected_ancestry_len = BLOCKS_TO_SKIP - finalized_block_number as usize;
+		let overseer_fut = overseer_process_active_leaves_update(
+			&mut virtual_overseer,
+			&chain,
+			finalized_block_number,
+			expected_ancestry_len,
+			|block_num| {
+				if block_num == TEST_TARGET_BLOCK_NUMBER {
+					get_backed_and_included_candidate_events(block_num)
+				} else {
+					vec![]
+				}
+			},
+		);
+		join(process_active_leaves_update(ctx.sender(), &mut scraper, next_update), overseer_fut)
+			.await;
+
+		let candidate = make_candidate_receipt(get_block_number_hash(TEST_TARGET_BLOCK_NUMBER));
+
+		// After `CANDIDATE_LIFETIME_AFTER_FINALIZATION` blocks the candidate should be removed
+		finalized_block_number =
+			TEST_TARGET_BLOCK_NUMBER + ChainScraper::CANDIDATE_LIFETIME_AFTER_FINALIZATION;
+		process_finalized_block(&mut scraper, &finalized_block_number);
+
+		assert!(!scraper.is_candidate_backed(&candidate.hash()));
+		assert!(!scraper.is_candidate_included(&candidate.hash()));
+	});
+}
+
+#[test]
+fn scraper_handles_backed_but_not_included_candidate() {
+	const TEST_TARGET_BLOCK_NUMBER: BlockNumber = 2;
+
+	// How many blocks should we skip before sending a leaf update.
+	const BLOCKS_TO_SKIP: usize = 3;
+
+	futures::executor::block_on(async {
+		let (state, mut virtual_overseer) = TestState::new().await;
+
+		let TestState { mut chain, mut scraper, mut ctx } = state;
+
+		let next_update = (1..BLOCKS_TO_SKIP as BlockNumber)
+			.map(|_| next_leaf(&mut chain))
+			.last()
+			.unwrap();
+
+		// Add `ActiveLeavesUpdate` containing `CandidateBacked` event for block `BLOCK_WITH_EVENTS`
+		let mut finalized_block_number = 1;
+		let expected_ancestry_len = BLOCKS_TO_SKIP - finalized_block_number as usize;
+		let overseer_fut = overseer_process_active_leaves_update(
+			&mut virtual_overseer,
+			&chain,
+			finalized_block_number,
+			expected_ancestry_len,
+			|block_num| {
+				if block_num == TEST_TARGET_BLOCK_NUMBER {
+					get_backed_candidate_event(block_num)
+				} else {
+					vec![]
+				}
+			},
+		);
+		join(process_active_leaves_update(ctx.sender(), &mut scraper, next_update), overseer_fut)
+			.await;
+
+		// Finalize blocks to enforce pruning of scraped events
+		finalized_block_number += 1;
+		process_finalized_block(&mut scraper, &finalized_block_number);
+
+		// `FIRST_TEST_BLOCK` is finalized, which is within `BACKED_CANDIDATE_LIFETIME_AFTER_FINALIZATION` window.
+		// The candidate should still be backed.
+		let candidate = make_candidate_receipt(get_block_number_hash(TEST_TARGET_BLOCK_NUMBER));
+		assert!(!scraper.is_candidate_included(&candidate.hash()));
+		assert!(scraper.is_candidate_backed(&candidate.hash()));
+
+		// Bump the finalized block outside `BACKED_CANDIDATE_LIFETIME_AFTER_FINALIZATION`.
+		// The candidate should be removed.
+		assert!(
+			finalized_block_number <
+				TEST_TARGET_BLOCK_NUMBER + ChainScraper::CANDIDATE_LIFETIME_AFTER_FINALIZATION
+		);
+		finalized_block_number +=
+			TEST_TARGET_BLOCK_NUMBER + ChainScraper::CANDIDATE_LIFETIME_AFTER_FINALIZATION;
+		process_finalized_block(&mut scraper, &finalized_block_number);
+
+		assert!(!scraper.is_candidate_included(&candidate.hash()));
+		assert!(!scraper.is_candidate_backed(&candidate.hash()));
+	});
+}
+
+#[test]
+fn scraper_handles_the_same_candidate_incuded_in_two_different_block_heights() {
+	// Same candidate will be inclued in these two leaves
+	let test_targets = vec![2, 3];
+
+	// How many blocks should we skip before sending a leaf update.
+	const BLOCKS_TO_SKIP: usize = 3;
+
+	futures::executor::block_on(async {
+		let (state, mut virtual_overseer) = TestState::new().await;
+
+		let TestState { mut chain, mut scraper, mut ctx } = state;
+
+		// 1 because `TestState` starts at leaf 1.
+		let next_update = (1..BLOCKS_TO_SKIP).map(|_| next_leaf(&mut chain)).last().unwrap();
+
+		// Now we will add the same magic candidate at two different block heights.
+		// Check `get_backed_and_included_magic_candidate_event` implementation
+		let mut finalized_block_number = 1;
+		let expected_ancestry_len = BLOCKS_TO_SKIP - finalized_block_number as usize;
+		let overseer_fut = overseer_process_active_leaves_update(
+			&mut virtual_overseer,
+			&chain,
+			finalized_block_number,
+			expected_ancestry_len,
+			|block_num| {
+				if test_targets.contains(&block_num) {
+					get_backed_and_included_magic_candidate_events(block_num)
+				} else {
+					vec![]
+				}
+			},
+		);
+		join(process_active_leaves_update(ctx.sender(), &mut scraper, next_update), overseer_fut)
+			.await;
+
+		// Finalize blocks to enforce pruning of scraped events.
+		// The magic candidate was added twice, so it shouldn't be removed if we finalize two more blocks.
+		finalized_block_number = test_targets.first().expect("there are two block nums") +
+			ChainScraper::CANDIDATE_LIFETIME_AFTER_FINALIZATION;
+		process_finalized_block(&mut scraper, &finalized_block_number);
+
+		let magic_candidate = make_candidate_receipt(get_magic_candidate_hash());
+		assert!(scraper.is_candidate_backed(&magic_candidate.hash()));
+		assert!(scraper.is_candidate_included(&magic_candidate.hash()));
+
+		// On the next finalization the magic candidate should be removed
+		finalized_block_number += 1;
+		process_finalized_block(&mut scraper, &finalized_block_number);
+
+		assert!(!scraper.is_candidate_backed(&magic_candidate.hash()));
+		assert!(!scraper.is_candidate_included(&magic_candidate.hash()));
+	});
+}

polkadot/node/core/dispute-coordinator/src/tests.rs

@@ -57,10 +57,10 @@ use polkadot_node_subsystem::{
 };
 use polkadot_node_subsystem_test_helpers::{make_subsystem_context, TestSubsystemContextHandle};
 use polkadot_primitives::v2::{
-	ApprovalVote, BlockNumber, CandidateCommitments, CandidateHash, CandidateReceipt,
-	DisputeStatement, GroupIndex, Hash, Header, IndexedVec, MultiDisputeStatementSet,
-	ScrapedOnChainVotes, SessionIndex, SessionInfo, SigningContext, ValidDisputeStatementKind,
-	ValidatorId, ValidatorIndex, ValidatorSignature,
+	ApprovalVote, BlockNumber, CandidateCommitments, CandidateEvent, CandidateHash,
+	CandidateReceipt, CoreIndex, DisputeStatement, GroupIndex, Hash, HeadData, Header, IndexedVec,
+	MultiDisputeStatementSet, ScrapedOnChainVotes, SessionIndex, SessionInfo, SigningContext,
+	ValidDisputeStatementKind, ValidatorId, ValidatorIndex, ValidatorSignature,
 };
 
 use crate::{
@@ -212,6 +212,7 @@ impl TestState {
 		virtual_overseer: &mut VirtualOverseer,
 		session: SessionIndex,
 		block_number: BlockNumber,
+		candidate_events: Vec<CandidateEvent>,
 	) {
 		assert!(block_number > 0);
 
@@ -239,8 +240,14 @@ impl TestState {
 			)))
 			.await;
 
-		self.handle_sync_queries(virtual_overseer, block_hash, block_number, session)
-			.await;
+		self.handle_sync_queries(
+			virtual_overseer,
+			block_hash,
+			block_number,
+			session,
+			candidate_events,
+		)
+		.await;
 	}
 
 	async fn handle_sync_queries(
@@ -249,6 +256,7 @@ impl TestState {
 		block_hash: Hash,
 		block_number: BlockNumber,
 		session: SessionIndex,
+		candidate_events: Vec<CandidateEvent>,
 	) {
 		// Order of messages is not fixed (different on initializing):
 		#[derive(Debug)]
@@ -352,7 +360,7 @@ impl TestState {
 							_new_leaf,
 							RuntimeApiRequest::CandidateEvents(tx),
 							)) => {
-						tx.send(Ok(Vec::new())).unwrap();
+						tx.send(Ok(candidate_events.clone())).unwrap();
 					}
 					);
 					gum::trace!("After answering runtime api request");
@@ -401,8 +409,14 @@ impl TestState {
 				)))
 				.await;
 
-			self.handle_sync_queries(virtual_overseer, *leaf, n as BlockNumber, session)
-				.await;
+			self.handle_sync_queries(
+				virtual_overseer,
+				*leaf,
+				n as BlockNumber,
+				session,
+				Vec::new(),
+			)
+			.await;
 		}
 	}
 
@@ -556,6 +570,26 @@ fn make_invalid_candidate_receipt() -> CandidateReceipt {
 	dummy_candidate_receipt_bad_sig(Default::default(), Some(Default::default()))
 }
 
+// Generate a `CandidateBacked` event from a `CandidateReceipt`. The rest is dummy data.
+fn make_candidate_backed_event(candidate_receipt: CandidateReceipt) -> CandidateEvent {
+	CandidateEvent::CandidateBacked(
+		candidate_receipt,
+		HeadData(Vec::new()),
+		CoreIndex(0),
+		GroupIndex(0),
+	)
+}
+
+// Generate a `CandidateIncluded` event from a `CandidateReceipt`. The rest is dummy data.
+fn make_candidate_included_event(candidate_receipt: CandidateReceipt) -> CandidateEvent {
+	CandidateEvent::CandidateIncluded(
+		candidate_receipt,
+		HeadData(Vec::new()),
+		CoreIndex(0),
+		GroupIndex(0),
+	)
+}
+
 /// Handle request for approval votes:
 pub async fn handle_approval_vote_request(
 	ctx_handle: &mut VirtualOverseer,
@@ -587,7 +621,9 @@ fn too_many_unconfirmed_statements_are_considered_spam() {
 			let candidate_receipt2 = make_invalid_candidate_receipt();
 			let candidate_hash2 = candidate_receipt2.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote1 = test_state
 				.issue_backing_statement_with_index(ValidatorIndex(3), candidate_hash1, session)
@@ -634,8 +670,9 @@ fn too_many_unconfirmed_statements_are_considered_spam() {
 			handle_approval_vote_request(&mut virtual_overseer, &candidate_hash1, HashMap::new())
 				.await;
 
-			// Participation has to fail, otherwise the dispute will be confirmed.
-			participation_missing_availability(&mut virtual_overseer).await;
+			// Participation has to fail here, otherwise the dispute will be confirmed. However
+			// participation won't happen at all because the dispute is neither backed, not confirmed
+			// nor the candidate is included. Or in other words - we'll refrain from participation.
 
 			{
 				let (tx, rx) = oneshot::channel();
@@ -718,7 +755,9 @@ fn approval_vote_import_works() {
 			let candidate_receipt1 = make_valid_candidate_receipt();
 			let candidate_hash1 = candidate_receipt1.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote1 = test_state
 				.issue_backing_statement_with_index(ValidatorIndex(3), candidate_hash1, session)
@@ -762,8 +801,8 @@ fn approval_vote_import_works() {
 			handle_approval_vote_request(&mut virtual_overseer, &candidate_hash1, approval_votes)
 				.await;
 
-			// Participation has to fail, otherwise the dispute will be confirmed.
-			participation_missing_availability(&mut virtual_overseer).await;
+			// Participation won't happen here because the dispute is neither backed, not confirmed
+			// nor the candidate is included. Or in other words - we'll refrain from participation.
 
 			{
 				let (tx, rx) = oneshot::channel();
@@ -814,7 +853,17 @@ fn dispute_gets_confirmed_via_participation() {
 			let candidate_receipt2 = make_invalid_candidate_receipt();
 			let candidate_hash2 = candidate_receipt2.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![
+						make_candidate_backed_event(candidate_receipt1.clone()),
+						make_candidate_backed_event(candidate_receipt2.clone()),
+					],
+				)
+				.await;
 
 			let valid_vote1 = test_state
 				.issue_explicit_statement_with_index(
@@ -963,7 +1012,9 @@ fn dispute_gets_confirmed_at_byzantine_threshold() {
 			let candidate_receipt2 = make_invalid_candidate_receipt();
 			let candidate_hash2 = candidate_receipt2.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote1 = test_state
 				.issue_explicit_statement_with_index(
@@ -1037,7 +1088,8 @@ fn dispute_gets_confirmed_at_byzantine_threshold() {
 			handle_approval_vote_request(&mut virtual_overseer, &candidate_hash1, HashMap::new())
 				.await;
 
-			participation_missing_availability(&mut virtual_overseer).await;
+			// Participation won't happen here because the dispute is neither backed, not confirmed
+			// nor the candidate is included. Or in other words - we'll refrain from participation.
 
 			{
 				let (tx, rx) = oneshot::channel();
@@ -1124,7 +1176,9 @@ fn backing_statements_import_works_and_no_spam() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote1 = test_state
 				.issue_backing_statement_with_index(ValidatorIndex(3), candidate_hash, session)
@@ -1187,6 +1241,15 @@ fn backing_statements_import_works_and_no_spam() {
 				.issue_backing_statement_with_index(ValidatorIndex(4), candidate_hash, session)
 				.await;
 
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
+
 			let (pending_confirmation, confirmation_rx) = oneshot::channel();
 			// Backing vote import should not have accounted to spam slots, so this should succeed
 			// as well:
@@ -1228,7 +1291,14 @@ fn conflicting_votes_lead_to_dispute_participation() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -1353,7 +1423,14 @@ fn positive_votes_dont_trigger_participation() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -1466,7 +1543,9 @@ fn wrong_validator_index_is_ignored() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -1544,7 +1623,14 @@ fn finality_votes_ignore_disputed_candidates() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -1654,7 +1740,14 @@ fn supermajority_valid_dispute_may_be_finalized() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let supermajority_threshold =
 				polkadot_primitives::v2::supermajority_threshold(test_state.validators.len());
@@ -1794,7 +1887,14 @@ fn concluded_supermajority_for_non_active_after_time() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let supermajority_threshold =
 				polkadot_primitives::v2::supermajority_threshold(test_state.validators.len());
@@ -1912,7 +2012,14 @@ fn concluded_supermajority_against_non_active_after_time() {
 
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let supermajority_threshold =
 				polkadot_primitives::v2::supermajority_threshold(test_state.validators.len());
@@ -2036,7 +2143,9 @@ fn resume_dispute_without_local_statement() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -2074,7 +2183,8 @@ fn resume_dispute_without_local_statement() {
 				.await;
 
 			// Missing availability -> No local vote.
-			participation_missing_availability(&mut virtual_overseer).await;
+			// Participation won't happen here because the dispute is neither backed, not confirmed
+			// nor the candidate is included. Or in other words - we'll refrain from participation.
 
 			assert_eq!(confirmation_rx.await, Ok(ImportStatementsResult::ValidImport));
 
@@ -2216,7 +2326,14 @@ fn resume_dispute_with_local_statement() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
 
 			let local_valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -2315,7 +2432,9 @@ fn resume_dispute_without_local_statement_or_local_key() {
 				let candidate_receipt = make_valid_candidate_receipt();
 				let candidate_hash = candidate_receipt.hash();
 
-				test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+				test_state
+					.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+					.await;
 
 				let valid_vote = test_state
 					.issue_explicit_statement_with_index(
@@ -2408,7 +2527,9 @@ fn resume_dispute_with_local_statement_without_local_key() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let local_valid_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -2517,7 +2638,9 @@ fn issue_local_statement_does_cause_distribution_but_not_duplicate_participation
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let other_vote = test_state
 				.issue_explicit_statement_with_index(
@@ -2590,7 +2713,9 @@ fn own_approval_vote_gets_distributed_on_dispute() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let statement = test_state.issue_approval_vote_with_index(
 				ValidatorIndex(0),
@@ -2681,7 +2806,9 @@ fn negative_issue_local_statement_only_triggers_import() {
 			let candidate_receipt = make_invalid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			virtual_overseer
 				.send(FromOrchestra::Communication {
@@ -2729,7 +2856,9 @@ fn redundant_votes_ignored() {
 			let candidate_receipt = make_valid_candidate_receipt();
 			let candidate_hash = candidate_receipt.hash();
 
-			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
 
 			let valid_vote = test_state
 				.issue_backing_statement_with_index(ValidatorIndex(1), candidate_hash, session)
@@ -2787,3 +2916,195 @@ fn redundant_votes_ignored() {
 		})
 	});
 }
+
+#[test]
+fn refrain_from_participation() {
+	test_harness(|mut test_state, mut virtual_overseer| {
+		Box::pin(async move {
+			let session = 1;
+
+			test_state.handle_resume_sync(&mut virtual_overseer, session).await;
+
+			let candidate_receipt = make_valid_candidate_receipt();
+			let candidate_hash = candidate_receipt.hash();
+
+			// activate leaf - no backing/included event
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
+
+			// generate two votes
+			let valid_vote = test_state
+				.issue_explicit_statement_with_index(
+					ValidatorIndex(1),
+					candidate_hash,
+					session,
+					true,
+				)
+				.await;
+
+			let invalid_vote = test_state
+				.issue_explicit_statement_with_index(
+					ValidatorIndex(2),
+					candidate_hash,
+					session,
+					false,
+				)
+				.await;
+
+			virtual_overseer
+				.send(FromOrchestra::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: vec![
+							(valid_vote, ValidatorIndex(1)),
+							(invalid_vote, ValidatorIndex(2)),
+						],
+						pending_confirmation: None,
+					},
+				})
+				.await;
+
+			handle_approval_vote_request(&mut virtual_overseer, &candidate_hash, HashMap::new())
+				.await;
+
+			{
+				let (tx, rx) = oneshot::channel();
+				virtual_overseer
+					.send(FromOrchestra::Communication {
+						msg: DisputeCoordinatorMessage::ActiveDisputes(tx),
+					})
+					.await;
+
+				assert_eq!(rx.await.unwrap().len(), 1);
+
+				let (tx, rx) = oneshot::channel();
+				virtual_overseer
+					.send(FromOrchestra::Communication {
+						msg: DisputeCoordinatorMessage::QueryCandidateVotes(
+							vec![(session, candidate_hash)],
+							tx,
+						),
+					})
+					.await;
+
+				let (_, _, votes) = rx.await.unwrap().get(0).unwrap().clone();
+				assert_eq!(votes.valid.len(), 1);
+				assert_eq!(votes.invalid.len(), 1);
+			}
+
+			// activate leaf - no backing event
+			test_state
+				.activate_leaf_at_session(&mut virtual_overseer, session, 1, Vec::new())
+				.await;
+
+			virtual_overseer.send(FromOrchestra::Signal(OverseerSignal::Conclude)).await;
+
+			// confirm that no participation request is made.
+			assert!(virtual_overseer.try_recv().await.is_none());
+
+			test_state
+		})
+	});
+}
+
+/// We have got no `participation_for_backed_candidates` test because most of the other tests (e.g.
+/// `dispute_gets_confirmed_via_participation`, `backing_statements_import_works_and_no_spam`) use
+/// candidate backing event to trigger participation. If they pass - that case works.
+#[test]
+fn participation_for_included_candidates() {
+	test_harness(|mut test_state, mut virtual_overseer| {
+		Box::pin(async move {
+			let session = 1;
+
+			test_state.handle_resume_sync(&mut virtual_overseer, session).await;
+
+			let candidate_receipt = make_valid_candidate_receipt();
+			let candidate_hash = candidate_receipt.hash();
+
+			// activate leaf - with candidate included event
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_included_event(candidate_receipt.clone())],
+				)
+				.await;
+
+			// generate two votes
+			let valid_vote = test_state
+				.issue_explicit_statement_with_index(
+					ValidatorIndex(1),
+					candidate_hash,
+					session,
+					true,
+				)
+				.await;
+
+			let invalid_vote = test_state
+				.issue_explicit_statement_with_index(
+					ValidatorIndex(2),
+					candidate_hash,
+					session,
+					false,
+				)
+				.await;
+
+			virtual_overseer
+				.send(FromOrchestra::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: vec![
+							(valid_vote, ValidatorIndex(1)),
+							(invalid_vote, ValidatorIndex(2)),
+						],
+						pending_confirmation: None,
+					},
+				})
+				.await;
+
+			handle_approval_vote_request(&mut virtual_overseer, &candidate_hash, HashMap::new())
+				.await;
+
+			participation_with_distribution(
+				&mut virtual_overseer,
+				&candidate_hash,
+				candidate_receipt.commitments_hash,
+			)
+			.await;
+
+			{
+				let (tx, rx) = oneshot::channel();
+				virtual_overseer
+					.send(FromOrchestra::Communication {
+						msg: DisputeCoordinatorMessage::ActiveDisputes(tx),
+					})
+					.await;
+
+				assert_eq!(rx.await.unwrap().len(), 1);
+
+				// check if we have participated (casted a vote)
+				let (tx, rx) = oneshot::channel();
+				virtual_overseer
+					.send(FromOrchestra::Communication {
+						msg: DisputeCoordinatorMessage::QueryCandidateVotes(
+							vec![(session, candidate_hash)],
+							tx,
+						),
+					})
+					.await;
+
+				let (_, _, votes) = rx.await.unwrap().get(0).unwrap().clone();
+				assert_eq!(votes.valid.len(), 2); // 2 => we have participated
+				assert_eq!(votes.invalid.len(), 1);
+			}
+
+			virtual_overseer.send(FromOrchestra::Signal(OverseerSignal::Conclude)).await;
+
+			test_state
+		})
+	});
+}