Dispute distribution implementation

Cargo.toml

@@ -64,6 +64,7 @@ members = [
 	"node/network/availability-recovery",
 	"node/network/collator-protocol",
 	"node/network/gossip-support",
+	"node/network/dispute-distribution",
 	"node/overseer",
 	"node/malus",
 	"node/primitives",

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

@@ -28,10 +28,11 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 
-use polkadot_node_primitives::{CandidateVotes, SignedDisputeStatement};
+use polkadot_node_primitives::{CandidateVotes, DISPUTE_WINDOW, SignedDisputeStatement};
 use polkadot_node_subsystem::{
 	messages::{
 		DisputeCoordinatorMessage, ChainApiMessage, DisputeParticipationMessage,
+		ImportStatementsResult,
 	},
 	Subsystem, SubsystemContext, FromOverseer, OverseerSignal, SpawnedSubsystem,
 	SubsystemError,
@@ -60,10 +61,6 @@ mod tests;
 
 const LOG_TARGET: &str = "parachain::dispute-coordinator";
 
-// It would be nice to draw this from the chain state, but we have no tools for it right now.
-// On Polkadot this is 1 day, and on Kusama it's 6 hours.
-const DISPUTE_WINDOW: SessionIndex = 6;
-
 struct State {
 	keystore: Arc<LocalKeystore>,
 	highest_session: Option<SessionIndex>,
@@ -131,6 +128,9 @@ pub enum Error {
 	#[error(transparent)]
 	Oneshot(#[from] oneshot::Canceled),
 
+	#[error("Oneshot send failed")]
+	OneshotSend,
+
 	#[error(transparent)]
 	Subsystem(#[from] SubsystemError),
 
@@ -301,6 +301,7 @@ async fn handle_incoming(
 			candidate_receipt,
 			session,
 			statements,
+			pending_confirmation,
 		} => {
 			handle_import_statements(
 				ctx,
@@ -312,6 +313,11 @@ async fn handle_incoming(
 				session,
 				statements,
 			).await?;
+			// TODO: Recover availability and report back unavailable data:
+			// /~https://github.com/paritytech/polkadot/issues/3399
+			pending_confirmation
+				.send(ImportStatementsResult::ValidImport)
+				.map_err(|_| Error::OneshotSend)?;
 		}
 		DisputeCoordinatorMessage::ActiveDisputes(rx) => {
 			let active_disputes = db::v1::load_active_disputes(store, &config.column_config())?

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

@@ -261,6 +261,7 @@ fn conflicting_votes_lead_to_dispute_participation() {
 			false,
 		).await;
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -270,6 +271,7 @@ fn conflicting_votes_lead_to_dispute_participation() {
 					(valid_vote, ValidatorIndex(0)),
 					(invalid_vote, ValidatorIndex(1)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -310,6 +312,7 @@ fn conflicting_votes_lead_to_dispute_participation() {
 			assert_eq!(votes.invalid.len(), 1);
 		}
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -318,6 +321,7 @@ fn conflicting_votes_lead_to_dispute_participation() {
 				statements: vec![
 					(invalid_vote_2, ValidatorIndex(2)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -371,6 +375,7 @@ fn positive_votes_dont_trigger_participation() {
 			true,
 		).await;
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -379,6 +384,7 @@ fn positive_votes_dont_trigger_participation() {
 				statements: vec![
 					(valid_vote, ValidatorIndex(0)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -404,6 +410,7 @@ fn positive_votes_dont_trigger_participation() {
 			assert!(votes.invalid.is_empty());
 		}
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -412,6 +419,7 @@ fn positive_votes_dont_trigger_participation() {
 				statements: vec![
 					(valid_vote_2, ValidatorIndex(1)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -472,6 +480,7 @@ fn wrong_validator_index_is_ignored() {
 			false,
 		).await;
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -481,6 +490,7 @@ fn wrong_validator_index_is_ignored() {
 					(valid_vote, ValidatorIndex(1)),
 					(invalid_vote, ValidatorIndex(0)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -541,6 +551,7 @@ fn finality_votes_ignore_disputed_candidates() {
 			false,
 		).await;
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -550,6 +561,7 @@ fn finality_votes_ignore_disputed_candidates() {
 					(valid_vote, ValidatorIndex(0)),
 					(invalid_vote, ValidatorIndex(1)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 		let _ = virtual_overseer.recv().await;
@@ -624,6 +636,7 @@ fn supermajority_valid_dispute_may_be_finalized() {
 			false,
 		).await;
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
@@ -633,6 +646,7 @@ fn supermajority_valid_dispute_may_be_finalized() {
 					(valid_vote, ValidatorIndex(0)),
 					(invalid_vote, ValidatorIndex(1)),
 				],
+				pending_confirmation,
 			},
 		}).await;
 
@@ -650,12 +664,14 @@ fn supermajority_valid_dispute_may_be_finalized() {
 			statements.push((vote, ValidatorIndex(i as _)));
 		};
 
+		let (pending_confirmation, _confirmation_rx) = oneshot::channel();
 		virtual_overseer.send(FromOverseer::Communication {
 			msg: DisputeCoordinatorMessage::ImportStatements {
 				candidate_hash,
 				candidate_receipt: candidate_receipt.clone(),
 				session,
 				statements,
+				pending_confirmation,
 			},
 		}).await;
 

node/network/availability-distribution/Cargo.toml

@@ -27,7 +27,6 @@ polkadot-subsystem-testhelpers = { package = "polkadot-node-subsystem-test-helpe
 sp-core = { git = "/~https://github.com/paritytech/substrate", branch = "master", features = ["std"] }
 sp-keyring = { git = "/~https://github.com/paritytech/substrate", branch = "master" }
 sp-tracing = { git = "/~https://github.com/paritytech/substrate", branch = "master" }
-sc-keystore = { git = "/~https://github.com/paritytech/substrate", branch = "master" }
 sc-network = { git = "/~https://github.com/paritytech/substrate", branch = "master" }
 futures-timer = "3.0.2"
 assert_matches = "1.4.0"

node/network/availability-distribution/src/pov_requester/mod.rs

@@ -19,10 +19,8 @@
 use futures::{FutureExt, channel::oneshot, future::BoxFuture};
 
 use polkadot_subsystem::jaeger;
-use polkadot_node_network_protocol::{
-	request_response::{OutgoingRequest, Recipient, request::{RequestError, Requests},
-	v1::{PoVFetchingRequest, PoVFetchingResponse}}
-};
+use polkadot_node_network_protocol::request_response::{OutgoingRequest, Recipient, request::{RequestError, Requests},
+	v1::{PoVFetchingRequest, PoVFetchingResponse}};
 use polkadot_primitives::v1::{
 	CandidateHash, Hash, ValidatorIndex,
 };
@@ -49,7 +47,7 @@ pub async fn fetch_pov<Context>(
 where
 	Context: SubsystemContext,
 {
-	let info = &runtime.get_session_info(ctx, parent).await?.session_info;
+	let info = &runtime.get_session_info(ctx.sender(), parent).await?.session_info;
 	let authority_id = info.discovery_keys.get(from_validator.0 as usize)
 		.ok_or(NonFatal::InvalidValidatorIndex)?
 		.clone();
@@ -131,10 +129,11 @@ mod tests {
 	use polkadot_node_primitives::BlockData;
 	use polkadot_subsystem_testhelpers as test_helpers;
 	use polkadot_subsystem::messages::{AvailabilityDistributionMessage, RuntimeApiMessage, RuntimeApiRequest};
+	use test_helpers::mock::make_ferdie_keystore;
 
 	use super::*;
 	use crate::LOG_TARGET;
-	use crate::tests::mock::{make_session_info, make_ferdie_keystore};
+	use crate::tests::mock::{make_session_info};
 
 	#[test]
 	fn rejects_invalid_pov() {

node/network/availability-distribution/src/requester/session_cache.rs

@@ -106,7 +106,7 @@ impl SessionCache {
 		Context: SubsystemContext,
 		F: FnOnce(&SessionInfo) -> R,
 	{
-		let session_index = runtime.get_session_index(ctx, parent).await?;
+		let session_index = runtime.get_session_index(ctx.sender(), parent).await?;
 
 		if let Some(o_info) = self.session_info_cache.get(&session_index) {
 			tracing::trace!(target: LOG_TARGET, session_index, "Got session from lru");
@@ -183,7 +183,7 @@ impl SessionCache {
 	where
 		Context: SubsystemContext,
 	{
-		let info = runtime.get_session_info_by_index(ctx, parent, session_index).await?;
+		let info = runtime.get_session_info_by_index(ctx.sender(), parent, session_index).await?;
 
 		let discovery_keys = info.session_info.discovery_keys.clone();
 		let mut validator_groups = info.session_info.validator_groups.clone();

node/network/availability-distribution/src/tests/mock.rs

@@ -19,30 +19,16 @@
 
 use std::sync::Arc;
 
-use sc_keystore::LocalKeystore;
 use sp_keyring::Sr25519Keyring;
-use sp_application_crypto::AppKey;
 
 use polkadot_erasure_coding::{branches, obtain_chunks_v1 as obtain_chunks};
 use polkadot_primitives::v1::{
 	CandidateCommitments, CandidateDescriptor, CandidateHash,
 	CommittedCandidateReceipt, GroupIndex, Hash, HeadData, Id as ParaId,
-	OccupiedCore, PersistedValidationData, SessionInfo, ValidatorId, ValidatorIndex
+	OccupiedCore, PersistedValidationData, SessionInfo, ValidatorIndex
 };
 use polkadot_node_primitives::{PoV, ErasureChunk, AvailableData, BlockData};
-use sp_keystore::{SyncCryptoStore, SyncCryptoStorePtr};
-
-/// Get mock keystore with `Ferdie` key.
-pub fn make_ferdie_keystore() -> SyncCryptoStorePtr {
-	let keystore: SyncCryptoStorePtr = Arc::new(LocalKeystore::in_memory());
-	SyncCryptoStore::sr25519_generate_new(
-		&*keystore,
-		ValidatorId::ID,
-		Some(&Sr25519Keyring::Ferdie.to_seed()),
-	)
-	.expect("Insert key into keystore");
-	keystore
-}
+
 
 /// Create dummy session info with two validator groups.
 pub fn make_session_info() -> SessionInfo {