//! The swarm-level job graph: its node and resource kinds, the worker that //! runs claimed nodes, and the `queue_*` fns that insert them. use std::sync::{Arc, Mutex}; use anyhow::Result; use swarm_authelia_bridge_sock::BridgeResponse; use crate::routes::AppState; use crate::routes::agents::{claiming_hives, read_declarations}; use crate::startup::PUBLIC_URL_ENV; use crate::{ agent_identity, agent_renewal, auth, builder, config_pr, forge, matrix_account, wanted, }; /// Node payload for the swarm-level job graph. Named `Swarm*` rather than /// the bare `NodeKind`/`Resource` `hive-c0re::job_queue::model` already /// uses, so a grep for either doesn't land on both crates. /// /// `CreateIdentity` was the first real variant — "the minimal shape for /// agent creation is creating the identity and wiring that in" (the design /// thread's own framing for why that landed before the forge-node work /// here). `CreateRepo`/`AddRepoMember`/`InitAgentConfigRepo` are three /// separate nodes rather than one combined "create the repo" step so each /// is independently retryable/observable in the job graph, same as every /// other multi-step provisioning flow in this codebase (`hive-c0re`'s own /// `NodeKind` never folds unrelated forge calls into one node either). #[derive(Clone, Debug)] pub(crate) enum SwarmNodeKind { /// Ensure `agent` exists as an authelia subject at the swarm level. CreateIdentity { agent: String }, /// Create the agent's repo in `forge::CONFIG_ORG` with the operator /// merge gate on its default branch. See `forge::Client::create_repo`. CreateRepo { agent: String }, /// Ensure `agent` exists as a Forgejo user account. Independent of /// `CreateIdentity`/`CreateRepo` (a forge user needs neither an /// authelia subject nor an existing repo) but a prerequisite for /// `AddRepoMember`, which fails with "user does not exist" against an /// account this node hasn't created yet. See /// `forge::Client::ensure_agent_user`. CreateForgeUser { agent: String }, /// Add `agent` as a write collaborator on its own repo. See /// `forge::Client::add_repo_member`. AddRepoMember { agent: String }, /// Add `agent` as a read collaborator on `internal/docs`. See /// `forge::docs_grant`. /// /// Carries no hive: the grant is on a swarm-wide repo. GrantDocsRead { agent: String }, /// Seed the agent's repo with `agent.nix` + `flake.nix`. See /// `forge::Client::seed_agent_config`. /// /// Deliberately carries no hive: seeding a config repo is the same /// work whichever hive the agent is bound for, and an agent's config /// states nothing about where it runs. The hive is an address the /// swarm routes a deploy message to — it belongs on the node that /// sends that message, not on this one. InitAgentConfigRepo { agent: String }, /// Mint `agent`'s own client certificate for the swarm secret store, /// publish it there, grant it, and log in with it. See /// `agent_identity::mint_and_verify` — including why this node does not /// report success on a write. /// /// Carries no hive: neither the certificate, the queue credential nor /// the agent's policy names one, so an agent keeps one store identity /// whichever hive it runs on. MintAgentIdentity { agent: String }, /// Make sure `agent` holds a live forge access token in the swarm secret /// store, minting one with the forge's admin API when it does not. See /// `forge::agent_token` — including why a rotation is a delete then a /// create. /// /// Carries no hive: the token's store path has no hive segment, and the /// agent pulls it from wherever it runs. MintAgentForgeToken { agent: String }, /// Make sure `agent` holds a live token for its own account on the swarm's /// homeserver, creating the account with the swarm's appservice token /// when it does not. See `matrix_account::agent_token` — including why a /// stored token is read back before anything is minted. /// /// Carries no hive: the path it writes has no hive segment, so an agent /// that moves between hives keeps one matrix identity. MintAgentMatrixAccount { agent: String }, /// Make sure `hive`'s sender account (`@hive-:`) holds a live token /// in the swarm store, where that hive's matrix sweep reads it. See /// `matrix_account::hive_sender`. MintHiveSenderToken { hive: String }, /// Make sure `hive`'s read-only forge account (`hive-`) exists, is in /// the `hives` team, and holds a live token in the swarm store, where /// that hive's hive-c0re reads it. See `forge::hive_token`. MintHiveForgeToken { hive: String }, /// Make sure the review classifier's forge account exists and holds a /// live token in the swarm store, where this daemon reads it to post the /// classifier's reviews. See `forge::classifier_account`. MintClassifierForgeToken, /// Make sure the swarm builder's read-only forge account exists, is in /// the `hives` team, and holds a live token in the swarm store, where the /// builder reads it to fetch config repos. See `forge::builder_account`. MintBuilderForgeToken, /// Make sure the CI metrics poller's read-only forge account exists, is in /// its team in every organisation, and holds a live token in the swarm /// store, where the poller reads it to list CI runs. See /// `forge::ci_metrics_account`. MintCiMetricsForgeToken, /// Build `agent`'s config at `rev`, the head of its config PR `pr`, on /// the swarm builder, and write the result as a commit status on `rev`. /// See `builder`. /// /// Carries no hive: the check builds on the builder's own agent base, /// not on any hive. BuildAgent { agent: String, rev: String, pr: u64 }, /// Re-mint `agent`'s queue secret if it is still stored and old enough. /// See `agent_renewal`. /// /// Carries no hive: the secret's store path has none. RenewAgentQueueCredential { 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 because 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. /// /// Carries the hive, unlike every variant above — this is the node /// `InitAgentConfigRepo`'s doc points at when it says the hive belongs /// on the node that sends the deploy message. It names the subject the /// message goes to, one per hive, so no other hive is woken by it. TriggerDeploy { hive: String, agent: String, /// The config commit to deploy, from a merged config PR. `None` from /// agent creation, where the hive deploys what it seeds. rev: Option, }, } impl hive_jobq_wire::WireNode for SwarmNodeKind { fn label(&self) -> String { match self { SwarmNodeKind::CreateIdentity { .. } => "create_identity".to_owned(), SwarmNodeKind::CreateRepo { .. } => "create_repo".to_owned(), SwarmNodeKind::CreateForgeUser { .. } => "create_forge_user".to_owned(), SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::GrantDocsRead { .. } => "grant_docs_read".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), SwarmNodeKind::MintAgentForgeToken { .. } => "mint_agent_forge_token".to_owned(), SwarmNodeKind::MintAgentMatrixAccount { .. } => "mint_agent_matrix_account".to_owned(), SwarmNodeKind::MintHiveSenderToken { .. } => "mint_hive_sender_token".to_owned(), SwarmNodeKind::MintHiveForgeToken { .. } => "mint_hive_forge_token".to_owned(), SwarmNodeKind::MintClassifierForgeToken => "mint_classifier_forge_token".to_owned(), SwarmNodeKind::MintBuilderForgeToken => "mint_builder_forge_token".to_owned(), SwarmNodeKind::MintCiMetricsForgeToken => "mint_ci_metrics_forge_token".to_owned(), SwarmNodeKind::BuildAgent { .. } => "build_agent".to_owned(), SwarmNodeKind::RenewAgentQueueCredential { .. } => { "renew_agent_queue_credential".to_owned() } SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(), SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), } } fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { // Spelled out as an or-pattern rather than a catch-all on purpose: a // new variant then fails to compile here instead of silently // rendering as an agent name. That gate has now fired once — // `TriggerDeploy` is the first node that is about an agent *and a // hive*, and a catch-all would have dropped the hive from the // viewer without anyone noticing. match self { SwarmNodeKind::CreateIdentity { agent } | SwarmNodeKind::CreateRepo { agent } | SwarmNodeKind::CreateForgeUser { agent } | SwarmNodeKind::AddRepoMember { agent } | SwarmNodeKind::GrantDocsRead { agent } | SwarmNodeKind::InitAgentConfigRepo { agent } | SwarmNodeKind::MintAgentIdentity { agent } | SwarmNodeKind::MintAgentForgeToken { agent } | SwarmNodeKind::MintAgentMatrixAccount { agent } | SwarmNodeKind::RenewAgentQueueCredential { agent } => { serde_json::json!({ "agent": agent }) } SwarmNodeKind::MintHiveSenderToken { hive } | SwarmNodeKind::MintHiveForgeToken { hive } => serde_json::json!({ "hive": hive }), SwarmNodeKind::MintClassifierForgeToken | SwarmNodeKind::MintBuilderForgeToken | SwarmNodeKind::MintCiMetricsForgeToken => serde_json::json!({}), SwarmNodeKind::BuildAgent { agent, rev, pr } => { serde_json::json!({ "agent": agent, "rev": rev, "pr": pr }) } SwarmNodeKind::TriggerDeploy { hive, agent, rev } => { serde_json::json!({ "agent": agent, "hive": hive, "rev": rev }) } SwarmNodeKind::SetAgentWanted { hive, agent } => { serde_json::json!({ "agent": agent, "hive": hive }) } } } } /// A named counting semaphore a node acquires before it starts. Every name /// has capacity 1: the scheduler's default, which no caller overrides. #[derive(Clone, Debug, PartialEq, Eq, Hash)] pub(crate) enum SwarmResourceKind { /// The swarm builder: one `BuildAgent` node runs at a time, and the rest /// wait pending, where a newer push can replace them. Builder, } impl hive_jobq_wire::WireResource for SwarmResourceKind { fn name(&self) -> String { match self { SwarmResourceKind::Builder => "builder".to_owned(), } } } /// A dependency this daemon may only gain after it starts serving: empty /// until then, filled once. Readers take what is there at the moment they ask. pub(crate) type Late = Arc>; /// Everything a claimed node's executor arm might need to reach outside /// this process — bundled into one `Clone` struct rather than growing /// `run_swarm_node`'s parameter list per node kind (three forge-shaped /// node kinds landed in one slice; a fourth parameter each would have made /// the signature the least readable part of this file). Each field is /// built once at startup (see `main`) and is `None` exactly when that /// dependency isn't configured on this host — every arm below treats /// absence as *this node's* failure, not a reason to skip silently. #[derive(Clone)] pub(crate) struct WorkerDeps { pub(crate) auth: Option>, /// Empty until the forge token has been read (`forge::enable`), so a node /// run before then fails as "no forge access", and one run after succeeds. pub(crate) forge: Late>, /// The swarm queue connection, for nodes whose effect is a published /// event rather than an API call. A handle rather than the status /// reader it is cloned from: a node that announces a deploy has no /// business reading hive status, and `status.rs`'s own doc says the /// connection living there is an accident of construction order, not a /// claim that events are a kind of status. pub(crate) queue: 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`. pub(crate) wanted: Option>, /// Client-server API base of the swarm's homeserver, for the node that /// mints agents' accounts on it. `None` when /// `matrix_account::DEFAULT_HOMESERVER_ENV` is unset, which that node /// **fails** on rather than skips, same as every other field here. pub(crate) matrix_homeserver: 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_wire::wanted::AgentState = swarm_wire::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. pub(crate) async fn run_swarm_node( _id: hive_jobq::NodeId, kind: SwarmNodeKind, builder: hive_jobq::builder::JobBuilder, deps: WorkerDeps, ) -> ( hive_jobq::builder::JobBuilder, hive_jobq::scheduler::Outcome, ) { use hive_jobq::scheduler::Outcome; let forge = deps.forge.get().cloned(); let outcome = match kind { SwarmNodeKind::CreateIdentity { agent } => match deps.auth { None => { Outcome::Failed("no swarm-authelia-bridge is configured on this host".to_owned()) } Some(bridge) => match bridge.ensure_agent_identity(&agent).await { // Matched rather than discarded with `Ok(_)`: a heal writes // to a subject this job did not create, and that has to // reach a human. `Outcome` has no success-carrying variant, // so the controller's own log is the only channel a // succeeding node has today. Ok(BridgeResponse::Healed) => { tracing::warn!( %agent, "swarm jobq: create_identity marked an EXISTING identity as an \ agent — verify this name did not belong to a person" ); Outcome::Done } Ok(_) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), }, }, SwarmNodeKind::CreateRepo { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" .to_owned(), ), Some(client) => match forge_node("create_repo", client.create_repo(&agent)).await { Ok(full_name) => { tracing::info!(%full_name, "swarm jobq: create_repo done"); Outcome::Done } Err(e) => Outcome::Failed(format!("{e:#}")), }, }, SwarmNodeKind::CreateForgeUser { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" .to_owned(), ), Some(client) => { match forge_node("create_forge_user", client.ensure_agent_user(&agent)).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } }, SwarmNodeKind::AddRepoMember { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" .to_owned(), ), Some(client) => { match forge_node("add_repo_member", client.add_repo_member(&agent, &agent)).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } }, SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(forge, &agent).await, SwarmNodeKind::InitAgentConfigRepo { agent } => init_agent_config_repo(forge, &agent).await, SwarmNodeKind::MintAgentIdentity { agent } => mint_identity(&agent).await, SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(forge, &agent).await, SwarmNodeKind::MintAgentMatrixAccount { agent } => { mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await } SwarmNodeKind::MintHiveSenderToken { hive } => { mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await } SwarmNodeKind::MintHiveForgeToken { hive } => mint_hive_forge_token(forge, &hive).await, SwarmNodeKind::MintClassifierForgeToken => mint_classifier_forge_token(forge).await, SwarmNodeKind::MintBuilderForgeToken => mint_builder_forge_token(forge).await, SwarmNodeKind::MintCiMetricsForgeToken => mint_ci_metrics_forge_token(forge).await, SwarmNodeKind::BuildAgent { agent, rev, pr } => { build_agent(&deps, forge, agent, rev, pr).await } SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await, 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, rev } => match deps.queue { None => Outcome::Failed( "no swarm queue is configured on this host, so no hive can be told to deploy" .to_owned(), ), Some(client) => publish_deploy(&client, &hive, &agent, rev).await, }, }; (builder, outcome) } /// A forge-calling node's work, run under [`forge::FORGE_DEADLINE`]. Elapsing /// is the node's own error, naming `node`, so it fails like any other refused /// step instead of holding its executor slot. async fn forge_node( node: &str, work: impl std::future::Future>, ) -> Result { forge::within_forge_deadline(node, work) .await .unwrap_or_else(|| { Err(anyhow::anyhow!( "{node}: no answer from the forge within {}s", forge::FORGE_DEADLINE.as_secs() )) }) } /// Logged at `warn`, with the agent and the error, when [`mint_identity`] /// fails. The Grafana rule `hh-agent-identity-mint-failed` /// (`nix/host-modules/swarm-grafana.nix`) matches this exact text: change both /// together. const MINT_IDENTITY_FAILED: &str = "swarm jobq: mint_agent_identity failed"; /// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under /// `clippy::too_many_lines`. pub(crate) async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; match agent_identity::mint_and_verify(agent).await { Ok(()) => Outcome::Done, Err(e) => { let error = format!("{e:#}"); tracing::warn!(%agent, %error, "{MINT_IDENTITY_FAILED}"); Outcome::Failed(error) } } } /// The `MintAgentForgeToken` arm, lifted out so `run_swarm_node` stays under /// `clippy::too_many_lines`. async fn mint_forge_token( forge: Option>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no agent forge token can be minted" .to_owned(), ); }; match forge_node( "mint_agent_forge_token", client.ensure_agent_forge_token(agent), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `GrantDocsRead` arm, lifted out for the same reason. pub(crate) async fn grant_docs_read( forge: Option>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no docs grant can be made" .to_owned(), ); }; match forge_node("grant_docs_read", client.grant_docs_read(agent)).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `InitAgentConfigRepo` arm, lifted out for the same reason. async fn init_agent_config_repo( forge: Option>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" .to_owned(), ); }; match forge_node( "init_agent_config_repo", client.seed_agent_config(agent, agent), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `RenewAgentQueueCredential` arm, lifted out for the same reason. async fn renew_queue_credential(agent: &str) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; match agent_renewal::renew(agent).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintHiveForgeToken` arm, lifted out for the same reason. async fn mint_hive_forge_token( forge: Option>, hive: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no hive forge token can be minted" .to_owned(), ); }; match forge_node( "mint_hive_forge_token", client.ensure_hive_forge_token(hive), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `BuildAgent` arm, lifted out for `clippy::too_many_lines`. pub(crate) async fn build_agent( deps: &WorkerDeps, forge: Option>, agent: String, rev: String, pr: u64, ) -> hive_jobq::scheduler::Outcome { let job = swarm_wire::builder::BuildAgent { agent, rev, pr }; let builds = builder::runs::Builds::Live(deps.queue.clone()); let link_base = std::env::var(PUBLIC_URL_ENV).ok(); builder::run( deps.queue.as_ref(), forge.as_deref(), &builds, link_base.as_deref(), job, ) .await } /// The `MintBuilderForgeToken` arm, lifted out for the same reason. async fn mint_builder_forge_token( forge: Option>, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no builder forge token can be minted" .to_owned(), ); }; match forge_node( "mint_builder_forge_token", client.ensure_builder_forge_token(), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintCiMetricsForgeToken` arm, lifted out for the same reason. async fn mint_ci_metrics_forge_token( forge: Option>, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no ci metrics forge token can be minted" .to_owned(), ); }; match forge_node( "mint_ci_metrics_forge_token", client.ensure_ci_metrics_forge_token(), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintClassifierForgeToken` arm, lifted out for the same reason. async fn mint_classifier_forge_token( forge: Option>, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so no review classifier forge token can be minted" .to_owned(), ); }; match forge_node( "mint_classifier_forge_token", client.ensure_classifier_forge_token(), ) .await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintAgentMatrixAccount` arm, lifted out for the same reason. async fn mint_matrix_account( homeserver: Option<&str>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; // Loud on purpose: an agent quietly created without a matrix account is // an agent nothing can talk to, and no hive mints one any more. let Some(base) = homeserver else { return Outcome::Failed(format!( "no matrix homeserver configured on this host ({} unset), so no agent \ matrix account can be minted", matrix_account::DEFAULT_HOMESERVER_ENV )); }; match matrix_account::agent_token::ensure_agent_matrix_account(base, agent).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintHiveSenderToken` arm, lifted out for the same reason. async fn mint_hive_sender(homeserver: Option<&str>, hive: &str) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(base) = homeserver else { return Outcome::Failed(format!( "no matrix homeserver configured on this host ({} unset), so no hive \ sender token can be minted", matrix_account::DEFAULT_HOMESERVER_ENV )); }; match matrix_account::hive_sender::ensure_hive_sender_token(base, hive).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`] — unless it turns /// out not to be new: an agent that already has a declaration (other than /// `Destroyed`, which this treats as reusable) is left alone, so a retried /// or migrating create is idempotent rather than pausing an already-running /// agent. `wanted.rs` itself has no such carve-out, so this node makes the /// call. async fn declare_new_agent( writer: &wanted::WantedWriter, hive: &str, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; use swarm_wire::wanted::AgentState; let existing = match writer.view(hive).await { Ok(declaration) => declaration.and_then(|d| d.agents.get(agent).map(|a| a.state)), Err(e) => return Outcome::Failed(format!("{e:#}")), }; if let Some(state) = existing && state != AgentState::Destroyed { tracing::debug!( hive, agent, ?state, "agent already declared, leaving its wanted state alone" ); return Outcome::Done; } 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 /// client's write buffer and returns, so a node that reported `Done` on /// that alone would be claiming a delivery it has no evidence for — the /// same ordering `webhook::announce_knowledge_change` documents. async fn publish_deploy( client: &async_nats::Client, hive: &str, agent: &str, rev: Option, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; // One subject per hive, so the other hives are never woken by this. let subject = swarm_queue_client::deploy_subject(hive); let request = swarm_queue_client::DeployRequest { agent: agent.to_owned(), rev, }; let payload = match serde_json::to_vec(&request) { Ok(payload) => payload, Err(e) => return Outcome::Failed(format!("encoding the deploy request failed: {e}")), }; if let Err(e) = client.publish(subject.clone(), payload.into()).await { return Outcome::Failed(format!("publishing to {subject} failed: {e}")); } if let Err(e) = client.flush().await { return Outcome::Failed(format!( "flushing the deploy event to {subject} failed: {e}" )); } tracing::info!(%subject, %hive, %agent, rev = ?request.rev, "swarm jobq: deploy requested"); Outcome::Done } /// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/ /// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node, /// spawn the future that runs + completes it, loop again immediately if /// something started (more may now be runnable), otherwise back off /// briefly before re-polling. /// /// No shutdown signal to wire in — unlike `hive-c0re`'s `coord.shutdown_rx()`, /// this daemon has no graceful-shutdown machinery at all yet (`main`'s /// `axum::serve` runs unconditionally to process exit), so this loop /// matches that: it rides the runtime down with the process, same as /// every in-flight HTTP request does. /// /// Cheap to run with an empty graph: `claim_next` on a graph nothing was /// ever inserted into just returns `None` every poll. /// /// `deps` is cloned per iteration (its fields are `Arc` clones, not /// reconnects) and moved into the closure `claim_next` takes ownership /// of — `run_swarm_node` needs its own owned copy since the claimed /// future may outlive this loop iteration. pub(crate) fn spawn_jobq_worker( sched: Arc>>, deps: WorkerDeps, ) { tokio::spawn(async move { loop { let deps = deps.clone(); let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { run_swarm_node(id, kind, builder, deps) }); match runner { Some(runner) => { tokio::spawn(async move { let (id, grew) = runner.await; if let Err(e) = grew { tracing::warn!( node = id.get(), error = %e, "swarm jobq: grown job rejected" ); } }); // Something just started — more may be runnable right // now, so loop again immediately rather than sleeping. } None => { // Nothing runnable. Bounded poll rather than an event // wake (unlike hive-c0re's `notify.notify_one()`, // there is no completion-signal channel here yet) — // fine at this daemon's scale (one graph, no // submitters yet); revisit if/when that stops holding. tokio::time::sleep(std::time::Duration::from_millis(200)).await; } } } }); } /// Queue a build check of a config PR's new head. Every head is built, on /// whichever hive the agent runs: the check builds on the builder's own /// agent base. pub(crate) fn queue_config_build(state: &AppState, changed: config_pr::ChangedConfigPr) { let (agent, rev, pr) = (changed.agent.clone(), changed.rev.clone(), changed.pr); let queued = queue_build( &mut state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), changed, ); match queued { Ok(superseded) => { tracing::info!(%agent, %rev, pr, superseded, "config-pr head: build queued"); } Err(e) => { tracing::warn!(%agent, %rev, pr, error = %e, "config-pr head: queueing the build failed"); } } } /// Cancel every pending build of the same PR, then insert the new one behind /// [`SwarmResourceKind::Builder`]. A build already running is left to finish: /// its status lands on its own commit, not on the new head. Returns how many /// pending builds were cancelled. pub(crate) fn queue_build( sched: &mut hive_jobq::scheduler::Scheduler, changed: config_pr::ChangedConfigPr, ) -> Result { let config_pr::ChangedConfigPr { agent, rev, pr } = changed; let stale: Vec = sched .graph() .nodes() .filter(|n| n.state == hive_jobq::State::Pending) .filter(|n| { matches!(&n.payload, SwarmNodeKind::BuildAgent { agent: a, pr: p, .. } if *a == agent && *p == pr) }) .map(|n| n.id) .collect(); let superseded = stale .into_iter() .filter(|&id| sched.cancel_node(id)) .count(); sched .insert_job(None, |b| { vec![ b.node(SwarmNodeKind::BuildAgent { agent, rev, pr }) .needs(SwarmResourceKind::Builder) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; Ok(superseded) } /// Queue the deploy of a merged config PR on the one hive that runs its agent. /// /// The hive is found by scanning every hive's wanted state, since nothing /// records an agent's hive (see `forge::initial_agent_nix`). Zero claimants or /// several deploy nothing: there is no hive to send it to, or no telling which. pub(crate) async fn queue_merged_config_deploy( state: &AppState, merged: config_pr::MergedConfigPr, ) { let config_pr::MergedConfigPr { agent, rev } = merged; let declared = match read_declarations(state, state.hives.iter()).await { Ok(declared) => declared, Err((hive, e)) => { tracing::warn!( %agent, %rev, %hive, error = %format!("{e:#}"), "config-pr merge: cannot read where the agent is placed; nothing deployed" ); return; } }; let claimants = claiming_hives(&agent, &declared); let [hive] = claimants.as_slice() else { tracing::warn!( %agent, %rev, ?claimants, "config-pr merge: not exactly one hive places this agent; nothing deployed" ); return; }; let hive = hive.clone(); let inserted = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert_job(None, |b| { vec![ b.node(SwarmNodeKind::TriggerDeploy { hive: hive.clone(), agent: agent.clone(), rev: Some(rev.clone()), }) .guid(), ] }); match inserted { Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"), Err(e) => { tracing::warn!(%agent, %rev, %hive, error = %e, "config-pr merge: queueing the deploy failed"); } } } /// The sub-DAG one agent creation is: the nodes, and the edges between them. /// /// A function rather than a closure inside [`create_agent`](crate::routes::agents::create_agent) so the endpoint's /// validation and the graph's shape can each be read without scrolling past /// the other — and so the one handle the response reports is returned from /// the place that decides which node it is. pub(crate) fn declare_agent_job( b: &hive_jobq::builder::JobBuilder, agent: &str, hive: &str, archetype: bool, ) -> Vec { let create_identity = b.node(SwarmNodeKind::CreateIdentity { agent: agent.to_owned(), }); // A second, independent root: a forge user needs neither an authelia // subject nor an existing repo, so it does not chain off // `create_identity` (see the doc comment above). let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser { agent: agent.to_owned(), }); // An archetype agent has no config repo: its hive builds it from the // archetype's module, so there is no repo to create, share or seed. let init_config = (!archetype).then(|| { let create_repo = b .node(SwarmNodeKind::CreateRepo { agent: agent.to_owned(), }) .after_ok(create_identity); // `AddRepoMember` needs both parents: the repo to add a collaborator // to, and the forge user to add as one — adding a nonexistent user is // a Forgejo validation error, not an idempotent no-op. // `InitAgentConfigRepo` needs only the repo — see the doc comment // above for why. let _add_repo_member = b .node(SwarmNodeKind::AddRepoMember { agent: agent.to_owned(), }) .after_ok(create_repo) .after_ok(create_forge_user); b.node(SwarmNodeKind::InitAgentConfigRepo { agent: agent.to_owned(), }) .after_ok(create_repo) .guid() }); // Read on `internal/docs`, a repo that exists whatever this agent is, so // it needs only the user. Not a parent of the deploy: the agent reads docs // once running, and a failed grant is retried by `forge::docs_grant::spawn`. let _grant_docs = b .node(SwarmNodeKind::GrantDocsRead { agent: agent.to_owned(), }) .after_ok(create_forge_user); // The agent's own identity at the swarm secret store, minted and // published at swarm level so no hive ever needs the capability to mint // one. After `create_identity` because the certificate names a principal // the swarm has agreed exists — not because anything in the store reads // authelia. let mint_identity = b .node(SwarmNodeKind::MintAgentIdentity { agent: agent.to_owned(), }) .after_ok(create_identity); // The agent's own forge token, in the store before the hive deploys // the container that pulls it. After the forge user, because the token // is minted for that user. let mint_forge_token = b .node(SwarmNodeKind::MintAgentForgeToken { agent: agent.to_owned(), }) .after_ok(create_forge_user); // The agent's account on the swarm's homeserver. A root of its own: // creating it needs neither an authelia subject nor a forge user nor a // repo, and chaining it behind one of those would make an unrelated // failure look like a matrix failure. let mint_matrix = b.node(SwarmNodeKind::MintAgentMatrixAccount { agent: agent.to_owned(), }); // 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 // swarm level actually put it on a hive, rather than leaving a // provisioned name nobody runs. An archetype agent has no repo, so its // deploy waits on the mints and the declaration alone. // // `after_any` on the mint, not `after_ok`: a hive cannot pass down a // certificate the swarm has not published, so the deploy must not // overtake the mint — but a host whose store or agent PKI is not set up // must still create agents. `after_ok` there would turn a store outage // into an agent nobody runs. // // The forge-token mint gets `after_any` for the same reason: a host with // no forge or no store must still create agents; only that node fails, // by name, and the agent's container fetches the token on its next // timer tick once one is minted. // // `after_ok` on the declaration, though: an agent deployed without its // pause landing first is the race this node exists to close, so a // declaration that did not land must not be deployed past. A host with no // queue at all fails this node — but `TriggerDeploy` has nothing to // publish to on such a host either, so the strict edge cancels a node // that could not have succeeded anyway. let trigger_deploy = b .node(SwarmNodeKind::TriggerDeploy { hive: hive.to_owned(), agent: agent.to_owned(), rev: None, }) .after_any(mint_identity) .after_any(mint_forge_token) // `after_any` for the same reason: the container reads this token // rather than producing it, so the deploy must not overtake the mint — // but a host with no homeserver configured must still create agents, // and only this node fails, by name. .after_any(mint_matrix) .after_ok(set_wanted); if let Some(init_config) = init_config { let _ = trigger_deploy.after_ok(init_config); } vec![create_identity.guid()] } /// Insert one job per agent: `CreateForgeUser`, then `MintAgentForgeToken` /// `after_ok` it — the same edge [`declare_agent_job`] uses. Returns the mint /// nodes' ids. /// /// The one place that node is queued outside agent creation: both the manual /// route below and `forge::agent_token::spawn`'s periodic pass come through /// here, so a backfilled mint is the same node a new agent gets. The user /// node is what lets the pass reach an agent that has a store identity but /// no forge account; for one that has an account it is a no-op, because /// `forge::Client::ensure_agent_user` folds "already exists" into success. pub(crate) fn queue_forge_token_mints( sched: &Mutex>, agents: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(agents.len()); for agent in agents { let queued = sched .insert_job(None, |b| { let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser { agent: agent.clone(), }); vec![ b.node(SwarmNodeKind::MintAgentForgeToken { agent }) .after_ok(create_forge_user) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one job per agent: `CreateForgeUser`, then `GrantDocsRead` `after_ok` /// it — the same edge [`declare_agent_job`] uses. Returns the grant nodes' ids. /// `forge::docs_grant::spawn`'s periodic pass comes through here, so a /// backfilled grant is the same node a new agent gets; the user node is a /// no-op for an agent that has an account. pub(crate) fn queue_docs_grants( sched: &Mutex>, agents: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(agents.len()); for agent in agents { let queued = sched .insert_job(None, |b| { let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser { agent: agent.clone(), }); vec![ b.node(SwarmNodeKind::GrantDocsRead { agent }) .after_ok(create_forge_user) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one `MintAgentMatrixAccount` job per agent and return the nodes' /// ids. `matrix_account::agent_token::spawn`'s periodic pass comes through /// here, so a backfilled mint is the same node a new agent gets. pub(crate) fn queue_matrix_account_mints( sched: &Mutex>, agents: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(agents.len()); for agent in agents { let queued = sched .insert_job(None, |b| { vec![ b.node(SwarmNodeKind::MintAgentMatrixAccount { agent }) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one `MintHiveSenderToken` job per hive and return the nodes' ids. /// `matrix_account::hive_sender::spawn`'s periodic pass is the one caller. pub(crate) fn queue_hive_sender_mints( sched: &Mutex>, hives: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(hives.len()); for hive in hives { let queued = sched .insert_job(None, |b| { vec![b.node(SwarmNodeKind::MintHiveSenderToken { hive }).guid()] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one `MintHiveForgeToken` job per hive and return the nodes' ids. /// `forge::hive_token::spawn`'s periodic pass is the one caller. pub(crate) fn queue_hive_forge_token_mints( sched: &Mutex>, hives: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(hives.len()); for hive in hives { let queued = sched .insert_job(None, |b| { vec![b.node(SwarmNodeKind::MintHiveForgeToken { hive }).guid()] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one `MintClassifierForgeToken` job and return its node's id. /// `forge::classifier_account::spawn`'s periodic pass is the one caller. pub(crate) fn queue_classifier_forge_token_mint( sched: &Mutex>, ) -> Result> { sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert_job(None, |b| { vec![b.node(SwarmNodeKind::MintClassifierForgeToken).guid()] }) .map_err(|e| anyhow::anyhow!("{e}")) } /// Insert one `MintBuilderForgeToken` job and return its node's id. /// `forge::builder_account::spawn`'s periodic pass is the one caller. pub(crate) fn queue_builder_forge_token_mint( sched: &Mutex>, ) -> Result> { sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert_job(None, |b| { vec![b.node(SwarmNodeKind::MintBuilderForgeToken).guid()] }) .map_err(|e| anyhow::anyhow!("{e}")) } /// Insert one `MintCiMetricsForgeToken` job and return its node's id. /// `forge::ci_metrics_account::spawn`'s periodic pass is the one caller. pub(crate) fn queue_ci_metrics_forge_token_mint( sched: &Mutex>, ) -> Result> { sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert_job(None, |b| { vec![b.node(SwarmNodeKind::MintCiMetricsForgeToken).guid()] }) .map_err(|e| anyhow::anyhow!("{e}")) } /// Insert one job per renewal and return the ids of its nodes: /// `MintAgentIdentity` for a certificate, `RenewAgentQueueCredential` for a /// queue secret, the second `after_any` the first when an agent needs both. /// `agent_renewal::spawn`'s periodic pass comes through here. /// /// Chained rather than parallel because `mint_and_verify` reads the queue /// secret back and compares it with the one it read first; a re-mint landing /// in between fails that node. `after_any`, so a certificate that could not be /// re-issued does not hold up the secret. pub(crate) fn queue_agent_renewals( sched: &Mutex>, renewals: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(renewals.len()); for renewal in renewals { let queued = sched .insert_job(None, |b| { let mut asked = Vec::new(); let cert = renewal.cert.then(|| { b.node(SwarmNodeKind::MintAgentIdentity { agent: renewal.agent.clone(), }) .guid() }); asked.extend(cert); if renewal.secret { let secret = b.node(SwarmNodeKind::RenewAgentQueueCredential { agent: renewal.agent.clone(), }); let secret = match cert { Some(cert) => secret.after_any(cert), None => secret, }; asked.push(secret.guid()); } asked }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } #[cfg(test)] mod tests { use super::*; use crate::routes::tests::state_with_roster; pub(crate) fn changed(agent: &str, rev: &str, pr: u64) -> crate::config_pr::ChangedConfigPr { crate::config_pr::ChangedConfigPr { agent: agent.to_owned(), rev: rev.to_owned(), pr, } } fn build_state( sched: &hive_jobq::scheduler::Scheduler, rev: &str, ) -> hive_jobq::State { sched .graph() .nodes() .find(|n| matches!(&n.payload, SwarmNodeKind::BuildAgent { rev: r, .. } if r == rev)) .map(|n| n.state) .expect("a build of that rev") } /// A newer head of the same PR replaces its pending build; another PR's /// build, and the same PR's running build, are left alone. #[tokio::test] async fn a_newer_push_replaces_the_prs_pending_build_only() { let sched = std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); let queue = |c| super::queue_build(&mut sched.lock().unwrap(), c).expect("queued"); assert_eq!(queue(changed("atlas", "r1", 3)), 0); // r1 takes the builder and stays running: its future is never awaited. let _running = hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move { (b, hive_jobq::scheduler::Outcome::Done) }) .expect("r1 is runnable"); assert_eq!( queue(changed("atlas", "r2", 3)), 0, "r1 is running, not pending" ); assert_eq!(queue(changed("argus", "o1", 4)), 0); assert_eq!(queue(changed("atlas", "r3", 3)), 1, "r2 was pending"); let guard = sched.lock().unwrap(); assert_eq!(build_state(&guard, "r1"), hive_jobq::State::Running); assert_eq!(build_state(&guard, "r2"), hive_jobq::State::Cancelled); assert_eq!(build_state(&guard, "r3"), hive_jobq::State::Pending); assert_eq!(build_state(&guard, "o1"), hive_jobq::State::Pending); } /// Builds share one builder slot: a second build is not claimable while /// the first runs, and is once it has finished. #[tokio::test] async fn one_build_runs_at_a_time() { let sched = std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); for c in [changed("atlas", "a1", 1), changed("argus", "b1", 2)] { super::queue_build(&mut sched.lock().unwrap(), c).expect("queued"); } let run = |sched| { hive_jobq::scheduler::Scheduler::claim_next(sched, |_, _, b| async move { (b, hive_jobq::scheduler::Outcome::Done) }) }; let first = run(&sched).expect("one build is runnable"); assert!( run(&sched).is_none(), "a second build started beside the first" ); first.await.1.expect("no growth"); run(&sched) .expect("the second build runs once the first is done") .await .1 .expect("no growth"); } /// Every node whose first request goes to the forge, run against a forge /// that accepts the connection and never answers: each ends `Failed` /// with an error naming the node instead of holding its executor slot. /// `MintAgentForgeToken` is not in the list: it logs in to the secret /// store before its first forge request, and no test here has a store. #[tokio::test(start_paused = true)] async fn a_forge_node_fails_when_the_forge_never_answers() { // Bound but never accepted: the kernel completes the handshake and // buffers the request, and no response ever comes. 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(); let forge = super::Late::>::default(); assert!( forge .set(std::sync::Arc::new(crate::forge::Client::for_test_forge( url ))) .is_ok() ); let agent = || "atlas".to_owned(); let nodes = [ ("create_repo", SwarmNodeKind::CreateRepo { agent: agent() }), ( "create_forge_user", SwarmNodeKind::CreateForgeUser { agent: agent() }, ), ( "add_repo_member", SwarmNodeKind::AddRepoMember { agent: agent() }, ), ( "grant_docs_read", SwarmNodeKind::GrantDocsRead { agent: agent() }, ), ( "init_agent_config_repo", SwarmNodeKind::InitAgentConfigRepo { agent: agent() }, ), ( "mint_hive_forge_token", SwarmNodeKind::MintHiveForgeToken { hive: "pr1ma".to_owned(), }, ), ( "mint_classifier_forge_token", SwarmNodeKind::MintClassifierForgeToken, ), ( "mint_builder_forge_token", SwarmNodeKind::MintBuilderForgeToken, ), ( "mint_ci_metrics_forge_token", SwarmNodeKind::MintCiMetricsForgeToken, ), ]; for (label, kind) in nodes { let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched.append(kind, Vec::new(), None).expect("insert"); let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); let deps = WorkerDeps { auth: None, forge: forge.clone(), queue: None, wanted: None, matrix_homeserver: 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"); tokio::time::timeout(std::time::Duration::from_mins(10), runner) .await .unwrap_or_else(|_| panic!("{label}: node still running after 10 minutes")) .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, "{label}"); let error = node.error.as_deref().unwrap_or_default(); assert!( error.contains(label) && error.contains("no answer from the forge"), "{label}: expected a deadline error naming the node, got {error:?}" ); } drop(listener); } #[tokio::test] async fn create_repo_node_runs_end_to_end_and_fails_without_forge_configured() { unsafe { std::env::remove_var("SWARM_CONTROLLER_FORGE_URL"); } let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched .append( SwarmNodeKind::CreateRepo { 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: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: 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("forge"), "expected a forge-not-configured error, got {:?}", node.error ); } /// Sibling of the `CreateRepo` test above, same shape: drives /// `SwarmNodeKind::CreateForgeUser` through the real scheduler with no /// forge configured, asserting the graceful-absence-is-failure branch — /// this node's happy path needs a live forge, same caveat as /// `CreateRepo`'s. #[tokio::test] async fn create_forge_user_node_runs_end_to_end_and_fails_without_forge_configured() { unsafe { std::env::remove_var("SWARM_CONTROLLER_FORGE_URL"); } let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched .append( SwarmNodeKind::CreateForgeUser { 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: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: 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("forge"), "expected a forge-not-configured error, got {:?}", node.error ); } /// Third sibling of the two above, and the same deliberate caveat: with /// no agent PKI named this reaches only the graceful-absence-is-failure /// branch, before any network call. The happy path is a live store and a /// real login, which is exactly why it is `mint_and_verify`'s own job to /// prove it at agent-creation time rather than a unit test's. /// /// What this does pin is the degrade: a host whose environment does not /// name the agent PKI fails this one node with a reason that names the /// variable, and creates the agent anyway. #[tokio::test] async fn mint_agent_identity_node_runs_end_to_end_and_fails_without_the_agent_pki_named() { let pki = [ crate::agent_identity::ENV_AGENT_PKI_MOUNT, crate::agent_identity::ENV_AGENT_PKI_ROLE, ]; assert!( pki.iter().all(|var| std::env::var_os(var).is_none()), "{} and {} must be unset: with them set this test would reach a live store", pki[0], pki[1] ); let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched .append( SwarmNodeKind::MintAgentIdentity { 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: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: 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); let error = node.error.as_deref().unwrap_or_default(); assert!( error.contains(crate::agent_identity::ENV_AGENT_PKI_MOUNT), "the reason must name the variable an operator has to set, got {error:?}" ); } /// 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: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: 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!( super::NEW_AGENT_WANTED_STATE, swarm_wire::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 = crate::routes::agents::create_agent( axum::extract::State(state), axum::Json(swarm_controller_api::agents::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), limits: None, inference: None, permissions: None, archetype: None, agents_md: None, }), ) .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`" ); } /// Node labels of the graph [`super::declare_agent_job`] builds, and the /// labels the deploy node waits on. fn creation_graph(archetype: bool) -> (Vec, Vec) { use hive_jobq_wire::WireNode as _; let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); sched .insert_job(None, |b| { super::declare_agent_job(b, "atlas", "pr1ma", archetype) }) .expect("the graph inserts"); let graph = sched.graph(); let labels = graph.nodes().map(|n| n.payload.label()).collect(); let deploy = graph .nodes() .find(|n| n.payload.label() == "trigger_deploy") .expect("every creation deploys"); let waits = deploy .deps .iter() .filter_map(|d| match d { hive_jobq::Dep::Node { id, .. } => graph.node(*id).map(|n| n.payload.label()), hive_jobq::Dep::Resource { .. } => None, }) .collect(); (labels, waits) } /// An archetype agent's graph has no repo nodes and its deploy waits on /// no repo; a custom agent's has all three, and its deploy waits on the /// seeded repo. #[test] fn an_archetype_agent_is_created_without_a_config_repo() { const REPO_NODES: [&str; 3] = ["create_repo", "add_repo_member", "init_agent_config_repo"]; let (custom, custom_waits) = creation_graph(false); for repo in REPO_NODES { assert!(custom.iter().any(|l| l == repo), "custom lacks {repo}"); } assert!(custom_waits.iter().any(|l| l == "init_agent_config_repo")); let (archetype, archetype_waits) = creation_graph(true); for repo in REPO_NODES { assert!(!archetype.iter().any(|l| l == repo), "archetype has {repo}"); } for kept in [ "create_identity", "create_forge_user", "grant_docs_read", "mint_agent_identity", "mint_agent_forge_token", "mint_agent_matrix_account", "set_agent_wanted", ] { assert!( archetype.iter().any(|l| l == kept), "archetype lacks {kept}" ); assert!(custom.iter().any(|l| l == kept), "custom lacks {kept}"); } assert!(archetype_waits.iter().any(|l| l == "set_agent_wanted")); assert_eq!(archetype.len() + REPO_NODES.len(), custom.len()); } /// The ordering the operator's ruling requires: a hive cannot pass down /// a certificate the swarm has not published, so the deploy message must /// not leave before the mint is terminal. /// /// Asserted on the graph `create_agent` builds, because the edge is one /// line in a builder closure and its absence changes nothing observable /// until a real agent boots without an identity. #[tokio::test] async fn the_deploy_waits_for_the_mint_and_is_not_cancelled_by_it() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); let _queued = crate::routes::agents::create_agent( axum::extract::State(state), axum::Json(swarm_controller_api::agents::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), limits: None, inference: None, permissions: None, archetype: None, agents_md: None, }), ) .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 mint = id_of("mint_agent_identity"); 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 == mint => Some(*when), _ => None, }) .expect("the deploy waits for the mint"); assert!( when.accepts(hive_jobq::TerminalState::Failed), "a failed or unconfigured agent PKI issue must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } /// The forge-token mint sits between the forge user and the deploy: it /// needs the user to exist, and the deploy must not overtake it — but a /// host with no forge or no store must still deploy the agent. #[tokio::test] async fn the_forge_token_mint_follows_the_user_and_does_not_block_the_deploy() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); let _queued = crate::routes::agents::create_agent( axum::extract::State(state), axum::Json(swarm_controller_api::agents::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), limits: None, inference: None, permissions: None, archetype: None, agents_md: None, }), ) .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 edge = |from: hive_jobq::NodeId, to: &str| { graph .node(id_of(to)) .expect("just found") .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == from => Some(*when), _ => None, }) .unwrap_or_else(|| panic!("{to} waits for node {}", from.get())) }; let user_to_mint = edge(id_of("create_forge_user"), "mint_agent_forge_token"); assert!( !user_to_mint.accepts(hive_jobq::TerminalState::Failed), "a token for a user that was never created cannot be minted; this \ edge has to be `after_ok`" ); let mint_to_deploy = edge(id_of("mint_agent_forge_token"), "trigger_deploy"); assert!( mint_to_deploy.accepts(hive_jobq::TerminalState::Failed), "a host with no forge or store must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } /// Agent creation grants `internal/docs` read once the forge user exists, /// and a failed grant does not hold up the deploy (the pass retries it). #[tokio::test] async fn the_docs_grant_follows_the_user_and_is_not_a_deploy_parent() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); let _queued = crate::routes::agents::create_agent( axum::extract::State(state), axum::Json(swarm_controller_api::agents::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), limits: None, inference: None, permissions: None, archetype: None, agents_md: None, }), ) .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 grant = graph.node(id_of("grant_docs_read")).expect("just found"); assert_eq!(grant.payload.data(grant.id.get())["agent"], "atlas"); let when = grant .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == id_of("create_forge_user") => { Some(*when) } _ => None, }) .expect("the grant waits for the forge user"); assert!( !when.accepts(hive_jobq::TerminalState::Failed), "adding a user that was never created as a collaborator is a forge error; \ this edge has to be `after_ok`" ); let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); assert!( !deploy.deps.iter().any(|d| matches!( d, hive_jobq::Dep::Node { id, .. } if *id == grant.id )), "the deploy must not wait on the docs grant" ); } /// The periodic pass queues the same pair of nodes agent creation does. #[test] fn a_queued_docs_grant_creates_the_forge_user_first() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_docs_grants(&sched, vec!["a".to_owned(), "b".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2, "one returned id per agent: its grant node"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let mut labels: Vec = graph.nodes().map(|n| n.payload.label()).collect(); labels.sort(); assert_eq!( labels, [ "create_forge_user", "create_forge_user", "grant_docs_read", "grant_docs_read" ] ); for grant in graph .nodes() .filter(|n| n.payload.label() == "grant_docs_read") { assert!(ids.contains(&grant.id)); assert!( grant .deps .iter() .any(|d| matches!(d, hive_jobq::Dep::Node { when, .. } if !when.accepts(hive_jobq::TerminalState::Failed))), "each grant waits `after_ok` on its user" ); } } #[test] fn a_forge_token_node_renders_the_agent() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::MintAgentForgeToken { agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "mint_agent_forge_token"); assert_eq!(kind.data(1)["agent"], "atlas"); } #[test] fn a_matrix_account_node_renders_the_agent() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::MintAgentMatrixAccount { agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "mint_agent_matrix_account"); assert_eq!(kind.data(1)["agent"], "atlas"); } /// The periodic pass's queueing: a certificate re-issue is the node agent /// creation mints with, a secret re-mint its own node, and an agent due /// both gets the secret `after_any` the certificate. #[test] fn a_queued_renewal_chains_the_secret_after_the_certificate() { use crate::agent_renewal::Renewal; use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let r = |agent: &str, cert, secret| Renewal { agent: agent.to_owned(), cert, secret, }; let ids = super::queue_agent_renewals( &sched, vec![ r("both", true, true), r("cert", true, false), r("secret", false, true), ], ) .expect("three jobs insert"); assert_eq!(ids.len(), 4, "one returned id per node"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let agent_of = |n: &hive_jobq::Node| { n.payload.data(n.id.get())["agent"] .as_str() .expect("agent is a string") .to_owned() }; let mut nodes: Vec<(String, String)> = graph .nodes() .map(|n| (agent_of(n), n.payload.label())) .collect(); nodes.sort(); assert_eq!( nodes, [ ("both".to_owned(), "mint_agent_identity".to_owned()), ("both".to_owned(), "renew_agent_queue_credential".to_owned()), ("cert".to_owned(), "mint_agent_identity".to_owned()), ( "secret".to_owned(), "renew_agent_queue_credential".to_owned() ), ] ); let find = |agent: &str, label: &str| { graph .nodes() .find(|n| n.payload.label() == label && agent_of(n) == agent) .unwrap_or_else(|| panic!("{agent} has a {label} node")) }; let cert = find("both", "mint_agent_identity").id; let secret = find("both", "renew_agent_queue_credential"); let when = secret .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == cert => Some(*when), _ => None, }) .expect("the secret waits for the certificate"); assert!( when.accepts(hive_jobq::TerminalState::Failed), "a certificate that could not be re-issued must not hold up the secret" ); assert!( find("secret", "renew_agent_queue_credential") .deps .is_empty(), "a secret on its own waits for nothing" ); } /// Agent creation mints the matrix account as a root of its own, and the /// deploy waits for it without being cancelled by it: a host with no /// homeserver configured must still deploy the agent. #[tokio::test] async fn the_matrix_account_mint_is_a_root_and_does_not_block_the_deploy() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); let _queued = crate::routes::agents::create_agent( axum::extract::State(state), axum::Json(swarm_controller_api::agents::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), limits: None, inference: None, permissions: None, archetype: None, agents_md: None, }), ) .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 find = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label) .unwrap_or_else(|| panic!("the graph holds a {label} node")) }; let mint = find("mint_agent_matrix_account"); assert!(mint.deps.is_empty(), "the mint waits for nothing"); let when = find("trigger_deploy") .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == mint.id => Some(*when), _ => None, }) .expect("the deploy waits for the matrix mint"); assert!( when.accepts(hive_jobq::TerminalState::Failed), "a host with no homeserver must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } /// The backfill's queueing: one mint node per agent, nothing else. #[test] fn a_queued_matrix_mint_is_one_node_per_agent() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_matrix_account_mints(&sched, vec!["a".to_owned(), "b".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut agents: Vec = guard .graph() .nodes() .map(|n| { assert_eq!(n.payload.label(), "mint_agent_matrix_account"); n.payload.data(n.id.get())["agent"] .as_str() .expect("agent is a string") .to_owned() }) .collect(); agents.sort(); assert_eq!(agents, ["a", "b"]); } /// The hive sender-token pass's queueing: one mint node per hive, and /// the node names the hive rather than an agent. #[test] fn a_queued_hive_sender_mint_is_one_node_per_hive() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_hive_sender_mints(&sched, vec!["alpha".to_owned(), "beta".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut hives: Vec = guard .graph() .nodes() .map(|n| { assert_eq!(n.payload.label(), "mint_hive_sender_token"); let data = n.payload.data(n.id.get()); assert!(data.get("agent").is_none(), "{data}"); data["hive"].as_str().expect("hive is a string").to_owned() }) .collect(); hives.sort(); assert_eq!(hives, ["alpha", "beta"]); } /// The hive forge-token pass's queueing: one node per hive, naming the /// hive rather than an agent. #[test] fn a_queued_hive_forge_token_mint_is_one_node_per_hive() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_hive_forge_token_mints( &sched, vec!["alpha".to_owned(), "beta".to_owned()], ) .expect("two jobs insert"); assert_eq!(ids.len(), 2); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut hives: Vec = guard .graph() .nodes() .map(|n| { assert_eq!(n.payload.label(), "mint_hive_forge_token"); let data = n.payload.data(n.id.get()); assert!(data.get("agent").is_none(), "{data}"); data["hive"].as_str().expect("hive is a string").to_owned() }) .collect(); hives.sort(); assert_eq!(hives, ["alpha", "beta"]); } /// The manual route and the periodic pass both go through /// `queue_forge_token_mints`, so this is the assertion that a backfilled /// mint is the same pair of nodes agent creation inserts: the forge user, /// then the mint `after_ok` it, per agent. #[test] fn a_queued_mint_creates_the_forge_user_first() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_forge_token_mints(&sched, vec!["a".to_owned(), "b".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2, "one returned id per agent: its mint node"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let agent_of = |n: &hive_jobq::Node| { n.payload.data(n.id.get())["agent"] .as_str() .expect("agent is a string") .to_owned() }; for agent in ["a", "b"] { let find = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label && agent_of(n) == agent) .unwrap_or_else(|| panic!("{agent} has a {label} node")) }; let user = find("create_forge_user").id; let mint = find("mint_agent_forge_token"); assert!(ids.contains(&mint.id), "the returned id is {agent}'s mint"); let when = mint .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == user => Some(*when), _ => None, }) .unwrap_or_else(|| panic!("{agent}'s mint waits for its own forge user")); assert!( !when.accepts(hive_jobq::TerminalState::Failed), "a token cannot be minted for a user that was never created; this \ edge has to be `after_ok`" ); } assert_eq!( graph.nodes().count(), 4, "two nodes per agent, nothing else" ); } /// The backfill end to end, minus the IO: an agent the pass observed with /// no forge user is planned, and the job it gets creates that user before /// it mints. #[test] fn an_agent_with_no_forge_user_gets_a_user_and_a_mint() { use crate::forge::agent_token::{Observed, plan}; use hive_jobq_wire::WireNode as _; let agents = plan(&[("ruth".to_owned(), Observed::NoForgeUser)]); let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); super::queue_forge_token_mints(&sched, agents).expect("one job inserts"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut labels: Vec = guard.graph().nodes().map(|n| n.payload.label()).collect(); labels.sort(); assert_eq!(labels, ["create_forge_user", "mint_agent_forge_token"]); } } #[cfg(test)] mod mint_identity_tests { use std::sync::{Arc, Mutex}; use hive_jobq::scheduler::Outcome; use tracing_subscriber::layer::SubscriberExt as _; use super::mint_identity; /// One event's level and its fields, each by name. type Event = (tracing::Level, Vec<(String, String)>); /// Every event. #[derive(Clone, Default)] struct Events(Arc>>); impl tracing_subscriber::Layer for Events { fn on_event( &self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>, ) { let mut fields = Vec::new(); event.record( &mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| { fields.push((field.name().to_owned(), format!("{value:?}"))); }, ); self.0 .lock() .expect("unpoisoned") .push((*event.metadata().level(), fields)); } } /// An agent name the store would refuse fails the mint before any store /// call, whatever this process's environment holds. The literal is the /// `_msg` that `hh-agent-identity-mint-failed` in /// `nix/host-modules/swarm-grafana.nix` matches. #[tokio::test] async fn a_failed_mint_logs_the_agent_and_the_error_on_the_line_the_alert_rule_matches() { let events = Events::default(); let _guard = tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); let Outcome::Failed(error) = mint_identity("a/b").await else { panic!("an illegal agent name cannot be minted"); }; let got = events.0.lock().expect("unpoisoned").clone(); let [(level, fields)] = got.as_slice() else { panic!("one event, got {got:?}"); }; assert_eq!(*level, tracing::Level::WARN); let field = |name: &str| { fields .iter() .find(|(n, _)| n == name) .map_or_else(|| panic!("no {name} in {fields:?}"), |(_, v)| v.clone()) }; assert_eq!(field("message"), "swarm jobq: mint_agent_identity failed"); assert_eq!(field("agent"), "a/b"); assert_eq!(field("error"), error); } }