fix(#1218): bulk topology move uses one git commit via new set-parent-bulk endpoint
This commit is contained in:
parent
644519f358
commit
7ec0a36d7a
4 changed files with 179 additions and 4 deletions
|
|
@ -976,9 +976,10 @@ window.marked = marked;
|
||||||
}
|
}
|
||||||
sel.disabled = true;
|
sel.disabled = true;
|
||||||
const failures = [];
|
const failures = [];
|
||||||
for (const name of names) {
|
if (names.length === 1) {
|
||||||
|
// Single agent — use the form-encoded endpoint (backwards compat).
|
||||||
try {
|
try {
|
||||||
const body = new URLSearchParams({ child: name, new_parent: newParent });
|
const body = new URLSearchParams({ child: names[0], new_parent: newParent });
|
||||||
const resp = await fetch('/api/topology/set-parent', {
|
const resp = await fetch('/api/topology/set-parent', {
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
|
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
|
||||||
|
|
@ -989,10 +990,30 @@ window.marked = marked;
|
||||||
|| (resp.status >= 200 && resp.status < 400);
|
|| (resp.status >= 200 && resp.status < 400);
|
||||||
if (!ok) {
|
if (!ok) {
|
||||||
const text = await resp.text().catch(() => '');
|
const text = await resp.text().catch(() => '');
|
||||||
failures.push(`${name}: http ${resp.status}${text ? ' — ' + text.slice(0, 200) : ''}`);
|
failures.push(`${names[0]}: http ${resp.status}${text ? ' — ' + text.slice(0, 200) : ''}`);
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
failures.push(`${name}: ${err}`);
|
failures.push(`${names[0]}: ${err}`);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// Multiple agents — use the bulk endpoint so all moves land in
|
||||||
|
// a single git commit instead of one per agent.
|
||||||
|
try {
|
||||||
|
const payload = names.map((n) => ({ child: n, new_parent: newParent || null }));
|
||||||
|
const resp = await fetch('/api/topology/set-parent-bulk', {
|
||||||
|
method: 'POST',
|
||||||
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
body: JSON.stringify(payload),
|
||||||
|
redirect: 'manual',
|
||||||
|
});
|
||||||
|
const ok = resp.ok || resp.type === 'opaqueredirect'
|
||||||
|
|| (resp.status >= 200 && resp.status < 400);
|
||||||
|
if (!ok) {
|
||||||
|
const text = await resp.text().catch(() => '');
|
||||||
|
failures.push(`bulk: http ${resp.status}${text ? ' — ' + text.slice(0, 200) : ''}`);
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
failures.push(`bulk: ${err}`);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
sel.disabled = false;
|
sel.disabled = false;
|
||||||
|
|
|
||||||
|
|
@ -716,6 +716,57 @@ impl Coordinator {
|
||||||
Ok(())
|
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.
|
||||||
|
pub async fn reparent_bulk_with_notify(
|
||||||
|
self: &Arc<Self>,
|
||||||
|
moves: &[(&str, Option<&str>)],
|
||||||
|
) -> std::result::Result<(), String> {
|
||||||
|
if moves.is_empty() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
// bulk_commit_topology applies all set_parent calls under one lock
|
||||||
|
// and returns (child, old_parent) for every move that changed.
|
||||||
|
let changed = crate::meta::bulk_commit_topology(moves).await?;
|
||||||
|
|
||||||
|
// Send per-agent notifications for each changed move.
|
||||||
|
for (child, old_parent) in &changed {
|
||||||
|
// Find the new parent from the moves slice.
|
||||||
|
let new_parent = moves
|
||||||
|
.iter()
|
||||||
|
.find(|(c, _)| *c == child)
|
||||||
|
.and_then(|(_, np)| *np);
|
||||||
|
let old_label = old_parent.as_deref().unwrap_or("<root>");
|
||||||
|
let new_label = new_parent.unwrap_or("<root>");
|
||||||
|
if let Some(op) = old_parent.as_deref() {
|
||||||
|
let _ = self.broker.send(&hive_sh4re::Message {
|
||||||
|
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
||||||
|
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::SYSTEM_SENDER.to_owned(),
|
||||||
|
to: np.to_owned(),
|
||||||
|
body: format!(
|
||||||
|
"{child} just moved into your subtree (was previously under {old_label})"
|
||||||
|
),
|
||||||
|
in_reply_to: None,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
let _ = self
|
||||||
|
.broker
|
||||||
|
.send_coalescing_reparent(child, old_label, new_label);
|
||||||
|
}
|
||||||
|
|
||||||
|
self.rescan_containers_and_emit().await;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Read-only snapshot of the last cached container view. Used by
|
/// Read-only snapshot of the last cached container view. Used by
|
||||||
/// `/api/state` to cold-load page-open clients without re-running
|
/// `/api/state` to cold-load page-open clients without re-running
|
||||||
/// `nixos-container list` themselves; the
|
/// `nixos-container list` themselves; the
|
||||||
|
|
|
||||||
|
|
@ -78,6 +78,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
||||||
.route("/retry-reminder/{id}", post(post_retry_reminder))
|
.route("/retry-reminder/{id}", post(post_retry_reminder))
|
||||||
.route("/request-spawn", post(post_request_spawn))
|
.route("/request-spawn", post(post_request_spawn))
|
||||||
.route("/api/topology/set-parent", post(post_set_parent))
|
.route("/api/topology/set-parent", post(post_set_parent))
|
||||||
|
.route("/api/topology/set-parent-bulk", post(post_set_parent_bulk))
|
||||||
.route("/api/tool-groups", get(get_tool_groups))
|
.route("/api/tool-groups", get(get_tool_groups))
|
||||||
.route("/api/tool-groups/{agent}", post(post_tool_groups))
|
.route("/api/tool-groups/{agent}", post(post_tool_groups))
|
||||||
.route("/api/capabilities", get(get_capabilities))
|
.route("/api/capabilities", get(get_capabilities))
|
||||||
|
|
@ -1054,6 +1055,15 @@ struct SetParentForm {
|
||||||
new_parent: Option<String>,
|
new_parent: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// One entry in a `POST /api/topology/set-parent-bulk` JSON array.
|
||||||
|
/// `new_parent`: absent/null/empty-string all mean "promote to root".
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct SetParentBulkEntry {
|
||||||
|
child: String,
|
||||||
|
#[serde(default)]
|
||||||
|
new_parent: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
#[derive(Deserialize)]
|
||||||
struct AnswerForm {
|
struct AnswerForm {
|
||||||
answer: String,
|
answer: String,
|
||||||
|
|
@ -2476,6 +2486,36 @@ async fn post_set_parent(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `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.
|
||||||
|
async fn post_set_parent_bulk(
|
||||||
|
State(state): State<AppState>,
|
||||||
|
axum::Json(body): axum::Json<Vec<SetParentBulkEntry>>,
|
||||||
|
) -> Response {
|
||||||
|
if body.is_empty() {
|
||||||
|
return (StatusCode::OK, "ok").into_response();
|
||||||
|
}
|
||||||
|
// Collect borrows for the coordinator call.
|
||||||
|
let moves: Vec<(&str, Option<&str>)> = 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)
|
||||||
|
})
|
||||||
|
.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}")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ── tool-group endpoints ──────────────────────────────────
|
// ── tool-group endpoints ──────────────────────────────────
|
||||||
|
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
|
|
|
||||||
|
|
@ -397,6 +397,69 @@ pub async fn commit_topology(
|
||||||
Ok(())
|
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 (no partial writes because
|
||||||
|
/// `topology::set_parent` rewrites the in-memory map atomically and we only
|
||||||
|
/// flush to disk after all moves have passed validation).
|
||||||
|
///
|
||||||
|
/// Returns a `Vec` of `(child, old_parent)` pairs for every move that
|
||||||
|
/// actually changed the topology (idempotent same-parent moves are skipped),
|
||||||
|
/// so the caller can send targeted notifications.
|
||||||
|
pub async fn bulk_commit_topology(
|
||||||
|
moves: &[(&str, Option<&str>)],
|
||||||
|
) -> std::result::Result<Vec<(String, Option<String>)>, String> {
|
||||||
|
if moves.is_empty() {
|
||||||
|
return Ok(vec![]);
|
||||||
|
}
|
||||||
|
let _guard = META_LOCK.lock().await;
|
||||||
|
// Snapshot parents before any writes so we can compute the diff.
|
||||||
|
let topo_before = crate::topology::read();
|
||||||
|
// Validate + apply all moves in sequence; first error aborts.
|
||||||
|
for (child, new_parent) in moves {
|
||||||
|
crate::topology::set_parent(child, *new_parent)?;
|
||||||
|
}
|
||||||
|
// Commit the whole batch as one git operation.
|
||||||
|
let dir = meta_dir();
|
||||||
|
let commit_msg = if moves.len() == 1 {
|
||||||
|
let (child, new_parent) = moves[0];
|
||||||
|
format!("topology: {} → {}", child, new_parent.unwrap_or("<root>"))
|
||||||
|
} else {
|
||||||
|
let names: Vec<&str> = moves.iter().map(|(c, _)| *c).collect();
|
||||||
|
let dest = moves[0].1.unwrap_or("<root>");
|
||||||
|
format!(
|
||||||
|
"topology: move {} agents → {} ({})",
|
||||||
|
moves.len(),
|
||||||
|
dest,
|
||||||
|
names.join(", ")
|
||||||
|
)
|
||||||
|
};
|
||||||
|
let stage = async {
|
||||||
|
git(&dir, &["add", "topology.json"]).await?;
|
||||||
|
if has_staged_changes(&dir).await? {
|
||||||
|
git_commit(&dir, &commit_msg).await?;
|
||||||
|
}
|
||||||
|
Ok::<_, anyhow::Error>(())
|
||||||
|
};
|
||||||
|
if let Err(e) = stage.await {
|
||||||
|
tracing::warn!(error = ?e, "bulk_commit_topology: topology written but git commit failed (sync_agents will recover)");
|
||||||
|
}
|
||||||
|
// Return (child, old_parent) for each move that changed state.
|
||||||
|
let changed = moves
|
||||||
|
.iter()
|
||||||
|
.filter_map(|(child, new_parent)| {
|
||||||
|
let old = topo_before.get(*child).cloned().flatten();
|
||||||
|
if old.as_deref() != *new_parent {
|
||||||
|
Some((child.to_string(), old))
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
Ok(changed)
|
||||||
|
}
|
||||||
|
|
||||||
fn render_flake(
|
fn render_flake(
|
||||||
hyperhive_flake: &str,
|
hyperhive_flake: &str,
|
||||||
nixpkgs_flake: &str,
|
nixpkgs_flake: &str,
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue