diff --git a/Cargo.lock b/Cargo.lock index 08d2a4c6..4a496e0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4880,6 +4880,7 @@ dependencies = [ "swarm-matrix-client", "swarm-queue-client", "swarm-secret-client", + "tempfile", "tokio", "tower", "tracing", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index e99d1243..8ae5ca1f 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -136,6 +136,8 @@ x509-cert.workspace = true # tests wait out its five-minute bound on paused time. [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +# A fresh directory per test `nats-server`, for its store and ports file. +tempfile = "3" # `ServiceExt::oneshot`, to drive a route through the real router in tests. tower = { version = "0.5", features = ["util"] } # A layer that captures `review_classifier::runs`'s log event in a test. diff --git a/swarm-controller/src/test_nats.rs b/swarm-controller/src/test_nats.rs index 36e93d0f..f24f5393 100644 --- a/swarm-controller/src/test_nats.rs +++ b/swarm-controller/src/test_nats.rs @@ -5,19 +5,26 @@ //! `nix/devshell.nix` export. A test that needs it panics when it is unset //! rather than skipping, so it never shows green without having run. -use std::path::PathBuf; +use std::path::Path; use std::process::{Child, Command, Stdio}; use std::time::Duration; /// The variable naming the `nats-server` binary. pub(crate) const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER"; -/// A running server on a loopback port, with its store in a fresh directory. -/// Dropping it kills the server and removes the store. +/// The subdirectory holding nothing but the server's ports file. +const PORTS_DIR: &str = "ports"; + +/// A running server on a loopback port it bound itself, with its store and +/// ports file under a directory of its own. Dropping it kills the server and +/// removes the directory. +/// +/// The port is the server's choice, read back from its ports file: a port +/// picked here and handed over would be free for a concurrent test to pick +/// too, and both tests would then talk to whichever server won the bind. pub(crate) struct TestNats { child: Child, - store: PathBuf, - url: String, + dir: tempfile::TempDir, } impl TestNats { @@ -25,42 +32,69 @@ impl TestNats { let bin = std::env::var(NATS_SERVER_ENV).unwrap_or_else(|_| { panic!("{NATS_SERVER_ENV} is exported by nix/checks.nix and nix/devshell.nix") }); - // Bound and released to learn a free port; the server binds it next. - let port = std::net::TcpListener::bind("127.0.0.1:0") - .and_then(|l| l.local_addr()) - .expect("a free loopback port") - .port(); - let store = std::env::temp_dir().join(format!("test-nats-{}-{port}", std::process::id())); + let dir = tempfile::Builder::new() + .prefix("test-nats-") + .tempdir() + .expect("a fresh directory for the server"); + std::fs::create_dir(dir.path().join(PORTS_DIR)).expect("the ports file directory"); let child = Command::new(&bin) - .args(["-a", "127.0.0.1", "-p", &port.to_string(), "-js", "-sd"]) - .arg(&store) + .args(["-a", "127.0.0.1", "-p", "-1", "-js", "-sd"]) + .arg(dir.path().join("store")) + .arg("--ports_file_dir") + .arg(dir.path().join(PORTS_DIR)) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .unwrap_or_else(|e| panic!("starting {bin}: {e}")); - Self { - child, - store, - url: format!("nats://127.0.0.1:{port}"), - } + Self { child, dir } } /// A client connected to the server, once it accepts connections. pub(crate) async fn client(&self) -> async_nats::Client { for _ in 0..100 { - if let Ok(client) = async_nats::connect(&self.url).await { + if let Some(url) = announced_url(&self.dir.path().join(PORTS_DIR)) + && let Ok(client) = async_nats::connect(&url).await + { return client; } tokio::time::sleep(Duration::from_millis(100)).await; } - panic!("nats-server at {} never accepted a connection", self.url); + panic!( + "nats-server in {} never accepted a connection", + self.dir.path().display() + ); } } +/// The client URL in the `_.ports` file, the one file the server +/// writes into `dir` once it listens, or `None` while there is none to read. +fn announced_url(dir: &Path) -> Option { + let ports = std::fs::read_dir(dir).ok()?.next()?.ok()?; + let ports: serde_json::Value = + serde_json::from_slice(&std::fs::read(ports.path()).ok()?).ok()?; + ports["nats"][0].as_str().map(str::to_owned) +} + impl Drop for TestNats { fn drop(&mut self) { let _ = self.child.kill(); let _ = self.child.wait(); - let _ = std::fs::remove_dir_all(&self.store); + } +} + +#[cfg(test)] +mod tests { + use super::TestNats; + + /// Servers started back to back are each their own, so no test reaches + /// another test's streams. + #[tokio::test] + async fn each_test_nats_is_its_own_server() { + let servers: Vec = (0..4).map(|_| TestNats::start()).collect(); + let mut ids = std::collections::BTreeSet::new(); + for nats in &servers { + ids.insert(nats.client().await.server_info().server_id); + } + assert_eq!(ids.len(), servers.len(), "server ids: {ids:?}"); } }