diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index a8dbc8cf..6b4cbcdb 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -53,30 +53,30 @@ Cheap — no build slot: -| Node | Behavior | -| -------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `MetaSync` | the rebuild's meta preamble — rebuild-dir prep, idempotent meta `sync_agents`, optional per-agent relock. Holds the `MetaWindow` resource (below); deliberately its own node so the window never covers `Prebuild`'s multi-minute build | -| `Provision` | first-spawn pre-create provisioning — proposed/applied repos, state subvolume, meta registration (`sync_agents`); runs ahead of `Create` so the `nixos-container create --flake meta#` ref resolves. Store/meta-only, no container yet | -| `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop | -| `Start` | mechanical container start — runtime dir + drop-ins, `start_with_fallback`, MCP listener registration, the manager kick. Fanned out by a `Reconcile` that observed `wanted = Up` and the container down | -| `Stop` | mechanical container stop — `nixos-container` kill, MCP listener unregister, the `Killed` manager notify. Fanned out by a `Reconcile` that observed `wanted = Offline` and up | -| `StopForUpdate` | mechanical `nixos-container stop` for the profile swap; never touches `wanted`; noop if already stopped | -| `RebuildBookkeeping` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the DAG's `EmitRebuilt` tail node emits the `Rebuilt` manager event, not here). Split out of `Swap` for dashboard visibility + retry granularity, declares no resources of its own — a coordinated child of the `AgentWindow` brace | -| `AgentWindow` | pure resource holder — the brace for one agent's rebuild. Declares the build slot + agent lease atomically and holds both for its whole subtree, so `Prebuild` and the `Signal`→`Drain` quiesce window run concurrently instead of one nested under the other. Performs no work; see _Braces_ | -| `Signal` | set the graceful fence + kick, so the harness runs one stop-checkpoint turn | -| `Drain` | await the harness clearing the fence, bounded by the 3-min graceful-stop timeout; resolves ok either way | -| `PauseSignal` | write the pause marker + mark `pause_pending`. No kick, unlike `Signal` — the harness's between-turns poll is already responsive enough, and `Signal`'s kick-message body ("you were just (re)started") would be actively misleading here | -| `PauseDrain` | await the harness reporting `PauseAcknowledged`, bounded timeout; best-effort like `Drain` | -| `DestroyContainer` | `nixos-container destroy` + un-registration (drop from the roster, clear the ephemeral runtime dir). Runs downstream of a `Stop`, so deliberately excluded from `takes_container_down` — the container is already down by the time it claims | -| `PurgeState` | the `purge = true` half of a destroy: delete the agent's state subvolume (via hive-priv) plus its state/applied dirs. Own node because it's conditional and the irreversible step | -| `DestroyBookkeeping` | the post-destroy tail — meta sync, drop the power intent, notify the manager, rescan, re-emit the tombstone. Same split rationale as `RebuildBookkeeping`/`Swap`. The destructive work is `PurgeState`'s | -| `SetWanted` | write the durable power intent (`wanted = Up`/`Offline`) as the head node of a power-op DAG. Takes the agent lease even though it's a store write, so the intent write and the tail `Reconcile` are atomic per-agent — two racing power ops can't clobber each other's intent before either reconciles | -| `EmitRebuilt` | tail of a rebuild — emit the agent's `Rebuilt` manager event (ok/fail per outcome, nothing on cancel). One node per agent _and_ per outcome | -| `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | -| `ForgeSweep` | one-shot boot-time forge user/token sweep for every container (`forge::ensure_all`) as a first-class node, so it shows as real work on the dashboard instead of running invisibly in a bare `tokio::spawn`. Agentless | -| `MatrixSweep` | matrix user/space sweep (`matrix::ensure_all`): the boot-time instance, plus one every 30 min from a loop in `main.rs`. Holds `Resource::MatrixSweep` (capacity 1), so two passes never overlap; each tick queues its own pass, which waits for the resource if one is already live. Agentless | -| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree; each trigger queues its own pass, which waits for the resource if one is already live. Agentless | -| `WantedPull` | one-shot boot-time pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. No background loop behind this one — boot is the whole cadence; the deploy event (`swarm_status`) is the fast path, this repairs a missed one. Agentless | +| Node | Behavior | +| -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `MetaSync` | the rebuild's meta preamble — rebuild-dir prep, idempotent meta `sync_agents`, optional per-agent relock. Holds the `MetaWindow` resource (below); deliberately its own node so the window never covers `Prebuild`'s multi-minute build | +| `Provision` | first-spawn pre-create provisioning — proposed/applied repos, state subvolume, meta registration (`sync_agents`); runs ahead of `Create` so the `nixos-container create --flake meta#` ref resolves. Store/meta-only, no container yet | +| `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop | +| `Start` | mechanical container start — runtime dir + drop-ins, `start_with_fallback`, MCP listener registration, the manager kick. Fanned out by a `Reconcile` that observed `wanted = Up` and the container down | +| `Stop` | mechanical container stop — `nixos-container` kill, MCP listener unregister, the `Killed` manager notify. Fanned out by a `Reconcile` that observed `wanted = Offline` and up | +| `StopForUpdate` | mechanical `nixos-container stop` for the profile swap; never touches `wanted`; noop if already stopped | +| `RebuildBookkeeping` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the DAG's `EmitRebuilt` tail node emits the `Rebuilt` manager event, not here). Split out of `Swap` for dashboard visibility + retry granularity, declares no resources of its own — a coordinated child of the `AgentWindow` brace | +| `AgentWindow` | pure resource holder — the brace for one agent's rebuild. Declares the build slot + agent lease atomically and holds both for its whole subtree, so `Prebuild` and the `Signal`→`Drain` quiesce window run concurrently instead of one nested under the other. Performs no work; see _Braces_ | +| `Signal` | set the graceful fence + kick, so the harness runs one stop-checkpoint turn | +| `Drain` | await the harness clearing the fence, bounded by the 3-min graceful-stop timeout; resolves ok either way | +| `PauseSignal` | write the pause marker + mark `pause_pending`. No kick, unlike `Signal` — the harness's between-turns poll is already responsive enough, and `Signal`'s kick-message body ("you were just (re)started") would be actively misleading here | +| `PauseDrain` | await the harness reporting `PauseAcknowledged`, bounded timeout; best-effort like `Drain` | +| `DestroyContainer` | `nixos-container destroy` + un-registration (drop from the roster, clear the ephemeral runtime dir). Runs downstream of a `Stop`, so deliberately excluded from `takes_container_down` — the container is already down by the time it claims | +| `PurgeState` | the `purge = true` half of a destroy: delete the agent's state subvolume (via hive-priv) plus its state/applied dirs. Own node because it's conditional and the irreversible step | +| `DestroyBookkeeping` | the post-destroy tail — meta sync, drop the power intent, notify the manager, rescan, re-emit the tombstone. Same split rationale as `RebuildBookkeeping`/`Swap`. The destructive work is `PurgeState`'s | +| `SetWanted` | write the durable power intent (`wanted = Up`/`Offline`) as the head node of a power-op DAG. Takes the agent lease even though it's a store write, so the intent write and the tail `Reconcile` are atomic per-agent — two racing power ops can't clobber each other's intent before either reconciles | +| `EmitRebuilt` | tail of a rebuild — emit the agent's `Rebuilt` manager event (ok/fail per outcome, nothing on cancel). One node per agent _and_ per outcome | +| `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | +| `ForgeSweep` | one-shot boot-time forge user/token sweep for every container (`forge::ensure_all`) as a first-class node, so it shows as real work on the dashboard instead of running invisibly in a bare `tokio::spawn`. Agentless | +| `MatrixSweep` | matrix user/space sweep (`matrix::ensure_all`): the boot-time instance, plus one every 30 min from a loop in `main.rs`. Holds `Resource::MatrixSweep` (capacity 1), so two passes never overlap; each tick queues its own pass, which waits for the resource if one is already live. Agentless | +| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree; each trigger queues its own pass, which waits for the resource if one is already live. Agentless | +| `WantedPull` | pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. Runs at boot, waiting up to a minute in total for the queue connection, and again after every attach of the wanted-state watch (which retries with a backoff when the attach fails or the stream ends); the deploy event (`swarm_status`) and the watch are the fast path, this repairs a missed one. A first deploy already queued or running for an agent isn't queued again. Agentless | diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 65caafe4..0533c4d6 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -190,6 +190,32 @@ impl JobQueue { Ok(named) } + /// Queue `agent`'s first deploy unless one is already queued or running. + /// + /// `None` means a `Provision` for the agent is still live. The check and + /// the insert share one lock, so two callers racing (the wanted-state pull + /// and the watch, or the deploy event) queue one deploy between them. A + /// deploy that finished or failed does not count: a lost one is retried. + /// + /// # Errors + /// Propagates a graph-insert error. + pub fn insert_first_deploy_once(&self, agent: &str) -> anyhow::Result>> { + let mut inner = self.lock(); + let live = inner.graph().nodes().any(|n| { + !n.state.is_terminal() + && matches!(&n.payload, model::NodeKind::Provision { agent: a } if a == agent) + }); + if live { + return Ok(None); + } + let named = inner + .insert_job(None, |b| templates::first_deploy(b, agent)) + .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; + drop(inner); + self.notify.notify_one(); + Ok(Some(named)) + } + /// The scheduler itself, for `hive_jobq`'s run-loop seam /// (`Scheduler::claim_next`), which takes exactly this type. /// diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 8247e104..bd2f42ce 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1537,3 +1537,29 @@ fn each_sweep_holds_its_own_resource() { vec![vec![Resource::KnowledgeTree]] ); } + +/// The pull, the watch and the deploy event can all see one declaration, so a +/// second first deploy for an agent that has one queued must not be inserted. +#[test] +fn a_first_deploy_is_not_queued_twice_for_one_agent() { + let q = JobQueue::new(1); + let first = q.insert_first_deploy_once("newbie").expect("insert"); + assert!(first.is_some(), "control: the first deploy is queued"); + let kinds = || declared_shape(&q).len(); + let queued = kinds(); + + assert!( + q.insert_first_deploy_once("newbie") + .expect("insert") + .is_none(), + "a live first deploy for the same agent blocks another" + ); + assert_eq!(kinds(), queued, "nothing was added"); + + assert!( + q.insert_first_deploy_once("other") + .expect("insert") + .is_some(), + "control: another agent's deploy is independent" + ); +} diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 6cff2da4..d2585e33 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -322,14 +322,18 @@ async fn handle_deploy_request( coord .job_queue .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true)) + .map(Some) } else { queue_first_deploy(coord, &agent) }; match inserted { - Ok(_) => { + Ok(Some(_)) => { tracing::info!(%agent, known, "swarm events: deploy requested by the swarm, queued"); coord.emit_rebuild_queue_snapshot(); } + Ok(None) => { + tracing::info!(%agent, "swarm events: first deploy already queued; not queueing another"); + } Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"), } } @@ -377,16 +381,17 @@ async fn apply_rev(agent: &str, rev: &str) -> bool { /// Setting the row up front closes that window. A second caller open-coding /// the insert would lose exactly that, and the agent would come up stopped for /// no visible reason. +/// +/// `Ok(None)` when a first deploy for the agent is already queued or running: +/// the pull, the watch and the deploy event can all see the same declaration. pub(crate) fn queue_first_deploy( coord: &std::sync::Arc, agent: &str, -) -> Result> { +) -> Result>> { if let Err(e) = coord.power.set(agent, crate::power::Wanted::Up) { tracing::warn!(%agent, error = ?e, "agent_power: seed on swarm first-deploy failed"); } - coord - .job_queue - .insert_job(|b| crate::job_queue::templates::first_deploy(b, agent)) + coord.job_queue.insert_first_deploy_once(agent) } /// Offer one snapshot: this hive's current readiness, under its own key. diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index af7bd5ca..28b599b5 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -27,10 +27,11 @@ //! behind that reconcile's back on every boot. //! //! One consequence, deliberate: a declaration outranks a local stop — an agent -//! declared `Up` returns at the next boot, because the controller owns that. +//! declared `Up` returns at the next boot or watch attach, because the controller owns that. use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use std::time::Duration; use anyhow::{Context, Result}; use swarm_wire::wanted::{AgentState, HiveWanted}; @@ -38,6 +39,19 @@ use swarm_wire::wanted::{AgentState, HiveWanted}; use crate::coordinator::Coordinator; use crate::power::Wanted; +/// Longest `pull`, and each watch session's attach and first pull together, wait +/// for the swarm queue to connect. +const CONNECT_WAIT: Duration = Duration::from_mins(1); + +/// How often the wait for a connected client looks at the connection state. +const CONNECT_POLL: Duration = Duration::from_millis(250); + +/// Wait after the first failed watch session; doubled after each further one. +const FIRST_RETRY_DELAY: Duration = Duration::from_secs(5); + +/// The longest wait between two watch sessions. +const MAX_RETRY_DELAY: Duration = Duration::from_mins(5); + /// Read this hive's declaration and converge the agents it names. /// /// `Ok(())` covers three quiet cases that are **not** an empty declaration: no @@ -47,6 +61,10 @@ use crate::power::Wanted; /// this hive cannot enumerate — is an error, so the node fails visibly on the /// dashboard instead of reading as "nothing was declared". /// +/// Waits up to [`CONNECT_WAIT`] in total, for the client and then for its +/// connection: at boot the shared client exists before it is connected, and a +/// pull that did not wait would fail every boot. +/// /// # Errors /// Propagates a failed queue connect, KV read, decode, or agent enumeration. pub async fn pull(coord: &Arc) -> Result<()> { @@ -54,16 +72,29 @@ pub async fn pull(coord: &Arc) -> Result<()> { tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read"); return Ok(()); }; - let Some(client) = crate::swarm_queue::client().await else { - // Absent or failed — either way already reported by the shared - // connector (an `info` log or a `swarm_queue_config` banner). - return Ok(()); - }; - // An unconnected client does not fail a JetStream request, it hangs on it. - // A hung node is worse than a failed one: the dashboard shows it running. - swarm_queue_client::ensure_connected(&client)?; + let deadline = tokio::time::Instant::now() + CONNECT_WAIT; + let client = + match tokio::time::timeout_at(deadline, crate::swarm_queue::wait_for_client()).await { + // Absent — already reported by the shared connector (an `info` log or + // a `swarm_queue_config` banner). + Ok(None) => return Ok(()), + Ok(Some(client)) => client, + Err(_) => anyhow::bail!("no swarm queue client within {}s", CONNECT_WAIT.as_secs()), + }; + pull_with(coord, &client, &hive, deadline).await +} - let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive) +/// [`pull`] over a client the caller already has, waiting for its connection +/// no later than `deadline`. +async fn pull_with( + coord: &Arc, + client: &async_nats::Client, + hive: &str, + deadline: tokio::time::Instant, +) -> Result<()> { + wait_connected(client, deadline).await?; + + let Some(store) = swarm_queue_client::wanted::open_read_only(client, hive) .await .context("opening this hive's wanted-state bucket")? else { @@ -71,7 +102,7 @@ pub async fn pull(coord: &Arc) -> Result<()> { return Ok(()); }; let Some(raw) = store - .get(&hive) + .get(hive) .await .context("reading this hive's wanted state")? else { @@ -84,11 +115,90 @@ pub async fn pull(coord: &Arc) -> Result<()> { converge(coord, &declared).await } +/// Wait until `deadline` for `client` to be connected. +/// +/// `swarm_queue::wait_for_client` returns a client that may still be +/// connecting. An unconnected client does not fail a `JetStream` request, it +/// hangs on it, and a hung node shows as running on the dashboard, so every +/// request here is preceded by this bounded wait and its +/// `ensure_connected` verdict. +async fn wait_connected(client: &async_nats::Client, deadline: tokio::time::Instant) -> Result<()> { + while swarm_queue_client::ensure_connected(client).is_err() + && tokio::time::Instant::now() < deadline + { + tokio::time::sleep(CONNECT_POLL).await; + } + Ok(swarm_queue_client::ensure_connected(client)?) +} + +/// How one watch session ended. +#[derive(Debug, PartialEq, Eq)] +enum SessionEnd { + /// The hive is shutting down. + Shutdown, + /// The watch could not be attached (`attached == false`) or its stream + /// ended. + Lost { attached: bool }, +} + +/// The wait before the next watch session after `failures` sessions in a row +/// never attached: [`FIRST_RETRY_DELAY`] doubling up to [`MAX_RETRY_DELAY`]. +fn retry_delay(failures: u32) -> Duration { + FIRST_RETRY_DELAY + .saturating_mul(1_u32 << failures.min(16)) + .min(MAX_RETRY_DELAY) +} + +/// Attach the watch, then pull once, and hand back the watch. +/// +/// The watch comes first: a put landing between a pull and a later attach +/// would be seen by neither. One landing between the two is seen by both, which +/// `converge` tolerates. `None` when the attach did not happen; a failed pull is +/// logged and does not drop the watch, since the next attach pulls again. +async fn attach_then_pull(attach: A, pull: P) -> Option +where + A: Future>, + P: Future>, +{ + let updates = attach.await?; + if let Err(e) = pull.await { + tracing::warn!(error = ?e, "wanted state: pull after attaching the watch failed"); + } + Some(updates) +} + +/// Run watch sessions until shutdown, waiting [`retry_delay`] between them. +async fn supervise(mut session: F, mut shutdown: tokio::sync::watch::Receiver) +where + F: FnMut() -> Fut, + Fut: Future, +{ + let mut failures = 0_u32; + loop { + match session().await { + SessionEnd::Shutdown => return, + SessionEnd::Lost { attached } => { + failures = if attached { + 0 + } else { + failures.saturating_add(1) + }; + } + } + tokio::select! { + () = tokio::time::sleep(retry_delay(failures)) => {} + _ = shutdown.changed() => return, + } + } +} + /// 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. +/// The **fast** path. Every attach, the first included, is followed by one +/// [`pull`]: the watch hears only what is published while it is attached, so +/// what was declared before an attach (a hive that was down, a deploy lost to +/// a restart) would otherwise wait for the next boot. A failed attach or an +/// ended stream is retried with a bounded backoff, forever. /// /// ⚠️ 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, @@ -98,45 +208,68 @@ pub async fn watch_declarations( client: async_nats::Client, coord: Arc, hive: String, - mut shutdown: tokio::sync::watch::Receiver, + shutdown: tokio::sync::watch::Receiver, ) { + Box::pin(supervise( + || watch_session(&client, &coord, &hive, shutdown.clone()), + shutdown.clone(), + )) + .await; +} + +/// One attach, one pull, then the updates until the stream ends. +async fn watch_session( + client: &async_nats::Client, + coord: &Arc, + hive: &str, + mut shutdown: tokio::sync::watch::Receiver, +) -> SessionEnd { use futures_util::StreamExt as _; - let mut updates = match swarm_queue_client::wanted::watch(&client, &hive).await { - Ok(Some(updates)) => updates, - Ok(None) => { - tracing::warn!( - %hive, - "wanted state: cannot watch this hive's bucket; changes will be \ - picked up at the next boot instead" - ); - return; + let deadline = tokio::time::Instant::now() + CONNECT_WAIT; + let attach = async { + if let Err(e) = wait_connected(client, deadline).await { + tracing::warn!(%hive, error = ?e, "wanted state: cannot watch this hive's bucket; retrying"); + return None; } - Err(e) => { - tracing::warn!( - %hive, - error = %swarm_queue_client::chain(&e), - "wanted state: cannot watch this hive's bucket; changes will be \ - picked up at the next boot instead" - ); - return; + match swarm_queue_client::wanted::watch(client, hive).await { + Ok(Some(updates)) => Some(updates), + Ok(None) => { + tracing::warn!(%hive, "wanted state: cannot watch this hive's bucket; retrying"); + None + } + Err(e) => { + tracing::warn!( + %hive, + error = %swarm_queue_client::chain(&e), + "wanted state: cannot watch this hive's bucket; retrying" + ); + None + } } }; + let Some(mut updates) = Box::pin(attach_then_pull( + attach, + pull_with(coord, client, hive, deadline), + )) + .await + else { + return SessionEnd::Lost { attached: false }; + }; 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; + // `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; re-attaching"); + return SessionEnd::Lost { attached: true }; }; match entry { - Ok(entry) => apply_entry(&coord, &hive, &entry).await, + 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"), @@ -144,7 +277,7 @@ pub async fn watch_declarations( } _ = shutdown.changed() => { tracing::info!("wanted state: shutdown signal received"); - return; + return SessionEnd::Shutdown; } } } @@ -255,7 +388,9 @@ async fn converge(coord: &Arc, declared: &HiveWanted) -> Result<()> let mut queued = false; for agent in &plan.deploy { match crate::swarm_status::queue_first_deploy(coord, agent) { - Ok(_) => queued = true, + Ok(Some(_)) => queued = true, + // Already queued or running: the pull and the watch overlap. + Ok(None) => {} // Per agent rather than the whole sweep: one agent that cannot be // queued is not a reason to leave the others undeclared. Err(e) => tracing::warn!(%agent, error = %e, "wanted state: queueing a deploy failed"), @@ -811,4 +946,171 @@ mod tests { ); } } + + /// A client that never connects: initial-connect retries run in the + /// background and `ensure_connected` keeps failing. + async fn never_connected() -> async_nats::Client { + async_nats::ConnectOptions::new() + .retry_on_initial_connect() + .connect("nats://127.0.0.1:1") + .await + .expect("a client that retries its first connect is returned unconnected") + } + + /// One deadline bounds the whole connect wait: time already spent before + /// `wait_connected` counts against it, so `pull`'s client wait and its + /// connection wait together stay inside one `CONNECT_WAIT`. + #[tokio::test(start_paused = true)] + async fn the_connect_wait_is_bounded_by_one_deadline() { + let client = never_connected().await; + let start = tokio::time::Instant::now(); + let deadline = start + super::CONNECT_WAIT; + tokio::time::sleep(std::time::Duration::from_secs(59)).await; + assert!(super::wait_connected(&client, deadline).await.is_err()); + assert!( + start.elapsed() <= super::CONNECT_WAIT + super::CONNECT_POLL, + "waited {:?} in total", + start.elapsed() + ); + } + + mod watch_supervision { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use super::super::{MAX_RETRY_DELAY, SessionEnd, attach_then_pull, retry_delay, supervise}; + + #[tokio::test] + async fn the_watch_attaches_before_the_pull_runs() { + let log = Arc::new(Mutex::new(Vec::new())); + let (a, p) = (Arc::clone(&log), Arc::clone(&log)); + let got = attach_then_pull( + async move { + a.lock().unwrap().push("attach"); + Some(7) + }, + async move { + p.lock().unwrap().push("pull"); + Ok(()) + }, + ) + .await; + assert_eq!(got, Some(7)); + assert_eq!(*log.lock().unwrap(), ["attach", "pull"]); + } + + #[tokio::test] + async fn a_failed_attach_does_not_pull() { + let log = Arc::new(Mutex::new(Vec::new())); + let p = Arc::clone(&log); + let got: Option = attach_then_pull(async { None }, async move { + p.lock().unwrap().push("pull"); + Ok(()) + }) + .await; + assert_eq!(got, None); + assert!(log.lock().unwrap().is_empty(), "no watch, no pull"); + } + + #[tokio::test] + async fn a_failed_pull_keeps_the_watch() { + let got = attach_then_pull(async { Some(1) }, async { anyhow::bail!("down") }).await; + assert_eq!(got, Some(1)); + } + + #[tokio::test(start_paused = true)] + async fn every_re_attach_pulls_again() { + let log = Arc::new(Mutex::new(Vec::new())); + let (_tx, shutdown) = tokio::sync::watch::channel(false); + let l = Arc::clone(&log); + supervise( + move || { + let l = Arc::clone(&l); + async move { + let (a, p) = (Arc::clone(&l), Arc::clone(&l)); + let _ = attach_then_pull( + async move { + a.lock().unwrap().push("attach"); + Some(()) + }, + async move { + p.lock().unwrap().push("pull"); + Ok(()) + }, + ) + .await; + if l.lock().unwrap().len() >= 6 { + SessionEnd::Shutdown + } else { + SessionEnd::Lost { attached: true } + } + } + }, + shutdown, + ) + .await; + assert_eq!( + *log.lock().unwrap(), + ["attach", "pull", "attach", "pull", "attach", "pull"] + ); + } + + #[tokio::test(start_paused = true)] + async fn failed_attaches_are_retried_not_abandoned() { + let tries = Arc::new(Mutex::new(0_u32)); + let (_tx, shutdown) = tokio::sync::watch::channel(false); + let t = Arc::clone(&tries); + supervise( + move || { + let t = Arc::clone(&t); + async move { + let mut n = t.lock().unwrap(); + *n += 1; + if *n >= 5 { + SessionEnd::Shutdown + } else { + SessionEnd::Lost { attached: false } + } + } + }, + shutdown, + ) + .await; + assert_eq!(*tries.lock().unwrap(), 5); + } + + #[tokio::test(start_paused = true)] + async fn an_attach_resets_the_backoff() { + let start = tokio::time::Instant::now(); + let starts = Arc::new(Mutex::new(Vec::new())); + let (_tx, shutdown) = tokio::sync::watch::channel(false); + let s = Arc::clone(&starts); + supervise( + move || { + let s = Arc::clone(&s); + async move { + let mut s = s.lock().unwrap(); + s.push(start.elapsed().as_secs()); + match s.len() { + 1 | 2 => SessionEnd::Lost { attached: false }, + 3 => SessionEnd::Lost { attached: true }, + _ => SessionEnd::Shutdown, + } + } + }, + shutdown, + ) + .await; + // Waits: 10 and 20 after two misses, then 5 once an attach worked. + assert_eq!(*starts.lock().unwrap(), [0, 10, 30, 35]); + } + + #[test] + fn the_retry_delay_doubles_and_is_capped() { + let secs: Vec = (0..8).map(|n| retry_delay(n).as_secs()).collect(); + assert_eq!(secs, [5, 10, 20, 40, 80, 160, 300, 300]); + assert_eq!(retry_delay(u32::MAX), MAX_RETRY_DELAY); + assert!(retry_delay(1_000) <= Duration::from_mins(5)); + } + } }