diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 5594654e..609fd7ab 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -105,6 +105,12 @@ enum SwarmNodeKind { /// its hive's shared queue credential, so the document cannot be rendered /// without knowing which hive the agent belongs to. MintAgentIdentity { hive: String, agent: String }, + /// Declare `agent` on `hive` as `Paused` in the swarm's wanted-state + /// store, so a freshly created agent does not start driving turns the + /// moment it's deployed — the operator has to explicitly flip it to `Up`. + /// Carries the hive for the same reason `TriggerDeploy`/`MintAgentIdentity` + /// do: the wanted-state bucket is keyed per hive. + SetAgentWanted { hive: String, agent: String }, /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy /// subject. The one node kind whose effect leaves this host. /// @@ -124,6 +130,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), + SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(), SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), } } @@ -144,7 +151,8 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { serde_json::json!({ "agent": agent }) } SwarmNodeKind::TriggerDeploy { hive, agent } - | SwarmNodeKind::MintAgentIdentity { hive, agent } => { + | SwarmNodeKind::MintAgentIdentity { hive, agent } + | SwarmNodeKind::SetAgentWanted { hive, agent } => { serde_json::json!({ "agent": agent, "hive": hive }) } } @@ -186,8 +194,22 @@ struct WorkerDeps { /// a per-node chance to read one. `None` on a host the operator has not /// given an authority — see `agent_identity::Authority::from_env`. agent_ca: Option>, + /// The wanted-state writer, for nodes that declare what a hive should + /// converge an agent to. Shares the status reader's queue connection — + /// see `wanted_writer`. `None` exactly when no swarm queue is configured + /// on this host, same as `queue`. + wanted: Option>, } +/// What [`SwarmNodeKind::SetAgentWanted`] declares a brand-new agent to be. +/// +/// A named constant rather than the literal inline, because this *is* the +/// decision the node exists to carry — writing a declaration at all is the +/// mechanism, `Paused` is the policy — and a test can pin a name where it +/// cannot pin a value written into a KV bucket no unit test has a server for. +const NEW_AGENT_WANTED_STATE: swarm_queue_client::wanted::AgentState = + swarm_queue_client::wanted::AgentState::Paused; + /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` /// variant turns into a real effect. @@ -286,6 +308,14 @@ async fn run_swarm_node( } } }, + SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted { + None => Outcome::Failed( + "no swarm queue is configured on this host, so no wanted-state \ + declaration can be written" + .to_owned(), + ), + Some(writer) => declare_new_agent(&writer, &hive, &agent).await, + }, SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue { None => Outcome::Failed( "no swarm queue is configured on this host, so no hive can be told to deploy" @@ -297,6 +327,25 @@ async fn run_swarm_node( (builder, outcome) } +/// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`]. +/// +/// A named function beside [`publish_deploy`] rather than the two lines +/// inline, for the same reason that one is: `run_swarm_node`'s match is a +/// per-variant index, and every arm that grows a body past a call pushes the +/// next reader further from the variant they came to read. +async fn declare_new_agent( + writer: &wanted::WantedWriter, + hive: &str, + agent: &str, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + match writer.set(hive, agent, NEW_AGENT_WANTED_STATE).await { + Ok(_) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + } +} + /// Publish one deploy request and report whether it left this process. /// /// The flush is not belt-and-braces: `publish` hands the message to the @@ -1316,6 +1365,15 @@ fn declare_agent_job( agent: agent.to_owned(), }) .after_ok(create_identity); + // Declared before the deploy trigger so the pause is visible in the + // wanted-state store before the hive brings the container up — see the + // node's own doc comment for why "before", not just "eventually". Needs + // no parent of its own: the declaration is written from the hive/agent + // names known at request time, same as `create_forge_user`. + let set_wanted = b.node(SwarmNodeKind::SetAgentWanted { + hive: hive.to_owned(), + agent: agent.to_owned(), + }); // Last, and specifically after the config repo is seeded: the hive // deploys by reading that repo, so a deploy asked for any earlier would // find nothing to build. This is the edge that makes creating an agent at @@ -1327,13 +1385,21 @@ fn declare_agent_job( // overtake the mint — but a host with no authority configured must still // create agents exactly as it does today. `after_ok` there would turn an // unconfigured option into an agent nobody runs. + // + // `after_ok` on the declaration, though: an agent deployed without its + // pause landing first is the race this node exists to close, and the + // only way the declaration fails without a live queue is a host that has + // no queue at all — on which `TriggerDeploy` has nothing to publish to + // either, so the strict edge cancels a node that could not have + // succeeded. let _trigger_deploy = b .node(SwarmNodeKind::TriggerDeploy { hive: hive.to_owned(), agent: agent.to_owned(), }) .after_ok(init_config) - .after_any(mint_identity); + .after_any(mint_identity) + .after_ok(set_wanted); vec![create_identity.guid()] } @@ -1664,6 +1730,7 @@ async fn main() -> Result<()> { forge: forge_client.clone(), queue: status.as_ref().map(|s| s.queue_client()), agent_ca: load_agent_authority(), + wanted: wanted_writer(status.as_ref()), }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( @@ -1882,8 +1949,9 @@ mod tests { }]), links: std::sync::Arc::new(Vec::new()), status: None, - // No queue, for the same reason as `status`: these tests drive - // agent creation, which publishes no declaration. + // No queue, for the same reason as `status`: these tests build + // the creation graph and read its shape, and never run the node + // that would write a declaration. wanted: None, agent_status: None, jobq: std::sync::Arc::clone(&sched), @@ -2031,6 +2099,7 @@ mod tests { forge: None, queue: None, agent_ca: None, + wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2084,6 +2153,7 @@ mod tests { forge: None, queue: None, agent_ca: None, + wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2137,6 +2207,7 @@ mod tests { forge: None, queue: None, agent_ca: None, + wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2159,6 +2230,137 @@ mod tests { ); } + /// Fourth sibling, same shape and the same caveat once more: with no + /// queue configured this reaches only the graceful-absence-is-failure + /// branch. Writing the declaration for real needs a `JetStream` server, + /// which no test in this crate has — `wanted::apply`'s own unit tests + /// cover what gets written, and [`a_new_agent_is_declared_paused`] below + /// pins the state this node asks for. + #[tokio::test] + async fn set_agent_wanted_node_runs_end_to_end_and_fails_without_a_queue() { + let mut sched = hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ); + let id = sched + .append( + SwarmNodeKind::SetAgentWanted { + hive: "pr1ma".to_owned(), + agent: "atlas".to_owned(), + }, + Vec::new(), + None, + ) + .expect("insert"); + let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); + + let deps = WorkerDeps { + auth: None, + forge: None, + queue: None, + agent_ca: None, + wanted: None, + }; + let runner = + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { + run_swarm_node(id, kind, builder, deps) + }) + .expect("the node just inserted is runnable"); + runner + .await + .1 + .expect("no growth declared, nothing to reject"); + + let guard = sched.lock().unwrap(); + let node = guard.graph().node(id).expect("node still present"); + assert_eq!(node.state, hive_jobq::State::Failed); + assert!( + node.error.as_deref().unwrap_or_default().contains("queue"), + "expected a no-queue-configured error, got {:?}", + node.error + ); + } + + /// The whole point of the node: a freshly created agent is declared + /// `Paused`, so it does not start driving turns the moment its hive + /// brings it up. Any other state here is the bug this fixes. + #[test] + fn a_new_agent_is_declared_paused() { + assert_eq!( + crate::NEW_AGENT_WANTED_STATE, + swarm_queue_client::wanted::AgentState::Paused + ); + } + + /// Same reasoning as `a_mint_node_renders_both_the_agent_and_the_hive`: + /// this variant carries a hive too, and has to have joined `data`'s + /// two-field or-pattern rather than the agent-only one. + #[test] + fn a_set_wanted_node_renders_both_the_agent_and_the_hive() { + use hive_jobq_wire::WireNode as _; + + let kind = SwarmNodeKind::SetAgentWanted { + hive: "pr1ma".to_owned(), + agent: "atlas".to_owned(), + }; + assert_eq!(kind.label(), "set_agent_wanted"); + let data = kind.data(1); + assert_eq!(data["agent"], "atlas"); + assert_eq!(data["hive"], "pr1ma"); + } + + /// The ordering the fix depends on: the pause has to be in the + /// wanted-state store *before* the hive is told to deploy, or a freshly + /// created agent runs undeclared for the window between the two. + /// + /// Asserted on the graph `create_agent` builds, same as the mint edge's + /// test above and for the same reason — the edge is one line in a + /// builder closure whose absence changes nothing observable until a real + /// agent boots and starts taking turns nobody asked for. + #[tokio::test] + async fn the_deploy_waits_for_the_pause_to_be_declared() { + use hive_jobq_wire::WireNode as _; + + let (state, sched) = state_with_roster(); + let _queued = super::create_agent( + axum::extract::State(state), + axum::Json(super::CreateAgentRequest { + name: "atlas".to_owned(), + hive: "pr1ma".to_owned(), + }), + ) + .await + .expect("a hive in the roster must be accepted"); + + let guard = sched + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = guard.graph(); + let id_of = |label: &str| { + graph + .nodes() + .find(|n| n.payload.label() == label) + .unwrap_or_else(|| panic!("the graph holds a {label} node")) + .id + }; + let set_wanted = id_of("set_agent_wanted"); + let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); + + let when = deploy + .deps + .iter() + .find_map(|d| match d { + hive_jobq::Dep::Node { id, when } if *id == set_wanted => Some(*when), + _ => None, + }) + .expect("the deploy waits for the declaration"); + assert!( + !when.accepts(hive_jobq::TerminalState::Failed), + "a declaration that never landed must not be deployed past; this \ + edge has to be `after_ok`, not `after_any`" + ); + } + /// The node carries the hive, and the viewer has to see it. The `data` /// match is an or-pattern on purpose (see its own comment), and this is /// the assertion that the new variant joined the two-field arm rather