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::{PluginLifecyclePhase, 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    /// Plugin 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 Plugin factory was not linked.
108    MissingPluginFactory,
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 Plugin generation reported a failure.
124    PluginFailure,
125    /// A finite Plugin restart budget was exhausted.
126    PluginRestartExhausted,
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::MissingPluginFactory { .. } => Self::MissingPluginFactory,
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::PluginFailure { .. } => Self::PluginFailure,
145            RuntimeFailure::PluginRestartExhausted { .. } => Self::PluginRestartExhausted,
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 Plugin 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 { plugin_count: usize },
196    /// Every selected Plugin generation has activated and App admission opened.
197    AppReady,
198    /// A Plugin lifecycle phase began.
199    LifecycleStarted {
200        instance: String,
201        generation: u64,
202        phase: PluginLifecyclePhase,
203    },
204    /// A Plugin lifecycle phase completed with a sanitized outcome and duration.
205    LifecycleCompleted {
206        instance: String,
207        generation: u64,
208        phase: PluginLifecyclePhase,
209        outcome: DiagnosticOutcome,
210        elapsed: Duration,
211    },
212    /// A typed request or stream operation began.
213    InvocationStarted {
214        requirement_id: Option<String>,
215        request_id: u64,
216        caller_instance: Option<String>,
217        provider_instance: Option<String>,
218        capability: &'static str,
219        operation: Option<&'static str>,
220    },
221    /// A typed request or stream operation completed.
222    InvocationCompleted {
223        requirement_id: Option<String>,
224        request_id: u64,
225        caller_instance: Option<String>,
226        provider_instance: Option<String>,
227        capability: &'static str,
228        operation: Option<&'static str>,
229        outcome: DiagnosticOutcome,
230        elapsed: Duration,
231    },
232    /// Bounded request admission rejected an operation.
233    AdmissionRejected {
234        requirement_id: Option<String>,
235        request_id: u64,
236        caller_instance: Option<String>,
237        provider_instance: Option<String>,
238        capability: &'static str,
239        operation: Option<&'static str>,
240        outcome: DiagnosticAdmission,
241    },
242    /// One Event subscriber received an independent admission outcome.
243    EventAdmission {
244        requirement_id: Option<String>,
245        request_id: u64,
246        publisher_instance: String,
247        subscriber_instance: String,
248        capability: &'static str,
249        operation: Option<&'static str>,
250        outcome: DiagnosticAdmission,
251    },
252    /// A provider generation became unavailable.
253    GenerationUnavailable { instance: String, generation: u64 },
254    /// A replacement provider generation became ready.
255    GenerationReady { instance: String, generation: u64 },
256    /// Supervision scheduled one bounded restart attempt.
257    RestartScheduled {
258        instance: String,
259        attempt: usize,
260        delay: Duration,
261    },
262    /// Supervision exhausted its finite restart budget.
263    RestartExhausted {
264        instance: String,
265        attempts: usize,
266        terminal: bool,
267    },
268    /// A Runtime Failure category was observed without its detail or payload.
269    RuntimeFailure {
270        instance: Option<String>,
271        kind: RuntimeFailureKind,
272    },
273    /// App admission closed and cooperative cancellation began.
274    ShutdownAdmissionClosed,
275    /// App cleanup began with one global timeout.
276    ShutdownCleanupStarted { timeout: Duration },
277    /// App cleanup completed with a sanitized outcome and duration.
278    ShutdownCompleted {
279        outcome: DiagnosticShutdownOutcome,
280        elapsed: Duration,
281    },
282}
283
284/// One sequenced, timestamped Runtime Diagnostic record.
285#[derive(Clone, Debug, Eq, PartialEq)]
286pub struct DiagnosticRecord {
287    /// Monotonic sequence within the supplied diagnostics port.
288    pub sequence: u64,
289    /// Driver-monotonic timestamp at emission.
290    pub timestamp: Duration,
291    /// Kernel subsystem that emitted the record.
292    pub source: DiagnosticSource,
293    /// Sanitized structural metadata.
294    pub event: DiagnosticEvent,
295}
296
297/// Error returned when an observer queue cannot be created.
298#[derive(Clone, Copy, Debug, Eq, PartialEq)]
299pub enum DiagnosticSubscribeError {
300    /// Every observer queue must have at least one slot.
301    ZeroCapacity,
302}
303
304#[derive(Debug, Default)]
305struct RuntimeDiagnosticsState {
306    observers: RefCell<Vec<Weak<DiagnosticObserverState>>>,
307    next_sequence: Cell<u64>,
308    invocation_probe: Option<Rc<dyn RuntimeInvocationProbe>>,
309}
310
311impl Drop for RuntimeDiagnosticsState {
312    fn drop(&mut self) {
313        for observer in self
314            .observers
315            .get_mut()
316            .drain(..)
317            .filter_map(|observer| observer.upgrade())
318        {
319            observer.connected.set(false);
320            observer.wake_receiver();
321        }
322    }
323}
324
325/// An opt-in Runtime Diagnostics port.
326///
327/// The port only stores local, ephemeral, best-effort records. It never calls
328/// observer code and never waits for a consumer. It is therefore unsuitable
329/// for audit, durable Story correctness, persistence, replay, or redelivery.
330#[derive(Clone, Debug)]
331pub struct RuntimeDiagnostics {
332    state: Rc<RuntimeDiagnosticsState>,
333}
334
335impl RuntimeDiagnostics {
336    /// Creates an empty diagnostics port with no observers.
337    pub fn new() -> Self {
338        Self {
339            state: Rc::new(RuntimeDiagnosticsState::default()),
340        }
341    }
342
343    /// Attaches a compact aggregate probe without enabling rich Invocation records.
344    #[doc(hidden)]
345    #[must_use]
346    pub fn with_invocation_probe(mut self, probe: Rc<dyn RuntimeInvocationProbe>) -> Self {
347        Rc::get_mut(&mut self.state)
348            .expect("a newly configured diagnostics port is uniquely owned")
349            .invocation_probe = Some(probe);
350        self
351    }
352
353    /// Adds an independently bounded, source-filtered observer queue.
354    pub fn subscribe(
355        &self,
356        filter: DiagnosticFilter,
357        capacity: usize,
358    ) -> Result<DiagnosticObserver, DiagnosticSubscribeError> {
359        if capacity == 0 {
360            return Err(DiagnosticSubscribeError::ZeroCapacity);
361        }
362        let observer = Rc::new(DiagnosticObserverState {
363            filter,
364            capacity,
365            queue: RefCell::new(VecDeque::with_capacity(capacity)),
366            dropped: Cell::new(0),
367            connected: Cell::new(true),
368            receiver_waker: RefCell::new(None),
369        });
370        let mut observers = self.state.observers.borrow_mut();
371        observers.retain(|observer| observer.upgrade().is_some());
372        observers.push(Rc::downgrade(&observer));
373        Ok(DiagnosticObserver { state: observer })
374    }
375
376    /// Adds an all-source observer queue.
377    pub fn subscribe_all(
378        &self,
379        capacity: usize,
380    ) -> Result<DiagnosticObserver, DiagnosticSubscribeError> {
381        self.subscribe(DiagnosticFilter::all(), capacity)
382    }
383
384    /// Returns the number of observers that are still connected to this port.
385    pub fn observer_count(&self) -> usize {
386        let mut observers = self.state.observers.borrow_mut();
387        observers.retain(|observer| observer.upgrade().is_some());
388        observers.len()
389    }
390
391    pub(crate) fn has_interested_observer(&self, source: DiagnosticSource) -> bool {
392        self.state
393            .observers
394            .borrow()
395            .iter()
396            .filter_map(Weak::upgrade)
397            .any(|observer| observer.filter.includes(source))
398    }
399
400    pub(crate) fn record_invocation(&self, caller_instance: &str, provider_instance: &str) {
401        if let Some(probe) = &self.state.invocation_probe {
402            probe.record(caller_instance, provider_instance);
403        }
404    }
405
406    pub(crate) fn emit<F>(&self, source: DiagnosticSource, timestamp: Duration, build: F)
407    where
408        F: FnOnce(u64) -> DiagnosticEvent,
409    {
410        if !self.has_interested_observer(source) {
411            return;
412        }
413
414        let sequence = self.state.next_sequence.get();
415        self.state.next_sequence.set(sequence.saturating_add(1));
416        let record = DiagnosticRecord {
417            sequence,
418            timestamp,
419            source,
420            event: build(sequence),
421        };
422        self.state.observers.borrow_mut().retain(|observer| {
423            let Some(observer) = observer.upgrade() else {
424                return false;
425            };
426            if observer.filter.includes(source) {
427                observer.enqueue(record.clone());
428            }
429            true
430        });
431    }
432
433    pub(crate) fn emit_runtime_failure(
434        &self,
435        timestamp: Duration,
436        instance: Option<&str>,
437        error: &RuntimeFailure,
438    ) {
439        let kind = RuntimeFailureKind::from(error);
440        self.emit(DiagnosticSource::RuntimeFailure, timestamp, |_| {
441            DiagnosticEvent::RuntimeFailure {
442                instance: instance.map(str::to_owned),
443                kind,
444            }
445        });
446    }
447}
448
449impl Default for RuntimeDiagnostics {
450    fn default() -> Self {
451        Self::new()
452    }
453}
454
455#[derive(Debug)]
456struct DiagnosticObserverState {
457    filter: DiagnosticFilter,
458    capacity: usize,
459    queue: RefCell<VecDeque<DiagnosticRecord>>,
460    dropped: Cell<u64>,
461    connected: Cell<bool>,
462    receiver_waker: RefCell<Option<Waker>>,
463}
464
465impl DiagnosticObserverState {
466    fn enqueue(&self, record: DiagnosticRecord) {
467        let mut queue = self.queue.borrow_mut();
468        if queue.len() >= self.capacity {
469            self.dropped.set(self.dropped.get().saturating_add(1));
470            return;
471        }
472        queue.push_back(record);
473        drop(queue);
474        self.wake_receiver();
475    }
476
477    fn wake_receiver(&self) {
478        if let Some(waker) = self.receiver_waker.borrow_mut().take() {
479            waker.wake();
480        }
481    }
482}
483
484/// The receiving side of one independently bounded diagnostics queue.
485#[derive(Debug)]
486pub struct DiagnosticObserver {
487    state: Rc<DiagnosticObserverState>,
488}
489
490impl DiagnosticObserver {
491    /// Waits asynchronously for the oldest pending record.
492    ///
493    /// The observer side may await; Kernel producers always use non-blocking
494    /// enqueue and never execute observer-owned code.
495    pub async fn recv(&mut self) -> Option<DiagnosticRecord> {
496        poll_fn(|context| {
497            if let Some(record) = self.try_recv() {
498                return Poll::Ready(Some(record));
499            }
500            if !self.state.connected.get() {
501                return Poll::Ready(None);
502            }
503            self.state
504                .receiver_waker
505                .replace(Some(context.waker().clone()));
506            if let Some(record) = self.try_recv() {
507                self.state.receiver_waker.borrow_mut().take();
508                return Poll::Ready(Some(record));
509            }
510            Poll::Pending
511        })
512        .await
513    }
514
515    /// Removes and returns the oldest pending record without waiting.
516    pub fn try_recv(&self) -> Option<DiagnosticRecord> {
517        self.state.queue.borrow_mut().pop_front()
518    }
519
520    /// Alias for [`Self::try_recv`].
521    pub fn try_next(&self) -> Option<DiagnosticRecord> {
522        self.try_recv()
523    }
524
525    /// Returns the number of records dropped because this queue was full.
526    pub fn dropped_count(&self) -> u64 {
527        self.state.dropped.get()
528    }
529
530    /// Returns the number of records currently buffered.
531    pub fn pending_count(&self) -> usize {
532        self.state.queue.borrow().len()
533    }
534
535    /// Returns the fixed queue capacity.
536    pub fn capacity(&self) -> usize {
537        self.state.capacity
538    }
539
540    /// Returns this observer's source filter.
541    pub fn filter(&self) -> DiagnosticFilter {
542        self.state.filter
543    }
544}
545
546pub(crate) fn diagnostic_operation(
547    operations: &'static [&'static str],
548    operation: &str,
549) -> Option<&'static str> {
550    operations
551        .iter()
552        .copied()
553        .find(|candidate| *candidate == operation)
554}
555
556#[cfg(test)]
557mod tests {
558    use super::{
559        DiagnosticEvent, DiagnosticFilter, DiagnosticSource, RuntimeDiagnostics,
560        RuntimeInvocationProbe,
561    };
562    use std::{cell::Cell, rc::Rc, time::Duration};
563
564    #[derive(Debug)]
565    struct CountProbe(Rc<Cell<u64>>);
566
567    impl RuntimeInvocationProbe for CountProbe {
568        fn record(&self, _caller_instance: &str, _provider_instance: &str) {
569            self.0.set(self.0.get() + 1);
570        }
571    }
572
573    #[test]
574    fn does_not_build_a_record_without_an_interested_observer() {
575        let diagnostics = RuntimeDiagnostics::new();
576        let built = std::cell::Cell::new(false);
577
578        diagnostics.emit(DiagnosticSource::Lifecycle, Duration::ZERO, |_| {
579            built.set(true);
580            DiagnosticEvent::AppReady
581        });
582
583        assert!(!built.get());
584    }
585
586    #[test]
587    fn filters_sources_before_building_a_record() {
588        let diagnostics = RuntimeDiagnostics::new();
589        let observer = diagnostics
590            .subscribe(DiagnosticFilter::only(DiagnosticSource::Invocation), 1)
591            .expect("observer capacity is positive");
592        let built = std::cell::Cell::new(false);
593
594        diagnostics.emit(DiagnosticSource::Lifecycle, Duration::ZERO, |_| {
595            built.set(true);
596            DiagnosticEvent::AppReady
597        });
598
599        assert!(!built.get());
600        assert!(observer.try_recv().is_none());
601    }
602
603    #[test]
604    fn compact_probe_does_not_enable_rich_invocation_records() {
605        let count = Rc::new(Cell::new(0));
606        let diagnostics =
607            RuntimeDiagnostics::new().with_invocation_probe(Rc::new(CountProbe(Rc::clone(&count))));
608
609        diagnostics.record_invocation("caller", "provider");
610
611        assert_eq!(count.get(), 1);
612        assert!(!diagnostics.has_interested_observer(DiagnosticSource::Invocation));
613    }
614}
lenso_kernel/diagnostics.rs

lenso_kernel/
diagnostics.rs
