Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: let each test nats-server bind its own port

TestNats picked a port by binding 127.0.0.1:0, dropping the listener
and handing the number to nats-server. Until that server bound it, the
port was free for a concurrent test's bind(0) to return again. Both
servers were then started on it (with the same store dir, since that was
keyed by pid and port); the loser exited silently and both tests'
clients reached the winner. term_stream's tests all use the stream
term-main-iris, and one of them creates it with max_message_size 4096,
so a test sharing its server got "stream name already in use with a
different configuration" (10058) from open_or_create's
get-then-create.

The server now binds a port of its choice (-p -1) and announces it in a
ports file, and each server has its own tempdir for the store and that
file, so a test can only reach the server it started.

Fixes #5261
This commit is contained in:
atlas 2026-10-14 16:27:16 +02:00
commit 41404598f5
3 changed files with 58 additions and 21 deletions

1
Cargo.lock generated
View file

@ -4880,6 +4880,7 @@ dependencies = [
"swarm-matrix-client", "swarm-matrix-client",
"swarm-queue-client", "swarm-queue-client",
"swarm-secret-client", "swarm-secret-client",
"tempfile",
"tokio", "tokio",
"tower", "tower",
"tracing", "tracing",

View file

@ -136,6 +136,8 @@ x509-cert.workspace = true
# tests wait out its five-minute bound on paused time. # tests wait out its five-minute bound on paused time.
[dev-dependencies] [dev-dependencies]
tokio = { workspace = true, features = ["test-util"] } 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. # `ServiceExt::oneshot`, to drive a route through the real router in tests.
tower = { version = "0.5", features = ["util"] } tower = { version = "0.5", features = ["util"] }
# A layer that captures `review_classifier::runs`'s log event in a test. # A layer that captures `review_classifier::runs`'s log event in a test.

View file

@ -5,19 +5,26 @@
//! `nix/devshell.nix` export. A test that needs it panics when it is unset //! `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. //! 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::process::{Child, Command, Stdio};
use std::time::Duration; use std::time::Duration;
/// The variable naming the `nats-server` binary. /// The variable naming the `nats-server` binary.
pub(crate) const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER"; pub(crate) const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER";
/// A running server on a loopback port, with its store in a fresh directory. /// The subdirectory holding nothing but the server's ports file.
/// Dropping it kills the server and removes the store. 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 { pub(crate) struct TestNats {
child: Child, child: Child,
store: PathBuf, dir: tempfile::TempDir,
url: String,
} }
impl TestNats { impl TestNats {
@ -25,42 +32,69 @@ impl TestNats {
let bin = std::env::var(NATS_SERVER_ENV).unwrap_or_else(|_| { 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") 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 dir = tempfile::Builder::new()
let port = std::net::TcpListener::bind("127.0.0.1:0") .prefix("test-nats-")
.and_then(|l| l.local_addr()) .tempdir()
.expect("a free loopback port") .expect("a fresh directory for the server");
.port(); std::fs::create_dir(dir.path().join(PORTS_DIR)).expect("the ports file directory");
let store = std::env::temp_dir().join(format!("test-nats-{}-{port}", std::process::id()));
let child = Command::new(&bin) let child = Command::new(&bin)
.args(["-a", "127.0.0.1", "-p", &port.to_string(), "-js", "-sd"]) .args(["-a", "127.0.0.1", "-p", "-1", "-js", "-sd"])
.arg(&store) .arg(dir.path().join("store"))
.arg("--ports_file_dir")
.arg(dir.path().join(PORTS_DIR))
.stdout(Stdio::null()) .stdout(Stdio::null())
.stderr(Stdio::null()) .stderr(Stdio::null())
.spawn() .spawn()
.unwrap_or_else(|e| panic!("starting {bin}: {e}")); .unwrap_or_else(|e| panic!("starting {bin}: {e}"));
Self { Self { child, dir }
child,
store,
url: format!("nats://127.0.0.1:{port}"),
}
} }
/// A client connected to the server, once it accepts connections. /// A client connected to the server, once it accepts connections.
pub(crate) async fn client(&self) -> async_nats::Client { pub(crate) async fn client(&self) -> async_nats::Client {
for _ in 0..100 { 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; return client;
} }
tokio::time::sleep(Duration::from_millis(100)).await; 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 `<exe>_<pid>.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<String> {
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 { impl Drop for TestNats {
fn drop(&mut self) { fn drop(&mut self) {
let _ = self.child.kill(); let _ = self.child.kill();
let _ = self.child.wait(); 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<TestNats> = (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:?}");
} }
} }