Skip to content

[adapters] NATS: isolate and verify test server startup #6741

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ryzhyk merged 1 commit into from
Aug 2, 2026
Merged

[adapters] NATS: isolate and verify test server startup #6741

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
327 changes: 247 additions & 80 deletions crates/adapters/src/transport/nats/input/test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,23 +23,80 @@ impl NatsTestRecord {
deserialize_without_context!(NatsTestRecord);

mod util {
use crate::test::wait;
use anyhow::{Result as AnyResult, anyhow};
use async_nats::{self, Client, jetstream};
use serde::{Deserialize, Serialize};
use std::env;
use std::fs;
use std::path::Path;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
use tempfile::TempDir;

/// How long to wait for a spawned `nats-server` to bind its client port.
const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);

/// A running `nats-server`, killed when the guard is dropped.
///
/// Each server owns a private scratch directory for its JetStream store and
/// its ports file. Sharing either across servers breaks tests that run
/// concurrently:
///
/// - Servers starting at the same moment into one JetStream store directory
/// race on `mkdir $G/streams`, and the loser dies with
/// `Can't start JetStream`. The window is open only while that directory
/// does not yet exist, so the race showed up on fresh CI containers and
/// almost never on a developer machine.
/// - A ports file is named `nats-server_<pid>.ports`, and a server killed by
/// SIGKILL leaves its file behind. A later server assigned the same pid
/// then finds the stale file and reports the dead server's port.
pub struct ProcessKillGuard {
process: Child,
/// Holds the JetStream store, the ports file, and the server log.
/// Deleted after `process` has been killed and reaped.
_scratch: TempDir,
store_dir: PathBuf,
ports_file: PathBuf,
log_path: PathBuf,
}

impl ProcessKillGuard {
fn new(process: Child) -> Self {
Self { process }
/// Wait until the server reports the address it bound.
///
/// `nats-server` writes its ports file only after binding the client
/// port, so the file proves *this* server owns the port. A TCP probe
/// would not: it also succeeds against an unrelated server that already
/// held the port.
fn wait_until_ready(&mut self) -> AnyResult<String> {
let deadline = Instant::now() + STARTUP_TIMEOUT;
loop {
// Parse before checking liveness, so a server that became ready
// and then died still counts as started. Parsing on every poll
// also tolerates reading the file while it is still being
// written: a partial read simply does not parse yet.
if let Some(addr) = fs::read_to_string(&self.ports_file)
.ok()
.and_then(|content| parse_ports_file(&content))
{
return Ok(addr);
}
if let Some(status) = self.process.try_wait()? {
return Err(anyhow!(
"nats-server exited with {status} before it became ready; log:\n{}",
self.log()
));
}
if Instant::now() >= deadline {
return Err(anyhow!(
"nats-server did not report a bound port within {STARTUP_TIMEOUT:?}; log:\n{}",
self.log()
));
}
std::thread::sleep(Duration::from_millis(10));
}
}

fn log(&self) -> String {
fs::read_to_string(&self.log_path).unwrap_or_else(|e| format!("<unreadable: {e}>"))
}
}

Expand All @@ -50,106 +107,114 @@ mod util {
}
}

pub async fn wait_for_nats_ready(addr: &str, timeout: Duration) -> anyhow::Result<Client> {
let deadline = Instant::now() + timeout;
loop {
match async_nats::connect(addr).await {
Ok(client) => return Ok(client),
Err(_) if Instant::now() < deadline => {
tokio::time::sleep(Duration::from_millis(100)).await
}
Err(e) => return Err(anyhow::anyhow!("Timeout waiting for NATS: {e}")),
}
/// Extract the first client address from a `nats-server` ports file.
/// Returns `None` while the file is empty or incomplete.
fn parse_ports_file(content: &str) -> Option<String> {
#[derive(Deserialize)]
struct PortsData {
nats: Vec<String>,
}
}

pub fn start_nats_and_get_address() -> AnyResult<(ProcessKillGuard, String)> {
const MAX_ATTEMPTS: usize = 2;
let mut last_error: Option<anyhow::Error> = None;
for attempt in 1..=MAX_ATTEMPTS {
match start_nats_and_get_address_once() {
Ok(result) => return Ok(result),
Err(error) => {
last_error = Some(error);
if attempt < MAX_ATTEMPTS {
std::thread::sleep(Duration::from_millis(250));
}
}
}
}

Err(last_error.expect("at least one attempt should have failed"))
serde_json::from_str::<PortsData>(content)
.ok()?
.nats
.into_iter()
.next()
}

fn start_nats_and_get_address_once() -> AnyResult<(ProcessKillGuard, String)> {
/// Spawn a JetStream-enabled `nats-server` on `port` and wait until it binds.
///
/// `port` is a `nats-server` port argument: a TCP port, or `-1` to let the
/// operating system choose one. Returns the guard and the bound address.
fn spawn_nats_server(port: &str) -> AnyResult<(ProcessKillGuard, String)> {
let nats_ip_addr = "127.0.0.1";
const RANDOM_PORT: &str = "-1";

let temp_dir = env::temp_dir();
let port_file_dir = temp_dir.join("nats_ports");
let scratch = TempDir::new()?;
let store_dir = scratch.path().join("jetstream");
let ports_file_dir = scratch.path().join("ports");
let log_path = scratch.path().join("nats-server.log");
fs::create_dir_all(&store_dir)?;
fs::create_dir_all(&ports_file_dir)?;

fs::create_dir_all(&port_file_dir)?;

let child = Command::new("nats-server")
let log = fs::File::create(&log_path)?;
let process = Command::new("nats-server")
.arg("-a")
.arg(nats_ip_addr)
.arg("-p")
.arg(RANDOM_PORT)
.arg(port)
.arg("--ports_file_dir")
.arg(port_file_dir.to_str().unwrap())
.arg(&ports_file_dir)
.arg("--jetstream")
.stdout(Stdio::null())
.stderr(Stdio::null())
.arg("--store_dir")
.arg(&store_dir)
.stdout(Stdio::from(log.try_clone()?))
.stderr(Stdio::from(log))
.spawn()?;

let pid = child.id();
let port_file_path = port_file_dir.join(format!("nats-server_{}.ports", pid));
let ports_file = ports_file_dir.join(format!("nats-server_{}.ports", process.id()));
let mut guard = ProcessKillGuard {
process,
_scratch: scratch,
store_dir,
ports_file,
log_path,
};

let child = ProcessKillGuard::new(child);
// On error the guard is dropped here, which kills the server.
let addr = guard.wait_until_ready()?;
Ok((guard, addr))
}

if wait(|| port_file_path.exists(), 10_000).is_err() {
return Err(anyhow!("Port file was not created within timeout period"));
}
pub fn start_nats_and_get_address() -> AnyResult<(ProcessKillGuard, String)> {
const MAX_ATTEMPTS: usize = 2;
const RANDOM_PORT: &str = "-1";

fn get_address_from_ports_file(file_path: &Path) -> AnyResult<String> {
#[derive(Deserialize)]
struct PortsData {
nats: Vec<String>,
let mut last_error: Option<anyhow::Error> = None;
for attempt in 1..=MAX_ATTEMPTS {
match spawn_nats_server(RANDOM_PORT) {
Ok(result) => return Ok(result),
Err(error) => {
last_error = Some(error);
if attempt < MAX_ATTEMPTS {
std::thread::sleep(Duration::from_millis(250));
}
}
}

let port_content = fs::read_to_string(file_path)?;
let ports_data: PortsData = serde_json::from_str(&port_content)
.map_err(|_| anyhow!("Could not parse ports file"))?;

ports_data
.nats
.into_iter()
.next()
.ok_or(anyhow!("No NATS addresses found in port file"))
}

let nats_addr = get_address_from_ports_file(&port_file_path)?;

Ok((child, nats_addr))
Err(last_error.expect("at least one attempt should have failed"))
}

/// Start a server on a specific port, as tests do when restarting a killed
/// server at the address its connector is configured with.
///
/// Fails if the port is no longer available. Nothing reserves the port while
/// the server is down, so an unrelated process may take it; failing here
/// reports that directly instead of letting a later action time out against
/// a foreign server.
pub fn start_nats_on_port(port: u16) -> AnyResult<(ProcessKillGuard, String)> {
let nats_ip_addr = "127.0.0.1";
let (guard, addr) = spawn_nats_server(&port.to_string())
.map_err(|e| e.context(format!("failed to restart nats-server on port {port}")))?;

let child = Command::new("nats-server")
.arg("-a")
.arg(nats_ip_addr)
.arg("-p")
.arg(port.to_string())
.arg("--jetstream")
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
if !addr.ends_with(&format!(":{port}")) {
return Err(anyhow!(
"nats-server was asked for port {port} but bound {addr}"
));
}
Ok((guard, addr))
}

Ok((
ProcessKillGuard::new(child),
format!("nats://{nats_ip_addr}:{port}"),
))
pub async fn wait_for_nats_ready(addr: &str, timeout: Duration) -> anyhow::Result<Client> {
let deadline = Instant::now() + timeout;
loop {
match async_nats::connect(addr).await {
Ok(client) => return Ok(client),
Err(_) if Instant::now() < deadline => {
tokio::time::sleep(Duration::from_millis(100)).await
}
Err(e) => return Err(anyhow::anyhow!("Timeout waiting for NATS: {e}")),
}
}
}

pub async fn create_stream(nats_url: &str, stream: &str, subject: &str) -> AnyResult<()> {
Expand Down Expand Up @@ -196,4 +261,106 @@ mod util {
js.delete_stream(stream).await?;
Ok(())
}

/// Tests for the server-spawning helpers themselves. Every NATS test depends
/// on them, so a silent failure here surfaces as an unrelated timeout in
/// whichever test happened to run.
#[cfg(test)]
mod tests {
use super::*;

/// A server that cannot bind its port must be reported as a failure.
///
/// `Command::spawn` succeeds even when the server dies immediately after,
/// so returning the guard unchecked left tests talking to whichever
/// process actually held the port.
#[test]
fn start_nats_on_port_fails_when_port_is_taken() {
let (_holder, addr) = start_nats_and_get_address().unwrap();
let port: u16 = addr.rsplit_once(':').unwrap().1.parse().unwrap();

let error = start_nats_on_port(port)
.err()
.expect("starting a second server on a taken port must fail");
let text = format!("{error:#}");
assert!(
text.contains(&format!("port {port}")),
"error should name the port, got: {text}"
);
assert!(
text.contains("address already in use"),
"error should include the server log explaining the failure, got: {text}"
);
}

/// Servers must not share the directories they write into, for the
/// reasons given on [`ProcessKillGuard`].
///
/// Asserted structurally rather than by observing a race: both races are
/// timing- and pid-dependent, so a test that waited to see one would pass
/// while the defect was still present.
#[test]
fn servers_do_not_share_directories() {
let (first, _) = start_nats_and_get_address().unwrap();
let (second, _) = start_nats_and_get_address().unwrap();

assert_ne!(
first.store_dir, second.store_dir,
"servers must not share a JetStream store directory"
);
assert_ne!(
first.ports_file.parent(),
second.ports_file.parent(),
"servers must not share a ports file directory"
);
}

/// Servers started at the same moment must all come up on distinct ports.
#[test]
fn concurrent_servers_start_independently() {
const SERVERS: usize = 16;

let barrier = std::sync::Barrier::new(SERVERS);
let servers: Vec<_> = std::thread::scope(|scope| {
let handles: Vec<_> = (0..SERVERS)
.map(|_| {
scope.spawn(|| {
barrier.wait();
start_nats_and_get_address()
})
})
.collect();
handles
.into_iter()
.map(|handle| handle.join().expect("spawn thread panicked"))
.collect()
});

let mut addrs = Vec::new();
for server in &servers {
match server {
Ok((_guard, addr)) => addrs.push(addr.clone()),
Err(error) => panic!("concurrent server failed to start: {error:#}"),
}
}

addrs.sort();
addrs.dedup();
assert_eq!(addrs.len(), SERVERS, "servers must bind distinct ports");
}

#[test]
fn parse_ports_file_rejects_incomplete_content() {
// A server writes its ports file after startup; polling can observe
// it empty or half-written, which must read as "not ready yet"
// rather than as an error or a bogus address.
assert_eq!(parse_ports_file(""), None);
assert_eq!(parse_ports_file("{\"nats\":[\"nats://127.0"), None);
assert_eq!(parse_ports_file("{\"nats\":[]}"), None);
assert_eq!(
parse_ports_file("{\"nats\":[\"nats://127.0.0.1:4222\"]}"),
Some("nats://127.0.0.1:4222".to_string())
);
}
}
}
Loading