fan out the meta pin by a recorded pin, not by the move

swarm-controller/src/jobq.rs

@@ -10,8 +10,8 @@ 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, meta_flake,
-    wanted,
+    agent_identity, agent_renewal, auth, builder, config_pr, forge, matrix_account, meta_fanout,
+    meta_flake, wanted,
 };
 
 /// Node payload for the swarm-level job graph. Named `Swarm*` rather than
@@ -135,12 +135,17 @@ pub(crate) enum SwarmNodeKind {
         rev: Option<String>,
     },
     /// Pin the swarm meta flake (`internal/meta`): render, lock on the swarm
-    /// builder, commit. See `meta_flake`. When the commit moves the hyperhive
-    /// pin, a `TriggerDeploy` per placed agent follows.
+    /// builder, commit. See `meta_flake`. When the pin `internal/meta` holds is
+    /// not the one `meta_fanout` recorded, a `TriggerDeploy` per placed agent
+    /// follows, then a `RecordFanout`.
     ///
     /// Carries nothing: it reads every head when it runs, so one pending node
     /// covers every trigger queued behind it.
     UpdateMeta,
+    /// Record `pin` in `meta_fanout` as sent to the hives. Runs only once every
+    /// `TriggerDeploy` declared beside it is done, so a failed or lost deploy
+    /// leaves the record stale and the next `UpdateMeta` sends them again.
+    RecordFanout { pin: String },
 }
 
 impl hive_jobq_wire::WireNode for SwarmNodeKind {
@@ -167,6 +172,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
             SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(),
             SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(),
             SwarmNodeKind::UpdateMeta => "update_meta".to_owned(),
+            SwarmNodeKind::RecordFanout { .. } => "record_fanout".to_owned(),
         }
     }
 
@@ -205,6 +211,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
             SwarmNodeKind::SetAgentWanted { hive, agent } => {
                 serde_json::json!({ "agent": agent, "hive": hive })
             }
+            SwarmNodeKind::RecordFanout { pin } => serde_json::json!({ "pin": pin }),
         }
     }
 }
@@ -396,6 +403,7 @@ pub(crate) async fn run_swarm_node(
             Some(client) => publish_deploy(&client, &hive, &agent, rev).await,
         },
         SwarmNodeKind::UpdateMeta => finish_update_meta(update_meta(&deps, forge).await, &builder),
+        SwarmNodeKind::RecordFanout { pin } => record_fanout(deps.queue.as_ref(), &pin).await,
     };
     (builder, outcome)
 }
@@ -565,18 +573,55 @@ pub(crate) async fn build_agent(
     .await
 }
 
+/// What an `UpdateMeta` pass decided: the deploys to send, and the placements
+/// to send them to.
+struct MetaPass {
+    fan_out: FanOut,
+    placements: std::collections::BTreeSet<(String, String)>,
+}
+
+/// What a pass sends the placed agents.
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum FanOut {
+    /// Nothing: the pin is the one recorded.
+    None,
+    /// A deploy each, then record `pin`.
+    Record { pin: String },
+    /// A deploy each, with nothing to record into: the record cannot be read.
+    Unrecorded,
+}
+
+/// The fan-out for the pin `report` leaves, given the pin `meta_fanout` has
+/// recorded. With no readable record, only a move of the pin sends deploys,
+/// and nothing can be recorded.
+fn decide_fan_out(report: &meta_flake::Report, recorded: Result<Option<String>>) -> FanOut {
+    match recorded {
+        Ok(Some(pin)) if pin == report.pin => FanOut::None,
+        Ok(_) => FanOut::Record {
+            pin: report.pin.clone(),
+        },
+        Err(e) => {
+            tracing::warn!(
+                error = %format!("{e:#}"),
+                moved = report.hyperhive_moved,
+                "update_meta: the recorded fan-out pin is unreadable; deploying only on a moved pin"
+            );
+            if report.hyperhive_moved {
+                FanOut::Unrecorded
+            } else {
+                FanOut::None
+            }
+        }
+    }
+}
+
 /// The `UpdateMeta` arm, lifted out for the same reason: what the update
