From 530137a7bba68f9f88e1549864f9f12241b63e4e Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 7 Nov 2026 15:26:54 +0100 Subject: [PATCH] hive-c0re: queue another knowledge pull soon after a failed one, backing off to 5 minutes --- docs/2-handbook/swarm/knowledge.md | 5 + .../4-implementation/scheduler/coordinator.md | 2 +- hive-c0re/src/job_queue/exec.rs | 139 ++++++++++++++- hive-c0re/src/job_queue/exec/tests.rs | 168 ++++++++++++++++-- hive-c0re/src/job_queue/model.rs | 3 +- 5 files changed, 298 insertions(+), 19 deletions(-) diff --git a/docs/2-handbook/swarm/knowledge.md b/docs/2-handbook/swarm/knowledge.md index 5dfb9d73..0b7d9d29 100644 --- a/docs/2-handbook/swarm/knowledge.md +++ b/docs/2-handbook/swarm/knowledge.md @@ -68,6 +68,11 @@ pull`, so agents see the new content on their next turn. shows as a failed `KnowledgePull` node on the job queue, counts toward the `knowledge_pull` hive warning, and doesn't affect the rest of the daemon. +A failed pull of either kind, or the one at boot, also queues a retry after 5 s, +doubling per failure in a row up to 5 min, until a pull succeeds. The +`knowledge_pull` hive warning appears after seven failures in a row, about five +minutes in, so a forge still starting at boot raises nothing. + 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 — each trigger queues its own pass, and a pass that lands diff --git a/docs/4-implementation/scheduler/coordinator.md b/docs/4-implementation/scheduler/coordinator.md index 9ac6c8d8..5238f37a 100644 --- a/docs/4-implementation/scheduler/coordinator.md +++ b/docs/4-implementation/scheduler/coordinator.md @@ -72,7 +72,7 @@ Cheap — no build slot: | `SetWanted` | record the power intent (`wanted = Up`/`Offline`, in memory) as the head node of a power-op DAG. Takes the agent lease even though it writes no container state, so the intent write and the tail `Reconcile` are atomic per-agent — two racing power ops can't clobber each other's intent before either reconciles | | `EmitRebuilt` | tail of a rebuild — emit the agent's `Rebuilt` manager event (ok/fail per outcome, nothing on cancel). One node per agent _and_ per outcome | | `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | -| `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 | +| `KnowledgePull` | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, hourly as a fallback, and after a failed pull (5 s, doubling to 5 min, until one succeeds). 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` | pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. Runs at boot, waiting up to a minute in total for the queue connection, and again after every attach of the wanted-state watch (which retries with a backoff when the attach fails or the stream ends); the deploy event (`swarm_status`) and the watch are the fast path, this repairs a missed one. A first deploy already queued or running for an agent isn't queued again. Agentless | diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 21bbf23f..f969f4fe 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -315,7 +315,16 @@ async fn dispatch( // `Finishing` so the nodes under it start. What they declare stays held // until their whole subtree settles. NodeKind::AgentWindow { .. } => Ok(()), - NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await, + NodeKind::KnowledgePull => { + run_knowledge_pull( + host, + coord, + KNOWLEDGE_PULL_DEADLINE, + knowledge_pull_health(), + &KNOWLEDGE_PULL_RETRY, + ) + .await + } NodeKind::WantedPull => run_wanted_pull(host, coord).await, }; (builder, result) @@ -323,11 +332,14 @@ async fn dispatch( /// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every /// pass reports here, whoever submitted it: as the node's own error on the -/// dashboard, and into the debounced warning. +/// dashboard, and into the debounced warning. A failed pass queues the next one +/// through `retry`. async fn run_knowledge_pull( host: &impl Host, coord: &Arc, deadline: std::time::Duration, + health: &Mutex, + retry: &'static PullRetry, ) -> Result<()> { let result = tokio::time::timeout(deadline, host.knowledge_pull(coord)) .await @@ -336,12 +348,18 @@ async fn run_knowledge_pull( "knowledge pull timed out after {deadline:?}" )) }); - let mut health = knowledge_pull_health() + let mut health = health .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); match &result { - Ok(()) => health.record_ok(), + Ok(()) => { + health.record_ok(); + retry.record_ok(); + } Err(e) => { + if let Some((wait, generation)) = retry.record_err() { + queue_pull_after(coord.clone(), wait, generation, retry); + } let err = format!("{e:#}"); health.record_err(|ctx| { let age = ctx.since_last_ok.map_or_else( @@ -359,6 +377,107 @@ async fn run_knowledge_pull( result } +/// The wait before the first retry of a failed pull; doubled by each further +/// failure in a row. +const KNOWLEDGE_RETRY_FIRST: std::time::Duration = std::time::Duration::from_secs(5); + +/// The longest wait between a failed pull and its retry. +const KNOWLEDGE_RETRY_MAX: std::time::Duration = std::time::Duration::from_mins(5); + +/// Consecutive failed pulls before the hive warning is raised. At the retry +/// schedule above the seventh failure comes about five minutes after the +/// first, so a forge that is still starting does not raise it. +const KNOWLEDGE_WARN_AFTER: u32 = 7; + +/// The retry of the node's pulls: the boot, event and hourly triggers share it. +static KNOWLEDGE_PULL_RETRY: PullRetry = PullRetry::new(); + +/// Retry bookkeeping for failed `/knowledge` pulls: the failures in a row and +/// the retry that is waiting, so any number of failing triggers leave at most +/// one retry queued, and a success cancels it. +struct PullRetry(Mutex); + +#[derive(Clone, Copy)] +struct RetryState { + failures: u32, + waiting: bool, + /// Bumped by every success; a retry queued under an older one is stale. + generation: u32, +} + +impl PullRetry { + const fn new() -> Self { + Self(Mutex::new(RetryState { + failures: 0, + waiting: false, + generation: 0, + })) + } + + fn state(&self) -> std::sync::MutexGuard<'_, RetryState> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// A pull succeeded: the next failure starts the backoff over, and a + /// retry that is waiting no longer queues a pull. + fn record_ok(&self) { + let mut s = self.state(); + s.failures = 0; + s.waiting = false; + s.generation = s.generation.wrapping_add(1); + } + + /// A pull failed. `Some((wait, generation))` when the caller must queue a + /// retry after `wait`, passing `generation` to [`Self::queued`]; `None` + /// when one is already waiting. + fn record_err(&self) -> Option<(std::time::Duration, u32)> { + let mut s = self.state(); + let wait = KNOWLEDGE_RETRY_FIRST + .saturating_mul(1_u32 << s.failures.min(16)) + .min(KNOWLEDGE_RETRY_MAX); + s.failures = s.failures.saturating_add(1); + if s.waiting { + return None; + } + s.waiting = true; + Some((wait, s.generation)) + } + + /// The waiting retry is due. `true` when it should queue its pull, `false` + /// when a success has cancelled it since `generation`. + fn queued(&self, generation: u32) -> bool { + let mut s = self.state(); + if s.generation != generation { + return false; + } + s.waiting = false; + true + } +} + +/// Queue one `/knowledge` pull after `wait`, unless a success cancels it first. +fn queue_pull_after( + coord: Arc, + wait: std::time::Duration, + generation: u32, + retry: &'static PullRetry, +) { + tokio::spawn(async move { + tokio::time::sleep(wait).await; + if !retry.queued(generation) { + return; + } + if let Err(e) = coord + .job_queue + .insert_job(|b| vec![super::templates::knowledge_pull(b).guid()]) + { + tracing::warn!(error = ?e, "knowledge pull retry: submit failed"); + } + }); +} + /// A pull still running holds [`Resource::KnowledgeTree`] against every later /// one. The `reset` / `clean` / `pull` git children are `kill_on_drop`, so the /// deadline kills them. @@ -366,11 +485,17 @@ async fn run_knowledge_pull( /// [`Resource::KnowledgeTree`]: super::resource::Resource::KnowledgeTree const KNOWLEDGE_PULL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(5); -/// Warning for a `/knowledge` pull that keeps failing: three misses in a row, -/// cleared by the next successful pull. +/// Warning for a `/knowledge` pull that keeps failing: [`KNOWLEDGE_WARN_AFTER`] +/// misses in a row, cleared by the next successful pull. fn knowledge_pull_health() -> &'static Mutex { static HEALTH: OnceLock> = OnceLock::new(); - HEALTH.get_or_init(|| Mutex::new(SweepHealth::new("knowledge_pull", "warn", 3))) + HEALTH.get_or_init(|| { + Mutex::new(SweepHealth::new( + "knowledge_pull", + "warn", + KNOWLEDGE_WARN_AFTER, + )) + }) } /// Boot-time swarm wanted-state pull as a DAG node — see diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index 7a3fa476..618dd8ed 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -11,15 +11,17 @@ use hive_jobq::NodeId; use hive_jobq::scheduler::Scheduler; use super::{ - Host, dispatch, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain, - run_emit_rebuilt, run_knowledge_pull, run_meta_sync, run_pause_drain, run_pause_signal, - run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping, run_reconcile, - run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap, - run_wanted_pull, run_write_dropin, + Host, KNOWLEDGE_WARN_AFTER, PullRetry, dispatch, run_create, run_destroy_bookkeeping, + run_destroy_container, run_drain, run_emit_rebuilt, run_knowledge_pull, run_meta_sync, + run_pause_drain, run_pause_signal, run_prebuild, run_provision, run_purge_state, + run_rebuild_bookkeeping, run_reconcile, run_set_wanted, run_signal, run_start, run_stop, + run_stop_for_update, run_swap, run_wanted_pull, run_write_dropin, }; use crate::coordinator::{AgentPaths, Coordinator, HiveEnv}; use crate::job_queue::model::NodeKind; use crate::power::Wanted; +use crate::stats::sweep_health::SweepHealth; +use crate::warnings::snapshot; use swarm_wire::agent_closures::AgentClosure; const REV: &str = "2222222222222222222222222222222222222222"; @@ -1056,16 +1058,30 @@ async fn emit_rebuilt_reports_success_and_failure_under_one_key() { // --- Sweeps and pulls --------------------------------------------------------- +/// A pull's health tracker, as production builds it, under a `kind` of its own +/// so tests do not share the warning. +fn pull_health(kind: &'static str) -> Mutex { + Mutex::new(SweepHealth::new(kind, "warn", KNOWLEDGE_WARN_AFTER)) +} + #[tokio::test] async fn knowledge_pull_passes_the_result_through_and_times_out() { + static RETRY: PullRetry = PullRetry::new(); + let health = pull_health("pull_result"); let (_dir, coord) = coordinator(); let secs = Duration::from_secs(5); - run_knowledge_pull(&FakeHost::up(), &coord, secs) + run_knowledge_pull(&FakeHost::up(), &coord, secs, &health, &RETRY) .await .expect("ok"); - let err = run_knowledge_pull(&FakeHost::up().failing("knowledge_pull"), &coord, secs) - .await - .expect_err("pull fails"); + let err = run_knowledge_pull( + &FakeHost::up().failing("knowledge_pull"), + &coord, + secs, + &health, + &RETRY, + ) + .await + .expect_err("pull fails"); assert!( format!("{err:#}").contains("knowledge_pull failed"), "{err:#}" @@ -1074,12 +1090,144 @@ async fn knowledge_pull_passes_the_result_through_and_times_out() { hang: true, ..FakeHost::up() }; - let err = run_knowledge_pull(&hang, &coord, Duration::from_millis(50)) + let err = run_knowledge_pull(&hang, &coord, Duration::from_millis(50), &health, &RETRY) .await .expect_err("deadline"); assert!(format!("{err:#}").contains("timed out"), "{err:#}"); } +fn queued_pulls(coord: &Arc) -> usize { + let sched = coord.job_queue.sched().lock().expect("job_queue mutex"); + sched + .graph() + .nodes() + .filter(|n| matches!(n.payload, NodeKind::KnowledgePull)) + .count() +} + +/// A failed pull queues one more after the first retry wait, however many +/// pulls fail meanwhile. +#[tokio::test(start_paused = true)] +async fn a_failed_pull_queues_one_retry_after_a_wait() { + static RETRY: PullRetry = PullRetry::new(); + let health = pull_health("pull_retry"); + let (_dir, coord) = coordinator(); + let secs = Duration::from_secs(5); + let failing = FakeHost::up().failing("knowledge_pull"); + run_knowledge_pull(&failing, &coord, secs, &health, &RETRY) + .await + .expect_err("pull fails"); + run_knowledge_pull(&failing, &coord, secs, &health, &RETRY) + .await + .expect_err("pull fails"); + tokio::time::sleep(Duration::from_secs(4)).await; + assert_eq!(queued_pulls(&coord), 0, "queued before the wait ran out"); + tokio::time::sleep(Duration::from_secs(2)).await; + assert_eq!(queued_pulls(&coord), 1, "one retry after the wait"); +} + +/// A pull that succeeds queues nothing, and cancels the retry a failed one +/// left waiting. +#[tokio::test(start_paused = true)] +async fn a_pull_that_succeeds_queues_no_retry() { + static RETRY: PullRetry = PullRetry::new(); + let health = pull_health("pull_ok"); + let (_dir, coord) = coordinator(); + let secs = Duration::from_secs(5); + run_knowledge_pull(&FakeHost::up(), &coord, secs, &health, &RETRY) + .await + .expect("ok"); + run_knowledge_pull( + &FakeHost::up().failing("knowledge_pull"), + &coord, + secs, + &health, + &RETRY, + ) + .await + .expect_err("pull fails"); + run_knowledge_pull(&FakeHost::up(), &coord, secs, &health, &RETRY) + .await + .expect("ok"); + tokio::time::sleep(Duration::from_mins(10)).await; + assert_eq!(queued_pulls(&coord), 0); +} + +#[test] +fn the_retry_wait_doubles_up_to_its_cap_and_starts_over_after_a_success() { + let retry = PullRetry::new(); + let mut waits = Vec::new(); + for _ in 0..8 { + let (wait, generation) = retry.record_err().expect("no retry is waiting"); + waits.push(wait.as_secs()); + assert!(retry.queued(generation)); + } + assert_eq!(waits, [5, 10, 20, 40, 80, 160, 300, 300]); + retry.record_ok(); + let first = retry.record_err().map(|(wait, _)| wait.as_secs()); + assert_eq!(first, Some(5)); + assert_eq!(retry.record_err(), None, "a retry is already waiting"); +} + +/// A retry that a success has cancelled does not queue, and does not stop the +/// next failure from scheduling its own. +#[test] +fn a_success_cancels_the_waiting_retry() { + let retry = PullRetry::new(); + let (_, stale) = retry.record_err().expect("first retry"); + retry.record_ok(); + let (_, current) = retry.record_err().expect("a new retry after the success"); + assert!(!retry.queued(stale), "the cancelled retry queued a pull"); + assert!(retry.queued(current)); +} + +/// The warning waits for a failure that has lasted minutes, which the first +/// few quick retries are not: the forge still starting at boot raises nothing. +#[tokio::test] +async fn the_warning_waits_for_a_sustained_failure() { + static RETRY: PullRetry = PullRetry::new(); + let health = pull_health("pull_warning"); + let (_dir, coord) = coordinator(); + let failing = FakeHost::up().failing("knowledge_pull"); + let raised = || snapshot().iter().any(|w| w.kind == "pull_warning"); + // The first three quick failures, then the rest short of the threshold. + for _ in (1..=3).chain(4..KNOWLEDGE_WARN_AFTER) { + run_knowledge_pull(&failing, &coord, Duration::from_secs(5), &health, &RETRY) + .await + .expect_err("pull fails"); + assert!(!raised(), "raised before the failure was sustained"); + } + run_knowledge_pull(&failing, &coord, Duration::from_secs(5), &health, &RETRY) + .await + .expect_err("pull fails"); + assert!(raised(), "not raised after a sustained failure"); + run_knowledge_pull( + &FakeHost::up(), + &coord, + Duration::from_secs(5), + &health, + &RETRY, + ) + .await + .expect("ok"); + assert!(!raised(), "not cleared by the next success"); +} + +#[test] +fn the_seventh_failure_comes_minutes_after_the_first() { + let retry = PullRetry::new(); + let mut elapsed = Duration::ZERO; + for _ in 1..KNOWLEDGE_WARN_AFTER { + let (wait, generation) = retry.record_err().expect("no retry is waiting"); + elapsed += wait; + assert!(retry.queued(generation)); + } + assert!( + elapsed >= Duration::from_mins(5), + "the warning would come after {elapsed:?}" + ); +} + #[tokio::test] async fn wanted_pull_passes_the_result_through() { let (_dir, coord) = coordinator(); diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 780cb4ee..9f3850c7 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -97,7 +97,8 @@ pub enum NodeKind { /// else `Offline`) as the head node of a power-op DAG. SetWanted { agent: String, up: bool }, /// `/knowledge` pull (`knowledge::pull`): at boot, on the swarm - /// knowledge-changed event, and hourly. Agentless. + /// knowledge-changed event, hourly, and after a failed pull (backed off). + /// Agentless. KnowledgePull, /// One-shot boot-time pull of the agent set the swarm controller /// declares for this hive (`wanted::pull`). Agentless.