CLI improvements & fixes (#4812)

These are a few changes I missed during the refactoring.

1. Initialization issue and boilerplate

    Most importantly: part of the `Configuration` initialization was done in `sc_cli::init`. This means the user can not benefit from this initialization boilerplate if they have multiple `Configuration` since `sc_cli::init` can only be called once.

2. Boilerplate for `VersionInfo` and `Configuration`

    I'm also answering to the critic of @bkchr on the initialization using version: https://github.com/paritytech/substrate/pull/4692/files/bea809d4c14a2ede953227ac885e3b3f9771c548#r372047238 This will allow initializing a `Configuration` and provide the version by default.

3. Loading the `chain_spec` explicitly

    In the past it was done automatically but in some cases we want to delay this. I moved the code to `Configuration.load_spec()` so it can be called later on. `chain_spec` can also be written directly to the `Configuration` without using this `load_spec` helper.

4. [deleted]

5. Fixing issue that prevents the user to override the port

    In the refactoring I introduced a bug by mistake that could potentially prevent the CLI user to override the ports if defaults where provided for these ports (only on cumulus).

6. Change task_executor from Box to Arc

    This is useful for cumulus where we have 2 nodes with 2 separate Configuration that need to spawn tasks to the same runtime.

7. Renamed TasksExecutorRequired to TaskExecutor

    For consistency.

This is related to https://github.com/paritytech/cumulus/issues/24

This is the continuation (and hopefully the end of) #4692

substrate/client/cli/src/lib.rs

@@ -35,6 +35,7 @@ use sc_service::{
 	RuntimeGenesis, ChainSpecExtension, PruningMode, ChainSpec,
 	AbstractService, Roles as ServiceRoles,
 };
+pub use sc_service::config::VersionInfo;
 use sc_network::{
 	self,
 	multiaddr::Protocol,
@@ -74,32 +75,11 @@ const DEFAULT_NETWORK_CONFIG_PATH : &'static str = "network";
 /// default sub directory to store database
 const DEFAULT_DB_CONFIG_PATH : &'static str = "db";
 /// default sub directory for the key store
-const DEFAULT_KEYSTORE_CONFIG_PATH : &'static str =  "keystore";
+const DEFAULT_KEYSTORE_CONFIG_PATH : &'static str = "keystore";
 
 /// The maximum number of characters for a node name.
 const NODE_NAME_MAX_LENGTH: usize = 32;
 
-/// Executable version. Used to pass version information from the root crate.
-#[derive(Clone)]
-pub struct VersionInfo {
-	/// Implementaiton name.
-	pub name: &'static str,
-	/// Implementation version.
-	pub version: &'static str,
-	/// SCM Commit hash.
-	pub commit: &'static str,
-	/// Executable file name.
-	pub executable_name: &'static str,
-	/// Executable file description.
-	pub description: &'static str,
-	/// Executable file author.
-	pub author: &'static str,
-	/// Support URL.
-	pub support_url: &'static str,
-	/// Copyright starting year (x-current year)
-	pub copyright_start_year: i32,
-}
-
 fn get_chain_key(cli: &SharedParams) -> String {
 	match cli.chain {
 		Some(ref chain) => chain.clone(),
@@ -120,8 +100,12 @@ fn generate_node_name() -> String {
 	result
 }
 
-/// Load spec give shared params and spec factory.
-pub fn load_spec<F, G, E>(cli: &SharedParams, factory: F) -> error::Result<ChainSpec<G, E>> where
+/// Load spec to `Configuration` from shared params and spec factory.
+pub fn load_spec<'a, G, E, F>(
+	mut config: &'a mut Configuration<G, E>,
+	cli: &SharedParams,
+	factory: F,
+) -> error::Result<&'a ChainSpec<G, E>> where
 	G: RuntimeGenesis,
 	E: ChainSpecExtension,
 	F: FnOnce(&str) -> Result<Option<ChainSpec<G, E>>, String>,
@@ -131,7 +115,13 @@ pub fn load_spec<F, G, E>(cli: &SharedParams, factory: F) -> error::Result<Chain
 		Some(spec) => spec,
 		None => ChainSpec::from_json_file(PathBuf::from(chain_key))?
 	};
-	Ok(spec)
+
+	config.network.boot_nodes = spec.boot_nodes().to_vec();
+	config.telemetry_endpoints = spec.telemetry_endpoints().clone();
+
+	config.chain_spec = Some(spec);
+
+	Ok(config.chain_spec.as_ref().unwrap())
 }
 
 fn base_path(cli: &SharedParams, version: &VersionInfo) -> PathBuf {
@@ -243,8 +233,8 @@ where
 	SL: AbstractService + Unpin,
 	SF: AbstractService + Unpin,
 {
-	init(&mut config, spec_factory, &run_cmd.shared_params, version)?;
-
+	init(&run_cmd.shared_params, version)?;
+	load_spec(&mut config, &run_cmd.shared_params, spec_factory)?;
 	run_cmd.run(config, new_light, new_full, version)
 }
 
@@ -266,30 +256,21 @@ where
 	<<<BB as BlockT>::Header as HeaderT>::Number as std::str::FromStr>::Err: std::fmt::Debug,
 	<BB as BlockT>::Hash: std::str::FromStr,
 {
-	init(&mut config, spec_factory, &subcommand.get_shared_params(), version)?;
+	let shared_params = subcommand.get_shared_params();
 
+	init(shared_params, version)?;
+	load_spec(&mut config, shared_params, spec_factory)?;
 	subcommand.run(config, builder)
 }
 
-/// Initialize substrate and its configuration
+/// Initialize substrate. This must be done only once.
 ///
 /// This method:
 ///
 /// 1.  set the panic handler
 /// 2.  raise the FD limit
 /// 3.  initialize the logger
-/// 4.  update the configuration provided with the chain specification, config directory,
-///     information (version, commit), database's path, boot nodes and telemetry endpoints
-pub fn init<G, E, F>(
-	mut config: &mut Configuration<G, E>,
-	spec_factory: F,
-	shared_params: &SharedParams,
-	version: &VersionInfo,
-) -> error::Result<()>
-where
-	G: RuntimeGenesis,
-	E: ChainSpecExtension,
-	F: FnOnce(&str) -> Result<Option<ChainSpec<G, E>>, String>,
+pub fn init(shared_params: &SharedParams, version: &VersionInfo) -> error::Result<()>
 {
 	let full_version = sc_service::config::full_version_from_strs(
 		version.version,
@@ -300,21 +281,6 @@ where
 	fdlimit::raise_fd_limit();
 	init_logger(shared_params.log.as_ref().map(|v| v.as_ref()).unwrap_or(""));
 
-	config.chain_spec = Some(load_spec(shared_params, spec_factory)?);
-	config.config_dir = Some(base_path(shared_params, version));
-	config.impl_commit = version.commit;
-	config.impl_version = version.version;
-
-	config.database = DatabaseConfig::Path {
-		path: config
-			.in_chain_config_dir(DEFAULT_DB_CONFIG_PATH)
-			.expect("We provided a base_path/config_dir."),
-		cache_size: None,
-	};
-
-	config.network.boot_nodes = config.expect_chain_spec().boot_nodes().to_vec();
-	config.telemetry_endpoints = config.expect_chain_spec().telemetry_endpoints().clone();
-
 	Ok(())
 }
 
@@ -419,8 +385,6 @@ fn fill_network_configuration(
 		];
 	}
 
-	config.public_addresses = Vec::new();
-
 	config.client_version = client_id;
 	config.node_key = node_key::node_key_config(cli.node_key_params, &config.net_config_path)?;
 
@@ -496,10 +460,8 @@ pub fn fill_import_params<G, E>(
 where
 	G: RuntimeGenesis,
 {
-	match config.database {
-		DatabaseConfig::Path { ref mut cache_size, .. } =>
-			*cache_size = Some(cli.database_cache_size),
-		DatabaseConfig::Custom(_) => {},
+	if let Some(DatabaseConfig::Path { ref mut cache_size, .. }) = config.database {
+		*cache_size = Some(cli.database_cache_size);
 	}
 
 	config.state_cache_size = cli.state_cache_size;
@@ -549,14 +511,30 @@ where
 	Ok(())
 }
 
-/// Update and prepare a `Configuration` with command line parameters of `RunCmd`
+/// Update and prepare a `Configuration` with command line parameters of `RunCmd` and `VersionInfo`
 pub fn update_config_for_running_node<G, E>(
 	mut config: &mut Configuration<G, E>,
 	cli: RunCmd,
+	version: &VersionInfo,
 ) -> error::Result<()>
 where
 	G: RuntimeGenesis,
 {
+	if config.config_dir.is_none() {
+		config.config_dir = Some(base_path(&cli.shared_params, version));
+	}
+
+	if config.database.is_none() {
+		// NOTE: the loading of the DatabaseConfig is voluntarily delayed to here
+		//       in case config.config_dir has been customized
+		config.database = Some(DatabaseConfig::Path {
+			path: config
+				.in_chain_config_dir(DEFAULT_DB_CONFIG_PATH)
+				.expect("We provided a base_path/config_dir."),
+			cache_size: None,
+		});
+	}
+
 	fill_config_keystore_password_and_path(&mut config, &cli)?;
 
 	let keyring = cli.get_keyring();
@@ -580,16 +558,13 @@ where
 		(_, Some(keyring)) => keyring.to_string(),
 		(None, None) => generate_node_name(),
 	};
-	match node_key::is_node_name_valid(&config.name) {
-		Ok(_) => (),
-		Err(msg) => Err(
-			error::Error::Input(
-				format!("Invalid node name '{}'. Reason: {}. If unsure, use none.",
-					config.name,
-					msg
-				)
+	if let Err(msg) = node_key::is_node_name_valid(&config.name) {
+		return Err(error::Error::Input(
+			format!("Invalid node name '{}'. Reason: {}. If unsure, use none.",
+				config.name,
+				msg,
 			)
-		)?
+		));
 	}
 
 	// set sentry mode (i.e. act as an authority but **never** actively participate)
@@ -625,16 +600,16 @@ where
 			}
 		});
 
-	if config.rpc_http.is_none() {
+	if config.rpc_http.is_none() || cli.rpc_port.is_some() {
 		let rpc_interface: &str = interface_str(cli.rpc_external, cli.unsafe_rpc_external, cli.validator)?;
 		config.rpc_http = Some(parse_address(&format!("{}:{}", rpc_interface, 9933), cli.rpc_port)?);
 	}
-	if config.rpc_ws.is_none() {
+	if config.rpc_ws.is_none() || cli.ws_port.is_some() {
 		let ws_interface: &str = interface_str(cli.ws_external, cli.unsafe_ws_external, cli.validator)?;
 		config.rpc_ws = Some(parse_address(&format!("{}:{}", ws_interface, 9944), cli.ws_port)?);
 	}
 
-	if config.grafana_port.is_none() {
+	if config.grafana_port.is_none() || cli.grafana_port.is_some() {
 		let grafana_interface: &str = if cli.grafana_external { "0.0.0.0" } else { "127.0.0.1" };
 		config.grafana_port = Some(
 			parse_address(&format!("{}:{}", grafana_interface, 9955), cli.grafana_port)?
@@ -781,6 +756,17 @@ fn kill_color(s: &str) -> String {
 mod tests {
 	use super::*;
 
+	const TEST_VERSION_INFO: &'static VersionInfo = &VersionInfo {
+		name: "node-test",
+		version: "0.1.0",
+		commit: "some_commit",
+		executable_name: "node-test",
+		description: "description",
+		author: "author",
+		support_url: "http://example.org",
+		copyright_start_year: 2020,
+	};
+
 	#[test]
 	fn keystore_path_is_generated_correctly() {
 		let chain_spec = ChainSpec::from_genesis(
@@ -805,6 +791,7 @@ mod tests {
 			update_config_for_running_node(
 				&mut node_config,
 				run_cmds.clone(),
+				TEST_VERSION_INFO,
 			).unwrap();
 
 			let expected_path = match keystore_path {
@@ -815,4 +802,60 @@ mod tests {
 			assert_eq!(expected_path, node_config.keystore.path().unwrap().to_owned());
 		}
 	}
+
+	#[test]
+	fn ensure_load_spec_provide_defaults() {
+		let chain_spec = ChainSpec::from_genesis(
+			"test",
+			"test-id",
+			|| (),
+			vec!["boo".to_string()],
+			Some(TelemetryEndpoints::new(vec![("foo".to_string(), 42)])),
+			None,
+			None,
+			None::<()>,
+		);
+
+		let args: Vec<&str> = vec![];
+		let cli = RunCmd::from_iter(args);
+
+		let mut config = Configuration::new(TEST_VERSION_INFO);
+		load_spec(&mut config, &cli.shared_params, |_| Ok(Some(chain_spec))).unwrap();
+
+		assert!(config.chain_spec.is_some());
+		assert!(!config.network.boot_nodes.is_empty());
+		assert!(config.telemetry_endpoints.is_some());
+	}
+
+	#[test]
+	fn ensure_update_config_for_running_node_provides_defaults() {
+		let chain_spec = ChainSpec::from_genesis(
+			"test",
+			"test-id",
+			|| (),
+			vec![],
+			None,
+			None,
+			None,
+			None::<()>,
+		);
+
+		let args: Vec<&str> = vec![];
+		let cli = RunCmd::from_iter(args);
+
+		let mut config = Configuration::new(TEST_VERSION_INFO);
+		config.chain_spec = Some(chain_spec);
+		update_config_for_running_node(&mut config, cli, TEST_VERSION_INFO).unwrap();
+
+		assert!(config.config_dir.is_some());
+		assert!(config.database.is_some());
+		if let Some(DatabaseConfig::Path { ref cache_size, .. }) = config.database {
+			assert!(cache_size.is_some());
+		} else {
+			panic!("invalid config.database variant");
+		}
+		assert!(!config.name.is_empty());
+		assert!(config.network.config_path.is_some());
+		assert!(!config.network.listen_addresses.is_empty());
+	}
 }

substrate/client/cli/src/params.rs

@@ -936,6 +936,7 @@ impl RunCmd {
 		crate::update_config_for_running_node(
 			&mut config,
 			self,
+			&version,
 		)?;
 
 		crate::run_node(config, new_light, new_full, &version)
@@ -1003,7 +1004,7 @@ impl ExportBlocksCmd {
 
 		crate::fill_config_keystore_in_memory(&mut config)?;
 
-		if let DatabaseConfig::Path { ref path, .. } = &config.database {
+		if let DatabaseConfig::Path { ref path, .. } = config.expect_database() {
 			info!("DB path: {}", path.display());
 		}
 		let from = self.from.as_ref().and_then(|f| f.parse().ok()).unwrap_or(1);
@@ -1124,7 +1125,7 @@ impl PurgeChainCmd {
 
 		crate::fill_config_keystore_in_memory(&mut config)?;
 
-		let db_path = match config.database {
+		let db_path = match config.expect_database() {
 			DatabaseConfig::Path { path, .. } => path,
 			_ => {
 				eprintln!("Cannot purge custom database implementation");

substrate/client/cli/src/runtime.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::sync::Arc;
+
 use futures::{Future, future, future::FutureExt};
 use futures::select;
 use futures::pin_mut;
@@ -91,7 +93,7 @@ where
 
 	config.task_executor = {
 		let runtime_handle = runtime.handle().clone();
-		Some(Box::new(move |fut| { runtime_handle.spawn(fut); }))
+		Some(Arc::new(move |fut| { runtime_handle.spawn(fut); }))
 	};
 
 	let f = future_builder(config)?;
@@ -117,7 +119,7 @@ where
 
 	config.task_executor = {
 		let runtime_handle = runtime.handle().clone();
-		Some(Box::new(move |fut| { runtime_handle.spawn(fut); }))
+		Some(Arc::new(move |fut| { runtime_handle.spawn(fut); }))
 	};
 
 	let service = service_builder(config)?;

substrate/client/service/src/builder.rs

@@ -196,7 +196,7 @@ fn new_full_parts<TBl, TRtApi, TExecDisp, TGen, TCSExt>(
 			state_cache_child_ratio:
 			config.state_cache_child_ratio.map(|v| (v, 100)),
 			pruning: config.pruning.clone(),
-			source: match &config.database {
+			source: match config.expect_database() {
 				DatabaseConfig::Path { path, cache_size } =>
 					sc_client_db::DatabaseSettingsSrc::Path {
 						path: path.clone(),
@@ -307,7 +307,7 @@ where TGen: RuntimeGenesis, TCSExt: Extension {
 				state_cache_child_ratio:
 					config.state_cache_child_ratio.map(|v| (v, 100)),
 				pruning: config.pruning.clone(),
-				source: match &config.database {
+				source: match config.expect_database() {
 					DatabaseConfig::Path { path, cache_size } =>
 						sc_client_db::DatabaseSettingsSrc::Path {
 							path: path.clone(),
@@ -1187,7 +1187,7 @@ ServiceBuilder<
 			task_executor: if let Some(exec) = config.task_executor {
 				exec
 			} else {
-				return Err(Error::TasksExecutorRequired);
+				return Err(Error::TaskExecutorRequired);
 			},
 			rpc_handlers,
 			_rpc: rpc,

substrate/client/service/src/config.rs

@@ -28,6 +28,27 @@ use sp_core::crypto::Protected;
 use target_info::Target;
 use sc_telemetry::TelemetryEndpoints;
 
+/// Executable version. Used to pass version information from the root crate.
+#[derive(Clone)]
+pub struct VersionInfo {
+	/// Implementation name.
+	pub name: &'static str,
+	/// Implementation version.
+	pub version: &'static str,
+	/// SCM Commit hash.
+	pub commit: &'static str,
+	/// Executable file name.
+	pub executable_name: &'static str,
+	/// Executable file description.
+	pub description: &'static str,
+	/// Executable file author.
+	pub author: &'static str,
+	/// Support URL.
+	pub support_url: &'static str,
+	/// Copyright starting year (x-current year)
+	pub copyright_start_year: i32,
+}
+
 /// Service configuration.
 pub struct Configuration<G, E = NoExtension> {
 	/// Implementation name
@@ -39,7 +60,7 @@ pub struct Configuration<G, E = NoExtension> {
 	/// Node roles.
 	pub roles: Roles,
 	/// How to spawn background tasks. Mandatory, otherwise creating a `Service` will error.
-	pub task_executor: Option<Box<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send>>,
+	pub task_executor: Option<Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send + Sync>>,
 	/// Extrinsic pool configuration.
 	pub transaction_pool: TransactionPoolOptions,
 	/// Network configuration.
@@ -49,7 +70,7 @@ pub struct Configuration<G, E = NoExtension> {
 	/// Configuration for the keystore.
 	pub keystore: KeystoreConfig,
 	/// Configuration for the database.
-	pub database: DatabaseConfig,
+	pub database: Option<DatabaseConfig>,
 	/// Size of internal state cache in Bytes
 	pub state_cache_size: usize,
 	/// Size in percent of cache size dedicated to child tries
@@ -147,7 +168,7 @@ pub enum DatabaseConfig {
 impl<G, E> Default for Configuration<G, E> {
 	/// Create a default config
 	fn default() -> Self {
-		let configuration = Configuration {
+		Configuration {
 			impl_name: "parity-substrate",
 			impl_version: "0.0.0",
 			impl_commit: "",
@@ -159,10 +180,7 @@ impl<G, E> Default for Configuration<G, E> {
 			transaction_pool: Default::default(),
 			network: Default::default(),
 			keystore: KeystoreConfig::None,
-			database: DatabaseConfig::Path {
-				path: Default::default(),
-				cache_size: Default::default(),
-			},
+			database: None,
 			state_cache_size: Default::default(),
 			state_cache_child_ratio: Default::default(),
 			pruning: PruningMode::default(),
@@ -183,14 +201,21 @@ impl<G, E> Default for Configuration<G, E> {
 			dev_key_seed: None,
 			tracing_targets: Default::default(),
 			tracing_receiver: Default::default(),
-		};
-
-		configuration
+		}
 	}
-
 }
 
 impl<G, E> Configuration<G, E> {
+	/// Create a default config using `VersionInfo`
+	pub fn new(version: &VersionInfo) -> Self {
+		let mut config = Configuration::default();
+		config.impl_name = version.name;
+		config.impl_version = version.version;
+		config.impl_commit = version.commit;
+
+		config
+	}
+
 	/// Returns full version string of this configuration.
 	pub fn full_version(&self) -> String {
 		full_version_from_strs(self.impl_version, self.impl_commit)
@@ -220,6 +245,15 @@ impl<G, E> Configuration<G, E> {
 	pub fn expect_chain_spec(&self) -> &ChainSpec<G, E> {
 		self.chain_spec.as_ref().expect("chain_spec must be specified")
 	}
+
+	/// Return a reference to the `DatabaseConfig` of this `Configuration`.
+	///
+	/// ### Panics
+	///
+	/// This method panic if the `database` is `None`
+	pub fn expect_database(&self) -> &DatabaseConfig {
+		self.database.as_ref().expect("database must be specified")
+	}
 }
 
 /// Returns platform info

substrate/client/service/src/error.rs

@@ -42,7 +42,7 @@ pub enum Error {
 	SelectChainRequired,
 	/// Tasks executor is missing.
 	#[display(fmt="Tasks executor hasn't been provided.")]
-	TasksExecutorRequired,
+	TaskExecutorRequired,
 	/// Other error.
 	Other(String),
 }

substrate/client/service/src/lib.rs

@@ -96,7 +96,7 @@ pub struct Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
 	/// Receiver for futures that must be spawned as background tasks.
 	to_spawn_rx: mpsc::UnboundedReceiver<(Pin<Box<dyn Future<Output = ()> + Send>>, Cow<'static, str>)>,
 	/// How to spawn background tasks.
-	task_executor: Box<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send>,
+	task_executor: Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send + Sync>,
 	rpc_handlers: sc_rpc_server::RpcHandler<sc_rpc::Metadata>,
 	_rpc: Box<dyn std::any::Any + Send + Sync>,
 	_telemetry: Option<sc_telemetry::Telemetry>,

substrate/client/service/test/src/lib.rs

@@ -133,7 +133,7 @@ fn node_config<G, E: Clone> (
 	index: usize,
 	spec: &ChainSpec<G, E>,
 	role: Roles,
-	task_executor: Box<dyn Fn(Pin<Box<dyn futures::Future<Output = ()> + Send>>) + Send>,
+	task_executor: Arc<dyn Fn(Pin<Box<dyn futures::Future<Output = ()> + Send>>) + Send + Sync>,
 	key_seed: Option<String>,
 	base_port: u16,
 	root: &TempDir,
@@ -183,10 +183,10 @@ fn node_config<G, E: Clone> (
 			password: None
 		},
 		config_dir: Some(root.clone()),
-		database: DatabaseConfig::Path {
+		database: Some(DatabaseConfig::Path {
 			path: root.join("db"),
 			cache_size: None
-		},
+		}),
 		state_cache_size: 16777216,
 		state_cache_child_ratio: None,
 		pruning: Default::default(),
@@ -256,7 +256,7 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		for (key, authority) in authorities {
 			let task_executor = {
 				let executor = executor.clone();
-				Box::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
+				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
 			};
 			let node_config = node_config(
 				self.nodes,
@@ -280,7 +280,7 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		for full in full {
 			let task_executor = {
 				let executor = executor.clone();
-				Box::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
+				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
 			};
 			let node_config = node_config(self.nodes, &self.chain_spec, Roles::FULL, task_executor, None, self.base_port, &temp);
 			let addr = node_config.network.listen_addresses.iter().next().unwrap().clone();
@@ -296,7 +296,7 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		for light in light {
 			let task_executor = {
 				let executor = executor.clone();
-				Box::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
+				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>| executor.spawn(fut.unit_error().compat()))
 			};
 			let node_config = node_config(self.nodes, &self.chain_spec, Roles::LIGHT, task_executor, None, self.base_port, &temp);
 			let addr = node_config.network.listen_addresses.iter().next().unwrap().clone();