dispute-coordinator: past session dispute slashing

.gitlab-ci.yml

@@ -38,7 +38,7 @@ variables:
   BUILDAH_COMMAND: "buildah --storage-driver overlay2"
   DOCKER_OS: "debian:stretch"
   ARCH: "x86_64"
-  ZOMBIENET_IMAGE: "docker.io/paritytech/zombienet:v1.3.50"
+  ZOMBIENET_IMAGE: "docker.io/paritytech/zombienet:v1.3.55"
 
 default:
   cache: {}

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

@@ -39,9 +39,11 @@ use polkadot_node_subsystem::{
 	},
 	overseer, ActivatedLeaf, ActiveLeavesUpdate, FromOrchestra, OverseerSignal,
 };
-use polkadot_node_subsystem_util::runtime::RuntimeInfo;
+use polkadot_node_subsystem_util::runtime::{
+	key_ownership_proof, submit_report_dispute_lost, RuntimeInfo,
+};
 use polkadot_primitives::{
-	BlockNumber, CandidateHash, CandidateReceipt, CompactStatement, DisputeStatement,
+	vstaging, BlockNumber, CandidateHash, CandidateReceipt, CompactStatement, DisputeStatement,
 	DisputeStatementSet, Hash, ScrapedOnChainVotes, SessionIndex, ValidDisputeStatementKind,
 	ValidatorId, ValidatorIndex,
 };
@@ -52,6 +54,7 @@ use crate::{
 	import::{CandidateEnvironment, CandidateVoteState},
 	is_potential_spam,
 	metrics::Metrics,
+	scraping::ScrapedUpdates,
 	status::{get_active_with_status, Clock},
 	DisputeCoordinatorSubsystem, LOG_TARGET,
 };
@@ -350,27 +353,165 @@ impl Initialized {
 				},
 			}
 
+			let ScrapedUpdates { unapplied_slashes, on_chain_votes, .. } = scraped_updates;
+
+			self.process_unapplied_slashes(ctx, new_leaf.hash, unapplied_slashes).await;
+
 			gum::trace!(
 				target: LOG_TARGET,
 				timestamp = now,
 				"Will process {} onchain votes",
-				scraped_updates.on_chain_votes.len()
+				on_chain_votes.len()
 			);
 
-			self.process_chain_import_backlog(
-				ctx,
-				overlay_db,
-				scraped_updates.on_chain_votes,
-				now,
-				new_leaf.hash,
-			)
-			.await;
+			self.process_chain_import_backlog(ctx, overlay_db, on_chain_votes, now, new_leaf.hash)
+				.await;
 		}
 
 		gum::trace!(target: LOG_TARGET, timestamp = now, "Done processing ActiveLeavesUpdate");
 		Ok(())
 	}
 
