job_queue: finish reparent call-site swap, delete dead sync path

This commit is contained in:
damocles 2026-07-26 19:28:08 +02:00 committed by mara
commit 18745f1a98
8 changed files with 94 additions and 216 deletions

View file

@ -298,30 +298,10 @@ pub fn apply_set_parent(
Ok(next) 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(&current, 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 /// Declare a brand-new agent's parent edge before the agent exists in
/// the container set. Unlike [`set_parent`] (which reparents an entry /// the container set. Unlike [`crate::meta::bulk_commit_topology`] /
/// that must already be present), this inserts a fresh `child -> parent` /// [`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 /// row. Used by the `InitConfig` approval to place a just-scaffolded
/// sub-agent under its requesting parent, so the edge is in place /// sub-agent under its requesting parent, so the edge is in place
/// before the first apply-commit spawns the container (and before /// before the first apply-commit spawns the container (and before

View file

@ -926,9 +926,11 @@ impl Coordinator {
} }
} }
/// Apply a topology reparent + fan the resulting notifications /// Apply topology reparent(s) + fan the resulting notifications out to
/// out to the three affected agents. On success, drops a /// the affected agents. Applies all moves under a single `META_LOCK`
/// one-line system message into the inbox of: /// 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 /// 1. The **old parent** (if any) — `"{child} moved out of your
/// subtree to {new_parent_or_root}"`. /// subtree to {new_parent_or_root}"`.
@ -946,76 +948,6 @@ impl Coordinator {
/// `from = hive_sh4re::SYSTEM_SENDER` so the dashboard renders /// `from = hive_sh4re::SYSTEM_SENDER` so the dashboard renders
/// them under the existing system-source styling. /// 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<Self>,
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("<root>");
let new_label = new_parent.unwrap_or("<root>");
// 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. /// First validation failure aborts the whole batch with no disk writes.
/// ///
/// # Errors /// # Errors

View file

@ -2,8 +2,12 @@
//! //!
//! Operator-driven agent reparenting — single (`/api/topology/set-parent`, //! Operator-driven agent reparenting — single (`/api/topology/set-parent`,
//! form-encoded) and bulk (`/api/topology/set-parent-bulk`, JSON array → //! form-encoded) and bulk (`/api/topology/set-parent-bulk`, JSON array →
//! one git commit). Both go through `Coordinator::reparent*_with_notify`, //! one git commit). Both submit a `NodeKind::Reparent` DAG to the job
//! which wraps `topology::set_parent` with the move-notification messages //! 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. //! and the `ContainerView` rescan.
use axum::{ use axum::{
@ -15,7 +19,8 @@ use serde::Deserialize;
use problem_details::ProblemDetails; 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. /// `POST /api/topology/set-parent` body. `child` is required.
/// `new_parent` may be: /// `new_parent` may be:
@ -44,11 +49,13 @@ pub(super) struct SetParentBulkEntry {
/// `POST /api/topology/set-parent` — operator-driven parent move. /// `POST /api/topology/set-parent` — operator-driven parent move.
/// Form fields: `child` (required, agent name), `new_parent` /// Form fields: `child` (required, agent name), `new_parent`
/// (optional — empty / absent string ⇒ promote to root). Refuses /// (optional — empty / absent string ⇒ promote to root). Refuses
/// cycles and unknown agents. The manager is reparentable like any /// cycles and unknown agents (surfaced async on the job view — this
/// other agent — its privileges come from the privileged MCP socket, /// handler only validates the identifiers, not the move itself). The
/// not its tree position. On success /// manager is reparentable like any other agent — its privileges come
/// re-emits container snapshots so the dashboard tree repaints /// from the privileged MCP socket, not its tree position. Submitting
/// without a refresh. /// 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( pub(super) async fn post_set_parent(
State(state): State<AppState>, State(state): State<AppState>,
Form(form): Form<SetParentForm>, Form(form): Form<SetParentForm>,
@ -58,6 +65,8 @@ pub(super) async fn post_set_parent(
return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST)
.with_detail("set-parent: `child` required")); .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 // Empty / whitespace-only `new_parent` ⇒ promote to root. Web
// forms submit the empty string for a "no value" radio button, // forms submit the empty string for a "no value" radio button,
// so this is the ergonomic encoding. // so this is the ergonomic encoding.
@ -66,54 +75,65 @@ pub(super) async fn post_set_parent(
.as_deref() .as_deref()
.map(str::trim) .map(str::trim)
.filter(|s| !s.is_empty()) .filter(|s| !s.is_empty())
.map(str::to_owned); .map(hive_types::Ident::parse)
// `reparent_with_notify` wraps `topology::set_parent` with the .transpose()
// three notification messages + the ContainerView rescan. .map_err(|e| error_problem(&format!("set-parent: `new_parent` {e}")))?;
// Idempotent same-parent calls skip both the messages and the tracing::info!(
// disk write per the topology fast-path. child = %child,
match state new_parent = ?new_parent,
.coord "operator: set-parent via dashboard"
.reparent_with_notify(&child, new_parent.as_deref()) );
.await submit::reparent(
{ &state.coord,
Ok(()) => { vec![(child, new_parent)],
tracing::info!( Source::Manual,
child = %child, "manual set-parent via dashboard".to_owned(),
new_parent = ?new_parent, );
"operator: set-parent via dashboard" Ok((StatusCode::OK, "ok").into_response())
);
Ok((StatusCode::OK, "ok").into_response())
}
Err(e) => Err(error_problem(&format!("set-parent {child} failed: {e}"))),
}
} }
/// `POST /api/topology/set-parent-bulk` — move multiple agents in a single /// `POST /api/topology/set-parent-bulk` — move multiple agents in a single
/// request, producing **one** git commit. JSON body: `[{"child":"name", /// request, producing **one** git commit. JSON body: `[{"child":"name",
/// "new_parent":"target-or-null"}, ...]`. Empty array is a no-op (200 OK). /// "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( pub(super) async fn post_set_parent_bulk(
State(state): State<AppState>, State(state): State<AppState>,
axum::Json(body): axum::Json<Vec<SetParentBulkEntry>>, axum::Json(body): axum::Json<Vec<SetParentBulkEntry>>,
) -> Response { ) -> Result<Response, ProblemDetails> {
if body.is_empty() { if body.is_empty() {
return (StatusCode::OK, "ok").into_response(); return Ok((StatusCode::OK, "ok").into_response());
} }
// Collect borrows for the coordinator call. // Collect into `Result<_, String>` first, not `ProblemDetails` directly —
let moves: Vec<(&str, Option<&str>)> = body // 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() .iter()
.map(|e| { .map(|e| {
let child: &str = &e.child; let child = hive_types::Ident::parse(e.child.trim())
let parent: Option<&str> = e.new_parent.as_deref().filter(|s| !s.is_empty()); .map_err(|err| format!("set-parent-bulk: `{}` {err}", e.child))?;
(child, parent) 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(); .collect::<Result<Vec<_>, String>>()
match state.coord.reparent_bulk_with_notify(&moves).await { .map_err(|e| error_problem(&e))?;
Ok(()) => { let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect();
let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard");
tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); submit::reparent(
(StatusCode::OK, "ok").into_response() &state.coord,
} moves,
Err(e) => error_response(&format!("set-parent-bulk failed: {e}")), Source::Manual,
} "manual set-parent-bulk via dashboard".to_owned(),
);
Ok((StatusCode::OK, "ok").into_response())
} }

View file

@ -196,12 +196,6 @@ pub enum NodeKind {
/// window. `(child, new_parent)` pairs, applied in order under one /// window. `(child, new_parent)` pairs, applied in order under one
/// `META_LOCK` acquisition / one git commit (`meta::bulk_commit_topology` /// `META_LOCK` acquisition / one git commit (`meta::bulk_commit_topology`
/// handles both the single- and multi-move case uniformly). /// 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 { Reparent {
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
}, },

View file

@ -422,21 +422,12 @@ pub fn meta_update(
} }
/// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs — /// 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 /// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like
/// `set-parent`/`set-parent-bulk` HTTP handlers or the `hivectl`/MCP /// everything else in this module: submits and returns a DAG id
/// `SetParent` surface — those still call `Coordinator::reparent*_with_notify` /// immediately, the caller learns the outcome async (dashboard job view /
/// directly, which blocks until the commit lands and returns a synchronous /// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest::
/// `ok`/`err`, unlike every other queue-backed op. Whether that call-site /// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/
/// swap should keep the synchronous contract or go fire-and-forget (submit /// `set-parent-bulk` handlers.
/// 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"
)]
pub fn reparent( pub fn reparent(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,

View file

@ -361,12 +361,6 @@ pub fn meta_update(
/// dedicated variant would) and the per-template history-retention bucket — /// dedicated variant would) and the per-template history-retention bucket —
/// both cosmetic. Swap this to whatever the eventual node-kind-derived /// both cosmetic. Swap this to whatever the eventual node-kind-derived
/// dispatch lands with, whenever it lands. /// 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( pub fn reparent(
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
source: Source, source: Source,

View file

@ -562,41 +562,10 @@ pub async fn commit_perms(
Ok(()) Ok(())
} }
/// Write the topology file and commit it atomically under `META_LOCK`. /// Applies every `(child, new_parent)` move under a single `META_LOCK`
/// Returns `Err(String)` on validation failure (unknown agent, cycle, /// acquisition and creates **one** git commit for all of them — a
/// etc.) — same shape as `topology::set_parent` — so callers can /// single-move call is just a one-element slice, so there's no separate
/// surface the error as a user-visible message. Git failures are /// non-batch entry point. Moves are applied in the order given; the first
/// 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("<root>")),
&["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
/// validation error short-circuits the whole batch. True atomic write: all /// validation error short-circuits the whole batch. True atomic write: all
/// moves are pre-validated against a cumulative in-memory state with /// moves are pre-validated against a cumulative in-memory state with
/// [`crate::topology::apply_set_parent`] before anything touches disk, then /// [`crate::topology::apply_set_parent`] before anything touches disk, then

View file

@ -180,19 +180,17 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
} }
HostRequest::SetParent { child, new_parent } => { HostRequest::SetParent { child, new_parent } => {
tracing::info!(%child, ?new_parent, "set_parent"); tracing::info!(%child, ?new_parent, "set_parent");
// `reparent_with_notify` wraps `topology::set_parent` // Fire-and-forget, like every other queue-backed op:
// with the three notification messages + the // submit returns a DAG id immediately, the caller polls
// ContainerView rescan. Idempotent same-parent calls // `QueueDag` (`hivectl`'s wait/progress loop) for the
// skip both the messages and the disk write per the // outcome instead of blocking here on the commit.
// topology fast-path. let id = crate::job_queue::submit::reparent(
coord &coord,
.reparent_with_notify( vec![(child.clone(), new_parent.clone())],
child.as_str(), crate::job_queue::Source::Manual,
new_parent.as_ref().map(hive_types::Ident::as_str), "manual set-parent via hivectl".to_owned(),
) );
.await HostResponse::queued(vec![id])
.map_err(anyhow::Error::msg)?;
HostResponse::success()
} }
HostRequest::SetResourceLimits { HostRequest::SetResourceLimits {
name, name,