From 18745f1a98cd9199c3e1f229819174313625a94d Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 26 Jul 2026 19:28:08 +0200 Subject: [PATCH] job_queue: finish reparent call-site swap, delete dead sync path --- hive-c0re/src/agent_config/topology.rs | 26 +----- hive-c0re/src/coordinator.rs | 78 ++---------------- hive-c0re/src/dashboard/topology.rs | 110 +++++++++++++++---------- hive-c0re/src/job_queue/model.rs | 6 -- hive-c0re/src/job_queue/submit.rs | 21 ++--- hive-c0re/src/job_queue/templates.rs | 6 -- hive-c0re/src/meta.rs | 39 +-------- hive-c0re/src/server.rs | 24 +++--- 8 files changed, 94 insertions(+), 216 deletions(-) diff --git a/hive-c0re/src/agent_config/topology.rs b/hive-c0re/src/agent_config/topology.rs index 1ac11516..05cb3634 100644 --- a/hive-c0re/src/agent_config/topology.rs +++ b/hive-c0re/src/agent_config/topology.rs @@ -298,30 +298,10 @@ pub fn apply_set_parent( Ok(next) } -/// Operator-driven parent move. Set `child`'s parent to `new_parent` -/// (or `None` to promote to root). See [`apply_set_parent`] for the -/// validation rules. The operator-set parent sticks across -/// `reconcile()` calls (which preserves existing entries). -/// -/// No bind-mount / container churn today — the hierarchy is -/// currently logical-only. Once sub-manager bind mounts land, the -/// caller adds an umount-old / mount-new / restart-cascade step on -/// top. -pub fn set_parent(child: &str, new_parent: Option<&str>) -> Result<(), String> { - let current = read(); - // Idempotent no-op fast path: skip the disk write when nothing - // changes. apply_set_parent still runs to surface validation - // errors (e.g. unknown child) so the caller gets a real signal. - let next = apply_set_parent(¤t, child, new_parent)?; - if next == current { - return Ok(()); - } - write(&next).map_err(|e| format!("write topology.json: {e}")) -} - /// Declare a brand-new agent's parent edge before the agent exists in -/// the container set. Unlike [`set_parent`] (which reparents an entry -/// that must already be present), this inserts a fresh `child -> parent` +/// the container set. Unlike [`crate::meta::bulk_commit_topology`] / +/// [`apply_set_parent`] (which reparent an entry that must already be +/// present), this inserts a fresh `child -> parent` /// row. Used by the `InitConfig` approval to place a just-scaffolded /// sub-agent under its requesting parent, so the edge is in place /// before the first apply-commit spawns the container (and before diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index d44fe4cc..cfe9aa34 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -926,9 +926,11 @@ impl Coordinator { } } - /// Apply a topology reparent + fan the resulting notifications - /// out to the three affected agents. On success, drops a - /// one-line system message into the inbox of: + /// Apply topology reparent(s) + fan the resulting notifications out to + /// the affected agents. Applies all moves under a single `META_LOCK` + /// acquisition (one git commit — a single-move call is just a + /// one-element slice) then, for each move that actually changed + /// topology, drops a one-line system message into the inbox of: /// /// 1. The **old parent** (if any) — `"{child} moved out of your /// subtree to {new_parent_or_root}"`. @@ -946,76 +948,6 @@ impl Coordinator { /// `from = hive_sh4re::SYSTEM_SENDER` so the dashboard renders /// them under the existing system-source styling. /// - /// Idempotent: if `topology::set_parent` skipped the disk write - /// (same-parent no-op), no messages fire. The `topology` module's - /// validation (`apply_set_parent`: unknown agent, cycle, etc.) - /// runs *before* any message is sent, so a refused move never - /// notifies anyone. - /// - /// Also drives a `rescan_containers_and_emit` after the write so - /// the dashboard tree re-renders without polling. - pub async fn reparent_with_notify( - self: &Arc, - child: &str, - new_parent: Option<&str>, - ) -> std::result::Result<(), String> { - // Snapshot the old parent BEFORE the write so the - // notifications can describe both sides of the change. - let topo_before = crate::topology::read(); - let old_parent = topo_before.get(child).cloned().flatten(); - - // The disk write + git commit happen here under META_LOCK, so - // the topology.json change is committed atomically and the - // working tree is never left dirty between the write and the - // next meta operation. Topology validation + idempotent - // fast-path inside `set_parent` may short-circuit (same - // parent → no-op). We mirror the same idempotent shape for - // the notifications: if nothing changed, send nothing. - crate::meta::commit_topology(child, new_parent).await?; - - let changed = old_parent.as_deref() != new_parent; - if changed { - let old_label = old_parent.as_deref().unwrap_or(""); - let new_label = new_parent.unwrap_or(""); - // System-source notifications. Best-effort: a failed broker - // send is logged + ignored so a transient sqlite error - // doesn't bubble out and unwind the topology write. - if let Some(op) = old_parent.as_deref() { - let _ = self.broker.send(&hive_sh4re::Message { - from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER), - to: op.to_owned(), - body: format!("{child} moved out of your subtree to {new_label}"), - in_reply_to: None, - }); - } - if let Some(np) = new_parent { - let _ = self.broker.send(&hive_sh4re::Message { - from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER), - to: np.to_owned(), - body: format!( - "{child} just moved into your subtree (was previously under {old_label})" - ), - in_reply_to: None, - }); - } - // Coalesce: if a prior move notification is already pending in - // the broker (agent was offline for multiple moves), update it - // in-place so the agent sees "A to C" not "A to B" then "B to C". - let _ = self - .broker - .send_coalescing_reparent(child, old_label, new_label); - } - - // Rescan + diff-emit regardless of whether messages fired — - // even an idempotent no-op might have happened concurrently - // with another lifecycle event the dashboard cares about. - self.rescan_containers_and_emit().await; - Ok(()) - } - - /// Batch version of [`reparent_with_notify`]: applies all moves under a - /// single `META_LOCK` acquisition (one git commit) then sends per-agent - /// notifications for each move that actually changed topology. /// First validation failure aborts the whole batch with no disk writes. /// /// # Errors diff --git a/hive-c0re/src/dashboard/topology.rs b/hive-c0re/src/dashboard/topology.rs index 2519b6af..8d180675 100644 --- a/hive-c0re/src/dashboard/topology.rs +++ b/hive-c0re/src/dashboard/topology.rs @@ -2,8 +2,12 @@ //! //! Operator-driven agent reparenting — single (`/api/topology/set-parent`, //! form-encoded) and bulk (`/api/topology/set-parent-bulk`, JSON array → -//! one git commit). Both go through `Coordinator::reparent*_with_notify`, -//! which wraps `topology::set_parent` with the move-notification messages +//! one git commit). Both submit a `NodeKind::Reparent` DAG to the job +//! queue (fire-and-forget, like every other queue-backed op — the +//! dashboard tree repaints off the queue's own snapshot/rescan once the +//! commit lands, same as a rebuild or restart). The executor delegates to +//! `Coordinator::reparent_bulk_with_notify`, which wraps +//! `crate::meta::bulk_commit_topology` with the move-notification messages //! and the `ContainerView` rescan. use axum::{ @@ -15,7 +19,8 @@ use serde::Deserialize; use problem_details::ProblemDetails; -use super::{AppState, error_problem, error_response}; +use super::{AppState, error_problem}; +use crate::job_queue::{Source, submit}; /// `POST /api/topology/set-parent` body. `child` is required. /// `new_parent` may be: @@ -44,11 +49,13 @@ pub(super) struct SetParentBulkEntry { /// `POST /api/topology/set-parent` — operator-driven parent move. /// Form fields: `child` (required, agent name), `new_parent` /// (optional — empty / absent string ⇒ promote to root). Refuses -/// cycles and unknown agents. The manager is reparentable like any -/// other agent — its privileges come from the privileged MCP socket, -/// not its tree position. On success -/// re-emits container snapshots so the dashboard tree repaints -/// without a refresh. +/// cycles and unknown agents (surfaced async on the job view — this +/// handler only validates the identifiers, not the move itself). The +/// manager is reparentable like any other agent — its privileges come +/// from the privileged MCP socket, not its tree position. Submitting +/// re-emits the queue snapshot immediately so the dashboard shows the +/// queued move without a refresh; the tree itself repaints once the +/// commit lands. pub(super) async fn post_set_parent( State(state): State, Form(form): Form, @@ -58,6 +65,8 @@ pub(super) async fn post_set_parent( return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) .with_detail("set-parent: `child` required")); } + let child = hive_types::Ident::parse(&child) + .map_err(|e| error_problem(&format!("set-parent: `child` {e}")))?; // Empty / whitespace-only `new_parent` ⇒ promote to root. Web // forms submit the empty string for a "no value" radio button, // so this is the ergonomic encoding. @@ -66,54 +75,65 @@ pub(super) async fn post_set_parent( .as_deref() .map(str::trim) .filter(|s| !s.is_empty()) - .map(str::to_owned); - // `reparent_with_notify` wraps `topology::set_parent` with the - // three notification messages + the ContainerView rescan. - // Idempotent same-parent calls skip both the messages and the - // disk write per the topology fast-path. - match state - .coord - .reparent_with_notify(&child, new_parent.as_deref()) - .await - { - Ok(()) => { - tracing::info!( - child = %child, - new_parent = ?new_parent, - "operator: set-parent via dashboard" - ); - Ok((StatusCode::OK, "ok").into_response()) - } - Err(e) => Err(error_problem(&format!("set-parent {child} failed: {e}"))), - } + .map(hive_types::Ident::parse) + .transpose() + .map_err(|e| error_problem(&format!("set-parent: `new_parent` {e}")))?; + tracing::info!( + child = %child, + new_parent = ?new_parent, + "operator: set-parent via dashboard" + ); + submit::reparent( + &state.coord, + vec![(child, new_parent)], + Source::Manual, + "manual set-parent via dashboard".to_owned(), + ); + Ok((StatusCode::OK, "ok").into_response()) } /// `POST /api/topology/set-parent-bulk` — move multiple agents in a single /// request, producing **one** git commit. JSON body: `[{"child":"name", /// "new_parent":"target-or-null"}, ...]`. Empty array is a no-op (200 OK). -/// First validation error aborts the whole batch. +/// First identifier that fails to parse aborts the whole batch before +/// anything is submitted — a partially-invalid bulk move never reaches +/// the queue. pub(super) async fn post_set_parent_bulk( State(state): State, axum::Json(body): axum::Json>, -) -> Response { +) -> Result { if body.is_empty() { - return (StatusCode::OK, "ok").into_response(); + return Ok((StatusCode::OK, "ok").into_response()); } - // Collect borrows for the coordinator call. - let moves: Vec<(&str, Option<&str>)> = body + // Collect into `Result<_, String>` first, not `ProblemDetails` directly — + // clippy::result_large_err flags a ~232-byte `Err` variant threaded + // through this closure's `?`. `String` is small enough to satisfy the + // lint; the single `map_err` below promotes it to a `ProblemDetails` + // once, after the fallible collect. + let moves = body .iter() .map(|e| { - let child: &str = &e.child; - let parent: Option<&str> = e.new_parent.as_deref().filter(|s| !s.is_empty()); - (child, parent) + let child = hive_types::Ident::parse(e.child.trim()) + .map_err(|err| format!("set-parent-bulk: `{}` {err}", e.child))?; + let new_parent = e + .new_parent + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(hive_types::Ident::parse) + .transpose() + .map_err(|err| format!("set-parent-bulk: `{}` {err}", e.child))?; + Ok((child, new_parent)) }) - .collect(); - match state.coord.reparent_bulk_with_notify(&moves).await { - Ok(()) => { - let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); - tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); - (StatusCode::OK, "ok").into_response() - } - Err(e) => error_response(&format!("set-parent-bulk failed: {e}")), - } + .collect::, String>>() + .map_err(|e| error_problem(&e))?; + let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); + tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); + submit::reparent( + &state.coord, + moves, + Source::Manual, + "manual set-parent-bulk via dashboard".to_owned(), + ); + Ok((StatusCode::OK, "ok").into_response()) } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index a679dfe1..5d353d31 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -196,12 +196,6 @@ pub enum NodeKind { /// window. `(child, new_parent)` pairs, applied in order under one /// `META_LOCK` acquisition / one git commit (`meta::bulk_commit_topology` /// handles both the single- and multi-move case uniformly). - #[allow( - dead_code, - reason = "constructed by templates::reparent(), landed ahead of the call-site swap \ - (pending an answer on whether that swap should be synchronous or \ - fire-and-forget) — exercised today by job_queue::tests only" - )] Reparent { moves: Vec<(hive_types::Ident, Option)>, }, diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 44371572..c9648162 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -422,21 +422,12 @@ pub fn meta_update( } /// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs — -/// one entry for `set-parent`, N for `set-parent-bulk`. Not yet wired to the -/// `set-parent`/`set-parent-bulk` HTTP handlers or the `hivectl`/MCP -/// `SetParent` surface — those still call `Coordinator::reparent*_with_notify` -/// directly, which blocks until the commit lands and returns a synchronous -/// `ok`/`err`, unlike every other queue-backed op. Whether that call-site -/// swap should keep the synchronous contract or go fire-and-forget (submit -/// returns a DAG id immediately, like everything else in this module) is an -/// open question — exists so the `NodeKind::Reparent` shape is exercised -/// end-to-end (tests, and any future caller) ahead of that call-site swap. -#[allow( - dead_code, - reason = "landed ahead of the server.rs/dashboard::topology call-site swap, pending an \ - answer on whether that swap should be synchronous or fire-and-forget — \ - exercised today by job_queue::tests" -)] +/// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like +/// everything else in this module: submits and returns a DAG id +/// immediately, the caller learns the outcome async (dashboard job view / +/// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest:: +/// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/ +/// `set-parent-bulk` handlers. pub fn reparent( coord: &Arc, moves: Vec<(hive_types::Ident, Option)>, diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 30cf05d1..5aec3c0a 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -361,12 +361,6 @@ pub fn meta_update( /// dedicated variant would) and the per-template history-retention bucket — /// both cosmetic. Swap this to whatever the eventual node-kind-derived /// dispatch lands with, whenever it lands. -#[allow( - dead_code, - reason = "landed ahead of the server.rs/dashboard::topology call-site swap, pending an \ - answer on whether that swap should be synchronous or fire-and-forget — \ - exercised today by job_queue::tests and submit::reparent" -)] pub fn reparent( moves: Vec<(hive_types::Ident, Option)>, source: Source, diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index c17a90bd..f9896c25 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -562,41 +562,10 @@ pub async fn commit_perms( Ok(()) } -/// Write the topology file and commit it atomically under `META_LOCK`. -/// Returns `Err(String)` on validation failure (unknown agent, cycle, -/// etc.) — same shape as `topology::set_parent` — so callers can -/// surface the error as a user-visible message. Git failures are -/// logged as warnings and don't propagate: the topology write already -/// succeeded, and `sync_agents` will pick up any un-committed change -/// on the next run as a safety net. -pub async fn commit_topology( - child: &str, - new_parent: Option<&str>, -) -> std::result::Result<(), String> { - let _guard = META_LOCK.lock().await; - crate::topology::set_parent(child, new_parent)?; - let dir = crate::paths::meta_root(); - let stage = async { - git(&dir, &["add", "topology.json"]).await?; - if paths_dirty(&dir, &["topology.json"]).await? { - git_commit_paths( - &dir, - &format!("topology: {} → {}", child, new_parent.unwrap_or("")), - &["topology.json"], - ) - .await?; - } - Ok::<_, anyhow::Error>(()) - }; - if let Err(e) = stage.await { - tracing::warn!(%child, ?new_parent, error = ?e, "commit_topology: topology written but git commit failed (sync_agents will recover)"); - } - Ok(()) -} - -/// Batch variant of [`commit_topology`]: applies every `(child, new_parent)` -/// move under a single `META_LOCK` acquisition and creates **one** git commit -/// for all of them. Moves are applied in the order given; the first +/// Applies every `(child, new_parent)` move under a single `META_LOCK` +/// acquisition and creates **one** git commit for all of them — a +/// single-move call is just a one-element slice, so there's no separate +/// non-batch entry point. Moves are applied in the order given; the first /// validation error short-circuits the whole batch. True atomic write: all /// moves are pre-validated against a cumulative in-memory state with /// [`crate::topology::apply_set_parent`] before anything touches disk, then diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 18cddca5..e560bb48 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -180,19 +180,17 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { } HostRequest::SetParent { child, new_parent } => { tracing::info!(%child, ?new_parent, "set_parent"); - // `reparent_with_notify` wraps `topology::set_parent` - // with the three notification messages + the - // ContainerView rescan. Idempotent same-parent calls - // skip both the messages and the disk write per the - // topology fast-path. - coord - .reparent_with_notify( - child.as_str(), - new_parent.as_ref().map(hive_types::Ident::as_str), - ) - .await - .map_err(anyhow::Error::msg)?; - HostResponse::success() + // Fire-and-forget, like every other queue-backed op: + // submit returns a DAG id immediately, the caller polls + // `QueueDag` (`hivectl`'s wait/progress loop) for the + // outcome instead of blocking here on the commit. + let id = crate::job_queue::submit::reparent( + &coord, + vec![(child.clone(), new_parent.clone())], + crate::job_queue::Source::Manual, + "manual set-parent via hivectl".to_owned(), + ); + HostResponse::queued(vec![id]) } HostRequest::SetResourceLimits { name,