diff --git a/src/controller/core/admission.rs b/src/controller/core/admission.rs index 9cb92b0..fbc60d1 100644 --- a/src/controller/core/admission.rs +++ b/src/controller/core/admission.rs @@ -13,6 +13,7 @@ use crate::{ }, events::{Event, EventKind}, identity::TaskId, + reasons, }; use super::{AdmissionResult, CompletionResult, Controller, RemovalResult, Submission}; @@ -179,7 +180,7 @@ impl Controller { | SlotPhase::Terminating { .. }, AdmissionPolicy::DropIfRunning, ) => { - let reason = format!("dropped: slot busy ({})", slot.status_label()); + let reason = format!("{} ({})", reasons::DROP_IF_RUNNING, slot.status_label()); self.bus.publish( Event::new(EventKind::ControllerRejected) .with_task(Arc::clone(&slot_name)) diff --git a/src/core/registry/listener.rs b/src/core/registry/listener.rs index f17ed80..c28e2b0 100644 --- a/src/core/registry/listener.rs +++ b/src/core/registry/listener.rs @@ -208,7 +208,7 @@ impl Registry { match handle.await { Ok(()) => true, Err(error) => { - self.bus.publish(Event::subscriber_panicked( + self.bus.publish(Event::runtime_failure( "registry", format!("listener join failed: {error}"), )); @@ -236,7 +236,7 @@ impl Registry { /// A panic while processing one command or completion is reported as a diagnostic event instead of killing the registry listener. async fn guarded(&self, who: &'static str, fut: impl Future) { if let Err(msg) = crate::core::panic_guard::guarded(fut).await { - self.bus.publish(Event::subscriber_panicked( + self.bus.publish(Event::runtime_failure( who, format!("listener panic: {msg}"), )); diff --git a/src/core/registry/removal.rs b/src/core/registry/removal.rs index 6d1fdb7..76fe8d8 100644 --- a/src/core/registry/removal.rs +++ b/src/core/registry/removal.rs @@ -17,6 +17,7 @@ use crate::{ core::{actor::ActorExitReason, outcome::TaskOutcome}, events::{Bus, Event, EventKind}, identity::TaskId, + reasons, }; /// Terminal result passed from the single join owner to registry cleanup. @@ -424,7 +425,7 @@ impl Registry { Event::new(EventKind::TaskRemoved) .with_task(Arc::clone(&entry.label)) .with_id(id) - .with_reason("force_terminated_after_grace"), + .with_reason(reasons::FORCE_TERMINATED_AFTER_GRACE), ); } } diff --git a/src/core/runtime/event_relay.rs b/src/core/runtime/event_relay.rs index 55b03a5..94a0cc4 100644 --- a/src/core/runtime/event_relay.rs +++ b/src/core/runtime/event_relay.rs @@ -71,7 +71,7 @@ impl SupervisorCore { ) .await { - set.emit_arc(Arc::new(Event::subscriber_panicked( + set.emit_arc(Arc::new(Event::runtime_failure( "subscriber_listener", format!("listener panic: {panic}"), ))); @@ -123,7 +123,7 @@ impl SupervisorCore { match handle.await { Ok(()) => true, Err(error) => { - self.subs.emit_arc(Arc::new(Event::subscriber_panicked( + self.subs.emit_arc(Arc::new(Event::runtime_failure( "subscriber_listener", format!("listener join failed: {error}"), ))); diff --git a/src/core/runtime/shutdown_workflow.rs b/src/core/runtime/shutdown_workflow.rs index 0e93c76..1db0128 100644 --- a/src/core/runtime/shutdown_workflow.rs +++ b/src/core/runtime/shutdown_workflow.rs @@ -295,7 +295,7 @@ impl SupervisorCore { /// Reports an internal shutdown panic without interrupting later cleanup phases. fn report_shutdown_panic(&self, phase: &str, panic: String) { - self.bus.publish(Event::subscriber_panicked( + self.bus.publish(Event::runtime_failure( "shutdown_owner", format!("{phase} panic: {panic}"), )); diff --git a/src/events/event.rs b/src/events/event.rs index 90bd405..adc99b3 100644 --- a/src/events/event.rs +++ b/src/events/event.rs @@ -83,6 +83,17 @@ pub enum EventKind { /// - `seq`: process-local sequence SubscriberPanicked, + /// An internal runtime component failed. + /// + /// This includes a caught panic or a worker that did not join cleanly. + /// + /// Sets: + /// - `task`: runtime component name + /// - `reason`: diagnostic failure details + /// - `at`: wall-clock timestamp + /// - `seq`: process-local sequence + RuntimeFailure, + /// An event was lost because a subscriber path fell behind or closed. /// /// Sets: @@ -237,6 +248,7 @@ pub enum EventKind { /// Sets: /// - `id`: task run identity /// - `task`: task name + /// - `reason`: [`FORCE_TERMINATED_AFTER_GRACE`](crate::reasons::FORCE_TERMINATED_AFTER_GRACE) when the actor was force-aborted or `None` /// - `at`: wall-clock timestamp /// - `seq`: process-local sequence TaskRemoved, @@ -328,6 +340,7 @@ impl EventKind { pub fn as_label(&self) -> &'static str { match self { EventKind::SubscriberPanicked => "subscriber_panicked", + EventKind::RuntimeFailure => "runtime_failure", EventKind::SubscriberOverflow => "subscriber_overflow", EventKind::ShutdownRequested => "shutdown_requested", EventKind::AllStoppedWithinGrace => "all_stopped_within_grace", @@ -569,13 +582,24 @@ impl Event { .with_reason(info) } + /// Creates an internal runtime failure event. + #[inline] + #[must_use] + pub fn runtime_failure(component: impl Into>, reason: impl Into>) -> Self { + Event::new(EventKind::RuntimeFailure) + .with_task(component) + .with_reason(reason) + } + /// Returns `true` for internal diagnostic events. #[inline] #[must_use] pub fn is_internal_diagnostic(&self) -> bool { matches!( self.kind, - EventKind::SubscriberOverflow | EventKind::SubscriberPanicked + EventKind::SubscriberOverflow + | EventKind::SubscriberPanicked + | EventKind::RuntimeFailure ) } } @@ -631,6 +655,7 @@ mod tests { fn event_kind_labels_are_stable() { let cases = [ (EventKind::SubscriberPanicked, "subscriber_panicked"), + (EventKind::RuntimeFailure, "runtime_failure"), (EventKind::SubscriberOverflow, "subscriber_overflow"), (EventKind::ShutdownRequested, "shutdown_requested"), (EventKind::AllStoppedWithinGrace, "all_stopped_within_grace"), @@ -714,15 +739,19 @@ mod tests { } #[test] - fn is_internal_diagnostic_covers_both_variants() { - for kind in [EventKind::SubscriberOverflow, EventKind::SubscriberPanicked] { + fn is_internal_diagnostic_covers_all_variants() { + for kind in [ + EventKind::SubscriberOverflow, + EventKind::SubscriberPanicked, + EventKind::RuntimeFailure, + ] { assert!(Event::new(kind).is_internal_diagnostic(), "{kind:?}"); } assert!(!Event::new(EventKind::TaskStarting).is_internal_diagnostic()); } #[test] - fn subscriber_factories_set_kind_task_and_reason() { + fn diagnostic_factories_set_kind_task_and_reason() { let overflow = Event::subscriber_overflow("my-sub", "full"); assert_eq!(overflow.kind, EventKind::SubscriberOverflow); assert_eq!( @@ -740,6 +769,14 @@ mod tests { assert_eq!(panicked.kind, EventKind::SubscriberPanicked); assert_eq!(panicked.task.as_deref(), Some("my-sub")); assert_eq!(panicked.reason.as_deref(), Some("boom")); + + let runtime_failure = Event::runtime_failure("registry", "listener join failed"); + assert_eq!(runtime_failure.kind, EventKind::RuntimeFailure); + assert_eq!(runtime_failure.task.as_deref(), Some("registry")); + assert_eq!( + runtime_failure.reason.as_deref(), + Some("listener join failed") + ); } #[test] diff --git a/src/reasons.rs b/src/reasons.rs index af451e2..b7d9dfe 100644 --- a/src/reasons.rs +++ b/src/reasons.rs @@ -21,17 +21,19 @@ //! //! ## Where each value appears //! -//! | Constant | Used by | Meaning | -//! |------------------------------|--------------------------------------------------------|----------------------------------------| -//! | [`POLICY_EXHAUSTED_SUCCESS`] | `ActorExhausted` | Normal one-shot finish. | -//! | [`TASK_RETURNED_CANCELED`] | `ActorExhausted` | The task stopped itself. | -//! | [`MAX_RETRIES_EXCEEDED`] | `ActorExhausted` (prefix) | Retry limit reached. | -//! | [`ALREADY_EXISTS`] | `TaskAddFailed`, `TaskOutcome::Rejected` | A task with this name already exists. | -//! | [`BATCH_REJECTED`] | `TaskAddFailed` | Another item rejected the whole batch. | -//! | [`QUEUE_FULL`] | `ControllerRejected`, `TaskOutcome::Rejected` (prefix) | The slot queue is full. | -//! | [`REMOVED_FROM_QUEUE`] | `ControllerRejected`, `TaskOutcome::Rejected` | A queued task was removed. | -//! | [`SUPERSEDED_BY_REPLACE`] | `ControllerRejected`, `TaskOutcome::Rejected` | A newer `Replace` task took its place. | -//! | [`CONTROLLER_SHUTTING_DOWN`] | `ControllerRejected`, `TaskOutcome::Rejected` | The controller is shutting down. | +//! | Constant | Used by | Meaning | +//! |--------------------------------------|--------------------------------------------------------|----------------------------------------| +//! | [`POLICY_EXHAUSTED_SUCCESS`] | `ActorExhausted` | Normal one-shot finish. | +//! | [`TASK_RETURNED_CANCELED`] | `ActorExhausted` | The task stopped itself. | +//! | [`MAX_RETRIES_EXCEEDED`] | `ActorExhausted` (prefix) | Retry limit reached. | +//! | [`ALREADY_EXISTS`] | `TaskAddFailed`, `TaskOutcome::Rejected` | A task with this name already exists. | +//! | [`BATCH_REJECTED`] | `TaskAddFailed` | Another item rejected the whole batch. | +//! | [`QUEUE_FULL`] | `ControllerRejected`, `TaskOutcome::Rejected` (prefix) | The slot queue is full. | +//! | [`DROP_IF_RUNNING`] | `ControllerRejected`, `TaskOutcome::Rejected` (prefix) | A busy slot rejected new work. | +//! | [`REMOVED_FROM_QUEUE`] | `ControllerRejected`, `TaskOutcome::Rejected` | A queued task was removed. | +//! | [`SUPERSEDED_BY_REPLACE`] | `ControllerRejected`, `TaskOutcome::Rejected` | A newer `Replace` task took its place. | +//! | [`CONTROLLER_SHUTTING_DOWN`] | `ControllerRejected`, `TaskOutcome::Rejected` | The controller is shutting down. | +//! | [`FORCE_TERMINATED_AFTER_GRACE`] | `TaskRemoved` | The actor was force-aborted. | //! //! Controller rejection values are used only with the `controller` feature. //! The constants themselves are always available. @@ -59,6 +61,15 @@ pub const SUPERSEDED_BY_REPLACE: &str = "superseded_by_replace"; /// Rejection reason: the controller is shutting down. pub const CONTROLLER_SHUTTING_DOWN: &str = "controller_shutting_down"; +/// `ControllerRejected` and `TaskOutcome::Rejected` reason **prefix**: `DropIfRunning` rejected a submission because its slot was busy. +/// +/// The full reason is `dropped: slot busy ()`. +/// Match with `starts_with`, not equality. +pub const DROP_IF_RUNNING: &str = "dropped: slot busy"; + +/// `TaskRemoved` reason: the actor did not stop within its grace window and was force-aborted. +pub const FORCE_TERMINATED_AFTER_GRACE: &str = "force_terminated_after_grace"; + /// `ActorExhausted` reason **prefix**: the task stopped after it used all retry attempts. /// /// The full reason is `max_retries_exceeded(/): `. diff --git a/src/subscribers/embedded/log.rs b/src/subscribers/embedded/log.rs index 3627486..ab82091 100644 --- a/src/subscribers/embedded/log.rs +++ b/src/subscribers/embedded/log.rs @@ -141,6 +141,13 @@ impl LogWriter { or(e.reason.as_deref(), "unknown") ); } + EventKind::RuntimeFailure => { + println!( + "{head} component={} reason=\"{}\"", + or(e.task.as_deref(), "none"), + or(e.reason.as_deref(), "unknown") + ); + } // Terminals. EventKind::ActorExhausted => { diff --git a/src/subscribers/embedded/tracing.rs b/src/subscribers/embedded/tracing.rs index 97c0053..883be93 100644 --- a/src/subscribers/embedded/tracing.rs +++ b/src/subscribers/embedded/tracing.rs @@ -45,9 +45,10 @@ pub struct TracingBridge; /// Maps an event to a tracing level. fn level_for(e: &Event) -> Level { match e.kind { - EventKind::TaskFailed | EventKind::ActorDead | EventKind::SubscriberPanicked => { - Level::ERROR - } + EventKind::TaskFailed + | EventKind::ActorDead + | EventKind::SubscriberPanicked + | EventKind::RuntimeFailure => Level::ERROR, EventKind::TimeoutHit | EventKind::GraceExceeded diff --git a/src/subscribers/subscriber_set.rs b/src/subscribers/subscriber_set.rs index 2703f7a..34be364 100644 --- a/src/subscribers/subscriber_set.rs +++ b/src/subscribers/subscriber_set.rs @@ -850,7 +850,11 @@ mod tests { #[tokio::test] async fn panic_on_internal_diagnostic_does_not_republish() { - for diagnostic in [EventKind::SubscriberPanicked, EventKind::SubscriberOverflow] { + for diagnostic in [ + EventKind::SubscriberPanicked, + EventKind::SubscriberOverflow, + EventKind::RuntimeFailure, + ] { let bus = Bus::new(64); let mut rx = bus.subscribe(); let set = SubscriberSet::new(vec![PanicSub::new()], bus.clone()); diff --git a/tests/controller.rs b/tests/controller.rs index b7bbfe7..bd8e331 100644 --- a/tests/controller.rs +++ b/tests/controller.rs @@ -496,10 +496,9 @@ async fn drop_if_running_rejects_busy_submission_without_starting_it() { event.task.as_deref() == Some("s") && event.kind == EventKind::ControllerRejected && event.id == Some(rejected_id) - && event - .reason - .as_deref() - .is_some_and(|reason| reason.contains("dropped: slot busy")) + && event.reason.as_deref().is_some_and(|reason| { + reason.starts_with(taskvisor::reasons::DROP_IF_RUNNING) + }) }) }) .await diff --git a/tests/identity.rs b/tests/identity.rs index aa0bf44..c009bfc 100644 --- a/tests/identity.rs +++ b/tests/identity.rs @@ -392,7 +392,8 @@ async fn individually_removed_stuck_task_is_force_aborted_after_grace() { events.iter().any(|event| { event.id == Some(id) && event.kind == EventKind::TaskRemoved - && event.reason.as_deref() == Some("force_terminated_after_grace") + && event.reason.as_deref() + == Some(taskvisor::reasons::FORCE_TERMINATED_AFTER_GRACE) }) }) .await,