appcore_core/
audit.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: audit.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/05/31 13:38:42 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 16:07:49 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Bounded in-memory audit log for command dispatch outcomes.
12
13use crate::audit_bounds::{
14    audit_entry_is_bounded_and_redacted, audit_entry_retained_bytes, audit_record_retained_bytes,
15    bound_audit_entry, bound_audit_text, bound_audit_trace, MAX_AUDIT_ID_BYTES,
16};
17use crate::ids::{AppId, CommandName, NodeId};
18use crate::operational_journal::{FileOperationalJournal, OperationalJournalRecord};
19use crate::trace::TraceContext;
20use crate::{redact_text, MAX_OPERATIONAL_TEXT_BYTES};
21use parking_lot::Mutex;
22use serde::ser::SerializeSeq;
23use serde::{Serialize, Serializer};
24use std::collections::VecDeque;
25use std::io::{self, Write};
26use std::sync::Arc;
27
28const MAX_AUDIT_RECORDS: usize = 10_000;
29/// Default aggregate memory budget for process-local audit records and entries.
30pub const DEFAULT_AUDIT_LOG_MAX_BYTES: usize = 16 * 1024 * 1024;
31
32/// Controlled outcome recorded for an audited operation.
33#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum AuditOutcome {
36    /// Operation completed successfully.
37    Accepted,
38    /// Operation was rejected by policy or validation.
39    Rejected,
40    /// Operation failed during execution.
41    Error,
42}
43
44/// Command-specific audit record.
45#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
46pub struct AuditRecord {
47    /// Command identity.
48    pub command_id: String,
49    /// Command name.
50    pub command_name: CommandName,
51    /// Application scope.
52    pub app_id: AppId,
53    /// Runtime node scope.
54    pub node_id: NodeId,
55    /// Command start timestamp in Unix milliseconds.
56    pub timestamp_ms: u64,
57    /// Recorded command outcome.
58    pub outcome: AuditOutcome,
59    /// Optional redacted detail.
60    pub message: Option<String>,
61    /// Optional distributed trace context.
62    pub trace: Option<TraceContext>,
63}
64
65/// Generic operational category associated with an audit entry.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
67#[serde(rename_all = "snake_case")]
68pub enum AuditCategory {
69    /// Command dispatch.
70    Command,
71    /// Query dispatch.
72    Query,
73    /// Event processing.
74    Event,
75    /// Scheduler execution.
76    Scheduler,
77    /// Control-plane operation.
78    ControlPlane,
79    /// Direct peer RPC operation.
80    PeerRpc,
81    /// Runtime lifecycle or infrastructure operation.
82    Runtime,
83}
84
85/// Transport-neutral append-only operational audit entry.
86#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
87pub struct AuditEntry {
88    /// Generic operation category.
89    pub category: AuditCategory,
90    /// Stable operation identity.
91    pub operation_id: String,
92    /// Stable operation name.
93    pub operation_name: String,
94    /// Optional application scope.
95    pub app_id: Option<String>,
96    /// Optional node scope.
97    pub node_id: Option<String>,
98    /// Start timestamp in Unix milliseconds.
99    pub started_at_ms: u64,
100    /// Completion timestamp in Unix milliseconds.
101    pub completed_at_ms: u64,
102    /// Saturating elapsed time in milliseconds.
103    pub latency_ms: u64,
104    /// Recorded operation outcome.
105    pub outcome: AuditOutcome,
106    /// Optional redacted detail.
107    pub message: Option<String>,
108    /// Optional distributed trace context.
109    pub trace: Option<TraceContext>,
110}
111
112impl AuditEntry {
113    /// Creates an unscoped operational audit entry.
114    pub fn new(
115        category: AuditCategory,
116        operation_id: impl Into<String>,
117        operation_name: impl Into<String>,
118        started_at_ms: u64,
119        completed_at_ms: u64,
120        outcome: AuditOutcome,
121    ) -> Self {
122        Self {
123            category,
124            operation_id: operation_id.into(),
125            operation_name: operation_name.into(),
126            app_id: None,
127            node_id: None,
128            started_at_ms,
129            completed_at_ms,
130            latency_ms: completed_at_ms.saturating_sub(started_at_ms),
131            outcome,
132            message: None,
133            trace: None,
134        }
135    }
136
137    /// Adds application and node scope.
138    pub fn with_runtime_scope(mut self, app_id: &AppId, node_id: &NodeId) -> Self {
139        self.app_id = Some(app_id.as_str().to_string());
140        self.node_id = Some(node_id.as_str().to_string());
141        self
142    }
143
144    /// Adds a redacted optional message.
145    pub fn with_message(mut self, message: Option<String>) -> Self {
146        self.message = message.map(|value| redact_text(&value));
147        self
148    }
149
150    /// Adds distributed trace context.
151    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
152        self.trace = trace;
153        self
154    }
155
156    pub(crate) fn into_bounded(self) -> Self {
157        bound_audit_entry(self)
158    }
159
160    pub(crate) fn is_bounded_and_redacted(&self) -> bool {
161        audit_entry_is_bounded_and_redacted(self)
162    }
163}
164
165/// Point-in-time pressure metrics for a process-local audit log.
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
167pub struct AuditLogStats {
168    /// Retained command records.
169    pub record_count: usize,
170    /// Retained generic audit entries.
171    pub entry_count: usize,
172    /// Estimated bytes retained by both snapshots.
173    pub used_bytes: usize,
174    /// Highest retained byte count observed by this log.
175    pub peak_bytes: usize,
176    /// Records or entries evicted to maintain count or byte limits.
177    pub evictions: u64,
178    /// Individual records or entries too large for the configured budget.
179    pub rejections: u64,
180    /// Aggregate configured byte budget.
181    pub max_bytes: usize,
182}
183
184/// Shared immutable point-in-time view of command audit records.
185#[derive(Clone, Debug)]
186pub struct AuditRecordsSnapshot {
187    records: Arc<VecDeque<Arc<AuditRecord>>>,
188}
189
190impl AuditRecordsSnapshot {
191    /// Returns the number of records captured by this snapshot.
192    #[must_use]
193    pub fn len(&self) -> usize {
194        self.records.len()
195    }
196
197    /// Reports whether this snapshot contains no records.
198    #[must_use]
199    pub fn is_empty(&self) -> bool {
200        self.records.is_empty()
201    }
202
203    /// Iterates over at most the newest `limit` records without cloning them.
204    pub fn recent(&self, limit: usize) -> impl Iterator<Item = &AuditRecord> {
205        let start = self.records.len().saturating_sub(limit);
206        self.records.iter().skip(start).map(Arc::as_ref)
207    }
208}
209
210impl Serialize for AuditRecordsSnapshot {
211    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
212    where
213        S: Serializer,
214    {
215        let mut sequence = serializer.serialize_seq(Some(self.records.len()))?;
216        for record in self.records.iter() {
217            sequence.serialize_element(record.as_ref())?;
218        }
219        sequence.end()
220    }
221}
222
223/// Shared immutable point-in-time view of generic audit entries.
224#[derive(Clone, Debug)]
225pub struct AuditEntriesSnapshot {
226    entries: Arc<VecDeque<Arc<OperationalJournalRecord>>>,
227}
228
229impl AuditEntriesSnapshot {
230    /// Returns the number of entries captured by this snapshot.
231    #[must_use]
232    pub fn len(&self) -> usize {
233        self.entries.len()
234    }
235
236    /// Reports whether this snapshot contains no entries.
237    #[must_use]
238    pub fn is_empty(&self) -> bool {
239        self.entries.is_empty()
240    }
241
242    /// Iterates over at most the newest `limit` entries without cloning them.
243    pub fn recent(&self, limit: usize) -> impl Iterator<Item = &AuditEntry> {
244        let start = self.entries.len().saturating_sub(limit);
245        self.entries.iter().skip(start).filter_map(audit_entry)
246    }
247}
248
249impl Serialize for AuditEntriesSnapshot {
250    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
251    where
252        S: Serializer,
253    {
254        let mut sequence = serializer.serialize_seq(Some(self.entries.len()))?;
255        for entry in self.entries.iter().filter_map(audit_entry) {
256            sequence.serialize_element(entry)?;
257        }
258        sequence.end()
259    }
260}
261
262#[derive(Debug, Clone)]
263struct AuditState {
264    records: Arc<VecDeque<Arc<AuditRecord>>>,
265    entries: Arc<VecDeque<Arc<OperationalJournalRecord>>>,
266    used_bytes: usize,
267    peak_bytes: usize,
268    evictions: u64,
269    rejections: u64,
270    max_bytes: usize,
271}
272
273impl AuditState {
274    fn new(max_bytes: usize) -> Self {
275        Self {
276            records: Arc::default(),
277            entries: Arc::default(),
278            used_bytes: 0,
279            peak_bytes: 0,
280            evictions: 0,
281            rejections: 0,
282            max_bytes: max_bytes.max(1),
283        }
284    }
285
286    fn push_record(&mut self, record: AuditRecord) {
287        let bytes = audit_record_retained_bytes(&record);
288        while self.records.len() >= MAX_AUDIT_RECORDS {
289            self.pop_record();
290        }
291        if !self.make_room(bytes) {
292            self.rejections = self.rejections.saturating_add(1);
293            return;
294        }
295        Arc::make_mut(&mut self.records).push_back(Arc::new(record));
296        self.add_bytes(bytes);
297    }
298
299    fn push_shared_entry(&mut self, record: Arc<OperationalJournalRecord>) {
300        let Some(entry) = audit_entry(&record) else {
301            self.rejections = self.rejections.saturating_add(1);
302            return;
303        };
304        let bytes = audit_entry_retained_bytes(entry);
305        while self.entries.len() >= MAX_AUDIT_RECORDS {
306            self.pop_entry();
307        }
308        if !self.make_room(bytes) {
309            self.rejections = self.rejections.saturating_add(1);
310            return;
311        }
312        Arc::make_mut(&mut self.entries).push_back(record);
313        self.add_bytes(bytes);
314    }
315
316    fn replace_entries(&mut self, entries: Vec<Arc<OperationalJournalRecord>>) {
317        while !self.entries.is_empty() {
318            self.pop_entry();
319        }
320        for entry in entries {
321            self.push_shared_entry(entry);
322        }
323    }
324
325    fn make_room(&mut self, incoming: usize) -> bool {
326        if incoming > self.max_bytes {
327            return false;
328        }
329        while self.used_bytes.saturating_add(incoming) > self.max_bytes {
330            if !self.pop_oldest() {
331                return false;
332            }
333        }
334        true
335    }
336
337    fn pop_oldest(&mut self) -> bool {
338        let record_time = self.records.front().map(|record| record.timestamp_ms);
339        let entry_time = self
340            .entries
341            .front()
342            .and_then(audit_entry)
343            .map(|entry| entry.started_at_ms);
344        match (record_time, entry_time) {
345            (Some(record), Some(entry)) if record <= entry => self.pop_record(),
346            (Some(_), Some(_)) | (None, Some(_)) => self.pop_entry(),
347            (Some(_), None) => self.pop_record(),
348            (None, None) => return false,
349        }
350        true
351    }
352
353    fn pop_record(&mut self) {
354        if let Some(record) = Arc::make_mut(&mut self.records).pop_front() {
355            self.used_bytes = self
356                .used_bytes
357                .saturating_sub(audit_record_retained_bytes(&record));
358            self.evictions = self.evictions.saturating_add(1);
359        }
360    }
361
362    fn pop_entry(&mut self) {
363        if let Some(entry) = Arc::make_mut(&mut self.entries).pop_front() {
364            if let Some(entry) = audit_entry(&entry) {
365                self.used_bytes = self
366                    .used_bytes
367                    .saturating_sub(audit_entry_retained_bytes(entry));
368            }
369            self.evictions = self.evictions.saturating_add(1);
370        }
371    }
372
373    fn add_bytes(&mut self, bytes: usize) {
374        self.used_bytes = self.used_bytes.saturating_add(bytes);
375        self.peak_bytes = self.peak_bytes.max(self.used_bytes);
376    }
377
378    fn stats(&self) -> AuditLogStats {
379        AuditLogStats {
380            record_count: self.records.len(),
381            entry_count: self.entries.len(),
382            used_bytes: self.used_bytes,
383            peak_bytes: self.peak_bytes,
384            evictions: self.evictions,
385            rejections: self.rejections,
386            max_bytes: self.max_bytes,
387        }
388    }
389}
390
391/// Bounded process-local audit log.
392#[derive(Debug)]
393pub struct AuditLog {
394    state: Mutex<AuditState>,
395    journal: Mutex<Option<Arc<FileOperationalJournal>>>,
396    journal_error: Mutex<Option<String>>,
397}
398
399impl Default for AuditLog {
400    fn default() -> Self {
401        Self::with_max_bytes(DEFAULT_AUDIT_LOG_MAX_BYTES)
402    }
403}
404
405impl Clone for AuditLog {
406    fn clone(&self) -> Self {
407        Self {
408            state: Mutex::new(self.state.lock().clone()),
409            journal: Mutex::new(self.journal.lock().clone()),
410            journal_error: Mutex::new(self.journal_error.lock().clone()),
411        }
412    }
413}
414
415impl AuditLog {
416    /// Creates an empty audit log.
417    pub fn new() -> Self {
418        Self::default()
419    }
420
421    /// Creates an empty audit log with an aggregate retained-byte budget.
422    pub fn with_max_bytes(max_bytes: usize) -> Self {
423        Self {
424            state: Mutex::new(AuditState::new(max_bytes)),
425            journal: Mutex::new(None),
426            journal_error: Mutex::new(None),
427        }
428    }
429
430    /// Attaches a journal, sharing safe entries and sanitizing any unsafe record.
431    pub fn attach_journal(&self, journal: Arc<FileOperationalJournal>) {
432        let mut entries = journal.shared_audit_records();
433        if entries.len() > MAX_AUDIT_RECORDS {
434            entries.drain(..entries.len() - MAX_AUDIT_RECORDS);
435        }
436        self.state.lock().replace_entries(entries);
437        *self.journal.lock() = Some(journal);
438        *self.journal_error.lock() = None;
439    }
440
441    /// Returns the last durable journal failure, when persistence degraded.
442    pub fn durability_error(&self) -> Option<String> {
443        self.journal_error.lock().clone()
444    }
445
446    /// Appends a command record and its generic audit projection.
447    pub fn push(&self, mut record: AuditRecord) {
448        record.command_id = bound_audit_text(&record.command_id, MAX_AUDIT_ID_BYTES);
449        record.message = record
450            .message
451            .map(|message| bound_audit_text(&message, MAX_OPERATIONAL_TEXT_BYTES));
452        bound_audit_trace(&mut record.trace);
453        let completed_at_ms = now_ms();
454        let entry = AuditEntry::new(
455            AuditCategory::Command,
456            record.command_id.clone(),
457            record.command_name.as_str(),
458            record.timestamp_ms,
459            completed_at_ms,
460            record.outcome,
461        )
462        .with_runtime_scope(&record.app_id, &record.node_id)
463        .with_message(record.message.clone())
464        .with_trace(record.trace.clone());
465        let entry = Arc::new(OperationalJournalRecord::Audit(entry));
466        self.persist_entry(Arc::clone(&entry));
467        let mut state = self.state.lock();
468        state.push_record(record);
469        state.push_shared_entry(entry);
470    }
471
472    /// Appends one generic audit entry after redaction.
473    pub fn push_entry(&self, entry: AuditEntry) {
474        let entry = entry.into_bounded();
475        let entry = Arc::new(OperationalJournalRecord::Audit(entry));
476        self.persist_entry(Arc::clone(&entry));
477        self.state.lock().push_shared_entry(entry);
478    }
479
480    fn persist_entry(&self, entry: Arc<OperationalJournalRecord>) {
481        if let Some(journal) = self.journal.lock().clone() {
482            if let Err(error) = journal.append_shared_audit(entry) {
483                *self.journal_error.lock() = Some(redact_text(&format!("{error:?}")));
484            }
485        }
486    }
487
488    /// Returns the number of command records.
489    pub fn len(&self) -> usize {
490        self.state.lock().records.len()
491    }
492
493    /// Reports whether no command records exist.
494    pub fn is_empty(&self) -> bool {
495        self.state.lock().records.is_empty()
496    }
497
498    /// Returns a point-in-time copy of command records.
499    pub fn records(&self) -> Vec<AuditRecord> {
500        self.records_snapshot()
501            .recent(usize::MAX)
502            .cloned()
503            .collect()
504    }
505
506    /// Returns a point-in-time copy of generic audit entries.
507    pub fn entries(&self) -> Vec<AuditEntry> {
508        self.entries_snapshot()
509            .recent(usize::MAX)
510            .cloned()
511            .collect()
512    }
513
514    /// Returns a shared immutable snapshot without cloning record fields.
515    pub fn records_snapshot(&self) -> AuditRecordsSnapshot {
516        AuditRecordsSnapshot {
517            records: Arc::clone(&self.state.lock().records),
518        }
519    }
520
521    /// Returns a shared immutable snapshot without cloning entry fields.
522    pub fn entries_snapshot(&self) -> AuditEntriesSnapshot {
523        AuditEntriesSnapshot {
524            entries: Arc::clone(&self.state.lock().entries),
525        }
526    }
527
528    /// Returns current count, byte-pressure, eviction, and rejection metrics.
529    pub fn stats(&self) -> AuditLogStats {
530        self.state.lock().stats()
531    }
532
533    /// Writes generic entries as JSONL from a shared immutable snapshot.
534    pub fn write_jsonl(&self, writer: &mut impl Write) -> io::Result<()> {
535        let snapshot = self.entries_snapshot();
536        for entry in snapshot.entries.iter().filter_map(audit_entry) {
537            serde_json::to_writer(&mut *writer, entry).map_err(io::Error::other)?;
538            writer.write_all(b"\n")?;
539        }
540        Ok(())
541    }
542
543    /// Exports generic entries as newline-delimited JSON.
544    pub fn export_jsonl(&self) -> Result<String, serde_json::Error> {
545        let mut output = Vec::new();
546        self.write_jsonl(&mut output)
547            .map_err(serde_json::Error::io)?;
548        String::from_utf8(output).map_err(|error| {
549            serde_json::Error::io(io::Error::new(io::ErrorKind::InvalidData, error))
550        })
551    }
552}
553
554fn audit_entry(record: &Arc<OperationalJournalRecord>) -> Option<&AuditEntry> {
555    match record.as_ref() {
556        OperationalJournalRecord::Audit(entry) => Some(entry),
557        OperationalJournalRecord::Event(_) => None,
558    }
559}
560
561fn now_ms() -> u64 {
562    std::time::SystemTime::now()
563        .duration_since(std::time::UNIX_EPOCH)
564        .map(|duration| duration.as_millis() as u64)
565        .unwrap_or(0)
566}
567
568#[cfg(test)]
569#[path = "audit_tests.rs"]
570mod tests;
appcore_core/audit.rs

appcore_core/
audit.rs
