malus: add the nemesis use cases

Cargo.toml

@@ -113,6 +113,7 @@ panic = "unwind"
 [features]
 runtime-benchmarks= [ "polkadot-cli/runtime-benchmarks" ]
 try-runtime = [ "polkadot-cli/try-runtime" ]
+disputes = [ "polkadot-cli/disputes" ]
 
 # Configuration for building a .deb package - for use with `cargo-deb`
 [package.metadata.deb]

cli/Cargo.toml

@@ -67,3 +67,4 @@ westend-native = [ "service/westend-native" ]
 rococo-native = [ "service/rococo-native" ]
 
 malus = [ "full-node", "service/malus" ]
+disputes = [ "service/disputes" ]

node/core/approval-voting/src/lib.rs

@@ -502,6 +502,8 @@ impl ApprovalState {
 }
 
 struct CurrentlyCheckingSet {
+	/// Invariant: The contained `Vec` needs to stay sorted as we are using `binary_search_by_key`
+	/// on it.
 	candidate_hash_map: HashMap<CandidateHash, Vec<Hash>>,
 	currently_checking: FuturesUnordered<BoxFuture<'static, ApprovalState>>,
 }
@@ -918,7 +920,7 @@ async fn handle_actions(
 
 				match confirmation_rx.await {
 					Err(oneshot::Canceled) => {
-						tracing::warn!(target: LOG_TARGET, "Dispute coordinator confirmation lost",)
+						tracing::debug!(target: LOG_TARGET, "Dispute coordinator confirmation lost",)
 					},
 					Ok(ImportStatementsResult::ValidImport) => {},
 					Ok(ImportStatementsResult::InvalidImport) => tracing::warn!(

node/core/backing/src/lib.rs

@@ -901,7 +901,7 @@ impl CandidateBackingJob {
 
 			match confirmation_rx.await {
 				Err(oneshot::Canceled) =>
-					tracing::warn!(target: LOG_TARGET, "Dispute coordinator confirmation lost",),
+					tracing::debug!(target: LOG_TARGET, "Dispute coordinator confirmation lost",),
 				Ok(ImportStatementsResult::ValidImport) => {},
 				Ok(ImportStatementsResult::InvalidImport) =>
 					tracing::warn!(target: LOG_TARGET, "Failed to import statements of validity",),

node/core/candidate-validation/src/lib.rs

@@ -70,6 +70,7 @@ pub struct Config {
 /// The candidate validation subsystem.
 pub struct CandidateValidationSubsystem {
 	metrics: Metrics,
+	pvf_metrics: polkadot_node_core_pvf::Metrics,
 	config: Config,
 }
 
@@ -78,8 +79,12 @@ impl CandidateValidationSubsystem {
 	/// strategy.
 	///
 	/// Check out [`IsolationStrategy`] to get more details.
-	pub fn with_config(config: Config, metrics: Metrics) -> Self {
-		CandidateValidationSubsystem { config, metrics }
+	pub fn with_config(
+		config: Config,
+		metrics: Metrics,
+		pvf_metrics: polkadot_node_core_pvf::Metrics,
+	) -> Self {
+		CandidateValidationSubsystem { config, metrics, pvf_metrics }
 	}
 }
 
@@ -89,17 +94,23 @@ where
 	Context: overseer::SubsystemContext<Message = CandidateValidationMessage>,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
-		let future =
-			run(ctx, self.metrics, self.config.artifacts_cache_path, self.config.program_path)
-				.map_err(|e| SubsystemError::with_origin("candidate-validation", e))
-				.boxed();
+		let future = run(
+			ctx,
+			self.metrics,
+			self.pvf_metrics,
+			self.config.artifacts_cache_path,
+			self.config.program_path,
+		)
+		.map_err(|e| SubsystemError::with_origin("candidate-validation", e))
+		.boxed();
 		SpawnedSubsystem { name: "candidate-validation-subsystem", future }
 	}
 }
 
 async fn run<Context>(
 	mut ctx: Context,
 	metrics: Metrics,
+	pvf_metrics: polkadot_node_core_pvf::Metrics,
 	cache_path: PathBuf,
 	program_path: PathBuf,
 ) -> SubsystemResult<()>
@@ -109,6 +120,7 @@ where
 {
 	let (mut validation_host, task) = polkadot_node_core_pvf::start(
 		polkadot_node_core_pvf::Config::new(cache_path, program_path),
+		pvf_metrics,
 	);
 	ctx.spawn_blocking("pvf-validation-host", task.boxed())?;
 

node/core/dispute-coordinator/Cargo.toml

@@ -28,3 +28,7 @@ sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 assert_matches = "1.4.0"
 polkadot-overseer = { path = "../../overseer" }
+
+[features]
+# If not enabled, the dispute coordinator will do nothing.
+disputes = []

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

@@ -0,0 +1,263 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Implements the dispute coordinator subsystem (dummy implementation).
+
+use std::sync::Arc;
+
+use polkadot_node_subsystem::{
+	errors::{ChainApiError, RuntimeApiError},
+	messages::DisputeCoordinatorMessage,
+	overseer, FromOverseer, OverseerSignal, SpawnedSubsystem, SubsystemContext, SubsystemError,
+};
+use polkadot_primitives::v1::BlockNumber;
+
+use futures::{channel::oneshot, prelude::*};
+use kvdb::KeyValueDB;
+use parity_scale_codec::{Decode, Encode, Error as CodecError};
+use polkadot_node_subsystem_util::metrics::{self, prometheus};
+use sc_keystore::LocalKeystore;
+
+const LOG_TARGET: &str = "parachain::dispute-coordinator";
+
+/// Candidate validation metrics placeholder.
+#[derive(Default, Clone)]
+pub struct Metrics;
+
+impl metrics::Metrics for Metrics {
+	fn try_register(_: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
+		Ok(Metrics)
+	}
+}
+
+/// Timestamp based on the 1 Jan 1970 UNIX base, which is persistent across node restarts and OS reboots.
+type Timestamp = u64;
+
+#[derive(Eq, PartialEq)]
+enum Participation {}
+
+struct State {}
+
+/// Configuration for the dispute coordinator subsystem.
+#[derive(Debug, Clone, Copy)]
+pub struct Config {
+	/// The data column in the store to use for dispute data.
+	pub col_data: u32,
+}
+
+/// An implementation of the dispute coordinator subsystem.
+pub struct DisputeCoordinatorSubsystem {}
+
+impl DisputeCoordinatorSubsystem {
+	/// Create a new instance of the subsystem.
+	pub fn new(_: Arc<dyn KeyValueDB>, _: Config, _: Arc<LocalKeystore>, _: Metrics) -> Self {
+		Self {}
+	}
+}
+
+impl<Context> overseer::Subsystem<Context, SubsystemError> for DisputeCoordinatorSubsystem
+where
+	Context: SubsystemContext<Message = DisputeCoordinatorMessage>,
+	Context: overseer::SubsystemContext<Message = DisputeCoordinatorMessage>,
+{
+	fn start(self, ctx: Context) -> SpawnedSubsystem {
+		let future = run(self, ctx).map(|_| Ok(())).boxed();
+
+		SpawnedSubsystem { name: "dispute-coordinator-subsystem", future }
+	}
+}
+
+#[derive(Debug, thiserror::Error)]
+#[allow(missing_docs)]
+pub enum Error {
+	#[error(transparent)]
+	RuntimeApi(#[from] RuntimeApiError),
+
+	#[error(transparent)]
+	ChainApi(#[from] ChainApiError),
+
+	#[error(transparent)]
+	Io(#[from] std::io::Error),
+
+	#[error(transparent)]
+	Oneshot(#[from] oneshot::Canceled),
+
+	#[error("Oneshot send failed")]
+	OneshotSend,
+
+	#[error(transparent)]
+	Subsystem(#[from] SubsystemError),
+
+	#[error(transparent)]
+	Codec(#[from] CodecError),
+}
+
+impl Error {
+	fn trace(&self) {
+		match self {
+			// don't spam the log with spurious errors
+			Self::RuntimeApi(_) | Self::Oneshot(_) =>
+				tracing::debug!(target: LOG_TARGET, err = ?self),
+			// it's worth reporting otherwise
+			_ => tracing::warn!(target: LOG_TARGET, err = ?self),
+		}
+	}
+}
+
+/// The status of dispute. This is a state machine which can be altered by the
+/// helper methods.
+#[derive(Debug, Clone, Copy, Encode, Decode, PartialEq)]
+pub enum DisputeStatus {
+	/// The dispute is active and unconcluded.
+	#[codec(index = 0)]
+	Active,
+	/// The dispute has been concluded in favor of the candidate
+	/// since the given timestamp.
+	#[codec(index = 1)]
+	ConcludedFor(Timestamp),
+	/// The dispute has been concluded against the candidate
+	/// since the given timestamp.
+	///
+	/// This takes precedence over `ConcludedFor` in the case that
+	/// both are true, which is impossible unless a large amount of
+	/// validators are participating on both sides.
+	#[codec(index = 2)]
+	ConcludedAgainst(Timestamp),
+}
+
+impl DisputeStatus {
+	/// Initialize the status to the active state.
+	pub fn active() -> DisputeStatus {
+		DisputeStatus::Active
+	}
+
+	/// Transition the status to a new status after observing the dispute has concluded for the candidate.
+	/// This may be a no-op if the status was already concluded.
+	pub fn concluded_for(self, now: Timestamp) -> DisputeStatus {
+		match self {
+			DisputeStatus::Active => DisputeStatus::ConcludedFor(now),
+			DisputeStatus::ConcludedFor(at) => DisputeStatus::ConcludedFor(std::cmp::min(at, now)),
+			against => against,
+		}
+	}
+
+	/// Transition the status to a new status after observing the dispute has concluded against the candidate.
+	/// This may be a no-op if the status was already concluded.
+	pub fn concluded_against(self, now: Timestamp) -> DisputeStatus {
+		match self {
+			DisputeStatus::Active => DisputeStatus::ConcludedAgainst(now),
+			DisputeStatus::ConcludedFor(at) =>
+				DisputeStatus::ConcludedAgainst(std::cmp::min(at, now)),
+			DisputeStatus::ConcludedAgainst(at) =>
+				DisputeStatus::ConcludedAgainst(std::cmp::min(at, now)),
+		}
+	}
+
+	/// Whether the disputed candidate is possibly invalid.
+	pub fn is_possibly_invalid(&self) -> bool {
+		match self {
+			DisputeStatus::Active | DisputeStatus::ConcludedAgainst(_) => true,
+			DisputeStatus::ConcludedFor(_) => false,
+		}
+	}
+
+	/// Yields the timestamp this dispute concluded at, if any.
+	pub fn concluded_at(&self) -> Option<Timestamp> {
+		match self {
+			DisputeStatus::Active => None,
+			DisputeStatus::ConcludedFor(at) | DisputeStatus::ConcludedAgainst(at) => Some(*at),
+		}
+	}
+}
+
+async fn run<Context>(subsystem: DisputeCoordinatorSubsystem, mut ctx: Context)
+where
+	Context: overseer::SubsystemContext<Message = DisputeCoordinatorMessage>,
+	Context: SubsystemContext<Message = DisputeCoordinatorMessage>,
+{
+	loop {
+		let res = run_until_error(&mut ctx, &subsystem).await;
+		match res {
+			Err(e) => {
+				e.trace();
+
+				if let Error::Subsystem(SubsystemError::Context(_)) = e {
+					break
+				}
+			},
+			Ok(()) => {
+				tracing::info!(target: LOG_TARGET, "received `Conclude` signal, exiting");
+				break
+			},
+		}
+	}
+}
+
+async fn run_until_error<Context>(
+	ctx: &mut Context,
+	_: &DisputeCoordinatorSubsystem,
+) -> Result<(), Error>
+where
+	Context: overseer::SubsystemContext<Message = DisputeCoordinatorMessage>,
+	Context: SubsystemContext<Message = DisputeCoordinatorMessage>,
+{
+	let mut state = State {};
+
+	loop {
+		match ctx.recv().await? {
+			FromOverseer::Signal(OverseerSignal::Conclude) => return Ok(()),
+			FromOverseer::Signal(OverseerSignal::ActiveLeaves(_)) => {},
+			FromOverseer::Signal(OverseerSignal::BlockFinalized(_, _)) => {},
+			FromOverseer::Communication { msg } => handle_incoming(ctx, &mut state, msg).await?,
+		}
+	}
+}
+
+async fn handle_incoming(
+	_: &mut impl SubsystemContext,
+	_: &mut State,
+	message: DisputeCoordinatorMessage,
+) -> Result<(), Error> {
+	match message {
+		DisputeCoordinatorMessage::ImportStatements { .. } => { /* just drop confirmation */ },
+		DisputeCoordinatorMessage::RecentDisputes(tx) => {
+			let _ = tx.send(Vec::new());
+		},
+		DisputeCoordinatorMessage::ActiveDisputes(tx) => {
+			let _ = tx.send(Vec::new());
+		},
+		DisputeCoordinatorMessage::QueryCandidateVotes(_, tx) => {
+			let _ = tx.send(Vec::new());
+		},
+		DisputeCoordinatorMessage::IssueLocalStatement(_, _, _, _) => {},
+		DisputeCoordinatorMessage::DetermineUndisputedChain {
+			base_number,
+			block_descriptions,
+			tx,
+		} => {
+			let undisputed_chain = block_descriptions
+				.last()
+				.map(|e| (base_number + block_descriptions.len() as BlockNumber, e.block_hash));
+
+			let _ = tx.send(undisputed_chain);
+		},
+	}
+
+	Ok(())
+}
+
+#[derive(Debug, thiserror::Error)]
+enum DisputeMessageCreationError {}