swarm-controller: archive a destroyed agent's config repo, unarchive it when the name is created again
This commit is contained in:
parent
814edd1d67
commit
7beb1193f4
5 changed files with 197 additions and 3 deletions
36
swarm-controller/src/agent_config_repo.rs
Normal file
36
swarm-controller/src/agent_config_repo.rs
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
//! A destroyed agent's config repo (`agent-configs/<agent>`): archived when the
|
||||
//! agent is declared `Destroyed`, so it stays readable but takes no more
|
||||
//! pushes or PRs, and unarchived when an agent of the same name is created
|
||||
//! again (`forge::Client::create_repo`).
|
||||
|
||||
use crate::routes::AppState;
|
||||
|
||||
/// Archive `agent`'s config repo. A failure is logged and does not stop the
|
||||
/// destroy, as in `agent_archetype::clear_on_destroy`: an agent that cannot be
|
||||
/// torn down is the larger harm, and declaring it destroyed again re-runs
|
||||
/// this.
|
||||
pub(crate) async fn archive_on_destroy(state: &AppState, hive: &str, agent: &str) {
|
||||
let Some(forge) = state.forge.get() else {
|
||||
tracing::warn!(
|
||||
hive,
|
||||
agent,
|
||||
"no forge access here, so the destroyed agent's config repo was left as it is"
|
||||
);
|
||||
return;
|
||||
};
|
||||
match forge.set_config_repo_archived(agent, true).await {
|
||||
Ok(true) => tracing::info!(hive, agent, "destroyed agent's config repo archived"),
|
||||
Ok(false) => tracing::debug!(
|
||||
hive,
|
||||
agent,
|
||||
"destroyed agent's config repo was already archived, or does not exist"
|
||||
),
|
||||
Err(e) => tracing::error!(
|
||||
hive,
|
||||
agent,
|
||||
error = %format!("{e:#}"),
|
||||
"archiving this agent's config repo FAILED; it was destroyed anyway, so the repo \
|
||||
still takes pushes. Declaring the agent destroyed again re-runs this."
|
||||
),
|
||||
}
|
||||
}
|
||||
|
|
@ -669,14 +669,55 @@ impl Client {
|
|||
/// Ensures the org and its [`OPERATORS_TEAM`] first: the periodic
|
||||
/// [`objects`] pass does too, but an agent can be created before that
|
||||
/// pass has succeeded once, and the rule below names the team.
|
||||
///
|
||||
/// An archived repo, left by an earlier agent of the same name that was
|
||||
/// destroyed ([`Self::set_config_repo_archived`]), is unarchived before the
|
||||
/// rule is applied: an archived repo refuses every write.
|
||||
pub async fn create_repo(&self, repo: &str) -> Result<String> {
|
||||
self.ensure_merge_gate_prerequisites().await?;
|
||||
self.ensure_org_repo(repo).await?;
|
||||
self.set_config_repo_archived(repo, false).await?;
|
||||
self.apply_operator_branch_protection(repo).await?;
|
||||
tracing::info!(%repo, "swarm forge: created repo in {CONFIG_ORG} with operator merge gate");
|
||||
Ok(format!("{CONFIG_ORG}/{repo}"))
|
||||
}
|
||||
|
||||
/// Archive or unarchive `repo` in [`CONFIG_ORG`]: read-only while archived,
|
||||
/// so a destroyed agent's config can no longer be pushed to or have PRs
|
||||
/// opened against it, and it stays readable.
|
||||
///
|
||||
/// Returns whether anything changed. A repo already in the requested state
|
||||
/// and a repo that does not exist both return `false`: destroying an agent
|
||||
/// that never got a config repo is not a fault, and a repeated destroy is
|
||||
/// ordinary.
|
||||
///
|
||||
/// # Errors
|
||||
/// A forge failure reading the repo or editing it.
|
||||
pub async fn set_config_repo_archived(&self, repo: &str, archived: bool) -> Result<bool> {
|
||||
let current = match self
|
||||
.api
|
||||
.call(|api| api.repo_get(CONFIG_ORG, repo).into_future())
|
||||
.await
|
||||
{
|
||||
Ok(r) => r.archived.unwrap_or(false),
|
||||
Err(e) if is_repo_not_found(&e) => return Ok(false),
|
||||
Err(e) => return Err(e).with_context(|| format!("get repo {CONFIG_ORG}/{repo}")),
|
||||
};
|
||||
if current == archived {
|
||||
return Ok(false);
|
||||
}
|
||||
let mut edit = objects::sparse_edit_repo_option();
|
||||
edit.archived = Some(archived);
|
||||
self.api
|
||||
.call(|api| api.repo_edit(CONFIG_ORG, repo, edit.clone()).into_future())
|
||||
.await
|
||||
.with_context(|| {
|
||||
let verb = if archived { "archive" } else { "unarchive" };
|
||||
format!("{verb} repo {CONFIG_ORG}/{repo}")
|
||||
})?;
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Add `agent` as a **write** collaborator on `repo` (can push + open
|
||||
/// PRs, cannot bypass branch protection) — the whole job of the
|
||||
/// `AddRepoMember` node. Idempotent.
|
||||
|
|
@ -2273,6 +2314,112 @@ impl Client {
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A stand-in forge holding `agent-configs/atlas`, archived or not, that
|
||||
/// records each PATCH body and flips the flag when it receives one. Any
|
||||
/// other repo is a 404.
|
||||
async fn config_repo_forge(
|
||||
archived: bool,
|
||||
) -> (Client, Arc<std::sync::Mutex<Vec<serde_json::Value>>>) {
|
||||
use axum::http::{Method, Uri};
|
||||
|
||||
let edits = Arc::new(std::sync::Mutex::new(Vec::new()));
|
||||
let state = Arc::new(std::sync::Mutex::new(archived));
|
||||
let (seen, flag) = (Arc::clone(&edits), Arc::clone(&state));
|
||||
let app = axum::Router::new().fallback(
|
||||
move |method: Method, uri: Uri, body: axum::body::Bytes| {
|
||||
let (seen, flag) = (Arc::clone(&seen), Arc::clone(&flag));
|
||||
async move {
|
||||
if uri.path() != "/api/v1/repos/agent-configs/atlas" {
|
||||
return (StatusCode::NOT_FOUND, r#"{"message":"no"}"#.to_owned());
|
||||
}
|
||||
if method == axum::http::Method::PATCH {
|
||||
let edit: serde_json::Value = serde_json::from_slice(&body).unwrap();
|
||||
if let Some(now) = edit["archived"].as_bool() {
|
||||
*flag.lock().unwrap() = now;
|
||||
}
|
||||
// The client sends every unset field as `null`, which the
|
||||
// forge reads as "leave alone"; keep what is actually set.
|
||||
let set: serde_json::Map<_, _> = edit
|
||||
.as_object()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|(_, v)| !v.is_null())
|
||||
.map(|(k, v)| (k.clone(), v.clone()))
|
||||
.collect();
|
||||
seen.lock().unwrap().push(serde_json::Value::Object(set));
|
||||
}
|
||||
let archived = *flag.lock().unwrap();
|
||||
(
|
||||
StatusCode::OK,
|
||||
format!(r#"{{"name":"atlas","archived":{archived}}}"#),
|
||||
)
|
||||
}
|
||||
},
|
||||
);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
|
||||
tokio::spawn(async move { axum::serve(listener, app).await });
|
||||
(Client::for_test_forge(url), edits)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_destroyed_agents_config_repo_is_archived_once() {
|
||||
let (client, edits) = config_repo_forge(false).await;
|
||||
assert!(
|
||||
client
|
||||
.set_config_repo_archived("atlas", true)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
assert_eq!(
|
||||
*edits.lock().unwrap(),
|
||||
[serde_json::json!({"archived": true})],
|
||||
"one edit, touching nothing but the flag"
|
||||
);
|
||||
assert!(
|
||||
!client
|
||||
.set_config_repo_archived("atlas", true)
|
||||
.await
|
||||
.unwrap(),
|
||||
"a repeated destroy changes nothing"
|
||||
);
|
||||
assert_eq!(edits.lock().unwrap().len(), 1, "and sends no second edit");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_new_agent_of_the_same_name_gets_its_repo_unarchived() {
|
||||
let (client, edits) = config_repo_forge(true).await;
|
||||
assert!(
|
||||
client
|
||||
.set_config_repo_archived("atlas", false)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
assert_eq!(
|
||||
*edits.lock().unwrap(),
|
||||
[serde_json::json!({"archived": false})]
|
||||
);
|
||||
assert!(
|
||||
!client
|
||||
.set_config_repo_archived("atlas", false)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_agent_without_a_config_repo_is_not_an_error_to_archive() {
|
||||
let (client, edits) = config_repo_forge(false).await;
|
||||
assert!(
|
||||
!client
|
||||
.set_config_repo_archived("ghost", true)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
assert!(edits.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
/// Commit stats sum per side; a commit with none, or a negative figure,
|
||||
/// adds zero.
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -594,7 +594,7 @@ fn seen<T, U>(res: Result<T, ForgejoError>, what: &str, f: impl FnOnce(T) -> U)
|
|||
|
||||
/// `EditRepoOption` with every field unset — repo edits only ever change the
|
||||
/// one field the caller sets on top (Forgejo leaves `None` fields untouched).
|
||||
fn sparse_edit_repo_option() -> EditRepoOption {
|
||||
pub(super) fn sparse_edit_repo_option() -> EditRepoOption {
|
||||
EditRepoOption {
|
||||
allow_fast_forward_only_merge: None,
|
||||
allow_manual_merge: None,
|
||||
|
|
@ -728,7 +728,13 @@ impl Client {
|
|||
return;
|
||||
}
|
||||
};
|
||||
for name in repos.into_iter().filter_map(|r| r.name) {
|
||||
// An archived repo is a destroyed agent's, and refuses every write, so
|
||||
// converging its rule would fail on each pass for as long as it stays.
|
||||
for name in repos
|
||||
.into_iter()
|
||||
.filter(|r| !r.archived.unwrap_or(false))
|
||||
.filter_map(|r| r.name)
|
||||
{
|
||||
let rule = seen(
|
||||
self.api
|
||||
.call(|api| {
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ use std::sync::{Arc, Mutex};
|
|||
use anyhow::{Context, Result};
|
||||
|
||||
mod agent_archetype;
|
||||
mod agent_config_repo;
|
||||
mod agent_control;
|
||||
mod agent_custom;
|
||||
mod agent_entity;
|
||||
|
|
|
|||
|
|
@ -219,6 +219,7 @@ pub(crate) async fn set_agent_state(
|
|||
agents_md_initial::clear_on_destroy(&state, &hive, &agent).await;
|
||||
egress_allow::clear_on_destroy(&state, &hive, &agent).await;
|
||||
egress_deny::clear_on_destroy(&state, &hive, &agent).await;
|
||||
crate::agent_config_repo::archive_on_destroy(&state, &hive, &agent).await;
|
||||
}
|
||||
Ok(Json(render(&declaration)))
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue