swarm-controller: check only a first declaration, and never a stop's name

The state route ran the name rules and the placed-elsewhere check on
every up/offline/paused declaration. An agent already declared on the
hive whose name breaks a rule, and which is not in the roster, could then
only be destroyed from the swarm.

Now an agent this hive already declares in a placing state passes both
checks. For a first declaration the placed-elsewhere check still applies
to up, offline and paused alike, and the name rules to up only: offline
and paused are how an operator stops an agent. The route reads this hive's
declaration to tell, and refuses with 503/500 when it cannot.

Refs #4804

swarm-controller/src/main.rs

@@ -1053,16 +1053,25 @@ 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.
+    // Only a first declaration of the name on this hive is checked, and only
-    // An agent with no declaration on this hive is still accepted: declaring
+    // against what `create_agent` refuses; `placement_checks` says which
-    // it is how the swarm adopts an agent that predates swarm-level creation.
+    // checks apply. An undeclared name that passes them is accepted: that 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) {
+        let here = writer.view(&hive).await.map_err(|e| {
+            let detail = format!(
+                "cannot tell whether {agent:?} is already declared on hive {hive:?}, so it was \
+                 not placed: {e:#}"
+            );
+            error_problem(wanted_error_status(&e), &detail)
+        })?;
         refuse_placement(
             &agent,
             &hive,
+            placement_checks(&agent, req.state, here.as_ref()),
             hive_types::reserved_names_raw().as_deref(),
             || in_roster(state.auth.as_deref(), &agent),
             || queued_placements_now(&state),
@@ -1099,9 +1108,41 @@ fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool {
     }
 }
 
+/// Which of [`refuse_placement`]'s checks a declaration must pass.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PlacementChecks {
+    Neither,
+    Elsewhere,
+    NameAndElsewhere,
+}
+
+/// The checks declaring `agent` into `state` must pass, given `here`, the
+/// hive's current declaration.
+///
+/// An agent `here` already declares in a placing state is placed on this
+/// hive, so it passes both. Otherwise the name rules apply to `Up` only:
+/// `Offline` and `Paused` are how an operator stops an agent, and that is
+/// never refused over its name.
+fn placement_checks(
+    agent: &str,
+    state: swarm_queue_client::wanted::AgentState,
+    here: Option<&swarm_queue_client::wanted::HiveWanted>,
+) -> PlacementChecks {
+    use swarm_queue_client::wanted::AgentState;
+    let declared_here = here
+        .and_then(|declaration| declaration.agents.get(agent))
+        .is_some_and(|wanted| places_agent(wanted.state));
+    match state {
+        _ if declared_here => PlacementChecks::Neither,
+        AgentState::Up => PlacementChecks::NameAndElsewhere,
+        AgentState::Offline | AgentState::Paused => PlacementChecks::Elsewhere,
+        AgentState::Destroyed => PlacementChecks::Neither,
+    }
+}
+
 /// 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.
+/// placed on another hive. `checks` says which of the two to run.
 ///
 /// 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
