swarm-controller: keep a hive's read grant in step with its declaration

The controller writes an agent's credential; the hive fetches it back with
its own certificate. Nothing said which paths that certificate may read, so
the read half of a delivery answers 403 with no way to tell why.

The grant is derived from the declaration, so it is re-rendered at the one
place the declaration changes -- WantedWriter::set -- rather than at its
caller, which would work today and break on the second caller.

Emitted before the KV write: a grant that lands late is a 403 on an agent's
first fetch, while one that shrinks early only affects an agent already
being torn down. A failed write then leaves a superset the next declaration
re-renders.

Destroyed agents are filtered out. The declared set is a hive's whole
history -- a destroyed entry stays so that redeclaring it Up is refused as
the terminal transition it is -- so granting every declared agent would
leave a torn-down agent's credentials readable forever.

The sink is a trait because a missed emission is that same untraceable 403:
the double pins which agents were published, and the no-sink and refusing
arms pin the two deployments that are not a happy path. Not covered: the
call site inside set(), which needs a live queue.

The cert role moves to its own module on the way past. It is the
controller's identity at the store, not something the matrix route owns,
and the policy writer needs the same login.

swarm-controller/src/wanted.rs

@@ -13,12 +13,14 @@
 //! declaration to reconcile against, because the current value can be read
 //! back from the queue whenever it is needed.
 
-use std::fmt;
+use std::{fmt, sync::Arc};
 
 use anyhow::{Context, Result};
 use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind};
 use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};
 
+use crate::read_policy::ReadPolicySink;
+
 /// The one failure out of [`apply`] that is the caller's mistake, not a
 /// server fault — worth a distinct type so `set_agent_state` can tell it
 /// apart from every other error this module raises (an undecodable
@@ -101,15 +103,60 @@ fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(Hive
     Ok((declaration, encoded))
 }
 
