fix(kerykeion): escalate background task failures to a clean shutdown

crates/kerykeion/src/collector.rs

@@ -495,10 +495,10 @@ impl Collector for MeshCollector { // kanon:ignore ARCHITECTURE/trait-impl-coloc
                     }
                 }
                 Some(task_result) = tasks.join_next() => {
-                    match task_result {
-                        Ok(Ok(())) => tracing::debug!("background task completed"),
-                        Ok(Err(e)) => tracing::warn!(error = %e, "background task error"),
-                        Err(e) => tracing::warn!(error = %e, "background task panicked"),
+                    if supervise_task_result(task_result) == TaskOutcome::Shutdown {
+                        tracing::error!("collector shutting down: a background task subsystem was lost");
+                        cancel.cancel();
+                        break;
                     }
                 }
             }
@@ -524,6 +524,44 @@ impl Collector for MeshCollector { // kanon:ignore ARCHITECTURE/trait-impl-coloc
     }
 }
 
+/// What the main receive loop should do after observing a background task's
+/// join result.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum TaskOutcome {
+    /// The task exited cleanly (or is merely being logged); keep running.
+    Continue,
+    /// The task panicked or returned an error; the collector must not keep
+    /// running silently degraded with that subsystem gone.
+    Shutdown,
+}
+
+/// Classify a background task's [`JoinSet::join_next`] result and log it.
+///
+/// WHY(#205): heartbeat, gateway health, discovery, and router-flush used to
+/// be logged-and-ignored on panic or error, leaving the collector running --
+/// and reporting healthy -- with a subsystem permanently gone. Losing
+/// router-flush in particular silently stops all outbound sending while the
+/// collector keeps receiving, so the operator has no signal that the tool
+/// has stopped doing half its job. Every background task is now supervised
+/// uniformly: any panic or error escalates to a clean shutdown rather than
+/// an indefinite silent partial failure.
+fn supervise_task_result(result: Result<Result<(), Error>, tokio::task::JoinError>) -> TaskOutcome {
+    match result {
+        Ok(Ok(())) => {
+            tracing::debug!("background task completed");
+            TaskOutcome::Continue
+        }
+        Ok(Err(e)) => {
+            tracing::error!(error = %e, "background task returned an error");
+            TaskOutcome::Shutdown
+        }
+        Err(e) => {
+            tracing::error!(error = %e, "background task panicked");
+            TaskOutcome::Shutdown
+        }
+    }
+}
+
 /// Bound on how long [`recv_yielding`] holds the connection lock during a
 /// single `recv()` poll before releasing it.
 const RECV_LOCK_YIELD: Duration = Duration::from_millis(250);

crates/kerykeion/src/collector_tests.rs

@@ -724,3 +724,84 @@ async fn router_flush_starts_ack_timeout_after_transmit() {
     token.cancel();
     handle.abort();
 }
+
+// ── akroasis#205: background task failures escalate, not just log ─────────
+
+#[test]
+fn supervise_task_result_continues_on_clean_completion() {
+    assert_eq!(
+        supervise_task_result(Ok(Ok(()))),
+        TaskOutcome::Continue,
+        "a task that returns Ok must not trigger shutdown"
+    );
+}
+
+#[test]
+fn supervise_task_result_shuts_down_on_task_error() {
+    let err = Error::ConnectionLost {
+        detail: "synthetic test error".into(),
+        location: snafu::location!(),
+    };
+    assert_eq!(
+        supervise_task_result(Ok(Err(err))),
+        TaskOutcome::Shutdown,
+        "a task that returns Err must escalate to shutdown rather than only log"
+    );
+}
+
+#[tokio::test]
+async fn supervise_task_result_shuts_down_on_join_error() {
+    // WHY: a real JoinError, not a hand-built stand-in -- JoinError has no
+    // public constructor. Aborting a task and awaiting its handle yields a
+    // genuine one without triggering an actual panic (clippy::panic denies
+    // `panic!()` outside a deliberately-scoped exception, and the
+    // classifier under test treats every `Err(JoinError)` identically
+    // regardless of whether it came from a panic or a cancellation).
+    let handle = tokio::spawn(std::future::pending::<()>());
+    handle.abort();
+    #[expect(clippy::expect_used, reason = "test-only")]
+    let join_err = handle
+        .await
+        .expect_err("an aborted task must yield Err from JoinHandle::await");
+    assert_eq!(
+        supervise_task_result(Err(join_err)),
+        TaskOutcome::Shutdown,
+        "a panicked or aborted task must escalate to shutdown rather than only log"
+    );
+}
+
+#[tokio::test]
+async fn losing_router_flush_shuts_down_the_collector_loop() {
+    // WHY(#205): this is the scenario the issue names as most consequential
+    // -- router-flush dying used to leave the collector receiving forever
+    // while silently never sending again. Drive the actual `select!` arm
+    // (via a JoinSet standing in for `tasks`) and assert the loop's exit
+    // condition, not just the pure classifier above.
+    let mut tasks: JoinSet<Result<(), Error>> = JoinSet::new();
+    tasks.spawn(async {
+        Err(Error::ConnectionLost {
+            detail: "router flush died".into(),
+            location: snafu::location!(),
+        })
+    });
+
+    let cancel = CancellationToken::new();
+    let mut shut_down = false;
+    tokio::select! {
+        Some(task_result) = tasks.join_next() => {
+            if supervise_task_result(task_result) == TaskOutcome::Shutdown {
+                cancel.cancel();
+                shut_down = true;
+            }
+        }
+    }
+
+    assert!(
+        shut_down,
+        "the loop must recognize a lost background task as a shutdown condition"
+    );
+    assert!(
+        cancel.is_cancelled(),
+        "losing a background task must cancel the collector, not just log it"
+    );
+}