Update dispute participation on active leaves update

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

@@ -269,8 +269,9 @@ impl Initialized {
 		update: ActiveLeavesUpdate,
 		now: u64,
 	) -> Result<()> {
-		let on_chain_votes =
+		let (on_chain_votes, included_receipts) =
 			self.scraper.process_active_leaves_update(ctx.sender(), &update).await?;
+		self.participation.prioritize_newly_included(ctx, &included_receipts).await;
 		self.participation.process_active_leaves_update(ctx, &update).await?;
 
 		if let Some(new_leaf) = update.activated {
@@ -416,6 +417,8 @@ impl Initialized {
 				})
 				.collect();
 
+			// Importantly, handling import statements for backing votes also 
+			// clears spam slots for any newly backed candidates
 			let import_result = self
 				.handle_import_statements(
 					ctx,
@@ -828,8 +831,15 @@ impl Initialized {
 		let new_state = import_result.new_state();
 
 		let is_included = self.scraper.is_candidate_included(&candidate_hash);
-
-		let potential_spam = !is_included && !new_state.is_confirmed() && !new_state.has_own_vote();
+		let is_backed = self.scraper.is_candidate_backed(&candidate_hash);
+		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_confirmed = new_state.is_confirmed();
+		let potential_spam = !is_included && !is_backed 
+			&& !new_state.is_confirmed() && !new_state.has_own_vote();
+		// We participate only in disputes which are included, backed or confirmed
+		let allow_participation = is_included || is_backed || is_confirmed;
 
 		gum::trace!(
 			target: LOG_TARGET,
@@ -842,8 +852,11 @@ impl Initialized {
 			"Is spam?"
 		);
 
+		// This check is responsible for all clearing of spam slots. It runs 
+		// whenever a vote is imported from on or off chain, and decrements 
+		// slots whenever a candidate is newly backed, confirmed, or has our
+		// own vote.
 		if !potential_spam {
-			// Former spammers have not been spammers after all:
 			self.spam_slots.clear(&(session, candidate_hash));
 
 		// Potential spam:
@@ -871,14 +884,6 @@ 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. Disputes should fall in one of the categories below, otherwise we will refrain
 		// from participation:

node/core/dispute-coordinator/src/participation/mod.rs

@@ -212,6 +212,19 @@ impl Participation {
 		Ok(())
 	}
 
+	/// Reprioritizes participation requests for disputes that are freshly included
+	pub async fn prioritize_newly_included<Context>(&mut self, ctx: &mut Context, included_receipts: &Vec<CandidateReceipt>) {
+		for receipt in included_receipts {
+			let r = self.queue.prioritize_if_present(ctx.sender(), receipt).await;
+			if let Err(queue_error) = r {
+				match queue_error {
+					QueueError::PriorityFull => return, // Avoid working through the rest of the vec
+					_ => (),
+				}
+			}
+		}
+	}
+
 	/// Dequeue until `MAX_PARALLEL_PARTICIPATIONS` is reached.
 	async fn dequeue_until_capacity<Context>(
 		&mut self,

node/core/dispute-coordinator/src/participation/queues/mod.rs

@@ -67,7 +67,7 @@ pub struct ParticipationRequest {
 }
 
 /// Whether a `ParticipationRequest` should be put on best-effort or the priority queue.
-#[derive(Debug)]
+#[derive(Debug, PartialEq)]
 pub enum ParticipationPriority {
 	BestEffort,
 	Priority,
@@ -103,6 +103,8 @@ pub enum QueueError {
 	BestEffortFull,
 	#[error("Request could not be queued, because priority queue was already full.")]
 	PriorityFull,
+	#[error("A comparator could not be generated for the given request.")]
+	CouldNotGenerateComparator,
 }
 
 impl ParticipationRequest {
@@ -159,6 +161,24 @@ impl Queues {
 		self.pop_best_effort().map(|d| d.1)
 	}
 
+	/// Reprioritizes any participation requests pertaining to the
+	/// passed candidates from best effort to priority.
+	pub async fn prioritize_if_present(
+		&mut self, 
+		sender: &mut impl overseer::DisputeCoordinatorSenderTrait,
+		receipt: &CandidateReceipt,
+	) -> std::result::Result<(), QueueError>{
+		if self.priority.len() >= PRIORITY_QUEUE_SIZE {
+			return Err(QueueError::PriorityFull)
+		}
+
+		let comparator = CandidateComparator::new(sender, receipt).await.map_err(|_e| QueueError::CouldNotGenerateComparator)?;
+		if let Some(request) = self.best_effort.remove(&comparator){
+			self.priority.insert(comparator, request);
+		}
+		Ok(())
+	}
+
 	fn queue_with_comparator(
 		&mut self,
 		comparator: CandidateComparator,

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

@@ -26,7 +26,7 @@ use polkadot_node_subsystem::{
 };
 use polkadot_node_subsystem_util::runtime::{get_candidate_events, get_on_chain_votes};
 use polkadot_primitives::v2::{
-	BlockNumber, CandidateEvent, CandidateHash, Hash, ScrapedOnChainVotes,
+	BlockNumber, CandidateEvent, CandidateHash, Hash, ScrapedOnChainVotes, CandidateReceipt,
 };
 
 use crate::{
@@ -114,7 +114,7 @@ impl ChainScraper {
 		};
 		let update =
 			ActiveLeavesUpdate { activated: Some(initial_head), deactivated: Default::default() };
-		let votes = s.process_active_leaves_update(sender, &update).await?;
+		let (votes, _) = s.process_active_leaves_update(sender, &update).await?;
 		Ok((s, votes))
 	}
 
@@ -137,13 +137,13 @@ impl ChainScraper {
 		&mut self,
 		sender: &mut Sender,
 		update: &ActiveLeavesUpdate,
-	) -> Result<Vec<ScrapedOnChainVotes>>
+	) -> Result<(Vec<ScrapedOnChainVotes>, Vec<CandidateReceipt>)>
 	where
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
 		let activated = match update.activated.as_ref() {
 			Some(activated) => activated,
-			None => return Ok(Vec::new()),
+			None => return Ok((Vec::new(), Vec::new())),
 		};
 
 		// Fetch ancestry up to last finalized block.
@@ -157,11 +157,13 @@ impl ChainScraper {
 
 		let block_hashes = std::iter::once(activated.hash).chain(ancestors);
 
+		let mut included_receipts: Vec<CandidateReceipt> = Vec::new();
 		let mut on_chain_votes = Vec::new();
 		for (block_number, block_hash) in block_numbers.zip(block_hashes) {
 			gum::trace!(?block_number, ?block_hash, "In ancestor processing.");
 
-			self.process_candidate_events(sender, block_number, block_hash).await?;
+			let events_for_block = self.process_candidate_events(sender, block_number, block_hash).await?;
+			included_receipts.extend(events_for_block);
 
 			if let Some(votes) = get_on_chain_votes(sender, block_hash).await? {
 				on_chain_votes.push(votes);
@@ -170,7 +172,7 @@ impl ChainScraper {
 
 		self.last_observed_blocks.put(activated.hash, ());
 
-		Ok(on_chain_votes)
+		Ok((on_chain_votes, included_receipts))
 	}
 
 	/// Prune finalized candidates.
@@ -201,12 +203,14 @@ impl ChainScraper {
 		sender: &mut Sender,
 		block_number: BlockNumber,
 		block_hash: Hash,
-	) -> Result<()>
+	) -> Result<Vec<CandidateReceipt>>
 	where
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
+		let events = get_candidate_events(sender, block_hash).await?;
+		let mut included_receipts: Vec<CandidateReceipt> = Vec::new();
 		// Get included and backed events:
-		for ev in get_candidate_events(sender, block_hash).await? {
+		for ev in events {
 			match ev {
 				CandidateEvent::CandidateIncluded(receipt, _, _, _) => {
 					let candidate_hash = receipt.hash();
@@ -217,6 +221,7 @@ impl ChainScraper {
 						"Processing included event"
 					);
 					self.included_candidates.insert(block_number, candidate_hash);
+					included_receipts.push(receipt);
 				},
 				CandidateEvent::CandidateBacked(receipt, _, _, _) => {
 					let candidate_hash = receipt.hash();
@@ -233,7 +238,7 @@ impl ChainScraper {
 				},
 			}
 		}
-		Ok(())
+		Ok(included_receipts)
 	}
 
 	/// Returns ancestors of `head` in the descending order, stopping

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

@@ -1287,7 +1287,7 @@ fn backing_statements_import_works_and_no_spam() {
 				})
 				.await;
 
-			// Result should be valid, because our node participated, so spam slots are cleared:
+			// Import should be valid, as spam slots were not filled
 			assert_matches!(confirmation_rx.await, Ok(ImportStatementsResult::ValidImport));
 
 			virtual_overseer.send(FromOrchestra::Signal(OverseerSignal::Conclude)).await;
@@ -3193,3 +3193,134 @@ fn participation_for_included_candidates() {
 		})
 	});
 }
+
+/// Shows that importing backing votes when a backing event is being processed
+/// results in participation.
+#[test]
+fn local_participation_in_dispute_for_backed_candidate() {
+	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();
+
+			// Step 1: Show that we don't participate when not backed, confirmed, or included
+
+			// activate leaf - without candidate backed event
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![],
+				)
+				.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;
+
+			assert_matches!(virtual_overseer.recv().timeout(TEST_TIMEOUT).await, None);
+
+			// Step 2: Show that once backing votes are processed we participate
+
+			// Activate leaf: With candidate backed event
+			test_state
+				.activate_leaf_at_session(
+					&mut virtual_overseer,
+					session,
+					1,
+					vec![make_candidate_backed_event(candidate_receipt.clone())],
+				)
+				.await;
+
+			let backing_valid = test_state
+				.issue_backing_statement_with_index(ValidatorIndex(3), candidate_hash, session).await;
+
+			virtual_overseer
+				.send(FromOrchestra::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: vec![
+							(backing_valid, ValidatorIndex(3))
+						],
+						pending_confirmation: None,
+					},
+				})
+				.await;
+
+			participation_with_distribution(
+				&mut virtual_overseer,
+				&candidate_hash,
+				candidate_receipt.commitments_hash,
+			)
+			.await;
+
+			// Check for our 1 active dispute
+			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(), 3); // 3 => 1 initial vote, 1 backing vote, and our vote
+			assert_eq!(votes.invalid.len(), 1);
+
+			// Wrap up
+			virtual_overseer.send(FromOrchestra::Signal(OverseerSignal::Conclude)).await;
+
+			test_state
+		})
+	});
+}