vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

wip(#3001): convert the last submit call sites; the binary compiles again

Browse source 
`server.rs`'s five sites move to `power::{stop,start,restart}_many` and direct
template inserts. `submit_single` routes through the `*_many` builders with a
one-element slice rather than keeping a parallel single-target shape.

`templates::rebuild` and `templates::reparent` now return the guids of the roots
they declare, so a caller that has to wait on them can name them; previously
only the void-returning form existed and every caller got an empty id list.

Two comments corrected while converting, both contradicted by the code they sit
above:

* `templates::rebuild` said its tail is edged onto "(MetaSync, Prebuild,
  Reconcile)" and that "Prebuild's roll-up carries the subtree" — the brace has
  been the middle root since the AgentWindow change.
* the restart handler described the per-agent shape as starting with SetWanted,
  while `restart_chain`'s own doc says a restart never rewrites `wanted` — that
  is the difference between restart and stop/start.

Error handling is no longer swallowed: a failed insert becomes a reported error
rather than a silently-absent id.
This commit is contained in:
atlas 2026-08-04 18:06:14 +02:00 • committed by mara
commit 0523b4f7de
2 changed files with 83 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

132
hive-c0re/src/server.rs
View file

@ -180,13 +180,19 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
// 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])
let inserted = coord.job_queue.insert_job(|b| {
vec![crate::job_queue::templates::reparent(
b,
vec![(child.clone(), new_parent.clone())],
)]
});
match inserted {
Ok(ids) => {
coord.emit_rebuild_queue_snapshot();
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
}
Err(e) => HostResponse::error(format!("queue reparent: {e}")),
}
}
HostRequest::SetResourceLimits {
name,
@ -777,39 +783,36 @@ enum Verb {
}
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
use crate::job_queue::{Source, submit};
let id = match verb {
use crate::job_queue::power;
// A single-target op is the N-target one with N = 1 — the shapes are
// identical, so there is no separate builder to keep in sync.
let targets = [name.to_owned()];
let ids = match verb {
Verb::Kill => {
tracing::info!(%name, "kill");
submit::stop(
coord,
name,
Source::Manual,
"manual kill via hivectl".to_owned(),
)
.await
power::stop_many(coord, &targets, false).await
}
Verb::Restart => {
tracing::info!(%name, "restart");
submit::restart(
coord,
name,
Source::Manual,
"manual restart via hivectl".to_owned(),
)
.await
power::restart_many(coord, &targets, false).await
}
Verb::Rebuild => {
tracing::info!(%name, "rebuild");
submit::rebuild(
coord,
name,
Source::Manual,
"manual rebuild via hivectl".to_owned(),
)
// Not a power op: a rebuild's shape doesn't depend on live state,
// so it is a plain template insert rather than a `*_many` gather.
let inserted = coord
.job_queue
.insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
if inserted.is_ok() {
coord.emit_rebuild_queue_snapshot();
}
inserted
}
};
HostResponse::queued(vec![id])
match ids {
Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
}
}
/// Stop the given `agents` (resolved logical names) then `infra` containers
@ -847,17 +850,13 @@ async fn handle_stop(
} else {
"manual via hivectl stop"
};
queued.push(
crate::job_queue::submit::stop_many(
coord,
agents,
graceful,
crate::job_queue::Source::Manual,
reason.to_owned(),
)
.await,
);
ok_items.extend(agents.iter().cloned());
match crate::job_queue::power::stop_many(coord, agents, graceful).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue stop: {e}")),
}
}
// Agents go down before infra so they're not mid-request against a
@ -944,16 +943,13 @@ async fn handle_start(
// hivectl's wait loop.
let mut queued: Vec<u64> = Vec::new();
if !agents.is_empty() {
queued.push(
crate::job_queue::submit::start_many(
coord,
agents,
crate::job_queue::Source::Manual,
"manual via hivectl start".to_owned(),
)
.await,
);
ok_items.extend(agents.iter().cloned());
match crate::job_queue::power::start_many(coord, agents).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue start: {e}")),
}
}
let mut resp = finish_lifecycle(ok_items, &errors);
@ -982,28 +978,20 @@ async fn handle_restart_scoped(
let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new();
// One DAG for all targeted agents — a per-agent restart subgraph each
// (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`),
// independent roots that run concurrently on their own leases. A
// hive-wide `hivectl restart` is now a single DAG, not N. No
// client-side stop-then-start composition — the whole restart survives
// a dropped connection because the DAG owns it.
// One insert for all targeted agents — a per-agent restart subgraph each
// (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
// restart converges to the agent's *existing* intent rather than rewriting
// it), independent roots running concurrently on their own leases. No
// client-side stop-then-start composition — the whole restart survives a
// dropped connection because the graph owns it.
if !agents.is_empty() {
queued.push(
crate::job_queue::submit::restart_many(
coord,
&agents,
graceful,
crate::job_queue::Source::Manual,
if graceful {
"manual via hivectl restart --graceful".to_owned()
} else {
"manual restart via hivectl restart".to_owned()
},
)
.await,
);
ok_items.extend(agents.iter().cloned());
match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue restart: {e}")),
}
}
for &container in &infra {