From 59c47e15aa28324086cd78d75341e50f2409ef25 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 16 Aug 2026 20:13:02 +0200 Subject: [PATCH] swarm-controller: run the swarm-level job-graph scheduler loop --- swarm-controller/src/main.rs | 112 ++++++++++++++++++++++++++++++----- 1 file changed, 98 insertions(+), 14 deletions(-) diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index a42aaa95..5be49dde 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -37,11 +37,16 @@ mod status; /// Placeholder node payload for the swarm-level job graph — uninhabited on /// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource` /// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't -/// land on both crates. This wires the graph and its read-only endpoints; -/// giving it real variants waits on there being an actual job (agent -/// creation) to run. `WireNode` is trivially satisfiable on an empty enum -/// (`match *self {}`), so the wire machinery below is real and typechecked -/// today, with nothing yet to put in it. +/// land on both crates. The *scheduler loop* below is real and running +/// (`spawn_jobq_worker`, mirroring `hive-c0re/src/job_queue/scheduler.rs`'s +/// `run_worker`) — what's still missing is a real job to give it: no +/// variant exists yet, so nothing is ever inserted into the graph and +/// `claim_next` always returns `None`. Giving this real variants (starting +/// with `CreateRepo`) is the next slice, landing together with +/// `swarm-controller::forge`, the client those nodes will call. `WireNode` +/// is trivially satisfiable on an empty enum (`match *self {}`), so the +/// wire machinery below is real and typechecked today, with nothing yet to +/// put in it. #[derive(Clone, Debug)] enum SwarmNodeKind {} @@ -66,6 +71,73 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { } } +/// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ +/// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` +/// variant turns into a real effect. Trivially exhaustive today +/// (`match kind {}`) because the enum has no variants yet; the first +/// real arm (`CreateRepo`, calling `swarm-controller::forge`) lands +/// alongside that variant, not before. +async fn run_swarm_node( + _id: hive_jobq::NodeId, + kind: SwarmNodeKind, + builder: hive_jobq::builder::JobBuilder, +) -> ( + hive_jobq::builder::JobBuilder, + hive_jobq::scheduler::Outcome, +) { + let _ = builder; + match kind {} +} + +/// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/ +/// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node, +/// spawn the future that runs + completes it, loop again immediately if +/// something started (more may now be runnable), otherwise back off +/// briefly before re-polling. +/// +/// No shutdown signal to wire in — unlike `hive-c0re`'s `coord.shutdown_rx()`, +/// this daemon has no graceful-shutdown machinery at all yet (`main`'s +/// `axum::serve` runs unconditionally to process exit), so this loop +/// matches that: it rides the runtime down with the process, same as +/// every in-flight HTTP request does. +/// +/// Cheap to run with an empty graph: `claim_next` on a graph nothing was +/// ever inserted into just returns `None` every poll, so this is a +/// harmless idle loop until the first real node kind exists. +fn spawn_jobq_worker( + sched: Arc>>, +) { + tokio::spawn(async move { + loop { + let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, run_swarm_node); + match runner { + Some(runner) => { + tokio::spawn(async move { + let (id, grew) = runner.await; + if let Err(e) = grew { + tracing::warn!( + node = id.get(), + error = %e, + "swarm jobq: grown job rejected" + ); + } + }); + // Something just started — more may be runnable right + // now, so loop again immediately rather than sleeping. + } + None => { + // Nothing runnable. Bounded poll rather than an event + // wake (unlike hive-c0re's `notify.notify_one()`, + // there is no completion-signal channel here yet) — + // fine at this daemon's scale (one graph, no + // submitters yet); revisit if/when that stops holding. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + } + } + }); +} + /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. /// /// A compiled-in default is legitimate here and is *not* the mistake that @@ -142,11 +214,15 @@ struct AppState { /// that is merely *unreachable* still yields a reader, because /// `async-nats` reconnects underneath it. status: Option>, - /// The swarm-level job graph. `std::sync::Mutex`, not `tokio`'s — every - /// lock scope below is synchronous (no `.await` while held). Always a - /// graph, never gated on the swarm queue: this is process state, not - /// something read over the network. - jobq: Arc>>, + /// The swarm-level job graph, wrapped in its + /// [`hive_jobq::scheduler::Scheduler`] now that something drives it + /// (`spawn_jobq_worker`) — the graph alone was enough for the + /// read-only endpoints, the scheduler is what a `claim_next` loop + /// needs. `std::sync::Mutex`, not `tokio`'s — every lock scope below + /// is synchronous (no `.await` while held). Always present, never + /// gated on the swarm queue: this is process state, not something + /// read over the network. + jobq: Arc>>, } /// Env var the controller's NixOS module sets from @@ -316,10 +392,11 @@ async fn get_jobq_graph( axum::extract::Query(q): axum::extract::Query, ) -> Json> { let states = hive_jobq_wire::parse_states(q.states.as_deref()); - let graph = state + let sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = sched.graph(); let roots: Vec = graph.roots().map(|n| n.id).collect(); let nodes = graph.wire_snapshot(roots); Json(hive_jobq_wire::filter_nodes_by_state( @@ -336,12 +413,13 @@ async fn get_jobq_graph( tag = "jobq" )] async fn get_jobq_rollup(State(state): State) -> Json> { - let graph = state + let sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = sched.graph(); let roots: Vec = graph.roots().map(|n| n.id).collect(); - Json(hive_jobq_wire::state_rollup(&graph, roots)) + Json(hive_jobq_wire::state_rollup(graph, roots)) } #[tokio::main] @@ -423,11 +501,17 @@ async fn main() -> Result<()> { }, }; + let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + spawn_jobq_worker(Arc::clone(&jobq)); + let state = AppState { hives: Arc::new(load_hives()), links: Arc::new(load_links()), status, - jobq: Arc::new(Mutex::new(hive_jobq::Graph::new())), + jobq, }; let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi())