78 lines
2.9 KiB
Rust
78 lines
2.9 KiB
Rust
//! Creating an agent's subagent stream. The names, subjects, grants and the
|
|
//! subagent list are `swarm_wire::subagent_term`.
|
|
//!
|
|
//! `swarm-controller` creates one stream per agent with the fixed config
|
|
//! [`crate::subagent_term::open_or_create`] spells.
|
|
|
|
use swarm_wire::subagent_term::{MAX_AGE, stream_name, stream_subjects};
|
|
|
|
/// How large one agent's stream may grow before the discard policy reclaims
|
|
/// space. 64 MiB is enough rows from a day of ordinary subagent chatter to
|
|
/// stay well inside it while still bounding the disk one compromised or
|
|
/// runaway agent can claim for 24h.
|
|
const MAX_BYTES: i64 = 64 * 1024 * 1024;
|
|
|
|
/// The config [`open_or_create`] creates `agent`'s stream with.
|
|
///
|
|
/// `max_message_size` is taken as a parameter rather than read from a
|
|
/// constant here: it tracks the queue's live `max_payload`
|
|
/// ([`crate::max_payload`]), the same value the publisher already sizes its
|
|
/// rows against, so the stream never rejects a row the queue itself
|
|
/// accepted.
|
|
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")),
|
|
subjects: vec![stream_subjects(agent)],
|
|
max_age: MAX_AGE,
|
|
max_bytes: MAX_BYTES,
|
|
discard: async_nats::jetstream::stream::DiscardPolicy::Old,
|
|
max_message_size,
|
|
..Default::default()
|
|
}
|
|
}
|
|
|
|
/// Open `agent`'s subagent stream, creating it if it does not exist yet. An
|
|
/// existing stream is opened as it is.
|
|
pub async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
agent: &str,
|
|
) -> Result<async_nats::jetstream::stream::Stream, crate::Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let name = stream_name(agent);
|
|
match js.get_stream(&name).await {
|
|
Ok(stream) => Ok(stream),
|
|
Err(e) => {
|
|
tracing::info!(stream = %name, reason = %e, "subagent terminal stream not available, creating it");
|
|
let max_message_size = i32::try_from(crate::max_payload(client)).unwrap_or(i32::MAX);
|
|
js.create_stream(stream_config(agent, max_message_size))
|
|
.await
|
|
.map_err(|source| crate::Error::CreateAgentStream {
|
|
stream: name,
|
|
source,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// The config the controller creates caps disk per agent and never
|
|
/// fails a publish because of that cap — `discard: Old` reclaims the
|
|
/// oldest rows instead of rejecting the newest.
|
|
#[test]
|
|
fn stream_config_caps_bytes_and_discards_the_oldest() {
|
|
let cfg = stream_config("iris", 8_388_608);
|
|
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);
|
|
}
|
|
}
|