@@ -1109,6 +1150,7 @@ fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool {
 async fn refuse_placement<R, D>(
     agent: &str,
     hive: &str,
+    checks: PlacementChecks,
     reserved: Option<&str>,
     read_roster: impl FnOnce() -> R,
     snapshot_queued: impl FnOnce() -> Vec<(String, String)>,
@@ -1123,6 +1165,10 @@ where
             >,
         >,
 {
+    if checks == PlacementChecks::Neither {
+        return Ok(());
+    }
+    if checks == PlacementChecks::NameAndElsewhere {
         // Both calls log what they push here; this response has nowhere to
         // carry warnings.
         let mut warnings = Vec::new();
@@ -1131,6 +1177,7 @@ where
             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!(
@@ -3251,15 +3298,22 @@ mod tests {
         assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty());
     }
 
-    /// `refuse_placement` of `agent` on `pr1ma`, with `admin` reserved,
+    /// The handler's two calls for declaring `agent` into `state` on `pr1ma`,
-    /// `agent` not in the roster and `declared` as every hive's wanted state.
+    /// with `admin` reserved, `agent` not in the roster, and `declared` as
+    /// every hive's wanted state, `pr1ma`'s included.
     async fn refuse_on_pr1ma(
         agent: &str,
+        state: swarm_queue_client::wanted::AgentState,
         declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>,
     ) -> Result<(), problem_details::ProblemDetails> {
+        let here = declared
+            .iter()
+            .find(|(hive, _)| hive == "pr1ma")
+            .map(|(_, declaration)| declaration.clone());
         super::refuse_placement(
             agent,
             "pr1ma",
+            super::placement_checks(agent, state, here.as_ref()),
             Some("admin"),
             || std::future::ready(Ok(false)),
             Vec::new,
@@ -3268,9 +3322,19 @@ mod tests {
         .await
     }
 
+    const PLACING: [swarm_queue_client::wanted::AgentState; 3] = [
+        swarm_queue_client::wanted::AgentState::Up,
+        swarm_queue_client::wanted::AgentState::Offline,
+        swarm_queue_client::wanted::AgentState::Paused,
+    ];
+
     #[tokio::test]
     async fn placing_a_new_reserved_name_is_refused() {
-        let err = refuse_on_pr1ma("admin", Vec::new())
+        let err = refuse_on_pr1ma(
+            "admin",
+            swarm_queue_client::wanted::AgentState::Up,
+            Vec::new(),
+        )
         .await
         .expect_err("a new reserved name must be refused");
         assert_eq!(
@@ -3280,41 +3344,101 @@ mod tests {
         );
     }
 
+    /// A destroyed agent is gone from its hive, so declaring it again is a
+    /// first declaration and its name is checked.
     #[tokio::test]
-    async fn placing_a_name_placed_on_another_hive_is_refused() {
+    async fn a_destroyed_declaration_here_does_not_exempt_the_name() {
         use swarm_queue_client::wanted::AgentState;
-        let err = refuse_on_pr1ma("atlas", vec![declaring("sec0nd", "atlas", AgentState::Up)])
+        let err = refuse_on_pr1ma(
+            "admin",
+            AgentState::Up,
+            vec![declaring("pr1ma", "admin", AgentState::Destroyed)],
+        )
+        .await
+        .expect_err("re-placing a destroyed reserved name must be refused");
+        assert_eq!(
+            err.status,
+            Some(axum::http::StatusCode::BAD_REQUEST),
+            "{err:?}"
+        );
+    }
+
+    /// Stopping is never refused over a name, even on a first declaration.
+    #[tokio::test]
+    async fn stopping_a_new_rule_breaking_name_is_accepted() {
+        use swarm_queue_client::wanted::AgentState;
+        for state in [AgentState::Offline, AgentState::Paused] {
+            refuse_on_pr1ma("admin", state, Vec::new())
+                .await
+                .unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
+        }
+    }
+
+    /// An agent already declared here keeps a rule-breaking name through
+    /// every state change, though the roster does not know it.
+    #[tokio::test]
+    async fn an_agent_declared_here_is_not_refused_over_its_name() {
+        use swarm_queue_client::wanted::AgentState;
+        for state in PLACING {
+            refuse_on_pr1ma(
+                "admin",
+                state,
+                vec![declaring("pr1ma", "admin", AgentState::Up)],
+            )
+            .await
+            .unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
+        }
+    }
+
+    #[tokio::test]
+    async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() {
+        use swarm_queue_client::wanted::AgentState;
+        for state in PLACING {
+            let err = refuse_on_pr1ma(
+                "atlas",
+                state,
+                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:?}"
+                "{state:?}: {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())
+        for state in PLACING {
+            refuse_on_pr1ma("legacy", state, Vec::new())
                 .await
-            .expect("an undeclared agent must still be adoptable");
+                .unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
+        }
     }
 
     #[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)])
+        for state in PLACING {
+            refuse_on_pr1ma(
+                "atlas",
+                state,
+                vec![declaring("pr1ma", "atlas", AgentState::Up)],
+            )
             .await
-            .expect("an agent's own hive must accept a new state for it");
+            .unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
+        }
     }
 
-    /// Through the handler: another hive whose wanted state cannot be read
+    /// Through the handler: a wanted state that cannot be read refuses the
-    /// refuses the declaration before it is written, as it refuses creation.
+    /// declaration before it is written, as it refuses creation.
     #[tokio::test]
-    async fn set_agent_state_refuses_when_another_hive_is_unreadable() {
+    async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() {
         let (state, _sched) = state_with_two_hives();
         let state = super::AppState {
             wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(