Do not run unneeded subsystems on collator and its alongside node (#3061)

Currently, collators and their alongside nodes spin up a full-scale
overseer running a bunch of subsystems that are not needed if the node
is not a validator. That was considered to be harmless; however, we've
got problems with unused subsystems getting stalled for a reason not
currently known, resulting in the overseer exiting and bringing down the
whole node.

This PR aims to only run needed subsystems on such nodes, replacing the
rest with `DummySubsystem`.

It also enables collator-optimized availability recovery subsystem
implementation.

Partially solves #1730.

polkadot/node/service/src/overseer.rs

@@ -16,6 +16,7 @@
 
 use super::{AuthorityDiscoveryApi, Block, Error, Hash, IsParachainNode, Registry};
 use polkadot_node_subsystem_types::DefaultSubsystemClient;
+use polkadot_overseer::{DummySubsystem, InitializedOverseerBuilder, SubsystemError};
 use sc_transaction_pool_api::OffchainTransactionPoolFactory;
 use sp_core::traits::SpawnNamed;
 
@@ -32,13 +33,10 @@ use polkadot_node_network_protocol::{
 	},
 };
 #[cfg(any(feature = "malus", test))]
-pub use polkadot_overseer::{
-	dummy::{dummy_overseer_builder, DummySubsystem},
-	HeadSupportsParachains,
-};
+pub use polkadot_overseer::{dummy::dummy_overseer_builder, HeadSupportsParachains};
 use polkadot_overseer::{
-	metrics::Metrics as OverseerMetrics, InitializedOverseerBuilder, MetricsTrait, Overseer,
-	OverseerConnector, OverseerHandle, SpawnGlue,
+	metrics::Metrics as OverseerMetrics, MetricsTrait, Overseer, OverseerConnector, OverseerHandle,
+	SpawnGlue,
 };
 
 use parking_lot::Mutex;
@@ -86,22 +84,14 @@ where
 	RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,
 	Spawner: 'static + SpawnNamed + Clone + Unpin,
 {
-	/// The keystore to use for i.e. validator keys.
-	pub keystore: Arc<LocalKeystore>,
 	/// Runtime client generic, providing the `ProvieRuntimeApi` trait besides others.
 	pub runtime_client: Arc<RuntimeClient>,
-	/// The underlying key value store for the parachains.
-	pub parachains_db: Arc<dyn polkadot_node_subsystem_util::database::Database>,
 	/// Underlying network service implementation.
 	pub network_service: Arc<sc_network::NetworkService<Block, Hash>>,
 	/// Underlying syncing service implementation.
 	pub sync_service: Arc<sc_network_sync::SyncingService<Block>>,
 	/// Underlying authority discovery service.
 	pub authority_discovery_service: AuthorityDiscoveryService,
-	/// POV request receiver.
-	pub pov_req_receiver: IncomingRequestReceiver<request_v1::PoVFetchingRequest>,
-	/// Erasure chunks request receiver.
-	pub chunk_req_receiver: IncomingRequestReceiver<request_v1::ChunkFetchingRequest>,
 	/// Collations request receiver for network protocol v1.
 	pub collation_req_v1_receiver: IncomingRequestReceiver<request_v1::CollationFetchingRequest>,
 	/// Collations request receiver for network protocol v2.
@@ -109,30 +99,12 @@ where
 	/// Receiver for available data requests.
 	pub available_data_req_receiver:
 		IncomingRequestReceiver<request_v1::AvailableDataFetchingRequest>,
-	/// Receiver for incoming large statement requests.
-	pub statement_req_receiver: IncomingRequestReceiver<request_v1::StatementFetchingRequest>,
-	/// Receiver for incoming candidate requests.
-	pub candidate_req_v2_receiver: IncomingRequestReceiver<request_v2::AttestedCandidateRequest>,
-	/// Receiver for incoming disputes.
-	pub dispute_req_receiver: IncomingRequestReceiver<request_v1::DisputeRequest>,
 	/// Prometheus registry, commonly used for production systems, less so for test.
 	pub registry: Option<&'a Registry>,
 	/// Task spawner to be used throughout the overseer and the APIs it provides.
 	pub spawner: Spawner,
 	/// Determines the behavior of the collator.
 	pub is_parachain_node: IsParachainNode,
-	/// Configuration for the approval voting subsystem.
-	pub approval_voting_config: ApprovalVotingConfig,
-	/// Configuration for the availability store subsystem.
-	pub availability_config: AvailabilityConfig,
-	/// Configuration for the candidate validation subsystem.
-	pub candidate_validation_config: Option<CandidateValidationConfig>,
-	/// Configuration for the chain selection subsystem.
-	pub chain_selection_config: ChainSelectionConfig,
-	/// Configuration for the dispute coordinator subsystem.
-	pub dispute_coordinator_config: DisputeCoordinatorConfig,
-	/// Enable PVF pre-checking
-	pub pvf_checker_enabled: bool,
 	/// Overseer channel capacity override.
 	pub overseer_message_channel_capacity_override: Option<usize>,
 	/// Request-response protocol names source.
@@ -145,39 +117,66 @@ where
 	pub notification_services: HashMap<PeerSet, Box<dyn NotificationService>>,
 }
 
