generics haggling

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

@@ -56,6 +56,7 @@ use async_trait::async_trait;
 const LOG_TARGET: &'static str = "parachain::candidate-validation";
 
 /// Configuration for the candidate validation subsystem
+#[derive(Clone)]
 pub struct Config {
 	/// The path where candidate validation can store compiled artifacts for PVFs.
 	pub artifacts_cache_path: PathBuf,

node/malus/Cargo.toml

@@ -19,6 +19,7 @@ publish = false
 [dependencies]
 polkadot-cli = { path = "../../cli", default-features = false, features = [ "cli", "malus" ]  }
 polkadot-node-subsystem = { path = "../subsystem" }
+polkadot-node-core-candidate-validation = { path = "../core/candidate-validation" }
 parity-util-mem = { version = "*", default-features = false, features = ["jemalloc-global"] }
 color-eyre = { version = "0.5.11", default-features = false }
 assert_matches = "1.5"

node/malus/src/lib.rs

@@ -20,8 +20,10 @@
 //! multiple subsystems and intercept or replace incoming and outgoing
 //! messages on the overseer level.
 
-use polkadot_node_subsystem::messages::AllMessages;
-use polkadot_node_subsystem::FromOverseer;
+pub use polkadot_node_subsystem::{
+	FromOverseer,
+	messages::AllMessages,
+};
 use polkadot_node_subsystem::*;
 use std::future::Future;
 use std::pin::Pin;
@@ -33,7 +35,7 @@ pub trait MsgFilter: Send + Sync + Clone + 'static {
 
 	/// Filter messages that are to be received by
 	/// the subsystem.
-	fn filter_in(&self, msg: Self::Message) -> Option<Self::Message> {
+	fn filter_in(&self, msg: FromOverseer<Self::Message>) -> Option<FromOverseer<Self::Message>> {
 		Some(msg)
 	}
 
@@ -80,19 +82,19 @@ where
 }
 
 /// A subsystem context, that filters the outgoing messages.
-pub struct FilteredContext<X: SubsystemContext, Fil: MsgFilter> {
-	inner: X,
+pub struct FilteredContext<Context: SubsystemContext, Fil: MsgFilter> {
+	inner: Context,
 	message_filter: Fil,
-	sender: FilteredSender<<X as SubsystemContext>::Sender, Fil>,
+	sender: FilteredSender<<Context as SubsystemContext>::Sender, Fil>,
 }
 
-impl<X, Fil> FilteredContext<X, Fil>
+impl<Context, Fil> FilteredContext<Context, Fil>
 where
-	X: SubsystemContext,
-	Fil: MsgFilter,
+	Context: SubsystemContext,
+	Fil: MsgFilter<Message=<Context as SubsystemContext>::Message>,
 {
-	pub fn new(mut inner: X, message_filter: Fil) -> Self {
-		let sender = FilteredSender::<<X as SubsystemContext>::Sender, Fil> {
+	pub fn new(mut inner: Context, message_filter: Fil) -> Self {
+		let sender = FilteredSender::<<Context as SubsystemContext>::Sender, Fil> {
 			inner: inner.sender().clone(),
 			message_filter: message_filter.clone(),
 		};
@@ -108,7 +110,7 @@ where
 impl<Context, Fil> SubsystemContext for FilteredContext<Context, Fil>
 where
 	Context: SubsystemContext,
-	Fil: MsgFilter<Message = FromOverseer<<Context as SubsystemContext>::Message>>,
+	Fil: MsgFilter<Message = <Context as SubsystemContext>::Message>,
 {
 	type Message = <Context as SubsystemContext>::Message;
 	type Sender = FilteredSender<<Context as SubsystemContext>::Sender, Fil>;
@@ -173,11 +175,13 @@ impl<Sub, Fil> FilteredSubsystem<Sub, Fil> {
 
 impl<Context, Sub, Fil> Subsystem<Context> for FilteredSubsystem<Sub, Fil>
 where
-	Sub: Subsystem<Context>,
+	Context: SubsystemContext + Sync + Send,
+	Sub: Subsystem<FilteredContext<Context, Fil>>,
+	FilteredContext<Context, Fil>: SubsystemContext,
 	Fil: MsgFilter<Message = <Context as SubsystemContext>::Message>,
-	Context: FilteredContext<impl SubsystemContext + Sync + Send, Fil>,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
-		self.subsystem.start(ctx)
+		let ctx = FilteredContext::new(ctx, self.message_filter);
+		Subsystem::<FilteredContext<Context, Fil>>::start(self.subsystem, ctx)
 	}
 }

node/malus/src/variant-a.rs

@@ -24,18 +24,31 @@
 
 use color_eyre::eyre;
 use polkadot_cli::{
+	create_default_subsystems,
 	service::{
-		create_default_subsystems, AuthorityDiscoveryApi, AuxStore, BabeApi, Block, Error,
+		AuthorityDiscoveryApi, AuxStore, BabeApi, Block, Error,
 		HeaderBackend, Overseer, OverseerGen, OverseerGenArgs, OverseerHandler, ParachainHost,
 		ProvideRuntimeApi, RealOverseerGen, SpawnNamed,
 	},
 	Cli,
 };
-use std::sync::atomic::AtomicUsize;
+
+// Import extra types relevant to the particular
+// subsystem.
+use polkadot_node_subsystem::messages::CandidateValidationMessage;
+use polkadot_node_core_candidate_validation::{
+	Metrics, CandidateValidationSubsystem,
+};
+
+// Filter wrapping related types.
+use malus::*;
+
+
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+
 use structopt::StructOpt;
 
-use malus::*;
 
 /// Silly example, just drop every second outgoing message.
 #[derive(Clone, Default, Debug)]
@@ -44,8 +57,8 @@ struct Skippy(Arc<AtomicUsize>);
 impl MsgFilter for Skippy {
 	type Message = CandidateValidationMessage;
 
-	fn filter_in(&self, msg: Self::Message) -> Option<FromOverseer<Self::Message>> {
-		if self.0.fetch_add(1, Ordering::Relaxed) % 2 {
+	fn filter_in(&self, msg: FromOverseer<Self::Message>) -> Option<FromOverseer<Self::Message>> {
+		if self.0.fetch_add(1, Ordering::Relaxed) % 2 == 0 {
 			Some(msg)
 		} else {
 			None
@@ -73,11 +86,16 @@ impl OverseerGen for BehaveMaleficient {
 		let leaves = args.leaves.clone();
 		let runtime_client = args.runtime_client.clone();
 		let registry = args.registry.clone();
-
+		let candidate_validation_config = args.candidate_validation_config.clone();
 		// modify the subsystem(s) as needed:
-		let all_subsystems = create_default_subsystems(args)?.replace_candidate_validation(
-			FilteredSubsystem::new(args.candidate_validation, Skippy::default()),
-		);
+		let all_subsystems = create_default_subsystems(args)?
+			.replace_candidate_validation(
+				// create the filtered subsystem
+				FilteredSubsystem::new(CandidateValidationSubsystem::from_config(
+					candidate_validation_config,
+					Metrics::register(registry)?,
+				), Skippy::default()),
+			);
 
 		Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
 			.map_err(|e| e.into())