diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 6b4cbcdb..a8dbc8cf 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` | 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 | +| 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 | diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 0533c4d6..65caafe4 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -190,32 +190,6 @@ 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 bd2f42ce..8247e104 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1537,29 +1537,3 @@ 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 d2585e33..6cff2da4 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -322,18 +322,14 @@ 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(Some(_)) => { + Ok(_) => { 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"), } } @@ -381,17 +377,16 @@ 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_first_deploy_once(agent) + coord + .job_queue + .insert_job(|b| crate::job_queue::templates::first_deploy(b, 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 28b599b5..af7bd5ca 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -27,11 +27,10 @@ //! 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 or watch attach, because the controller owns that. +//! declared `Up` returns at the next boot, 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}; @@ -39,19 +38,6 @@ 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 @@ -61,10 +47,6 @@ const MAX_RETRY_DELAY: Duration = Duration::from_mins(5); /// 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<()> { @@ -72,29 +54,16 @@ pub async fn pull(coord: &Arc) -> Result<()> { tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read"); return Ok(()); }; - 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(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)?; -/// [`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) + let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive) .await .context("opening this hive's wanted-state bucket")? else { @@ -102,7 +71,7 @@ async fn pull_with( return Ok(()); }; let Some(raw) = store - .get(hive) + .get(&hive) .await .context("reading this hive's wanted state")? else { @@ -115,90 +84,11 @@ async fn pull_with( 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. 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. +/// 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, @@ -208,68 +98,45 @@ pub async fn watch_declarations( client: async_nats::Client, coord: Arc, hive: String, - 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 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; + 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; } - 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 - } + 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; } }; - 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 { - // `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 }; + // 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, + 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"), @@ -277,7 +144,7 @@ async fn watch_session( } _ = shutdown.changed() => { tracing::info!("wanted state: shutdown signal received"); - return SessionEnd::Shutdown; + return; } } } @@ -388,9 +255,7 @@ 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(Some(_)) => queued = true, - // Already queued or running: the pull and the watch overlap. - Ok(None) => {} + Ok(_) => queued = true, // 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"), @@ -946,171 +811,4 @@ 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)); - } - } }