From 2b9927fda21f7e3accb5d61951d435ad6889300b Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 25 Oct 2026 02:05:24 +0200 Subject: [PATCH] fan out the meta pin by a recorded pin, not by the move --- swarm-controller/src/jobq.rs | 451 +++++++++++++++++++++++----- swarm-controller/src/main.rs | 1 + swarm-controller/src/meta_fanout.rs | 98 ++++++ swarm-controller/src/meta_flake.rs | 14 +- 4 files changed, 489 insertions(+), 75 deletions(-) create mode 100644 swarm-controller/src/meta_fanout.rs diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index a15c5f80..18e50825 100644 --- a/swarm-controller/src/jobq.rs +++ b/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, }, /// 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>) -> 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>, -) -> Result< - ( - meta_flake::Report, - std::collections::BTreeSet<(String, String)>, - ), - String, -> { +) -> Result { 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, builder: &hive_jobq::builder::JobBuilder, ) -> 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, - 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 = 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)> { + /// A fan-out record held in memory; `unreadable` stands for a bucket that + /// cannot be opened. + #[derive(Default)] + struct FakePins { + pin: Mutex>, + unreadable: bool, + } + + impl FakePins { + fn holding(pin: &str) -> Self { + Self { + pin: Mutex::new(Some(pin.to_owned())), + unreadable: false, + } + } + + fn held(&self) -> Option { + self.pin.lock().unwrap().clone() + } + } + + impl meta_fanout::PinStore for FakePins { + async fn recorded(&self) -> Result> { + 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; + + /// 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, + report: &meta_flake::Report, + placed: &[(&str, &str)], + failing_agent: Option<&'static str>, + crash_before_record: bool, + ) -> (FanOut, Arc>) { 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, label: &str) -> Vec { + 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 diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index fa9754f9..9c99a446 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -50,6 +50,7 @@ mod jobq; mod linked_accounts; mod matrix_account; mod message_stream; +mod meta_fanout; mod meta_flake; mod notify_flows; mod notify_router_api; diff --git a/swarm-controller/src/meta_fanout.rs b/swarm-controller/src/meta_fanout.rs new file mode 100644 index 00000000..c8d2aed7 --- /dev/null +++ b/swarm-controller/src/meta_fanout.rs @@ -0,0 +1,98 @@ +//! The hyperhive pin the controller last sent the hives to build from. +//! +//! One key ([`KEY`]) in [`BUCKET`], which only this daemon reads or writes. +//! The `UpdateMeta` node fans out whenever the pin `internal/meta` holds is +//! not the one recorded here, and the `RecordFanout` node writes it once every +//! `TriggerDeploy` of that fan-out is done. A pass that dies, or a deploy that +//! fails, before then leaves the record stale, so the next pass sends the +//! deploys again. A deploy is idempotent per pin, so sending twice is safe and +//! sending never is not. +//! +//! With no record, as on the first pass of a controller that has never written +//! one, the pin is not the recorded one and the agents are sent a deploy. + +use std::future::Future; + +use anyhow::{Context as _, Result}; +use async_nats::jetstream::kv::Store; + +/// The bucket holding the record. Agents hold no grant on it. +pub(crate) const BUCKET: &str = "meta-fanout"; + +/// The one key in [`BUCKET`]: the pin, as a rev string. +const KEY: &str = "pin"; + +/// Where the record lives: the bucket in production, a fake in the tests. +pub(crate) trait PinStore { + /// The recorded pin; `None` when none was ever written. + fn recorded(&self) -> impl Future>> + Send; + fn record(&self, pin: &str) -> impl Future> + Send; +} + +impl PinStore for Store { + async fn recorded(&self) -> Result> { + let entry = self + .get(KEY) + .await + .with_context(|| format!("read {BUCKET}/{KEY}"))?; + Ok(entry.map(|b| String::from_utf8_lossy(&b).into_owned())) + } + + async fn record(&self, pin: &str) -> Result<()> { + self.put(KEY, pin.to_owned().into()) + .await + .map(|_| ()) + .with_context(|| format!("write {BUCKET}/{KEY}")) + } +} + +/// Open [`BUCKET`], creating it if nothing has yet. +async fn open(client: &async_nats::Client) -> Result { + let js = async_nats::jetstream::new(client.clone()); + if let Ok(store) = js.get_key_value(BUCKET).await { + return Ok(store); + } + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "The hyperhive pin the controller last sent the hives to build from" + .to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {BUCKET} bucket")) +} + +/// The pin recorded in the bucket on `client`, opening or creating it. +/// +/// # Errors +/// The bucket cannot be opened or created, or its key cannot be read. +pub(crate) async fn recorded(client: &async_nats::Client) -> Result> { + open(client).await?.recorded().await +} + +/// Record `pin` as sent, in the bucket on `client`. +/// +/// # Errors +/// The bucket cannot be opened or created, or the key cannot be written. +pub(crate) async fn record(client: &async_nats::Client, pin: &str) -> Result<()> { + open(client).await?.record(pin).await +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The bucket on a real server: absent until written, then the pin + /// written, and a second write replaces it. + #[tokio::test] + async fn the_pin_round_trips_through_the_bucket() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + assert_eq!(recorded(&client).await.expect("opens"), None); + record(&client, "aaaa").await.expect("written"); + assert_eq!(recorded(&client).await.expect("read"), Some("aaaa".into())); + record(&client, "bbbb").await.expect("replaced"); + assert_eq!(recorded(&client).await.expect("read"), Some("bbbb".into())); + } +} diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs index 518835ad..42f54098 100644 --- a/swarm-controller/src/meta_flake.rs +++ b/swarm-controller/src/meta_flake.rs @@ -205,9 +205,12 @@ pub(crate) enum Updated { #[derive(Debug, PartialEq, Eq)] pub(crate) struct Report { pub(crate) updated: Updated, - /// The commit moved the hyperhive pin from one rev to another, which is - /// what the placed agents are rebuilt for. The first commit moves nothing: - /// no agent was ever told to build from this repo. + /// The hyperhive rev `internal/meta` pins once the update has run. + pub(crate) pin: String, + /// The commit moved the hyperhive pin from one rev to another. The first + /// commit moves nothing. The placed agents are rebuilt for the pin + /// `meta_fanout` has not recorded; this decides only while that record + /// cannot be read. pub(crate) hyperhive_moved: bool, } @@ -258,6 +261,7 @@ pub(crate) async fn update( if flake_file.is_some_and(|f| f.bytes == flake.as_bytes()) && sha_of("flake.lock").is_some() { return Ok(Report { updated: Updated::Unchanged, + pin: pin.rev, hyperhive_moved: false, }); } @@ -289,6 +293,7 @@ pub(crate) async fn update( .await?; Ok(Report { updated: Updated::Committed, + pin: pin.rev, hyperhive_moved: pin.moved, }) } @@ -713,6 +718,7 @@ mod tests { got, Report { updated: Updated::Committed, + pin: rev('1'), hyperhive_moved: false } ); @@ -757,6 +763,7 @@ mod tests { let locker = FakeLocker::locked(); let got = run(&forge, &locker, &settings()).await; assert_eq!(got.updated, Updated::Unchanged); + assert_eq!(got.pin, rev('1')); assert!(!got.hyperhive_moved); assert!(locker.sent.lock().unwrap().is_empty()); assert!(forge.commits.lock().unwrap().is_empty()); @@ -865,6 +872,7 @@ mod tests { got, Report { updated: Updated::Committed, + pin: rev('3'), hyperhive_moved: true } );