diff --git a/crates/kerykeion/src/collector.rs b/crates/kerykeion/src/collector.rs index 74db40e..d2df6bb 100644 --- a/crates/kerykeion/src/collector.rs +++ b/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, 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); diff --git a/crates/kerykeion/src/collector_tests.rs b/crates/kerykeion/src/collector_tests.rs index 741c911..4eb98c0 100644 --- a/crates/kerykeion/src/collector_tests.rs +++ b/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> = 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" + ); +}