//! The per-agent status KV bucket: one key per `(hive, agent)` pair. //! //! Deliberately **not** a field inside [`crate::status::BUCKET`]'s hive //! document, even though a hive already publishes its own status there. //! Three reasons, none of which show up until the second write: //! //! - **Cardinality and write rate.** Hive status is one value per hive. //! Agent status is N values per hive, changing independently and far //! more often — folded into the hive document, every agent's change //! rewrites the whole thing, and two agents changing at once race on //! the same key. //! - **A KV key is the unit of update and of watch.** A consumer that //! cares about one agent wants a per-key read with its own revision, //! not a hive-wide document to diff. //! - **Staleness means different things.** A hive can be healthy while //! one agent is wedged, and the reverse; one timestamp cannot answer //! both questions honestly. //! //! The transport *is* shared — this rides the same [`async_nats::Client`] //! and the same auth callout as [`crate::status`] and [`crate::wanted`], //! just a different bucket. #[cfg(feature = "kv")] use crate::Error; /// The KV bucket per-agent status snapshots are published into, keyed by /// [`key`]. /// /// A constant and not an option, for the reason [`crate::status::BUCKET`] /// gives: writer (a hive) and reader (the controller) must name the same /// bucket, and an option is a way for the two to disagree about which one /// that is. pub const BUCKET: &str = "agent-status"; /// The bucket key for one agent on one hive. /// /// `/`-joined rather than NATS's usual `.`-joined subject style: this is a /// KV key, not a subject, and neither `hive` nor `agent` names can contain /// `/` (`hive_types::Ident`'s charset is `[a-z0-9-]`), so the join is /// unambiguous to split back apart if a consumer ever needs to. /// /// Plain code span above, deliberately not an intra-doc link — this crate /// doesn't depend on `hive_types`, and a bracketed reference to its `Ident` /// type fails `cargo doc` with "no item named `hive_types` in scope" under /// this workspace's `docs-rustdoc` CI job, which treats broken doc links as /// errors. #[must_use] pub fn key(hive: &str, agent: &str) -> String { format!("{hive}/{agent}") } /// The inverse of [`key`]: split a bucket key back into `(hive, agent)`. /// /// `None` for a key with zero or more than one `/` — a hive publishing /// under [`key`] never produces one, so a malformed key means something /// else wrote this bucket. Splitting on the *first* `/` would be equally /// valid today (neither name can contain one), but this rejects rather /// than guesses, so a future name-charset change can't silently start /// misreading old keys. #[must_use] pub fn split_key(key: &str) -> Option<(&str, &str)> { let mut parts = key.splitn(3, '/'); let hive = parts.next()?; let agent = parts.next()?; if parts.next().is_some() { return None; } Some((hive, agent)) } /// One agent's status snapshot, as published under [`key`]. /// /// Mirrors the tuple `container_view::read_agent_status_live` returns on /// the publishing side — this is that value, not a recomputation of it. /// Nothing here stamps a time: same rule as [`crate::status`], freshness is /// derived by the reader from when the value landed in the bucket, so a /// hive with a wrong clock only skews its own payload. #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct AgentStatus { /// The agent's free-text status (`set_status`), or `None` if it has /// never set one, or if the container isn't running. pub status_text: Option, /// Unix timestamp the status was last set. `None` alongside /// `status_text: None` — the two are never independently absent. pub status_set_at: Option, /// Whether the container is currently running. `false` forces the two /// fields above to `None` even if an on-disk status file is stale from /// before the container stopped — the same "stopped containers have /// stale state" rule `read_agent_status_live` already enforces. pub running: bool, } /// Open the agent-status bucket, creating it if nothing has yet. /// /// `history: 1`, same rationale as [`crate::status::open_or_create`]: a /// consumer wants the last thing each agent said, not a log of everything /// it has ever said. #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, ) -> Result { let js = async_nats::jetstream::new(client.clone()); match js.get_key_value(BUCKET).await { Ok(store) => Ok(store), Err(e) => { tracing::info!( bucket = BUCKET, reason = %e, "agent-status bucket not available, creating it" ); js.create_key_value(async_nats::jetstream::kv::Config { bucket: BUCKET.to_owned(), description: "Last status snapshot offered by each agent".to_owned(), history: 1, ..Default::default() }) .await .map_err(|source| Error::CreateBucket { bucket: BUCKET, source, }) } } } #[cfg(test)] mod tests { use super::{AgentStatus, key, split_key}; #[test] fn key_joins_hive_and_agent_with_a_slash() { assert_eq!(key("prod", "iris"), "prod/iris"); } #[test] fn split_key_is_the_inverse_of_key() { assert_eq!(split_key(&key("prod", "iris")), Some(("prod", "iris"))); } #[test] fn split_key_rejects_a_key_with_no_slash() { assert_eq!(split_key("iris"), None); } #[test] fn split_key_rejects_a_key_with_two_slashes() { assert_eq!(split_key("prod/iris/extra"), None); } #[test] fn agent_status_round_trips() { let status = AgentStatus { status_text: Some("reviewing a pull request".to_owned()), status_set_at: Some(1_725_000_000), running: true, }; let json = serde_json::to_string(&status).expect("serialises"); let decoded: AgentStatus = serde_json::from_str(&json).expect("decodes"); assert_eq!(decoded, status); } #[test] fn a_stopped_agent_serialises_with_no_status() { let status = AgentStatus::default(); assert_eq!( serde_json::to_string(&status).expect("serialises"), r#"{"status_text":null,"status_set_at":null,"running":false}"# ); } }