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

copy the cache's pull facts to each hive and create them in the in-swarm cache

Browse source 
the in-swarm cache stores its address, name and public key beside its tokens, create-only. swarm-controller copies the four cache facts to swarm/hives/<hive>/attic/* every five minutes, with read on the sources and write on the copies. part 1 of 3: nothing reads the copies until the hive reader and probe land.
This commit is contained in:
atlas 2026-10-27 15:51:13 +01:00 • committed by mara
commit 1e5070796d
3 changed files with 640 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

323
swarm-controller/src/attic_copy.rs Normal file
View file

@ -0,0 +1,323 @@
//! The cache facts a hive pulls with, copied to where the hive may read them.
//!
//! The cache's operator keeps four facts under `swarm/services/attic/`
//! ([`swarm_secret_client::attic`]). A hive's store identity reads
//! `swarm/hives/<hive>/*` and nothing under `swarm/services/`, so every
//! [`RECONCILE_INTERVAL`] this writes each fact that is stored to every hive's
//! `attic/` prefix, and leaves a copy that already holds the value alone.
//!
//! A fact that is not stored is not copied, and a copy whose source is gone
//! stays: the pass never deletes. A swarm with no cache stores none, so the
//! pass writes nothing.
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
use swarm_secret_client::SecretStore;
use swarm_secret_client::attic::{Fact, Leaf, hive_path, service_path};
/// How often [`spawn`] copies.
const RECONCILE_INTERVAL: Duration = Duration::from_mins(5);
/// A pass that has not finished by now is given up on; the next tick retries.
const PASS_DEADLINE: Duration = Duration::from_mins(2);
/// The two store calls a pass makes, so a test can stand in for the store.
pub(crate) trait FactStore {
/// The fact at `path`; `None` when nothing is stored there.
async fn read(&self, path: &str) -> Result<Option<String>>;
/// Store `value` at `path`, replacing what is there.
async fn write(&self, path: &str, value: &str) -> Result<()>;
}
impl FactStore for SecretStore {
async fn read(&self, path: &str) -> Result<Option<String>> {
Ok(self.read_optional::<Fact>(path).await?.map(|f| f.value))
}
async fn write(&self, path: &str, value: &str) -> Result<()> {
let fact = Fact {
value: value.to_owned(),
};
Ok(SecretStore::write(self, path, &fact).await?)
}
}
/// What one pass did.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct Copied {
/// Copies written because they were absent or held another value.
pub written: usize,
/// Hives whose copies could not all be checked or written.
pub failed: usize,
}
/// The stored facts, as `(leaf, value)`. An empty value counts as not stored.
async fn sources(store: &impl FactStore) -> Result<Vec<(Leaf, String)>> {
let mut found = Vec::new();
for leaf in Leaf::ALL {
let path = service_path(leaf)?;
let stored = store
.read(&path)
.await
.with_context(|| format!("reading {path}"))?;
if let Some(value) = stored.filter(|v| !v.trim().is_empty()) {
found.push((leaf, value));
}
}
Ok(found)
}
/// Bring `hive`'s copies up to `sources`; how many were written.
async fn copy_hive(
store: &impl FactStore,
hive: &str,
sources: &[(Leaf, String)],
) -> Result<usize> {
let mut written = 0;
for (leaf, value) in sources {
let path = hive_path(hive, *leaf)?;
let current = store
.read(&path)
.await
.with_context(|| format!("reading {path}"))?;
if current.as_deref() == Some(value.as_str()) {
continue;
}
store
.write(&path, value)
.await
.with_context(|| format!("writing {path}"))?;
written += 1;
}
Ok(written)
}
/// One pass over `hives`.
///
/// # Errors
/// When a source cannot be read, which stops the pass before any copy: a
/// failed read is not an absent fact. A hive that fails is counted in
/// [`Copied::failed`] and logged, and the others are still copied.
pub(crate) async fn copy_pass(store: &impl FactStore, hives: &[String]) -> Result<Copied> {
let sources = sources(store).await?;
let mut done = Copied::default();
for hive in hives {
match copy_hive(store, hive, &sources).await {
Ok(n) => done.written += n,
Err(e) => {
tracing::warn!(hive, error = %format!("{e:#}"), "attic facts: copy failed");
done.failed += 1;
}
}
}
Ok(done)
}
/// Copy now and every [`RECONCILE_INTERVAL`] after. A pass that fails is
/// logged and retried on the next tick; it never stops the daemon.
pub fn spawn(hives: Vec<String>) {
let hives = Arc::new(hives);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
loop {
ticker.tick().await;
let pass = async {
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
copy_pass(&store, &hives).await
};
match tokio::time::timeout(PASS_DEADLINE, pass).await {
Ok(Ok(done)) if done.written > 0 || done.failed > 0 => tracing::info!(
written = done.written,
failed = done.failed,
"attic facts: copied to the hives"
),
Ok(Ok(_)) => tracing::debug!("attic facts: every hive's copy is current"),
Ok(Err(e)) => tracing::warn!(
error = %format!("{e:#}"),
retry_in_s = RECONCILE_INTERVAL.as_secs(),
"attic facts: pass failed; retrying next tick"
),
Err(_) => tracing::warn!(
deadline_s = PASS_DEADLINE.as_secs(),
"attic facts: pass timed out; retrying next tick"
),
}
}
});
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::sync::Mutex;
use super::*;
/// An in-memory store that counts writes and can refuse reads or writes
/// under a prefix.
#[derive(Default)]
struct Memory {
kv: Mutex<BTreeMap<String, String>>,
writes: Mutex<Vec<String>>,
refuse: Option<&'static str>,
}
impl Memory {
fn with(pairs: &[(&str, &str)]) -> Self {
let m = Self::default();
for (path, value) in pairs {
m.kv.lock()
.unwrap()
.insert((*path).to_owned(), (*value).to_owned());
}
m
}
fn get(&self, path: &str) -> Option<String> {
self.kv.lock().unwrap().get(path).cloned()
}
}
impl FactStore for Memory {
async fn read(&self, path: &str) -> Result<Option<String>> {
Ok(self.get(path))
}
async fn write(&self, path: &str, value: &str) -> Result<()> {
if self.refuse.is_some_and(|p| path.starts_with(p)) {
anyhow::bail!("denied");
}
self.writes.lock().unwrap().push(path.to_owned());
self.kv
.lock()
.unwrap()
.insert(path.to_owned(), value.to_owned());
Ok(())
}
}
const SOURCES: [(&str, &str); 4] = [
("swarm/services/attic/pull-token", "tok"),
("swarm/services/attic/url", "https://cache.example"),
("swarm/services/attic/cache", "swarm"),
("swarm/services/attic/public-key", "swarm:AAAA"),
];
fn hives() -> Vec<String> {
vec!["alpha".to_owned(), "beta".to_owned()]
}
#[tokio::test]
async fn present_facts_reach_every_hive() {
let store = Memory::with(&SOURCES);
let done = copy_pass(&store, &hives()).await.expect("pass");
assert_eq!(
done,
Copied {
written: 8,
failed: 0
}
);
for hive in hives() {
assert_eq!(
store
.get(&format!("swarm/hives/{hive}/attic/pull-token"))
.as_deref(),
Some("tok")
);
assert_eq!(
store
.get(&format!("swarm/hives/{hive}/attic/public-key"))
.as_deref(),
Some("swarm:AAAA")
);
}
// Presence control for the absence cases below: copies are the only
// thing a pass writes under a hive.
assert!(store.get("swarm/hives/alpha/forge/token").is_none());
}
#[tokio::test]
async fn absent_facts_write_nothing() {
let store = Memory::default();
let done = copy_pass(&store, &hives()).await.expect("pass");
assert_eq!(done, Copied::default());
assert!(store.writes.lock().unwrap().is_empty());
}
#[tokio::test]
async fn only_the_stored_facts_are_copied_and_an_empty_one_is_not_stored() {
let store = Memory::with(&[
("swarm/services/attic/pull-token", "tok"),
("swarm/services/attic/url", " \n"),
]);
copy_pass(&store, &hives()).await.expect("pass");
assert_eq!(
store.get("swarm/hives/alpha/attic/pull-token").as_deref(),
Some("tok")
);
assert!(store.get("swarm/hives/alpha/attic/url").is_none());
assert!(store.get("swarm/hives/alpha/attic/cache").is_none());
}
#[tokio::test]
async fn a_current_copy_is_not_rewritten_and_a_changed_value_is() {
let store = Memory::with(&SOURCES);
copy_pass(&store, &hives()).await.expect("first");
let again = copy_pass(&store, &hives()).await.expect("second");
assert_eq!(again, Copied::default(), "unchanged values are not written");
store.kv.lock().unwrap().insert(
"swarm/services/attic/pull-token".to_owned(),
"rotated".to_owned(),
);
let rotated = copy_pass(&store, &hives()).await.expect("third");
assert_eq!(
rotated,
Copied {
written: 2,
failed: 0
}
);
assert_eq!(
store.get("swarm/hives/beta/attic/pull-token").as_deref(),
Some("rotated")
);
}
#[tokio::test]
async fn a_removed_source_leaves_the_copies() {
let store = Memory::with(&SOURCES);
copy_pass(&store, &hives()).await.expect("first");
store
.kv
.lock()
.unwrap()
.remove("swarm/services/attic/pull-token");
copy_pass(&store, &hives()).await.expect("second");
assert_eq!(
store.get("swarm/hives/alpha/attic/pull-token").as_deref(),
Some("tok")
);
}
#[tokio::test]
async fn one_hive_failing_does_not_stop_the_others() {
let mut store = Memory::with(&SOURCES);
store.refuse = Some("swarm/hives/alpha/");
let done = copy_pass(&store, &hives()).await.expect("pass");
assert_eq!(
done,
Copied {
written: 4,
failed: 1
}
);
assert!(store.get("swarm/hives/beta/attic/cache").is_some());
assert!(store.get("swarm/hives/alpha/attic/cache").is_none());
}
}

1
swarm-controller/src/main.rs
View file

@ -37,6 +37,7 @@ mod agents_md;
mod agents_md_initial;
mod alerts;
mod archetypes;
mod attic_copy;
mod auth;
mod builder;
mod claude_login;

4
swarm-controller/src/startup.rs
View file

@ -367,7 +367,8 @@ fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[H
/// The per-agent background passes: subagent and terminal streams, and
/// schedule keys, and the permissions bucket and its import, and the
/// archetypes agents are created from.
/// archetypes agents are created from. Beside them, the
/// copy of the cache facts to each hive.
/// Lifted out of `main` for `clippy::too_many_lines`.
pub(crate) fn spawn_per_agent_upkeep(
status: Option<&Arc<status::StatusReader>>,
@ -376,6 +377,7 @@ pub(crate) fn spawn_per_agent_upkeep(
spawn_subagent_streams(status, hives);
spawn_schedule_reconcile(status, hives);
spawn_agent_permissions(status, hives);
crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
if let Some(status) = status {
archetypes::spawn(status.queue_client());
}