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

swarm-controller: refuse a state declaration create_agent would refuse

Browse source 
PUT /api/hives/{hive}/agents/{agent}/state wrote any identifier into any
hive's wanted state, and the hive first-deploys a declared agent it has
no container for. That skipped create_agent's checks on the name.

A declaration that places the agent (up, paused, offline) is now refused
with 400 for a new name that breaks a naming rule, and with 409 for a
name the swarm has placed on another hive. Both reuse create_agent's
helpers (broken_name_rules/name_verdict, placements_elsewhere), under the
same gate, and an unreadable wanted state on another hive refuses with
503/500 as creation does. `destroyed` places nothing and is not checked,
so the revocation path is unchanged.

An agent with no declaration at all is still accepted: swarm-ui declares
state for agents that predate swarm-level creation, which is how the swarm
adopts them. Whether to refuse such names instead is left open on #4804.

Refs #4804
This commit is contained in:
atlas 2026-09-29 16:17:42 +02:00 • committed by mara
commit 8b892f508b
1 changed files with 193 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -687,8 +687,10 @@ struct AppState {
/// forge-backed field here uses.
forge: Option<Arc<forge::Client>>,
/// Held by `create_agent` from reading where a name is placed until its
/// graph is queued, so two creations of one name cannot both find it
/// unplaced. `tokio`'s mutex, unlike `jobq`'s: the read awaits the queue.
/// graph is queued, and by `set_agent_state` from that read until its
/// declaration is written, so two placements of one name cannot both
/// find it unplaced. `tokio`'s mutex, unlike `jobq`'s: the read awaits
/// the queue.
create_gate: Arc<tokio::sync::Mutex<()>>,
}
@ -1033,9 +1035,10 @@ fn declaration_target(
request_body = SetAgentStateRequest,
responses(
(status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>),
(status = 400, description = "a name is not an identifier, the hive is not in this swarm, or the state is unknown (problem+json)", body = String),
(status = 400, description = "a name is not an identifier, the hive is not in this swarm, the state is unknown, or `agent` is new and reserved or refused by the forge (problem+json)", body = String),
(status = 409, description = "the swarm has already placed `agent` on a different hive (problem+json)", body = String),
(status = 503, description = "no swarm queue is wired up, or it is not connected (problem+json)", body = String),
(status = 500, description = "the declaration could not be published (problem+json)", body = String),
(status = 500, description = "the declaration could not be published, or another hive's wanted state could not be read (problem+json)", body = String),
),
tag = "agents"
)]
@ -1050,6 +1053,24 @@ async fn set_agent_state(
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string();
// Refuses only what `create_agent` would refuse for the same placement.
// An agent with no declaration on this hive is still accepted: declaring
// it is how the swarm adopts an agent that predates swarm-level creation.
// Deliberate, pending the operator's decision on whether a name the swarm
// never placed should be refused here instead.
let _gate = state.create_gate.lock().await;
if places_agent(req.state) {
refuse_placement(
&agent,
&hive,
hive_types::reserved_names_raw().as_deref(),
|| in_roster(state.auth.as_deref(), &agent),
|| queued_placements_now(&state),
|| declarations_elsewhere(&state, &agent, &hive),
)
.await?;
}
let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| {
tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
error_problem(wanted_error_status(&e), &format!("{e:#}"))
@ -1066,6 +1087,62 @@ async fn set_agent_state(
Ok(Json(render(&declaration)))
}
/// Whether a declaration in `state` places the agent on its hive.
///
/// Exhaustive for the reason [`revokes_queue_credential`] is. A `Destroyed`
/// agent is gone from that hive, and the hive deploys nothing for it.
fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool {
use swarm_queue_client::wanted::AgentState;
match state {
AgentState::Up | AgentState::Offline | AgentState::Paused => true,
AgentState::Destroyed => false,
}
}
/// Refuse a placement of `agent` on `hive` that `create_agent` refuses: 400
/// for a new name that breaks a naming rule, 409 for a name the swarm has
/// placed on another hive.
///
/// The roster is read only for a name that breaks a rule, as in
/// `create_agent`, and an unreadable one warns rather than refuses (see
/// [`name_verdict`]).
async fn refuse_placement<R, D>(
agent: &str,
hive: &str,
reserved: Option<&str>,
read_roster: impl FnOnce() -> R,
snapshot_queued: impl FnOnce() -> Vec<(String, String)>,
read_declared: impl FnOnce() -> D,
) -> Result<(), problem_details::ProblemDetails>
where
R: std::future::Future<Output = Result<bool, String>>,
D: std::future::Future<
Output = Result<
Vec<(String, swarm_queue_client::wanted::HiveWanted)>,
problem_details::ProblemDetails,
>,
>,
{
// Both calls log what they push here; this response has nowhere to
// carry warnings.
let mut warnings = Vec::new();
let broken = broken_name_rules(agent, reserved, &mut warnings);
if !broken.is_empty() {
name_verdict(agent, &broken, read_roster().await, &mut warnings)
.map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?;
}
let elsewhere = placements_elsewhere(agent, hive, snapshot_queued, read_declared).await?;
if !elsewhere.is_empty() {
let detail = format!(
"agent {agent:?} is placed on hive {}, not {hive:?} — agent names are unique across \
the swarm; declare its state there, or destroy it there first to move it",
elsewhere.join(", ")
);
return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
}
Ok(())
}
/// Whether declaring an agent into `state` ends its queue credential's life.
///
/// Exhaustive on purpose, like every other match on
@ -1437,15 +1514,7 @@ async fn create_agent(
let elsewhere = placements_elsewhere(
&agent,
&hive,
|| {
queued_placements(
state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph(),
)
},
|| queued_placements_now(&state),
|| declarations_elsewhere(&state, &agent, &hive),
)
.await?;
@ -1525,7 +1594,7 @@ async fn declarations_elsewhere(
Err(e) => {
let detail = format!(
"cannot tell whether {agent:?} already exists on hive {:?}, so it was not \
created: {e:#}",
placed: {e:#}",
other.name
);
return Err(error_problem(wanted_error_status(&e), &detail));
@ -1535,6 +1604,17 @@ async fn declarations_elsewhere(
Ok(declared)
}
/// [`queued_placements`] of the swarm's own queue, as it is now.
fn queued_placements_now(state: &AppState) -> Vec<(String, String)> {
queued_placements(
state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph(),
)
}
/// `(hive, agent)` of every `SetAgentWanted` node not yet settled: a creation
/// accepted but not yet visible in the wanted state it is about to write.
fn queued_placements(
@ -1552,21 +1632,20 @@ fn queued_placements(
/// The hives other than `hive` the swarm has placed `agent` on, sorted.
///
/// A placement is a declaration in any state but `Destroyed`, or a queued
/// one. A `Destroyed` agent is gone from that hive, so creating it on another
/// moves it.
/// A placement is a declaration in a state that [places][places_agent] the
/// agent, or a queued one. A `Destroyed` agent is gone from that hive, so
/// placing it on another moves it.
fn placed_elsewhere(
agent: &str,
hive: &str,
declared: &[(String, swarm_queue_client::wanted::HiveWanted)],
queued: &[(String, String)],
) -> Vec<String> {
use swarm_queue_client::wanted::AgentState;
let declared = declared.iter().filter(|(_, declaration)| {
declaration
.agents
.get(agent)
.is_some_and(|wanted| wanted.state != AgentState::Destroyed)
.is_some_and(|wanted| places_agent(wanted.state))
});
let queued = queued.iter().filter(|(_, queued)| queued == agent);
let hives: std::collections::BTreeSet<&str> = declared
@ -1595,7 +1674,7 @@ fn broken_name_rules(
None => {
tracing::error!(
var = hive_types::RESERVED_NAMES_ENV,
"create_agent: reserved-name check could not run — variable not set"
"reserved-name check could not run — variable not set"
);
warnings.push(format!(
"the reserved-name check did not run: {} is unset, so {agent:?} was accepted \
@ -1654,7 +1733,7 @@ fn name_verdict(
match existing {
Ok(false) => Err(format!("{}; choose another name", broken.join("; "))),
Ok(true) => {
tracing::warn!(agent, "create_agent: existing agent breaks a naming rule");
tracing::warn!(agent, "existing agent breaks a naming rule");
warnings.extend(
broken
.iter()
@ -1663,11 +1742,7 @@ fn name_verdict(
Ok(())
}
Err(why) => {
tracing::warn!(
agent,
why,
"create_agent: naming rule broken, roster unreadable"
);
tracing::warn!(agent, why, "naming rule broken, roster unreadable");
warnings.extend(broken.iter().map(|rule| {
format!(
"{rule} — not refused, because whether {agent:?} already exists could not \
@ -3176,6 +3251,98 @@ mod tests {
assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty());
}
/// `refuse_placement` of `agent` on `pr1ma`, with `admin` reserved,
/// `agent` not in the roster and `declared` as every hive's wanted state.
async fn refuse_on_pr1ma(
agent: &str,
declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>,
) -> Result<(), problem_details::ProblemDetails> {
super::refuse_placement(
agent,
"pr1ma",
Some("admin"),
|| std::future::ready(Ok(false)),
Vec::new,
|| std::future::ready(Ok(declared)),
)
.await
}
#[tokio::test]
async fn placing_a_new_reserved_name_is_refused() {
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("a new reserved name must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
}
#[tokio::test]
async fn placing_a_name_placed_on_another_hive_is_refused() {
use swarm_queue_client::wanted::AgentState;
let err = refuse_on_pr1ma("atlas", vec![declaring("sec0nd", "atlas", AgentState::Up)])
.await
.expect_err("a name placed on another hive must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::CONFLICT),
"{err:?}"
);
assert!(format!("{err:?}").contains("sec0nd"), "{err:?}");
}
/// An agent that runs on `pr1ma` but was never declared anywhere, and is
/// not in the roster, is one the swarm adopts by declaring it.
#[tokio::test]
async fn adopting_an_undeclared_agent_on_its_own_hive_is_accepted() {
refuse_on_pr1ma("legacy", Vec::new())
.await
.expect("an undeclared agent must still be adoptable");
}
#[tokio::test]
async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() {
use swarm_queue_client::wanted::AgentState;
refuse_on_pr1ma("atlas", vec![declaring("pr1ma", "atlas", AgentState::Up)])
.await
.expect("an agent's own hive must accept a new state for it");
}
/// Through the handler: another hive whose wanted state cannot be read
/// refuses the declaration before it is written, as it refuses creation.
#[tokio::test]
async fn set_agent_state_refuses_when_another_hive_is_unreadable() {
let (state, _sched) = state_with_two_hives();
let state = super::AppState {
wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
disconnected_client().await,
))),
..state
};
let err = super::set_agent_state(
axum::extract::State(state),
axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())),
axum::Json(SetAgentStateRequest {
state: swarm_queue_client::wanted::AgentState::Up,
}),
)
.await
.expect_err("an unreadable placement must refuse");
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
assert!(
format!("{err:?}").contains("was not placed"),
"the refusal must come from the placement check, got: {err:?}"
);
}
/// Forgejo v16.0.5's reserved usernames our charset can spell, plus the
/// bare `-` — the forge section of `nix/reserved-names.nix`. Read from
/// the list nix exports, so dropping one from the file reds this.