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-<agent>. 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

@@ -94,7 +94,10 @@ into a stream named `term-sub-<agent>` that swarm-controller creates for every
 declared agent. The stream keeps rows for 24 hours. swarm-controller serves the list as
 `GET /api/agents/<name>/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/<name>/subagents/<subagent>/term/stream`. An agent with no
 store identity, or no queue address, publishes no subagent terminals.
 

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<Result<i64, String>> = 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<i64> {
-    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<i64, String> {
+    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:?}")),
     }
 }
 

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-<agent>`, 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.<bucket>.<hive>` or an `extra_hive_subjects`