imporove Monerod spawn stability (#66)

* output monerod logs when a thread panics

* always spawn a monerod don't attempt to re-use

* set zmq port and `non-interactive`

* check monerod has started before test

* remove test panic

* review changes

Cargo.lock

@@ -608,7 +608,6 @@ dependencies = [
  "futures",
  "monero-p2p",
  "monero-wire",
- "rand",
  "reqwest",
  "tar",
  "tokio",

p2p/monero-p2p/tests/handshake.rs

@@ -111,11 +111,7 @@ async fn handshake_cuprate_to_monerod() {
     let semaphore = Arc::new(Semaphore::new(10));
     let permit = semaphore.acquire_owned().await.unwrap();
 
-    let (monerod, _) = monerod(
-        vec!["--fixed-difficulty=1".into(), "--out-peers=0".into()],
-        false,
-    )
-    .await;
+    let monerod = monerod(["--fixed-difficulty=1", "--out-peers=0"]).await;
 
     let our_basic_node_data = BasicNodeData {
         my_port: 0,
@@ -141,7 +137,7 @@ async fn handshake_cuprate_to_monerod() {
         .await
         .unwrap()
         .call(ConnectRequest {
-            addr: monerod,
+            addr: monerod.p2p_addr(),
             permit,
         })
         .await

test-utils/Cargo.toml

@@ -15,8 +15,6 @@ bytes = { workspace = true, features = ["std"] }
 
 borsh = { workspace = true, features = ["derive"]}
 
-rand = { workspace = true, features = ["std", "std_rng"] }
-
 [target.'cfg(unix)'.dependencies]
 tar = "0.4.40"
 bzip2 = "0.4.4"

test-utils/src/monerod.rs

@@ -4,169 +4,151 @@
 //! this to test compatibility with monerod.
 //!
 use std::{
-    collections::HashMap,
-    net::{IpAddr, Ipv4Addr, SocketAddr},
-    path::PathBuf,
-    process::Stdio,
-    sync::OnceLock,
+    ffi::OsStr,
+    io::Read,
+    net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener},
+    process::{Child, Command, Stdio},
+    str::from_utf8,
+    thread::panicking,
     time::Duration,
 };
 
-use rand::Rng;
-use tokio::{
-    process::{Child, Command},
-    sync::{mpsc, oneshot},
-};
+use tokio::{task::yield_now, time::timeout};
 
 mod download;
 
-const LOCAL_HOST: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+const LOCALHOST: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 const MONEROD_VERSION: &str = "v0.18.3.1";
+const MONEROD_STARTUP_TEXT: &str =
+    "The daemon will start synchronizing with the network. This may take a long time to complete.";
 
-#[allow(clippy::type_complexity)]
-static MONEROD_HANDLER_CHANNEL: OnceLock<
-    mpsc::Sender<(MoneroDRequest, oneshot::Sender<(SocketAddr, SocketAddr)>)>,
-> = OnceLock::new();
+const MONEROD_SHUTDOWN_TEXT: &str = "Stopping cryptonote protocol";
 
-/// Spawns monerod and returns the p2p address and rpc address.
-///
-/// When spawning monerod, this module will try to use an already spawned instance to reduce the amount
-/// of instances that need to be spawned.
+/// Spawns monerod and returns [`SpawnedMoneroD`].
 ///
 /// This function will set `regtest` and the P2P/ RPC ports so these can't be included in the flags.
-pub async fn monerod(flags: Vec<String>, mutable: bool) -> (SocketAddr, SocketAddr) {
-    // TODO: sort flags so the same flags in a different order will give the same monerod?
+pub async fn monerod<T: AsRef<OsStr>>(flags: impl IntoIterator<Item = T>) -> SpawnedMoneroD {
+    let path_to_monerod = download::check_download_monerod().await.unwrap();
 
-    // We only actually need these channels on first run so this might be wasteful
-    let (tx, rx) = mpsc::channel(3);
-    let mut should_spawn = false;
+    let rpc_port = get_available_port(&[]);
+    let p2p_port = get_available_port(&[rpc_port]);
+    let zmq_port = get_available_port(&[rpc_port, p2p_port]);
 
-    let monero_handler_tx = MONEROD_HANDLER_CHANNEL.get_or_init(|| {
-        should_spawn = true;
-        tx
-    });
-
-    if should_spawn {
-        // If this call was the first call to start a monerod instance then start the handler.
-        let manager = MoneroDManager::new().await;
-        tokio::task::spawn(manager.run(rx));
-    }
-
-    let (tx, rx) = oneshot::channel();
-
-    monero_handler_tx
-        .send((MoneroDRequest { mutable, flags }, tx))
-        .await
+    // TODO: set a random DB location
+    let mut monerod = Command::new(path_to_monerod)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .args(flags)
+        .arg("--regtest")
+        .arg("--log-level=2")
+        .arg(format!("--p2p-bind-port={}", p2p_port))
+        .arg(format!("--rpc-bind-port={}", rpc_port))
+        .arg(format!("--zmq-rpc-bind-port={}", zmq_port))
+        .arg("--non-interactive")
+        .spawn()
         .unwrap();
 
-    // Give monerod some time to start
-    tokio::time::sleep(Duration::from_secs(5)).await;
-    rx.await.unwrap()
+    let mut logs = String::new();
+
+    timeout(Duration::from_secs(30), async {
+        loop {
+            let mut next_str = [0];
+            let _ = monerod
+                .stdout
+                .as_mut()
+                .unwrap()
+                .read(&mut next_str)
+                .unwrap();
+
+            logs.push_str(from_utf8(&next_str).unwrap());
+
+            if logs.contains(MONEROD_SHUTDOWN_TEXT) {
+                panic!("Failed to start monerod, logs: \n {logs}");
+            }
+
+            if logs.contains(MONEROD_STARTUP_TEXT) {
+                break;
+            }
+            // this is blocking code but as this is for tests performance isn't a priority. However we should still yield so
+            // the timeout works.
+            yield_now().await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("Failed to start monerod in time, logs: {logs}"));
+
+    SpawnedMoneroD {
+        process: monerod,
+        rpc_port,
+        p2p_port,
+        start_up_logs: logs,
+    }
 }
 
-/// A request sent to get an address to a monerod instance.
-struct MoneroDRequest {
-    /// Whether we plan to change the state of the spawned monerod's blockchain.
-    mutable: bool,
-    /// Start flags to start monerod with.
-    flags: Vec<String>,
+fn get_available_port(already_taken: &[u16]) -> u16 {
+    loop {
+        // Using `0` makes the OS return a random available port.
+        let port = TcpListener::bind("127.0.0.1:0")
+            .unwrap()
+            .local_addr()
+            .unwrap()
+            .port();
+
+        if !already_taken.contains(&port) {
+            return port;
+        }
+    }
 }
 
 /// A struct representing a spawned monerod.
-struct SpawnedMoneroD {
-    /// A marker for if the test that spawned this monerod is going to mutate it.
-    mutable: bool,
+pub struct SpawnedMoneroD {
     /// A handle to the monerod process, monerod will be stopped when this is dropped.
-    #[allow(dead_code)]
     process: Child,
     /// The RPC port of the monerod instance.
     rpc_port: u16,
     /// The P2P port of the monerod instance.
     p2p_port: u16,
+
+    start_up_logs: String,
 }
 
-/// A manger of spawned monerods.
-struct MoneroDManager {
-    /// A map of start flags to monerods.
-    monerods: HashMap<Vec<String>, Vec<SpawnedMoneroD>>,
-    /// The path to the monerod binary.
-    path_to_monerod: PathBuf,
+impl SpawnedMoneroD {
+    /// Returns the p2p port of the spawned monerod
+    pub fn p2p_addr(&self) -> SocketAddr {
+        SocketAddr::new(LOCALHOST, self.p2p_port)
+    }
+
+    /// Returns the RPC port of the spawned monerod
+    pub fn rpc_port(&self) -> SocketAddr {
+        SocketAddr::new(LOCALHOST, self.rpc_port)
+    }
 }
 
-impl MoneroDManager {
-    pub async fn new() -> Self {
-        let path_to_monerod = download::check_download_monerod().await.unwrap();
-
-        Self {
-            monerods: Default::default(),
-            path_to_monerod,
-        }
+impl Drop for SpawnedMoneroD {
+    fn drop(&mut self) {
+        if self.process.kill().is_err() {
+            println!("Failed to kill monerod, process id: {}", self.process.id())
         }
 
-    pub async fn run(
-        mut self,
-        mut rx: mpsc::Receiver<(MoneroDRequest, oneshot::Sender<(SocketAddr, SocketAddr)>)>,
-    ) {
-        while let Some((req, tx)) = rx.recv().await {
-            let (p2p_port, rpc_port) = self.get_monerod_with_flags(req.flags, req.mutable);
-            let _ = tx.send((
-                SocketAddr::new(LOCAL_HOST, p2p_port),
-                SocketAddr::new(LOCAL_HOST, rpc_port),
-            ));
-        }
+        if panicking() {
+            // If we are panicking then a test failed so print monerod's logs.
+
+            let mut out = String::new();
+
+            if self
+                .process
+                .stdout
+                .as_mut()
+                .unwrap()
+                .read_to_string(&mut out)
+                .is_err()
+            {
+                println!("Failed to get monerod's logs.");
             }
 
-    /// Tries to get a current monerod instance or spans one if there is not an appropriate one to use.
-    /// Returns the p2p port and then the RPC port of the spawned monerd.
-    fn get_monerod_with_flags(&mut self, flags: Vec<String>, mutable: bool) -> (u16, u16) {
-        // If we need to mutate monerod's blockchain then we can't reuse one.
-        if !mutable {
-            if let Some(monerods) = &self.monerods.get(&flags) {
-                for monerod in monerods.iter() {
-                    if !monerod.mutable {
-                        return (monerod.p2p_port, monerod.rpc_port);
+            println!("-----START-MONEROD-LOGS-----");
+            println!("{}{out}", self.start_up_logs);
+            println!("------END-MONEROD-LOGS------");
         }
     }
-            }
-        }
-
-        let mut rng = rand::thread_rng();
-        // Use random ports and *hope* we don't get a collision (TODO: just keep a counter and increment?)
-        let rpc_port: u16 = rng.gen_range(1500..u16::MAX);
-        let p2p_port: u16 = rng.gen_range(1500..u16::MAX);
-
-        // TODO: set a different DB location per node
-        let monerod = Command::new(&self.path_to_monerod)
-            .stdout(Stdio::null())
-            .stdin(Stdio::piped())
-            .args(&flags)
-            .arg("--regtest")
-            .arg(format!("--p2p-bind-port={}", p2p_port))
-            .arg(format!("--rpc-bind-port={}", rpc_port))
-            .kill_on_drop(true)
-            .spawn()
-            .unwrap();
-
-        let spawned_monerod = SpawnedMoneroD {
-            mutable,
-            process: monerod,
-            rpc_port,
-            p2p_port,
-        };
-
-        self.monerods
-            .entry(flags.clone())
-            .or_default()
-            .push(spawned_monerod);
-        let Some(monerods) = self.monerods.get(&flags) else {
-            unreachable!()
-        };
-
-        for monerod in monerods {
-            if !monerod.mutable {
-                return (monerod.p2p_port, monerod.rpc_port);
-            }
-        }
-        unreachable!()
-    }
 }

test-utils/src/monerod/download.rs

@@ -14,9 +14,13 @@ use std::{
 #[cfg(unix)]
 use bytes::Buf;
 use reqwest::{get, Error as ReqError};
+use tokio::sync::Mutex;
 
 use super::MONEROD_VERSION;
 
+/// A mutex to make sure only one thread at a time downloads monerod.
+static DOWNLOAD_MONEROD_MUTEX: Mutex<()> = Mutex::const_new(());
+
 /// Returns the file name to download and the expected extracted folder name.
 fn file_name(version: &str) -> (String, String) {
     let download_file = match (OS, ARCH) {
@@ -80,6 +84,9 @@ fn find_target() -> PathBuf {
 
 /// Checks if we have monerod or downloads it if we don't and then returns the path to it.
 pub async fn check_download_monerod() -> Result<PathBuf, ReqError> {
+    // make sure no other threads are downloading monerod at the same time.
+    let _guard = DOWNLOAD_MONEROD_MUTEX.lock().await;
+
     let path_to_store = find_target();
 
     let (file_name, dir_name) = file_name(MONEROD_VERSION);