From 5bb5a88aa07db2ded6cd70d72d5bb66ac1dea1d8 Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 16 Jul 2026 20:05:27 +0200 Subject: [PATCH] restart preserves wanted intent instead of forcing all agents up (#2540) --- docs/coordinator.md | 31 ++++++++------- hive-c0re/src/job_queue/submit.rs | 58 ++++++++++++++++------------ hive-c0re/src/job_queue/templates.rs | 6 ++- hive-c0re/src/job_queue/tests.rs | 52 +++++++++++-------------- 4 files changed, 78 insertions(+), 69 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index 3a578cd5..02160057 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -75,21 +75,24 @@ container build: ### Every operation as a DAG -The power ops write the durable `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. The hive-wide power ops — -`restart`, `stop`, and `start` — take an agent *list*: a hive-wide `hivectl -restart` / `stop` / `start` is ONE DAG with a per-agent subgraph each -(independent roots, run concurrently on their own leases), not N separate -DAGs. +The `stop` / `start` power ops write the durable `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 +lets the tail `Reconcile` converge to the agent's existing `wanted`, so a +deliberately-stopped agent is not forced back up by a hive-wide restart. The +hive-wide power ops — `restart`, `stop`, and `start` — take an agent *list*: +a hive-wide `hivectl restart` / `stop` / `start` is ONE DAG with a per-agent +subgraph each (independent roots, run concurrently on their own leases), not +N separate DAGs. **These are built dynamically from each agent's live running state** (an async `lifecycle::is_running` read), so they live in `job_queue/submit.rs`, -not the pure/sync `templates.rs`. Per-agent shape rule: the head `SetWanted` -(intent) and the tail `Reconcile` (convergence guarantee — cheap, noops when -already converged) are ALWAYS present; only the *mechanical* nodes -(`Signal`/`Drain`/`StopForUpdate`) are state-conditional — skipped for a -*down* agent (nothing to quiesce/stop). Keeping `Reconcile` in every shape +not the pure/sync `templates.rs`. Per-agent shape rule: `stop`/`start` carry +a head `SetWanted` (intent) — `restart` does not; the tail `Reconcile` +(convergence guarantee — cheap, noops when already converged) is ALWAYS +present; only the *mechanical* nodes (`Signal`/`Drain`/`StopForUpdate`) are +state-conditional — skipped for a *down* agent (nothing to quiesce/stop). Keeping `Reconcile` in every shape closes the TOCTOU window: if an agent flips state between the `is_running` read and node exec, the tail `Reconcile` still converges it in-DAG (with `StopForUpdate`-noop as the backstop) — no reliance on an external reconcile @@ -100,8 +103,8 @@ agent's subgraph is a rebuild-then-start). rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(after-ok) PostSwap(a) →(after-any) Reconcile(a) stop(a..): online a: SetWanted(a,Off) → [Signal→Drain→ if graceful] Reconcile(a) offline a: SetWanted(a,Off) → Reconcile(a) (N subgraphs, 1 DAG) -restart(a..): online a: SetWanted(a,Up) → [Signal→Drain→ if graceful] StopForUpdate(a) → Reconcile(a) - offline a: SetWanted(a,Up) → Reconcile(a) (nothing to stop — it's a start) +restart(a..): online a: [Signal→Drain→ if graceful] StopForUpdate(a) → Reconcile(a) (no SetWanted) + offline a: Reconcile(a) (nothing to stop; Reconcile converges to existing wanted) start(a..): a: SetWanted(a,Up) → Reconcile(a) (down+stale ⇒ SetWanted(a,Up) → «rebuild subgraph») spawn(a): [wanted=Up at approve] Create(a) → WriteDropin(a) → Reconcile(a) perm-change(a): WritePermFile(a) → «rebuild subgraph» diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 5e5ea391..805e4d78 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -9,9 +9,11 @@ //! (`templates::{node, after_ok, rebuild_nodes}`), and concatenate them into //! ONE DAG (independent per-agent roots, concurrent on their own leases). //! -//! Dynamic shape rule: the head `SetWanted(w)` (durable intent) and the tail -//! `Reconcile` (the convergence guarantee — cheap, noops when already -//! converged) are ALWAYS present; only the *mechanical* nodes +//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable +//! intent write) — `restart` does NOT (it bounces the container but leaves +//! `wanted` untouched, so a deliberately-stopped agent isn't forced up). The +//! tail `Reconcile` (the convergence guarantee — cheap, noops when already +//! converged) is ALWAYS present; only the *mechanical* nodes //! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a //! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape //! closes the TOCTOU window: if an agent flips state between the `is_running` @@ -85,26 +87,32 @@ fn start_chain(agent: &str, running: bool, stale: bool) -> Vec { n } -/// One agent's **restart** subgraph. `SetWanted(Up)` head + `Reconcile` -/// tail; the stop portion (`Signal → Drain` when graceful, then -/// `StopForUpdate`) only when the agent is running — a restart of a down -/// agent is really a start (`SetWanted(Up) → Reconcile`). +/// One agent's **restart** subgraph. Restart NEVER rewrites `wanted` +/// intent (no `SetWanted` head, unlike stop/start): it bounces the +/// container and lets the tail `Reconcile` converge to the agent's +/// EXISTING intent, so a deliberately-stopped (`wanted = Off`) agent is +/// not forced back up by a hive-wide restart. A running agent gets the +/// mechanical stop (`Signal → Drain` when graceful, then `StopForUpdate`) +/// before `Reconcile`; a down agent gets just `Reconcile`, which +/// converges to intent — a stopped (`wanted = Off`) agent stays stopped, +/// a crashed (`wanted = Up`) agent comes back up. fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec { - let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())]; - if running { - let mut prev = 0u32; - if graceful { - n.push(node(agent, NodeKind::Signal, after_ok(prev))); - prev += 1; - n.push(node(agent, NodeKind::Drain, after_ok(prev))); - prev += 1; - } - n.push(node(agent, NodeKind::StopForUpdate, after_ok(prev))); - prev += 1; - n.push(node(agent, NodeKind::Reconcile, after_ok(prev))); - } else { - n.push(node(agent, NodeKind::Reconcile, after_ok(0))); + if !running { + // Nothing to bounce — a lone Reconcile converges to intent. + return vec![node(agent, NodeKind::Reconcile, Vec::new())]; } + // Running: mechanical stop then Reconcile. The first stop node is the + // subgraph root (no SetWanted head) and acquires the agent lease. + let mut n = Vec::new(); + if graceful { + n.push(node(agent, NodeKind::Signal, Vec::new())); + n.push(node(agent, NodeKind::Drain, after_ok(0))); + n.push(node(agent, NodeKind::StopForUpdate, after_ok(1))); + } else { + n.push(node(agent, NodeKind::StopForUpdate, Vec::new())); + } + let stop_idx = u32::try_from(n.len() - 1).unwrap_or(0); + n.push(node(agent, NodeKind::Reconcile, after_ok(stop_idx))); n } @@ -258,9 +266,11 @@ pub async fn graceful_restart( /// Restart `agents` (one or many) in a **single** DAG — one per-agent /// subgraph each, built dynamically from live running state and run /// concurrently on their own leases. A running agent gets the stop→reconcile -/// chain (`graceful` prepends signal→drain); a down agent gets just -/// `SetWanted(Up) → Reconcile` (there's nothing to stop). The whole -/// hive-wide `hivectl restart` / `restart-all` is one DAG. +/// chain (`graceful` prepends signal→drain); a down agent gets just a lone +/// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the +/// tail `Reconcile` converges each agent to its EXISTING intent — a +/// deliberately-stopped agent stays down. The whole hive-wide +/// `hivectl restart` / `restart-all` is one DAG. pub async fn restart_many( coord: &Arc, agents: &[String], diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index df16a2ce..898f7b96 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -11,9 +11,11 @@ //! assembled dynamically in `submit.rs` out of the shared pure primitives //! this module exports (`node`, `after_ok`, `rebuild_nodes`) — one //! independent per-agent subgraph each, concurrent on its own lease, ONE -//! DAG for the whole hive-wide op. Power ops write the durable `wanted` +//! DAG for the whole hive-wide op. `stop`/`start` write the durable `wanted` //! intent via a head `SetWanted(w)` node (holding the agent lease, so -//! intent+reconcile is atomic per-agent). +//! intent+reconcile is atomic per-agent); `restart` writes no intent — it +//! bounces the container and lets the tail `Reconcile` converge to the +//! agent's existing `wanted`. //! //! ```text //! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(ok) PostSwap(a) →(any) Reconcile(a) diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index c903b921..eae41d91 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -18,9 +18,9 @@ fn rebuild(agent: &str, reason: &str) -> DagSpec { } /// Restart DAG spec with every agent treated as **running** — the online -/// shape (`SetWanted → [Signal→Drain→] StopForUpdate → Reconcile`) most -/// queue-mechanics tests assume. Mirrors the pre-dynamic `templates::restart` -/// (which is now the state-aware `submit::restart_spec`). +/// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) +/// most queue-mechanics tests assume. Mirrors the pre-dynamic +/// `templates::restart` (which is now the state-aware `submit::restart_spec`). fn restart_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned()) @@ -229,27 +229,23 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { None, ), ); - // Restart's head SetWanted takes the lease; stop's Reconcile must - // wait even though slots are free. + // Restart's first node (StopForUpdate) takes the lease; stop's + // Reconcile must wait even though slots are free. let first = claim_one(&q); assert_eq!(first.dag_id, restart); - assert_eq!(first.kind.as_str(), "set_wanted"); + assert_eq!(first.kind.as_str(), "stop_for_update"); assert!(first.lease_acquired); q.complete_node(restart, first.node_id, Ok(())); - // Same DAG keeps the lease through StopForUpdate then Reconcile. + // Same DAG keeps the lease through the tail Reconcile. let second = claim_one(&q); assert_eq!(second.dag_id, restart); - assert_eq!(second.kind.as_str(), "stop_for_update"); + assert_eq!(second.kind.as_str(), "reconcile"); assert!(!second.lease_acquired, "lease already held by this DAG"); q.complete_node(restart, second.node_id, Ok(())); - let third = claim_one(&q); - assert_eq!(third.dag_id, restart); - assert_eq!(third.kind.as_str(), "reconcile"); - q.complete_node(restart, third.node_id, Ok(())); // Restart terminal → lease released → stop's Reconcile runs. - let fourth = claim_one(&q); - assert_eq!(fourth.dag_id, stop); - q.complete_node(stop, fourth.node_id, Ok(())); + let third = claim_one(&q); + assert_eq!(third.dag_id, stop); + q.complete_node(stop, third.node_id, Ok(())); assert_eq!(state_of(&q, restart), State::Done); assert_eq!(state_of(&q, stop), State::Done); } @@ -314,9 +310,9 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { ); // A hive-wide restart is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); - // Each agent's subgraph head (SetWanted) is a root, so both are - // claimable at once — each takes its OWN agent's lease (no contention - // across distinct agents), all inside the single DAG. + // Each agent's subgraph head (StopForUpdate, since both are running) is + // a root, so both are claimable at once — each takes its OWN agent's + // lease (no contention across distinct agents), all inside the single DAG. let claims = q.claim_ready(); assert!(claims.iter().all(|c| c.dag_id == id)); let mut heads: Vec<(&str, &str, bool)> = claims @@ -327,8 +323,8 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { assert_eq!( heads, vec![ - ("agent-a", "set_wanted", true), - ("agent-b", "set_wanted", true), + ("agent-a", "stop_for_update", true), + ("agent-b", "stop_for_update", true), ], "both per-agent subgraphs start concurrently, each acquiring its own lease" ); @@ -470,8 +466,9 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { "stop down".to_owned(), ), ); - // Offline restart → SetWanted(Up) → Reconcile (no StopForUpdate): a - // restart of a down agent is really a start. + // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): + // nothing to bounce, and restart never rewrites intent, so the tail + // Reconcile converges the down agent to its existing `wanted`. let restart = submit( &q, submit::restart_spec( @@ -498,8 +495,8 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { ); assert_eq!( shape(restart), - vec!["set_wanted".to_owned(), "reconcile".to_owned()], - "offline restart skips StopForUpdate, keeps Reconcile (it's a start)" + vec!["reconcile".to_owned()], + "offline restart is a lone Reconcile (no SetWanted head, nothing to stop)" ); } @@ -749,11 +746,8 @@ fn cancel_refuses_running_dag() { fn terminal_dag_reported_exactly_once_and_lease_released() { let q = JobQueue::new(1); let id = submit(&q, restart_online(&["agent-a"], false, "r")); - // restart = SetWanted → StopForUpdate → Reconcile; not terminal until - // the last node completes. - let set_wanted = claim_one(&q); - q.complete_node(id, set_wanted.node_id, Ok(())); - assert!(q.drain_terminal().is_empty(), "dag not terminal yet"); + // restart = StopForUpdate → Reconcile; not terminal until the last + // node completes. let stop = claim_one(&q); q.complete_node(id, stop.node_id, Ok(())); assert!(q.drain_terminal().is_empty(), "dag not terminal yet");