appcore_ops/
observation.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: observation.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/21 23:21:21 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Bounded, tool-independent runtime observation events.
12
13use appcore_core::redact_text_with_limit;
14use parking_lot::{Mutex, RwLock};
15use serde::{Deserialize, Serialize};
16use std::collections::{BTreeMap, VecDeque};
17use std::sync::Arc;
18
19/// Maximum retained attributes per observation.
20pub const MAX_OBSERVATION_ATTRIBUTES: usize = 32;
21/// Maximum UTF-8 bytes retained in an observation name.
22pub const MAX_OBSERVATION_NAME_BYTES: usize = 128;
23/// Maximum UTF-8 bytes retained in an observation attribute key.
24pub const MAX_OBSERVATION_KEY_BYTES: usize = 64;
25/// Maximum UTF-8 bytes retained in an observation attribute value.
26pub const MAX_OBSERVATION_VALUE_BYTES: usize = 1_024;
27/// Maximum UTF-8 bytes retained in a trace identifier.
28pub const MAX_OBSERVATION_TRACE_BYTES: usize = 256;
29/// Maximum events retained by one process-local observation sink.
30pub const MAX_IN_MEMORY_OBSERVATION_ITEMS: usize = 65_536;
31/// Absolute aggregate retained-byte ceiling for one process-local sink.
32pub const MAX_IN_MEMORY_OBSERVATION_BYTES: usize = 16 * 1024 * 1024;
33/// Maximum operational drains attached to one process-local observation sink.
34pub const MAX_OBSERVATION_DRAINS: usize = 32;
35const OBSERVATION_FIXED_BYTES: usize = std::mem::size_of::<ObservationEvent>()
36    + std::mem::size_of::<Arc<ObservationEvent>>()
37    + std::mem::size_of::<usize>() * 2;
38const ATTRIBUTE_FIXED_BYTES: usize =
39    std::mem::size_of::<(String, String)>() + std::mem::size_of::<usize>() * 4;
40
41/// Runtime subsystem that produced an observation.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum ObservationKind {
45    /// Process lifecycle.
46    Lifecycle,
47    /// Runtime or deployment configuration.
48    Configuration,
49    /// Health evaluation.
50    Health,
51    /// Authentication, authorization or secret boundary.
52    Security,
53    /// Storage operation.
54    Storage,
55    /// Control-plane operation.
56    ControlPlane,
57    /// Direct peer RPC operation.
58    PeerRpc,
59    /// Scheduler operation.
60    Scheduler,
61    /// Synchronization operation.
62    Sync,
63    /// Audit operation.
64    Audit,
65    /// Diagnostic operation.
66    Diagnostic,
67}
68
69/// Severity of one runtime observation.
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
71#[serde(rename_all = "snake_case")]
72pub enum ObservationSeverity {
73    /// Developer diagnostic fact.
74    Debug,
75    /// Normal operational fact.
76    Info,
77    /// Recoverable or degraded condition.
78    Warning,
79    /// Failed operation.
80    Error,
81}
82
83/// Generic runtime fact suitable for logs, diagnostics, metrics and audit sinks.
84#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85pub struct ObservationEvent {
86    /// Runtime subsystem category.
87    pub kind: ObservationKind,
88    /// Observation severity.
89    pub severity: ObservationSeverity,
90    /// Stable observation name.
91    pub name: String,
92    /// Timestamp in Unix milliseconds.
93    pub timestamp_ms: u64,
94    /// Optional trace identity.
95    pub trace_id: Option<String>,
96    /// Bounded non-sensitive dimensions.
97    pub attributes: BTreeMap<String, String>,
98}
99
100impl ObservationEvent {
101    /// Creates an event without attributes.
102    pub fn new(
103        kind: ObservationKind,
104        severity: ObservationSeverity,
105        name: impl Into<String>,
106        timestamp_ms: u64,
107    ) -> Self {
108        Self {
109            kind,
110            severity,
111            name: name.into(),
112            timestamp_ms,
113            trace_id: None,
114            attributes: BTreeMap::new(),
115        }
116    }
117
118    /// Attaches a trace identity.
119    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
120        self.trace_id = Some(trace_id.into());
121        self
122    }
123
124    /// Adds one attribute. Sensitive keys and values are redacted immediately.
125    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
126        let key = key.into();
127        let sensitive = is_sensitive_key(&key);
128        let key = redact_text_with_limit(&key, MAX_OBSERVATION_KEY_BYTES);
129        if self.attributes.len() >= MAX_OBSERVATION_ATTRIBUTES
130            && !self.attributes.contains_key(&key)
131        {
132            return self;
133        }
134        let value = if sensitive {
135            "[REDACTED]".to_string()
136        } else {
137            redact_text_with_limit(&value.into(), MAX_OBSERVATION_VALUE_BYTES)
138        };
139        self.attributes.insert(key, value);
140        self
141    }
142
143    pub(crate) fn redacted(mut self) -> Self {
144        self.name = redact_text_with_limit(&self.name, MAX_OBSERVATION_NAME_BYTES);
145        self.name.shrink_to_fit();
146        self.trace_id = self.trace_id.map(|value| {
147            let mut value = redact_text_with_limit(&value, MAX_OBSERVATION_TRACE_BYTES);
148            value.shrink_to_fit();
149            value
150        });
151        let mut attributes = BTreeMap::new();
152        for (key, value) in std::mem::take(&mut self.attributes)
153            .into_iter()
154            .take(MAX_OBSERVATION_ATTRIBUTES)
155        {
156            let sensitive = is_sensitive_key(&key);
157            let mut key = redact_text_with_limit(&key, MAX_OBSERVATION_KEY_BYTES);
158            let mut value = if sensitive {
159                "[REDACTED]".to_string()
160            } else {
161                redact_text_with_limit(&value, MAX_OBSERVATION_VALUE_BYTES)
162            };
163            key.shrink_to_fit();
164            value.shrink_to_fit();
165            attributes.insert(key, value);
166        }
167        self.attributes = attributes;
168        self
169    }
170}
171
172/// Redacted, bounded observation payload shared across multiple sinks.
173///
174/// Construction reapplies the same validation as [`ObservationSink::emit`],
175/// so a sink may retain the immutable payload without copying its owned fields.
176#[derive(Debug, Clone)]
177pub struct SharedObservationEvent {
178    event: Arc<ObservationEvent>,
179}
180
181impl SharedObservationEvent {
182    /// Redacts, bounds and shares one observation event.
183    pub fn new(event: ObservationEvent) -> Self {
184        Self {
185            event: Arc::new(event.redacted()),
186        }
187    }
188
189    /// Borrows the validated observation payload.
190    pub fn as_event(&self) -> &ObservationEvent {
191        &self.event
192    }
193
194    fn clone_event_arc(&self) -> Arc<ObservationEvent> {
195        Arc::clone(&self.event)
196    }
197}
198
199/// Destination for generic observation events.
200pub trait ObservationSink: Send + Sync {
201    /// Emits one event without blocking on external tooling.
202    fn emit(&self, event: ObservationEvent);
203
204    /// Emits an already validated shared event.
205    ///
206    /// Existing sinks remain compatible through this owned fallback. Sinks
207    /// that retain or only inspect events should override it to avoid copying
208    /// the payload.
209    fn emit_shared(&self, event: &SharedObservationEvent) {
210        self.emit(event.as_event().clone());
211    }
212}
213
214/// Immutable shared view of retained observations, ordered oldest to newest.
215#[derive(Debug, Clone, Default)]
216pub struct ObservationSnapshot {
217    events: Arc<VecDeque<Arc<ObservationEvent>>>,
218}
219
220impl ObservationSnapshot {
221    /// Returns the number of observations in the snapshot.
222    pub fn len(&self) -> usize {
223        self.events.len()
224    }
225
226    /// Reports whether the snapshot contains no observations.
227    pub fn is_empty(&self) -> bool {
228        self.events.is_empty()
229    }
230
231    /// Iterates from the oldest observation to the newest without cloning it.
232    pub fn iter(&self) -> impl DoubleEndedIterator<Item = &ObservationEvent> {
233        self.events.iter().map(AsRef::as_ref)
234    }
235
236    /// Iterates over at most the newest `limit` observations in chronological order.
237    pub fn recent(&self, limit: usize) -> impl Iterator<Item = &ObservationEvent> {
238        self.events
239            .iter()
240            .skip(self.events.len().saturating_sub(limit))
241            .map(AsRef::as_ref)
242    }
243}
244
245/// Point-in-time retention pressure for a process-local observation sink.
246#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
247pub struct InMemoryObservationPressure {
248    /// Current retained event count.
249    pub entries: usize,
250    /// Configured retained-event ceiling.
251    pub max_entries: usize,
252    /// Estimated bytes retained by events and their owned fields.
253    pub used_bytes: usize,
254    /// Highest estimated retained byte count observed since creation.
255    pub peak_bytes: usize,
256    /// Configured aggregate retained-byte ceiling.
257    pub max_bytes: usize,
258    /// Oldest events discarded to enforce count or byte limits.
259    pub evictions: u64,
260    /// Events not retained because one event exceeded the byte ceiling.
261    pub oversized_rejections: u64,
262    /// Drains rejected because the attachment ceiling was full.
263    pub drain_rejections: u64,
264}
265
266#[derive(Debug)]
267struct ObservationState {
268    events: Arc<VecDeque<Arc<ObservationEvent>>>,
269    pressure: InMemoryObservationPressure,
270}
271
272type ObservationDrain = Arc<dyn ObservationSink>;
273type ObservationDrainGeneration = Arc<Vec<ObservationDrain>>;
274
275/// Bounded in-memory sink used by diagnostics and embedded runtimes.
276///
277/// Drain configuration is copy-on-write. Emission borrows one immutable drain
278/// generation and releases its configuration lock before invoking callbacks.
279#[derive(Clone)]
280pub struct InMemoryObservationSink {
281    capacity: usize,
282    state: Arc<Mutex<ObservationState>>,
283    drains: Arc<RwLock<ObservationDrainGeneration>>,
284}
285
286impl std::fmt::Debug for InMemoryObservationSink {
287    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
288        formatter
289            .debug_struct("InMemoryObservationSink")
290            .field("capacity", &self.capacity)
291            .field("event_count", &self.len())
292            .field("pressure", &self.pressure())
293            .field("drain_count", &self.drains.read().len())
294            .finish()
295    }
296}
297
298impl InMemoryObservationSink {
299    /// Creates a sink that retains the newest `capacity` events.
300    pub fn new(capacity: usize) -> Self {
301        let capacity = capacity.clamp(1, MAX_IN_MEMORY_OBSERVATION_ITEMS);
302        let max_bytes = default_max_bytes(capacity);
303        Self::with_limits(capacity, max_bytes)
304    }
305
306    /// Creates a sink with a tighter aggregate retained-byte ceiling.
307    ///
308    /// Count is clamped to the public safety ceiling. The byte limit is clamped
309    /// to `1..=default_max_bytes(capacity)` and is observable through
310    /// [`Self::pressure`].
311    pub fn with_max_bytes(capacity: usize, max_bytes: usize) -> Self {
312        let capacity = capacity.clamp(1, MAX_IN_MEMORY_OBSERVATION_ITEMS);
313        Self::with_limits(capacity, max_bytes.clamp(1, default_max_bytes(capacity)))
314    }
315
316    fn with_limits(capacity: usize, max_bytes: usize) -> Self {
317        Self {
318            capacity,
319            state: Arc::new(Mutex::new(ObservationState {
320                events: Arc::new(VecDeque::new()),
321                pressure: InMemoryObservationPressure {
322                    max_entries: capacity,
323                    max_bytes,
324                    ..InMemoryObservationPressure::default()
325                },
326            })),
327            drains: Arc::new(RwLock::new(Arc::new(Vec::new()))),
328        }
329    }
330
331    /// Adds an operational drain that receives future redacted events.
332    pub fn add_drain(&self, drain: Arc<dyn ObservationSink>) {
333        let _ = self.try_add_drain(drain);
334    }
335
336    /// Attempts to add an operational drain under the attachment ceiling.
337    pub fn try_add_drain(&self, drain: Arc<dyn ObservationSink>) -> bool {
338        let accepted = {
339            let mut drains = self.drains.write();
340            if drains.len() >= MAX_OBSERVATION_DRAINS {
341                false
342            } else {
343                Arc::make_mut(&mut drains).push(drain);
344                true
345            }
346        };
347        if !accepted {
348            let mut state = self.state.lock();
349            state.pressure.drain_rejections = state.pressure.drain_rejections.saturating_add(1);
350        }
351        accepted
352    }
353
354    /// Returns the number of attached operational drains.
355    pub fn drain_count(&self) -> usize {
356        self.drains.read().len()
357    }
358
359    /// Returns a stable snapshot from oldest to newest.
360    pub fn snapshot(&self) -> Vec<ObservationEvent> {
361        self.shared_snapshot().iter().cloned().collect()
362    }
363
364    /// Returns an immutable snapshot without cloning retained observations.
365    pub fn shared_snapshot(&self) -> ObservationSnapshot {
366        ObservationSnapshot {
367            events: Arc::clone(&self.state.lock().events),
368        }
369    }
370
371    /// Returns current count and retained-memory pressure.
372    pub fn pressure(&self) -> InMemoryObservationPressure {
373        self.state.lock().pressure
374    }
375
376    /// Returns the number of retained events.
377    pub fn len(&self) -> usize {
378        self.state.lock().pressure.entries
379    }
380
381    /// Reports whether no events are retained.
382    pub fn is_empty(&self) -> bool {
383        self.len() == 0
384    }
385}
386
387impl Default for InMemoryObservationSink {
388    fn default() -> Self {
389        Self::new(1_024)
390    }
391}
392
393impl ObservationSink for InMemoryObservationSink {
394    fn emit(&self, event: ObservationEvent) {
395        self.emit_shared(&SharedObservationEvent::new(event));
396    }
397
398    fn emit_shared(&self, event: &SharedObservationEvent) {
399        let retained_bytes = observation_retained_bytes(event.as_event());
400        let mut state = self.state.lock();
401        if retained_bytes > state.pressure.max_bytes {
402            state.pressure.oversized_rejections =
403                state.pressure.oversized_rejections.saturating_add(1);
404        } else {
405            while state.pressure.entries >= self.capacity
406                || state.pressure.used_bytes.saturating_add(retained_bytes)
407                    > state.pressure.max_bytes
408            {
409                let Some(removed) = Arc::make_mut(&mut state.events).pop_front() else {
410                    break;
411                };
412                state.pressure.entries = state.pressure.entries.saturating_sub(1);
413                state.pressure.used_bytes = state
414                    .pressure
415                    .used_bytes
416                    .saturating_sub(observation_retained_bytes(&removed));
417                state.pressure.evictions = state.pressure.evictions.saturating_add(1);
418            }
419            Arc::make_mut(&mut state.events).push_back(event.clone_event_arc());
420            state.pressure.entries = state.pressure.entries.saturating_add(1);
421            state.pressure.used_bytes = state.pressure.used_bytes.saturating_add(retained_bytes);
422            state.pressure.peak_bytes = state.pressure.peak_bytes.max(state.pressure.used_bytes);
423        }
424        drop(state);
425        let drains = Arc::clone(&self.drains.read());
426        for drain in drains.iter() {
427            drain.emit_shared(event);
428        }
429    }
430}
431
432fn default_max_bytes(capacity: usize) -> usize {
433    capacity
434        .saturating_mul(maximum_observation_retained_bytes())
435        .clamp(1, MAX_IN_MEMORY_OBSERVATION_BYTES)
436}
437
438fn maximum_observation_retained_bytes() -> usize {
439    OBSERVATION_FIXED_BYTES
440        .saturating_add(MAX_OBSERVATION_NAME_BYTES)
441        .saturating_add(MAX_OBSERVATION_TRACE_BYTES)
442        .saturating_add(
443            MAX_OBSERVATION_ATTRIBUTES.saturating_mul(
444                ATTRIBUTE_FIXED_BYTES
445                    .saturating_add(MAX_OBSERVATION_KEY_BYTES)
446                    .saturating_add(MAX_OBSERVATION_VALUE_BYTES),
447            ),
448        )
449}
450
451fn observation_retained_bytes(event: &ObservationEvent) -> usize {
452    OBSERVATION_FIXED_BYTES
453        .saturating_add(event.name.capacity())
454        .saturating_add(event.trace_id.as_ref().map_or(0, String::capacity))
455        .saturating_add(event.attributes.iter().fold(0usize, |total, (key, value)| {
456            total
457                .saturating_add(ATTRIBUTE_FIXED_BYTES)
458                .saturating_add(key.capacity())
459                .saturating_add(value.capacity())
460        }))
461}
462
463fn is_sensitive_key(key: &str) -> bool {
464    ["secret", "password", "token", "credential", "private_key"]
465        .iter()
466        .any(|fragment| {
467            key.as_bytes()
468                .windows(fragment.len())
469                .any(|candidate| candidate.eq_ignore_ascii_case(fragment.as_bytes()))
470        })
471}
472
473#[cfg(test)]
474#[path = "observation_tests.rs"]
475mod tests;
appcore_ops/observation.rs

appcore_ops/
observation.rs