+/// Which of a declaration's agents a hive may read credentials for.
+///
+/// Not all of them: the declared set is the hive's whole history, destroyed
+/// entries included — they stay so that redeclaring one back to `Up` can be
+/// refused as the terminal transition it is, rather than silently
+/// un-destroying an agent. A torn-down agent keeping read access on its
+/// credentials is what this filter exists to stop.
+fn readable_agents(declaration: &HiveWanted) -> Vec<&str> {
+    declaration
+        .agents
+        .iter()
+        .filter(|(_, wanted)| wanted.state != AgentState::Destroyed)
+        .map(|(agent, _)| agent.as_str())
+        .collect()
+}
+
+/// Re-render `hive`'s read grant from the declaration about to be published.
+///
+/// Separate from [`WantedWriter::set`] so the trigger is assertable: `set`
+/// needs a live queue, this needs nothing, and "was the grant emitted, over
+/// exactly which agents" is the half that fails as a 403 nobody can trace.
+async fn publish_read_grant(
+    sink: Option<&Arc<dyn ReadPolicySink>>,
+    hive: &str,
+    declaration: &HiveWanted,
+) -> Result<()> {
+    let Some(sink) = sink else {
+        return Ok(());
+    };
+    let agents = readable_agents(declaration);
+    sink.publish(hive, &agents)
+        .await
+        .with_context(|| format!("publishing the read grant for {hive}"))
+}
+
 /// Writes the wanted-state bucket, and reads it back.
 pub struct WantedWriter {
     client: async_nats::Client,
+    /// Where a hive's read grant is published, or `None` on a deployment
+    /// with no store identity — see [`WantedWriter::new`].
+    read_policy: Option<Arc<dyn ReadPolicySink>>,
 }
 
 impl WantedWriter {
+    /// `read_policy` is `None` when this deployment has no secret store to
+    /// authenticate to. A declaration still publishes; there is simply no
+    /// grant to keep in step with it, and the credential delivery it would
+    /// have served does not exist either.
     #[must_use]
-    pub fn new(client: async_nats::Client) -> Self {
-        Self { client }
+    pub fn new(client: async_nats::Client, read_policy: Option<Arc<dyn ReadPolicySink>>) -> Self {
+        Self {
+            client,
+            read_policy,
+        }
     }
 
     /// One hive's bucket handle, created on first use if nothing has made it
@@ -170,6 +217,14 @@ impl WantedWriter {
             let (declaration, encoded) =
                 apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?;
 
+            // Before the write, not after: a grant that lands late is a 403
+            // on the agent's first credential fetch, while
+            // a grant that shrinks a moment early only affects an agent
+            // already being torn down. If the write below then fails, the
+            // grant is a harmless superset and the next declaration
+            // re-renders it.
+            publish_read_grant(self.read_policy.as_ref(), hive, &declaration).await?;
+
             // `update` and `create` have separate error types, and only one
             // variant of each means "someone else got there first". Every
             // other failure returns immediately: retrying a disconnect or a
@@ -209,8 +264,114 @@ impl WantedWriter {
 
 #[cfg(test)]
 mod tests {
-    use super::apply;
-    use swarm_queue_client::wanted::AgentState;
+    use super::{apply, publish_read_grant, readable_agents};
+    use std::sync::{Arc, Mutex};
+    use swarm_queue_client::wanted::{AgentState, HiveWanted};
+
+    /// Records what a caller published instead of talking to a store.
+    #[derive(Default)]
+    struct Recorder {
+        published: Mutex<Vec<(String, Vec<String>)>>,
+    }
+
+    #[async_trait::async_trait]
+    impl super::ReadPolicySink for Recorder {
+        async fn publish(&self, hive: &str, agents: &[&str]) -> anyhow::Result<()> {
+            self.published.lock().unwrap().push((
+                hive.to_owned(),
+                agents.iter().map(|a| (*a).to_owned()).collect(),
+            ));
+            Ok(())
+        }
+    }
+
+    /// A sink that always refuses, to pin that a failed grant is not swallowed.
+    struct Refuses;
+
+    #[async_trait::async_trait]
+    impl super::ReadPolicySink for Refuses {
+        async fn publish(&self, _hive: &str, _agents: &[&str]) -> anyhow::Result<()> {
+            anyhow::bail!("the store said no")
+        }
+    }
+
+    fn declaration(pairs: &[(&str, AgentState)]) -> HiveWanted {
+        let mut declaration = HiveWanted::default();
+        for (agent, state) in pairs {
+            let (d, _) = apply(
+                Some(&serde_json::to_vec(&declaration).unwrap()),
+                agent,
+                *state,
+            )
+            .unwrap();
+            declaration = d;
+        }
+        declaration
+    }
+
+    #[test]
+    fn a_destroyed_agent_is_not_granted_read_while_the_live_ones_are() {
+        let d = declaration(&[
+            ("iris", AgentState::Up),
+            ("argus", AgentState::Offline),
+            ("atlas", AgentState::Destroyed),
+        ]);
+        let mut granted = readable_agents(&d);
+        granted.sort_unstable();
+        assert_eq!(
+            granted,
+            ["argus", "iris"],
+            "destroyed keeps its declaration"
+        );
+    }
+
+    /// The control on the filter above: without it every assertion there
+    /// could pass because `readable_agents` returns nothing at all.
+    #[test]
+    fn a_declaration_with_no_destroyed_agent_grants_every_one_of_them() {
+        let d = declaration(&[("iris", AgentState::Up), ("argus", AgentState::Offline)]);
+        assert_eq!(readable_agents(&d).len(), 2);
+    }
+
+    #[tokio::test]
+    async fn declaring_an_agent_publishes_the_grant_for_exactly_the_live_set() {
+        let recorder = Arc::new(Recorder::default());
+        let sink: Arc<dyn super::ReadPolicySink> = recorder.clone();
+        let d = declaration(&[("iris", AgentState::Up), ("atlas", AgentState::Destroyed)]);
+
+        publish_read_grant(Some(&sink), "pr1ma", &d).await.unwrap();
+
+        let published = recorder.published.lock().unwrap().clone();
+        assert_eq!(
+            published,
+            vec![("pr1ma".to_owned(), vec!["iris".to_owned()])]
+        );
+    }
+
+    /// A deployment with no store still declares agent state — there is just
+    /// no grant to keep in step with it.
+    #[tokio::test]
+    async fn no_sink_is_a_no_op_rather_than_a_failure() {
+        let d = declaration(&[("iris", AgentState::Up)]);
+        publish_read_grant(None, "pr1ma", &d)
+            .await
+            .expect("a deployment with no store is not an error");
+    }
+
+    /// The failure this ordering exists for: a refused grant must stop the
+    /// declaration, not publish one whose reader will 403.
+    #[tokio::test]
+    async fn a_refused_grant_is_an_error_the_caller_sees() {
+        let sink: Arc<dyn super::ReadPolicySink> = Arc::new(Refuses);
+        let d = declaration(&[("iris", AgentState::Up)]);
+        let err = publish_read_grant(Some(&sink), "pr1ma", &d)
+            .await
+            .expect_err("the sink refused");
+        assert!(
+            format!("{err:#}").contains("the store said no"),
+            "the underlying refusal should survive the context: {err:#}"
+        );
+    }
 
     #[test]
     fn declaring_one_agent_preserves_every_other() {