+	/// For each unapplied (past-session) slash, report an unsigned extrinsic
+	/// to the runtime.
+	async fn process_unapplied_slashes<Context>(
+		&mut self,
+		ctx: &mut Context,
+		relay_parent: Hash,
+		unapplied_slashes: Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>,
+	) {
+		for (session_index, candidate_hash, pending) in unapplied_slashes {
+			gum::info!(
+				target: LOG_TARGET,
+				?session_index,
+				?candidate_hash,
+				n_slashes = pending.keys.len(),
+				"Processing unapplied validator slashes",
+			);
+
+			let inclusions = self.scraper.get_blocks_including_candidate(&candidate_hash);
+			if inclusions.is_empty() {
+				gum::info!(
+					target: LOG_TARGET,
+					"Couldn't find inclusion parent for an unapplied slash",
+				);
+				return
+			}
+
+			// Find the first inclusion parent that we can use
+			// to generate key ownership proof on.
+			// We use inclusion parents because of the proper session index.
+			let mut key_ownership_proofs = Vec::new();
+			let mut dispute_proofs = Vec::new();
+
+			for (_height, inclusion_parent) in inclusions {
+				for (validator_index, validator_id) in pending.keys.iter() {
+					let res =
+						key_ownership_proof(ctx.sender(), inclusion_parent, validator_id.clone())
+							.await;
+
+					match res {
+						Ok(Some(key_ownership_proof)) => {
+							key_ownership_proofs.push(key_ownership_proof);
+							let time_slot = vstaging::slashing::DisputesTimeSlot::new(
+								session_index,
+								candidate_hash,
+							);
+							let dispute_proof = vstaging::slashing::DisputeProof {
+								time_slot,
+								kind: pending.kind,
+								validator_index: *validator_index,
+								validator_id: validator_id.clone(),
+							};
+							dispute_proofs.push(dispute_proof);
+						},
+						Ok(None) => {},
+						Err(error) => {
+							gum::debug!(
+								target: LOG_TARGET,
+								?error,
+								?session_index,
+								?candidate_hash,
+								?validator_id,
+								"Could not generate key ownership proof",
+							);
+						},
+					}
+				}
+
+				if !key_ownership_proofs.is_empty() {
+					debug_assert_eq!(key_ownership_proofs.len(), pending.keys.len());
+					break
+				}
+			}
+
+			let expected_keys = pending.keys.len();
+			let resolved_keys = key_ownership_proofs.len();
+			if resolved_keys < expected_keys {
+				gum::warn!(
+					target: LOG_TARGET,
+					?session_index,
+					?candidate_hash,
+					"Could not generate key ownership proofs for {} keys",
+					expected_keys - resolved_keys,
+				);
+			}
+			debug_assert_eq!(resolved_keys, dispute_proofs.len());
+
+			for (key_ownership_proof, dispute_proof) in
+				key_ownership_proofs.into_iter().zip(dispute_proofs.into_iter())
+			{
+				let validator_id = dispute_proof.validator_id.clone();
+
+				gum::info!(
+					target: LOG_TARGET,
+					?session_index,
+					?candidate_hash,
+					key_ownership_proof_len = key_ownership_proof.len(),
+					"Trying to submit a slashing report",
+				);
+
+				let res = submit_report_dispute_lost(
+					ctx.sender(),
+					relay_parent,
+					dispute_proof,
+					key_ownership_proof,
+				)
+				.await;
+
+				match res {
+					Err(error) => {
+						gum::warn!(
+							target: LOG_TARGET,
+							?error,
+							?session_index,
+							?candidate_hash,
+							"Error reporting pending slash",
+						);
+					},
+					Ok(Some(())) => {
+						gum::info!(
+							target: LOG_TARGET,
+							?session_index,
+							?candidate_hash,
+							?validator_id,
+							"Successfully reported pending slash",
+						);
+					},
+					Ok(None) => {
+						gum::debug!(
+							target: LOG_TARGET,
+							?session_index,
+							?candidate_hash,
+							?validator_id,
+							"Duplicate pending slash report",
+						);
+					},
+				}
+			}
+		}
+	}
+
 	/// Process one batch of our `chain_import_backlog`.
 	///
 	/// `new_votes` will be appended beforehand.
@@ -477,10 +618,11 @@ impl Initialized {
 							validator_public.clone(),
 							validator_signature.clone(),
 						).is_ok(),
-						"Scraped backing votes had invalid signature! candidate: {:?}, session: {:?}, validator_public: {:?}",
+						"Scraped backing votes had invalid signature! candidate: {:?}, session: {:?}, validator_public: {:?}, validator_index: {}",
 						candidate_hash,
 						session,
 						validator_public,
