diff --git a/docs/2-handbook/agent-lifecycle/persistence.md b/docs/2-handbook/agent-lifecycle/persistence.md index d9b864d0..a24b5941 100644 --- a/docs/2-handbook/agent-lifecycle/persistence.md +++ b/docs/2-handbook/agent-lifecycle/persistence.md @@ -42,29 +42,23 @@ schemas, file layouts, and internal migration mechanics. ### `/var/lib/hyperhive/db/broker.sqlite` (host) -Three tables, all in one file — the message queue, a small key/value -table, and the per-agent power-intent registry: +One table, the message queue: - `messages` — every inter-agent / operator-authored message. `sender / recipient / body / sent_at / delivered_at / acked_at / in_reply_to / priority`. `in_reply_to` links a reply to its parent row id; the dashboard and per-agent inbox render these as threaded rows. -- `kv` — small persistent key/value store (`key PK / value`) for - host-side bookkeeping that doesn't warrant its own table. - ⚠️ The `mcp__hyperhive__remind` queue lives in a harness-local, - per-agent store — see [`/harness/` contents - below](#state-dirs-per-agent) for where reminders (and todos) - live. -- `agent_power` — one row per agent: `agent PK / wanted (up | - offline) / updated_at`, owned by `hive-c0re/src/stores/power.rs`. - This is the durable power *intent* the job queue reconciles the - observed container state against; intent survives hive-c0re - restarts even though in-flight queue work doesn't. See - [`docs/4-implementation/scheduler/coordinator.md`'s Desired-state - section](../../4-implementation/scheduler/coordinator.md#desired-state-spec-vs-status) for who - writes and reads it and how reconciliation works. +⚠️ The `mcp__hyperhive__remind` queue lives in a harness-local, +per-agent store — see [`/harness/` contents +below](#state-dirs-per-agent) for where reminders (and todos) +live. + +The swarm owns power intent, so hive-c0re keeps none on disk. An older build +also kept a `kv` table (the agents running at the last `hivectl stop`) and an +`agent_power` table (the durable `wanted` intent) here; opening the file drops +both. On a host whose hive-c0re once kept scheduled prompts, the file also holds `scheduled_prompts` and `scheduled_prompt_targets`. Nothing opens @@ -80,8 +74,6 @@ Retention: rows are always kept — the harness `ack_turn`s only after a successful turn, so `requeue_inflight` can still requeue an unacked row on a crash. -- `agent_power` rows live until the operator destroys the agent (one row per - agent — nothing to vacuum). ### `/harness/hyperhive-events.sqlite` (per agent) diff --git a/docs/2-handbook/tools/hivectl-cli.md b/docs/2-handbook/tools/hivectl-cli.md index d52fc4d0..f3b19a15 100644 --- a/docs/2-handbook/tools/hivectl-cli.md +++ b/docs/2-handbook/tools/hivectl-cli.md @@ -454,7 +454,7 @@ Bare `hivectl stop` stops every agent and the gateway; scope flags narrow it to Start containers hive-wide — the inverse of `hivectl stop`. -Bare `hivectl start` restores the agents stopped by the last broad-scope `stop` (or starts everything if none); scope flags narrow it. The swarm's forge and matrix are never in scope. +Bare `hivectl start` starts every agent; scope flags narrow it. The swarm's forge and matrix are never in scope. **Usage:** `hivectl start [OPTIONS]` diff --git a/docs/4-implementation/scheduler/coordinator.md b/docs/4-implementation/scheduler/coordinator.md index 0591c60b..f5b6523b 100644 --- a/docs/4-implementation/scheduler/coordinator.md +++ b/docs/4-implementation/scheduler/coordinator.md @@ -69,7 +69,7 @@ Cheap — no build slot: | `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 dir. 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 | +| `SetWanted` | record the power intent (`wanted = Up`/`Offline`, in memory) as the head node of a power-op DAG. Takes the agent lease even though it writes no container state, 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 | | `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 | @@ -82,7 +82,7 @@ and `agent_set::refresh` serializes itself with its own mutex. ### Every operation as a DAG -The `stop` / `start` power ops write the durable `wanted` intent via a head +The `stop` / `start` power ops record the in-memory `wanted` intent via a head `SetWanted` node (not a pre-submit side effect) — it holds the agent lease, so intent-write + reconcile is atomic per-agent. `restart` is the exception: it writes _no_ intent (no `SetWanted` head) — it bounces the container and @@ -138,11 +138,14 @@ Notable collapses: ### Desired-state (spec vs status) -Per-agent power _intent_ — `wanted: Up | Offline` — is durable as the -`agent_power` table in the coordinator DB (`hive-c0re/src/stores/power.rs`). +Per-agent power _intent_ — `wanted: Up | Offline` — is what the operator last +asked of an agent through a power op, held in memory by `PowerStore` +(`hive-c0re/src/stores/power.rs`) until hive-c0re restarts. Nothing is durable: +the swarm owns lasting intent, and declares each agent's state in the hive's +wanted bucket (`workers/wanted.rs`), which outranks the local intent. `container_view` remains the observed _status_; `Reconcile` nodes converge the two. Setting `wanted` is never a queued node: the power layer -(`job_queue/power.rs`) writes the row synchronously, then inserts the DAG +(`job_queue/power.rs`) writes it synchronously, then inserts the DAG whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins. Every operator power surface — dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill` — @@ -151,8 +154,12 @@ and crash-watch suppression can't drift per surface; the only direct starts left are the gateway's (no lease, no harness). Cancelling a still-queued power DAG reverts `wanted` to the observed state — a cancel means "don't do it," not "do it later." hive-c0re -seeds agents without a row from observed state on first touch (running ⇒ -`Up`); destroy removes the row. +seeds an agent without an intent from observed state on first touch (running ⇒ +`Up`); destroy removes it. After a restart no agent has an intent, so each one +is seeded from what its container is doing and nothing starts or stops by +itself; the swarm's declaration then starts the agents it declares `Up` and +stops the ones it declares `Offline`. A broad `hivectl start` starts every +agent in scope. The admin-socket responses carry the submitted DAG ids; `hivectl` polls `HostRequest::QueueNodes` (~1s) and prints a progress line per DAG — roll-up @@ -314,8 +321,8 @@ already attempted, so a closure that fails to deploy gets one try per process. The boot log line records the rebuild / deferred / up-to-date counts so the operator sees at a glance how much work the boot triggered. `auto_update::run` -seeds agents without an `agent_power` row from observed state during -classification (the one-time migration; thereafter the DB is authoritative). +seeds agents without an intent from observed state during classification, so a +boot with no operator action in this process never finds drift of its own. ## Meta flake diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 691c8060..8445931f 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -21,12 +21,6 @@ use crate::socket_server::{self, AgentSocket}; /// fine, the seq dedupe makes a reconnect resync safe. const DASHBOARD_CHANNEL: usize = 256; -/// Broker `kv` key under which the broad-stop running-agents snapshot is -/// persisted, so a `hivectl start` after a hive-c0re restart can still -/// restore only the previously-running agents. See -/// `Coordinator::set_last_stopped_running`. -const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running"; - pub struct Coordinator { pub broker: Arc, /// Full build-log capture. `lifecycle::run` `start()`s a row per attempt, @@ -107,17 +101,6 @@ pub struct Coordinator { /// orchestrations that can be in flight for different agents /// simultaneously. pause_pending: Mutex>, - /// Logical agent names that were running at the last broad-scope - /// `hivectl stop`. A subsequent broad-scope `hivectl start` restores - /// only this set (intersected with the requested scope) rather than - /// every configured container, so agents the operator intentionally - /// left stopped stay stopped. `None` when no broad stop has happened - /// since the last start (or since daemon boot) — start then falls - /// back to "start all". Mirrored to the broker `kv` table - /// (`last_stopped_running` key) so the snapshot also survives a - /// hive-c0re restart between the stop-all and the start: this - /// in-memory copy is the fast path, the persisted copy the backstop. - last_stopped_running: Mutex>>, /// Unified wire-facing event channel feeding the dashboard SSE /// stream. Carries broker messages (mirrored from `broker.subscribe` /// by the forwarder task in `main.rs`) and dashboard-only mutation @@ -365,7 +348,7 @@ impl Coordinator { // without us having to thread an `Arc` through every public entry // point in the lifecycle surface. crate::build_logs::install(build_logs.clone()); - let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?); + let power = Arc::new(crate::power::PowerStore::default()); let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL); let (shutdown_tx, _) = watch::channel(false); Ok(Self { @@ -382,7 +365,6 @@ impl Coordinator { recent_transient: Mutex::new(HashMap::new()), graceful_stop_pending: Mutex::new(HashSet::new()), pause_pending: Mutex::new(HashSet::new()), - last_stopped_running: Mutex::new(None), dashboard_events, event_seq: AtomicU64::new(0), @@ -677,48 +659,6 @@ impl Coordinator { self.pause_pending.lock().unwrap().remove(name); } - /// Record the set of agents that were running at a broad-scope - /// `hivectl stop`, so the next broad-scope `start` restores exactly - /// this set. See the `last_stopped_running` field doc. Persists a - /// copy to the broker `kv` table (best-effort) so the snapshot - /// survives a hive-c0re restart; the in-memory copy is the fast path. - pub fn set_last_stopped_running(&self, agents: Vec) { - match serde_json::to_string(&agents) { - Ok(json) => { - if let Err(e) = self.broker.kv_set(LAST_STOPPED_RUNNING_KEY, &json) { - tracing::warn!(error = ?e, "persist last_stopped_running failed (in-memory copy still set)"); - } - } - Err(e) => tracing::warn!(error = ?e, "serialise last_stopped_running failed"), - } - *self.last_stopped_running.lock().unwrap() = Some(agents); - } - - /// Take (and clear) the recorded broad-stop running set, if any. A - /// broad-scope `start` uses this to restore only the previously - /// running agents; `None` means "no record — start all". Falls back - /// to the persisted `kv` copy when the in-memory snapshot is empty - /// (hive-c0re restarted between the stop-all and the start). Clears - /// the persisted copy either way — the snapshot is one-shot. - pub fn take_last_stopped_running(&self) -> Option> { - let result = self - .last_stopped_running - .lock() - .unwrap() - .take() - .or_else(|| { - self.broker - .kv_get(LAST_STOPPED_RUNNING_KEY) - .ok() - .flatten() - .and_then(|json| serde_json::from_str::>(&json).ok()) - }); - if let Err(e) = self.broker.kv_delete(LAST_STOPPED_RUNNING_KEY) { - tracing::warn!(error = ?e, "clear persisted last_stopped_running failed"); - } - result - } - /// Per-agent `deliberate_stop` for transients cleared within the last /// `grace` seconds — i.e. agents the operator just acted on, whose stop the /// crash watcher should NOT classify as a crash. Lazily reaps entries older diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 962a8eb7..95cd594b 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -8,7 +8,7 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex, OnceLock}; -use anyhow::{Context as _, Result}; +use anyhow::Result; use swarm_wire::agent_closures::AgentClosure; use hive_jobq::NodeId; @@ -305,7 +305,10 @@ async fn dispatch( run_emit_rebuilt(host, coord, agent, coord.job_queue.root_of(id), *ok).await; Ok(()) } - NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up), + NodeKind::SetWanted { up, .. } => { + run_set_wanted(coord, agent, *up); + Ok(()) + } // Braces carry no work of their own; completing one lets it reach // `Finishing` so the nodes under it start. What they declare stays held // until their whole subtree settles. @@ -425,7 +428,7 @@ async fn run_purge_state(host: &impl Host, agent: &str) { /// Post-destroy bookkeeping. Infallible by construction: every step is /// warn-and-continue, because the destroy it follows has already succeeded and -/// none of this is undoable — a failed agent-set refresh or power-store write is a +/// none of this is undoable — a failed agent-set refresh is a /// bookkeeping drift to log, not a reason to red a DAG whose container is /// already gone. async fn run_destroy_bookkeeping(host: &impl Host, coord: &Arc, agent: &str) { @@ -433,11 +436,9 @@ async fn run_destroy_bookkeeping(host: &impl Host, coord: &Arc, age if let Err(e) = host.refresh_agent_set().await { tracing::warn!(error = ?e, %agent, "agent_set refresh after destroy failed"); } - // Drop the durable power intent — a future agent of the same name seeds - // fresh from its observed state. - if let Err(e) = coord.power.remove(agent) { - tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed"); - } + // Drop the power intent — a future agent of the same name seeds fresh from + // its observed state. + coord.power.remove(agent); host.notify( format!("destroyed:{agent}"), format!("agent '{agent}' destroyed"), @@ -467,25 +468,18 @@ async fn run_emit_rebuilt( host.notify(format!("rebuilt:{agent}"), summary).await; } -/// Write the agent's durable power intent — the DAG-node form of the old -/// pre-submit `set_wanted` side effect. Store-only (no container touch), so +/// Record the agent's power intent — the DAG-node form of the old pre-submit +/// `set_wanted` side effect. Memory-only (no container touch), so /// build-slot-exempt; but it declares the agent's lifecycle lease /// (`Resource::Agent`) so the whole power-op DAG is atomic per-agent. -/// The downstream `Reconcile` reads the intent this writes. Unlike the old -/// warn-and-continue write, a failed write fails the node (cancel-downstream -/// cancels the `Reconcile`) rather than letting it converge to a stale -/// intent — that atomicity is the point of moving it into the DAG. -fn run_set_wanted(coord: &Arc, agent: &str, up: bool) -> Result<()> { +/// The downstream `Reconcile` reads the intent this writes. +fn run_set_wanted(coord: &Arc, agent: &str, up: bool) { let wanted = if up { crate::power::Wanted::Up } else { crate::power::Wanted::Offline }; - coord - .power - .set(agent, wanted) - .with_context(|| format!("set wanted={} for agent {agent}", <&str>::from(wanted)))?; - Ok(()) + coord.power.set(agent, wanted); } /// The rebuild's preamble: runtime-dir prep, the archetype and agent-set @@ -719,7 +713,7 @@ async fn run_reconcile( name: &str, ) -> Result> { let running = host.is_running(name).await; - let wanted = coord.power.get_or_seed(name, running)?; + let wanted = coord.power.get_or_seed(name, running); Ok(match reconcile_action(wanted, running) { ReconcileAction::Start => Some(NodeKind::Start { agent: name.to_owned(), @@ -780,13 +774,11 @@ async fn run_stop(host: &impl Host, coord: &Arc, name: &str) -> Res /// noop when already stopped. async fn run_stop_for_update(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { if host.is_running(name).await { - // Seed a missing agent_power row from the PRE-stop observation + // Seed a missing intent from the PRE-stop observation // — the DAG's tail `Reconcile` observes only the mechanically // stopped state and would otherwise seed a running-but-unknown // agent as `Offline`, stranding it down after its own rebuild. - if let Err(e) = coord.power.get_or_seed(name, true) { - tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed"); - } + coord.power.get_or_seed(name, true); host.kill(name).await?; coord.rescan_containers_and_emit().await; } diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index 0d7592d2..7a3fa476 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -572,21 +572,18 @@ async fn stop_for_update_kills_a_running_container_and_seeds_its_intent_up() { .expect("stop ok"); assert_eq!(host.calls(), ["is_running alice", "kill alice"]); // Seeded from the pre-stop observation, so the tail Reconcile restarts it. - assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up)); + assert_eq!(coord.power.get("alice"), Some(Wanted::Up)); } #[tokio::test] async fn stop_for_update_keeps_an_existing_offline_intent() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Offline).expect("seed"); + coord.power.set("alice", Wanted::Offline); let host = FakeHost::up(); run_stop_for_update(&host, &coord, "alice") .await .expect("stop ok"); - assert_eq!( - coord.power.get("alice").expect("read"), - Some(Wanted::Offline) - ); + assert_eq!(coord.power.get("alice"), Some(Wanted::Offline)); } #[tokio::test] @@ -597,7 +594,7 @@ async fn stop_for_update_is_a_noop_on_a_stopped_container() { .await .expect("noop ok"); assert_eq!(host.calls(), ["is_running alice"]); - assert_eq!(coord.power.get("alice").expect("read"), None); + assert_eq!(coord.power.get("alice"), None); } #[tokio::test] @@ -607,7 +604,7 @@ async fn stop_for_update_failure_fails_the_node_after_seeding() { run_stop_for_update(&host, &coord, "alice") .await .expect_err("a failed stop fails the node"); - assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up)); + assert_eq!(coord.power.get("alice"), Some(Wanted::Up)); } // --- Reconcile -------------------------------------------------------------- @@ -615,7 +612,7 @@ async fn stop_for_update_failure_fails_the_node_after_seeding() { #[tokio::test] async fn reconcile_fans_out_start_for_wanted_up_and_down() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Up).expect("set"); + coord.power.set("alice", Wanted::Up); let sub = run_reconcile(&FakeHost::down(), &coord, "alice") .await .expect("ok"); @@ -625,7 +622,7 @@ async fn reconcile_fans_out_start_for_wanted_up_and_down() { #[tokio::test] async fn reconcile_fans_out_stop_for_wanted_offline_and_up() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Offline).expect("set"); + coord.power.set("alice", Wanted::Offline); let sub = run_reconcile(&FakeHost::up(), &coord, "alice") .await .expect("ok"); @@ -635,7 +632,7 @@ async fn reconcile_fans_out_stop_for_wanted_offline_and_up() { #[tokio::test] async fn reconcile_is_a_noop_when_state_matches_intent() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Up).expect("set"); + coord.power.set("alice", Wanted::Up); let sub = run_reconcile(&FakeHost::up(), &coord, "alice") .await .expect("ok"); @@ -649,7 +646,7 @@ async fn reconcile_seeds_a_missing_intent_from_the_observed_state() { .await .expect("ok"); assert!(sub.is_none()); - assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up)); + assert_eq!(coord.power.get("alice"), Some(Wanted::Up)); } // --- Set wanted ------------------------------------------------------------- @@ -657,13 +654,10 @@ async fn reconcile_seeds_a_missing_intent_from_the_observed_state() { #[tokio::test] async fn set_wanted_writes_the_intent() { let (_dir, coord) = coordinator(); - run_set_wanted(&coord, "alice", false).expect("ok"); - assert_eq!( - coord.power.get("alice").expect("read"), - Some(Wanted::Offline) - ); - run_set_wanted(&coord, "alice", true).expect("ok"); - assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up)); + run_set_wanted(&coord, "alice", false); + assert_eq!(coord.power.get("alice"), Some(Wanted::Offline)); + run_set_wanted(&coord, "alice", true); + assert_eq!(coord.power.get("alice"), Some(Wanted::Up)); } // --- Signal / Drain --------------------------------------------------------- @@ -1015,7 +1009,7 @@ async fn purge_removes_every_state_dir_even_when_the_subvolume_delete_fails() { #[tokio::test] async fn destroy_bookkeeping_drops_the_power_row_and_announces() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Offline).expect("set"); + coord.power.set("alice", Wanted::Offline); let host = FakeHost::up(); run_destroy_bookkeeping(&host, &coord, "alice").await; assert_eq!( @@ -1026,10 +1020,7 @@ async fn destroy_bookkeeping_drops_the_power_row_and_announces() { ] ); // The Offline row is gone, so a fresh seed from "running" gives Up. - assert_eq!( - coord.power.get_or_seed("alice", true).expect("seed"), - Wanted::Up - ); + assert_eq!(coord.power.get_or_seed("alice", true), Wanted::Up); } #[tokio::test] @@ -1257,10 +1248,7 @@ async fn dispatch_set_wanted_writes_the_intent_and_emit_rebuilt_announces() { }, ) .await; - assert_eq!( - coord.power.get("alice").expect("read"), - Some(Wanted::Offline) - ); + assert_eq!(coord.power.get("alice"), Some(Wanted::Offline)); let settled = run_dag( &host, &coord, @@ -1325,7 +1313,7 @@ async fn dispatch_reports_a_failing_executor_as_a_failed_node() { #[tokio::test] async fn reconcile_whose_start_fails_fails_the_dag_without_registering() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Up).expect("set"); + coord.power.set("alice", Wanted::Up); let host = FakeHost::down().failing("start"); let settled = run_dag(&host, &coord, alice(|agent| NodeKind::Reconcile { agent })).await; let by_kind = |k: &str| settled.iter().find(|s| s.0 == k).cloned(); @@ -1346,7 +1334,7 @@ async fn reconcile_whose_start_fails_fails_the_dag_without_registering() { #[tokio::test] async fn reconcile_fans_out_a_stop_that_kills_and_announces() { let (_dir, coord) = coordinator(); - coord.power.set("alice", Wanted::Offline).expect("set"); + coord.power.set("alice", Wanted::Offline); let host = FakeHost::up(); let settled = run_dag(&host, &coord, alice(|agent| NodeKind::Reconcile { agent })).await; assert!(settled.iter().any(|s| s.0 == "stop")); diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index e260a280..001acb1c 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -145,32 +145,10 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { // graceful-stop queue, to re-expand it). let agents = scoped_agents(scope).await?; let infra = scoped_infra(scope); - // On a broad stop, remember which agents were actually - // running so a later broad `start` restores only those - // (not every configured container). A targeted `--agent` - // stop must not redefine the restore set. - if is_broad_scope(scope) { - let mut running = Vec::new(); - for a in &agents { - if lifecycle::is_running(a).await { - running.push(a.clone()); - } - } - coord.set_last_stopped_running(running); - } handle_stop(&coord, &agents, &infra, *graceful).await? } HostRequest::Start { scope } => { - let mut agents = scoped_agents(scope).await?; - // A broad start restores only the set recorded at the - // last broad stop, if any. No record (cold "bring the - // hive up", or a daemon restart since the stop) → start - // all. Targeted `--agent` start is never filtered. - if is_broad_scope(scope) - && let Some(prev) = coord.take_last_stopped_running() - { - agents.retain(|a| prev.contains(a)); - } + let agents = scoped_agents(scope).await?; let infra = scoped_infra(scope); handle_start(&coord, &agents, &infra).await? } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 4b67ba9d..e7bc0cda 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -11,7 +11,7 @@ use chrono::Utc; use hive_sh4re::inbox::{InboxRow, Message}; use crate::db::Migration; -use rusqlite::{Connection, OptionalExtension, params}; +use rusqlite::{Connection, params}; use serde::Serialize; use tokio::sync::broadcast; @@ -29,10 +29,11 @@ CREATE TABLE IF NOT EXISTS messages ( CREATE INDEX IF NOT EXISTS idx_messages_undelivered ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL; -CREATE TABLE IF NOT EXISTS kv ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL -); +-- Tables an earlier build kept here: the running-agents snapshot (`kv`) and +-- the durable power intent (`agent_power`). The swarm owns power intent now, +-- so a file from that build sheds them on its next open. +DROP TABLE IF EXISTS kv; +DROP TABLE IF EXISTS agent_power; "; /// Capacity of the live event channel. Slow subscribers (e.g. an idle browser) @@ -163,52 +164,6 @@ impl Broker { self.events.subscribe() } - /// Set a small persistent key/value pair (upsert). The `kv` table is a - /// general single-value store for state that must survive a hive-c0re - /// restart but isn't worth a dedicated table — currently just the - /// `hivectl start` running-agents snapshot (see - /// `Coordinator::set_last_stopped_running`). - /// - /// # Errors - /// - /// Returns an error if the sqlite upsert fails. - pub fn kv_set(&self, key: &str, value: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "INSERT INTO kv (key, value) VALUES (?1, ?2) - ON CONFLICT(key) DO UPDATE SET value = excluded.value", - params![key, value], - )?; - Ok(()) - } - - /// Read a persistent key/value pair. `None` when the key is absent. - /// - /// # Errors - /// - /// Returns an error if the sqlite query fails. - pub fn kv_get(&self, key: &str) -> Result> { - let conn = self.conn.lock().unwrap(); - let value: Option = conn - .query_row("SELECT value FROM kv WHERE key = ?1", params![key], |row| { - row.get(0) - }) - .optional()?; - Ok(value) - } - - /// Delete a persistent key/value pair. Idempotent — deleting an absent - /// key is a no-op. - /// - /// # Errors - /// - /// Returns an error if the sqlite delete fails. - pub fn kv_delete(&self, key: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM kv WHERE key = ?1", params![key])?; - Ok(()) - } - pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); let now = Utc::now().timestamp(); @@ -664,6 +619,42 @@ mod tests { TmpBroker { path, broker } } + /// A file from the build that kept the running-agents snapshot (`kv`) and + /// the durable power intent (`agent_power`) in it sheds both tables on its + /// next open and keeps its messages. + #[test] + fn opening_an_older_file_drops_the_kv_and_power_tables() { + let first = open_broker(); + first.broker.send(&msg("alice", "bob", "kept")).unwrap(); + let path = first.path.clone(); + { + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE agent_power ( + agent TEXT PRIMARY KEY, wanted TEXT NOT NULL, updated_at INTEGER NOT NULL); + INSERT INTO kv VALUES ('last_stopped_running', '[]'); + INSERT INTO agent_power VALUES ('alice', 'up', 0);", + ) + .unwrap(); + } + + let reopened = Broker::open(&path).unwrap(); + let conn = Connection::open(&path).unwrap(); + let left: Vec = conn + .prepare("SELECT name FROM sqlite_master WHERE name IN ('kv', 'agent_power')") + .unwrap() + .query_map([], |r| r.get(0)) + .unwrap() + .collect::>() + .unwrap(); + assert!(left.is_empty(), "left behind: {left:?}"); + assert_eq!( + pop_one(&reopened, "bob").map(|d| d.message.body), + Some("kept".to_owned()) + ); + } + fn msg(from: &str, to: &str, body: &str) -> Message { Message { from: hive_types::Ident::parse(from).expect("test sender must be a valid ident"), @@ -681,24 +672,6 @@ mod tests { batch.pop() } - #[test] - fn kv_set_get_delete_roundtrip() { - let tb = open_broker(); - let b = &tb.broker; - // Absent key -> None. - assert_eq!(b.kv_get("k").unwrap(), None); - // Set then get. - b.kv_set("k", "v1").unwrap(); - assert_eq!(b.kv_get("k").unwrap(), Some("v1".to_owned())); - // Upsert overwrites. - b.kv_set("k", "v2").unwrap(); - assert_eq!(b.kv_get("k").unwrap(), Some("v2".to_owned())); - // Delete clears; deleting again is a no-op. - b.kv_delete("k").unwrap(); - assert_eq!(b.kv_get("k").unwrap(), None); - b.kv_delete("k").unwrap(); - } - /// Happy path: send → recv → `ack_turn` drains the in-memory list /// and marks the row `acked_at IS NOT NULL`. A second recv finds /// nothing pending (the row stays in the table for vacuum). diff --git a/hive-c0re/src/stores/power.rs b/hive-c0re/src/stores/power.rs index 8fb1ecda..ca9c31d6 100644 --- a/hive-c0re/src/stores/power.rs +++ b/hive-c0re/src/stores/power.rs @@ -1,33 +1,24 @@ -//! Durable per-agent power *intent* (`wanted: Up | Offline`) — the -//! spec half of spec-vs-status desired-state reconciliation. -//! `container_view` remains the observed *status*; the job queue's -//! `Reconcile` nodes are the mechanism that converges the two. +//! Per-agent power *intent* (`wanted: Up | Offline`) this hive holds in +//! memory — what the operator last asked of an agent through `hivectl` or a +//! queued power op, until hive-c0re restarts. `container_view` remains the +//! observed *status*; the job queue's `Reconcile` nodes are the mechanism that +//! converges the two. //! -//! Stored as the `agent_power` table in the coordinator DB -//! (`/var/lib/hyperhive/db/broker.sqlite`, one tiny row per agent) — -//! same one-file-many-modules pattern as `broker`, each with its own -//! connection. Intent persists across hive-c0re restarts; in-flight -//! queue work deliberately does not. Setting `wanted` is never a -//! queued node: operator/intent actions update the row synchronously -//! at request time, then submit the DAG whose terminal `Reconcile` -//! reads the fresh value — rapid toggles are last-writer-wins and the -//! reconciles converge. +//! Nothing here is durable, on purpose: the swarm owns lasting power intent. +//! It declares each agent's state in the hive's wanted bucket (see +//! `crate::workers::wanted`), and that declaration outranks whatever this map +//! holds. After a restart the map is empty, an agent with no entry is seeded +//! from what its container is doing (so nothing starts or stops by itself), +//! and the declaration brings the declared agents to their state. +//! +//! Setting `wanted` is never a queued node's job to *read back*: operator +//! actions update the map synchronously at request time, then submit the DAG +//! whose terminal `Reconcile` reads the fresh value — rapid toggles are +//! last-writer-wins and the reconciles converge. -use std::path::Path; +use std::collections::HashMap; use std::sync::Mutex; -use anyhow::{Context, Result}; -use chrono::Utc; -use rusqlite::{Connection, OptionalExtension, params}; - -const SCHEMA: &str = " -CREATE TABLE IF NOT EXISTS agent_power ( - agent TEXT PRIMARY KEY, - wanted TEXT NOT NULL, - updated_at INTEGER NOT NULL -); -"; - /// Per-agent power intent. #[derive(Debug, Clone, Copy, PartialEq, Eq, strum::IntoStaticStr, strum::EnumString)] #[strum(serialize_all = "snake_case")] @@ -37,8 +28,8 @@ pub enum Wanted { } impl Wanted { - /// Seed value from an observed running state (first boot after - /// this store lands, or an agent spawned outside the normal path). + /// Seed value from an observed running state (an agent no operator or + /// declaration has said anything about since this process started). pub fn from_running(running: bool) -> Self { if running { Wanted::Up } else { Wanted::Offline } } @@ -62,96 +53,56 @@ pub fn reconcile_action(wanted: Wanted, running: bool) -> ReconcileAction { } } -/// Sqlite-backed store. `Arc`-friendly: all methods take `&self`, the -/// internal `Mutex` serializes access. +/// The in-memory intent map. `Arc`-friendly: all methods take `&self`, the +/// internal `Mutex` serializes access. +#[derive(Default)] pub struct PowerStore { - conn: Mutex, + intents: Mutex>, } impl PowerStore { - /// Open (a connection to) the shared coordinator DB and ensure the - /// `agent_power` table exists. `db_path` is the same sqlite file - /// the broker store opens. - pub fn open(db_path: &Path) -> Result { - let conn = crate::db::open(db_path, "agent_power")?; - conn.execute_batch(SCHEMA) - .context("apply agent_power schema")?; - Ok(Self { - conn: Mutex::new(conn), - }) - } - - /// In-memory store for tests. - #[cfg(test)] - pub fn open_in_memory() -> Result { - let conn = Connection::open_in_memory().context("open in-memory agent_power db")?; - conn.execute_batch(SCHEMA) - .context("apply agent_power schema")?; - Ok(Self { - conn: Mutex::new(conn), - }) - } - - /// Read an agent's intent. `None` when the agent has no row yet - /// (callers seed from observed state via [`Self::get_or_seed`]). - /// - /// A row that exists but holds an unparseable value is an **error**, not - /// `None`. The two used to collapse, and the damage was not that a caller - /// read the wrong value: [`Self::get_or_seed`] takes `None` to mean "never - /// seeded" and writes what the container is *currently doing*, so a - /// corrupted row silently overwrote itself with the observed state this - /// store exists to overrule. - pub fn get(&self, agent: &str) -> Result> { - let conn = self.conn.lock().expect("agent_power mutex poisoned"); - let row: Option = conn - .query_row( - "SELECT wanted FROM agent_power WHERE agent = ?1", - params![agent], - |r| r.get(0), - ) - .optional() - .context("select agent_power")?; - match row { - None => Ok(None), - Some(stored) => stored.parse::().ok().map(Some).with_context(|| { - format!("agent_power row for {agent} holds an unknown intent {stored:?}") - }), - } + /// Read an agent's intent. `None` when nothing has set or seeded one since + /// this process started (callers seed from observed state via + /// [`Self::get_or_seed`]). + pub fn get(&self, agent: &str) -> Option { + self.intents + .lock() + .expect("power intents mutex poisoned") + .get(agent) + .copied() } /// Write an agent's intent (last-writer-wins, synchronous at /// request time). - pub fn set(&self, agent: &str, wanted: Wanted) -> Result<()> { - let conn = self.conn.lock().expect("agent_power mutex poisoned"); - conn.execute( - "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3) - ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3", - params![agent, <&str>::from(wanted), Utc::now().timestamp()], - ) - .context("upsert agent_power")?; - Ok(()) + pub fn set(&self, agent: &str, wanted: Wanted) { + self.intents + .lock() + .expect("power intents mutex poisoned") + .insert(agent.to_owned(), wanted); } - /// Read an agent's intent, seeding the row from the observed - /// running state when absent — the migration rule for agents that - /// predate this store (running ⇒ `Up`, stopped ⇒ `Offline`), after - /// which the DB is authoritative. - pub fn get_or_seed(&self, agent: &str, running: bool) -> Result { - if let Some(w) = self.get(agent)? { - return Ok(w); - } - let seeded = Wanted::from_running(running); - self.set(agent, seeded)?; - tracing::info!(%agent, wanted = <&str>::from(seeded), "agent_power: seeded from observed state"); - Ok(seeded) + /// Read an agent's intent, seeding it from the observed running state when + /// absent (running ⇒ `Up`, stopped ⇒ `Offline`), after which the map is + /// authoritative until the process restarts. + pub fn get_or_seed(&self, agent: &str, running: bool) -> Wanted { + *self + .intents + .lock() + .expect("power intents mutex poisoned") + .entry(agent.to_owned()) + .or_insert_with(|| { + let seeded = Wanted::from_running(running); + tracing::info!(%agent, wanted = <&str>::from(seeded), "power intent: seeded from observed state"); + seeded + }) } - /// Drop an agent's row (container destroyed). - pub fn remove(&self, agent: &str) -> Result<()> { - let conn = self.conn.lock().expect("agent_power mutex poisoned"); - conn.execute("DELETE FROM agent_power WHERE agent = ?1", params![agent]) - .context("delete agent_power")?; - Ok(()) + /// Drop an agent's intent (container destroyed). + pub fn remove(&self, agent: &str) { + self.intents + .lock() + .expect("power intents mutex poisoned") + .remove(agent); } } @@ -175,75 +126,18 @@ mod tests { ); } - /// Write a value `set` cannot produce, the way corruption would. - fn write_raw(store: &PowerStore, agent: &str, wanted: &str) { - let conn = store.conn.lock().expect("agent_power mutex poisoned"); - conn.execute( - "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, 0)", - params![agent, wanted], - ) - .expect("insert raw row"); - } - - /// A row that exists but does not parse is an error, not an absent one. - /// The two rows beside it are the control: this must not pass by breaking - /// `get` for everything. - #[test] - fn an_unparseable_row_is_an_error_not_an_absent_one() { - let store = PowerStore::open_in_memory().expect("open"); - store.set("valid", Wanted::Up).expect("set"); - write_raw(&store, "corrupt", "sideways"); - - assert_eq!(store.get("valid").expect("valid reads"), Some(Wanted::Up)); - assert_eq!(store.get("missing").expect("absent reads"), None); - - let err = store - .get("corrupt") - .expect_err("corrupt must not read as absent"); - assert!( - format!("{err:#}").contains("sideways"), - "the error should name the offending value: {err:#}" - ); - } - - /// The damage the error arm exists to prevent: `get_or_seed` treats - /// `None` as "never seeded" and writes observed state, so a corrupt row - /// used to overwrite itself with whatever the container was doing. - #[test] - fn a_corrupt_row_is_not_reseeded_from_observed_state() { - let store = PowerStore::open_in_memory().expect("open"); - write_raw(&store, "corrupt", "sideways"); - - assert!(store.get_or_seed("corrupt", true).is_err()); - - let stored: String = store - .conn - .lock() - .expect("agent_power mutex poisoned") - .query_row( - "SELECT wanted FROM agent_power WHERE agent = ?1", - params!["corrupt"], - |r| r.get(0), - ) - .expect("row still there"); - assert_eq!(stored, "sideways", "the corrupt row must be left alone"); - } - #[test] fn get_set_roundtrip_and_seed() { - let store = PowerStore::open_in_memory().expect("open"); - assert_eq!(store.get("alice").expect("get"), None); + let store = PowerStore::default(); + assert_eq!(store.get("alice"), None); // Seed from observed running state, once. - assert_eq!(store.get_or_seed("alice", true).expect("seed"), Wanted::Up); - // Thereafter the DB is authoritative — observed state no longer + assert_eq!(store.get_or_seed("alice", true), Wanted::Up); + // Thereafter the map is authoritative — observed state no longer // overrides. - assert_eq!( - store.get_or_seed("alice", false).expect("seeded"), - Wanted::Up - ); - store.set("alice", Wanted::Offline).expect("set"); - assert_eq!(store.get("alice").expect("get"), Some(Wanted::Offline)); - store.remove("alice").expect("remove"); - assert_eq!(store.get("alice").expect("get"), None); + assert_eq!(store.get_or_seed("alice", false), Wanted::Up); + store.set("alice", Wanted::Offline); + assert_eq!(store.get("alice"), Some(Wanted::Offline)); + store.remove("alice"); + assert_eq!(store.get("alice"), None); } } diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 6f281291..ca46ae10 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -355,11 +355,11 @@ async fn handle_deploy_request( /// /// Shared with [`crate::workers::wanted`] for the **seeding**, not the insert. /// `first_deploy`'s tail `Reconcile` seeds `Wanted` from the container's -/// currently observed running state when no row exists yet -/// (`power::Store::get_or_seed`), and at that point the container is freshly -/// created but not started — so an unseeded row locks the agent to `Offline` +/// currently observed running state when no intent is set yet +/// (`PowerStore::get_or_seed`), and at that point the container is freshly +/// created but not started — so an unseeded agent is locked to `Offline` /// on its very first reconcile and `Reconcile` never emits the `Start` node. -/// Setting the row up front closes that window. A second caller open-coding +/// Setting the intent 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. /// @@ -369,9 +369,7 @@ pub(crate) fn queue_first_deploy( coord: &std::sync::Arc, agent: &str, ) -> 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.power.set(agent, crate::power::Wanted::Up); coord.job_queue.insert_first_deploy_once(agent) } diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index b002bb57..e15c311a 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -95,9 +95,9 @@ async fn classify( .collect(); logical_names.sort(); - // `get_or_seed` doubles as the one-time migration: an - // agent without an `agent_power` row is seeded from its observed - // state (running ⇒ Up), after which the DB is authoritative. + // `get_or_seed`: nothing remembers an intent across a restart, so an agent + // starts here with the one its observed state implies (running ⇒ Up) and + // the swarm's declaration (`workers::wanted`) moves it from there. // stale ∧ wanted=Up → rebuild. The bool is the agent's observed // running state, used to order running agents first before submit. let mut rebuilds: Vec<(String, bool)> = Vec::new(); @@ -106,13 +106,7 @@ async fn classify( let mut skipped = 0usize; for name in &logical_names { let running = lifecycle::is_running(name).await; - let wanted = match coord.power.get_or_seed(name, running) { - Ok(w) => Some(w), - Err(e) => { - tracing::warn!(%name, error = ?e, "agent_power read failed — reconciling so it surfaces"); - None - } - }; + let wanted = coord.power.get_or_seed(name, running); let pending = crate::attic::closure::pending_path(name).await; let fresh = is_fresh( pending.as_deref(), @@ -139,7 +133,6 @@ async fn classify( drifted.push(name.clone()); } } - BootAction::Unreadable => drifted.push(name.clone()), } } @@ -217,22 +210,9 @@ enum BootAction { Defer { reconcile: bool }, /// Already running the declared closure, or nothing to deploy. Skip { reconcile: bool }, - /// The intent could not be read. - Unreadable, } -/// `wanted` is `None` when the store could not answer for this agent. -/// -/// That case has no neutral stand-in, which is why it is a variant rather -/// than a fallback value: `Wanted::from_running(running)` is precisely the -/// value for which `reconcile_action` returns `Noop`, so guessing it here -/// puts the agent beyond every check below — a corrupt row then leaves one -/// `warn!` per boot and no other trace. The queued reconcile's own -/// `get_or_seed` fails as a per-agent node instead. -fn boot_action(wanted: Option, fresh: bool, running: bool) -> BootAction { - let Some(wanted) = wanted else { - return BootAction::Unreadable; - }; +fn boot_action(wanted: crate::power::Wanted, fresh: bool, running: bool) -> BootAction { let reconcile = crate::power::reconcile_action(wanted, running) != crate::power::ReconcileAction::Noop; if fresh { @@ -345,30 +325,14 @@ mod tests { use super::{BootAction, boot_action, is_fresh}; use crate::power::Wanted; - // The regression the `Unreadable` variant exists for: the arm used to - // substitute `from_running(running)`, which is `Noop` against BOTH - // observations — so no combination of inputs could reach a reconcile. - #[test] - fn an_unreadable_intent_is_never_settled_whatever_the_agent_is_doing() { - for fresh in [true, false] { - for running in [true, false] { - assert_eq!( - boot_action(None, fresh, running), - BootAction::Unreadable, - "fresh={fresh} running={running}" - ); - } - } - } - #[test] fn a_fresh_agent_doing_what_it_should_is_skipped_without_a_reconcile() { assert_eq!( - boot_action(Some(Wanted::Up), true, true), + boot_action(Wanted::Up, true, true), BootAction::Skip { reconcile: false } ); assert_eq!( - boot_action(Some(Wanted::Offline), true, false), + boot_action(Wanted::Offline, true, false), BootAction::Skip { reconcile: false } ); } @@ -376,35 +340,29 @@ mod tests { #[test] fn a_fresh_agent_that_drifted_from_its_intent_is_reconciled() { assert_eq!( - boot_action(Some(Wanted::Up), true, false), + boot_action(Wanted::Up, true, false), BootAction::Skip { reconcile: true } ); assert_eq!( - boot_action(Some(Wanted::Offline), true, true), + boot_action(Wanted::Offline, true, true), BootAction::Skip { reconcile: true } ); } #[test] fn a_stale_agent_wanted_up_is_rebuilt_running_or_not() { - assert_eq!( - boot_action(Some(Wanted::Up), false, true), - BootAction::Rebuild - ); - assert_eq!( - boot_action(Some(Wanted::Up), false, false), - BootAction::Rebuild - ); + assert_eq!(boot_action(Wanted::Up, false, true), BootAction::Rebuild); + assert_eq!(boot_action(Wanted::Up, false, false), BootAction::Rebuild); } #[test] fn a_stale_agent_wanted_offline_defers_and_still_reports_drift() { assert_eq!( - boot_action(Some(Wanted::Offline), false, false), + boot_action(Wanted::Offline, false, false), BootAction::Defer { reconcile: false } ); assert_eq!( - boot_action(Some(Wanted::Offline), false, true), + boot_action(Wanted::Offline, false, true), BootAction::Defer { reconcile: true } ); } @@ -415,13 +373,10 @@ mod tests { #[test] fn a_changed_declaration_for_a_wanted_up_agent_is_rebuilt_unprompted() { let fresh = is_fresh(Some("/nix/store/new"), None, false); - assert_eq!( - boot_action(Some(Wanted::Up), fresh, true), - BootAction::Rebuild - ); + assert_eq!(boot_action(Wanted::Up, fresh, true), BootAction::Rebuild); let fresh = is_fresh(None, None, false); assert_eq!( - boot_action(Some(Wanted::Up), fresh, true), + boot_action(Wanted::Up, fresh, true), BootAction::Skip { reconcile: false }, "control: nothing declared, nothing rebuilt" ); diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 33885908..47517766 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -4,8 +4,7 @@ //! [`swarm_wire::wanted`]); this reads its own and acts on it. It is //! the **repair** path, not the fast one: a deploy event //! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears -//! it, and until this existed that left the hive on its old shape with nothing -//! looking wrong. +//! it and would stay on its old shape with nothing looking wrong. //! //! # Two things a declaration does not mean //! @@ -21,13 +20,14 @@ //! //! # What "diverged" is measured against //! -//! The hive's **stored power intent**, not the container's observed running -//! state. An agent down while its intent says `Up` is already the boot +//! The power intent this process holds, not the container's observed running +//! state. An agent down while its intent says `Up` is already a queued //! reconcile's work; a loop reading `is_running` would insert a start DAG -//! behind that reconcile's back on every boot. +//! behind that reconcile's back. //! -//! 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. +//! A declaration outranks a local stop, since the controller owns lasting +//! power intent. Nothing is remembered across a restart, so the first pass +//! after one queues a (no-op) start or stop for each declared agent. use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; @@ -341,26 +341,17 @@ async fn converge(coord: &Arc, declared: &HiveWanted) -> Result<()> .map(|spec| spec.name) .collect(); - let mut intents = BTreeMap::new(); - for agent in declared.agents.keys() { - match coord.power.get(agent) { - Ok(intent) => { - intents.insert(agent.clone(), intent); - } - // Left out of the map, which is what makes `plan` skip it. - Err(e) => { - tracing::warn!(%agent, error = ?e, "wanted state: cannot read power intent; skipping"); - } - } - } + let intents: BTreeMap> = declared + .agents + .keys() + .map(|agent| (agent.clone(), coord.power.get(agent))) + .collect(); - // The pause marker is a stat, not a store read — it cannot fail the way - // `coord.power.get` above can, so there is no "unreadable, skip" branch - // to mirror. A name that does not parse as an `Ident` is skipped with a - // warning instead, the same defensive shape `crash_watch` uses for the - // same reason: a declaration is attacker-adjacent input (published by - // the swarm controller, not typed by an operator at this hive), so a - // malformed name should not panic the convergence loop. + // A name that does not parse as an `Ident` is skipped with a warning, the + // same defensive shape `crash_watch` uses for the same reason: a + // declaration is attacker-adjacent input (published by the swarm + // controller, not typed by an operator at this hive), so a malformed name + // should not panic the convergence loop. let mut paused = BTreeMap::new(); for agent in declared.agents.keys() { match hive_types::Ident::parse(agent) { @@ -463,10 +454,9 @@ struct Plan { /// Turn one declaration into that plan — pure, so what the loop does with a /// given declaration is testable without a queue, a container or a store. /// -/// `intents` is the hive's stored power intent per agent, and an agent -/// **missing from the map** is one whose intent could not be read: it is -/// skipped. That is not `Some(None)` — "no row yet" is a reading and gets -/// converged, "the read failed" is not and would be a guess. +/// `intents` is the power intent this process holds per agent: `None` is an +/// agent nothing has set an intent for since the process started, which gets +/// converged like any other that disagrees with the declaration. /// /// An agent the declaration does not name reaches none of the three lists. /// That is the whole of "absence is not a deletion order": the agents this @@ -479,10 +469,6 @@ fn plan( ) -> Plan { let mut plan = Plan::default(); for (agent, decl) in &declared.agents { - // The pause axis does not gate on a readable power intent the way - // the match below does — it has no intent row to fail reading, only - // a stat that always answers — so it is evaluated unconditionally, - // even for an agent this pass otherwise skips. let is_present = present.contains(agent); let is_paused = paused.get(agent).copied().unwrap_or(false); match decide_pause(decl.state, is_present, is_paused) { @@ -491,10 +477,8 @@ fn plan( PauseConverge::Nothing => {} } - let Some(intent) = intents.get(agent) else { - continue; - }; - match decide(decl.state, is_present, *intent) { + let intent = intents.get(agent).copied().flatten(); + match decide(decl.state, is_present, intent) { Converge::Deploy => plan.deploy.push(agent.clone()), Converge::Start => plan.start.push(agent.clone()), Converge::Stop => plan.stop.push(agent.clone()), @@ -659,37 +643,6 @@ mod tests { ); } - /// An intent that could not be read is skipped rather than treated as "no - /// row". `readable` is the control for the same call. - #[test] - fn an_unreadable_intent_skips_only_that_agent() { - let declared = declaration(&[("unreadable", AgentState::Up), ("readable", AgentState::Up)]); - let planned = plan( - &declared, - &set(&["unreadable", "readable"]), - &intents(&[("readable", None)]), - &paused(&[]), - ); - assert_eq!(planned.start, vec!["readable".to_owned()]); - assert!(planned.deploy.is_empty() && planned.stop.is_empty()); - } - - /// The pause axis does not gate on a readable intent row at all — a - /// `paused` agent this hive has no power-intent row for still gets its - /// pause marker converged, unlike the power axis right above. - #[test] - fn pause_converges_even_when_the_power_intent_is_unreadable() { - let declared = declaration(&[("unreadable", AgentState::Paused)]); - let planned = plan( - &declared, - &set(&["unreadable"]), - &intents(&[]), - &paused(&[("unreadable", false)]), - ); - assert_eq!(planned.pause, vec!["unreadable".to_owned()]); - assert!(planned.start.is_empty() && planned.deploy.is_empty()); - } - /// A declared-paused agent that is present, running and already carries /// the marker needs nothing on either axis. #[test] @@ -738,7 +691,7 @@ mod tests { #[test] fn a_declared_agent_this_hive_does_not_have_is_deployed() { assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy); - // Even holding an intent: a row can outlive its container. + // Even holding an intent: it can outlive its container. assert_eq!( decide(AgentState::Up, false, Some(Wanted::Up)), Converge::Deploy @@ -769,11 +722,11 @@ mod tests { ); } - /// No row is not agreement. An agent that predates the power store is - /// converged to what it was declared as, not left on whatever it happens - /// to be doing. + /// No intent is not agreement. An agent nothing has set an intent for since + /// this process started is converged to what it was declared as, not left + /// on whatever it happens to be doing. #[test] - fn an_agent_with_no_intent_row_is_converged() { + fn an_agent_with_no_intent_is_converged() { assert_eq!(decide(AgentState::Up, true, None), Converge::Start); assert_eq!(decide(AgentState::Offline, true, None), Converge::Stop); } diff --git a/hivectl/src/cli.rs b/hivectl/src/cli.rs index df44b23c..169e5ee0 100644 --- a/hivectl/src/cli.rs +++ b/hivectl/src/cli.rs @@ -103,9 +103,8 @@ pub enum Cmd { }, /// Start containers hive-wide — the inverse of `hivectl stop`. /// - /// Bare `hivectl start` restores the agents stopped by the last - /// broad-scope `stop` (or starts everything if none); scope flags - /// narrow it. The swarm's forge and matrix are never in scope. + /// Bare `hivectl start` starts every agent; scope flags narrow it. The + /// swarm's forge and matrix are never in scope. Start { #[command(flatten)] scope: ScopeArgs,