malus - mockable overseer mvp (#3224)

polkadot/node/service/src/lib.rs

@@ -22,6 +22,17 @@ pub mod chain_spec;
 mod grandpa_support;
 mod parachains_db;
 
+#[cfg(feature = "full-node")]
+mod overseer;
+
+#[cfg(feature = "full-node")]
+pub use self::overseer::{
+	OverseerGen,
+	OverseerGenArgs,
+	RealOverseerGen,
+	create_default_subsystems,
+};
+
 #[cfg(feature = "full-node")]
 use {
 	tracing::info,
@@ -30,22 +41,26 @@ use {
 	polkadot_node_core_av_store::Error as AvailabilityError,
 	polkadot_node_core_approval_voting::Config as ApprovalVotingConfig,
 	polkadot_node_core_candidate_validation::Config as CandidateValidationConfig,
-	polkadot_overseer::{AllSubsystems, BlockInfo, Overseer, OverseerHandler},
-	polkadot_primitives::v1::ParachainHost,
-	sc_authority_discovery::Service as AuthorityDiscoveryService,
-	sp_authority_discovery::AuthorityDiscoveryApi,
-	sp_blockchain::HeaderBackend,
+	polkadot_overseer::BlockInfo,
 	sp_trie::PrefixedMemoryDB,
-	sc_client_api::{AuxStore, ExecutorProvider},
-	sc_keystore::LocalKeystore,
-	sp_consensus_babe::BabeApi,
+	sc_client_api::ExecutorProvider,
 	grandpa::{self, FinalityProofProvider as GrandpaFinalityProofProvider},
 	beefy_primitives::ecdsa::AuthoritySignature as BeefySignature,
 	sp_runtime::traits::Header as HeaderT,
 };
 
-use sp_core::traits::SpawnNamed;
+#[cfg(feature = "full-node")]
+pub use {
+	sp_blockchain::HeaderBackend,
+	sp_consensus_babe::BabeApi,
+	sp_authority_discovery::AuthorityDiscoveryApi,
+	sc_client_api::AuxStore,
+	polkadot_primitives::v1::ParachainHost,
+	polkadot_overseer::{Overseer, OverseerHandler},
+};
+pub use sp_core::traits::SpawnNamed;
 
+#[cfg(feature = "full-node")]
 use polkadot_subsystem::jaeger;
 
 use std::sync::Arc;
@@ -53,7 +68,9 @@ use std::time::Duration;
 
 use prometheus_endpoint::Registry;
 use service::RpcHandlers;
-use telemetry::{Telemetry, TelemetryWorker, TelemetryWorkerHandle};
+#[cfg(feature = "full-node")]
+use telemetry::{Telemetry, TelemetryWorkerHandle};
+use telemetry::TelemetryWorker;
 
 #[cfg(feature = "rococo-native")]
 pub use polkadot_client::RococoExecutor;
@@ -92,6 +109,7 @@ pub use rococo_runtime;
 pub use westend_runtime;
 
 /// The maximum number of active leaves we forward to the [`Overseer`] on startup.
+#[cfg(any(test,feature = "full-node"))]
 const MAX_ACTIVE_LEAVES: usize = 4;
 
 #[derive(thiserror::Error, Debug)]
@@ -182,6 +200,7 @@ fn set_prometheus_registry(config: &mut Configuration) -> Result<(), Error> {
 
 /// Initialize the `Jeager` collector. The destination must listen
 /// on the given address and port for `UDP` packets.
+#[cfg(any(test,feature = "full-node"))]
 fn jaeger_launch_collector_with_agent(spawner: impl SpawnNamed, config: &Configuration, agent: Option<std::net::SocketAddr>) -> Result<(), Error> {
 	if let Some(agent) = agent {
 		let cfg = jaeger::JaegerConfig::builder()
@@ -404,145 +423,6 @@ fn new_partial<RuntimeApi, Executor>(
 	})
 }
 
-#[cfg(feature = "full-node")]
-fn real_overseer<Spawner, RuntimeClient>(
-	leaves: impl IntoIterator<Item = BlockInfo>,
-	keystore: Arc<LocalKeystore>,
-	runtime_client: Arc<RuntimeClient>,
-	parachains_db: Arc<dyn kvdb::KeyValueDB>,
-	availability_config: AvailabilityConfig,
-	approval_voting_config: ApprovalVotingConfig,
-	network_service: Arc<sc_network::NetworkService<Block, Hash>>,
-	authority_discovery: AuthorityDiscoveryService,
-	request_multiplexer: RequestMultiplexer,
-	registry: Option<&Registry>,
-	spawner: Spawner,
-	is_collator: IsCollator,
-	candidate_validation_config: CandidateValidationConfig,
-) -> Result<(Overseer<Spawner, Arc<RuntimeClient>>, OverseerHandler), 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;
-
-	use polkadot_availability_distribution::AvailabilityDistributionSubsystem;
-	use polkadot_node_core_av_store::AvailabilityStoreSubsystem;
-	use polkadot_availability_bitfield_distribution::BitfieldDistribution as BitfieldDistributionSubsystem;
-	use polkadot_node_core_bitfield_signing::BitfieldSigningSubsystem;
-	use polkadot_node_core_backing::CandidateBackingSubsystem;
-	use polkadot_node_core_candidate_validation::CandidateValidationSubsystem;
-	use polkadot_node_core_chain_api::ChainApiSubsystem;
-	use polkadot_node_collation_generation::CollationGenerationSubsystem;
-	use polkadot_collator_protocol::{CollatorProtocolSubsystem, ProtocolSide};
-	use polkadot_network_bridge::NetworkBridge as NetworkBridgeSubsystem;
-	use polkadot_node_core_provisioner::ProvisioningSubsystem as ProvisionerSubsystem;
-	use polkadot_node_core_runtime_api::RuntimeApiSubsystem;
-	use polkadot_statement_distribution::StatementDistribution as StatementDistributionSubsystem;
-	use polkadot_availability_recovery::AvailabilityRecoverySubsystem;
-	use polkadot_approval_distribution::ApprovalDistribution as ApprovalDistributionSubsystem;
-	use polkadot_node_core_approval_voting::ApprovalVotingSubsystem;
-	use polkadot_gossip_support::GossipSupport as GossipSupportSubsystem;
-
-	let all_subsystems = AllSubsystems {
-		availability_distribution: AvailabilityDistributionSubsystem::new(
-			keystore.clone(),
-			Metrics::register(registry)?,
-		),
-		availability_recovery: AvailabilityRecoverySubsystem::with_chunks_only(
-		),
-		availability_store: AvailabilityStoreSubsystem::new(
-			parachains_db.clone(),
-			availability_config,
-			Metrics::register(registry)?,
-		),
-		bitfield_distribution: BitfieldDistributionSubsystem::new(
-			Metrics::register(registry)?,
-		),
-		bitfield_signing: BitfieldSigningSubsystem::new(
-			spawner.clone(),
-			keystore.clone(),
-			Metrics::register(registry)?,
-		),
-		candidate_backing: CandidateBackingSubsystem::new(
-			spawner.clone(),
-			keystore.clone(),
-			Metrics::register(registry)?,
-		),
-		candidate_validation: CandidateValidationSubsystem::with_config(
-			candidate_validation_config,
-			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_collator {
-				IsCollator::Yes(collator_pair) => ProtocolSide::Collator(
-					network_service.local_peer_id().clone(),
-					collator_pair,
-					Metrics::register(registry)?,
-				),
-				IsCollator::No => ProtocolSide::Validator {
-					keystore: keystore.clone(),
-					eviction_policy: Default::default(),
-					metrics: Metrics::register(registry)?,
-				},
-			};
-			CollatorProtocolSubsystem::new(
-				side,
-			)
-		},
-		network_bridge: NetworkBridgeSubsystem::new(
-			network_service.clone(),
-			authority_discovery,
-			request_multiplexer,
-			Box::new(network_service.clone()),
-			Metrics::register(registry)?,
-		),
-		provisioner: ProvisionerSubsystem::new(
-			spawner.clone(),
-			(),
-			Metrics::register(registry)?,
-		),
-		runtime_api: RuntimeApiSubsystem::new(
-			runtime_client.clone(),
-			Metrics::register(registry)?,
-			spawner.clone(),
-		),
-		statement_distribution: StatementDistributionSubsystem::new(
-			keystore.clone(),
-			Metrics::register(registry)?,
-		),
-		approval_distribution: ApprovalDistributionSubsystem::new(
-			Metrics::register(registry)?,
-		),
-		approval_voting: ApprovalVotingSubsystem::with_config(
-			approval_voting_config,
-			parachains_db,
-			keystore.clone(),
-			Box::new(network_service.clone()),
-			Metrics::register(registry)?,
-		),
-		gossip_support: GossipSupportSubsystem::new(
-			keystore.clone(),
-		),
-	};
-
-	Overseer::new(
-		leaves,
-		all_subsystems,
-		registry,
-		runtime_client.clone(),
-		spawner,
-	).map_err(|e| e.into())
-}
-
 #[cfg(feature = "full-node")]
 pub struct NewFull<C> {
 	pub task_manager: TaskManager,
@@ -652,7 +532,7 @@ where
 /// This is an advanced feature and not recommended for general use. Generally, `build_full` is
 /// a better choice.
 #[cfg(feature = "full-node")]
-pub fn new_full<RuntimeApi, Executor>(
+pub fn new_full<RuntimeApi, Executor, OverseerGenerator>(
 	mut config: Configuration,
 	is_collator: IsCollator,
 	grandpa_pause: Option<(u32, u32)>,
@@ -660,12 +540,14 @@ pub fn new_full<RuntimeApi, Executor>(
 	jaeger_agent: Option<std::net::SocketAddr>,
 	telemetry_worker_handle: Option<TelemetryWorkerHandle>,
 	program_path: Option<std::path::PathBuf>,
+	overseer_gen: OverseerGenerator,
 ) -> Result<NewFull<Arc<FullClient<RuntimeApi, Executor>>>, Error>
 	where
 		RuntimeApi: ConstructRuntimeApi<Block, FullClient<RuntimeApi, Executor>> + Send + Sync + 'static,
 		RuntimeApi::RuntimeApi:
 		RuntimeApiCollection<StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,
 		Executor: NativeExecutionDispatch + 'static,
+		OverseerGenerator: OverseerGen,
 {
 	let role = config.role.clone();
 	let force_authoring = config.force_authoring;
@@ -840,20 +722,25 @@ pub fn new_full<RuntimeApi, Executor>(
 		.and_then(move |k| authority_discovery_service.map(|a| (a, k)));
 
 	let overseer_handler = if let Some((authority_discovery_service, keystore)) = maybe_params {
-		let (overseer, overseer_handler) = real_overseer(
-			active_leaves,
-			keystore,
-			overseer_client.clone(),
-			parachains_db,
-			availability_config,
-			approval_voting_config,
-			network.clone(),
-			authority_discovery_service,
-			request_multiplexer,
-			prometheus_registry.as_ref(),
-			spawner,
-			is_collator,
-			candidate_validation_config,
+		let (overseer, overseer_handler) = overseer_gen.generate::<
+			service::SpawnTaskHandle,
+			FullClient<RuntimeApi, Executor>,
+		>(
+			OverseerGenArgs {
+				leaves: active_leaves,
+				keystore,
+				runtime_client: overseer_client.clone(),
+				parachains_db,
+				availability_config,
+				approval_voting_config,
+				network_service: network.clone(),
+				authority_discovery_service,
+				request_multiplexer,
+				registry: prometheus_registry.as_ref(),
+				spawner,
+				is_collator,
+				candidate_validation_config,
+			}
 		)?;
 		let overseer_handler_clone = overseer_handler.clone();
 
@@ -1285,10 +1172,11 @@ pub fn build_full(
 	disable_beefy: bool,
 	jaeger_agent: Option<std::net::SocketAddr>,
 	telemetry_worker_handle: Option<TelemetryWorkerHandle>,
+	overseer_gen: impl OverseerGen,
 ) -> Result<NewFull<Client>, Error> {
 	#[cfg(feature = "rococo-native")]
 	if config.chain_spec.is_rococo() || config.chain_spec.is_wococo() {
-		return new_full::<rococo_runtime::RuntimeApi, RococoExecutor>(
+		return new_full::<rococo_runtime::RuntimeApi, RococoExecutor, _>(
 			config,
 			is_collator,
 			grandpa_pause,
@@ -1296,12 +1184,13 @@ pub fn build_full(
 			jaeger_agent,
 			telemetry_worker_handle,
 			None,
+			overseer_gen,
 		).map(|full| full.with_client(Client::Rococo))
 	}
 
 	#[cfg(feature = "kusama-native")]
 	if config.chain_spec.is_kusama() {
-		return new_full::<kusama_runtime::RuntimeApi, KusamaExecutor>(
+		return new_full::<kusama_runtime::RuntimeApi, KusamaExecutor, _>(
 			config,
 			is_collator,
 			grandpa_pause,
@@ -1309,12 +1198,13 @@ pub fn build_full(
 			jaeger_agent,
 			telemetry_worker_handle,
 			None,
+			overseer_gen,
 		).map(|full| full.with_client(Client::Kusama))
 	}
 
 	#[cfg(feature = "westend-native")]
 	if config.chain_spec.is_westend() {
-		return new_full::<westend_runtime::RuntimeApi, WestendExecutor>(
+		return new_full::<westend_runtime::RuntimeApi, WestendExecutor, _>(
 			config,
 			is_collator,
 			grandpa_pause,
@@ -1322,10 +1212,11 @@ pub fn build_full(
 			jaeger_agent,
 			telemetry_worker_handle,
 			None,
+			overseer_gen,
 		).map(|full| full.with_client(Client::Westend))
 	}
 
-	new_full::<polkadot_runtime::RuntimeApi, PolkadotExecutor>(
+	new_full::<polkadot_runtime::RuntimeApi, PolkadotExecutor, _>(
 		config,
 		is_collator,
 		grandpa_pause,
@@ -1333,5 +1224,6 @@ pub fn build_full(
 		jaeger_agent,
 		telemetry_worker_handle,
 		None,
+		overseer_gen,
 	).map(|full| full.with_client(Client::Polkadot))
 }