diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index ea477e35..d235653b 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -34,16 +34,37 @@ struct NodeDone { /// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`. /// /// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true signal -/// the loop exits immediately; already-running node tasks ride the runtime down -/// with the process, and pending `Queued` DAGs are dropped — desired state is -/// re-derived on next boot (boot sweep + reconcile), so the in-memory queue is -/// deliberately not durable. +/// the loop exits — checked explicitly at the top of every iteration (not just +/// in the `select!` below), so a sustained stream of ready work can't defer +/// exit until the queue happens to drain. Already-running node tasks ride the +/// runtime down with the process, and pending `Queued` DAGs are dropped — +/// desired state is re-derived on next boot (boot sweep + reconcile), so the +/// in-memory queue is deliberately not durable. +/// +/// That's safe only because every [`super::NodeKind`] is +/// idempotent-convergent: re-running a half-applied node converges instead of +/// double-applying (e.g. a dropped `Swap` re-derives to the same +/// `nixos-container update`, which is itself declarative). That's a property +/// of the node set, not of the queue — it holds today but isn't enforced by +/// the type system, so it's worth re-checking whenever a new `NodeKind` is +/// added. The side-effect tails (`EmitRebuilt`, `ResolveApproval`) sit +/// closest to the edge: dropping or re-running one affects something outside +/// the graph (today, a missed or duplicated notification) rather than +/// reconverging silently. pub async fn run_worker(coord: Arc) { let mut shutdown = coord.shutdown_rx(); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); // (DAG id, agent) → transient guard held for that agent's lease window. let mut transients: HashMap<(u64, String), crate::coordinator::TransientGuard> = HashMap::new(); loop { + // Checked every iteration, not just in the `select!` below — a + // continuous stream of ready claims never reaches the `select!`, so + // relying on it alone as the only shutdown observation point defers + // exit until the queue drains. See the doc comment above. + if *shutdown.borrow() { + tracing::info!("job_queue: scheduler exiting on shutdown"); + return; + } reconcile_transients(&coord, &mut transients); let claims = coord.job_queue.claim_ready(); if !claims.is_empty() {