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

lenso_kernel/
diagnostics.rs
