Merge remote-tracking branch 'origin/master' into lexnv/metadata_v15

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

Cargo.lock

@@ -1624,6 +1624,7 @@ dependencies = [
  "sp-core",
  "sp-keyring",
  "sp-runtime",
+ "substrate-runner",
  "subxt",
  "subxt-codegen",
  "subxt-metadata",
@@ -3481,6 +3482,10 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "substrate-runner"
+version = "0.28.0"
+
 [[package]]
 name = "subtle"
 version = "2.4.1"
@@ -3647,9 +3652,10 @@ dependencies = [
 name = "test-runtime"
 version = "0.28.0"
 dependencies = [
+ "impl-serde",
  "jsonrpsee",
- "parity-scale-codec",
+ "serde",
- "sp-runtime",
+ "substrate-runner",
  "subxt",
  "tokio",
  "which",

Cargo.toml

@@ -3,6 +3,7 @@ members = [
     "cli",
     "codegen",
     "examples",
+    "testing/substrate-runner",
     "testing/test-runtime",
     "testing/integration-tests",
     "testing/ui-tests",

testing/integration-tests/Cargo.toml

@@ -36,3 +36,4 @@ tracing = "0.1.34"
 tracing-subscriber = "0.3.11"
 wabt = "0.10.0"
 which = "4.4.0"
+substrate-runner = { path = "../substrate-runner" }

testing/integration-tests/src/utils/context.rs

@@ -22,11 +22,9 @@ pub async fn test_context_with(key: AccountKeyring) -> TestContext {
         SUBSTRATE_NODE_PATH.to_string()
     });
 
-    let proc = TestContext::build(path.as_str())
+    let mut proc = TestContext::build(path.as_str());
-        .with_authority(key)
+    proc.with_authority(key);
-        .spawn::<SubstrateConfig>()
+    proc.spawn::<SubstrateConfig>().await.unwrap()
-        .await;
-    proc.unwrap()
 }
 
 pub type TestContext = TestNodeProcess<SubstrateConfig>;

testing/integration-tests/src/utils/node_proc.rs

@@ -3,28 +3,17 @@
 // see LICENSE for license details.
 
 use sp_keyring::AccountKeyring;
-use std::{
+use std::ffi::{OsStr, OsString};
-    ffi::{OsStr, OsString},
+use substrate_runner::SubstrateNode;
-    io::{BufRead, BufReader, Read},
-    process,
-};
 use subxt::{Config, OnlineClient};
 
 /// Spawn a local substrate node for testing subxt.
 pub struct TestNodeProcess<R: Config> {
-    proc: process::Child,
+    // Keep a handle to the node; once it's dropped the node is killed.
+    _proc: SubstrateNode,
     client: OnlineClient<R>,
 }
 
-impl<R> Drop for TestNodeProcess<R>
-where
-    R: Config,
-{
-    fn drop(&mut self) {
-        let _ = self.kill();
-    }
-}
-
 impl<R> TestNodeProcess<R>
 where
     R: Config,
@@ -37,17 +26,6 @@ where
         TestNodeProcessBuilder::new(program)
     }
 
-    /// Attempt to kill the running substrate process.
-    pub fn kill(&mut self) -> Result<(), String> {
-        tracing::info!("Killing node process {}", self.proc.id());
-        if let Err(err) = self.proc.kill() {
-            let err = format!("Error killing node process {}: {}", self.proc.id(), err);
-            tracing::error!("{}", err);
-            return Err(err);
-        }
-        Ok(())
-    }
-
     /// Returns the subxt client connected to the running node.
     pub fn client(&self) -> OnlineClient<R> {
         self.client.clone()
@@ -78,78 +56,31 @@ impl TestNodeProcessBuilder {
     }
 
     /// Spawn the substrate node at the given path, and wait for rpc to be initialized.
-    pub async fn spawn<R>(&self) -> Result<TestNodeProcess<R>, String>
+    pub async fn spawn<R>(self) -> Result<TestNodeProcess<R>, String>
     where
         R: Config,
     {
-        let mut cmd = process::Command::new(&self.node_path);
+        let mut node_builder = SubstrateNode::builder();
-        cmd.env("RUST_LOG", "info")
+
-            .arg("--dev")
+        node_builder.binary_path(self.node_path);
-            .arg("--tmp")
-            .stdout(process::Stdio::piped())
-            .stderr(process::Stdio::piped())
-            .arg("--port=0")
-            .arg("--rpc-port=0")
-            .arg("--ws-port=0");
 
         if let Some(authority) = self.authority {
             let authority = format!("{authority:?}");
-            let arg = format!("--{}", authority.as_str().to_lowercase());
+            node_builder.arg(authority.as_str().to_lowercase());
-            cmd.arg(arg);
         }
 
-        let mut proc = cmd.spawn().map_err(|e| {
+        // Spawn the node and retrieve a URL to it:
-            format!(
+        let proc = node_builder.spawn().map_err(|e| e.to_string())?;
-                "Error spawning substrate node '{}': {}",
+        let ws_url = format!("ws://127.0.0.1:{}", proc.ws_port());
-                self.node_path.to_string_lossy(),
-                e
-            )
-        })?;
-
-        // Wait for RPC port to be logged (it's logged to stderr):
-        let stderr = proc.stderr.take().unwrap();
-        let ws_port = find_substrate_port_from_output(stderr);
-        let ws_url = format!("ws://127.0.0.1:{ws_port}");
 
         // Connect to the node with a subxt client:
         let client = OnlineClient::from_url(ws_url.clone()).await;
         match client {
-            Ok(client) => Ok(TestNodeProcess { proc, client }),
+            Ok(client) => Ok(TestNodeProcess {
-            Err(err) => {
+                _proc: proc,
-                let err = format!("Failed to connect to node rpc at {ws_url}: {err}");
+                client,
-                tracing::error!("{}", err);
+            }),
-                proc.kill().map_err(|e| {
+            Err(err) => Err(format!("Failed to connect to node rpc at {ws_url}: {err}")),
-                    format!("Error killing substrate process '{}': {}", proc.id(), e)
-                })?;
-                Err(err)
-            }
         }
     }
 }
-
-// Consume a stderr reader from a spawned substrate command and
-// locate the port number that is logged out to it.
-fn find_substrate_port_from_output(r: impl Read + Send + 'static) -> u16 {
-    BufReader::new(r)
-        .lines()
-        .find_map(|line| {
-            let line = line.expect("failed to obtain next line from stdout for port discovery");
-
-            // does the line contain our port (we expect this specific output from substrate).
-            let line_end = line
-                .rsplit_once("Listening for new connections on 127.0.0.1:")
-                .or_else(|| line.rsplit_once("Running JSON-RPC WS server: addr=127.0.0.1:"))
-                .map(|(_, port_str)| port_str)?;
-
-            // trim non-numeric chars from the end of the port part of the line.
-            let port_str = line_end.trim_end_matches(|b: char| !b.is_ascii_digit());
-
-            // expect to have a number here (the chars after '127.0.0.1:') and parse them into a u16.
-            let port_num = port_str
-                .parse()
-                .unwrap_or_else(|_| panic!("valid port expected for log line, got '{port_str}'"));
-
-            Some(port_num)
-        })
-        .expect("We should find a port before the reader ends")
-}

testing/substrate-runner/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "substrate-runner"
+version = "0.28.0"
+edition = "2021"
+publish = false
+
+[dependencies]

testing/substrate-runner/README.md

@@ -0,0 +1,3 @@
+# substrate-runner
+
+A small crate whose sole purpose is starting up a substrate node on some free port and handing back a handle to it with the port that it started on.

testing/substrate-runner/src/error.rs

@@ -0,0 +1,23 @@
+// Copyright 2019-2023 Parity Technologies (UK) Ltd.
+// This file is dual-licensed as Apache-2.0 or GPL-3.0.
+// see LICENSE for license details.
+
+#[derive(Debug)]
+pub enum Error {
+    Io(std::io::Error),
+    CouldNotExtractPort,
+}
+
+impl std::fmt::Display for Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Error::Io(err) => write!(f, "IO error: {err}"),
+            Error::CouldNotExtractPort => write!(
+                f,
+                "could not extract port from running substrate node's stdout"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for Error {}

testing/substrate-runner/src/lib.rs

@@ -0,0 +1,143 @@
+// Copyright 2019-2023 Parity Technologies (UK) Ltd.
+// This file is dual-licensed as Apache-2.0 or GPL-3.0.
+// see LICENSE for license details.
+
+mod error;
+
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::ffi::OsString;
+use std::io::{BufRead, BufReader, Read};
+use std::process::{self, Command};
+
+pub use error::Error;
+
+type CowStr = Cow<'static, str>;
+
+pub struct SubstrateNodeBuilder {
+    binary_path: OsString,
+    custom_flags: HashMap<CowStr, Option<CowStr>>,
+}
+
+impl Default for SubstrateNodeBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SubstrateNodeBuilder {
+    /// Configure a new Substrate node.
+    pub fn new() -> Self {
+        SubstrateNodeBuilder {
+            binary_path: "substrate".into(),
+            custom_flags: Default::default(),
+        }
+    }
+
+    /// Set the path to the `substrate` binary; defaults to "substrate".
+    pub fn binary_path(&mut self, path: impl Into<OsString>) -> &mut Self {
+        self.binary_path = path.into();
+        self
+    }
+
+    /// Provide a boolean argument like `--alice`
+    pub fn arg(&mut self, s: impl Into<CowStr>) -> &mut Self {
+        self.custom_flags.insert(s.into(), None);
+        self
+    }
+
+    /// Provide an argument with a value.
+    pub fn arg_val(&mut self, key: impl Into<CowStr>, val: impl Into<CowStr>) -> &mut Self {
+        self.custom_flags.insert(key.into(), Some(val.into()));
+        self
+    }
+
+    /// Spawn the node, handing back an object which, when dropped, will stop it.
+    pub fn spawn(self) -> Result<SubstrateNode, Error> {
+        let mut cmd = Command::new(self.binary_path);
+
+        cmd.env("RUST_LOG", "info")
+            .stdout(process::Stdio::piped())
+            .stderr(process::Stdio::piped())
+            .arg("--dev")
+            .arg("--port=0");
+
+        for (key, val) in self.custom_flags {
+            let arg = match val {
+                Some(val) => format!("--{key}={val}"),
+                None => format!("--{key}"),
+            };
+            cmd.arg(arg);
+        }
+
+        let mut proc = cmd.spawn().map_err(Error::Io)?;
+
+        // Wait for RPC port to be logged (it's logged to stderr).
+        let stderr = proc.stderr.take().unwrap();
+        let ws_port =
+            try_find_substrate_port_from_output(stderr).ok_or(Error::CouldNotExtractPort)?;
+
+        Ok(SubstrateNode { proc, ws_port })
+    }
+}
+
+pub struct SubstrateNode {
+    proc: process::Child,
+    ws_port: u16,
+}
+
+impl SubstrateNode {
+    /// Configure and spawn a new [`SubstrateNode`].
+    pub fn builder() -> SubstrateNodeBuilder {
+        SubstrateNodeBuilder::new()
+    }
+
+    /// Return the ID of the running process.
+    pub fn id(&self) -> u32 {
+        self.proc.id()
+    }
+
+    /// Return the port that WS connections are accepted on.
+    pub fn ws_port(&self) -> u16 {
+        self.ws_port
+    }
+
+    /// Kill the process.
+    pub fn kill(&mut self) -> std::io::Result<()> {
+        self.proc.kill()
+    }
+}
+
+impl Drop for SubstrateNode {
+    fn drop(&mut self) {
+        let _ = self.kill();
+    }
+}
+
+// Consume a stderr reader from a spawned substrate command and
+// locate the port number that is logged out to it.
+fn try_find_substrate_port_from_output(r: impl Read + Send + 'static) -> Option<u16> {
+    BufReader::new(r).lines().take(50).find_map(|line| {
+        let line = line.expect("failed to obtain next line from stdout for port discovery");
+
+        // does the line contain our port (we expect this specific output from substrate).
+        let line_end = line
+            // oldest message:
+            .rsplit_once("Listening for new connections on 127.0.0.1:")
+            // slightly newer message:
+            .or_else(|| line.rsplit_once("Running JSON-RPC WS server: addr=127.0.0.1:"))
+            // newest message (jsonrpsee merging http and ws servers):
+            .or_else(|| line.rsplit_once("Running JSON-RPC server: addr=127.0.0.1:"))
+            .map(|(_, port_str)| port_str)?;
+
+        // trim non-numeric chars from the end of the port part of the line.
+        let port_str = line_end.trim_end_matches(|b: char| !b.is_ascii_digit());
+
+        // expect to have a number here (the chars after '127.0.0.1:') and parse them into a u16.
+        let port_num = port_str
+            .parse()
+            .unwrap_or_else(|_| panic!("valid port expected for log line, got '{port_str}'"));
+
+        Some(port_num)
+    })
+}

testing/test-runtime/Cargo.toml

@@ -6,11 +6,11 @@ publish = false
 
 [dependencies]
 subxt = { path = "../../subxt" }
-sp-runtime = "23.0.0"
-codec = { package = "parity-scale-codec", version = "3.0.0", default-features = false, features = ["derive", "full", "bit-vec"] }
 
 [build-dependencies]
-subxt = { path = "../../subxt" }
+substrate-runner = { path = "../substrate-runner" }
+impl-serde = "0.4.0"
+serde = "1.0.160"
 tokio = { version = "1.27", features = ["macros", "rt-multi-thread"] }
 which = "4.4.0"
 jsonrpsee = { version = "0.16.0", features = ["async-client", "client-ws-transport"] }

testing/test-runtime/build.rs

@@ -2,14 +2,8 @@
 // This file is dual-licensed as Apache-2.0 or GPL-3.0.
 // see LICENSE for license details.
 
-use std::{
+use std::{env, fs, path::Path};
-    env, fs,
+use substrate_runner::{Error as SubstrateNodeError, SubstrateNode};
-    net::TcpListener,
-    ops::{Deref, DerefMut},
-    path::Path,
-    process::Command,
-    thread, time,
-};
 
 static SUBSTRATE_BIN_ENV_VAR: &str = "SUBSTRATE_NODE_PATH";
 
@@ -22,19 +16,15 @@ async fn run() {
     // Select substrate binary to run based on env var.
     let substrate_bin = env::var(SUBSTRATE_BIN_ENV_VAR).unwrap_or_else(|_| "substrate".to_owned());
 
-    // Run binary.
+    let mut node_builder = SubstrateNode::builder();
-    let port = next_open_port().expect("Cannot spawn substrate: no available ports");
+    node_builder.binary_path(substrate_bin.clone());
-    let cmd = Command::new(&substrate_bin)
+
-        .arg("--dev")
+    let node = match node_builder.spawn() {
-        .arg("--tmp")
+        Ok(node) => node,
-        .arg(format!("--ws-port={port}"))
+        Err(SubstrateNodeError::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => {
-        .spawn();
-    let mut cmd = match cmd {
-        Ok(cmd) => KillOnDrop(cmd),
-        Err(ref e) if e.kind() == std::io::ErrorKind::NotFound => {
             panic!(
                 "A substrate binary should be installed on your path for testing purposes. \
-            See https://github.com/paritytech/subxt/tree/master#integration-testing"
+                 See https://github.com/paritytech/subxt/tree/master#integration-testing"
             )
         }
         Err(e) => {
@@ -42,35 +32,20 @@ async fn run() {
         }
     };
 
-    // Download metadata from binary; retry until successful, or a limit is hit.
+    let port = node.ws_port();
-    let metadata_bytes: subxt::rpc::types::Bytes = {
-        const MAX_RETRIES: usize = 6;
-        let mut retries = 0;
 
-        loop {
+    // Download metadata from binary. Avoid Subxt dep on `subxt::rpc::types::Bytes`and just impl here.
-            if retries >= MAX_RETRIES {
+    // This may at least prevent this script from running so often (ie whenever we change Subxt).
-                panic!("Cannot connect to substrate node after {retries} retries");
+    #[derive(serde::Deserialize)]
-            }
+    pub struct Bytes(#[serde(with = "impl_serde::serialize")] pub Vec<u8>);
-
+    let metadata_bytes: Bytes = {
-            // It might take a while for substrate node that spin up the RPC server.
+        use client::ClientT;
-            // Thus, the connection might get rejected a few times.
+        client::build(&format!("ws://localhost:{port}"))
-            use client::ClientT;
+            .await
-            let res = match client::build(&format!("ws://localhost:{port}")).await {
+            .unwrap_or_else(|e| panic!("Failed to connect to node: {e}"))
-                Ok(c) => c.request("state_getMetadata", client::rpc_params![]).await,
+            .request("state_getMetadata", client::rpc_params![])
-                Err(e) => Err(e),
+            .await
-            };
+            .unwrap_or_else(|e| panic!("Failed to obtain metadata from node: {e}"))
-
-            match res {
-                Ok(res) => {
-                    let _ = cmd.kill();
-                    break res;
-                }
-                _ => {
-                    thread::sleep(time::Duration::from_secs(1 << retries));
-                    retries += 1;
-                }
-            };
-        }
     };
 
     // Save metadata to a file:
@@ -110,43 +85,6 @@ async fn run() {
     println!("cargo:rerun-if-changed=build.rs");
 }
 
-/// Returns the next open port, or None if no port found.
-fn next_open_port() -> Option<u16> {
-    match TcpListener::bind(("127.0.0.1", 0)) {
-        Ok(listener) => {
-            if let Ok(address) = listener.local_addr() {
-                Some(address.port())
-            } else {
-                None
-            }
-        }
-        Err(_) => None,
-    }
-}
-
-/// If the substrate process isn't explicitly killed on drop,
-/// it seems that panics that occur while the command is running
-/// will leave it running and block the build step from ever finishing.
-/// Wrapping it in this prevents this from happening.
-struct KillOnDrop(std::process::Child);
-
-impl Deref for KillOnDrop {
-    type Target = std::process::Child;
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-impl DerefMut for KillOnDrop {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0
-    }
-}
-impl Drop for KillOnDrop {
-    fn drop(&mut self) {
-        let _ = self.0.kill();
-    }
-}
-
 // Use jsonrpsee to obtain metadata from the node.
 mod client {
     pub use jsonrpsee::{