diff --git a/swarm-controller-api/src/permissions.rs b/swarm-controller-api/src/permissions.rs index 0a052dc7..8b8e8030 100644 --- a/swarm-controller-api/src/permissions.rs +++ b/swarm-controller-api/src/permissions.rs @@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize}; #[serde(deny_unknown_fields)] pub struct Permissions { /// Permission names, `snake_case`: `messaging`, `meta`, `inbox`, - /// `execution`, `web_tools`, `manage_root_agent`. + /// `execution`, `web_tools`, `manage_root_agent`, `logs`, `metrics`. #[serde(default, skip_serializing_if = "Option::is_none")] #[cfg_attr( feature = "schema", diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index eeadf51d..0b6311a8 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -23,11 +23,13 @@ use problem_details::ProblemDetails; use serde::Serialize; use swarm_controller_api::permissions::Permissions; use swarm_queue_client::agent_permissions; +use swarm_secret_client::observability::Grant; use swarm_wire::agent_permissions::{ AgentPermissions, Import, ImportReply, hive_from_import_subject, import_subject_wildcard, }; use utoipa::ToSchema; +use crate::observability_grant::{self, grant_of, mirror, require_tolerant_hive}; use crate::routes::{AppState, error_problem, wanted_error_status}; use crate::schedules::{live, queue, read_or_default, refused, write, write_revision}; use crate::wanted::WantedWriter; @@ -102,6 +104,31 @@ fn logged(op: &'static str, agent: &str, problem: ProblemDetails) -> ProblemDeta problem } +/// Bring the store's mirror of `agent`'s `logs`/`metrics` grant in line with +/// a record that went from `before` to `after`, after the record is written. +/// A failure is an error to the caller, who repeats the request: `mirror` +/// rewrites whenever either record holds a grant. +async fn mirrored( + agent: &str, + before: &AgentPermissions, + after: &AgentPermissions, +) -> Result<(), ProblemDetails> { + let held = |v: &AgentPermissions| v.permissions.clone().unwrap_or_default(); + mirror(&observability_grant::Store, agent, &held(before), &held(after)) + .await + .map_err(|e| { + tracing::warn!(agent, error = %format!("{e:#}"), "agent permissions: mirroring the grant failed"); + error_problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("the permissions are stored, but mirroring logs/metrics to the secret store failed: {e:#}"), + ) + }) +} + +fn hive_names(state: &AppState) -> Vec { + state.hives.iter().map(|h| h.name.clone()).collect() +} + fn queue_problem(e: &anyhow::Error) -> ProblemDetails { tracing::warn!(error = %format!("{e:#}"), "agent permissions: queue failure"); error_problem(wanted_error_status(e), &format!("{e:#}")) @@ -185,6 +212,19 @@ pub(crate) async fn put_agent_permissions( &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), ))); } + if let Some(perms) = &req.permissions + && grant_of(perms) != Grant::default() + { + let (Some(wanted), hives) = (state.wanted.as_ref(), hive_names(&state)) else { + return Err(refuse(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + ))); + }; + require_tolerant_hive(&client, wanted, &hives, &agent) + .await + .map_err(|why| refuse(error_problem(StatusCode::CONFLICT, &why)))?; + } let store = open_or_create(&client).await?; for _ in 0..MAX_ATTEMPTS { let (current, revision, undecodable) = read_or_default::(&store, &agent) @@ -200,6 +240,7 @@ pub(crate) async fn put_agent_permissions( .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set agent permissions"); + mirrored(&agent, ¤t, &value).await?; return Ok(Json(view(value))); } } @@ -255,6 +296,7 @@ pub(crate) async fn delete_agent_permissions( continue; } tracing::info!(%agent, "cleared agent permissions"); + mirrored(&agent, ¤t, &kept).await?; return Ok(StatusCode::NO_CONTENT); } Err(queue_problem(&anyhow::anyhow!( @@ -271,10 +313,17 @@ pub(crate) async fn delete_agent_permissions( pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) { let cleared = async { let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?; - agent_permissions::open_or_create(&client) - .await? - .delete(agent) - .await?; + let store = agent_permissions::open_or_create(&client).await?; + let (current, _, _) = read_or_default::(&store, agent).await?; + store.delete(agent).await?; + // The mirrored grant goes with the record. + mirror( + &observability_grant::Store, + agent, + ¤t.permissions.unwrap_or_default(), + &[], + ) + .await?; anyhow::Ok(()) } .await; @@ -457,10 +506,15 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives return; }; tokio::spawn({ - let (store, wanted, hives) = (store.clone(), Arc::clone(&wanted), hives.clone()); + let (client, store, wanted, hives) = ( + client.clone(), + store.clone(), + Arc::clone(&wanted), + hives.clone(), + ); async move { // Retries without end, so it answers only once it has worked. - let _ = (|| seed_ruth_grant(&store, &wanted, &hives)) + let _ = (|| seed_ruth_grant(&client, &store, &wanted, &hives)) .retry(every_five_seconds()) .notify(|e: &anyhow::Error, _| { tracing::warn!( @@ -539,6 +593,7 @@ const SEEDED_AGENT: &str = "ruth"; /// `ruth` whose permissions were unset; this writes that grant into her key /// so she keeps her recovery mounts. Idempotent: a set key is never touched. async fn seed_ruth_grant( + client: &async_nats::Client, store: &async_nats::jetstream::kv::Store, wanted: &WantedWriter, hives: &[String], @@ -549,16 +604,27 @@ async fn seed_ruth_grant( declarations.push(d); } } - seed_into(store, &crate::agent_renewal::live_agents(&declarations)).await?; - Ok(()) + let declared = crate::agent_renewal::live_agents(&declarations); + if !declared.contains(SEEDED_AGENT) { + return Ok(()); + } + let tolerant = + observability_grant::hive_is_tolerant(client, wanted, hives, SEEDED_AGENT).await?; + seed_into(store, &declared, tolerant).await?; + // At every start, so a mirror that failed to write is retried. + let (current, _, _) = read_or_default::(store, SEEDED_AGENT).await?; + let held = current.permissions.unwrap_or_default(); + mirror(&observability_grant::Store, SEEDED_AGENT, &[], &held).await } -/// Write [`Permission::DEFAULT`] plus `manage_root_agent` under -/// [`SEEDED_AGENT`]'s key when `declared` holds her and her `permissions` -/// are unset (no key, a deleted key, or `{}`). `true` when it wrote. +/// Write [`Permission::DEFAULT`] plus `manage_root_agent`, and `logs` and +/// `metrics` when her hive is `tolerant`, under [`SEEDED_AGENT`]'s key when +/// `declared` holds her and her `permissions` are unset (no key, a deleted +/// key, or `{}`). `true` when it wrote. async fn seed_into( store: &async_nats::jetstream::kv::Store, declared: &BTreeSet, + tolerant: bool, ) -> anyhow::Result { if !declared.contains(SEEDED_AGENT) { return Ok(false); @@ -570,6 +636,9 @@ async fn seed_into( } let mut perms = Permission::DEFAULT.to_vec(); perms.push(Permission::ManageRootAgent); + if tolerant { + perms.extend([Permission::Logs, Permission::Metrics]); + } let value = AgentPermissions { permissions: Some(perms), ..AgentPermissions::default() @@ -616,7 +685,7 @@ async fn import_into( } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::{ Permissions, delete_agent_permissions, get_agent_permissions, import_into, put_agent_permissions, seed_into, @@ -701,7 +770,7 @@ mod tests { ); } - async fn published(client: &async_nats::Client, agent: &str) -> Option { + pub(crate) async fn published(client: &async_nats::Client, agent: &str) -> Option { let store = swarm_queue_client::agent_permissions::open_or_create(client) .await .expect("bucket"); @@ -945,6 +1014,91 @@ mod tests { assert_eq!(published(&client, "atlas").await.as_deref(), Some(older)); } + async fn publish_snapshot(client: &async_nats::Client, json: &str) { + swarm_queue_client::status::open_or_create(client) + .await + .expect("bucket") + .put("pr1ma", json.as_bytes().to_vec().into()) + .await + .expect("snapshot"); + } + + fn observing(perms: &[Permission]) -> Permissions { + Permissions { + permissions: Some(perms.to_vec()), + unrestricted_internet: None, + internet_log: None, + } + } + + /// `logs` and `metrics` are refused, naming the hive and writing nothing, + /// while the agent's hive has published no snapshot or one without the + /// capability. Control: a grant that names neither passes the same hive. + #[tokio::test] + async fn logs_and_metrics_need_a_hive_that_advertises_tolerant_decoding() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + + for perm in [Permission::Logs, Permission::Metrics] { + let none = put(&state, "atlas", observing(&[perm])) + .await + .expect_err("no snapshot"); + assert_eq!(none.status, Some(StatusCode::CONFLICT)); + assert!(none.detail.as_deref().unwrap_or_default().contains("pr1ma")); + } + + publish_snapshot(&client, r#"{"status":"ok","warnings":[]}"#).await; + let old = put(&state, "atlas", observing(&[Permission::Logs])) + .await + .expect_err("old hive"); + assert_eq!(old.status, Some(StatusCode::CONFLICT)); + assert_eq!(published(&client, "atlas").await, None, "nothing written"); + + let _ = put(&state, "atlas", observing(&[Permission::Messaging])) + .await + .expect("control: no log or metric grant"); + } + + /// With the capability advertised the gate passes: the record is written, + /// and the mirror (which needs the secret store, absent here) is what + /// fails. + #[tokio::test] + async fn an_advertising_hive_passes_the_gate() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + publish_snapshot(&client, r#"{"capabilities":["tolerant-permissions"]}"#).await; + + let problem = put(&state, "atlas", observing(&[Permission::Metrics])) + .await + .expect_err("no secret store to mirror to"); + assert_eq!(problem.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"permissions":["metrics"]}"#) + ); + } + + /// With her hive tolerant ruth is seeded with `logs` and `metrics`; with + /// an old hive she gets the set she always had. + #[tokio::test] + async fn the_seed_includes_logs_and_metrics_on_a_tolerant_hive() { + let nats = TestNats::start(); + let client = nats.client().await; + let store = swarm_queue_client::agent_permissions::open_or_create(&client) + .await + .expect("bucket"); + let declared = ["ruth".to_owned()].into(); + assert!(seed_into(&store, &declared, true).await.expect("seeded")); + assert_eq!( + published(&client, "ruth").await.as_deref(), + Some( + r#"{"permissions":["messaging","meta","inbox","execution","manage_root_agent","logs","metrics"]}"# + ) + ); + } + /// A name no hive declares, or a name that is not an identifier, is a 400 /// and writes nothing. #[tokio::test] @@ -1170,25 +1324,29 @@ mod tests { .expect("bucket"); let declared = ["atlas".to_owned(), "ruth".to_owned()].into(); - assert!(seed_into(&store, &declared).await.expect("seeded")); + assert!(seed_into(&store, &declared, false).await.expect("seeded")); assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED)); assert_eq!(published(&client, "atlas").await, None, "only ruth"); - assert!(!seed_into(&store, &declared).await.expect("idempotent")); + assert!( + !seed_into(&store, &declared, false) + .await + .expect("idempotent") + ); store.delete("ruth").await.expect("delete"); - assert!(seed_into(&store, &declared).await.expect("seeded")); + assert!(seed_into(&store, &declared, false).await.expect("seeded")); assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED)); store .put("ruth", b"{}".to_vec().into()) .await .expect("unset"); - assert!(seed_into(&store, &declared).await.expect("seeded")); + assert!(seed_into(&store, &declared, false).await.expect("seeded")); assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED)); } /// A ruth key the swarm already set, even to nothing, is left as it is; - /// so is one this build cannot decode. + /// so is one holding only a name this build does not know. #[tokio::test] async fn the_seed_leaves_a_set_ruth_key_alone() { let nats = TestNats::start(); @@ -1206,7 +1364,10 @@ mod tests { .put("ruth", set.as_bytes().to_vec().into()) .await .expect("set"); - assert!(!seed_into(&store, &declared).await.expect("read"), "{set}"); + assert!( + !seed_into(&store, &declared, false).await.expect("read"), + "{set}" + ); assert_eq!(published(&client, "ruth").await.as_deref(), Some(set)); } } @@ -1222,12 +1383,12 @@ mod tests { .expect("bucket"); let declared = ["atlas".to_owned()].into(); - assert!(!seed_into(&store, &declared).await.expect("read")); + assert!(!seed_into(&store, &declared, false).await.expect("read")); assert_eq!(published(&client, "ruth").await, None); assert_eq!(published(&client, "atlas").await, None); let declared = ["atlas".to_owned(), "ruth".to_owned()].into(); - assert!(seed_into(&store, &declared).await.expect("seeded")); + assert!(seed_into(&store, &declared, false).await.expect("seeded")); assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED)); } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index dcc4d468..bac04742 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -64,6 +64,7 @@ mod meta_fanout; mod meta_flake; mod notify_flows; mod notify_router_api; +mod observability_grant; mod otel_http_client; mod proxy_credential; mod pulls; diff --git a/swarm-controller/src/observability_grant.rs b/swarm-controller/src/observability_grant.rs new file mode 100644 index 00000000..7f9cedcc --- /dev/null +++ b/swarm-controller/src/observability_grant.rs @@ -0,0 +1,248 @@ +//! The `logs` and `metrics` permissions beyond the queue's permissions bucket: +//! which hives may be granted them, and the mirror of the grant that the store +//! gateways read. +//! +//! A hive that cannot skip a permission name it does not know would fail to +//! decode a record that holds one, so the controller grants them only on a hive +//! whose status snapshot advertises `swarm_wire::hive_status::TOLERANT_PERMISSIONS`. +//! `swarm-logs-auth` has no queue access, so each grant is also written to +//! `swarm_secret_client::observability::grant_path`; a leaf that is absent is no +//! grant. + +use std::future::Future; + +use anyhow::{Context as _, Result}; +use hive_types::Permission; +use swarm_secret_client::observability::{Grant, grant_path}; +use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS}; + +use crate::wanted::WantedWriter; + +/// The grant `perms` hold. +pub(crate) fn grant_of(perms: &[Permission]) -> Grant { + Grant { + logs: perms.contains(&Permission::Logs), + metrics: perms.contains(&Permission::Metrics), + } +} + +/// Where a mirrored grant is written. A test stands in for the store. +pub(crate) trait GrantSink { + fn write(&self, agent: &str, grant: Grant) -> impl Future> + Send; +} + +/// The swarm secret store, under the controller's own identity. +pub(crate) struct Store; + +impl GrantSink for Store { + async fn write(&self, agent: &str, grant: Grant) -> Result<()> { + let path = grant_path(agent)?; + crate::store::connect() + .await + .context("logging in to the swarm secret store")? + .write(&path, &grant) + .await + .with_context(|| format!("writing {path}")) + } +} + +/// Bring the mirror in line when `agent`'s permissions go from `before` to +/// `after`. Nothing is written when neither holds `logs` or `metrics`, which +/// leaves the leaf absent. +/// +/// # Errors +/// The write failed; the responder then keeps the grant it last read, or none. +pub(crate) async fn mirror( + sink: &impl GrantSink, + agent: &str, + before: &[Permission], + after: &[Permission], +) -> Result<()> { + if grant_of(before) == Grant::default() && grant_of(after) == Grant::default() { + return Ok(()); + } + sink.write(agent, grant_of(after)).await +} + +/// `Ok` when `hive`'s snapshot advertises tolerant decoding. +/// +/// # Errors +/// Naming the hive, when it has published nothing or does not advertise it. +pub(crate) fn check_capability( + hive: &str, + snapshot: Option<&serde_json::Value>, +) -> Result<(), String> { + match snapshot { + Some(s) if Capabilities::of_snapshot(s).has(TOLERANT_PERMISSIONS) => Ok(()), + Some(_) => Err(format!( + "hive {hive} does not advertise {TOLERANT_PERMISSIONS}: update it first, \ + then grant logs or metrics" + )), + None => Err(format!( + "hive {hive} has published no status snapshot, so it is not known to \ + advertise {TOLERANT_PERMISSIONS}" + )), + } +} + +/// The hive whose declaration keeps `agent` alive. +async fn hive_of(wanted: &WantedWriter, hives: &[String], agent: &str) -> Option { + for hive in hives { + if let Ok(d) = wanted.view(hive).await + && crate::agent_renewal::live_agents(d.as_slice()).contains(agent) + { + return Some(hive.clone()); + } + } + None +} + +/// `hive`'s last status snapshot, `None` when it has published none. +async fn snapshot(client: &async_nats::Client, hive: &str) -> Result> { + swarm_queue_client::ensure_connected(client)?; + let store = swarm_queue_client::status::open_or_create(client).await?; + let Some(entry) = store.get(hive).await? else { + return Ok(None); + }; + Ok(serde_json::from_slice(&entry).ok()) +} + +/// Whether the hive running `agent` advertises tolerant decoding, waiting +/// (an error, for a retrying caller) on a hive that has published no snapshot +/// yet. `false` when no hive declares the agent. +/// +/// # Errors +/// The queue could not be read, or the hive has not published a snapshot. +pub(crate) async fn hive_is_tolerant( + client: &async_nats::Client, + wanted: &WantedWriter, + hives: &[String], + agent: &str, +) -> Result { + let Some(hive) = hive_of(wanted, hives, agent).await else { + return Ok(false); + }; + let Some(snapshot) = snapshot(client, &hive).await? else { + anyhow::bail!("hive {hive} has published no status snapshot yet"); + }; + Ok(check_capability(&hive, Some(&snapshot)).is_ok()) +} + +/// `Ok` when the hive running `agent` advertises tolerant decoding, so the +/// agent's record may name `logs` or `metrics`. +/// +/// # Errors +/// The reason: no hive declares the agent, the hive's snapshot is missing or +/// lacks the capability, or the queue could not be read. +pub(crate) async fn require_tolerant_hive( + client: &async_nats::Client, + wanted: &WantedWriter, + hives: &[String], + agent: &str, +) -> Result<(), String> { + let hive = hive_of(wanted, hives, agent) + .await + .ok_or_else(|| format!("no hive declares agent {agent}"))?; + require_tolerant(client, &hive).await +} + +/// `Ok` when `hive`'s last status snapshot advertises tolerant decoding. +/// +/// # Errors +/// The reason, naming the hive. +pub(crate) async fn require_tolerant( + client: &async_nats::Client, + hive: &str, +) -> Result<(), String> { + let snapshot = snapshot(client, hive) + .await + .map_err(|e| format!("reading hive {hive}'s status: {e:#}"))?; + check_capability(hive, snapshot.as_ref()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + #[derive(Default)] + struct Leaves(Mutex>); + + impl GrantSink for &Leaves { + async fn write(&self, agent: &str, grant: Grant) -> Result<()> { + self.0.lock().unwrap().push((agent.to_owned(), grant)); + Ok(()) + } + } + + struct Down; + + impl GrantSink for Down { + async fn write(&self, _: &str, _: Grant) -> Result<()> { + anyhow::bail!("store unreachable") + } + } + + #[tokio::test] + async fn a_grant_and_its_revocation_are_written() { + let leaves = Leaves::default(); + let both = [Permission::Messaging, Permission::Logs, Permission::Metrics]; + mirror(&&leaves, "atlas", &[], &both).await.unwrap(); + mirror(&&leaves, "atlas", &both, &[Permission::Messaging]) + .await + .unwrap(); + let written = leaves.0.lock().unwrap().clone(); + let yes = Grant { + logs: true, + metrics: true, + }; + assert_eq!( + written, + [ + ("atlas".to_owned(), yes), + ("atlas".to_owned(), Grant::default()) + ] + ); + } + + /// An agent that never held either, before or after, gets no leaf; one + /// that holds only one of them gets exactly that one. + #[tokio::test] + async fn only_the_permissions_held_are_mirrored() { + let leaves = Leaves::default(); + let plain = [Permission::Messaging]; + mirror(&&leaves, "atlas", &plain, &plain).await.unwrap(); + assert!(leaves.0.lock().unwrap().is_empty()); + mirror(&&leaves, "atlas", &plain, &[Permission::Metrics]) + .await + .unwrap(); + let written = leaves.0.lock().unwrap().clone(); + assert_eq!(written.len(), 1); + assert_eq!((written[0].1.logs, written[0].1.metrics), (false, true)); + } + + #[tokio::test] + async fn a_failed_write_is_an_error() { + let held = [Permission::Logs]; + assert!(mirror(&Down, "atlas", &[], &held).await.is_err()); + // Control: nothing to write is not a failure, whatever the store does. + assert!(mirror(&Down, "atlas", &[], &[]).await.is_ok()); + } + + #[test] + fn a_hive_without_the_capability_is_refused_by_name() { + let old = serde_json::json!({"status": "ok", "warnings": []}); + let new = serde_json::json!({"capabilities": [TOLERANT_PERMISSIONS]}); + let err = check_capability("pr1ma", Some(&old)).unwrap_err(); + assert!( + err.contains("pr1ma") && err.contains(TOLERANT_PERMISSIONS), + "{err}" + ); + assert!( + check_capability("pr1ma", None) + .unwrap_err() + .contains("pr1ma") + ); + assert!(check_capability("pr1ma", Some(&new)).is_ok()); + } +} diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index 0272a5d4..e44c7131 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -10,9 +10,11 @@ use axum::{ }; use serde::{Deserialize, Serialize}; use swarm_controller_api::agents::{CreateAgentRequest, CreateAgentResponse}; +use swarm_secret_client::observability::Grant; use utoipa::ToSchema; use crate::jobq::{SwarmNodeKind, SwarmResourceKind, declare_agent_job}; +use crate::observability_grant::{self, grant_of, mirror, require_tolerant}; use crate::routes::hives::HiveEntry; use crate::routes::{AppState, StatusUnavailable, error_problem, wanted_error_status}; use crate::{ @@ -623,26 +625,18 @@ pub(crate) async fn create_agent( // Checked after the roster, so a mistyped hive stays a 400 while the queue // is down. let settings_queue = settings.queue(&state).await?; + let requested = settings.permissions.permissions.clone().unwrap_or_default(); + if grant_of(&requested) != Grant::default() { + require_tolerant(&settings_queue, &hive) + .await + .map_err(|why| error_problem(axum::http::StatusCode::CONFLICT, &why))?; + } // Agent names are one swarm-wide namespace: identity, forge user, store // path and matrix id all carry the bare name. The same name on the same // hive is that agent being re-created, and goes through. let _gate = state.create_gate.lock().await; - let elsewhere = placements_elsewhere( - &agent, - &hive, - || queued_placements_now(&state), - || declarations_elsewhere(&state, &agent, &hive), - ) - .await?; - if !elsewhere.is_empty() { - let detail = format!( - "agent name {agent:?} is already placed on hive {} — agent names are unique across \ - the swarm; destroy it there first to move it, or choose another name", - elsewhere.join(", ") - ); - return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail)); - } + refuse_if_placed_elsewhere(&state, &agent, &hive).await?; // Stored ahead of the job graph, not once the agent is declared: the hive // reads these by agent name when it first runs the agent. A failure from // here on puts back what was stored, so a refused creation leaves no @@ -657,6 +651,22 @@ pub(crate) async fn create_agent( settings.agents_md.as_deref(), ) .await?; + // The record this create replaced is what the mirror may still describe. + // A create that names no list keeps the stored one, and its mirror with it. + let earlier = stored + .permissions + .as_ref() + .filter(|_| settings.permissions.permissions.is_some()) + .and_then(|replaced| replaced.previous()) + .and_then(|record| record.permissions.clone()) + .unwrap_or_default(); + if let Err(e) = mirror(&observability_grant::Store, &agent, &earlier, &requested).await { + stored.roll_back(&settings_queue, &agent).await; + return Err(error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("mirroring logs/metrics to the secret store failed: {e:#}"), + )); + } let queued = { let mut sched = state .jobq @@ -670,6 +680,8 @@ pub(crate) async fn create_agent( Ok(ids) => ids, Err(e) => { stored.roll_back(&settings_queue, &agent).await; + // Best effort, like the roll back: the mirror follows the record. + let _ = mirror(&observability_grant::Store, &agent, &requested, &earlier).await; return Err(error_problem( axum::http::StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), @@ -685,6 +697,30 @@ pub(crate) async fn create_agent( })) } +/// A 409 when `agent` is already placed on a hive other than `hive`. +async fn refuse_if_placed_elsewhere( + state: &AppState, + agent: &str, + hive: &str, +) -> Result<(), problem_details::ProblemDetails> { + let elsewhere = placements_elsewhere( + agent, + hive, + || queued_placements_now(state), + || declarations_elsewhere(state, agent, hive), + ) + .await?; + if elsewhere.is_empty() { + return Ok(()); + } + let detail = format!( + "agent name {agent:?} is already placed on hive {} — agent names are unique across \ + the swarm; destroy it there first to move it, or choose another name", + elsewhere.join(", ") + ); + Err(error_problem(axum::http::StatusCode::CONFLICT, &detail)) +} + /// The settings a `POST /api/agents` body gives, as they are stored. struct CreateSettings { limits: Option, @@ -1552,6 +1588,35 @@ mod tests { assert_eq!(node_count(&state.jobq), 0, "nothing may be queued"); } + /// A create naming `logs` or `metrics` on a hive that does not advertise + /// tolerant decoding is a 409 that stores and queues nothing. Control: the + /// same create without them is accepted by that hive. + #[tokio::test] + async fn a_create_granting_logs_needs_a_tolerant_hive() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let create = |perms: &str| { + let req: CreateAgentRequest = serde_json::from_str(&format!( + r#"{{"name":"atlas","hive":"pr1ma","permissions":[{perms}]}}"# + )) + .expect("body parses"); + super::create_agent(axum::extract::State(state.clone()), axum::Json(req)) + }; + let err = create(r#""messaging","logs""#).await.expect_err("refused"); + assert_eq!( + axum::response::IntoResponse::into_response(err).status(), + axum::http::StatusCode::CONFLICT + ); + assert_eq!( + crate::agent_permissions::tests::published(&client, "atlas").await, + None + ); + let _ = create(r#""messaging""#) + .await + .expect("control: no log grant"); + } + /// `state_with_roster` with a second hive, `sec0nd`. async fn state_with_two_hives() -> ( crate::test_nats::TestNats, diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index 0d014417..94daf218 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -503,6 +503,11 @@ impl Replaced { Self { previous, written } } + /// The value the write replaced, `None` when the key held none. + pub(crate) fn previous(&self) -> Option<&T> { + self.previous.as_ref() + } + /// Put `previous` back, or remove the key if it held none, against the /// revision the write made. `Ok(false)` when another write has landed /// since: that one is newer than what is being undone and stays. diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index bfedbf5e..4df2383c 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -358,7 +358,8 @@ struct PermissionsSetArgs { /// Agent name. A hive must declare it. agent: String, /// The permissions to grant, comma-separated: any of `messaging`, - /// `meta`, `inbox`, `execution`, `web_tools`, `manage_root_agent`. + /// `meta`, `inbox`, `execution`, `web_tools`, `manage_root_agent`, `logs`, + /// `metrics`. /// `--permissions ''` grants none. #[arg( long,