-/// Obtain a prepared `OverseerBuilder`, that is initialized
-/// with all default values.
-pub fn prepared_overseer_builder<Spawner, RuntimeClient>(
+pub struct ExtendedOverseerGenArgs {
+	/// The keystore to use for i.e. validator keys.
+	pub keystore: Arc<LocalKeystore>,
+	/// The underlying key value store for the parachains.
+	pub parachains_db: Arc<dyn polkadot_node_subsystem_util::database::Database>,
+	/// Configuration for the candidate validation subsystem.
+	pub candidate_validation_config: Option<CandidateValidationConfig>,
+	/// Configuration for the availability store subsystem.
+	pub availability_config: AvailabilityConfig,
+	/// POV request receiver.
+	pub pov_req_receiver: IncomingRequestReceiver<request_v1::PoVFetchingRequest>,
+	/// Erasure chunks request receiver.
+	pub chunk_req_receiver: IncomingRequestReceiver<request_v1::ChunkFetchingRequest>,
+	/// Receiver for incoming large statement requests.
+	pub statement_req_receiver: IncomingRequestReceiver<request_v1::StatementFetchingRequest>,
+	/// Receiver for incoming candidate requests.
+	pub candidate_req_v2_receiver: IncomingRequestReceiver<request_v2::AttestedCandidateRequest>,
+	/// Configuration for the approval voting subsystem.
+	pub approval_voting_config: ApprovalVotingConfig,
+	/// Receiver for incoming disputes.
+	pub dispute_req_receiver: IncomingRequestReceiver<request_v1::DisputeRequest>,
+	/// Configuration for the dispute coordinator subsystem.
+	pub dispute_coordinator_config: DisputeCoordinatorConfig,
+	/// Configuration for the chain selection subsystem.
+	pub chain_selection_config: ChainSelectionConfig,
+}
+
+/// Obtain a prepared validator `Overseer`, that is initialized with all default values.
+pub fn validator_overseer_builder<Spawner, RuntimeClient>(
 	OverseerGenArgs {
-		keystore,
 		runtime_client,
-		parachains_db,
 		network_service,
 		sync_service,
 		authority_discovery_service,
-		pov_req_receiver,
-		chunk_req_receiver,
-		collation_req_v1_receiver,
-		collation_req_v2_receiver,
+		collation_req_v1_receiver: _,
+		collation_req_v2_receiver: _,
 		available_data_req_receiver,
-		statement_req_receiver,
-		candidate_req_v2_receiver,
-		dispute_req_receiver,
 		registry,
 		spawner,
 		is_parachain_node,
-		approval_voting_config,
-		availability_config,
-		candidate_validation_config,
-		chain_selection_config,
-		dispute_coordinator_config,
-		pvf_checker_enabled,
 		overseer_message_channel_capacity_override,
 		req_protocol_names,
 		peerset_protocol_names,
 		offchain_transaction_pool_factory,
 		notification_services,
 	}: OverseerGenArgs<Spawner, RuntimeClient>,
+	ExtendedOverseerGenArgs {
+		keystore,
+		parachains_db,
+		candidate_validation_config,
+		availability_config,
+		pov_req_receiver,
+		chunk_req_receiver,
+		statement_req_receiver,
+		candidate_req_v2_receiver,
+		approval_voting_config,
+		dispute_req_receiver,
+		dispute_coordinator_config,
+		chain_selection_config,
+	}: ExtendedOverseerGenArgs,
 ) -> Result<
 	InitializedOverseerBuilder<
 		SpawnGlue<Spawner>,
@@ -280,23 +279,15 @@ where
 			Metrics::register(registry)?, // candidate-validation metrics
 			Metrics::register(registry)?, // validation host metrics
 		))
