lenso_kernel/
diagnostics.rs

1use std::{
2    cell::{Cell, RefCell},
3    collections::VecDeque,
4    future::poll_fn,
5    rc::{Rc, Weak},
6    task::{Poll, Waker},
7    time::Duration,
8};
9
10use super::{ModuleLifecyclePhase, RuntimeFailure};
11
12/// Allocation-free request probe used by host runtimes for aggregate placement metrics.
13///
14/// This hook receives borrowed, Plan-owned identities and must return immediately. Rich,
15/// buffered diagnostics remain available through [`RuntimeDiagnostics::subscribe`].
16#[doc(hidden)]
17pub trait RuntimeInvocationProbe: std::fmt::Debug {
18    /// Records one request edge observed by the local runtime.
19    fn record(&self, caller_instance: &str, provider_instance: &str);
20}
21
22/// The Kernel subsystem that produced one Runtime Diagnostic.
23#[derive(Clone, Copy, Debug, Eq, PartialEq)]
24#[repr(u8)]
25pub enum DiagnosticSource {
26    /// Module generation preparation, activation, readiness, or deactivation.
27    Lifecycle = 0,
28    /// A Request or Stream-open operation entered or left the Kernel.
29    Invocation = 1,
30    /// Bounded work admission or Event delivery was accepted or rejected.
31    Admission = 2,
32    /// Provider generation replacement and restart-budget decisions.
33    Supervision = 3,
34    /// App shutdown admission and cleanup.
35    Shutdown = 4,
36    /// A sanitized Runtime Failure fact.
37    RuntimeFailure = 5,
38}
39
40impl DiagnosticSource {
41    const COUNT: u8 = 6;
42
43    const fn bit(self) -> u8 {
44        1 << (self as u8)
45    }
46}
47
48/// A compact source allowlist for one observer.
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
50pub struct DiagnosticFilter {
51    mask: u8,
52}
53
54impl DiagnosticFilter {
55    /// Matches no diagnostic source.
56    pub const fn none() -> Self {
57        Self { mask: 0 }
58    }
59
60    /// Matches every Kernel diagnostic source.
61    pub const fn all() -> Self {
62        Self {
63            mask: (1 << DiagnosticSource::COUNT) - 1,
64        }
65    }
66
67    /// Matches exactly one diagnostic source.
68    pub const fn only(source: DiagnosticSource) -> Self {
69        Self { mask: source.bit() }
70    }
71
72    /// Returns a filter that also matches `source`.
73    #[must_use]
74    pub const fn with_source(self, source: DiagnosticSource) -> Self {
75        Self {
76            mask: self.mask | source.bit(),
77        }
78    }
79
80    /// Returns whether this filter accepts `source`.
81    pub const fn includes(self, source: DiagnosticSource) -> bool {
82        self.mask & source.bit() != 0
83    }
84}
85
86impl Default for DiagnosticFilter {
87    fn default() -> Self {
88        Self::all()
89    }
90}
91
92/// A sanitized category of Runtime Failure.
93///
94/// Details, payloads, configuration, and opaque values are intentionally not
95/// represented. Observers can use the category with structural fields from a
96/// [`DiagnosticEvent`] without receiving business data.
97#[derive(Clone, Copy, Debug, Eq, PartialEq)]
98pub enum RuntimeFailureKind {
99    /// No current provider generation is available.
100    Unavailable,
101    /// The requested Operation is not in the resolved Descriptor.
102    UnknownOperation,
103    /// A singular handle was used for several providers.
104    AmbiguousBinding,
105    /// The generated contract and endpoint disagreed.
106    ProtocolViolation,
107    /// A selected Module factory was not linked.
108    MissingModuleFactory,
109    /// The selected Execution Adapter is unavailable.
110    UnavailableExecutionClass,
111    /// The resolved Plan or prepared endpoint set is invalid.
112    InvalidResolvedPlan,
113    /// New work was rejected because App admission is closed.
114    AdmissionClosed,
115    /// A bounded admission queue was full.
116    ResourceExhausted,
117    /// A monotonic invocation deadline expired.
118    DeadlineExceeded,
119    /// Invocation cancellation won the race.
120    Cancelled,
121    /// The Driver or Adapter reported an internal failure.
122    Internal,
123    /// A Module generation reported a failure.
124    ModuleFailure,
125    /// A finite Module restart budget was exhausted.
126    ModuleRestartExhausted,
127}
128
129impl From<&RuntimeFailure> for RuntimeFailureKind {
130    fn from(error: &RuntimeFailure) -> Self {
131        match error {
132            RuntimeFailure::Unavailable { .. } => Self::Unavailable,
133            RuntimeFailure::UnknownOperation { .. } => Self::UnknownOperation,
134            RuntimeFailure::AmbiguousBinding { .. } => Self::AmbiguousBinding,
135            RuntimeFailure::ProtocolViolation { .. } => Self::ProtocolViolation,
136            RuntimeFailure::MissingModuleFactory { .. } => Self::MissingModuleFactory,
137            RuntimeFailure::UnavailableExecutionClass { .. } => Self::UnavailableExecutionClass,
138            RuntimeFailure::InvalidResolvedPlan { .. } => Self::InvalidResolvedPlan,
139            RuntimeFailure::AdmissionClosed => Self::AdmissionClosed,
140            RuntimeFailure::ResourceExhausted { .. } => Self::ResourceExhausted,
141            RuntimeFailure::DeadlineExceeded { .. } => Self::DeadlineExceeded,
142            RuntimeFailure::Cancelled { .. } => Self::Cancelled,
143            RuntimeFailure::Internal { .. } => Self::Internal,
144            RuntimeFailure::ModuleFailure { .. } => Self::ModuleFailure,
145            RuntimeFailure::ModuleRestartExhausted { .. } => Self::ModuleRestartExhausted,
146        }
147    }
148}
149
150/// An outcome that is safe to expose without including a Domain Error body.
151#[derive(Clone, Copy, Debug, Eq, PartialEq)]
152pub enum DiagnosticOutcome {
153    /// The operation or lifecycle phase completed successfully.
154    Succeeded,
155    /// The Capability returned a Domain Error; its body is deliberately absent.
156    DomainError,
157    /// The Kernel returned a sanitized Runtime Failure category.
158    RuntimeFailure(RuntimeFailureKind),
159}
160
161/// A bounded admission outcome safe to expose to a diagnostic observer.
162#[derive(Clone, Copy, Debug, Eq, PartialEq)]
163pub enum DiagnosticAdmission {
164    /// The value or operation entered the selected bounded queue.
165    Accepted,
166    /// The selected provider or subscriber generation is unavailable.
167    Unavailable,
168    /// The selected bounded queue is full.
169    Exhausted,
170    /// App admission was already closed.
171    Closed,
172}
173
174/// A sanitized App shutdown outcome.
175#[derive(Clone, Copy, Debug, Eq, PartialEq)]
176pub enum DiagnosticShutdownOutcome {
177    /// All managed work and resources were released.
178    Clean,
179    /// Cleanup reported a Runtime Failure.
180    RuntimeFailure,
181    /// The global cleanup deadline expired.
182    Timeout,
183}
184
185/// Structural, lossy metadata emitted by the Kernel.
186///
187/// This enum intentionally has no payload, configuration, secret, opaque
188/// extension, `ActorAssertion`, or Domain Error fields. Delivery of these
189/// records is not itself observed, so exporting a record cannot recurse into
190/// the diagnostic feed. Caller identities are present only when they resolve
191/// to a Module Instance in the immutable App Plan.
192#[derive(Clone, Debug, Eq, PartialEq)]
193pub enum DiagnosticEvent {
194    /// The Kernel created a running App runtime.
195    AppStarted { module_count: usize },
196    /// Every selected Module generation has activated and App admission opened.
197    AppReady,
198    /// A Module lifecycle phase began.
199    LifecycleStarted {
200        instance: String,
201        generation: u64,
202        phase: ModuleLifecyclePhase,
203    },
204    /// A Module lifecycle phase completed with a sanitized outcome and duration.
205    LifecycleCompleted {
206        instance: String,
207        generation: u64,
208        phase: ModuleLifecyclePhase,
209        outcome: DiagnosticOutcome,
210        elapsed: Duration,
211    },
212    /// A typed request or stream operation began.
213    InvocationStarted {
214        request_id: u64,
215        caller_instance: Option<String>,
216        provider_instance: Option<String>,
217        capability: &'static str,
218        operation: Option<&'static str>,
219    },
220    /// A typed request or stream operation completed.
221    InvocationCompleted {
222        request_id: u64,
223        caller_instance: Option<String>,
224        provider_instance: Option<String>,
225        capability: &'static str,
226        operation: Option<&'static str>,
227        outcome: DiagnosticOutcome,
228        elapsed: Duration,
229    },
230    /// Bounded request admission rejected an operation.
231    AdmissionRejected {
232        request_id: u64,
233        caller_instance: Option<String>,
234        provider_instance: Option<String>,
235        capability: &'static str,
236        operation: Option<&'static str>,
237        outcome: DiagnosticAdmission,
238    },
239    /// One Event subscriber received an independent admission outcome.
240    EventAdmission {
241        request_id: u64,
242        publisher_instance: String,
243        subscriber_instance: String,
244        capability: &'static str,
245        operation: Option<&'static str>,
246        outcome: DiagnosticAdmission,
247    },
248    /// A provider generation became unavailable.
249    GenerationUnavailable { instance: String, generation: u64 },
250    /// A replacement provider generation became ready.
251    GenerationReady { instance: String, generation: u64 },
252    /// Supervision scheduled one bounded restart attempt.
253    RestartScheduled {
254        instance: String,
255        attempt: usize,
256        delay: Duration,
257    },
258    /// Supervision exhausted its finite restart budget.
259    RestartExhausted {
260        instance: String,
261        attempts: usize,
262        terminal: bool,
263    },
264    /// A Runtime Failure category was observed without its detail or payload.
265    RuntimeFailure {
266        instance: Option<String>,
267        kind: RuntimeFailureKind,
268    },
269    /// App admission closed and cooperative cancellation began.
270    ShutdownAdmissionClosed,
271    /// App cleanup began with one global timeout.
272    ShutdownCleanupStarted { timeout: Duration },
273    /// App cleanup completed with a sanitized outcome and duration.
274    ShutdownCompleted {
275        outcome: DiagnosticShutdownOutcome,
276        elapsed: Duration,
277    },
278}
279
280/// One sequenced, timestamped Runtime Diagnostic record.
281#[derive(Clone, Debug, Eq, PartialEq)]
282pub struct DiagnosticRecord {
283    /// Monotonic sequence within the supplied diagnostics port.
284    pub sequence: u64,
285    /// Driver-monotonic timestamp at emission.
286    pub timestamp: Duration,
287    /// Kernel subsystem that emitted the record.
288    pub source: DiagnosticSource,
289    /// Sanitized structural metadata.
290    pub event: DiagnosticEvent,
291}
292
293/// Error returned when an observer queue cannot be created.
294#[derive(Clone, Copy, Debug, Eq, PartialEq)]
295pub enum DiagnosticSubscribeError {
296    /// Every observer queue must have at least one slot.
297    ZeroCapacity,
298}
299
300#[derive(Debug, Default)]
301struct RuntimeDiagnosticsState {
302    observers: RefCell<Vec<Weak<DiagnosticObserverState>>>,
303    next_sequence: Cell<u64>,
304    invocation_probe: Option<Rc<dyn RuntimeInvocationProbe>>,
305}
306
307impl Drop for RuntimeDiagnosticsState {
308    fn drop(&mut self) {
309        for observer in self
310            .observers
311            .get_mut()
312            .drain(..)
313            .filter_map(|observer| observer.upgrade())
314        {
315            observer.connected.set(false);
316            observer.wake_receiver();
317        }
318    }
319}
320
321/// An opt-in Runtime Diagnostics port.
322///
323/// The port only stores local, ephemeral, best-effort records. It never calls
324/// observer code and never waits for a consumer. It is therefore unsuitable
325/// for audit, durable Story correctness, persistence, replay, or redelivery.
326#[derive(Clone, Debug)]
327pub struct RuntimeDiagnostics {
328    state: Rc<RuntimeDiagnosticsState>,
329}
330
331impl RuntimeDiagnostics {
332    /// Creates an empty diagnostics port with no observers.
333    pub fn new() -> Self {
334        Self {
335            state: Rc::new(RuntimeDiagnosticsState::default()),
336        }
337    }
338
339    /// Attaches a compact aggregate probe without enabling rich Invocation records.
340    #[doc(hidden)]
341    #[must_use]
342    pub fn with_invocation_probe(mut self, probe: Rc<dyn RuntimeInvocationProbe>) -> Self {
343        Rc::get_mut(&mut self.state)
344            .expect("a newly configured diagnostics port is uniquely owned")
345            .invocation_probe = Some(probe);
346        self
347    }
348
349    /// Adds an independently bounded, source-filtered observer queue.
350    pub fn subscribe(
351        &self,
352        filter: DiagnosticFilter,
353        capacity: usize,
354    ) -> Result<DiagnosticObserver, DiagnosticSubscribeError> {
355        if capacity == 0 {
356            return Err(DiagnosticSubscribeError::ZeroCapacity);
357        }
358        let observer = Rc::new(DiagnosticObserverState {
359            filter,
360            capacity,
361            queue: RefCell::new(VecDeque::with_capacity(capacity)),
362            dropped: Cell::new(0),
363            connected: Cell::new(true),
364            receiver_waker: RefCell::new(None),
365        });
366        let mut observers = self.state.observers.borrow_mut();
367        observers.retain(|observer| observer.upgrade().is_some());
368        observers.push(Rc::downgrade(&observer));
369        Ok(DiagnosticObserver { state: observer })
370    }
371
372    /// Adds an all-source observer queue.
373    pub fn subscribe_all(
374        &self,
375        capacity: usize,
376    ) -> Result<DiagnosticObserver, DiagnosticSubscribeError> {
377        self.subscribe(DiagnosticFilter::all(), capacity)
378    }
379
380    /// Returns the number of observers that are still connected to this port.
381    pub fn observer_count(&self) -> usize {
382        let mut observers = self.state.observers.borrow_mut();
383        observers.retain(|observer| observer.upgrade().is_some());
384        observers.len()
385    }
386
387    pub(crate) fn has_interested_observer(&self, source: DiagnosticSource) -> bool {
388        self.state
389            .observers
390            .borrow()
391            .iter()
392            .filter_map(Weak::upgrade)
393            .any(|observer| observer.filter.includes(source))
394    }
395
396    pub(crate) fn record_invocation(&self, caller_instance: &str, provider_instance: &str) {
397        if let Some(probe) = &self.state.invocation_probe {
398            probe.record(caller_instance, provider_instance);
399        }
400    }
401
402    pub(crate) fn emit<F>(&self, source: DiagnosticSource, timestamp: Duration, build: F)
403    where
404        F: FnOnce(u64) -> DiagnosticEvent,
405    {
406        if !self.has_interested_observer(source) {
407            return;
408        }
409
410        let sequence = self.state.next_sequence.get();
411        self.state.next_sequence.set(sequence.saturating_add(1));
412        let record = DiagnosticRecord {
413            sequence,
414            timestamp,
415            source,
416            event: build(sequence),
417        };
418        self.state.observers.borrow_mut().retain(|observer| {
419            let Some(observer) = observer.upgrade() else {
420                return false;
421            };
422            if observer.filter.includes(source) {
423                observer.enqueue(record.clone());
424            }
425            true
426        });
427    }
428
429    pub(crate) fn emit_runtime_failure(
430        &self,
431        timestamp: Duration,
432        instance: Option<&str>,
433        error: &RuntimeFailure,
434    ) {
435        let kind = RuntimeFailureKind::from(error);
436        self.emit(DiagnosticSource::RuntimeFailure, timestamp, |_| {
437            DiagnosticEvent::RuntimeFailure {
438                instance: instance.map(str::to_owned),
439                kind,
440            }
441        });
442    }
443}
444
445impl Default for RuntimeDiagnostics {
446    fn default() -> Self {
447        Self::new()
448    }
449}
450
451#[derive(Debug)]
452struct DiagnosticObserverState {
453    filter: DiagnosticFilter,
454    capacity: usize,
455    queue: RefCell<VecDeque<DiagnosticRecord>>,
456    dropped: Cell<u64>,
457    connected: Cell<bool>,
458    receiver_waker: RefCell<Option<Waker>>,
459}
460
461impl DiagnosticObserverState {
462    fn enqueue(&self, record: DiagnosticRecord) {
463        let mut queue = self.queue.borrow_mut();
464        if queue.len() >= self.capacity {
465            self.dropped.set(self.dropped.get().saturating_add(1));
466            return;
467        }
468        queue.push_back(record);
469        drop(queue);
470        self.wake_receiver();
471    }
472
473    fn wake_receiver(&self) {
474        if let Some(waker) = self.receiver_waker.borrow_mut().take() {
475            waker.wake();
476        }
477    }
478}
479
480/// The receiving side of one independently bounded diagnostics queue.
481#[derive(Debug)]
482pub struct DiagnosticObserver {
483    state: Rc<DiagnosticObserverState>,
484}
485
486impl DiagnosticObserver {
487    /// Waits asynchronously for the oldest pending record.
488    ///
489    /// The observer side may await; Kernel producers always use non-blocking
490    /// enqueue and never execute observer-owned code.
491    pub async fn recv(&mut self) -> Option<DiagnosticRecord> {
492        poll_fn(|context| {
493            if let Some(record) = self.try_recv() {
494                return Poll::Ready(Some(record));
495            }
496            if !self.state.connected.get() {
497                return Poll::Ready(None);
498            }
499            self.state
500                .receiver_waker
501                .replace(Some(context.waker().clone()));
502            if let Some(record) = self.try_recv() {
503                self.state.receiver_waker.borrow_mut().take();
504                return Poll::Ready(Some(record));
505            }
506            Poll::Pending
507        })
508        .await
509    }
510
511    /// Removes and returns the oldest pending record without waiting.
512    pub fn try_recv(&self) -> Option<DiagnosticRecord> {
513        self.state.queue.borrow_mut().pop_front()
514    }
515
516    /// Alias for [`Self::try_recv`].
517    pub fn try_next(&self) -> Option<DiagnosticRecord> {
518        self.try_recv()
519    }
520
521    /// Returns the number of records dropped because this queue was full.
522    pub fn dropped_count(&self) -> u64 {
523        self.state.dropped.get()
524    }
525
526    /// Returns the number of records currently buffered.
527    pub fn pending_count(&self) -> usize {
528        self.state.queue.borrow().len()
529    }
530
531    /// Returns the fixed queue capacity.
532    pub fn capacity(&self) -> usize {
533        self.state.capacity
534    }
535
536    /// Returns this observer's source filter.
537    pub fn filter(&self) -> DiagnosticFilter {
538        self.state.filter
539    }
540}
541
542pub(crate) fn diagnostic_operation(
543    operations: &'static [&'static str],
544    operation: &str,
545) -> Option<&'static str> {
546    operations
547        .iter()
548        .copied()
549        .find(|candidate| *candidate == operation)
550}
551
552#[cfg(test)]
553mod tests {
554    use super::{
555        DiagnosticEvent, DiagnosticFilter, DiagnosticSource, RuntimeDiagnostics,
556        RuntimeInvocationProbe,
557    };
558    use std::{cell::Cell, rc::Rc, time::Duration};
559
560    #[derive(Debug)]
561    struct CountProbe(Rc<Cell<u64>>);
562
563    impl RuntimeInvocationProbe for CountProbe {
564        fn record(&self, _caller_instance: &str, _provider_instance: &str) {
565            self.0.set(self.0.get() + 1);
566        }
567    }
568
569    #[test]
570    fn does_not_build_a_record_without_an_interested_observer() {
571        let diagnostics = RuntimeDiagnostics::new();
572        let built = std::cell::Cell::new(false);
573
574        diagnostics.emit(DiagnosticSource::Lifecycle, Duration::ZERO, |_| {
575            built.set(true);
576            DiagnosticEvent::AppReady
577        });
578
579        assert!(!built.get());
580    }
581
582    #[test]
583    fn filters_sources_before_building_a_record() {
584        let diagnostics = RuntimeDiagnostics::new();
585        let observer = diagnostics
586            .subscribe(DiagnosticFilter::only(DiagnosticSource::Invocation), 1)
587            .expect("observer capacity is positive");
588        let built = std::cell::Cell::new(false);
589
590        diagnostics.emit(DiagnosticSource::Lifecycle, Duration::ZERO, |_| {
591            built.set(true);
592            DiagnosticEvent::AppReady
593        });
594
595        assert!(!built.get());
596        assert!(observer.try_recv().is_none());
597    }
598
599    #[test]
600    fn compact_probe_does_not_enable_rich_invocation_records() {
601        let count = Rc::new(Cell::new(0));
602        let diagnostics =
603            RuntimeDiagnostics::new().with_invocation_probe(Rc::new(CountProbe(Rc::clone(&count))));
604
605        diagnostics.record_invocation("caller", "provider");
606
607        assert_eq!(count.get(), 1);
608        assert!(!diagnostics.has_interested_observer(DiagnosticSource::Invocation));
609    }
610}
lenso_kernel/diagnostics.rs

lenso_kernel/
diagnostics.rs
