vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: drop the relay subscription to the refused hive-scoped agent subject

Browse source
This commit is contained in:
atlas 2026-10-04 21:43:33 +02:00 • committed by mara
commit 8a7733b3f5
3 changed files with 37 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

13
swarm-controller/src/agent_state_stream.rs
View file

@ -2,8 +2,7 @@
//! turn-state header, sourced straight from the swarm queue. //! turn-state header, sourced straight from the swarm queue.
//! //!
//! Sibling of [`crate::term_stream`] in every structural respect — the //! Sibling of [`crate::term_stream`] in every structural respect — the
//! reasoning there for resolving the hive at request time rather than //! reasoning there for keeping the hive out of the path, and for a live tail with no replay, applies
//! naming it in the path, and for a live tail with no replay, applies
//! unchanged here and is not repeated. What differs is only what the two //! 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 //! subjects carry: a terminal row is an event that happened, a header is
//! the agent's current state. //! the agent's current state.
@ -33,8 +32,8 @@ use crate::AppState;
/// Subject family carrying agent turn-state headers — must match /// Subject family carrying agent turn-state headers — must match
/// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two /// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two
/// ends never see the constant together. `crate::term_stream::agent_subjects` /// ends never see the constant together. `crate::term_stream::agent_subject`
/// spells the full subjects; as with the terminal relay's own copy, there is no /// spells the full subject; as with the terminal relay's own copy, there is no
/// way to check the two sides agree short of this comment and the module /// way to check the two sides agree short of this comment and the module
/// docs on both ends staying honest about it. /// docs on both ends staying honest about it.
const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; const SUBJECT_PREFIX: &str = "$SWARM.agent-state";
@ -85,7 +84,7 @@ pub(crate) async fn stream_agent_state(
// One-agent "roster": reuses `AgentStatusReader::view`'s already-tested // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested
// row logic rather than a second, narrower lookup that could disagree // row logic rather than a second, narrower lookup that could disagree
// with it about what "never reported" means. // with it about what "never reported" means.
let hive = agent_status agent_status
.view(std::slice::from_ref(&agent), SystemTime::now()) .view(std::slice::from_ref(&agent), SystemTime::now())
.await .await
.map_err(|e| { .map_err(|e| {
@ -117,9 +116,9 @@ pub(crate) async fn stream_agent_state(
) )
})?; })?;
let subscriber = crate::term_stream::subscribe_all( let subscriber = crate::term_stream::subscribe(
&client, &client,
&crate::term_stream::agent_subjects(SUBJECT_PREFIX, &hive, &agent), crate::term_stream::agent_subject(SUBJECT_PREFIX, &agent),
"state", "state",
) )
.await?; .await?;

9
swarm-controller/src/subagent_term.rs
View file

