Passed candidate events from scraper to participation

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

@@ -190,7 +190,7 @@ impl Initialized {
 		if let Some(first_leaf) = first_leaf.take() {
 			// Also provide first leaf to participation for good measure.
 			self.participation
-				.process_active_leaves_update(ctx, &ActiveLeavesUpdate::start_work(first_leaf))
+				.process_active_leaves_update(ctx, &ActiveLeavesUpdate::start_work(first_leaf), Vec::new())
 				.await?;
 		}
 
@@ -269,9 +269,9 @@ impl Initialized {
 		update: ActiveLeavesUpdate,
 		now: u64,
 	) -> Result<()> {
-		let on_chain_votes =
+		let (on_chain_votes, candidate_events) =
 			self.scraper.process_active_leaves_update(ctx.sender(), &update).await?;
-		self.participation.process_active_leaves_update(ctx, &update).await?;
+		self.participation.process_active_leaves_update(ctx, &update, candidate_events).await?;
 
 		if let Some(new_leaf) = update.activated {
 			match self

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

@@ -31,7 +31,7 @@ use polkadot_node_subsystem::{
 	overseer, ActiveLeavesUpdate, RecoveryError,
 };
 use polkadot_node_subsystem_util::runtime::get_validation_code_by_hash;
-use polkadot_primitives::v2::{BlockNumber, CandidateHash, CandidateReceipt, Hash, SessionIndex};
+use polkadot_primitives::v2::{BlockNumber, CandidateHash, CandidateReceipt, Hash, SessionIndex, CandidateEvent};
 
 use crate::LOG_TARGET;
 
@@ -195,6 +195,7 @@ impl Participation {
 		&mut self,
 		ctx: &mut Context,
 		update: &ActiveLeavesUpdate,
+		candidate_events: Vec<(Vec<CandidateEvent>, BlockNumber)>,
 	) -> FatalResult<()> {
 		if let Some(activated) = &update.activated {
 			match self.recent_block {
@@ -208,6 +209,8 @@ impl Participation {
 				},
 				Some(_) => {},
 			}
+
+			self.handle_candidate_events(candidate_events);
 		}
 		Ok(())
 	}
@@ -245,6 +248,26 @@ impl Participation {
 		}
 		Ok(())
 	}
+
+	/// Update priority queue and best effort queue to account for
+	/// freshly backed or included disputed candidates
+	fn handle_candidate_events(&mut self, candidate_events : Vec<(Vec<CandidateEvent>, BlockNumber)>) {
+		for (events_for_block, block_number) in candidate_events {
+			for event in events_for_block {
+				match event {
+					CandidateEvent::CandidateIncluded(receipt, _, _, _) => {
+						let candidate_hash = receipt.hash();
+					},
+					CandidateEvent::CandidateBacked(receipt, _, _, _) => {
+						let candidate_hash = receipt.hash();
+					},
+					_ => {
+						// skip the rest
+					},
+				}
+			}
+		}
+	}
 }
 
 async fn participate(

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

@@ -101,6 +101,7 @@ async fn activate_leaf<Context>(
 				number: block_number,
 				status: LeafStatus::Fresh,
 			}),
+			Vec::new(),
 		)
 		.await
 }

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

@@ -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<(Vec<CandidateEvent>, BlockNumber)>)>
 	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 candidate_events: Vec<(Vec<CandidateEvent>, BlockNumber)> = 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?;
+			candidate_events.push((events_for_block, block_number));
 
 			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, candidate_events))
 	}
 
 	/// Prune finalized candidates.
@@ -201,12 +203,13 @@ impl ChainScraper {
 		sender: &mut Sender,
 		block_number: BlockNumber,
 		block_hash: Hash,
-	) -> Result<()>
+	) -> Result<Vec<CandidateEvent>>
 	where
 		Sender: overseer::DisputeCoordinatorSenderTrait,
 	{
+		let events = get_candidate_events(sender, block_hash).await?;
 		// 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();
@@ -233,7 +236,7 @@ impl ChainScraper {
 				},
 			}
 		}
-		Ok(())
+		Ok(events)
 	}
 
 	/// Returns ancestors of `head` in the descending order, stopping