f08a9cadb6
A replay consumer now has a 5s idle heartbeat, and the row stream ends at the first consumer error instead of skipping it. When the server drops the consumer, as on a restart, two missed heartbeats end the SSE response and the browser reconnects with Last-Event-ID. Before, the response stayed open and delivered nothing. A Last-Event-ID above the stream's newest sequence now replays as a fresh reader. The server accepts a start past the end and delivers nothing until the stream reaches it, so a stale id stalled the replay. swarm-controller's stream tests run against a real nats-server on loopback (`test_nats`, binary from HIVE_NATS_SERVER, exported by checks.nix and devshell.nix). They cover: ensuring the stream twice, an existing stream with other limits being opened as it is, the replay arriving before live rows, a reconnect getting every later row exactly once, and a dropped consumer ending the row stream. The limits test now compares against the subagent stream's own config. Two comments that described the agent's terminal route as live-only are corrected.
66 lines
2.4 KiB
Rust
66 lines
2.4 KiB
Rust
//! A throwaway `nats-server` with `JetStream`, for tests whose subject is the
|
|
//! server's own behaviour: stream creation, consumer delivery order.
|
|
//!
|
|
//! The binary comes from [`NATS_SERVER_ENV`], which `nix/checks.nix` and
|
|
//! `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::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.
|
|
pub(crate) struct TestNats {
|
|
child: Child,
|
|
store: PathBuf,
|
|
url: String,
|
|
}
|
|
|
|
impl TestNats {
|
|
pub(crate) fn start() -> Self {
|
|
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 child = Command::new(&bin)
|
|
.args(["-a", "127.0.0.1", "-p", &port.to_string(), "-js", "-sd"])
|
|
.arg(&store)
|
|
.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}"),
|
|
}
|
|
}
|
|
|
|
/// 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 {
|
|
return client;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
panic!("nats-server at {} never accepted a connection", self.url);
|
|
}
|
|
}
|
|
|
|
impl Drop for TestNats {
|
|
fn drop(&mut self) {
|
|
let _ = self.child.kill();
|
|
let _ = self.child.wait();
|
|
let _ = std::fs::remove_dir_all(&self.store);
|
|
}
|
|
}
|