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

swarm-archive: grant only once an archive is up; keep logins across a gap in placements

Browse source 
The archive host creates a ready marker in the store and swarm-controller
writes no credentials before it exists, so a swarm without an archive, and a
mixed-version swarm, leave every hive backup a logged no-op. A hive with no
declaration no longer counts as placing nothing: its copies and its agents'
user lines stay. The restic password is pinned by tests across a revoke, a
re-placement and a move.
This commit is contained in:
atlas 2026-11-10 08:18:45 +01:00
commit 91f36f39d2
2 changed files with 241 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

156
swarm-controller/src/archive_access.rs
View file

@ -199,21 +199,37 @@ async fn leaves(store: &impl ArchiveStore, prefix: &str) -> Result<Vec<String>>
/// One pass: grant every agent in `placed`, then revoke what no placement
/// justifies. `hives` is every hive of the swarm, whose copies are checked.
///
/// Nothing happens until the archive has announced itself at
/// [`archive::ready_path`]: a swarm without one gets no credentials. A hive in
/// `undeclared` has no declaration to read, which does not show it places
/// nothing, so its copies, and the user lines of the agents they name, stay.
///
/// # Errors
/// Never: a failure is counted in [`Done::failed`] and logged, and the rest of
/// the pass still runs.
pub(crate) async fn reconcile(
store: &impl ArchiveStore,
placed: &BTreeSet<(String, String)>,
undeclared: &BTreeSet<String>,
hives: &[String],
) -> Done {
let mut done = Done::default();
match store.read(&archive::ready_path().unwrap_or_default()).await {
Ok(Some(_)) => {}
Ok(None) => return done,
Err(e) => {
tracing::warn!(error = %format!("{e:#}"), "archive: reading whether the archive is up failed");
done.failed += 1;
return done;
}
}
let mut holders: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for (hive, agent) in placed {
holders.entry(agent).or_default().push(hive);
}
let mut ambiguous = BTreeSet::new();
let mut unconfirmed = BTreeSet::new();
for (agent, on) in &holders {
if let [hive] = on.as_slice() {
match grant(store, hive, agent).await {
@ -243,6 +259,10 @@ pub(crate) async fn reconcile(
continue;
}
};
if undeclared.contains(hive) {
unconfirmed.extend(copies);
continue;
}
for agent in copies {
let keep = ambiguous.contains(agent.as_str())
|| placed.contains(&(hive.clone(), agent.clone()));
@ -255,7 +275,7 @@ pub(crate) async fn reconcile(
match leaves(store, &archive::users_prefix().unwrap_or_default()).await {
Ok(users) => {
for agent in users {
if !holders.contains_key(agent.as_str()) {
if !holders.contains_key(agent.as_str()) && !unconfirmed.contains(&agent) {
revoke(store, archive::user_path(&agent), &mut done).await;
}
}
@ -303,11 +323,12 @@ pub fn spawn(wanted: Arc<WantedWriter>, hives: Vec<String>) {
loop {
ticker.tick().await;
let pass = async {
let placed = crate::jobq::placements(&wanted, &hives).await?;
let (placed, undeclared) =
crate::jobq::placements_with_gaps(&wanted, &hives).await?;
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
anyhow::Ok(reconcile(&store, &placed, &hives).await)
anyhow::Ok(reconcile(&store, &placed, &undeclared, &hives).await)
};
match tokio::time::timeout(PASS_DEADLINE, pass).await {
Ok(Ok(done)) if done != Done::default() => tracing::info!(
@ -347,6 +368,16 @@ mod tests {
}
impl Memory {
/// A store whose archive has announced itself.
fn up() -> Self {
let store = Self::default();
store.kv.lock().unwrap().insert(
archive::ready_path().unwrap(),
fields(&[(archive::READY, "1")]),
);
store
}
fn get(&self, path: &str) -> Option<Fields> {
self.kv.lock().unwrap().get(path).cloned()
}
@ -414,6 +445,10 @@ mod tests {
.collect()
}
fn none() -> BTreeSet<String> {
BTreeSet::new()
}
fn hives() -> Vec<String> {
vec!["alpha".to_owned(), "beta".to_owned()]
}
@ -434,8 +469,8 @@ mod tests {
#[tokio::test]
async fn a_placed_agent_gets_its_hive_a_password_and_the_archive_its_hash() {
let store = Memory::default();
let done = reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
let store = Memory::up();
let done = reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
assert_eq!(
done,
Done {
@ -460,23 +495,23 @@ mod tests {
#[tokio::test]
async fn a_current_grant_is_not_rewritten() {
let store = Memory::default();
let store = Memory::up();
let one = placed(&[("alpha", "atlas")]);
reconcile(&store, &one, &hives()).await;
reconcile(&store, &one, &none(), &hives()).await;
let before = store.get(ALPHA_ATLAS);
let done = reconcile(&store, &one, &hives()).await;
let done = reconcile(&store, &one, &none(), &hives()).await;
assert_eq!(done, Done::default());
assert_eq!(store.get(ALPHA_ATLAS), before);
}
#[tokio::test]
async fn a_move_replaces_the_password_and_takes_the_old_hives_copy() {
let store = Memory::default();
reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
let store = Memory::up();
reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
let old_rest = store.value(ALPHA_ATLAS, REST_PASSWORD).unwrap();
let restic = store.value(RESTIC_ATLAS, PASSWORD).unwrap();
let done = reconcile(&store, &placed(&[("beta", "atlas")]), &hives()).await;
let done = reconcile(&store, &placed(&[("beta", "atlas")]), &none(), &hives()).await;
assert_eq!(done.granted, 1);
assert_eq!(done.revoked, 1);
assert!(store.get(ALPHA_ATLAS).is_none(), "the old hive loses read");
@ -499,14 +534,15 @@ mod tests {
#[tokio::test]
async fn a_destroyed_agent_loses_every_credential_but_the_restic_password() {
let store = Memory::default();
let store = Memory::up();
reconcile(
&store,
&placed(&[("alpha", "atlas"), ("beta", "argus")]),
&none(),
&hives(),
)
.await;
let done = reconcile(&store, &placed(&[("beta", "argus")]), &hives()).await;
let done = reconcile(&store, &placed(&[("beta", "argus")]), &none(), &hives()).await;
assert_eq!(done.revoked, 2);
assert!(store.get(ALPHA_ATLAS).is_none());
assert!(store.get(USER_ATLAS).is_none());
@ -521,30 +557,34 @@ mod tests {
#[tokio::test]
async fn an_agent_two_hives_place_is_left_as_it_is() {
let store = Memory::default();
reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
let store = Memory::up();
reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
let before = (store.get(ALPHA_ATLAS), store.get(USER_ATLAS));
let both = placed(&[("alpha", "atlas"), ("beta", "atlas")]);
let done = reconcile(&store, &both, &hives()).await;
let done = reconcile(&store, &both, &none(), &hives()).await;
assert_eq!(done, Done::default());
assert_eq!((store.get(ALPHA_ATLAS), store.get(USER_ATLAS)), before);
assert!(store.get(BETA_ATLAS).is_none());
// Neither hive is granted when the agent is new and both place it.
let fresh = Memory::default();
let done = reconcile(&fresh, &both, &hives()).await;
let fresh = Memory::up();
let done = reconcile(&fresh, &both, &none(), &hives()).await;
assert_eq!(done, Done::default());
assert!(fresh.kv.lock().unwrap().is_empty());
assert_eq!(
fresh.kv.lock().unwrap().keys().collect::<Vec<_>>(),
[&archive::ready_path().unwrap()],
"only the archive's own marker is stored"
);
}
#[tokio::test]
async fn a_lost_user_line_is_rewritten_from_the_hives_password() {
let store = Memory::default();
let store = Memory::up();
let one = placed(&[("alpha", "atlas")]);
reconcile(&store, &one, &hives()).await;
reconcile(&store, &one, &none(), &hives()).await;
let line = store.value(USER_ATLAS, HTPASSWD);
store.kv.lock().unwrap().remove(USER_ATLAS);
let done = reconcile(&store, &one, &hives()).await;
let done = reconcile(&store, &one, &none(), &hives()).await;
assert_eq!(done.granted, 1);
assert_eq!(store.value(USER_ATLAS, HTPASSWD), line);
}
@ -553,11 +593,12 @@ mod tests {
async fn a_refused_write_counts_as_failed_and_the_other_agents_still_get_theirs() {
let store = Memory {
refuse: Some("swarm/hives/alpha/"),
..Memory::default()
..Memory::up()
};
let done = reconcile(
&store,
&placed(&[("alpha", "atlas"), ("beta", "argus")]),
&none(),
&hives(),
)
.await;
@ -569,4 +610,73 @@ mod tests {
"no hash without the hive's copy"
);
}
#[tokio::test]
async fn the_restic_password_survives_a_revoke_and_a_new_grant_and_a_move() {
let store = Memory::up();
let atlas = |hive: &str| placed(&[(hive, "atlas")]);
reconcile(&store, &atlas("alpha"), &none(), &hives()).await;
let restic = store.value(RESTIC_ATLAS, PASSWORD).expect("minted once");
let rest = store.value(ALPHA_ATLAS, REST_PASSWORD).expect("granted");
// Nothing places the agent: every credential goes, the restic one stays.
let nowhere = placed(&[]);
reconcile(&store, &nowhere, &none(), &hives()).await;
assert!(store.get(ALPHA_ATLAS).is_none() && store.get(USER_ATLAS).is_none());
assert_eq!(store.value(RESTIC_ATLAS, PASSWORD).as_ref(), Some(&restic));
// Placed again: the login is new, the repository's password is the one it was created with.
reconcile(&store, &atlas("alpha"), &none(), &hives()).await;
assert_ne!(store.value(ALPHA_ATLAS, REST_PASSWORD), Some(rest));
assert_eq!(
store.value(ALPHA_ATLAS, RESTIC_PASSWORD).as_ref(),
Some(&restic)
);
assert_eq!(store.value(RESTIC_ATLAS, PASSWORD).as_ref(), Some(&restic));
// And across a move.
reconcile(&store, &atlas("beta"), &none(), &hives()).await;
assert_eq!(
store.value(BETA_ATLAS, RESTIC_PASSWORD).as_ref(),
Some(&restic)
);
assert_eq!(store.value(RESTIC_ATLAS, PASSWORD).as_ref(), Some(&restic));
}
#[tokio::test]
async fn a_hive_with_no_declaration_keeps_its_copies_and_its_agents_user_lines() {
let store = Memory::up();
reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
let (copy, line) = (store.get(ALPHA_ATLAS), store.get(USER_ATLAS));
// alpha's declaration is not there: it does not place atlas, and that proves nothing.
let gap: BTreeSet<String> = ["alpha".to_owned()].into();
let done = reconcile(&store, &placed(&[]), &gap, &hives()).await;
assert_eq!(done, Done::default());
assert_eq!(
(store.get(ALPHA_ATLAS), store.get(USER_ATLAS)),
(copy, line)
);
// Control: with the declaration read, the same placements revoke.
let done = reconcile(&store, &placed(&[]), &none(), &hives()).await;
assert_eq!(done.revoked, 2);
assert!(store.get(ALPHA_ATLAS).is_none() && store.get(USER_ATLAS).is_none());
}
#[tokio::test]
async fn a_swarm_whose_archive_has_not_announced_itself_gets_nothing() {
let store = Memory::default();
let done = reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
assert_eq!(done, Done::default());
assert!(store.kv.lock().unwrap().is_empty());
// Control: the same placements, once the archive is up.
store.kv.lock().unwrap().insert(
archive::ready_path().unwrap(),
fields(&[(archive::READY, "1")]),
);
let done = reconcile(&store, &placed(&[("alpha", "atlas")]), &none(), &hives()).await;
assert_eq!(done.granted, 1);
}
}

17
swarm-controller/src/jobq.rs
View file

@ -965,13 +965,28 @@ pub(crate) async fn placements(
wanted: &wanted::WantedWriter,
hives: &[String],
) -> Result<std::collections::BTreeSet<(String, String)>> {
Ok(placements_with_gaps(wanted, hives).await?.0)
}
/// [`placements`], and the hives that have no declaration at all. Such a hive
/// places nothing here, but the queue cannot tell a hive that never had agents
/// from one whose declaration is not there (yet).
pub(crate) async fn placements_with_gaps(
wanted: &wanted::WantedWriter,
hives: &[String],
) -> Result<(
std::collections::BTreeSet<(String, String)>,
std::collections::BTreeSet<String>,
)> {
let mut placed = std::collections::BTreeSet::new();
let mut undeclared = std::collections::BTreeSet::new();
for hive in hives {
let Some(declared) = wanted
.view(hive)
.await
.map_err(|e| e.context(format!("read where hive {hive} places agents")))?
else {
undeclared.insert(hive.clone());
continue;
};
placed.extend(
@ -982,7 +997,7 @@ pub(crate) async fn placements(
.map(|(name, _)| (hive.clone(), name)),
);
}
Ok(placed)
Ok((placed, undeclared))
}
/// Every agent in `placements`, with its archetype.