ic_logging/log/
mod.rs

1//! Heap log admission, immutable rows, bounded retention and cursor paging.
2//!
3//! Callers own authorization, scheduling, time and any durable storage.
4
5use alloc::{collections::VecDeque, string::String, vec::Vec};
6use core::num::NonZeroUsize;
7
8const TRUNCATION_SUFFIX: &str = "...[truncated]";
9
10/// Severity ordering used by minimum-level filters.
11#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
12pub enum Level {
13    /// Development detail.
14    Debug,
15    /// Ordinary informational observation.
16    Info,
17    /// Successful outcome, ordered above informational observations.
18    Ok,
19    /// Recoverable or suspicious condition.
20    Warn,
21    /// Failed operation or invariant observation.
22    Error,
23}
24
25/// Fixed limits chosen by the host before constructing a log.
26#[derive(Clone, Copy, Debug, Eq, PartialEq)]
27pub struct Limits {
28    /// Maximum retained rows; count eviction removes the oldest row.
29    pub max_entries: NonZeroUsize,
30    /// Maximum retained message bytes, including any truncation suffix.
31    pub max_message_bytes: usize,
32    /// Maximum UTF-8 bytes in each source or topic label.
33    pub max_label_bytes: usize,
34}
35
36/// Borrowed append input; bounded retained strings are copied only after admission.
37#[derive(Clone, Copy, Debug)]
38pub struct EntryInput<'a> {
39    /// Caller-supplied whole seconds; may not regress within this log.
40    pub created_at_secs: u64,
41    /// Observation severity.
42    pub level: Level,
43    /// Application or component label.
44    pub source: &'a str,
45    /// Optional application-defined topic.
46    pub topic: Option<&'a str>,
47    /// Message to retain, truncated to the configured byte ceiling.
48    pub message: &'a str,
49}
50
51/// Immutable retained row; sequence identities belong to one log instance.
52#[derive(Clone, Debug, Eq, PartialEq)]
53pub struct Entry {
54    /// Monotonic identity, starting at zero and never reused within one log.
55    pub sequence: u64,
56    /// Admitted caller-supplied timestamp.
57    pub created_at_secs: u64,
58    /// Observation severity.
59    pub level: Level,
60    /// Bounded application or component label.
61    pub source: String,
62    /// Optional bounded topic label.
63    pub topic: Option<String>,
64    /// Bounded message text.
65    pub message: String,
66    /// Whether the original message exceeded its byte limit.
67    pub truncated: bool,
68}
69
70/// Label rejected by append admission.
71#[derive(Clone, Copy, Debug, Eq, PartialEq)]
72pub enum Label {
73    /// Application or component source.
74    Source,
75    /// Optional topic.
76    Topic,
77}
78
79/// Append refusal; every variant leaves entries and watermarks unchanged.
80#[derive(Clone, Copy, Debug, Eq, PartialEq)]
81pub enum AppendError {
82    /// A label exceeded its configured byte ceiling.
83    LabelTooLong {
84        /// Rejected field.
85        label: Label,
86        /// Supplied UTF-8 byte length.
87        actual_bytes: usize,
88        /// Configured ceiling.
89        max_bytes: usize,
90    },
91    /// The timestamp is older than the last admitted append, even after pruning.
92    TimestampRegressed {
93        /// Last admitted timestamp.
94        previous: u64,
95        /// Rejected timestamp.
96        current: u64,
97    },
98    /// All sequence identities have been issued for this log instance.
99    SequenceExhausted,
100}
101
102/// Optional exact label filters and a minimum severity; all supplied filters apply.
103#[derive(Clone, Copy, Debug, Default)]
104pub struct Filter<'a> {
105    /// Match this exact source when present.
106    pub source: Option<&'a str>,
107    /// Match this exact topic when present.
108    pub topic: Option<&'a str>,
109    /// Include this severity and higher when present.
110    pub min_level: Option<Level>,
111}
112
113impl Filter<'_> {
114    fn matches(self, entry: &Entry) -> bool {
115        let source_matches = self.source.is_none_or(|source| source == entry.source);
116        let topic_matches = self
117            .topic
118            .is_none_or(|topic| Some(topic) == entry.topic.as_deref());
119        let level_matches = self.min_level.is_none_or(|level| entry.level >= level);
120        source_matches && topic_matches && level_matches
121    }
122}
123
124/// One newest-first scan using an exclusive sequence cursor.
125#[derive(Clone, Copy, Debug)]
126pub struct PageRequest {
127    /// Scan entries older than this sequence, or begin at the newest entry.
128    pub before: Option<u64>,
129    /// Maximum matching rows to return.
130    pub max_results: NonZeroUsize,
131    /// Maximum candidate rows to examine after binary-search positioning.
132    pub max_scanned: NonZeroUsize,
133}
134
135/// Borrowed page; a continuation can be present even when no row matched.
136#[derive(Debug)]
137pub struct Page<'a> {
138    /// Matching rows, newest first.
139    pub entries: Vec<&'a Entry>,
140    /// Number of candidate rows examined, not the total number of matches.
141    pub scanned: usize,
142    /// Exclusive cursor for the next older scan, if retained candidates remain.
143    pub next_before: Option<u64>,
144}
145
146/// Result of a bounded age-retention call.
147#[derive(Clone, Copy, Debug, Eq, PartialEq)]
148pub struct RetentionBatch {
149    /// Number of rows removed by this call.
150    pub removed: usize,
151    /// Whether the oldest remaining row is also expired.
152    pub more_due: bool,
153}
154
155/// Caller-owned heap history with fixed limits and monotonic append watermarks.
156#[derive(Debug)]
157pub struct Log {
158    limits: Limits,
159    entries: VecDeque<Entry>,
160    next_sequence: Option<u64>,
161    last_created_at_secs: Option<u64>,
162}
163
164impl Log {
165    /// Construct an empty log without preallocating the configured maximum.
166    #[must_use]
167    pub const fn new(limits: Limits) -> Self {
168        Self {
169            limits,
170            entries: VecDeque::new(),
171            next_sequence: Some(0),
172            last_created_at_secs: None,
173        }
174    }
175
176    /// Return the retained row count.
177    #[must_use]
178    pub fn len(&self) -> usize {
179        self.entries.len()
180    }
181
182    /// Whether this log currently has no retained rows.
183    #[must_use]
184    pub fn is_empty(&self) -> bool {
185        self.entries.is_empty()
186    }
187
188    /// Admit one row and evict the oldest if the count ceiling is reached.
189    ///
190    /// # Errors
191    /// Returns a typed refusal for oversized labels, regressed time or exhausted
192    /// sequence identities. Validation completes before any eviction or mutation.
193    pub fn append(&mut self, input: EntryInput<'_>) -> Result<u64, AppendError> {
194        self.validate_label(Label::Source, input.source)?;
195        if let Some(topic) = input.topic {
196            self.validate_label(Label::Topic, topic)?;
197        }
198        if let Some(previous) = self.last_created_at_secs
199            && input.created_at_secs < previous
200        {
201            return Err(AppendError::TimestampRegressed {
202                previous,
203                current: input.created_at_secs,
204            });
205        }
206        let sequence = self.next_sequence.ok_or(AppendError::SequenceExhausted)?;
207        let entry = Entry {
208            sequence,
209            created_at_secs: input.created_at_secs,
210            level: input.level,
211            source: String::from(input.source),
212            topic: input.topic.map(String::from),
213            message: bounded_message(input.message, self.limits.max_message_bytes),
214            truncated: input.message.len() > self.limits.max_message_bytes,
215        };
216        if self.entries.len() == self.limits.max_entries.get() {
217            self.entries.pop_front();
218        }
219        self.entries.push_back(entry);
220        self.next_sequence = sequence.checked_add(1);
221        self.last_created_at_secs = Some(input.created_at_secs);
222        Ok(sequence)
223    }
224
225    /// Remove at most `max_remove` entries strictly older than the age cutoff.
226    /// An entry remains available through its full configured age second.
227    pub fn prune_expired(
228        &mut self,
229        now_secs: u64,
230        max_age_secs: u64,
231        max_remove: usize,
232    ) -> RetentionBatch {
233        let cutoff = now_secs.saturating_sub(max_age_secs);
234        let mut removed = 0;
235        while removed < max_remove && self.expired_before(cutoff) {
236            self.entries.pop_front();
237            removed += 1;
238        }
239        RetentionBatch {
240            removed,
241            more_due: self.expired_before(cutoff),
242        }
243    }
244
245    /// First second at which the oldest row expires; `None` for empty history or overflow.
246    #[must_use]
247    pub fn next_expiry_secs(&self, max_age_secs: u64) -> Option<u64> {
248        self.entries
249            .front()?
250            .created_at_secs
251            .checked_add(max_age_secs)?
252            .checked_add(1)
253    }
254
255    /// Scan newest-first without cloning retained messages or counting every match.
256    /// Follow `next_before` with the same filters, including after an empty page.
257    #[must_use]
258    pub fn page(&self, filter: Filter<'_>, request: PageRequest) -> Page<'_> {
259        let mut cursor = request.before.map_or(self.entries.len(), |sequence| {
260            self.entries
261                .binary_search_by_key(&sequence, |entry| entry.sequence)
262                .unwrap_or_else(|index| index)
263        });
264        let mut entries = Vec::new();
265        let mut scanned = 0;
266        let mut last_scanned = None;
267        while cursor > 0
268            && scanned < request.max_scanned.get()
269            && entries.len() < request.max_results.get()
270        {
271            cursor -= 1;
272            let entry = &self.entries[cursor];
273            scanned += 1;
274            last_scanned = Some(entry.sequence);
275            if filter.matches(entry) {
276                entries.push(entry);
277            }
278        }
279        Page {
280            entries,
281            scanned,
282            next_before: if cursor > 0 { last_scanned } else { None },
283        }
284    }
285
286    const fn validate_label(&self, label: Label, text: &str) -> Result<(), AppendError> {
287        if text.len() > self.limits.max_label_bytes {
288            return Err(AppendError::LabelTooLong {
289                label,
290                actual_bytes: text.len(),
291                max_bytes: self.limits.max_label_bytes,
292            });
293        }
294        Ok(())
295    }
296
297    fn expired_before(&self, cutoff: u64) -> bool {
298        self.entries
299            .front()
300            .is_some_and(|entry| entry.created_at_secs < cutoff)
301    }
302}
303
304fn bounded_message(message: &str, max_bytes: usize) -> String {
305    if message.len() <= max_bytes {
306        return String::from(message);
307    }
308    let suffix = if max_bytes > TRUNCATION_SUFFIX.len() {
309        TRUNCATION_SUFFIX
310    } else {
311        ""
312    };
313    let mut end = max_bytes - suffix.len();
314    while !message.is_char_boundary(end) {
315        end -= 1;
316    }
317    let mut retained = String::from(&message[..end]);
318    retained.push_str(suffix);
319    retained
320}
321
322#[cfg(test)]
323mod tests;
ic_logging/log/mod.rs

ic_logging/log/
mod.rs
