vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:de68f175b5e5cc2a9fac6be2c5c102b2aed7674f
Branches Tags
vinzenz:main
..
compare: vinzenz:d989636d980803f6ecec599a4ff3cd622e0e8f0d
Branches Tags
vinzenz:main
5 changed files with 70 additions and 429 deletions
Download patch file Download diff file Expand all files Collapse all files

48
docs/scheduler/coordinator.md
View file

@ -53,30 +53,30 @@ Cheap — no build slot:
<!-- vale write-good.Passive = NO -->
| 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#<name>` 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#<name>` 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 |
<!-- vale write-good.Passive = YES -->

26
hive-c0re/src/job_queue/mod.rs
View file

@ -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<Option<Vec<NodeId>>> {
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.
///

26
hive-c0re/src/job_queue/tests.rs
View file

@ -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"
);
}

15
hive-c0re/src/swarm_status.rs
View file

@ -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<crate::coordinator::Coordinator>,
agent: &str,
) -> Result<Option<Vec<hive_jobq::NodeId>>> {
) -> Result<Vec<hive_jobq::NodeId>> {
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.

384
hive-c0re/src/workers/wanted.rs
View file

@ -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<Coordinator>) -> Result<()> {
@ -72,29 +54,16 @@ pub async fn pull(coord: &Arc<Coordinator>) -> 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<Coordinator>,
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<U, A, P>(attach: A, pull: P) -> Option<U>
where
A: Future<Output = Option<U>>,
P: Future<Output = Result<()>>,
{
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<F, Fut>(mut session: F, mut shutdown: tokio::sync::watch::Receiver<bool>)
where
F: FnMut() -> Fut,
Fut: Future<Output = SessionEnd>,
{
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<Coordinator>,
hive: String,
shutdown: tokio::sync::watch::Receiver<bool>,
) {
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<Coordinator>,
hive: &str,
mut shutdown: tokio::sync::watch::Receiver<bool>,
) -> 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<Coordinator>, 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<u8> = 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<u64> = (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));
}
}
}