From d81fb2236fea97a70ffc09ba051d896b3a2c61e4 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 4 Oct 2026 19:30:20 +0200 Subject: [PATCH] swarm: grant the controller STREAM.MSG.GET; list subagents concurrently GET /api/agents/{name}/subagents reads each subagent's newest row with get_last_raw_message_by_subject, which publishes $JS.API.STREAM.MSG.GET.term-sub-. The controller's NATS grant had no STREAM.MSG.GET, so every lookup was refused and waited the 10s request timeout, one subagent after another; six subagents passed nginx's 60s. Grant `$JS.API.STREAM.MSG.GET.*`. A wildcard matches a whole subject token only, so a `term-sub-*` spelling would name one stream literally; `*` is the narrowest grant that covers the per-agent streams, the same width as the controller's STREAM.INFO.* grant. The lookups now run concurrently (16 at a time) with a 2s timeout each. A lookup that fails or times out lists its subagent with a null last_row_unix, and one warn per request names how many failed and the first error. --- docs/swarm/ui.md | 5 +- swarm-controller/src/subagent_term.rs | 70 +++++++++++++++++---------- swarm-nats-auth/src/policy.rs | 27 +++++++++++ 3 files changed, 76 insertions(+), 26 deletions(-) diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 9e6fb0db..1a910e9e 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -94,7 +94,10 @@ into a stream named `term-sub-` that swarm-controller creates for every declared agent. The stream keeps rows for 24 hours. swarm-controller serves the list as `GET /api/agents//subagents`, the subagents named by the subjects in that stream, each with `last_row_unix`, the stream's publish time of its newest row, -and relays one subagent's rows as SSE on +or `null` when that row isn't read within two seconds. The lookups run +concurrently, sixteen at a time, so the list waits at most two seconds for +each sixteen subagents. +swarm-controller also relays one subagent's rows as SSE on `GET /api/agents//subagents//term/stream`. An agent with no store identity, or no queue address, publishes no subagent terminals. diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 8b19caf2..3372d636 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -41,6 +41,15 @@ use crate::wanted::WantedWriter; /// declared agent's subagents can go unlisted. const ENSURE_INTERVAL: Duration = Duration::from_mins(1); +/// How long [`list_subagents`] waits for one subagent's newest row before +/// listing it without a time. The lookups run concurrently, so a stream that +/// does not answer costs the list this long per [`LOOKUP_CONCURRENCY`] +/// subagents. +const LOOKUP_TIMEOUT: Duration = Duration::from_secs(2); + +/// How many newest-row lookups [`list_subagents`] has in flight at once. +const LOOKUP_CONCURRENCY: usize = 16; + /// Create the subagent stream of every live agent that has none, now and every /// [`ENSURE_INTERVAL`] after. An existing stream is left as it is. /// @@ -137,39 +146,50 @@ pub(crate) async fn list_subagents( .await .map_err(|e| read_failed(&agent, &e))?; let names = subagent_term::subagent_names(&agent, subjects.iter().map(|(s, _)| s.as_str())); - let mut entries = Vec::with_capacity(names.len()); - for name in names { - let last_row_unix = last_row_unix(&stream, &agent, &name).await; - entries.push(SubagentEntry { - name, - last_row_unix, - }); + // Owned names and `async move`, not a closure over `&names`: a borrowed + // item type trips a "`FnOnce` is not general enough" inference error where + // `routes!` checks this fn as a handler. + let (stream_ref, agent_ref) = (&stream, &agent); + let lookups: Vec> = futures_util::stream::iter(names.clone()) + .map(|name| async move { last_row_unix(stream_ref, agent_ref, &name).await }) + .buffered(LOOKUP_CONCURRENCY) + .collect() + .await; + let failed = lookups.iter().filter(|l| l.is_err()).count(); + if let Some(Err(error)) = lookups.iter().find(|l| l.is_err()) { + tracing::warn!( + %agent, + failed, + subagents = names.len(), + %error, + "listing subagents: reading the newest row failed (first error shown)" + ); } + let entries = names + .into_iter() + .zip(lookups) + .map(|(name, lookup)| SubagentEntry { + name, + last_row_unix: lookup.ok(), + }) + .collect(); Ok(Json(entries)) } -/// The publish time of a subagent's newest row. A row that cannot be read, -/// aged out between the listing and this lookup say, is a missing time and -/// not a failed list. +/// The publish time of a subagent's newest row, or why it could not be read. +/// A row that cannot be read, aged out between the listing and this lookup +/// say, is a missing time and not a failed list. async fn last_row_unix( stream: &async_nats::jetstream::stream::Stream, agent: &str, subagent: &str, -) -> Option { - match stream - .get_last_raw_message_by_subject(&subagent_term::subject(agent, subagent)) - .await - { - Ok(row) => Some(row.time.unix_timestamp()), - Err(e) => { - tracing::debug!( - %agent, - %subagent, - error = %swarm_queue_client::chain(&e), - "listing subagents: reading the newest row failed" - ); - None - } +) -> Result { + let subject = subagent_term::subject(agent, subagent); + let lookup = stream.get_last_raw_message_by_subject(&subject); + match tokio::time::timeout(LOOKUP_TIMEOUT, lookup).await { + Ok(Ok(row)) => Ok(row.time.unix_timestamp()), + Ok(Err(e)) => Err(format!("{subagent}: {}", swarm_queue_client::chain(&e))), + Err(_) => Err(format!("{subagent}: no answer in {LOOKUP_TIMEOUT:?}")), } } diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 85be9c9a..a465960f 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -540,6 +540,14 @@ impl Policy { "$JS.API.STREAM.INFO.*".to_owned(), "$JS.API.STREAM.CREATE.*".to_owned(), "$JS.API.DIRECT.GET.*.>".to_owned(), + // The newest row of one subagent in `term-sub-`, which the + // subagent list reads by subject. Those streams do not allow + // direct get, so `DIRECT.GET` above does not reach them. `*` is a + // whole token: `term-sub-*` would name one stream literally, and + // the agent is not known here. Also reaches every other stream's + // `MSG.GET`, the same width as `STREAM.INFO.*` above. A refused + // publish reaches the client as a timeout. + "$JS.API.STREAM.MSG.GET.*".to_owned(), "$KV.*.>".to_owned(), ]); // The per-agent status bucket, read-side only. Same five subjects as @@ -1171,6 +1179,25 @@ mod tests { assert!(p.publish.contains(&"$JS.API.STREAM.CREATE.*".to_owned())); } + #[test] + fn the_controller_may_read_a_subagents_newest_row() { + let p = policy() + .permissions("swarm-controller") + .expect("a reader is admitted"); + assert!(p.publish.contains(&"$JS.API.STREAM.MSG.GET.*".to_owned())); + // `*` only matches a whole token, so a prefix spelling grants one + // stream literally named that and reads nothing. + assert!(!p.publish.iter().any(|s| s.contains("term-sub-*"))); + // Control: a hive's grant holds no `MSG.GET`, so the subject above is + // the controller's alone. + assert!( + !policy() + .hive_subjects("alpha") + .iter() + .any(|s| s.starts_with("$JS.API.STREAM.MSG.GET")) + ); + } + #[test] fn the_notices_stream_grant_is_identical_across_hives() { // Unlike `$KV..` or an `extra_hive_subjects`