refactor+feat: allow subsystems to send only declared messages, generate graphviz (#5314)

Closes #3774
Closes #3826

polkadot/node/subsystem-util/src/determine_new_blocks.rs

@@ -39,7 +39,7 @@ pub async fn determine_new_blocks<E, Sender>(
 	lower_bound_number: BlockNumber,
 ) -> Result<Vec<(Hash, Header)>, E>
 where
-	Sender: SubsystemSender,
+	Sender: SubsystemSender<ChainApiMessage>,
 {
 	const ANCESTRY_STEP: usize = 4;
 

polkadot/node/subsystem-util/src/lib.rs

@@ -26,9 +26,7 @@
 
 use polkadot_node_subsystem::{
 	errors::{RuntimeApiError, SubsystemError},
-	messages::{
-		AllMessages, BoundToRelayParent, RuntimeApiMessage, RuntimeApiRequest, RuntimeApiSender,
-	},
+	messages::{BoundToRelayParent, RuntimeApiMessage, RuntimeApiRequest, RuntimeApiSender},
 	overseer, ActivatedLeaf, ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem,
 	SubsystemContext, SubsystemSender,
 };
@@ -144,7 +142,7 @@ pub async fn request_from_runtime<RequestBuilder, Response, Sender>(
 ) -> RuntimeApiReceiver<Response>
 where
 	RequestBuilder: FnOnce(RuntimeApiSender<Response>) -> RuntimeApiRequest,
-	Sender: SubsystemSender,
+	Sender: SubsystemSender<RuntimeApiMessage>,
 {
 	let (tx, rx) = oneshot::channel();
 
@@ -176,7 +174,7 @@ macro_rules! specialize_requests {
 			$(
 				$param_name: $param_ty,
 			)*
-			sender: &mut impl SubsystemSender,
+			sender: &mut impl overseer::SubsystemSender<RuntimeApiMessage>,
 		) -> RuntimeApiReceiver<$return_ty>
 		{
 			request_from_runtime(parent, sender, |tx| RuntimeApiRequest::$request_variant(
@@ -329,11 +327,14 @@ pub struct Validator {
 
 impl Validator {
 	/// Get a struct representing this node's validator if this node is in fact a validator in the context of the given block.
-	pub async fn new(
+	pub async fn new<S>(
 		parent: Hash,
 		keystore: SyncCryptoStorePtr,
-		sender: &mut impl SubsystemSender,
-	) -> Result<Self, Error> {
+		sender: &mut S,
+	) -> Result<Self, Error>
+	where
+		S: SubsystemSender<RuntimeApiMessage>,
+	{
 		// Note: request_validators and request_session_index_for_child do not and cannot
 		// run concurrently: they both have a mutable handle to the same sender.
 		// However, each of them returns a oneshot::Receiver, and those are resolved concurrently.
@@ -397,14 +398,14 @@ impl Drop for AbortOnDrop {
 }
 
 /// A `JobHandle` manages a particular job for a subsystem.
-struct JobHandle<ToJob> {
+struct JobHandle<Consumes> {
 	_abort_handle: AbortOnDrop,
-	to_job: mpsc::Sender<ToJob>,
+	to_job: mpsc::Sender<Consumes>,
 }
 
-impl<ToJob> JobHandle<ToJob> {
+impl<Consumes> JobHandle<Consumes> {
 	/// Send a message to the job.
-	async fn send_msg(&mut self, msg: ToJob) -> Result<(), Error> {
+	async fn send_msg(&mut self, msg: Consumes) -> Result<(), Error> {
 		self.to_job.send(msg).await.map_err(Into::into)
 	}
 }
@@ -418,49 +419,25 @@ pub enum FromJobCommand {
 }
 
 /// A sender for messages from jobs, as well as commands to the overseer.
-pub struct JobSender<S: SubsystemSender> {
+pub struct JobSender<S> {
 	sender: S,
 	from_job: mpsc::Sender<FromJobCommand>,
 }
 
 // A custom clone impl, since M does not need to impl `Clone`
 // which `#[derive(Clone)]` requires.
-impl<S: SubsystemSender> Clone for JobSender<S> {
+impl<S: Clone> Clone for JobSender<S> {
 	fn clone(&self) -> Self {
 		Self { sender: self.sender.clone(), from_job: self.from_job.clone() }
 	}
 }
 
-impl<S: SubsystemSender> JobSender<S> {
+impl<S> JobSender<S> {
 	/// Get access to the underlying subsystem sender.
 	pub fn subsystem_sender(&mut self) -> &mut S {
 		&mut self.sender
 	}
 
-	/// Send a direct message to some other `Subsystem`, routed based on message type.
-	pub async fn send_message(&mut self, msg: impl Into<AllMessages>) {
-		self.sender.send_message(msg.into()).await
-	}
-
-	/// Send multiple direct messages to other `Subsystem`s, routed based on message type.
-	pub async fn send_messages<T, M>(&mut self, msgs: T)
-	where
-		T: IntoIterator<Item = M> + Send,
-		T::IntoIter: Send,
-		M: Into<AllMessages>,
-	{
-		self.sender.send_messages(msgs.into_iter().map(|m| m.into())).await
-	}
-
-	/// Send a message onto the unbounded queue of some other `Subsystem`, routed based on message
-	/// type.
-	///
-	/// This function should be used only when there is some other bounding factor on the messages
-	/// sent with it. Otherwise, it risks a memory leak.
-	pub fn send_unbounded_message(&mut self, msg: impl Into<AllMessages>) {
-		self.sender.send_unbounded_message(msg.into())
-	}
-
 	/// Send a command to the subsystem, to be relayed onwards to the overseer.
 	pub async fn send_command(&mut self, msg: FromJobCommand) -> Result<(), mpsc::SendError> {
 		self.from_job.send(msg).await
@@ -470,23 +447,23 @@ impl<S: SubsystemSender> JobSender<S> {
 #[async_trait::async_trait]
 impl<S, M> overseer::SubsystemSender<M> for JobSender<S>
 where
-	M: Send + 'static + Into<AllMessages>,
-	S: SubsystemSender + Clone,
+	M: Send + 'static,
+	S: SubsystemSender<M> + Clone,
 {
 	async fn send_message(&mut self, msg: M) {
-		self.sender.send_message(msg.into()).await
+		self.sender.send_message(msg).await
 	}
 
-	async fn send_messages<T>(&mut self, msgs: T)
+	async fn send_messages<I>(&mut self, msgs: I)
 	where
-		T: IntoIterator<Item = M> + Send,
-		T::IntoIter: Send,
+		I: IntoIterator<Item = M> + Send,
+		I::IntoIter: Send,
 	{
-		self.sender.send_messages(msgs.into_iter().map(|m| m.into())).await
+		self.sender.send_messages(msgs).await
 	}
 
 	fn send_unbounded_message(&mut self, msg: M) {
-		self.sender.send_unbounded_message(msg.into())
+		self.sender.send_unbounded_message(msg)
 	}
 }
 
@@ -506,6 +483,14 @@ impl fmt::Debug for FromJobCommand {
 pub trait JobTrait: Unpin + Sized {
 	/// Message type used to send messages to the job.
 	type ToJob: 'static + BoundToRelayParent + Send;
+
+	/// The set of outgoing messages to be accumulated into.
+	type OutgoingMessages: 'static + Send;
+
+	/// The sender to send outgoing messages.
+	// The trait bounds are rather minimal.
+	type Sender: 'static + Send + Clone;
+
 	/// Job runtime error.
 	type Error: 'static + std::error::Error + Send;
 	/// Extra arguments this job needs to run properly.
@@ -525,12 +510,12 @@ pub trait JobTrait: Unpin + Sized {
 	/// Run a job for the given relay `parent`.
 	///
 	/// The job should be ended when `receiver` returns `None`.
-	fn run<S: SubsystemSender>(
+	fn run(
 		leaf: ActivatedLeaf,
 		run_args: Self::RunArgs,
 		metrics: Self::Metrics,
 		receiver: mpsc::Receiver<Self::ToJob>,
-		sender: JobSender<S>,
+		sender: JobSender<Self::Sender>,
 	) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send>>;
 }
 
@@ -572,15 +557,14 @@ where
 	}
 
 	/// Spawn a new job for this `parent_hash`, with whatever args are appropriate.
-	fn spawn_job<Job, Sender>(
+	fn spawn_job<Job>(
 		&mut self,
 		leaf: ActivatedLeaf,
 		run_args: Job::RunArgs,
 		metrics: Job::Metrics,
-		sender: Sender,
+		sender: Job::Sender,
 	) where
 		Job: JobTrait<ToJob = ToJob>,
-		Sender: SubsystemSender,
 	{
 		let hash = leaf.hash;
 		let (to_job_tx, to_job_rx) = mpsc::channel(JOB_CHANNEL_CAPACITY);
@@ -697,8 +681,12 @@ impl<Job: JobTrait, Spawner> JobSubsystem<Job, Spawner> {
 	pub async fn run<Context>(self, mut ctx: Context)
 	where
 		Spawner: SpawnNamed + Send + Clone + Unpin + 'static,
-		Context: SubsystemContext<Message = <Job as JobTrait>::ToJob, Signal = OverseerSignal>,
-		<Context as SubsystemContext>::Sender: SubsystemSender,
+		Context: SubsystemContext<
+			Message = <Job as JobTrait>::ToJob,
+			OutgoingMessages = <Job as JobTrait>::OutgoingMessages,
+			Sender = <Job as JobTrait>::Sender,
+			Signal = OverseerSignal,
+		>,
 		Job: 'static + JobTrait + Send,
 		<Job as JobTrait>::RunArgs: Clone + Sync,
 		<Job as JobTrait>::ToJob:
@@ -719,7 +707,7 @@ impl<Job: JobTrait, Spawner> JobSubsystem<Job, Spawner> {
 						}))) => {
 							for activated in activated {
 								let sender = ctx.sender().clone();
-								jobs.spawn_job::<Job, _>(
+								jobs.spawn_job::<Job>(
 									activated,
 									run_args.clone(),
 									metrics.clone(),
@@ -773,11 +761,15 @@ impl<Job: JobTrait, Spawner> JobSubsystem<Job, Spawner> {
 impl<Context, Job, Spawner> Subsystem<Context, SubsystemError> for JobSubsystem<Job, Spawner>
 where
 	Spawner: SpawnNamed + Send + Clone + Unpin + 'static,
-	Context: SubsystemContext<Message = Job::ToJob, Signal = OverseerSignal>,
+	Context: SubsystemContext<
+		Message = Job::ToJob,
+		Signal = OverseerSignal,
+		OutgoingMessages = <Job as JobTrait>::OutgoingMessages,
+		Sender = <Job as JobTrait>::Sender,
+	>,
 	Job: 'static + JobTrait + Send,
 	Job::RunArgs: Clone + Sync,
-	<Job as JobTrait>::ToJob:
-		Sync + From<<Context as polkadot_overseer::SubsystemContext>::Message>,
+	<Job as JobTrait>::ToJob: Sync + From<<Context as SubsystemContext>::Message>,
 	Job::Metrics: Sync,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {

polkadot/node/subsystem-util/src/rolling_session_window.rs

@@ -26,7 +26,7 @@ use futures::channel::oneshot;
 use polkadot_node_subsystem::{
 	errors::RuntimeApiError,
 	messages::{RuntimeApiMessage, RuntimeApiRequest},
-	overseer, SubsystemContext,
+	overseer,
 };
 use thiserror::Error;
 
@@ -94,16 +94,19 @@ pub struct RollingSessionWindow {
 
 impl RollingSessionWindow {
 	/// Initialize a new session info cache with the given window size.
-	pub async fn new(
-		ctx: &mut (impl SubsystemContext + overseer::SubsystemContext),
+	pub async fn new<Sender>(
+		mut sender: Sender,
 		window_size: SessionWindowSize,
 		block_hash: Hash,
-	) -> Result<Self, SessionsUnavailable> {
-		let session_index = get_session_index_for_child(ctx, block_hash).await?;
+	) -> Result<Self, SessionsUnavailable>
+	where
+		Sender: overseer::SubsystemSender<RuntimeApiMessage>,
+	{
+		let session_index = get_session_index_for_child(&mut sender, block_hash).await?;
 
 		let window_start = session_index.saturating_sub(window_size.get() - 1);
 
-		match load_all_sessions(ctx, block_hash, window_start, session_index).await {
+		match load_all_sessions(&mut sender, block_hash, window_start, session_index).await {
 			Err(kind) => Err(SessionsUnavailable {
 				kind,
 				info: Some(SessionsUnavailableInfo {
@@ -154,10 +157,10 @@ impl RollingSessionWindow {
 	/// some backwards drift in session index is acceptable.
 	pub async fn cache_session_info_for_head(
 		&mut self,
-		ctx: &mut (impl SubsystemContext + overseer::SubsystemContext),
+		sender: &mut impl overseer::SubsystemSender<RuntimeApiMessage>,
 		block_hash: Hash,
 	) -> Result<SessionWindowUpdate, SessionsUnavailable> {
-		let session_index = get_session_index_for_child(ctx, block_hash).await?;
+		let session_index = get_session_index_for_child(sender, block_hash).await?;
 
 		let old_window_start = self.earliest_session;
 
@@ -177,7 +180,7 @@ impl RollingSessionWindow {
 
 		let fresh_start = if latest < window_start { window_start } else { latest + 1 };
 
-		match load_all_sessions(ctx, block_hash, fresh_start, session_index).await {
+		match load_all_sessions(sender, block_hash, fresh_start, session_index).await {
 			Err(kind) => Err(SessionsUnavailable {
 				kind,
 				info: Some(SessionsUnavailableInfo {
@@ -215,17 +218,18 @@ impl RollingSessionWindow {
 // cleaner to just call the runtime API directly without needing to create an instance
 // of `RuntimeInfo`.
 async fn get_session_index_for_child(
-	ctx: &mut (impl SubsystemContext + overseer::SubsystemContext),
+	sender: &mut impl overseer::SubsystemSender<RuntimeApiMessage>,
 	block_hash: Hash,
 ) -> Result<SessionIndex, SessionsUnavailable> {
 	let (s_tx, s_rx) = oneshot::channel();
 
 	// We're requesting session index of a child to populate the cache in advance.
-	ctx.send_message(RuntimeApiMessage::Request(
-		block_hash,
-		RuntimeApiRequest::SessionIndexForChild(s_tx),
-	))
-	.await;
+	sender
+		.send_message(RuntimeApiMessage::Request(
+			block_hash,
+			RuntimeApiRequest::SessionIndexForChild(s_tx),
+		))
+		.await;
 
 	match s_rx.await {
 		Ok(Ok(s)) => Ok(s),
@@ -243,7 +247,7 @@ async fn get_session_index_for_child(
 }
 
 async fn load_all_sessions(
-	ctx: &mut (impl SubsystemContext + overseer::SubsystemContext),
+	sender: &mut impl overseer::SubsystemSender<RuntimeApiMessage>,
 	block_hash: Hash,
 	start: SessionIndex,
 	end_inclusive: SessionIndex,
@@ -251,11 +255,12 @@ async fn load_all_sessions(
 	let mut v = Vec::new();
 	for i in start..=end_inclusive {
 		let (tx, rx) = oneshot::channel();
-		ctx.send_message(RuntimeApiMessage::Request(
-			block_hash,
-			RuntimeApiRequest::SessionInfo(i, tx),
-		))
-		.await;
+		sender
+			.send_message(RuntimeApiMessage::Request(
+				block_hash,
+				RuntimeApiRequest::SessionInfo(i, tx),
+			))
+			.await;
 
 		let session_info = match rx.await {
 			Ok(Ok(Some(s))) => s,
@@ -274,7 +279,10 @@ async fn load_all_sessions(
 mod tests {
 	use super::*;
 	use assert_matches::assert_matches;
-	use polkadot_node_subsystem::messages::{AllMessages, AvailabilityRecoveryMessage};
+	use polkadot_node_subsystem::{
+		messages::{AllMessages, AvailabilityRecoveryMessage},
+		SubsystemContext,
+	};
 	use polkadot_node_subsystem_test_helpers::make_subsystem_context;
 	use polkadot_primitives::v2::Header;
 	use sp_core::testing::TaskExecutor;
@@ -319,13 +327,16 @@ mod tests {
 
 		let hash = header.hash();
 
+		let sender = ctx.sender();
+
 		let test_fut = {
 			Box::pin(async move {
 				let window = match window {
-					None =>
-						RollingSessionWindow::new(&mut ctx, TEST_WINDOW_SIZE, hash).await.unwrap(),
+					None => RollingSessionWindow::new(sender.clone(), TEST_WINDOW_SIZE, hash)
+						.await
+						.unwrap(),
 					Some(mut window) => {
-						window.cache_session_info_for_head(&mut ctx, hash).await.unwrap();
+						window.cache_session_info_for_head(sender, hash).await.unwrap();
 						window
 					},
 				};
@@ -495,8 +506,9 @@ mod tests {
 		let hash = header.hash();
 
 		let test_fut = {
+			let sender = ctx.sender().clone();
 			Box::pin(async move {
-				let res = RollingSessionWindow::new(&mut ctx, TEST_WINDOW_SIZE, hash).await;
+				let res = RollingSessionWindow::new(sender, TEST_WINDOW_SIZE, hash).await;
 				assert!(res.is_err());
 			})
 		};
@@ -555,8 +567,9 @@ mod tests {
 
 		let test_fut = {
 			Box::pin(async move {
+				let sender = ctx.sender().clone();
 				let window =
-					RollingSessionWindow::new(&mut ctx, TEST_WINDOW_SIZE, hash).await.unwrap();
+					RollingSessionWindow::new(sender, TEST_WINDOW_SIZE, hash).await.unwrap();
 
 				assert_eq!(window.earliest_session, session);
 				assert_eq!(window.session_info, vec![dummy_session_info(session)]);

polkadot/node/subsystem-util/src/runtime/mod.rs

@@ -25,7 +25,7 @@ use sp_application_crypto::AppKey;
 use sp_core::crypto::ByteArray;
 use sp_keystore::{CryptoStore, SyncCryptoStorePtr};
 
-use polkadot_node_subsystem::{SubsystemContext, SubsystemSender};
+use polkadot_node_subsystem::{messages::RuntimeApiMessage, overseer, SubsystemSender};
 use polkadot_primitives::v2::{
 	CandidateEvent, CoreState, EncodeAs, GroupIndex, GroupRotationInfo, Hash, OccupiedCore,
 	ScrapedOnChainVotes, SessionIndex, SessionInfo, Signed, SigningContext, UncheckedSigned,
@@ -123,7 +123,7 @@ impl RuntimeInfo {
 		parent: Hash,
 	) -> Result<SessionIndex>
 	where
-		Sender: SubsystemSender,
+		Sender: SubsystemSender<RuntimeApiMessage>,
 	{
 		match self.session_index_cache.get(&parent) {
 			Some(index) => Ok(*index),
@@ -143,7 +143,7 @@ impl RuntimeInfo {
 		relay_parent: Hash,
 	) -> Result<&'a ExtendedSessionInfo>
 	where
-		Sender: SubsystemSender,
+		Sender: SubsystemSender<RuntimeApiMessage>,
 	{
 		let session_index = self.get_session_index_for_child(sender, relay_parent).await?;
 
@@ -161,7 +161,7 @@ impl RuntimeInfo {
 		session_index: SessionIndex,
 	) -> Result<&'a ExtendedSessionInfo>
 	where
-		Sender: SubsystemSender,
+		Sender: SubsystemSender<RuntimeApiMessage>,
 	{
 		if !self.session_info_cache.contains(&session_index) {
 			let session_info =
@@ -190,7 +190,7 @@ impl RuntimeInfo {
 		std::result::Result<Signed<Payload, RealPayload>, UncheckedSigned<Payload, RealPayload>>,
 	>
 	where
-		Sender: SubsystemSender,
+		Sender: SubsystemSender<RuntimeApiMessage>,
 		Payload: EncodeAs<RealPayload> + Clone,
 		RealPayload: Encode + Clone,
 	{
@@ -257,25 +257,25 @@ where
 }
 
 /// Request availability cores from the runtime.
-pub async fn get_availability_cores<Context>(
-	ctx: &mut Context,
+pub async fn get_availability_cores<Sender>(
+	sender: &mut Sender,
 	relay_parent: Hash,
 ) -> Result<Vec<CoreState>>
 where
-	Context: SubsystemContext,
+	Sender: overseer::SubsystemSender<RuntimeApiMessage>,
 {
-	recv_runtime(request_availability_cores(relay_parent, ctx.sender()).await).await
+	recv_runtime(request_availability_cores(relay_parent, sender).await).await
 }
 
 /// Variant of `request_availability_cores` that only returns occupied ones.
-pub async fn get_occupied_cores<Context>(
-	ctx: &mut Context,
+pub async fn get_occupied_cores<Sender>(
+	sender: &mut Sender,
 	relay_parent: Hash,
 ) -> Result<Vec<OccupiedCore>>
 where
-	Context: SubsystemContext,
+	Sender: overseer::SubsystemSender<RuntimeApiMessage>,
 {
-	let cores = get_availability_cores(ctx, relay_parent).await?;
+	let cores = get_availability_cores(sender, relay_parent).await?;
 
 	Ok(cores
 		.into_iter()
@@ -290,17 +290,16 @@ where
 }
 
 /// Get group rotation info based on the given `relay_parent`.
-pub async fn get_group_rotation_info<Context>(
-	ctx: &mut Context,
+pub async fn get_group_rotation_info<Sender>(
+	sender: &mut Sender,
 	relay_parent: Hash,
 ) -> Result<GroupRotationInfo>
 where
-	Context: SubsystemContext,
+	Sender: overseer::SubsystemSender<RuntimeApiMessage>,
 {
 	// We drop `groups` here as we don't need them, because of `RuntimeInfo`. Ideally we would not
 	// fetch them in the first place.
-	let (_, info) =
-		recv_runtime(request_validator_groups(relay_parent, ctx.sender()).await).await?;
+	let (_, info) = recv_runtime(request_validator_groups(relay_parent, sender).await).await?;
 	Ok(info)
 }
 
@@ -310,7 +309,7 @@ pub async fn get_candidate_events<Sender>(
 	relay_parent: Hash,
 ) -> Result<Vec<CandidateEvent>>
 where
-	Sender: SubsystemSender,
+	Sender: SubsystemSender<RuntimeApiMessage>,
 {
 	recv_runtime(request_candidate_events(relay_parent, sender).await).await
 }
@@ -321,7 +320,7 @@ pub async fn get_on_chain_votes<Sender>(
 	relay_parent: Hash,
 ) -> Result<Option<ScrapedOnChainVotes>>
 where
-	Sender: SubsystemSender,
+	Sender: SubsystemSender<RuntimeApiMessage>,
 {
 	recv_runtime(request_on_chain_votes(relay_parent, sender).await).await
 }
@@ -333,7 +332,7 @@ pub async fn get_validation_code_by_hash<Sender>(
 	validation_code_hash: ValidationCodeHash,
 ) -> Result<Option<ValidationCode>>
 where
-	Sender: SubsystemSender,
+	Sender: SubsystemSender<RuntimeApiMessage>,
 {
 	recv_runtime(request_validation_code_by_hash(relay_parent, validation_code_hash, sender).await)
 		.await

polkadot/node/subsystem-util/src/tests.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
+#![cfg(test)]
+
 use super::*;
 use assert_matches::assert_matches;
 use executor::block_on;
@@ -44,8 +46,9 @@ use thiserror::Error;
 
 // job structs are constructed within JobTrait::run
 // most will want to retain the sender and receiver, as well as whatever other data they like
-struct FakeCollatorProtocolJob {
+struct FakeCollatorProtocolJob<Sender> {
 	receiver: mpsc::Receiver<CollatorProtocolMessage>,
+	_phantom: std::marker::PhantomData<Sender>,
 }
 
 // Error will mostly be a wrapper to make the try operator more convenient;
@@ -57,8 +60,18 @@ enum Error {
 	Sending(#[from] mpsc::SendError),
 }
 
-impl JobTrait for FakeCollatorProtocolJob {
+impl<Sender> JobTrait for FakeCollatorProtocolJob<Sender>
+where
+	Sender: overseer::CollatorProtocolSenderTrait
+		+ std::marker::Unpin
+		+ overseer::SubsystemSender<CollatorProtocolMessage>,
+	JobSender<Sender>: overseer::CollatorProtocolSenderTrait
+		+ std::marker::Unpin
+		+ overseer::SubsystemSender<CollatorProtocolMessage>,
+{
 	type ToJob = CollatorProtocolMessage;
+	type OutgoingMessages = overseer::CollatorProtocolOutgoingMessages;
+	type Sender = Sender;
 	type Error = Error;
 	type RunArgs = bool;
 	type Metrics = ();
@@ -68,20 +81,21 @@ impl JobTrait for FakeCollatorProtocolJob {
 	/// Run a job for the parent block indicated
 	//
 	// this function is in charge of creating and executing the job's main loop
-	fn run<S: SubsystemSender>(
+	fn run(
 		_: ActivatedLeaf,
 		run_args: Self::RunArgs,
 		_metrics: Self::Metrics,
 		receiver: mpsc::Receiver<CollatorProtocolMessage>,
-		mut sender: JobSender<S>,
+		mut sender: JobSender<Sender>,
 	) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send>> {
 		async move {
-			let job = FakeCollatorProtocolJob { receiver };
+			let job =
+				FakeCollatorProtocolJob { receiver, _phantom: std::marker::PhantomData::<Sender> };
 
 			if run_args {
 				sender
 					.send_message(CollatorProtocolMessage::Invalid(
-						Default::default(),
+						dummy_hash(),
 						dummy_candidate_receipt(dummy_hash()),
 					))
 					.await;
@@ -95,7 +109,10 @@ impl JobTrait for FakeCollatorProtocolJob {
 	}
 }
 
-impl FakeCollatorProtocolJob {
+impl<Sender> FakeCollatorProtocolJob<Sender>
+where
+	Sender: overseer::CollatorProtocolSenderTrait,
+{
 	async fn run_loop(mut self) -> Result<(), Error> {
 		loop {
 			match self.receiver.next().await {
@@ -111,7 +128,8 @@ impl FakeCollatorProtocolJob {
 }
 
 // with the job defined, it's straightforward to get a subsystem implementation.
-type FakeCollatorProtocolSubsystem<Spawner> = JobSubsystem<FakeCollatorProtocolJob, Spawner>;
+type FakeCollatorProtocolSubsystem<Spawner> =
+	JobSubsystem<FakeCollatorProtocolJob<test_helpers::TestSubsystemSender>, Spawner>;
 
 // this type lets us pretend to be the overseer
 type OverseerHandle = test_helpers::TestSubsystemContextHandle<CollatorProtocolMessage>;