-		.pvf_checker(PvfCheckerSubsystem::new(
-			pvf_checker_enabled,
-			keystore.clone(),
-			Metrics::register(registry)?,
-		))
+		.pvf_checker(PvfCheckerSubsystem::new(keystore.clone(), Metrics::register(registry)?))
 		.chain_api(ChainApiSubsystem::new(runtime_client.clone(), Metrics::register(registry)?))
 		.collation_generation(CollationGenerationSubsystem::new(Metrics::register(registry)?))
 		.collator_protocol({
 			let side = match is_parachain_node {
-				IsParachainNode::Collator(collator_pair) => ProtocolSide::Collator {
-					peer_id: network_service.local_peer_id(),
-					collator_pair,
-					request_receiver_v1: collation_req_v1_receiver,
-					request_receiver_v2: collation_req_v2_receiver,
-					metrics: Metrics::register(registry)?,
-				},
-				IsParachainNode::FullNode => ProtocolSide::None,
+				IsParachainNode::Collator(_) | IsParachainNode::FullNode =>
+					return Err(Error::Overseer(SubsystemError::Context(
+						"build validator overseer for parachain node".to_owned(),
+					))),
 				IsParachainNode::No => ProtocolSide::Validator {
 					keystore: keystore.clone(),
 					eviction_policy: Default::default(),
@@ -352,23 +343,173 @@ where
 		.metrics(metrics)
 		.spawner(spawner);
 
-	if let Some(capacity) = overseer_message_channel_capacity_override {
-		Ok(builder.message_channel_capacity(capacity))
+	let builder = if let Some(capacity) = overseer_message_channel_capacity_override {
+		builder.message_channel_capacity(capacity)
 	} else {
-		Ok(builder)
-	}
+		builder
+	};
+	Ok(builder)
+}
+
+/// Obtain a prepared collator `Overseer`, that is initialized with all default values.
+pub fn collator_overseer_builder<Spawner, RuntimeClient>(
+	OverseerGenArgs {
+		runtime_client,
+		network_service,
+		sync_service,
+		authority_discovery_service,
+		collation_req_v1_receiver,
+		collation_req_v2_receiver,
+		available_data_req_receiver,
+		registry,
+		spawner,
+		is_parachain_node,
+		overseer_message_channel_capacity_override,
+		req_protocol_names,
+		peerset_protocol_names,
+		offchain_transaction_pool_factory,
+		notification_services,
+	}: OverseerGenArgs<Spawner, RuntimeClient>,
+) -> Result<
+	InitializedOverseerBuilder<
+		SpawnGlue<Spawner>,
+		Arc<DefaultSubsystemClient<RuntimeClient>>,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		AvailabilityRecoverySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		RuntimeApiSubsystem<DefaultSubsystemClient<RuntimeClient>>,
+		DummySubsystem,
+		NetworkBridgeRxSubsystem<
+			Arc<sc_network::NetworkService<Block, Hash>>,
+			AuthorityDiscoveryService,
+		>,
+		NetworkBridgeTxSubsystem<
+			Arc<sc_network::NetworkService<Block, Hash>>,
+			AuthorityDiscoveryService,
+		>,
+		ChainApiSubsystem<RuntimeClient>,
+		CollationGenerationSubsystem,
+		CollatorProtocolSubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		DummySubsystem,
+		ProspectiveParachainsSubsystem,
+	>,
+	Error,
+>
+where
+	RuntimeClient: 'static + ProvideRuntimeApi<Block> + HeaderBackend<Block> + AuxStore,
+	RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,
+	Spawner: 'static + SpawnNamed + Clone + Unpin,
+{
+	use polkadot_node_subsystem_util::metrics::Metrics;
+
+	let metrics = <OverseerMetrics as MetricsTrait>::register(registry)?;
+	let notification_sinks = Arc::new(Mutex::new(HashMap::new()));
+
+	let spawner = SpawnGlue(spawner);
+
+	let network_bridge_metrics: NetworkBridgeMetrics = Metrics::register(registry)?;
+
+	let runtime_api_client = Arc::new(DefaultSubsystemClient::new(
+		runtime_client.clone(),
+		offchain_transaction_pool_factory,
+	));
+
+	let builder = Overseer::builder()
+		.network_bridge_tx(NetworkBridgeTxSubsystem::new(
+			network_service.clone(),
+			authority_discovery_service.clone(),
+			network_bridge_metrics.clone(),
+			req_protocol_names,
+			peerset_protocol_names.clone(),
+			notification_sinks.clone(),
+		))
+		.network_bridge_rx(NetworkBridgeRxSubsystem::new(
+			network_service.clone(),
+			authority_discovery_service.clone(),
+			Box::new(sync_service.clone()),
+			network_bridge_metrics,
+			peerset_protocol_names,
+			notification_services,
+			notification_sinks,
+		))
+		.availability_distribution(DummySubsystem)
+		.availability_recovery(AvailabilityRecoverySubsystem::for_collator(
+			available_data_req_receiver,
+			Metrics::register(registry)?,
+		))
+		.availability_store(DummySubsystem)
+		.bitfield_distribution(DummySubsystem)
+		.bitfield_signing(DummySubsystem)
+		.candidate_backing(DummySubsystem)
+		.candidate_validation(DummySubsystem)
+		.pvf_checker(DummySubsystem)
+		.chain_api(ChainApiSubsystem::new(runtime_client.clone(), Metrics::register(registry)?))
+		.collation_generation(CollationGenerationSubsystem::new(Metrics::register(registry)?))
+		.collator_protocol({
+			let side = match is_parachain_node {
+				IsParachainNode::No =>
+					return Err(Error::Overseer(SubsystemError::Context(
+						"build parachain node overseer for validator".to_owned(),
+					))),
+				IsParachainNode::Collator(collator_pair) => ProtocolSide::Collator {
+					peer_id: network_service.local_peer_id(),
+					collator_pair,
+					request_receiver_v1: collation_req_v1_receiver,
+					request_receiver_v2: collation_req_v2_receiver,
+					metrics: Metrics::register(registry)?,
+				},
+				IsParachainNode::FullNode => ProtocolSide::None,
+			};
+			CollatorProtocolSubsystem::new(side)
+		})
+		.provisioner(DummySubsystem)
+		.runtime_api(RuntimeApiSubsystem::new(
+			runtime_api_client.clone(),
+			Metrics::register(registry)?,
+			spawner.clone(),
+		))
+		.statement_distribution(DummySubsystem)
+		.approval_distribution(DummySubsystem)
+		.approval_voting(DummySubsystem)
+		.gossip_support(DummySubsystem)
+		.dispute_coordinator(DummySubsystem)
+		.dispute_distribution(DummySubsystem)
+		.chain_selection(DummySubsystem)
+		.prospective_parachains(ProspectiveParachainsSubsystem::new(Metrics::register(registry)?))
+		.activation_external_listeners(Default::default())
+		.span_per_active_leaf(Default::default())
+		.active_leaves(Default::default())
+		.supports_parachains(runtime_api_client)
+		.metrics(metrics)
+		.spawner(spawner);
+
+	let builder = if let Some(capacity) = overseer_message_channel_capacity_override {
+		builder.message_channel_capacity(capacity)
+	} else {
+		builder
+	};
+	Ok(builder)
 }
 
 /// Trait for the `fn` generating the overseer.
-///
-/// Default behavior is to create an unmodified overseer, as `RealOverseerGen`
-/// would do.
 pub trait OverseerGen {
 	/// Overwrite the full generation of the overseer, including the subsystems.
 	fn generate<Spawner, RuntimeClient>(
 		&self,
 		connector: OverseerConnector,
 		args: OverseerGenArgs<Spawner, RuntimeClient>,
+		ext_args: Option<ExtendedOverseerGenArgs>,
 	) -> Result<
 		(Overseer<SpawnGlue<Spawner>, Arc<DefaultSubsystemClient<RuntimeClient>>>, OverseerHandle),
 		Error,
@@ -376,24 +517,22 @@ pub trait OverseerGen {
 	where
 		RuntimeClient: 'static + ProvideRuntimeApi<Block> + HeaderBackend<Block> + AuxStore,
 		RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,
-		Spawner: 'static + SpawnNamed + Clone + Unpin,
-	{
-		let gen = RealOverseerGen;
-		RealOverseerGen::generate::<Spawner, RuntimeClient>(&gen, connector, args)
-	}
+		Spawner: 'static + SpawnNamed + Clone + Unpin;
+
 	// It would be nice to make `create_subsystems` part of this trait,
 	// but the amount of generic arguments that would be required as
 	// as consequence make this rather annoying to implement and use.
 }
 
 /// The regular set of subsystems.
-pub struct RealOverseerGen;
+pub struct ValidatorOverseerGen;
 
-impl OverseerGen for RealOverseerGen {
+impl OverseerGen for ValidatorOverseerGen {
 	fn generate<Spawner, RuntimeClient>(
 		&self,
 		connector: OverseerConnector,
 		args: OverseerGenArgs<Spawner, RuntimeClient>,
+		ext_args: Option<ExtendedOverseerGenArgs>,
 	) -> Result<
 		(Overseer<SpawnGlue<Spawner>, Arc<DefaultSubsystemClient<RuntimeClient>>>, OverseerHandle),
 		Error,
@@ -403,7 +542,35 @@ impl OverseerGen for RealOverseerGen {
 		RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,
 		Spawner: 'static + SpawnNamed + Clone + Unpin,
 	{
-		prepared_overseer_builder(args)?
+		let ext_args = ext_args.ok_or(Error::Overseer(SubsystemError::Context(
+			"create validator overseer as mandatory extended arguments were not provided"
+				.to_owned(),
+		)))?;
+		validator_overseer_builder(args, ext_args)?
+			.build_with_connector(connector)
+			.map_err(|e| e.into())
+	}
+}
+
+/// Reduced set of subsystems, to use in collator and collator's full node.
+pub struct CollatorOverseerGen;
+
+impl OverseerGen for CollatorOverseerGen {
+	fn generate<Spawner, RuntimeClient>(
+		&self,
+		connector: OverseerConnector,
+		args: OverseerGenArgs<Spawner, RuntimeClient>,
+		_ext_args: Option<ExtendedOverseerGenArgs>,
+	) -> Result<
+		(Overseer<SpawnGlue<Spawner>, Arc<DefaultSubsystemClient<RuntimeClient>>>, OverseerHandle),
+		Error,
+	>
+	where
+		RuntimeClient: 'static + ProvideRuntimeApi<Block> + HeaderBackend<Block> + AuxStore,
+		RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,
+		Spawner: 'static + SpawnNamed + Clone + Unpin,
+	{
+		collator_overseer_builder(args)?
 			.build_with_connector(connector)
 			.map_err(|e| e.into())
 	}