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

keep the operator list under one key and read it with one direct get

Browse source 
The agent read the operator-profiles bucket with a keys() listing, an
ordered consumer whose grant (STREAM.INFO, CONSUMER.CREATE) and flow-control
reply had never run against a permissioned queue. The controller now writes
the whole list as one JSON array under one key, only when it changed, and
the agent reads it with one DIRECT.GET the way agent_inference::read_own
reads its key. The agent grant shrinks to that one subject.
This commit is contained in:
atlas 2026-10-17 12:19:43 +02:00 • committed by mara
commit 49015eb553
12 changed files with 254 additions and 201 deletions
Download patch file Download diff file Expand all files Collapse all files

189
swarm-controller/src/forge/operator_profiles.rs
View file

@ -1,21 +1,21 @@
//! Each operator's forge-profile pronouns, published to the `operator-profiles`
//! KV bucket (`swarm_queue_client::operator_profiles`) at start and every
//! [`RECONCILE_INTERVAL`] after. Every agent reads that bucket into its system
//! Each operator's forge-profile pronouns, published as one list under the
//! `operator-profiles` KV bucket's one key
//! (`swarm_queue_client::operator_profiles`) at start and every
//! [`RECONCILE_INTERVAL`] after. Every agent reads that key into its system
//! prompt.
//!
//! An operator is a member of authelia's operator group, read the way
//! `crate::matrix_account::operator_admin` reads it. A pass writes a key only
//! when its value changed and removes the key of a name no longer in the
//! group. An operator whose forge profile could not be read keeps the value
//! the bucket holds; one with no forge account yet is listed with no pronouns.
//! `crate::matrix_account::operator_admin` reads it. A pass writes the key
//! only when the list changed, so a name no longer in the group drops out of
//! it. An operator whose forge profile could not be read keeps the entry the
//! stored list holds (name only when it holds none); one with no forge account
//! yet is listed with no pronouns.
use std::collections::BTreeSet;
use std::sync::Arc;
use anyhow::{Context, Result};
use async_nats::jetstream::kv::Store;
use futures_util::TryStreamExt as _;
use swarm_queue_client::operator_profiles::{self, OperatorProfile};
use swarm_queue_client::operator_profiles::{self, KEY, OperatorProfile};
use super::Client;
use super::agent_token::is_not_found;
@ -47,91 +47,68 @@ impl Profiles for Client {
/// What one pass did.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct Outcome {
written: usize,
removed: usize,
/// Operators whose profile could not be read, left as they were.
/// The list differed from the stored one and was written.
written: bool,
/// Operators whose profile could not be read, left as stored.
kept: usize,
}
/// Whether `name` can be a key: one subject token of the characters NATS
/// accepts in one.
fn is_key(name: &str) -> bool {
!name.is_empty()
&& name
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
}
/// Bring `store` to one key per name in `operators`, each holding that
/// operator's pronouns as `forge` reports them.
/// Bring [`KEY`] in `store` to the list of `operators`, each with its
/// pronouns as `forge` reports them, sorted by name.
///
/// # Errors
/// When the bucket cannot be listed, read or written. A forge read that fails
/// is logged and leaves that operator's key alone.
/// When the key cannot be read or written. A forge read that fails is logged
/// and keeps that operator's stored entry. A stored value this build cannot
/// decode reads as an empty list, so the pass replaces it.
pub(crate) async fn sync(
operators: &[String],
forge: &impl Profiles,
store: &Store,
) -> Result<Outcome> {
let mut stale: BTreeSet<String> = store
.keys()
let stored: Vec<OperatorProfile> = store
.get(KEY)
.await
.context("listing the operator-profiles bucket")?
.try_collect()
.await
.context("listing the operator-profiles bucket")?;
.context("reading the operator list")?
.and_then(|value| serde_json::from_slice(&value).ok())
.unwrap_or_default();
let mut outcome = Outcome::default();
let mut list = Vec::with_capacity(operators.len());
for name in operators {
stale.remove(name);
if !is_key(name) {
tracing::warn!(
operator = %name,
"operator profiles: the username is no legal key; skipped"
);
continue;
}
let pronouns = match forge.pronouns(name).await {
Ok(pronouns) => pronouns,
Err(e) => {
tracing::warn!(
operator = %name,
error = %format!("{e:#}"),
"operator profiles: forge profile unreadable; keeping the stored value"
"operator profiles: forge profile unreadable; keeping the stored entry"
);
outcome.kept += 1;
continue;
stored
.iter()
.find(|o| o.name == *name)
.and_then(|o| o.pronouns.clone())
}
};
let profile = OperatorProfile {
list.push(OperatorProfile {
name: name.clone(),
pronouns,
};
let stored = store
.get(name)
.await
.with_context(|| format!("reading the profile of {name}"))?
.and_then(|value| serde_json::from_slice::<OperatorProfile>(&value).ok());
if stored.as_ref() != Some(&profile) {
let value = serde_json::to_vec(&profile).context("encoding an operator profile")?;
store
.put(name, value.into())
.await
.with_context(|| format!("writing the profile of {name}"))?;
outcome.written += 1;
}
});
}
for name in stale {
list.sort_by(|a, b| a.name.cmp(&b.name));
list.dedup_by(|a, b| a.name == b.name);
if list != stored {
let value = serde_json::to_vec(&list).context("encoding the operator list")?;
store
.delete(&name)
.put(KEY, value.into())
.await
.with_context(|| format!("removing the profile of {name}"))?;
outcome.removed += 1;
.context("writing the operator list")?;
outcome.written = true;
}
Ok(outcome)
}
/// One pass: read the operator group, then [`sync`] the bucket to it. Fails
/// as a whole, changing nothing, when the group cannot be read.
/// One pass: read the operator group, then [`sync`] the key to it. Fails as a
/// whole, changing nothing, when the group cannot be read.
async fn pass(forge: &Client, bridge: &AuthBridge, queue: &async_nats::Client) -> Result<Outcome> {
let operators = bridge
.list_operators()
@ -158,11 +135,9 @@ pub fn spawn(forge: Arc<Client>, bridge: Arc<AuthBridge>, queue: async_nats::Cli
continue;
};
match outcome {
Ok(outcome) if outcome.written > 0 || outcome.removed > 0 => tracing::info!(
written = outcome.written,
removed = outcome.removed,
Ok(outcome) if outcome.written => tracing::info!(
kept = outcome.kept,
"operator profiles: bucket updated"
"operator profiles: operator list updated"
),
Ok(outcome) => tracing::debug!(kept = outcome.kept, "operator profiles: unchanged"),
Err(e) => tracing::warn!(
@ -199,18 +174,8 @@ mod tests {
}
async fn stored(store: &Store) -> Vec<OperatorProfile> {
let keys: Vec<String> = store.keys().await.unwrap().try_collect().await.unwrap();
let mut out = Vec::new();
for key in keys {
let value = store
.get(&key)
.await
.unwrap()
.expect("a listed key has a value");
out.push(serde_json::from_slice(&value).unwrap());
}
out.sort_by(|a: &OperatorProfile, b| a.name.cmp(&b.name));
out
let value = store.get(KEY).await.unwrap().expect("the list is written");
serde_json::from_slice(&value).unwrap()
}
fn profile(name: &str, pronouns: Option<&str>) -> OperatorProfile {
@ -221,34 +186,36 @@ mod tests {
}
#[tokio::test]
async fn profiles_are_written_and_a_departed_operator_is_removed() {
async fn the_list_is_written_once_and_a_departed_operator_drops_out() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let store = operator_profiles::open_or_create(&client).await.unwrap();
let forge = Fake([("mara", Ok(Some("she/her"))), ("sol", Ok(None))].into());
let outcome = sync(&names(&["mara", "sol"]), &forge, &store)
let outcome = sync(&names(&["sol", "mara"]), &forge, &store)
.await
.unwrap();
assert_eq!(outcome.written, 2);
assert!(outcome.written);
assert_eq!(
stored(&store).await,
[profile("mara", Some("she/her")), profile("sol", None)]
);
let revision = store.entry(KEY).await.unwrap().unwrap().revision;
// An unchanged profile is not rewritten.
// An unchanged list is not rewritten.
let outcome = sync(&names(&["mara", "sol"]), &forge, &store)
.await
.unwrap();
assert_eq!(outcome, Outcome::default());
assert_eq!(store.entry(KEY).await.unwrap().unwrap().revision, revision);
let outcome = sync(&names(&["mara"]), &forge, &store).await.unwrap();
assert_eq!(outcome.removed, 1);
assert!(outcome.written);
assert_eq!(stored(&store).await, [profile("mara", Some("she/her"))]);
}
#[tokio::test]
async fn a_failed_forge_read_keeps_the_stored_value() {
async fn a_failed_forge_read_keeps_the_stored_entry() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let store = operator_profiles::open_or_create(&client).await.unwrap();
@ -264,27 +231,57 @@ mod tests {
.unwrap();
// Control: the same sync does change a value the forge reads anew.
let failing = Fake([("mara", Err(())), ("sol", Ok(Some("he/him")))].into());
let outcome = sync(&names(&["mara", "sol"]), &failing, &store)
// `ren` is new and unreadable, so it is listed by name only.
let failing = Fake(
[
("mara", Err(())),
("sol", Ok(Some("he/him"))),
("ren", Err(())),
]
.into(),
);
let outcome = sync(&names(&["mara", "sol", "ren"]), &failing, &store)
.await
.unwrap();
assert_eq!(outcome.kept, 1);
assert_eq!(outcome.removed, 0);
assert_eq!(outcome.kept, 2);
assert_eq!(
stored(&store).await,
[
profile("mara", Some("she/her")),
profile("ren", None),
profile("sol", Some("he/him"))
]
);
}
#[test]
fn a_dotted_or_empty_name_is_no_key() {
assert!(is_key("mara"));
assert!(is_key("op_2-x"));
assert!(!is_key(""));
assert!(!is_key("ma.ra"));
assert!(!is_key("ma*"));
/// The agent's read (`operator_profiles::read`, one direct get) finds
/// what a pass wrote, an empty list for a bucket with no key yet, and
/// `NoBucket` before the bucket exists.
#[tokio::test]
async fn the_agent_read_finds_the_list_and_falls_back_without_one() {
use swarm_queue_client::operator_profiles::{Read, read};
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
assert!(matches!(read(&client).await.unwrap(), Read::NoBucket));
let store = operator_profiles::open_or_create(&client).await.unwrap();
let Read::Current(empty) = read(&client).await.unwrap() else {
panic!("the bucket exists");
};
assert_eq!(empty, []);
let forge = Fake([("mara", Ok(Some("she/her")))].into());
sync(&names(&["mara"]), &forge, &store).await.unwrap();
let Read::Current(list) = read(&client).await.unwrap() else {
panic!("the bucket exists");
};
assert_eq!(list, [profile("mara", Some("she/her"))]);
store.delete(KEY).await.unwrap();
let Read::Current(deleted) = read(&client).await.unwrap() else {
panic!("the bucket exists");
};
assert_eq!(deleted, []);
}
}