@ -11,9 +11,8 @@
//! Each subagent carries the stream's own publish time of its newest row, so //! Each subagent carries the stream's own publish time of its newest row, so
//! a reader can tell a subagent that is running from one that has gone quiet. //! a reader can tell a subagent that is running from one that has gone quiet.
//! //!
//! **No hive lookup.** Unlike [`crate::term_stream`], there is no //! **No agent-status lookup.** Unlike [`crate::term_stream`], an agent that
//! hive-scoped subject to follow: only the agent's own credential is granted //! has never reported gets an empty stream, not a 404.
//! these subjects.
//! //!
//! **Live tail only, output only.** The relay is a core subscription, the //! **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 //! same as the agent's own terminal route, so a reader sees rows published
@ -221,9 +220,9 @@ pub(crate) async fn stream_subagent_term(
let agent = ident(&agent).map_err(bad_request)?; let agent = ident(&agent).map_err(bad_request)?;
let subagent = ident(&subagent).map_err(bad_request)?; let subagent = ident(&subagent).map_err(bad_request)?;
let client = connected_client(&state).map_err(|e| unavailable(&e))?; let client = connected_client(&state).map_err(|e| unavailable(&e))?;
let subscriber = crate::term_stream::subscribe_all( let subscriber = crate::term_stream::subscribe(
&client, &client,
&[subagent_term::subject(&agent, &subagent)], subagent_term::subject(&agent, &subagent),
"subagent term", "subagent term",
) )
.await?; .await?;

78
swarm-controller/src/term_stream.rs
View file

@ -3,15 +3,13 @@
//! //!
//! **No `hive` in the path, on purpose.** An agent is not pinned to a //! **No `hive` in the path, on purpose.** An agent is not pinned to a
//! hive forever — it can move — so a URL naming one would go stale the //! hive forever — it can move — so a URL naming one would go stale the
//! moment it did. The hive is resolved at request time from //! moment it did. `crate::agent_status`'s bucket is read only to answer
//! `crate::agent_status`'s bucket instead, the same "an agent's hive //! 404 for an agent that has never reported.
//! comes from its own last report" rule that module already documents.
//! //!
//! **Live tail only, on purpose.** `hive-agent` publishes each //! **Live tail only, on purpose.** `hive-agent` publishes each
//! already-classified `TermMsg` row to the core subject //! already-classified `TermMsg` row to the core subject
//! `$SWARM.term.{agent}` (`$SWARM.term.{hive}.{agent}` from an agent on its //! `$SWARM.term.{agent}` (see `hive-agent::swarm_term`'s module doc) —
//! hive's shared credential; see `hive-agent::swarm_term`'s module //! a core subject, not `JetStream`, so a subscriber who was not
//! doc) — a core subject, not `JetStream`, so a subscriber who was not
//! listening missed the row, same as on the agent's own local SSE //! listening missed the row, same as on the agent's own local SSE
//! stream. This handler relays exactly that: no replay, no last-N //! stream. This handler relays exactly that: no replay, no last-N
//! history. The publish side's own retention half was deliberately //! history. The publish side's own retention half was deliberately
@ -35,8 +33,8 @@ use crate::AppState;
/// Subject family carrying agent terminal rows — must match /// Subject family carrying agent terminal rows — must match
/// `hive-agent::swarm_term::SUBJECT_PREFIX` exactly, since the two ends /// `hive-agent::swarm_term::SUBJECT_PREFIX` exactly, since the two ends
/// never see the constant together. [`agent_subjects`] spells the full /// never see the constant together. [`agent_subject`] spells the full
/// subjects; there is no way to check the two sides agree short of /// subject; there is no way to check the two sides agree short of
/// this comment and the module docs on both ends staying honest about it. /// this comment and the module docs on both ends staying honest about it.
const SUBJECT_PREFIX: &str = "$SWARM.term"; const SUBJECT_PREFIX: &str = "$SWARM.term";
@ -84,7 +82,7 @@ pub(crate) async fn stream_agent_term(
// One-agent "roster": reuses `AgentStatusReader::view`'s already-tested // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested
// row logic rather than a second, narrower lookup that could disagree // row logic rather than a second, narrower lookup that could disagree
// with it about what "never reported" means. // with it about what "never reported" means.
let hive = agent_status agent_status
.view(std::slice::from_ref(&agent), SystemTime::now()) .view(std::slice::from_ref(&agent), SystemTime::now())
.await .await
.map_err(|e| { .map_err(|e| {
@ -117,63 +115,41 @@ pub(crate) async fn stream_agent_term(
) )
})?; })?;
let subscriber = subscribe_all( let subscriber = subscribe(&client, agent_subject(SUBJECT_PREFIX, &agent), "term").await?;
&client,
&agent_subjects(SUBJECT_PREFIX, &hive, &agent),
"term",
)
.await?;
let stream = let stream =
subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload)))); subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
Ok(Sse::new(stream).keep_alive(KeepAlive::default())) Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
} }
/// The subjects one agent publishes on under `prefix`: its own /// The subject one agent publishes on under `prefix`: `<prefix>.<agent>`.
/// `<prefix>.<agent>`, and `<prefix>.<hive>.<agent>`, which an agent still on pub(crate) fn agent_subject(prefix: &str, agent: &str) -> String {
/// its hive's shared queue credential publishes to. format!("{prefix}.{agent}")
pub(crate) fn agent_subjects(prefix: &str, hive: &str, agent: &str) -> [String; 2] {
[
format!("{prefix}.{agent}"),
format!("{prefix}.{hive}.{agent}"),
]
} }
/// Subscribe to every one of `subjects`, merged into one stream. `what` names /// Subscribe to `subject`. `what` names the relay in logs and errors.
/// the relay in logs and errors. pub(crate) async fn subscribe(
pub(crate) async fn subscribe_all(
client: &async_nats::Client, client: &async_nats::Client,
subjects: &[String], subject: String,
what: &str, what: &str,
) -> Result<futures_util::stream::SelectAll<async_nats::Subscriber>, problem_details::ProblemDetails> ) -> Result<async_nats::Subscriber, problem_details::ProblemDetails> {
{ let subscriber = client.subscribe(subject.clone()).await.map_err(|e| {
let mut subscribers = Vec::with_capacity(subjects.len()); tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed");
for subject in subjects { crate::error_problem(
let subscriber = client.subscribe(subject.clone()).await.map_err(|e| { axum::http::StatusCode::INTERNAL_SERVER_ERROR,
tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed"); &format!("subscribing to {subject} failed: {e}"),
crate::error_problem( )
axum::http::StatusCode::INTERNAL_SERVER_ERROR, })?;
&format!("subscribing to {subject} failed: {e}"), tracing::info!(%subject, "{what} stream: client attached");
) Ok(subscriber)
})?;
subscribers.push(subscriber);
}
tracing::info!(?subjects, "{what} stream: client attached");
Ok(futures_util::stream::select_all(subscribers))
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
/// Both shapes, so an agent on either credential is streamed. /// The subject the agent's own queue credential is granted.
#[test] #[test]
fn a_relay_subscribes_to_the_agents_own_subject_and_its_hive_scoped_one() { fn a_relay_subscribes_to_the_agents_own_subject() {
assert_eq!( assert_eq!(agent_subject(SUBJECT_PREFIX, "atlas"), "$SWARM.term.atlas");
agent_subjects(SUBJECT_PREFIX, "h1", "atlas"),
[
"$SWARM.term.atlas".to_owned(),
"$SWARM.term.h1.atlas".to_owned(),
]
);
} }
} }