From f08a9cadb63283d93f87754edf2542c4ac458a6d Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 11 Oct 2026 23:25:27 +0200 Subject: [PATCH] swarm-controller: end a dead replay, clamp stale resume ids, test on a real server 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. --- docs/swarm/ui.md | 5 +- nix/checks.nix | 5 + nix/devshell.nix | 3 + swarm-controller/src/agent_state_stream.rs | 10 +- swarm-controller/src/main.rs | 5 +- swarm-controller/src/subagent_term.rs | 6 +- swarm-controller/src/term_stream.rs | 242 +++++++++++++++++++-- swarm-controller/src/test_nats.rs | 66 ++++++ swarm-queue-client/src/agent_term.rs | 50 ++++- swarm-queue-client/src/subagent_term.rs | 5 +- 10 files changed, 357 insertions(+), 40 deletions(-) create mode 100644 swarm-controller/src/test_nats.rs diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 72cbad9d..15eee72b 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -113,7 +113,10 @@ below `$SWARM.term..sub`, so they never overlap. `GET /api/agents//term/stream` creates a short-lived consumer on it that starts 100 rows back, so the SSE stream sends those rows and then each new one. Each event's `id` is the row's stream sequence, and a request with -`Last-Event-ID` resumes after it. Until the stream exists, the route sends +`Last-Event-ID` resumes after it. An id past the newest row of the stream, as +after swarm-controller recreates the stream, starts over as a fresh reader. When the +consumer stops sending its heartbeat, as after a queue restart, the route ends +the response and the browser reconnects. Until the stream exists, the route sends live rows only. diff --git a/nix/checks.nix b/nix/checks.nix index a7c24d60..01b56173 100644 --- a/nix/checks.nix +++ b/nix/checks.nix @@ -222,6 +222,11 @@ in # quietly does nothing is worse than no drift test, because it still # shows up green. HIVE_RESERVED_NAMES = pkgs.lib.concatStringsSep " " (import ./reserved-names.nix); + # swarm-controller's stream and replay tests run against a real + # `nats-server` on loopback (`swarm-controller/src/test_nats.rs`), since + # what they check is the server's own create and delivery behaviour. + # Exported here and in `devshell.nix`; the tests panic when it is unset. + HIVE_NATS_SERVER = "${pkgs.nats-server}/bin/nats-server"; }; # Rustdoc gate. Builds the workspace's docs and turns rustdoc's own diff --git a/nix/devshell.nix b/nix/devshell.nix index 9e3de16c..5c971abf 100644 --- a/nix/devshell.nix +++ b/nix/devshell.nix @@ -11,6 +11,9 @@ # developer's local run and CI would disagree about a test that exists # precisely to stop two spellings of one fact from drifting. HIVE_RESERVED_NAMES = pkgs.lib.concatStringsSep " " (import ./reserved-names.nix); + # The `nats-server` swarm-controller's stream tests start, the same one + # `checks.nix` gives the test derivation. Those tests panic without it. + HIVE_NATS_SERVER = "${pkgs.nats-server}/bin/nats-server"; packages = rust.nativeBuildInputs ++ (with pkgs; [ diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs index ebd02eb4..97893e79 100644 --- a/swarm-controller/src/agent_state_stream.rs +++ b/swarm-controller/src/agent_state_stream.rs @@ -1,11 +1,11 @@ //! `GET /api/agents/{name}/state/stream` — a live SSE relay of one agent's //! turn-state header, sourced straight from the swarm queue. //! -//! Sibling of [`crate::term_stream`] in every structural respect — the -//! reasoning there for keeping the hive out of the path, and for a live tail with no replay, applies -//! unchanged here and is not repeated. What differs is only what the two -//! subjects carry: a terminal row is an event that happened, a header is -//! the agent's current state. +//! Sibling of [`crate::term_stream`]: the reasoning there for keeping the +//! hive out of the path applies unchanged here and is not repeated. Unlike +//! that route this one is a live tail with no replay, because no stream keeps +//! headers. The two subjects also differ in what they carry: a terminal row +//! is an event that happened, a header is the agent's current state. //! //! **That difference is the one thing worth reading twice.** `hive-agent` //! publishes a header on transition and re-sends it once a minute, to a core diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 8d6af936..a2038ed0 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -64,6 +64,8 @@ mod status; mod store; mod subagent_term; mod term_stream; +#[cfg(test)] +mod test_nats; mod vcs_metrics; mod wanted; mod webhook; @@ -2631,7 +2633,8 @@ fn spawn_subagent_streams(status: Option<&Arc>, hives: &[H subagent_term::spawn(status.queue_client(), wanted, hives); } -/// The per-agent background passes: terminal streams and schedule keys. +/// The per-agent background passes: subagent and terminal streams, and +/// schedule keys. /// Lifted out of `main` for `clippy::too_many_lines`. fn spawn_per_agent_upkeep(status: Option<&Arc>, hives: &[HiveEntry]) { spawn_subagent_streams(status, hives); diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 307c304b..47dcdd88 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -15,9 +15,9 @@ //! **No agent-status lookup.** Unlike [`crate::term_stream`], an agent that //! has never reported gets an empty stream, not a 404. //! -//! **Live tail only, output only.** The relay is a core subscription, the -//! same as the agent's own terminal route, so a reader sees rows published -//! while it is attached. Nothing is sent toward a subagent: subagents take no +//! **Live tail only, output only.** The relay is a core subscription, so a +//! reader sees rows published while it is attached; unlike the agent's own +//! terminal route, it replays nothing from the stream. Nothing is sent toward a subagent: subagents take no //! input from the swarm. use std::collections::BTreeSet; diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 47f63c06..6a333cdf 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -117,7 +117,7 @@ pub(crate) async fn stream_agent_term( })?; let stream = match replay(&client, &agent, last_event_id(&headers)).await { - Ok(stream) => stream, + Ok(rows) => rows.map(|row| Ok(row_event(&row))).boxed(), Err(error) => { tracing::warn!(%agent, %error, "term stream: no replay, live rows only"); let subscriber = @@ -130,13 +130,20 @@ pub(crate) async fn stream_agent_term( Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } +/// One row read from the agent's terminal stream. +struct Row { + /// Its stream sequence, when the delivery names one. + seq: Option, + payload: bytes::Bytes, +} + /// The agent's rows from its terminal stream: the replay /// [`agent_term::replay_start`] picks, then every new row. async fn replay( client: &async_nats::Client, agent: &str, resume_after: Option, -) -> Result>, String> { +) -> Result, String> { let js = async_nats::jetstream::new(client.clone()); let name = agent_term::stream_name(agent); let stream = js @@ -164,28 +171,39 @@ async fn replay( .await .map_err(|e| format!("subscribing to {name}: {}", swarm_queue_client::chain(&e)))?; tracing::info!(stream = %name, start, "term stream: client attached"); - let agent = agent.to_owned(); - Ok(messages - .filter_map(move |msg| { - let event = match msg { - Ok(msg) => Some(Ok(row_event(&msg))), + Ok(rows_until_error(messages, agent.to_owned())) +} + +/// The rows of a consumer's deliveries, ending at the first error. A missed +/// heartbeat is such an error, and ending there closes the SSE response, so +/// the browser reconnects with `Last-Event-ID` and gets a fresh consumer. +fn rows_until_error( + messages: impl Stream> + Send + 'static, + agent: String, +) -> BoxStream<'static, Row> { + messages + .scan((), move |(), msg| { + let row = match msg { + Ok(msg) => Some(Row { + seq: msg.info().ok().map(|info| info.stream_sequence), + payload: msg.message.payload, + }), Err(e) => { - tracing::warn!(%agent, error = %e, "term stream: row skipped"); + tracing::warn!(%agent, error = %e, "term stream: consumer failed, ending the stream"); None } }; - std::future::ready(event) + std::future::ready(row) }) - .boxed()) + .boxed() } -/// One row as an SSE event, carrying its stream sequence as the event `id` -/// when the delivery names one. -fn row_event(msg: &async_nats::jetstream::Message) -> Event { - let event = Event::default().data(String::from_utf8_lossy(&msg.payload)); - match msg.info() { - Ok(info) => event.id(info.stream_sequence.to_string()), - Err(_) => event, +/// One row as an SSE event, carrying its stream sequence as the event `id`. +fn row_event(row: &Row) -> Event { + let event = Event::default().data(String::from_utf8_lossy(&row.payload)); + match row.seq { + Some(seq) => event.id(seq.to_string()), + None => event, } } @@ -222,6 +240,196 @@ pub(crate) async fn subscribe( mod tests { use super::*; + use std::time::Duration; + + use crate::test_nats::TestNats; + + const AGENT: &str = "iris"; + + async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive) { + for i in rows { + client + .publish( + swarm_queue_client::term_subject(AGENT), + format!("row{i}").into(), + ) + .await + .expect("publish"); + } + client.flush().await.expect("flush"); + } + + /// Wait until the stream holds row `last`, so the next step sees it. + async fn stored(client: &async_nats::Client, last: u64) { + let js = async_nats::jetstream::new(client.clone()); + for _ in 0..100 { + let mut stream = js + .get_stream(agent_term::stream_name(AGENT)) + .await + .expect("the stream"); + if stream.info().await.expect("info").state.last_sequence >= last { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the stream never stored row {last}"); + } + + async fn take(rows: &mut BoxStream<'static, Row>, n: usize) -> Vec<(u64, String)> { + let mut out = Vec::with_capacity(n); + for _ in 0..n { + let row = tokio::time::timeout(Duration::from_secs(5), rows.next()) + .await + .unwrap_or_else(|_| panic!("row {} of {n} never came; got {out:?}", out.len() + 1)) + .expect("the row stream ended"); + out.push(( + row.seq.expect("a delivered row names its sequence"), + String::from_utf8_lossy(&row.payload).into_owned(), + )); + } + out + } + + fn rows(range: std::ops::RangeInclusive) -> Vec<(u64, String)> { + range.map(|i| (i, format!("row{i}"))).collect() + } + + async fn nothing_more(rows: &mut BoxStream<'static, Row>) { + if let Ok(Some(row)) = tokio::time::timeout(Duration::from_millis(500), rows.next()).await { + panic!("an extra row: seq {:?}", row.seq); + } + } + + /// A consumer the server dropped, as on a restart, ends the row stream + /// once its heartbeats stop, so the SSE response closes and the browser + /// reconnects instead of keeping a tail that never delivers. + #[tokio::test] + async fn a_dropped_consumer_ends_the_row_stream() { + use futures_util::TryStreamExt as _; + + let nats = TestNats::start(); + let client = nats.client().await; + let stream = agent_term::open_or_create(&client, AGENT) + .await + .expect("the stream"); + publish_rows(&client, 1..=3).await; + stored(&client, 3).await; + let mut replayed = replay(&client, AGENT, None).await.expect("a replay"); + assert_eq!(take(&mut replayed, 3).await, rows(1..=3)); + let consumers: Vec = stream + .consumer_names() + .try_collect() + .await + .expect("consumer names"); + let [consumer] = consumers.as_slice() else { + panic!("expected the one replay consumer, got {consumers:?}"); + }; + stream + .delete_consumer(consumer) + .await + .expect("deleting the consumer"); + let end = tokio::time::timeout(agent_term::IDLE_HEARTBEAT * 4, replayed.next()) + .await + .expect("the row stream ends once heartbeats stop"); + assert!( + end.is_none(), + "a row arrived after the consumer was deleted" + ); + } + + #[tokio::test] + async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() { + let nats = TestNats::start(); + let client = nats.client().await; + let created = agent_term::open_or_create(&client, AGENT) + .await + .expect("the first ensure creates the stream"); + publish_rows(&client, 1..=3).await; + stored(&client, 3).await; + agent_term::open_or_create(&client, AGENT) + .await + .expect("a second ensure succeeds"); + // Read back from the server, not from what the ensure returned. + let after = async_nats::jetstream::new(client.clone()) + .get_stream(agent_term::stream_name(AGENT)) + .await + .expect("the stream") + .info() + .await + .expect("its info") + .clone(); + assert_eq!(after.config, created.cached_info().config); + assert_eq!(after.state.messages, 3); + } + + /// An existing stream is opened, not created again: one made when the + /// queue announced a different `max_payload` keeps its own config and + /// rows, where a second create would be refused for the mismatch. + #[tokio::test] + async fn an_existing_terminal_stream_with_other_limits_is_opened_as_it_is() { + let nats = TestNats::start(); + let client = nats.client().await; + let js = async_nats::jetstream::new(client.clone()); + js.create_stream(async_nats::jetstream::stream::Config { + name: agent_term::stream_name(AGENT), + subjects: vec![agent_term::stream_subject(AGENT)], + max_message_size: 4096, + ..Default::default() + }) + .await + .expect("an earlier stream"); + publish_rows(&client, 1..=2).await; + stored(&client, 2).await; + let opened = agent_term::open_or_create(&client, AGENT) + .await + .expect("an existing stream opens"); + assert_eq!(opened.cached_info().config.max_message_size, 4096); + assert_eq!(opened.cached_info().state.messages, 2); + } + + /// Live rows published after the reader attached, before it read any + /// replayed row, still arrive after the whole replay. + #[tokio::test] + async fn the_replay_reaches_the_reader_before_the_live_tail() { + let nats = TestNats::start(); + let client = nats.client().await; + agent_term::open_or_create(&client, AGENT) + .await + .expect("the stream"); + publish_rows(&client, 1..=150).await; + stored(&client, 150).await; + let mut replayed = replay(&client, AGENT, None).await.expect("a replay"); + publish_rows(&client, 151..=153).await; + assert_eq!(take(&mut replayed, 103).await, rows(51..=153)); + nothing_more(&mut replayed).await; + } + + /// A reader that reconnects with the last id it was sent gets every row + /// after it exactly once, including those published while it was away. + #[tokio::test] + async fn a_reconnect_gets_every_later_row_once() { + let nats = TestNats::start(); + let client = nats.client().await; + agent_term::open_or_create(&client, AGENT) + .await + .expect("the stream"); + publish_rows(&client, 1..=150).await; + stored(&client, 150).await; + let mut first = replay(&client, AGENT, None).await.expect("a replay"); + let seen = take(&mut first, 100).await; + drop(first); + publish_rows(&client, 151..=160).await; + stored(&client, 160).await; + let last_id = seen.last().expect("rows were seen").0.to_string(); + let resume_after = last_event_id(&with_last_event_id(&last_id)); + let mut resumed = replay(&client, AGENT, resume_after) + .await + .expect("a resumed replay"); + publish_rows(&client, 161..=163).await; + assert_eq!(take(&mut resumed, 13).await, rows(151..=163)); + nothing_more(&mut resumed).await; + } + fn with_last_event_id(value: &str) -> HeaderMap { let mut headers = HeaderMap::new(); headers.insert( diff --git a/swarm-controller/src/test_nats.rs b/swarm-controller/src/test_nats.rs new file mode 100644 index 00000000..36e93d0f --- /dev/null +++ b/swarm-controller/src/test_nats.rs @@ -0,0 +1,66 @@ +//! 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); + } +} diff --git a/swarm-queue-client/src/agent_term.rs b/swarm-queue-client/src/agent_term.rs index 7699e6b6..f0bdd060 100644 --- a/swarm-queue-client/src/agent_term.rs +++ b/swarm-queue-client/src/agent_term.rs @@ -37,6 +37,11 @@ const MAX_BYTES: i64 = 64 * 1024 * 1024; /// events as its first history page. pub const REPLAY_ROWS: u64 = 100; +/// How often the server sends an idle replay consumer a heartbeat. Two missed +/// in a row end the reader's row stream, so a consumer the server dropped, +/// as on a restart, ends the SSE response and the browser reconnects. +pub const IDLE_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(5); + /// The stream holding `agent`'s terminal rows: `term-main-`. #[must_use] pub fn stream_name(agent: &str) -> String { @@ -75,11 +80,16 @@ pub fn consumer_create_grant() -> String { /// it. A fresh reader gets the newest [`REPLAY_ROWS`] below `last_sequence`, /// the stream's newest row when it was read. A sequence the stream has since /// discarded starts the consumer at the oldest row it still has. +/// +/// A `resume_after` above `last_sequence` names no row this stream ever held, +/// as after the stream was created again, and is read as a fresh reader: the +/// server accepts a start past the end and delivers nothing until the stream +/// reaches it. #[must_use] pub fn replay_start(last_sequence: u64, resume_after: Option) -> u64 { match resume_after { - Some(seen) => seen.saturating_add(1), - None => last_sequence.saturating_sub(REPLAY_ROWS - 1).max(1), + Some(seen) if seen <= last_sequence => seen + 1, + _ => last_sequence.saturating_sub(REPLAY_ROWS - 1).max(1), } } @@ -148,6 +158,7 @@ pub fn replay_consumer_config( start_sequence, }, ack_policy: async_nats::jetstream::consumer::AckPolicy::None, + idle_heartbeat: IDLE_HEARTBEAT, ..Default::default() } } @@ -243,6 +254,15 @@ mod tests { assert_eq!(replay_start(1000, Some(1000)), 1001); } + /// An id past the stream's newest row, as after the stream was created + /// again, replays like a fresh reader instead of waiting for that number. + #[test] + fn an_id_past_the_newest_row_replays_as_a_fresh_reader() { + assert_eq!(replay_start(5, Some(99)), replay_start(5, None)); + assert_eq!(replay_start(1000, Some(1001)), replay_start(1000, None)); + assert_eq!(replay_start(0, Some(7)), 1); + } + /// The consumer starts at a stored sequence, not at the next new row, so /// the replay reaches the reader before the live tail, in stream order, /// through the one subscription. @@ -263,19 +283,25 @@ mod tests { async_nats::jetstream::consumer::AckPolicy::None ); assert!(cfg.durable_name.is_none()); + // Without it a dropped consumer is never noticed. + assert_eq!(cfg.idle_heartbeat, IDLE_HEARTBEAT); + assert!(!cfg.idle_heartbeat.is_zero()); } - #[cfg(feature = "agent-term")] + /// Every limit is read from the subagent stream's own config, so the two + /// fail this test the moment they drift apart. + #[cfg(all(feature = "agent-term", feature = "subagent-term"))] #[test] fn stream_config_matches_the_subagent_streams_limits() { - let cfg = stream_config("iris", 8_388_608); - assert_eq!(cfg.subjects, ["$SWARM.term.iris"]); - assert_eq!(cfg.max_age, crate::subagent_term::MAX_AGE); - assert_eq!(cfg.max_bytes, MAX_BYTES); - assert_eq!( - cfg.discard, - async_nats::jetstream::stream::DiscardPolicy::Old - ); - assert_eq!(cfg.max_message_size, 8_388_608); + let main = stream_config("iris", 8_388_608); + let sub = crate::subagent_term::stream_config("iris", 8_388_608); + assert_eq!(main.subjects, ["$SWARM.term.iris"]); + assert_eq!(main.max_age, sub.max_age); + assert_eq!(main.max_bytes, sub.max_bytes); + assert_eq!(main.discard, sub.discard); + assert_eq!(main.max_message_size, sub.max_message_size); + assert_eq!(main.retention, sub.retention); + assert_eq!(main.storage, sub.storage); + assert_eq!(main.num_replicas, sub.num_replicas); } } diff --git a/swarm-queue-client/src/subagent_term.rs b/swarm-queue-client/src/subagent_term.rs index b340d0d4..aeef3d85 100644 --- a/swarm-queue-client/src/subagent_term.rs +++ b/swarm-queue-client/src/subagent_term.rs @@ -99,7 +99,10 @@ pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator async_nats::jetstream::stream::Config { +pub(crate) fn stream_config( + agent: &str, + max_message_size: i32, +) -> async_nats::jetstream::stream::Config { async_nats::jetstream::stream::Config { name: stream_name(agent), description: Some(format!("Terminal rows of {agent}'s subagents")),