Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
de68f175b5 | ||
|
|
784507f554 |
5 changed files with 429 additions and 70 deletions
|
|
@ -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` | 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#<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 |
|
||||
|
||||
<!-- vale write-good.Passive = YES -->
|
||||
|
||||
|
|
|
|||
|
|
@ -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<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.
|
||||
///
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<crate::coordinator::Coordinator>,
|
||||
agent: &str,
|
||||
) -> Result<Vec<hive_jobq::NodeId>> {
|
||||
) -> Result<Option<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_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.
|
||||
|
|
|
|||
|
|
@ -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<Coordinator>) -> Result<()> {
|
||||
|
|
@ -54,16 +72,29 @@ pub async fn pull(coord: &Arc<Coordinator>) -> 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<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)
|
||||
.await
|
||||
.context("opening this hive's wanted-state bucket")?
|
||||
else {
|
||||
|
|
@ -71,7 +102,7 @@ pub async fn pull(coord: &Arc<Coordinator>) -> 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<Coordinator>) -> 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<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, 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<Coordinator>,
|
||||
hive: String,
|
||||
mut shutdown: tokio::sync::watch::Receiver<bool>,
|
||||
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 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<Coordinator>, 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<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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue