diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index b0c63070..fe27b098 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -89,6 +89,15 @@ pub fn spawn( // the other was still down. `async_nats::Client` is a handle, so the // clone is cheap. tokio::spawn(drain_swarm_events( + client.clone(), + std::sync::Arc::clone(&coord), + hive.clone(), + shutdown.clone(), + )); + + // The wanted-state watch rides the same connection for the same + // reason, and is a third consumer rather than a second connection. + tokio::spawn(crate::workers::wanted::watch_declarations( client.clone(), coord, hive.clone(), diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 4ade198a..ad1a90c1 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -81,6 +81,105 @@ pub async fn pull(coord: &Arc) -> Result<()> { converge(coord, &declared).await } +/// Converge again each time the controller republishes this hive's declaration. +/// +/// The **fast** path, and it does not replace [`pull`]: this hears only what is +/// published while it is listening, so a hive that was down still learns the +/// current declaration from the boot read. Both, not either. +/// +/// ⚠️ A refused watch and a quiet one are told apart here, unlike the core-NATS +/// subscriptions in [`crate::swarm_status`]: a watch is a `JetStream` consumer, +/// so the request is answered, and a hive lacking the grant gets `None` rather +/// than silence. That is why this warns instead of returning quietly. +pub async fn watch_declarations( + client: async_nats::Client, + coord: Arc, + hive: String, + mut shutdown: tokio::sync::watch::Receiver, +) { + use futures_util::StreamExt as _; + + let Some(mut updates) = swarm_queue_client::wanted::watch(&client, &hive).await else { + tracing::warn!( + %hive, + "wanted state: cannot watch this hive's bucket; changes will be \ + picked up at the next boot instead" + ); + return; + }; + tracing::info!(%hive, "wanted state: watching for declarations"); + + loop { + tokio::select! { + entry = updates.next() => { + let Some(entry) = entry else { + // Same reading as the sibling subscriptions: `async-nats` + // reconnects underneath a live watch, so an ended stream is + // the connection going away for good rather than a blip to + // spin on. + tracing::warn!(%hive, "wanted state: watch closed"); + return; + }; + match entry { + Ok(entry) => apply_entry(&coord, &hive, &entry).await, + // The watch survives one bad entry; the stream ending is + // the case above. + Err(e) => tracing::warn!(%hive, error = %e, "wanted state: watch error"), + } + } + _ = shutdown.changed() => { + tracing::info!("wanted state: shutdown signal received"); + return; + } + } + } +} + +/// Converge one watched update, or decline to. +/// +/// A delete is **not** a deletion order — the module docs' rule, and the reason +/// this is not simply "decode and converge": the controller removing the key +/// says nothing about the agents this hive runs, so acting on it would tear +/// down the very set that absence is defined not to touch. +async fn apply_entry( + coord: &Arc, + hive: &str, + entry: &async_nats::jetstream::kv::Entry, +) { + if !carries_a_declaration(entry.operation) { + tracing::info!(%hive, "wanted state: declaration withdrawn; nothing to converge"); + return; + } + let declared: HiveWanted = match serde_json::from_slice(&entry.value) { + Ok(declared) => declared, + // Warned rather than propagated: this end and the controller share one + // type, so a decode failure means they disagree about it — and the + // watch must keep running to pick up the next, possibly good, value. + Err(e) => { + tracing::warn!(%hive, error = %e, "wanted state: undecodable declaration"); + return; + } + }; + if let Err(e) = converge(coord, &declared).await { + tracing::warn!(%hive, error = ?e, "wanted state: converging a watched update failed"); + } +} + +/// Whether a watched operation carries a declaration to converge to. +/// +/// Pure, and separate from [`apply_entry`], so the module's "absence is not a +/// deletion order" rule is enforced by a test rather than only asserted in +/// prose — converging on a removed key is the one mistake here that would tear +/// down agents nobody asked to stop. +fn carries_a_declaration(operation: async_nats::jetstream::kv::Operation) -> bool { + use async_nats::jetstream::kv::Operation; + + match operation { + Operation::Put => true, + Operation::Delete | Operation::Purge => false, + } +} + /// Queue whatever the declaration asks for and this hive is not already doing. async fn converge(coord: &Arc, declared: &HiveWanted) -> Result<()> { // Fail closed, for the reason the deploy event's own arm gives: without @@ -223,7 +322,7 @@ fn decide(state: AgentState, present: bool, intent: Option) -> Converge mod tests { use std::collections::{BTreeMap, BTreeSet}; - use super::{Converge, Plan, decide, plan}; + use super::{Converge, Plan, carries_a_declaration, decide, plan}; use crate::power::Wanted; use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; @@ -333,6 +432,22 @@ mod tests { assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing); } + /// The watch's half of "absence is not a deletion order". `Put` is the + /// control: without it this would pass on a function that refused + /// everything, which would silently stop the fast path converging at all. + #[test] + fn only_a_put_carries_a_declaration_to_converge_to() { + use async_nats::jetstream::kv::Operation; + + assert!(carries_a_declaration(Operation::Put)); + for withdrawn in [Operation::Delete, Operation::Purge] { + assert!( + !carries_a_declaration(withdrawn), + "{withdrawn:?} must not converge" + ); + } + } + /// Version skew is handled one layer up, at the decode: `AgentState` is /// closed, so an unknown value never reaches [`decide`] — it fails the /// whole declaration in `swarm-queue-client`, which owns that test diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 5af35c63..3592bd38 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -154,6 +154,27 @@ pub async fn open_read_only( js.get_key_value(bucket(hive)).await.ok() } +/// Watch this hive's own declaration for changes. +/// +/// **Hive-side.** `None` on the same terms as [`open_read_only`] — no bucket +/// yet — plus one more: a watch is a `JetStream` *consumer*, so it needs a grant +/// a plain `get` does not. A hive missing `CONSUMER.CREATE` on its own stream +/// gets `None` here while `get` keeps working, which is why the caller must +/// treat this as "not watching yet" and retry rather than as a dead end. +/// +/// Updates only, deliberately: the boot-time read already has the current +/// value, and a watch that replayed history would re-converge the whole +/// declaration on every reconnect for nothing. +#[cfg(feature = "kv")] +pub async fn watch( + client: &async_nats::Client, + hive: &str, +) -> Option { + // The bucket holds exactly one key, named for the hive — same key + // `open_read_only`'s caller reads, so both paths address one declaration. + open_read_only(client, hive).await?.watch(hive).await.ok() +} + #[cfg(test)] mod tests { use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket};