Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-wire/src/agent_status.rs
damocles 748ef2da08 hives report an agent being updated: its status shows updating from the update's stop until the container is back up
A hive marks an agent updating when StopForUpdate stops its running container and clears it when the tail Reconcile or Start settles it, success or failure. The flag rides in the agent status snapshot (absent when false, so older readers and documents are unchanged), the status publisher sends a change at once instead of at the next sweep, and the swarm UI card shows an updating badge.
2026-11-11 16:51:42 +01:00

184 lines
7.7 KiB
Rust

//! The per-agent status KV bucket: one key per `(hive, agent)` pair.
//!
//! Deliberately **not** a field inside `swarm_queue_client::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 `swarm_queue_client::status` and
//! `swarm_queue_client::wanted`,
//! just a different bucket.
/// The KV bucket per-agent status snapshots are published into, keyed by
/// [`key`].
///
/// A constant and not an option, for the reason `swarm_queue_client::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, and the separator is load-bearing rather than cosmetic: a KV
/// entry is published to `$KV.<bucket>.<key>`, and NATS wildcards match
/// whole `.`-delimited tokens. Two tokens make a hive's write grant
/// expressible as `$KV.agent-status.<hive>.*` — every agent of one hive and
/// nothing else. Joined by any character that is not `.`, the key is a
/// single token, and the only grants available are one exact subject per
/// agent or a bucket-wide wildcard that lets any hive overwrite any other
/// hive's agents.
///
/// Neither name can be empty or contain a `.` (`hive_types::Ident` is
/// `[a-z0-9-]`, non-empty), so the join stays unambiguous to split back
/// apart — and a leading or trailing `.` would be rejected outright by the
/// client's own KV key validation.
#[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 `swarm_queue_client::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<String>,
/// Unix timestamp the status was last set. `None` alongside
/// `status_text: None` — the two are never independently absent.
pub status_set_at: Option<i64>,
/// 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,
/// Whether the hive is updating the agent: from the moment an update stops
/// its container until the container is back up (or the update gave up and
/// the hive settled the agent's power state). The agent reads as stopped
/// across that window, and this says why. Left out of the document when
/// `false`, so a snapshot of an agent not being updated is byte for byte
/// what a hive without the field publishes, and a document without it
/// reads as `false`.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub updating: bool,
}
#[cfg(test)]
mod tests {
use super::{AgentStatus, BUCKET, key, split_key};
#[test]
fn key_joins_hive_and_agent_with_a_dot() {
assert_eq!(key("prod", "iris"), "prod.iris");
}
#[test]
fn the_published_subject_carries_the_hive_as_its_own_token() {
// This is what the hive's write grant `$KV.agent-status.<hive>.*`
// matches on, so the separator is pinned here rather than left as a
// property of `key`'s formatting string.
assert_eq!(
format!("$KV.{BUCKET}.{}", key("prod", "iris")),
"$KV.agent-status.prod.iris"
);
// Control: a single-token join lands in the same bucket and would
// pass any assertion that only checked the prefix — it is the hive
// token that the grant needs, and only the `.` produces one.
assert!(!format!("$KV.{BUCKET}.prod/iris").starts_with("$KV.agent-status.prod."));
}
#[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_dot() {
assert_eq!(split_key("iris"), None);
}
#[test]
fn split_key_rejects_a_key_with_two_dots() {
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,
updating: false,
};
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 an_updating_agent_says_so_and_a_document_without_the_field_does_not() {
let updating = AgentStatus {
running: false,
updating: true,
..AgentStatus::default()
};
let json = serde_json::to_string(&updating).expect("serialises");
assert_eq!(
json,
r#"{"status_text":null,"status_set_at":null,"running":false,"updating":true}"#
);
let decoded: AgentStatus = serde_json::from_str(&json).expect("decodes");
assert!(decoded.updating);
// A hive that predates the field publishes none.
let old: AgentStatus =
serde_json::from_str(r#"{"status_text":null,"status_set_at":null,"running":true}"#)
.expect("decodes");
assert!(!old.updating);
}
#[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}"#
);
}
}