+						validator_index.0,
 					);
 					let signed_dispute_statement =
 						SignedDisputeStatement::new_unchecked_from_trusted_source(

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

@@ -27,9 +27,12 @@ use polkadot_node_subsystem::{
 	messages::ChainApiMessage, overseer, ActivatedLeaf, ActiveLeavesUpdate, ChainApiError,
 	SubsystemSender,
 };
-use polkadot_node_subsystem_util::runtime::{get_candidate_events, get_on_chain_votes};
+use polkadot_node_subsystem_util::runtime::{
+	get_candidate_events, get_on_chain_votes, get_unapplied_slashes,
+};
 use polkadot_primitives::{
-	BlockNumber, CandidateEvent, CandidateHash, CandidateReceipt, Hash, ScrapedOnChainVotes,
+	vstaging, BlockNumber, CandidateEvent, CandidateHash, CandidateReceipt, Hash,
+	ScrapedOnChainVotes, SessionIndex,
 };
 
 use crate::{
@@ -64,11 +67,16 @@ const LRU_OBSERVED_BLOCKS_CAPACITY: NonZeroUsize = match NonZeroUsize::new(20) {
 pub struct ScrapedUpdates {
 	pub on_chain_votes: Vec<ScrapedOnChainVotes>,
 	pub included_receipts: Vec<CandidateReceipt>,
+	pub unapplied_slashes: Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>,
 }
 
 impl ScrapedUpdates {
 	pub fn new() -> Self {
-		Self { on_chain_votes: Vec::new(), included_receipts: Vec::new() }
+		Self {
+			on_chain_votes: Vec::new(),
+			included_receipts: Vec::new(),
+			unapplied_slashes: Vec::new(),
+		}
 	}
 }
 
@@ -120,7 +128,7 @@ impl Inclusions {
 			.retain(|_, blocks_including| blocks_including.keys().len() > 0);
 	}
 
-	pub fn get(&mut self, candidate: &CandidateHash) -> Vec<(BlockNumber, Hash)> {
+	pub fn get(&self, candidate: &CandidateHash) -> Vec<(BlockNumber, Hash)> {
 		let mut inclusions_as_vec: Vec<(BlockNumber, Hash)> = Vec::new();
 		if let Some(blocks_including) = self.inclusions_inner.get(candidate) {
 			for (height, blocks_at_height) in blocks_including.iter() {
@@ -256,6 +264,22 @@ impl ChainScraper {
 			}
 		}
 
+		// for unapplied slashes, we only look at the latest activated hash,
+		// it should accumulate them all
+		match get_unapplied_slashes(sender, activated.hash).await {
+			Ok(unapplied_slashes) => {
+				scraped_updates.unapplied_slashes = unapplied_slashes;
+			},
+			Err(error) => {
+				gum::debug!(
+					target: LOG_TARGET,
+					block_hash = ?activated.hash,
+					?error,
+					"Error fetching unapplied slashes.",
+				);
+			},
+		}
+
 		self.last_observed_blocks.put(activated.hash, ());
 
 		Ok(scraped_updates)
@@ -403,7 +427,7 @@ impl ChainScraper {
 	}
 
 	pub fn get_blocks_including_candidate(
-		&mut self,
+		&self,
 		candidate: &CandidateHash,
 	) -> Vec<(BlockNumber, Hash)> {
 		self.inclusions.get(candidate)

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

@@ -81,6 +81,7 @@ impl TestState {
 			)
 			.await;
 			assert_chain_vote_request(&mut ctx_handle, &chain).await;
+			assert_unapplied_slashes_request(&mut ctx_handle, &chain).await;
 		};
 
 		let (scraper, _) = join(ChainScraper::new(ctx.sender(), leaf.clone()), overseer_fut)
@@ -242,6 +243,18 @@ async fn assert_chain_vote_request(virtual_overseer: &mut VirtualOverseer, _chai
 	);
 }
 
+async fn assert_unapplied_slashes_request(virtual_overseer: &mut VirtualOverseer, _chain: &[Hash]) {
+	assert_matches!(
+		overseer_recv(virtual_overseer).await,
+		AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+			_hash,
+			RuntimeApiRequest::UnappliedSlashes(tx),
+		)) => {
+			tx.send(Ok(Vec::new())).unwrap();
+		}
+	);
+}
+
 async fn assert_finalized_block_number_request(
 	virtual_overseer: &mut VirtualOverseer,
 	response: BlockNumber,
@@ -287,6 +300,7 @@ async fn overseer_process_active_leaves_update<F>(
 		assert_candidate_events_request(virtual_overseer, chain, event_generator.clone()).await;
 		assert_chain_vote_request(virtual_overseer, chain).await;
 	}
+	assert_unapplied_slashes_request(virtual_overseer, chain).await;
 }
 
 #[test]

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

@@ -382,6 +382,12 @@ impl TestState {
 					})))
 					.unwrap();
 				},
+				AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+					_new_leaf,
+					RuntimeApiRequest::UnappliedSlashes(tx),
+				)) => {
+					tx.send(Ok(Vec::new())).unwrap();
+				},
 				AllMessages::ChainApi(ChainApiMessage::Ancestors { hash, k, response_channel }) => {
 					let target_header = self
 						.headers