From 329d556ab97810c9429787e74b819bd3cc86568c Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 1 Nov 2026 18:27:55 +0100 Subject: [PATCH] order grant mirror writes by direction, re-mirror every five minutes, gate on the agent's own harness --- hive-agent/src/swarm_agent_state.rs | 27 +- swarm-controller/src/agent_login.rs | 57 ++- swarm-controller/src/agent_permissions.rs | 493 +++++++++++++++++--- swarm-controller/src/observability_grant.rs | 322 ++++++++++--- swarm-wire/src/hive_status.rs | 18 +- 5 files changed, 796 insertions(+), 121 deletions(-) diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 0c41e852..4c538ced 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -35,6 +35,7 @@ use tokio::time::{Interval, MissedTickBehavior}; use swarm_wire::agent_inference::Effective; use swarm_wire::agent_login::Login; +use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS}; use swarm_wire::wanted::AgentState; use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; @@ -100,6 +101,10 @@ struct AgentStateMsg { /// `needs_login`, from this process's own [`LoginState`], so always set. #[serde(flatten)] login: Login, + /// What this harness can do, so swarm-controller grants an agent a + /// permission only when the harness running it can decode it. + #[serde(flatten)] + capabilities: Capabilities, } /// What this process can honestly report about the agent-state axis. @@ -140,6 +145,13 @@ fn login(state: LoginState) -> Login { /// Build the current header from the bus, the login state and the pause /// marker. +/// The capabilities this build reports. +fn capabilities() -> Capabilities { + Capabilities { + capabilities: vec![TOLERANT_PERMISSIONS.to_owned()], + } +} + fn snapshot(bus: &Bus, login_state: &Mutex) -> AgentStateMsg { let (turn_state, since_unix) = bus.state_snapshot(); let model = bus.model(); @@ -158,6 +170,7 @@ fn snapshot(bus: &Bus, login_state: &Mutex) -> AgentStateMsg { ctx_usage: bus.last_ctx_usage(), cost_usage: bus.last_cost_usage(), login: login(*login_state.lock().unwrap_or_else(PoisonError::into_inner)), + capabilities: capabilities(), } } @@ -332,12 +345,14 @@ async fn publish( #[cfg(test)] mod tests { use super::{ - AgentStateMsg, REFRESH_INTERVAL, Trigger, due, login, next_trigger, refresh_timer, snapshot, + AgentStateMsg, REFRESH_INTERVAL, Trigger, capabilities, due, login, next_trigger, + refresh_timer, snapshot, }; use crate::events::{BusEvent, LiveEvent, TurnState}; use crate::login::LoginState; use std::sync::Mutex; use swarm_wire::agent_inference::{Effective, Source}; + use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS}; use swarm_wire::wanted::AgentState; use tokio::sync::broadcast; use tokio::time::{Duration, Instant}; @@ -387,6 +402,7 @@ mod tests { "cache_creation_input_tokens": 88, }))), login: login(LoginState::NeedsLogin), + capabilities: capabilities(), } } @@ -406,6 +422,7 @@ mod tests { keys, [ "agent_state", + "capabilities", "context_window_tokens", "cost_usage", "ctx_usage", @@ -505,6 +522,14 @@ mod tests { } } + /// swarm-controller reads the harness's capabilities back out of a header. + #[test] + fn the_header_advertises_tolerant_permissions() { + let bytes = serde_json::to_vec(&msg()).expect("serialises"); + let decoded: Capabilities = serde_json::from_slice(&bytes).expect("decodes"); + assert!(decoded.has(TOLERANT_PERMISSIONS)); + } + /// A login landing, or a logout, changes the header the next bus event /// rebuilds, so it goes out without waiting for a refresh. #[test] diff --git a/swarm-controller/src/agent_login.rs b/swarm-controller/src/agent_login.rs index 708c752d..103bdb85 100644 --- a/swarm-controller/src/agent_login.rs +++ b/swarm-controller/src/agent_login.rs @@ -6,6 +6,11 @@ //! when it changes and re-send it once a minute, and the queue keeps none, so //! an agent has no entry for up to a minute after this daemon starts — the //! same gap [`crate::inference::Reported`] has on the same subject. +//! +//! The same header carries the harness's `capabilities` +//! (`swarm_wire::hive_status`), which [`Reported::advertises`] answers from: +//! the controller grants `logs` or `metrics` only to an agent whose running +//! harness reports it can decode them. use std::collections::HashMap; use std::sync::{Arc, Mutex, PoisonError}; @@ -13,11 +18,14 @@ use std::time::Duration; use futures_util::StreamExt as _; use swarm_wire::agent_login::Login; +use swarm_wire::hive_status::Capabilities; /// The latest login flag each agent reported, by agent. #[derive(Debug, Default)] pub(crate) struct Reported { latest: Mutex>, + /// The capabilities each agent's latest header listed. + capabilities: Mutex>, } impl Reported { @@ -51,10 +59,22 @@ impl Reported { /// Keep one header's login flag. A header without it, from an agent that /// does not report it, removes what was kept. - fn record(&self, subject: &str, payload: &[u8]) { + pub(crate) fn record(&self, subject: &str, payload: &[u8]) { let Some(agent) = swarm_queue_client::agent_from_state_subject(subject) else { return; }; + let mut capabilities = self + .capabilities + .lock() + .unwrap_or_else(PoisonError::into_inner); + match serde_json::from_slice::(payload) { + Ok(listed) => { + capabilities.insert(agent.to_owned(), listed); + } + Err(_) => { + capabilities.remove(agent); + } + } let mut latest = self.latest.lock().unwrap_or_else(PoisonError::into_inner); match serde_json::from_slice::(payload) { Ok(Login { @@ -72,6 +92,17 @@ impl Reported { } } + /// Whether `agent`'s latest header lists `capability`. `false` for an + /// agent that has sent none since this daemon started, or whose harness + /// predates the field. + pub(crate) fn advertises(&self, agent: &str, capability: &str) -> bool { + self.capabilities + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(agent) + .is_some_and(|c| c.has(capability)) + } + /// `agent`'s latest flag; `None` when no header since this daemon /// started has carried one. pub(crate) fn get(&self, agent: &str) -> Option { @@ -142,4 +173,28 @@ mod tests { assert_eq!(reported.get("iris"), Some(true)); assert_eq!(reported.get("atlas"), None); } + + /// Only a header that lists the capability counts: one from a harness that + /// predates the field, a garbled one and another agent's do not. + #[test] + fn an_agent_advertises_what_its_latest_header_lists() { + let reported = Reported::default(); + let subject = swarm_queue_client::agent_state_subject("iris"); + let with = |caps: &[&str]| { + let mut value: serde_json::Value = serde_json::from_slice(&header(None)).unwrap(); + value["capabilities"] = serde_json::json!(caps); + serde_json::to_vec(&value).unwrap() + }; + let tolerant = swarm_wire::hive_status::TOLERANT_PERMISSIONS; + reported.record(&subject, &header(Some(false))); + assert!(!reported.advertises("iris", tolerant), "old harness"); + reported.record(&subject, &with(&["from-the-future", tolerant])); + assert!(reported.advertises("iris", tolerant)); + assert!(!reported.advertises("atlas", tolerant), "another agent"); + reported.record(&subject, &with(&["from-the-future"])); + assert!(!reported.advertises("iris", tolerant), "latest header wins"); + reported.record(&subject, &with(&[tolerant])); + reported.record(&subject, b"not json"); + assert!(!reported.advertises("iris", tolerant), "garbled clears it"); + } } diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index 0b6311a8..9fc845d3 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -8,7 +8,7 @@ //! restarts it when `manage_root_agent` changes. An unset field leaves the //! agent on the compiled-in default; an empty list grants nothing. -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use std::time::Duration; @@ -29,7 +29,9 @@ use swarm_wire::agent_permissions::{ }; use utoipa::ToSchema; -use crate::observability_grant::{self, grant_of, mirror, require_tolerant_hive}; +use crate::observability_grant::{ + self, GrantLeaves, adds, grant_of, mirror, require_tolerant_agent, revokes, +}; 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; @@ -104,25 +106,51 @@ 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( +/// The permissions a record holds. +fn held(record: &AgentPermissions) -> Vec { + record.permissions.clone().unwrap_or_default() +} + +/// The error for a mirror write that failed. `stored` says which side of the +/// change landed first: a revoke writes the leaf before the record, so the +/// record is untouched and the revoke did not happen; a grant writes the +/// record first, so the gateway keeps denying until the request is repeated +/// or the next pass of [`spawn`] writes the leaf. +fn mirror_problem(agent: &str, e: &anyhow::Error, stored: bool) -> ProblemDetails { + tracing::warn!(agent, error = %format!("{e:#}"), stored, "agent permissions: mirroring the grant failed"); + let outcome = if stored { + "the permissions are stored, but the log and metric gateways deny until the grant is mirrored" + } else { + "nothing was changed" + }; + error_problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("mirroring logs/metrics to the secret store failed; {outcome}: {e:#}"), + ) +} + +/// A 409 unless `agent` may be granted `logs` or `metrics`, see +/// [`require_tolerant_agent`]. +async fn require_tolerant_for( + state: &AppState, + client: &async_nats::Client, 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:#}"), - ) - }) + let Some(wanted) = state.wanted.as_ref() else { + return Err(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + )); + }; + require_tolerant_agent( + client, + wanted, + &hive_names(state), + state.agent_login.as_deref(), + agent, + ) + .await + .map_err(|why| error_problem(StatusCode::CONFLICT, &why)) } fn hive_names(state: &AppState) -> Vec { @@ -197,6 +225,22 @@ pub(crate) async fn put_agent_permissions( State(state): State, Path(name): Path, req: Result, JsonRejection>, +) -> Result, ProblemDetails> { + put_with(&observability_grant::Store, state, name, req).await +} + +/// [`put_agent_permissions`] against `leaves`. +/// +/// The record and the leaf are written in the order that fails towards +/// denying: a change that drops `logs` or `metrics` writes the leaf first, so a +/// failed write leaves both granted and the request an honest error; a change +/// that adds one writes the record first, so a failed write leaves the gateway +/// denying. +async fn put_with( + leaves: &impl GrantLeaves, + state: AppState, + name: String, + req: Result, JsonRejection>, ) -> Result, ProblemDetails> { let refuse = |p| logged("put", &name, p); let Json(req) = @@ -212,19 +256,6 @@ 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) @@ -234,13 +265,29 @@ pub(crate) async fn put_agent_permissions( if let Some(why) = log_without_restriction(&value) { return Err(refuse(error_problem(StatusCode::BAD_REQUEST, why))); } + let (before, after) = (held(¤t), held(&value)); + if adds(&before, &after) { + require_tolerant_for(&state, &client, &agent) + .await + .map_err(refuse)?; + } + let revoking = revokes(&before, &after); + if revoking { + mirror(leaves, &agent, &before, &after) + .await + .map_err(|e| mirror_problem(&agent, &e, false))?; + } if (current == value && !undecodable) || write(&store, &agent, &value, revision) .await .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set agent permissions"); - mirrored(&agent, ¤t, &value).await?; + if !revoking { + mirror(leaves, &agent, &before, &after) + .await + .map_err(|e| mirror_problem(&agent, &e, true))?; + } return Ok(Json(view(value))); } } @@ -267,6 +314,16 @@ pub(crate) async fn put_agent_permissions( pub(crate) async fn delete_agent_permissions( State(state): State, Path(name): Path, +) -> Result { + delete_with(&observability_grant::Store, state, name).await +} + +/// [`delete_agent_permissions`] against `leaves`. A clear that drops `logs` or +/// `metrics` writes the leaf first, as [`put_with`] does for a revoke. +async fn delete_with( + leaves: &impl GrantLeaves, + state: AppState, + name: String, ) -> Result { let refuse = |p| logged("delete", &name, p); let agent = agent_name(&name).map_err(|r| refuse(refused(r)))?; @@ -284,6 +341,9 @@ pub(crate) async fn delete_agent_permissions( internet_log: current.internet_log, ..AgentPermissions::default() }; + mirror(leaves, &agent, &held(¤t), &held(&kept)) + .await + .map_err(|e| mirror_problem(&agent, &e, false))?; if kept == AgentPermissions::default() { store .delete(&agent) @@ -296,7 +356,6 @@ 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!( @@ -309,21 +368,20 @@ pub(crate) async fn delete_agent_permissions( /// key left here would carry over to the next agent of that name. /// /// Like the credential revoke it runs beside, a failure is logged and does -/// not stop the destroy; declaring the agent destroyed again re-runs it. +/// not stop the destroy; declaring the agent destroyed again re-runs it. The +/// mirrored grant goes first: if that write fails the record stays too, so the +/// two agree. pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) { + clear_with(&observability_grant::Store, state, hive, agent).await; +} + +async fn clear_with(leaves: &impl GrantLeaves, state: &AppState, hive: &str, agent: &str) { let cleared = async { let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?; let store = agent_permissions::open_or_create(&client).await?; let (current, _, _) = read_or_default::(&store, agent).await?; + mirror(leaves, agent, &held(¤t), &[]).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; @@ -525,6 +583,27 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives .await; } }); + tokio::spawn({ + let (store, wanted, hives) = (store.clone(), Arc::clone(&wanted), hives.clone()); + async move { + // The seed's own retry has finished before the first pass, so + // a seed that waits on a hive does not race it. + loop { + match remirror(&store, &wanted, &hives).await { + Ok(0) => {} + Ok(n) => tracing::warn!( + n, + "agent permissions: corrected drifted log/metric grants" + ), + Err(e) => tracing::warn!( + error = %format!("{e:#}"), + "agent permissions: re-mirroring the log/metric grants failed" + ), + } + tokio::time::sleep(observability_grant::REMIRROR_EVERY).await; + } + } + }); let subject = import_subject_wildcard(); loop { match client.subscribe(subject.clone()).await { @@ -551,6 +630,41 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives }); } +/// The grant each agent's record holds: every agent with a key, and every live +/// agent without one (the empty grant), so a leaf left for an agent whose +/// record is gone is found too. +async fn desired_grants( + store: &async_nats::jetstream::kv::Store, + wanted: &WantedWriter, + hives: &[String], +) -> anyhow::Result> { + let mut desired = BTreeMap::new(); + let mut keys = store.keys().await?; + while let Some(key) = keys.next().await { + let key = key?; + let (record, _, _) = read_or_default::(store, &key).await?; + desired.insert(key, grant_of(&held(&record))); + } + for agent in crate::subagent_term::live_agents(wanted, hives).await { + desired.entry(agent).or_default(); + } + Ok(desired) +} + +/// One pass of [`observability_grant::reconcile`] over every agent: the leaf +/// the gateways read becomes what the record says, so a grant whose mirror +/// write failed, or a revoke whose record write failed after its leaf, ends +/// consistent. `Ok` is the number of leaves corrected. +async fn remirror( + store: &async_nats::jetstream::kv::Store, + wanted: &WantedWriter, + hives: &[String], +) -> anyhow::Result { + let desired = desired_grants(store, wanted, hives).await?; + let leaves = observability_grant::Connected::login().await?; + observability_grant::reconcile(&leaves, &desired).await +} + async fn answer_import( store: &async_nats::jetstream::kv::Store, wanted: &WantedWriter, @@ -690,10 +804,12 @@ pub(crate) mod tests { Permissions, delete_agent_permissions, get_agent_permissions, import_into, put_agent_permissions, seed_into, }; + use crate::observability_grant::fake::Leaves; use crate::test_nats::TestNats; use axum::extract::{Path, State}; use axum::http::StatusCode; use hive_types::Permission; + use swarm_secret_client::observability::Grant; use swarm_wire::agent_permissions::{Import, ImportReply}; fn stored(json: &str) -> swarm_wire::agent_permissions::AgentPermissions { @@ -1060,23 +1176,296 @@ pub(crate) mod tests { .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. + const TOLERANT_HIVE: &str = r#"{"capabilities":["tolerant-permissions"]}"#; + const LOGS_AND_METRICS: Grant = Grant { + logs: true, + metrics: true, + }; + + /// `state` with `agent_login` holding what each `(agent, capabilities)` + /// last reported. + fn with_reports( + state: crate::routes::AppState, + reports: &[(&str, &[&str])], + ) -> crate::routes::AppState { + let reported = crate::agent_login::Reported::default(); + for (agent, capabilities) in reports { + reported.record( + &swarm_queue_client::agent_state_subject(agent), + serde_json::json!({ "capabilities": capabilities }) + .to_string() + .as_bytes(), + ); + } + crate::routes::AppState { + agent_login: Some(std::sync::Arc::new(reported)), + ..state + } + } + + /// A state whose hive and whose atlas both advertise tolerant decoding. + async fn tolerant(client: &async_nats::Client) -> crate::routes::AppState { + let state = crate::agent_limits::tests::declared(client).await; + publish_snapshot(client, TOLERANT_HIVE).await; + with_reports(state, &[("atlas", &["tolerant-permissions"])]) + } + + async fn put_via( + leaves: &Leaves, + state: &crate::routes::AppState, + perms: &[Permission], + ) -> Result, problem_details::ProblemDetails> { + super::put_with( + leaves, + state.clone(), + "atlas".to_owned(), + Ok(axum::Json(observing(perms))), + ) + .await + } + + async fn record(client: &async_nats::Client, perms: &str) { + swarm_queue_client::agent_permissions::open_or_create(client) + .await + .expect("bucket") + .put( + "atlas", + format!(r#"{{"permissions":{perms}}}"#).into_bytes().into(), + ) + .await + .expect("record"); + } + + /// A tolerant hive is not enough: the agent's running harness has to have + /// reported the capability too. A harness that predates it, one that has + /// not reported yet and a hive that never published each refuse the grant. #[tokio::test] - async fn an_advertising_hive_passes_the_gate() { + async fn the_agents_own_harness_must_report_the_capability() { 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; + publish_snapshot(&client, TOLERANT_HIVE).await; + let leaves = Leaves::default(); - let problem = put(&state, "atlas", observing(&[Permission::Metrics])) + for (reports, why) in [ + (&[][..], "no report yet"), + (&[("atlas", &[][..])][..], "a harness that lists none"), + ( + &[("iris", &["tolerant-permissions"][..])][..], + "another agent", + ), + ] { + let state = with_reports(state.clone(), reports); + let err = put_via(&leaves, &state, &[Permission::Logs]) + .await + .expect_err(why); + assert_eq!(err.status, Some(StatusCode::CONFLICT), "{why}"); + assert!(err.detail.as_deref().unwrap_or_default().contains("atlas")); + assert_eq!( + published(&client, "atlas").await, + None, + "{why}: nothing written" + ); + } + assert_eq!(leaves.get("atlas"), None); + + // Control: with both reporting, the same request is accepted. + let state = with_reports(state, &[("atlas", &["tolerant-permissions"])]); + let _ = put_via(&leaves, &state, &[Permission::Logs]) .await - .expect_err("no secret store to mirror to"); - assert_eq!(problem.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); + .expect("granted"); + assert_eq!(leaves.get("atlas").map(|g| g.logs), Some(true)); + } + + /// Only a change that adds a grant needs the gate: an agent that holds + /// `logs` can have its other permissions changed while stopped. + #[tokio::test] + async fn keeping_a_grant_needs_no_report() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let leaves = Leaves::default(); + leaves.put("atlas", LOGS_AND_METRICS); + record(&client, r#"["logs","metrics"]"#).await; + + let _ = put_via( + &leaves, + &state, + &[Permission::Logs, Permission::Metrics, Permission::WebTools], + ) + .await + .expect("kept"); + assert_eq!(leaves.get("atlas"), Some(LOGS_AND_METRICS)); + } + + /// A revoke writes the leaf first. With the store down it fails as an + /// error and changes nothing: the record and the leaf both still grant. + #[tokio::test] + async fn a_revoke_that_cannot_reach_the_store_changes_nothing() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let leaves = Leaves::default(); + leaves.put("atlas", LOGS_AND_METRICS); + record(&client, r#"["logs","metrics"]"#).await; + + leaves.fail(true); + let err = put_via(&leaves, &state, &[Permission::Messaging]) + .await + .expect_err("store down"); + assert_eq!(err.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); assert_eq!( published(&client, "atlas").await.as_deref(), - Some(r#"{"permissions":["metrics"]}"#) + Some(r#"{"permissions":["logs","metrics"]}"#), + "the record still grants" + ); + leaves.fail(false); + assert_eq!( + leaves.get("atlas"), + Some(LOGS_AND_METRICS), + "and so does the leaf" + ); + + // Control: with the store back the same request revokes both. + let _ = put_via(&leaves, &state, &[Permission::Messaging]) + .await + .expect("revoked"); + assert_eq!(leaves.get("atlas"), Some(Grant::default())); + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"permissions":["messaging"]}"#) + ); + } + + /// A clear that drops a grant is ordered the same way. + #[tokio::test] + async fn a_clear_that_cannot_reach_the_store_changes_nothing() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let leaves = Leaves::default(); + leaves.put("atlas", LOGS_AND_METRICS); + record(&client, r#"["logs","metrics"]"#).await; + + leaves.fail(true); + let err = super::delete_with(&leaves, state.clone(), "atlas".to_owned()) + .await + .expect_err("store down"); + assert_eq!(err.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"permissions":["logs","metrics"]}"#) + ); + leaves.fail(false); + assert_eq!(leaves.get("atlas"), Some(LOGS_AND_METRICS)); + + super::delete_with(&leaves, state, "atlas".to_owned()) + .await + .expect("cleared"); + assert_eq!(leaves.get("atlas"), Some(Grant::default())); + assert_eq!(published(&client, "atlas").await, None); + } + + /// Destroying an agent drops the grant leaf first too. + #[tokio::test] + async fn a_destroy_that_cannot_reach_the_store_keeps_both() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let leaves = Leaves::default(); + leaves.put("atlas", LOGS_AND_METRICS); + record(&client, r#"["logs"]"#).await; + + leaves.fail(true); + super::clear_with(&leaves, &state, "pr1ma", "atlas").await; + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"permissions":["logs"]}"#) + ); + leaves.fail(false); + super::clear_with(&leaves, &state, "pr1ma", "atlas").await; + assert_eq!(published(&client, "atlas").await, None); + assert_eq!(leaves.get("atlas"), Some(Grant::default())); + } + + /// A grant writes the record first. With the store down the request is an + /// error, the record grants and the leaf does not, so the gateway denies; + /// the periodic pass then writes the leaf. + #[tokio::test] + async fn a_grant_that_cannot_reach_the_store_leaves_the_gateway_denying_until_the_pass() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = tolerant(&client).await; + let leaves = Leaves::default(); + + leaves.fail(true); + let err = put_via(&leaves, &state, &[Permission::Logs, Permission::Metrics]) + .await + .expect_err("store down"); + assert_eq!(err.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"permissions":["logs","metrics"]}"#), + "the record grants" + ); + leaves.fail(false); + assert_eq!(leaves.get("atlas"), None, "the leaf does not"); + + let store = swarm_queue_client::agent_permissions::open_or_create(&client) + .await + .expect("bucket"); + let wanted = state.wanted.as_ref().expect("wanted"); + let desired = super::desired_grants(&store, wanted, &["pr1ma".to_owned()]) + .await + .expect("desired"); + assert_eq!( + crate::observability_grant::reconcile(&leaves, &desired) + .await + .expect("pass"), + 1 + ); + assert_eq!(leaves.get("atlas"), Some(LOGS_AND_METRICS)); + } + + /// The pass reads the records, not the leaves: it revokes a leaf whose + /// agent has no record or an empty one, and names every live agent. + #[tokio::test] + async fn the_pass_makes_leaves_follow_records_in_both_directions() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let store = swarm_queue_client::agent_permissions::open_or_create(&client) + .await + .expect("bucket"); + store + .put("iris", br#"{"permissions":["metrics"]}"#.to_vec().into()) + .await + .expect("iris"); + let leaves = Leaves::default(); + leaves.put("atlas", LOGS_AND_METRICS); // live, no record: revoke + leaves.put("iris", Grant::default()); // record grants metrics: grant + + let wanted = state.wanted.as_ref().expect("wanted"); + let desired = super::desired_grants(&store, wanted, &["pr1ma".to_owned()]) + .await + .expect("desired"); + assert_eq!( + desired.keys().map(String::as_str).collect::>(), + ["atlas", "iris"] + ); + assert_eq!( + crate::observability_grant::reconcile(&leaves, &desired) + .await + .expect("pass"), + 2 + ); + assert_eq!(leaves.get("atlas"), Some(Grant::default())); + assert_eq!( + leaves.get("iris"), + Some(Grant { + logs: false, + metrics: true + }) ); } diff --git a/swarm-controller/src/observability_grant.rs b/swarm-controller/src/observability_grant.rs index 7f9cedcc..48653e7c 100644 --- a/swarm-controller/src/observability_grant.rs +++ b/swarm-controller/src/observability_grant.rs @@ -1,23 +1,38 @@ //! 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 +//! which agents 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 +//! An agent that cannot skip a permission name it does not know would drop to +//! no permissions on reading a record that holds one, so the controller grants +//! them only when the agent's hive status snapshot and its running harness's +//! turn-state header both advertise +//! `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. +//! +//! **The leaf is what the gateway enforces, the record is what the operator +//! sees**, so a change is ordered to fail towards denying: a revoke writes the +//! leaf first and the record second, a grant the record first and the leaf +//! second. [`reconcile`] makes every leaf equal its record's grant, every +//! [`REMIRROR_EVERY`], for the writes that failed half way. +use std::collections::BTreeMap; use std::future::Future; +use std::time::Duration; use anyhow::{Context as _, Result}; use hive_types::Permission; +use swarm_secret_client::SecretStore; use swarm_secret_client::observability::{Grant, grant_path}; use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS}; use crate::wanted::WantedWriter; +/// How often [`reconcile`] runs, matching the controller's other periodic +/// passes over the swarm's agents (`agent_renewal`). +pub(crate) const REMIRROR_EVERY: Duration = Duration::from_mins(5); + /// The grant `perms` hold. pub(crate) fn grant_of(perms: &[Permission]) -> Grant { Grant { @@ -26,20 +41,65 @@ pub(crate) fn grant_of(perms: &[Permission]) -> Grant { } } -/// Where a mirrored grant is written. A test stands in for the store. -pub(crate) trait GrantSink { +/// Whether `after` holds a grant `before` did not. +pub(crate) fn adds(before: &[Permission], after: &[Permission]) -> bool { + let (b, a) = (grant_of(before), grant_of(after)); + (a.logs && !b.logs) || (a.metrics && !b.metrics) +} + +/// Whether `after` drops a grant `before` held. +pub(crate) fn revokes(before: &[Permission], after: &[Permission]) -> bool { + adds(after, before) +} + +/// Where a mirrored grant is read and written. A test stands in for the store. +pub(crate) trait GrantLeaves { + /// The leaf, `None` when nothing is stored there. + fn read(&self, agent: &str) -> impl Future>> + Send; fn write(&self, agent: &str, grant: Grant) -> impl Future> + Send; } -/// The swarm secret store, under the controller's own identity. +/// The swarm secret store, under the controller's own identity, logging in +/// for each call. pub(crate) struct Store; -impl GrantSink for Store { +impl GrantLeaves for Store { + async fn read(&self, agent: &str) -> Result> { + Connected(connect().await?).read(agent).await + } + + async fn write(&self, agent: &str, grant: Grant) -> Result<()> { + Connected(connect().await?).write(agent, grant).await + } +} + +async fn connect() -> Result { + crate::store::connect() + .await + .context("logging in to the swarm secret store") +} + +/// One login, for a pass that touches many leaves. +pub(crate) struct Connected(pub(crate) SecretStore); + +impl Connected { + pub(crate) async fn login() -> Result { + Ok(Self(connect().await?)) + } +} + +impl GrantLeaves for Connected { + async fn read(&self, agent: &str) -> Result> { + let path = grant_path(agent)?; + self.0 + .read_optional(&path) + .await + .with_context(|| format!("reading {path}")) + } + 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")? + self.0 .write(&path, &grant) .await .with_context(|| format!("writing {path}")) @@ -51,9 +111,9 @@ impl GrantSink for Store { /// leaves the leaf absent. /// /// # Errors -/// The write failed; the responder then keeps the grant it last read, or none. +/// The write failed; the gateway then keeps the grant it last read, or none. pub(crate) async fn mirror( - sink: &impl GrantSink, + leaves: &impl GrantLeaves, agent: &str, before: &[Permission], after: &[Permission], @@ -61,7 +121,52 @@ pub(crate) async fn mirror( if grant_of(before) == Grant::default() && grant_of(after) == Grant::default() { return Ok(()); } - sink.write(agent, grant_of(after)).await + leaves.write(agent, grant_of(after)).await +} + +/// Make each agent's leaf equal `desired`, the grant its record holds; an +/// absent leaf equals the empty grant. Every correction is logged at `warn`. +/// All agents are tried; `Ok` is the number corrected. +/// +/// # Errors +/// How many agents could not be read or written. +pub(crate) async fn reconcile( + leaves: &impl GrantLeaves, + desired: &BTreeMap, +) -> Result { + let (mut corrected, mut failed) = (0, 0); + for (agent, want) in desired { + let have = match leaves.read(agent).await { + Ok(leaf) => leaf.unwrap_or_default(), + Err(e) => { + tracing::warn!(agent, error = %format!("{e:#}"), "observability grant: reading the leaf failed"); + failed += 1; + continue; + } + }; + if have == *want { + continue; + } + match leaves.write(agent, *want).await { + Ok(()) => { + tracing::warn!( + agent, + ?have, + ?want, + "observability grant: leaf differed from the record; corrected" + ); + corrected += 1; + } + Err(e) => { + tracing::warn!(agent, error = %format!("{e:#}"), "observability grant: correcting the leaf failed"); + failed += 1; + } + } + } + if failed > 0 { + anyhow::bail!("{failed} agent(s) could not be reconciled"); + } + Ok(corrected) } /// `Ok` when `hive`'s snapshot advertises tolerant decoding. @@ -128,22 +233,34 @@ pub(crate) async fn hive_is_tolerant( 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`. +/// `Ok` when `agent` may be granted `logs` or `metrics`: its hive advertises +/// tolerant decoding and so does the harness it is running (its latest +/// turn-state header, which an agent that is stopped, not yet restarted on a +/// tolerant build, or not reporting yet does not send). /// /// # 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( +/// lacks the capability, the harness has not reported it, or the queue could +/// not be read. +pub(crate) async fn require_tolerant_agent( client: &async_nats::Client, wanted: &WantedWriter, hives: &[String], + reported: Option<&crate::agent_login::Reported>, 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 + require_tolerant(client, &hive).await?; + if reported.is_some_and(|r| r.advertises(agent, TOLERANT_PERMISSIONS)) { + return Ok(()); + } + Err(format!( + "agent {agent}'s harness has not reported {TOLERANT_PERMISSIONS}: it must be running \ + a build that decodes the new permissions (update it, then wait a minute for its next \ + status)" + )) } /// `Ok` when `hive`'s last status snapshot advertises tolerant decoding. @@ -160,49 +277,89 @@ pub(crate) async fn require_tolerant( check_capability(hive, snapshot.as_ref()) } +/// A stand-in for the store's leaves, shared by the tests that mirror. #[cfg(test)] -mod tests { - use super::*; +pub(crate) mod fake { + use super::{GrantLeaves, Result}; + use std::collections::BTreeMap; use std::sync::Mutex; + use std::sync::atomic::{AtomicBool, Ordering}; + use swarm_secret_client::observability::Grant; + /// Leaves by agent. While `failing`, every read and write is an error. #[derive(Default)] - struct Leaves(Mutex>); + pub(crate) struct Leaves { + pub(crate) held: Mutex>, + pub(crate) failing: AtomicBool, + } - impl GrantSink for &Leaves { - async fn write(&self, agent: &str, grant: Grant) -> Result<()> { - self.0.lock().unwrap().push((agent.to_owned(), grant)); - Ok(()) + impl Leaves { + pub(crate) fn fail(&self, on: bool) { + self.failing.store(on, Ordering::SeqCst); + } + + pub(crate) fn get(&self, agent: &str) -> Option { + self.held.lock().unwrap().get(agent).copied() + } + + pub(crate) fn put(&self, agent: &str, grant: Grant) { + self.held.lock().unwrap().insert(agent.to_owned(), grant); } } - struct Down; - - impl GrantSink for Down { - async fn write(&self, _: &str, _: Grant) -> Result<()> { - anyhow::bail!("store unreachable") + impl GrantLeaves for Leaves { + async fn read(&self, agent: &str) -> Result> { + anyhow::ensure!(!self.failing.load(Ordering::SeqCst), "store unreachable"); + Ok(self.get(agent)) } + + async fn write(&self, agent: &str, grant: Grant) -> Result<()> { + anyhow::ensure!(!self.failing.load(Ordering::SeqCst), "store unreachable"); + self.put(agent, grant); + Ok(()) + } + } +} + +#[cfg(test)] +mod tests { + use super::fake::Leaves; + use super::*; + + const YES: Grant = Grant { + logs: true, + metrics: true, + }; + const LOGS: Grant = Grant { + logs: true, + metrics: false, + }; + const BOTH: [Permission; 3] = [Permission::Messaging, Permission::Logs, Permission::Metrics]; + const PLAIN: [Permission; 1] = [Permission::Messaging]; + + #[test] + fn adds_and_revokes_are_the_two_directions() { + assert!(adds(&PLAIN, &BOTH) && !revokes(&PLAIN, &BOTH)); + assert!(revokes(&BOTH, &PLAIN) && !adds(&BOTH, &PLAIN)); + assert!( + adds(&[Permission::Logs], &[Permission::Metrics]), + "swap adds" + ); + assert!( + revokes(&[Permission::Logs], &[Permission::Metrics]), + "and revokes" + ); + assert!(!adds(&BOTH, &BOTH) && !revokes(&BOTH, &BOTH)); + assert!(!adds(&PLAIN, &PLAIN) && !revokes(&PLAIN, &PLAIN)); } #[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()) - ] - ); + mirror(&leaves, "atlas", &[], &BOTH).await.unwrap(); + assert_eq!(leaves.get("atlas"), Some(YES)); + mirror(&leaves, "atlas", &BOTH, &PLAIN).await.unwrap(); + assert_eq!(leaves.get("atlas"), Some(Grant::default())); } /// An agent that never held either, before or after, gets no leaf; one @@ -210,23 +367,70 @@ mod tests { #[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]) + mirror(&leaves, "atlas", &PLAIN, &PLAIN).await.unwrap(); + assert_eq!(leaves.get("atlas"), None); + mirror(&leaves, "atlas", &PLAIN, &[Permission::Logs]) .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)); + assert_eq!(leaves.get("atlas"), Some(LOGS)); } #[tokio::test] async fn a_failed_write_is_an_error() { - let held = [Permission::Logs]; - assert!(mirror(&Down, "atlas", &[], &held).await.is_err()); + let leaves = Leaves::default(); + leaves.fail(true); + assert!(mirror(&leaves, "atlas", &[], &BOTH).await.is_err()); // Control: nothing to write is not a failure, whatever the store does. - assert!(mirror(&Down, "atlas", &[], &[]).await.is_ok()); + assert!(mirror(&leaves, "atlas", &[], &[]).await.is_ok()); + } + + /// The pass corrects a leaf that is granted where the record is not, one + /// that is missing or short where the record grants, and leaves the + /// agreeing ones (and an absent leaf for an empty grant) alone. + #[tokio::test] + async fn the_pass_corrects_drift_in_both_directions() { + let leaves = Leaves::default(); + leaves.put("stale", YES); + leaves.put("short", LOGS); + leaves.put("same", YES); + let desired = BTreeMap::from([ + ("stale".to_owned(), Grant::default()), + ("short".to_owned(), YES), + ("missing".to_owned(), YES), + ("same".to_owned(), YES), + ("none".to_owned(), Grant::default()), + ]); + assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 3); + assert_eq!( + leaves.get("stale"), + Some(Grant::default()), + "revoke repaired" + ); + assert_eq!(leaves.get("short"), Some(YES)); + assert_eq!(leaves.get("missing"), Some(YES), "grant repaired"); + assert_eq!(leaves.get("same"), Some(YES)); + assert_eq!(leaves.get("none"), None, "no leaf made for an empty grant"); + assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 0, "converged"); + } + + /// A store that is down is an error and changes nothing; the next pass, + /// once it is back, fixes it. + #[tokio::test] + async fn a_failed_pass_changes_nothing_and_the_next_one_fixes_it() { + let leaves = Leaves::default(); + leaves.put("stale", YES); + let desired = BTreeMap::from([ + ("stale".to_owned(), Grant::default()), + ("missing".to_owned(), YES), + ]); + leaves.fail(true); + assert!(reconcile(&leaves, &desired).await.is_err()); + leaves.fail(false); + assert_eq!(leaves.get("stale"), Some(YES)); + assert_eq!(leaves.get("missing"), None); + assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 2); + assert_eq!(leaves.get("stale"), Some(Grant::default())); + assert_eq!(leaves.get("missing"), Some(YES)); } #[test] diff --git a/swarm-wire/src/hive_status.rs b/swarm-wire/src/hive_status.rs index 82f8b65a..babb5e6d 100644 --- a/swarm-wire/src/hive_status.rs +++ b/swarm-wire/src/hive_status.rs @@ -1,13 +1,15 @@ -//! What a hive's status snapshot (`swarm_queue_client::status`) says about -//! what the hive can do, next to its readiness. +//! What a hive or an agent's harness says it can do, in the same `capabilities` +//! list of two documents: a hive's status snapshot +//! (`swarm_queue_client::status`, next to its readiness) and an agent's +//! turn-state header (`swarm_queue_client::agent_state_subject`). //! -//! The snapshot is opaque to the controller except for this field. A snapshot -//! from a hive that predates it has none, which reads as an empty list. +//! Readers take nothing else from the snapshot or header here. A document that +//! predates the field has none, which reads as an empty list. -/// The hive decodes a permission list that names a permission it does not -/// know by skipping that name, and runs agents on a harness that does too. -/// swarm-controller writes `logs` or `metrics` into an agent's record only -/// when the agent's hive advertises this. +/// The publisher decodes a permission list that names a permission it does not +/// know by skipping that name. swarm-controller writes `logs` or `metrics` into +/// an agent's record only when both its hive and its running harness advertise +/// this. pub const TOLERANT_PERMISSIONS: &str = "tolerant-permissions"; /// The part of a hive's status snapshot this module describes.