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

swarm-wire: take agent_limits's wire types

Browse source
This commit is contained in:
atlas 2026-10-18 12:29:18 +02:00 • committed by mara
commit 9cce9c470f
11 changed files with 122 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

106
swarm-queue-client/src/agent_limits.rs
View file

@ -1,58 +1,6 @@
//! The per-agent resource-cap KV bucket: one key per agent, holding the CPU
//! and memory caps the swarm sets for it.
//!
//! **swarm-controller is the only writer**, and creates the bucket on its
//! first write. Every field is optional: an absent field leaves the agent on
//! its hive's own answer, so an agent with no key runs exactly as it would
//! with no swarm.
//!
//! **The hive running the agent reads the key**, not the agent: the caps go
//! into the container's systemd drop-in, which the host writes. A hive reads
//! with one `DIRECT.GET` per agent
//! ([`crate::agent_limits::direct_get_subject`]), and its grant covers every
//! key of this bucket, read-only. Scoping it to the hive's own agents would
//! need a roster in the auth responder.
//!
//! Keyed by the agent alone, like [`crate::agent_inference`]: an agent can
//! move between hives, and its caps go with it.
/// The KV bucket caps are published into, keyed by agent name.
pub const BUCKET: &str = "agent-limits";
/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`.
pub const STREAM: &str = "KV_agent-limits";
/// The subject `agent`'s entry is stored under.
#[must_use]
pub fn key_subject(agent: &str) -> String {
format!("$KV.{BUCKET}.{agent}")
}
/// The request subject that reads `agent`'s current entry.
#[must_use]
pub fn direct_get_subject(agent: &str) -> String {
format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent))
}
/// One agent's caps: the value under `BUCKET`/`<agent>`, as systemd values
/// carried verbatim. `None` is "not capped by the swarm", never "uncapped".
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentCaps {
/// systemd `CPUQuota=`, e.g. `"400%"`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cpu_quota: Option<String>,
/// systemd `MemoryMax=`, e.g. `"8G"`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub memory_max: Option<String>,
}
impl AgentCaps {
/// True when no field is set, which a hive reads the same as no key.
#[must_use]
pub fn is_empty(&self) -> bool {
self.cpu_quota.is_none() && self.memory_max.is_none()
}
}
//! The connection side of the per-agent resource-cap KV bucket: opening it for
//! writing or reading, and the hive's direct read. The bucket's names, subjects
//! and stored type are `swarm_wire::agent_limits`.
/// Open the caps bucket for writing, creating it if nothing has yet.
///
@ -64,7 +12,7 @@ pub async fn open_or_create(
let js = async_nats::jetstream::new(client.clone());
crate::kv::open_or_create(
&js,
BUCKET,
swarm_wire::agent_limits::BUCKET,
"agent-limits",
"Resource caps the swarm controller sets for each agent".to_owned(),
)
@ -85,11 +33,11 @@ pub async fn open(
client: &async_nats::Client,
) -> Result<Option<async_nats::jetstream::kv::Store>, crate::Error> {
let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await {
match js.get_key_value(swarm_wire::agent_limits::BUCKET).await {
Ok(store) => Ok(Some(store)),
Err(e) if crate::wanted::is_bucket_not_found(&e) => Ok(None),
Err(source) => Err(crate::Error::OpenBucket {
bucket: BUCKET.to_owned(),
bucket: swarm_wire::agent_limits::BUCKET.to_owned(),
source,
}),
}
@ -100,7 +48,7 @@ pub async fn open(
#[derive(Debug)]
pub enum Read {
/// The bucket answered; an absent or deleted key has every field unset.
Current(AgentCaps),
Current(swarm_wire::agent_limits::AgentCaps),
/// Nothing answers for the bucket's stream: no cap has been written yet,
/// or `JetStream` is not serving it right now. Not an unset value, since
/// the second case says nothing about the agent's caps.
@ -119,6 +67,7 @@ pub async fn read(
agent: &str,
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
use async_nats::jetstream::stream::DirectGetErrorKind;
use swarm_wire::agent_limits::{AgentCaps, STREAM, key_subject};
let js = async_nats::jetstream::new(client.clone());
let stream = js.get_stream_no_info(STREAM).await?;
@ -149,42 +98,3 @@ fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool {
if status == async_nats::StatusCode::NO_RESPONDERS
)
}
#[cfg(test)]
mod tests {
use super::*;
/// The names the hive's grant in `swarm-nats-auth` is built from.
#[test]
fn names_match_the_queue_grant() {
assert_eq!(STREAM, format!("KV_{BUCKET}"));
assert_eq!(key_subject("iris"), "$KV.agent-limits.iris");
assert_eq!(
direct_get_subject("iris"),
"$JS.API.DIRECT.GET.KV_agent-limits.$KV.agent-limits.iris"
);
}
/// An unset field is left out on the wire, and `{}` is every field unset.
#[test]
fn unset_fields_are_absent() {
let value = AgentCaps {
memory_max: Some("8G".to_owned()),
..AgentCaps::default()
};
assert_eq!(
serde_json::to_string(&value).unwrap(),
r#"{"memory_max":"8G"}"#
);
let empty: AgentCaps = serde_json::from_str("{}").unwrap();
assert!(empty.is_empty());
}
/// Fields a newer controller adds must not cost an older hive its caps.
#[test]
fn decoding_tolerates_unknown_fields() {
let decoded: AgentCaps =
serde_json::from_str(r#"{"cpu_quota":"400%","io_weight":"10"}"#).unwrap();
assert_eq!(decoded.cpu_quota.as_deref(), Some("400%"));
}
}