hive-c0re: queue another knowledge pull soon after a failed one, backing off to 5 minutes

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<Coordinator>,
     deadline: std::time::Duration,
+    health: &Mutex<SweepHealth>,
+    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<RetryState>);
+
+#[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<Coordinator>,
+    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<SweepHealth> {
     static HEALTH: OnceLock<Mutex<SweepHealth>> = 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