diff --git a/docs/integrations/knowledge.md b/docs/integrations/knowledge.md index 3a113f52..76a97064 100644 --- a/docs/integrations/knowledge.md +++ b/docs/integrations/knowledge.md @@ -69,11 +69,10 @@ pull`, so agents see the new content on their next turn. Both paths, and the one pull at boot, queue the same `KnowledgePull` job node. It holds the working tree for its duration, so two pulls never run -over each other. An event that arrives while a pull is waiting to start -folds into it; one that arrives while a pull is running queues one more -behind it, since the running pull may have fetched before the push. The -node runs `knowledge::pull()`, which also handles the change notice -below. +over each other — each trigger queues its own pass, and a pass that lands +while another is already queued or running simply waits its turn, since the +one ahead of it may have fetched before the push. The node runs +`knowledge::pull()`, which also handles the change notice below. ### Change notice diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 0c9552c9..853d41d7 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -81,9 +81,9 @@ Cheap — no build slot: | `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | | `WritePermFile` | commit `tool-groups.json` / `capabilities.json` (single git commit under `META_LOCK`) + emit the P3RM1SS10NS snapshots | | `ForgeSweep` | one-shot boot-time forge user/token sweep for every container (`forge::ensure_all`) as a first-class node, so it shows as real work on the dashboard instead of running invisibly in a bare `tokio::spawn`. Agentless | -| `MatrixSweep` | matrix user/space sweep (`matrix::ensure_all`): the boot-time instance, plus one every 30 min from a loop in `main.rs`. Holds `Resource::MatrixSweep` (capacity 1), so two passes never overlap; a periodic tick finding one queued or running folds into it instead of stacking. Agentless | +| `MatrixSweep` | matrix user/space sweep (`matrix::ensure_all`): the boot-time instance, plus one every 30 min from a loop in `main.rs`. Holds `Resource::MatrixSweep` (capacity 1), so two passes never overlap; each tick queues its own pass, which waits for the resource if one is already live. Agentless | | `WebhookRegister` | one-shot boot-time Forgejo webhook registration (`internal/knowledge` push→pull, `agent-configs` PR→approval). No-op until the core token, hive domain, and HMAC secret are all available. Agentless | -| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree. The event folds into a queued pull but queues behind a running one; the hourly tick folds into either. Agentless | +| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree; each trigger queues its own pass, which waits for the resource if one is already live. Agentless | | `WantedPull` | one-shot boot-time pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. No background loop behind this one — boot is the whole cadence; the deploy event (`swarm_status`) is the fast path, this repairs a missed one. Agentless | diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f4130adf..91a44c57 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -192,39 +192,6 @@ impl JobQueue { Ok(named) } - /// Insert the single node `declare` names, unless a node of `kind` is - /// already in one of the `fold_into` states. `None` means the caller's - /// request folded into that node and nothing was inserted. - /// - /// For a standalone sweep, where a pass that has not started yet already - /// covers "one more pass": it reads whatever is current when it runs. The - /// read and the insert share one lock, so two racing callers cannot both - /// miss the other. - /// - /// # Errors - /// Propagates a graph-insert error. - pub fn insert_unless_live( - &self, - kind: &NodeKind, - fold_into: &[State], - declare: impl FnOnce(&JobBuilder) -> Handle<'_>, - ) -> anyhow::Result> { - let mut inner = self.lock(); - let live = inner - .graph() - .nodes() - .any(|n| &n.payload == kind && fold_into.contains(&n.state)); - if live { - return Ok(None); - } - let named = inner - .insert_job(None, |b| vec![declare(b).guid()]) - .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; - drop(inner); - self.notify.notify_one(); - Ok(named.first().copied()) - } - /// The scheduler itself, for `hive_jobq`'s run-loop seam /// (`Scheduler::claim_next`), which takes exactly this type. /// diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index b27a6b7e..ff75dddb 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -210,7 +210,7 @@ fn reconcile_transients(coord: &Arc, prev: &mut TransientSeen) { mod tests { use hive_jobq::scheduler::{Outcome, Scheduler}; - use super::super::{JobQueue, NodeKind, State, templates}; + use super::super::{JobQueue, State, templates}; /// Claim one runnable node through the same seam [`super::run_worker`] /// uses. The returned future completes the node when awaited. `None` when @@ -271,38 +271,28 @@ mod tests { second.await; } + /// A tick landing while a pass is already live queues a second pass; the + /// shared `Resource::KnowledgeTree` dep holds it back until the first + /// finishes, so the two never run together. #[tokio::test] - async fn a_tick_folds_into_a_live_pull_and_an_event_queues_one_behind_it() { + async fn a_tick_during_a_live_pull_queues_one_pass_behind_it() { let q = JobQueue::new(1); - let kind = NodeKind::KnowledgePull; - let queued = [State::Pending]; - let live = [State::Pending, State::Running]; - let submit = |fold_into: &[State]| { - q.insert_unless_live(&kind, fold_into, templates::knowledge_pull) - .expect("insert") - .is_some() - }; + q.insert_job(|b| vec![templates::knowledge_pull(b).guid()]) + .expect("insert"); + let live = claim(&q).expect("the first pull is runnable"); + assert_eq!(running(&q), ["knowledge_pull"]); - assert!(submit(&live), "nothing live: a tick inserts"); - let pull = claim(&q).expect("the pull is runnable"); - assert!(!submit(&live), "a tick folds into the running pull"); - assert!( - submit(&queued), - "an event queues a pull behind the running one" - ); - assert!( - !submit(&queued), - "a second event folds into the queued pull" - ); - assert!(!submit(&live), "a tick folds into the queued pull"); + q.insert_job(|b| vec![templates::knowledge_pull(b).guid()]) + .expect("insert"); assert_eq!(count(&q, "knowledge_pull"), 2); - assert!( - q.insert_unless_live(&NodeKind::MatrixSweep, &live, templates::matrix_sweep) - .expect("insert") - .is_some(), - "a live pull does not fold a different sweep" + claim(&q).is_none(), + "the second pull waits for the resource the first one holds" ); - pull.await; + + live.await; + let second = claim(&q).expect("the second pull runs once the first is done"); + assert_eq!(running(&q), ["knowledge_pull"]); + second.await; } } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index b1a84742..160fcf58 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -226,23 +226,17 @@ async fn main() -> Result<()> { } } -/// One periodic sweep tick. It folds into a pass already queued or running -/// instead of stacking another behind it, since that pass does the same work. +/// One periodic sweep tick. Queues unconditionally; the node's own +/// capacity-1 resource dep (`Resource::MatrixSweep` / `KnowledgeTree`) is +/// what keeps two passes of the same kind from running together, so a tick +/// landing while one is live just queues one more pass behind it. fn submit_periodic_sweep( coord: &Coordinator, kind: &job_queue::NodeKind, declare: impl FnOnce(&job_queue::JobBuilder) -> job_queue::Handle<'_>, ) { - let live = [job_queue::State::Pending, job_queue::State::Running]; - match coord.job_queue.insert_unless_live(kind, &live, declare) { - Ok(Some(_)) => {} - Ok(None) => tracing::debug!( - kind = <&str>::from(kind), - "periodic sweep: one already queued or running; folded into it" - ), - Err(e) => { - tracing::warn!(kind = <&str>::from(kind), error = ?e, "periodic sweep: submit failed"); - } + if let Err(e) = coord.job_queue.insert_job(|b| vec![declare(b).guid()]) { + tracing::warn!(kind = <&str>::from(kind), error = ?e, "periodic sweep: submit failed"); } } diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 43c89552..5c4b8516 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -238,16 +238,13 @@ async fn drain_swarm_events( return; } tracing::info!(%subject, "swarm events: knowledge change announced, pulling"); - // Folds into a pull that hasn't started, but queues behind a - // running one: that pull may have fetched before this push. - match coord.job_queue.insert_unless_live( - &crate::job_queue::NodeKind::KnowledgePull, - &[crate::job_queue::State::Pending], - crate::job_queue::templates::knowledge_pull, - ) { - Ok(Some(_)) => {} - Ok(None) => tracing::debug!("swarm events: a knowledge pull is already queued"), - Err(e) => tracing::warn!(error = ?e, "swarm events: knowledge pull submit failed"), + // Queues unconditionally; the node's `Resource::KnowledgeTree` + // dep keeps it from running alongside a pull already in + // flight, which may have fetched before this push landed. + if let Err(e) = coord.job_queue.insert_job(|b| { + vec![crate::job_queue::templates::knowledge_pull(b).guid()] + }) { + tracing::warn!(error = ?e, "swarm events: knowledge pull submit failed"); } } msg = deploy_sub.next() => {