Less sleeps

bin/node/cli/Cargo.toml

@@ -124,6 +124,11 @@ regex = "1"
 platforms = "1.1"
 async-std = { version = "1.6.5", features = ["attributes"] }
 soketto = "0.4.2"
+jsonrpsee-ws-client = { version = "0.3.0", default-features = false, features = [
+	"tokio1",
+] }
+tokio = { version = "1.10", features = ["macros", "time"] }
+wait-timeout = "0.2"
 
 [build-dependencies]
 structopt = { version = "0.3.8", optional = true }
@@ -135,7 +140,7 @@ try-runtime-cli = { version = "0.10.0-dev", optional = true, path = "../../../ut
 sc-cli = { version = "0.10.0-dev", path = "../../../client/cli", optional = true }
 
 [features]
-default = [ "cli" ]
+default = ["cli"]
 cli = [
 	"node-executor/wasmi-errno",
 	"node-inspect",

bin/node/cli/src/chain_spec.rs

@@ -265,7 +265,7 @@ pub fn testnet_genesis(
 		.map(|x| &x.0)
 		.chain(initial_nominators.iter())
 		.for_each(|x| {
-			if !endowed_accounts.contains(&x) {
+			if !endowed_accounts.contains(x) {
 				endowed_accounts.push(x.clone())
 			}
 		});

bin/node/cli/src/service.rs

@@ -86,7 +86,7 @@ pub fn new_partial(
 
 	let (client, backend, keystore_container, task_manager) =
 		sc_service::new_full_parts::<Block, RuntimeApi, _>(
-			&config,
+			config,
 			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),
 			executor,
 		)?;
@@ -277,7 +277,7 @@ pub fn new_full_base(
 
 	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {
 		config,
-		backend: backend.clone(),
+		backend,
 		client: client.clone(),
 		keystore: keystore_container.sync_keystore(),
 		network: network.clone(),
@@ -507,7 +507,7 @@ pub fn new_light_base(
 		babe_block_import,
 		Some(Box::new(justification_import)),
 		client.clone(),
-		select_chain.clone(),
+		select_chain,
 		move |_, ()| async move {
 			let timestamp = sp_timestamp::InherentDataProvider::from_system_time();
 

bin/node/cli/tests/check_block_works.rs

@@ -24,11 +24,11 @@ use tempfile::tempdir;
 
 pub mod common;
 
-#[test]
-fn check_block_works() {
+#[tokio::test]
+async fn check_block_works() {
 	let base_path = tempdir().expect("could not create a temp dir");
 
-	common::run_node_for_a_while(base_path.path(), &["--dev"]);
+	common::run_node_for_a_while(base_path.path(), &["--dev"]).await;
 
 	let status = Command::new(cargo_bin("substrate"))
 		.args(&["check-block", "--dev", "--pruning", "archive", "-d"])

bin/node/cli/tests/common.rs

@@ -25,56 +25,128 @@ use nix::{
 };
 use std::{
 	convert::TryInto,
+	ops::{Deref, DerefMut},
 	path::Path,
 	process::{Child, Command, ExitStatus},
-	thread,
 	time::Duration,
 };
+use tokio::time::timeout;
+
+static LOCALHOST_WS: &str = "ws://127.0.0.1:9944/";
 
 /// Wait for the given `child` the given number of `secs`.
 ///
 /// Returns the `Some(exit status)` or `None` if the process did not finish in the given time.
-pub fn wait_for(child: &mut Child, secs: usize) -> Option<ExitStatus> {
-	for i in 0..secs {
-		match child.try_wait().unwrap() {
-			Some(status) => {
-				if i > 5 {
-					eprintln!("Child process took {} seconds to exit gracefully", i);
-				}
-				return Some(status)
-			},
-			None => thread::sleep(Duration::from_secs(1)),
+pub fn wait_for(child: &mut Child, secs: u64) -> Result<ExitStatus, ()> {
+	assert!(secs > 5);
+
+	let result =
+		wait_timeout::ChildExt::wait_timeout(child, Duration::from_secs(5)).map_err(|_| ())?;
+	if let Some(exit_status) = result {
+		Ok(exit_status)
+	} else {
+		eprintln!("Child process taking over 5 seconds to exit gracefully");
+		let result = wait_timeout::ChildExt::wait_timeout(child, Duration::from_secs(secs - 5))
+			.map_err(|_| ())?;
+		if let Some(exit_status) = result {
+			Ok(exit_status)
+		} else {
+			eprintln!("Took too long to exit (> {} seconds). Killing...", secs);
+			let _ = child.kill();
+			child.wait().unwrap();
+			Err(())
 		}
 	}
-	eprintln!("Took too long to exit (> {} seconds). Killing...", secs);
-	let _ = child.kill();
-	child.wait().unwrap();
+}
 
-	None
+pub async fn wait_n_blocks(n: usize, timeout_secs: u64) -> Result<(), tokio::time::error::Elapsed> {
+	timeout(Duration::from_secs(timeout_secs), wait_n_blocks_from(n, LOCALHOST_WS)).await
 }
 
-/// Run the node for a while (30 seconds)
-pub fn run_node_for_a_while(base_path: &Path, args: &[&str]) {
+/// Wait for at least n blocks to be produced
+///
+/// Eg. to wait for 3 blocks or a timeout of 30 seconds:
+/// ```
+/// timeout(Duration::from_secs(30), wait_n_blocks("ws://127.0.0.1:9944/", 3)).await;
+/// ```
+pub async fn wait_n_blocks_from(n: usize, url: &str) {
+	let mut built_blocks = std::collections::HashSet::new();
+	let mut interval = tokio::time::interval(Duration::from_secs(2));
+
+	loop {
+		// We could call remote-externalities like this:
+		// = remote_externalities::rpc_api::get_finalized_head::<String,
+		// String>("ws://127.0.0.1:9944/".to_string()).await;
+		// but then we'd need to gen the types to get the BlockT type.
+		// https://github.com/paritytech/substrate-subxt/blob/aj-metadata-vnext/proc-macro/src/generate_types.rs
+
+		if let Ok(ws_client) = jsonrpsee_ws_client::WsClientBuilder::default().build(url).await {
+			let block_result: Result<String, _> =
+				<jsonrpsee_ws_client::WsClient as jsonrpsee_ws_client::types::traits::Client>::request(
+					&ws_client,
+					"chain_getFinalizedHead",
+					jsonrpsee_ws_client::types::v2::params::JsonRpcParams::NoParams,
+				)
+				.await;
+			if let Ok(block) = block_result {
+				built_blocks.insert(block);
+				if built_blocks.len() > n {
+					break
+				}
+			}
+		}
+		interval.tick().await;
+	}
+}
+
+/// Run the node for a while (3 blocks)
+pub async fn run_node_for_a_while(base_path: &Path, args: &[&str]) {
 	let mut cmd = Command::new(cargo_bin("substrate"));
 
-	let mut cmd = cmd.args(args).arg("-d").arg(base_path).spawn().unwrap();
+	let mut child = KillChildOnDrop(cmd.args(args).arg("-d").arg(base_path).spawn().unwrap());
 
 	// Let it produce some blocks.
-	thread::sleep(Duration::from_secs(30));
-	assert!(cmd.try_wait().unwrap().is_none(), "the process should still be running");
+	wait_n_blocks(3, 30).await;
+
+	assert!(child.try_wait().unwrap().is_none(), "the process should still be running");
 
 	// Stop the process
-	kill(Pid::from_raw(cmd.id().try_into().unwrap()), SIGINT).unwrap();
-	assert!(wait_for(&mut cmd, 40).map(|x| x.success()).unwrap_or_default());
+	kill(Pid::from_raw(child.id().try_into().unwrap()), SIGINT).unwrap();
+	assert!(wait_for(&mut child, 40).map(|x| x.success()).unwrap());
 }
 
 /// Run the node asserting that it fails with an error
 pub fn run_node_assert_fail(base_path: &Path, args: &[&str]) {
 	let mut cmd = Command::new(cargo_bin("substrate"));
 
-	let mut cmd = cmd.args(args).arg("-d").arg(base_path).spawn().unwrap();
+	let mut child = KillChildOnDrop(cmd.args(args).arg("-d").arg(base_path).spawn().unwrap());
 
-	// Let it produce some blocks.
-	thread::sleep(Duration::from_secs(10));
-	assert!(cmd.try_wait().unwrap().is_some(), "the process should not be running anymore");
+	// Let it produce some blocks, but it should die within 10 seconds.
+	assert_ne!(
+		wait_timeout::ChildExt::wait_timeout(&mut *child, Duration::from_secs(10)).unwrap(),
+		None,
+		"the process should not be running anymore"
+	);
+}
+
+pub struct KillChildOnDrop(pub Child);
+
+impl Drop for KillChildOnDrop {
+	fn drop(&mut self) {
+		let _ = self.0.kill();
+	}
+}
+
+impl Deref for KillChildOnDrop {
+	type Target = Child;
+
+	fn deref(&self) -> &Self::Target {
+		&self.0
+	}
+}
+
+impl DerefMut for KillChildOnDrop {
+	fn deref_mut(&mut self) -> &mut Self::Target {
+		&mut self.0
+	}
 }

bin/node/cli/tests/database_role_subdir_migration.rs

@@ -25,9 +25,9 @@ use tempfile::tempdir;
 
 pub mod common;
 
-#[test]
+#[tokio::test]
 #[cfg(unix)]
-fn database_role_subdir_migration() {
+async fn database_role_subdir_migration() {
 	type Block = RawBlock<ExtrinsicWrapper<u64>>;
 
 	let base_path = tempdir().expect("could not create a temp dir");
@@ -62,7 +62,8 @@ fn database_role_subdir_migration() {
 			"44445",
 			"--no-prometheus",
 		],
-	);
+	)
+	.await;
 
 	// check if the database dir had been migrated
 	assert!(!path.join("db_version").exists());

bin/node/cli/tests/export_import_flow.rs

@@ -182,13 +182,13 @@ impl<'a> ExportImportRevertExecutor<'a> {
 	}
 }
 
-#[test]
-fn export_import_revert() {
+#[tokio::test]
+async fn export_import_revert() {
 	let base_path = tempdir().expect("could not create a temp dir");
 	let exported_blocks_file = base_path.path().join("exported_blocks");
 	let db_path = base_path.path().join("db");
 
-	common::run_node_for_a_while(base_path.path(), &["--dev"]);
+	common::run_node_for_a_while(base_path.path(), &["--dev"]).await;
 
 	let mut executor = ExportImportRevertExecutor::new(&base_path, &exported_blocks_file, &db_path);
 

bin/node/cli/tests/inspect_works.rs

@@ -24,11 +24,11 @@ use tempfile::tempdir;
 
 pub mod common;
 
-#[test]
-fn inspect_works() {
+#[tokio::test]
+async fn inspect_works() {
 	let base_path = tempdir().expect("could not create a temp dir");
 
-	common::run_node_for_a_while(base_path.path(), &["--dev"]);
+	common::run_node_for_a_while(base_path.path(), &["--dev"]).await;
 
 	let status = Command::new(cargo_bin("substrate"))
 		.args(&["inspect", "--dev", "--pruning", "archive", "-d"])

bin/node/cli/tests/purge_chain_works.rs

@@ -22,12 +22,12 @@ use tempfile::tempdir;
 
 pub mod common;
 
-#[test]
+#[tokio::test]
 #[cfg(unix)]
-fn purge_chain_works() {
+async fn purge_chain_works() {
 	let base_path = tempdir().expect("could not create a temp dir");
 
-	common::run_node_for_a_while(base_path.path(), &["--dev"]);
+	common::run_node_for_a_while(base_path.path(), &["--dev"]).await;
 
 	let status = Command::new(cargo_bin("substrate"))
 		.args(&["purge-chain", "--dev", "-d"])