-/// reported, and the placements it rendered for. `Err` is the node's failure.
+/// decided to send, and the placements it rendered for. `Err` is the node's
+/// failure.
 async fn update_meta(
     deps: &WorkerDeps,
     forge: Option<Arc<forge::Client>>,
-) -> Result<
-    (
-        meta_flake::Report,
-        std::collections::BTreeSet<(String, String)>,
-    ),
-    String,
-> {
+) -> Result<MetaPass, String> {
     let Some(client) = forge else {
         return Err(
             "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
@@ -610,50 +655,79 @@ async fn update_meta(
     .await
     .map_err(|e| format!("{e:#}"))?;
     tracing::info!(?report, agents = agents.len(), "update_meta: done");
-    Ok((report, placements))
+    let fan_out = decide_fan_out(&report, meta_fanout::recorded(queue).await);
+    Ok(MetaPass {
+        fan_out,
+        placements,
+    })
 }
 
-/// The `UpdateMeta` node's outcome from what [`update_meta`] reported, with the
-/// deploys a moved pin calls for declared into `builder` (the scheduler inserts
-/// them under the node when it completes). Synchronous: the builder is not
-/// `Sync`, so it cannot be held across an await.
+/// The `UpdateMeta` node's outcome from what [`update_meta`] decided, with the
+/// deploys declared into `builder` (the scheduler inserts them under the node
+/// when it completes). Synchronous: the builder is not `Sync`, so it cannot be
+/// held across an await.
 fn finish_update_meta(
-    result: Result<
-        (
-            meta_flake::Report,
-            std::collections::BTreeSet<(String, String)>,
-        ),
-        String,
-    >,
+    result: Result<MetaPass, String>,
     builder: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
 ) -> hive_jobq::scheduler::Outcome {
     use hive_jobq::scheduler::Outcome;
 
     match result {
-        Ok((report, placements)) => {
-            fan_out_if_moved(builder, &report, &placements);
+        Ok(pass) => {
+            declare_fan_out(builder, &pass.fan_out, &pass.placements);
             Outcome::Done
         }
         Err(e) => Outcome::Failed(e),
     }
 }
 
-/// Declare one rebuild of each `(hive, agent)` placement when `report` says
-/// the agents' hyperhive moved: nothing else tells them to build again.
-fn fan_out_if_moved(
+/// The `RecordFanout` arm, lifted out so `run_swarm_node` stays under the line limit.
+async fn record_fanout(
+    queue: Option<&async_nats::Client>,
+    pin: &str,
+) -> hive_jobq::scheduler::Outcome {
+    use hive_jobq::scheduler::Outcome;
+
+    let Some(client) = queue else {
+        return Outcome::Failed(
+            "no swarm queue is configured on this host, so the fan-out cannot be recorded"
+                .to_owned(),
+        );
+    };
+    match meta_fanout::record(client, pin).await {
+        Ok(()) => Outcome::Done,
+        Err(e) => Outcome::Failed(format!("{e:#}")),
+    }
+}
+
+/// Declare one rebuild of each `(hive, agent)` placement, which nothing else
+/// tells to build again, and for a [`FanOut::Record`] the `RecordFanout` that
+/// waits on every one of them.
+fn declare_fan_out(
     grown: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
-    report: &meta_flake::Report,
+    fan_out: &FanOut,
     placements: &std::collections::BTreeSet<(String, String)>,
 ) {
-    if !report.hyperhive_moved {
+    if *fan_out == FanOut::None {
         return;
     }
-    for (hive, agent) in placements {
-        let _ = grown.node(SwarmNodeKind::TriggerDeploy {
-            hive: hive.clone(),
-            agent: agent.clone(),
-            rev: None,
-        });
+    let deploys: Vec<hive_jobq::builder::NodeGuid> = placements
+        .iter()
+        .map(|(hive, agent)| {
+            grown
+                .node(SwarmNodeKind::TriggerDeploy {
+                    hive: hive.clone(),
+                    agent: agent.clone(),
+                    rev: None,
+                })
+                .guid()
+        })
+        .collect();
+    if let FanOut::Record { pin } = fan_out {
+        let mut record = grown.node(SwarmNodeKind::RecordFanout { pin: pin.clone() });
+        for deploy in deploys {
+            record = record.after_ok(deploy);
+        }
     }
 }
 
@@ -2092,67 +2166,300 @@ mod tests {
         );
     }
 
-    /// Run an `UpdateMeta` node that reports `report`, and return the
-    /// `(hive, agent, rev)` of every deploy it grew.
-    async fn grown_deploys(report: meta_flake::Report) -> Vec<(String, String, Option<String>)> {
+    /// A fan-out record held in memory; `unreadable` stands for a bucket that
+    /// cannot be opened.
+    #[derive(Default)]
+    struct FakePins {
+        pin: Mutex<Option<String>>,
+        unreadable: bool,
+    }
+
+    impl FakePins {
+        fn holding(pin: &str) -> Self {
+            Self {
+                pin: Mutex::new(Some(pin.to_owned())),
+                unreadable: false,
+            }
+        }
+
+        fn held(&self) -> Option<String> {
+            self.pin.lock().unwrap().clone()
+        }
+    }
+
+    impl meta_fanout::PinStore for FakePins {
+        async fn recorded(&self) -> Result<Option<String>> {
+            anyhow::ensure!(!self.unreadable, "the bucket did not open");
+            Ok(self.held())
+        }
+
+        async fn record(&self, pin: &str) -> Result<()> {
+            *self.pin.lock().unwrap() = Some(pin.to_owned());
+            Ok(())
+        }
+    }
+
+    fn report(pin: &str, moved: bool) -> meta_flake::Report {
+        meta_flake::Report {
+            updated: meta_flake::Updated::Committed,
+            pin: pin.to_owned(),
+            hyperhive_moved: moved,
+        }
+    }
+
+    type TestScheduler = hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>;
+
+    /// One `UpdateMeta` pass against `pins`, with every node it grows run: the
+    /// deploy to `failing_agent` fails, and `RecordFanout` dies before it
+    /// writes when `crash_before_record`. Returns what the pass decided and the
+    /// scheduler it ran in.
+    async fn pass(
+        pins: &Arc<FakePins>,
+        report: &meta_flake::Report,
+        placed: &[(&str, &str)],
+        failing_agent: Option<&'static str>,
+        crash_before_record: bool,
+    ) -> (FanOut, Arc<Mutex<TestScheduler>>) {
         use hive_jobq::scheduler::{Outcome, Scheduler};
+        use meta_fanout::PinStore as _;
 
         let mut sched = Scheduler::new(
             hive_jobq::Graph::new(),
             hive_jobq::resources::ResourceTable::new(),
         );
-        let update = sched
+        sched
             .append(SwarmNodeKind::UpdateMeta, Vec::new(), None)
             .expect("insert");
-        let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
-        let placements: std::collections::BTreeSet<(String, String)> =
-            [("pr1ma", "atlas"), ("pr1ma", "argus"), ("sec0nd", "iris")]
-                .into_iter()
-                .map(|(h, a)| (h.to_owned(), a.to_owned()))
-                .collect();
-        let runner = Scheduler::claim_next(&sched, move |_id, _kind, builder| async move {
-            fan_out_if_moved(&builder, &report, &placements);
+        let sched = Arc::new(Mutex::new(sched));
+        let placements: std::collections::BTreeSet<(String, String)> = placed
+            .iter()
+            .map(|(h, a)| ((*h).to_owned(), (*a).to_owned()))
+            .collect();
+        let fan_out = decide_fan_out(report, pins.recorded().await);
+        let decided = fan_out.clone();
+        let update = Scheduler::claim_next(&sched, move |_id, _kind, builder| async move {
+            declare_fan_out(&builder, &fan_out, &placements);
             (builder, Outcome::Done)
         })
         .expect("the node just inserted is runnable");
-        runner.await.1.expect("the grown nodes insert");
-        let guard = sched.lock().unwrap();
-        guard
+        update.await.1.expect("the grown nodes insert");
+        while let Some(runner) = Scheduler::claim_next(&sched, {
+            let (pins, sched) = (Arc::clone(pins), Arc::clone(&sched));
+            move |_id, kind, builder| async move {
+                let outcome = match kind {
+                    SwarmNodeKind::TriggerDeploy { agent, .. }
+                        if failing_agent == Some(agent.as_str()) =>
+                    {
+                        Outcome::Failed("the hive did not take it".to_owned())
+                    }
+                    SwarmNodeKind::RecordFanout { .. } if crash_before_record => {
+                        Outcome::Failed("the controller died".to_owned())
+                    }
+                    SwarmNodeKind::RecordFanout { pin } => {
+                        let deploys_done = sched
+                            .lock()
+                            .unwrap()
+                            .graph()
+                            .nodes()
+                            .filter(|n| matches!(n.payload, SwarmNodeKind::TriggerDeploy { .. }))
+                            .all(|n| n.state == hive_jobq::State::Done);
+                        if deploys_done {
+                            pins.record(&pin).await.expect("recorded");
+                            Outcome::Done
+                        } else {
+                            Outcome::Failed("recorded before the deploys were done".to_owned())
+                        }
+                    }
+                    _ => Outcome::Done,
+                };
+                (builder, outcome)
+            }
+        }) {
+            runner.await.1.expect("no growth declared");
+        }
+        (decided, sched)
+    }
+
+    /// The states of the nodes labelled `label`.
+    fn states_of(sched: &Mutex<TestScheduler>, label: &str) -> Vec<hive_jobq::State> {
+        use hive_jobq_wire::WireNode as _;
+
+        sched
+            .lock()
+            .unwrap()
+            .graph()
+            .nodes()
+            .filter(|n| n.payload.label() == label)
+            .map(|n| n.state)
+            .collect()
+    }
+
+    const PLACED: [(&str, &str); 3] = [("pr1ma", "atlas"), ("pr1ma", "argus"), ("sec0nd", "iris")];
+
+    /// A controller with no record sends every placed agent a deploy, whether
+    /// or not the pin moved, and records the pin once they are done.
+    #[tokio::test]
+    async fn a_pin_with_no_record_deploys_every_placed_agent_and_is_recorded() {
+        let pins = Arc::new(FakePins::default());
+        let (fan_out, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::Record { pin: "p1".into() });
+        assert_eq!(
+            states_of(&sched, "trigger_deploy"),
+            [hive_jobq::State::Done; 3]
+        );
+        assert_eq!(states_of(&sched, "record_fanout"), [hive_jobq::State::Done]);
+        assert_eq!(pins.held().as_deref(), Some("p1"));
+    }
+
+    /// The deploys declared are one per placement, to the hive that places it.
+    #[tokio::test]
+    async fn each_deploy_goes_to_the_hive_that_places_the_agent() {
+        let pins = Arc::new(FakePins::default());
+        let (_, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        let mut deploys: Vec<_> = sched
+            .lock()
+            .unwrap()
             .graph()
             .nodes()
             .filter_map(|n| match &n.payload {
                 SwarmNodeKind::TriggerDeploy { hive, agent, rev } => {
-                    assert_eq!(n.parent, Some(update), "grown under the update");
                     Some((hive.clone(), agent.clone(), rev.clone()))
                 }
                 _ => None,
             })
-            .collect()
-    }
-
-    #[tokio::test]
-    async fn a_moved_pin_rebuilds_every_placed_agent_and_an_unmoved_one_none() {
-        let moved = grown_deploys(meta_flake::Report {
-            updated: meta_flake::Updated::Committed,
-            hyperhive_moved: true,
-        })
-        .await;
-        let mut moved: Vec<_> = moved.into_iter().collect();
-        moved.sort();
+            .collect();
+        deploys.sort();
         assert_eq!(
-            moved,
+            deploys,
             vec![
                 ("pr1ma".to_owned(), "argus".to_owned(), None),
                 ("pr1ma".to_owned(), "atlas".to_owned(), None),
                 ("sec0nd".to_owned(), "iris".to_owned(), None),
             ]
         );
-        let unmoved = grown_deploys(meta_flake::Report {
-            updated: meta_flake::Updated::Committed,
-            hyperhive_moved: false,
-        })
-        .await;
-        assert!(unmoved.is_empty(), "{unmoved:?}");
+    }
+
+    /// The record, not whether the pin moved, decides: a pin already sent
+    /// sends nothing even on a pass that moved it, and a different one does.
+    #[tokio::test]
+    async fn a_pin_that_is_recorded_deploys_nothing() {
+        let pins = Arc::new(FakePins::holding("p1"));
+        let (fan_out, sched) = pass(&pins, &report("p1", true), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::None);
+        assert!(states_of(&sched, "trigger_deploy").is_empty());
+        assert!(states_of(&sched, "record_fanout").is_empty());
+
+        let (fan_out, sched) = pass(&pins, &report("p2", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::Record { pin: "p2".into() });
+        assert_eq!(states_of(&sched, "trigger_deploy").len(), 3);
+        assert_eq!(pins.held().as_deref(), Some("p2"));
+    }
+
+    /// A controller that dies after the commit and before the record is
+    /// written finds the commit unchanged on its next pass, and still sends.
+    #[tokio::test]
+    async fn a_pass_that_dies_before_the_record_is_written_sends_again() {
+        let pins = Arc::new(FakePins::holding("p0"));
+        let (_, sched) = pass(&pins, &report("p1", true), &PLACED, None, true).await;
+        assert_eq!(states_of(&sched, "trigger_deploy").len(), 3);
+        assert_eq!(pins.held().as_deref(), Some("p0"), "nothing was recorded");
+
+        // The restart: the commit is the render now, so nothing moved.
+        let (fan_out, _) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::Record { pin: "p1".into() });
+        assert_eq!(pins.held().as_deref(), Some("p1"));
+
+        let (fan_out, _) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::None);
+    }
+
+    /// The record waits on every deploy of its pass, for all of them to be
+    /// done, not merely finished.
+    #[tokio::test]
+    async fn the_record_waits_on_every_deploy_being_done() {
+        let pins = Arc::new(FakePins::default());
+        let (_, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        let guard = sched.lock().unwrap();
+        let graph = guard.graph();
+        let record = graph
+            .nodes()
+            .find(|n| matches!(n.payload, SwarmNodeKind::RecordFanout { .. }))
+            .expect("a record node");
+        let mut waits: Vec<_> = record
+            .deps
+            .iter()
+            .map(|d| match d {
+                hive_jobq::Dep::Node { id, when } => (*id, *when),
+                hive_jobq::Dep::Resource { .. } => panic!("the record needs no resource"),
+            })
+            .collect();
+        waits.sort_by_key(|(id, _)| id.get());
+        let mut deploys: Vec<_> = graph
+            .nodes()
+            .filter(|n| matches!(n.payload, SwarmNodeKind::TriggerDeploy { .. }))
+            .map(|n| (n.id, hive_jobq::DepWhen::AFTER_OK))
+            .collect();
+        deploys.sort_by_key(|(id, _)| id.get());
+        assert_eq!(deploys.len(), 3);
+        assert_eq!(waits, deploys);
+    }
+
+    /// One deploy failing keeps the pin unrecorded: the record is ruled out,
+    /// the others still run, and the next pass sends again.
+    #[tokio::test]
+    async fn a_failed_deploy_leaves_the_pin_unrecorded() {
+        let pins = Arc::new(FakePins::holding("p0"));
+        let (_, sched) = pass(&pins, &report("p1", true), &PLACED, Some("argus"), false).await;
+        let mut deploys = states_of(&sched, "trigger_deploy");
+        deploys.sort_by_key(|s| format!("{s:?}"));
+        assert_eq!(
+            deploys,
+            [
+                hive_jobq::State::Done,
+                hive_jobq::State::Done,
+                hive_jobq::State::Failed
+            ]
+        );
+        assert_eq!(
+            states_of(&sched, "record_fanout"),
+            [hive_jobq::State::Skipped]
+        );
+        assert_eq!(pins.held().as_deref(), Some("p0"));
+
+        let (fan_out, _) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::Record { pin: "p1".into() });
+    }
+
+    /// With nothing placed there is nothing to wait on, and the pin is
+    /// recorded all the same.
+    #[tokio::test]
+    async fn a_pin_with_no_placed_agent_is_recorded() {
+        let pins = Arc::new(FakePins::default());
+        let (_, sched) = pass(&pins, &report("p1", false), &[], None, false).await;
+        assert!(states_of(&sched, "trigger_deploy").is_empty());
+        assert_eq!(states_of(&sched, "record_fanout"), [hive_jobq::State::Done]);
+        assert_eq!(pins.held().as_deref(), Some("p1"));
+    }
+
+    /// While the record cannot be read, a moved pin deploys as it did before
+    /// the record existed, and there is nothing to record into.
+    #[tokio::test]
+    async fn an_unreadable_record_falls_back_to_the_moved_pin() {
+        let pins = Arc::new(FakePins {
+            unreadable: true,
+            ..FakePins::default()
+        });
+        let (fan_out, sched) = pass(&pins, &report("p1", true), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::Unrecorded);
+        assert_eq!(
+            states_of(&sched, "trigger_deploy"),
+            [hive_jobq::State::Done; 3]
+        );
+        assert!(states_of(&sched, "record_fanout").is_empty());
+
+        let (fan_out, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await;
+        assert_eq!(fan_out, FanOut::None);
+        assert!(states_of(&sched, "trigger_deploy").is_empty());
     }
 
     /// Node labels of the graph [`super::declare_agent_job`] builds, and the