dispute-coordinator: past session dispute slashing

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

@@ -25,6 +25,7 @@ use futures::{
 
 use sc_keystore::LocalKeystore;
 
+use parity_scale_codec::Encode;
 use polkadot_node_primitives::{
 	disputes::ValidCandidateVotes, CandidateVotes, DisputeStatus, SignedDisputeStatement, Timestamp,
 };
@@ -35,11 +36,12 @@ use polkadot_node_subsystem::{
 	},
 	overseer, ActivatedLeaf, ActiveLeavesUpdate, FromOrchestra, OverseerSignal,
 };
-use polkadot_node_subsystem_util::rolling_session_window::{
-	RollingSessionWindow, SessionWindowUpdate, SessionsUnavailable,
+use polkadot_node_subsystem_util::{
+	rolling_session_window::{RollingSessionWindow, SessionWindowUpdate, SessionsUnavailable},
+	runtime::{key_ownership_proof, submit_report_dispute_lost},
 };
 use polkadot_primitives::{
-	BlockNumber, CandidateHash, CandidateReceipt, CompactStatement, DisputeStatement,
+	vstaging, BlockNumber, CandidateHash, CandidateReceipt, CompactStatement, DisputeStatement,
 	DisputeStatementSet, Hash, ScrapedOnChainVotes, SessionIndex, SessionInfo,
 	ValidDisputeStatementKind, ValidatorId, ValidatorIndex,
 };
@@ -49,6 +51,7 @@ use crate::{
 	import::{CandidateEnvironment, CandidateVoteState},
 	is_potential_spam,
 	metrics::Metrics,
+	scraping::ScrapedUpdates,
 	status::{get_active_with_status, Clock},
 	DisputeCoordinatorSubsystem, LOG_TARGET,
 };
@@ -314,9 +317,12 @@ impl Initialized {
 				Ok(SessionWindowUpdate::Unchanged) => {},
 			};
 
+			let ScrapedUpdates { unapplied_slashes, on_chain_votes, .. } = scraped_updates;
+
+			self.process_unapplied_slashes(ctx, new_leaf.hash, unapplied_slashes).await;
 			// The `runtime-api` subsystem has an internal queue which serializes the execution,
 			// so there is no point in running these in parallel.
-			for votes in scraped_updates.on_chain_votes {
+			for votes in on_chain_votes {
 				let _ = self.process_on_chain_votes(ctx, overlay_db, votes, now).await.map_err(
 					|error| {
 						gum::warn!(
@@ -332,6 +338,137 @@ impl Initialized {
 		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);
+			debug_assert!(
+				!inclusions.is_empty(),
+				"Couldn't find inclusion parent for an unapplied slash",
+			);
+
+			// 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.clone(),
+							);
+							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();
+				let opaque_key_ownership_proof =
+					vstaging::slashing::OpaqueKeyOwnershipProof::new(key_ownership_proof.encode());
+
+				let res = submit_report_dispute_lost(
+					ctx.sender(),
+					relay_parent,
+					dispute_proof,
+					opaque_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",
+						);
+					},
+				}
+			}
+		}
+	}
+
 	/// Scrapes on-chain votes (backing votes and concluded disputes) for a active leaf of the
 	/// relay chain.
 	async fn process_on_chain_votes<Context>(

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() {
@@ -252,6 +260,11 @@ impl ChainScraper {
 			}
 		}
 
+		// for unapplied slashes, we only look at the latest activated hash,
+		// it should accumulate them all
+		let unapplied_slashes = get_unapplied_slashes(sender, activated.hash).await?;
+		scraped_updates.unapplied_slashes = unapplied_slashes;
+
 		self.last_observed_blocks.put(activated.hash, ());
 
 		Ok(scraped_updates)
@@ -396,7 +409,7 @@ impl ChainScraper {
 	}
 
 	pub fn get_blocks_including_candidate(
-		&mut self,
+		&self,
 		candidate: &CandidateHash,
 	) -> Vec<(BlockNumber, Hash)> {
 		self.inclusions.get(candidate)

node/core/runtime-api/src/cache.rs

@@ -20,12 +20,11 @@ use lru::LruCache;
 use sp_consensus_babe::Epoch;
 
 use polkadot_primitives::{
-	vstaging::ExecutorParams, AuthorityDiscoveryId, BlockNumber, CandidateCommitments,
-	CandidateEvent, CandidateHash, CommittedCandidateReceipt, CoreState, DisputeState,
-	GroupRotationInfo, Hash, Id as ParaId, InboundDownwardMessage, InboundHrmpMessage,
-	OccupiedCoreAssumption, PersistedValidationData, PvfCheckStatement, ScrapedOnChainVotes,
-	SessionIndex, SessionInfo, ValidationCode, ValidationCodeHash, ValidatorId, ValidatorIndex,
-	ValidatorSignature,
+	vstaging, AuthorityDiscoveryId, BlockNumber, CandidateCommitments, CandidateEvent,
+	CandidateHash, CommittedCandidateReceipt, CoreState, DisputeState, GroupRotationInfo, Hash,
+	Id as ParaId, InboundDownwardMessage, InboundHrmpMessage, OccupiedCoreAssumption,
+	PersistedValidationData, PvfCheckStatement, ScrapedOnChainVotes, SessionIndex, SessionInfo,
+	ValidationCode, ValidationCodeHash, ValidatorId, ValidatorIndex, ValidatorSignature,
 };
 
 /// For consistency we have the same capacity for all caches. We use 128 as we'll only need that
@@ -52,7 +51,7 @@ pub(crate) struct RequestResultCache {
 	validation_code_by_hash: LruCache<ValidationCodeHash, Option<ValidationCode>>,
 	candidate_pending_availability: LruCache<(Hash, ParaId), Option<CommittedCandidateReceipt>>,
 	candidate_events: LruCache<Hash, Vec<CandidateEvent>>,
-	session_executor_params: LruCache<SessionIndex, Option<ExecutorParams>>,
+	session_executor_params: LruCache<SessionIndex, Option<vstaging::ExecutorParams>>,
 	session_info: LruCache<SessionIndex, SessionInfo>,
 	dmq_contents: LruCache<(Hash, ParaId), Vec<InboundDownwardMessage<BlockNumber>>>,
 	inbound_hrmp_channels_contents:
@@ -64,6 +63,10 @@ pub(crate) struct RequestResultCache {
 		LruCache<(Hash, ParaId, OccupiedCoreAssumption), Option<ValidationCodeHash>>,
 	version: LruCache<Hash, u32>,
 	disputes: LruCache<Hash, Vec<(SessionIndex, CandidateHash, DisputeState<BlockNumber>)>>,
+	unapplied_slashes:
+		LruCache<Hash, Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>>,
+	key_ownership_proof:
+		LruCache<(Hash, ValidatorId), Option<vstaging::slashing::OpaqueKeyOwnershipProof>>,
 }
 
 impl Default for RequestResultCache {
@@ -91,6 +94,8 @@ impl Default for RequestResultCache {
 			validation_code_hash: LruCache::new(DEFAULT_CACHE_CAP),
 			version: LruCache::new(DEFAULT_CACHE_CAP),
 			disputes: LruCache::new(DEFAULT_CACHE_CAP),
+			unapplied_slashes: LruCache::new(DEFAULT_CACHE_CAP),
+			key_ownership_proof: LruCache::new(DEFAULT_CACHE_CAP),
 		}
 	}
 }
@@ -269,14 +274,14 @@ impl RequestResultCache {
 	pub(crate) fn session_executor_params(
 		&mut self,
 		session_index: SessionIndex,
-	) -> Option<&Option<ExecutorParams>> {
+	) -> Option<&Option<vstaging::ExecutorParams>> {
 		self.session_executor_params.get(&session_index)
 	}
 
 	pub(crate) fn cache_session_executor_params(
 		&mut self,
 		session_index: SessionIndex,
-		value: Option<ExecutorParams>,
+		value: Option<vstaging::ExecutorParams>,
 	) {
 		self.session_executor_params.put(session_index, value);
 	}
@@ -386,6 +391,44 @@ impl RequestResultCache {
 	) {
 		self.disputes.put(relay_parent, value);
 	}
+
+	pub(crate) fn unapplied_slashes(
+		&mut self,
+		relay_parent: &Hash,
+	) -> Option<&Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>> {
+		self.unapplied_slashes.get(relay_parent)
+	}
+
+	pub(crate) fn cache_unapplied_slashes(
+		&mut self,
+		relay_parent: Hash,
+		value: Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>,
+	) {
+		self.unapplied_slashes.put(relay_parent, value);
+	}
+
+	pub(crate) fn key_ownership_proof(
+		&mut self,
+		key: (Hash, ValidatorId),
+	) -> Option<&Option<vstaging::slashing::OpaqueKeyOwnershipProof>> {
+		self.key_ownership_proof.get(&key)
+	}
+
+	pub(crate) fn cache_key_ownership_proof(
+		&mut self,
+		key: (Hash, ValidatorId),
+		value: Option<vstaging::slashing::OpaqueKeyOwnershipProof>,
+	) {
+		self.key_ownership_proof.put(key, value);
+	}
+
+	// This request is never cached, hence always returns `None`.
+	pub(crate) fn submit_report_dispute_lost(
+		&mut self,
+		_key: (Hash, vstaging::slashing::DisputeProof, vstaging::slashing::OpaqueKeyOwnershipProof),
+	) -> Option<&Option<()>> {
+		None
+	}
 }
 
 pub(crate) enum RequestResult {
@@ -407,7 +450,7 @@ pub(crate) enum RequestResult {
 	ValidationCodeByHash(Hash, ValidationCodeHash, Option<ValidationCode>),
 	CandidatePendingAvailability(Hash, ParaId, Option<CommittedCandidateReceipt>),
 	CandidateEvents(Hash, Vec<CandidateEvent>),
-	SessionExecutorParams(Hash, SessionIndex, Option<ExecutorParams>),
+	SessionExecutorParams(Hash, SessionIndex, Option<vstaging::ExecutorParams>),
 	SessionInfo(Hash, SessionIndex, Option<SessionInfo>),
 	DmqContents(Hash, ParaId, Vec<InboundDownwardMessage<BlockNumber>>),
 	InboundHrmpChannelsContents(
@@ -423,4 +466,13 @@ pub(crate) enum RequestResult {
 	ValidationCodeHash(Hash, ParaId, OccupiedCoreAssumption, Option<ValidationCodeHash>),
 	Version(Hash, u32),
 	Disputes(Hash, Vec<(SessionIndex, CandidateHash, DisputeState<BlockNumber>)>),
+	UnappliedSlashes(Hash, Vec<(SessionIndex, CandidateHash, vstaging::slashing::PendingSlashes)>),
+	KeyOwnershipProof(Hash, ValidatorId, Option<vstaging::slashing::OpaqueKeyOwnershipProof>),
+	// This is a request with side-effects.
+	SubmitReportDisputeLost(
+		Hash,
+		vstaging::slashing::DisputeProof,
+		vstaging::slashing::OpaqueKeyOwnershipProof,
+		Option<()>,
+	),
 }