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/// Complete passive checkpoint, with rows ordered oldest first.
156///
157/// This is not a byte format or an instance identity. Restore only through
158/// [`Log::restore`]; consumers own durable encoding, identity and allocation.
159#[derive(Clone, Debug, Eq, PartialEq)]
160pub struct LogSnapshot {
161    /// Fixed bounds of the original log.
162    pub limits: Limits,
163    /// Retained contiguous suffix of issued identities, oldest first.
164    pub entries: Vec<Entry>,
165    /// Next unissued identity; `None` means permanent sequence exhaustion.
166    pub next_sequence: Option<u64>,
167    /// Last admitted timestamp, retained even when no rows remain.
168    pub last_created_at_secs: Option<u64>,
169}
170
171/// Refusal of a checkpoint that the maintained engine cannot produce.
172#[derive(Clone, Copy, Debug, Eq, PartialEq)]
173pub enum RestoreError {
174    /// Retained row count exceeds the fixed limit.
175    TooManyEntries,
176    /// Rows are not contiguous or the next identity contradicts the retained tail.
177    InvalidSequence,
178    /// Row times regress or the timestamp watermark contradicts the retained tail.
179    InvalidTimestamp,
180    /// A label/message exceeds its bound or truncated text is inconsistent.
181    InvalidEntry,
182}
183
184impl core::fmt::Display for RestoreError {
185    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
186        formatter.write_str(match self {
187            Self::TooManyEntries => "log checkpoint exceeds its entry limit",
188            Self::InvalidSequence => "invalid log checkpoint sequence",
189            Self::InvalidTimestamp => "invalid log checkpoint timestamp",
190            Self::InvalidEntry => "invalid bounded log checkpoint entry",
191        })
192    }
193}
194
195impl core::error::Error for RestoreError {}
196
197/// Caller-owned heap history with fixed limits and monotonic append watermarks.
198#[derive(Debug)]
199pub struct Log {
200    limits: Limits,
201    entries: VecDeque<Entry>,
202    next_sequence: Option<u64>,
203    last_created_at_secs: Option<u64>,
204}
205
206impl Log {
207    /// Construct an empty log without preallocating the configured maximum.
208    #[must_use]
209    pub const fn new(limits: Limits) -> Self {
210        Self {
211            limits,
212            entries: VecDeque::new(),
213            next_sequence: Some(0),
214            last_created_at_secs: None,
215        }
216    }
217
218    /// Return the fixed bounds without copying retained messages.
219    #[must_use]
220    pub const fn limits(&self) -> Limits {
221        self.limits
222    }
223
224    /// Copy all retained rows and both append watermarks for explicit checkpointing.
225    /// Work and allocation scale with retained history, not a cursor-page limit.
226    #[must_use]
227    pub fn snapshot(&self) -> LogSnapshot {
228        LogSnapshot {
229            limits: self.limits,
230            entries: self.entries.iter().cloned().collect(),
231            next_sequence: self.next_sequence,
232            last_created_at_secs: self.last_created_at_secs,
233        }
234    }
235
236    /// Admit a complete checkpoint without replaying entries or reissuing identities.
237    ///
238    /// Empty retained history preserves its original append watermarks. Admission
239    /// proves internal consistency, not source authenticity or host continuity.
240    /// Byte decoders must bound allocation before constructing the checkpoint.
241    ///
242    /// # Errors
243    /// Rejects bounds, row ordering, watermarks and truncation states that cannot
244    /// result from this engine's append/retention operations.
245    pub fn restore(snapshot: LogSnapshot) -> Result<Self, RestoreError> {
246        let LogSnapshot {
247            limits,
248            entries,
249            next_sequence,
250            last_created_at_secs,
251        } = snapshot;
252        if entries.len() > limits.max_entries.get() {
253            return Err(RestoreError::TooManyEntries);
254        }
255        if (next_sequence == Some(0)) != last_created_at_secs.is_none() {
256            return Err(RestoreError::InvalidTimestamp);
257        }
258        let mut previous: Option<&Entry> = None;
259        for entry in &entries {
260            if entry.source.len() > limits.max_label_bytes
261                || entry
262                    .topic
263                    .as_ref()
264                    .is_some_and(|topic| topic.len() > limits.max_label_bytes)
265                || !valid_retained_message(entry, limits.max_message_bytes)
266            {
267                return Err(RestoreError::InvalidEntry);
268            }
269            if let Some(previous) = previous {
270                if previous.sequence.checked_add(1) != Some(entry.sequence) {
271                    return Err(RestoreError::InvalidSequence);
272                }
273                if entry.created_at_secs < previous.created_at_secs {
274                    return Err(RestoreError::InvalidTimestamp);
275                }
276            }
277            previous = Some(entry);
278        }
279        if let Some(last) = previous {
280            if last.sequence.checked_add(1) != next_sequence {
281                return Err(RestoreError::InvalidSequence);
282            }
283            if Some(last.created_at_secs) != last_created_at_secs {
284                return Err(RestoreError::InvalidTimestamp);
285            }
286        }
287        Ok(Self {
288            limits,
289            entries: entries.into(),
290            next_sequence,
291            last_created_at_secs,
292        })
293    }
294
295    /// Return the retained row count.
296    #[must_use]
297    pub fn len(&self) -> usize {
298        self.entries.len()
299    }
300
301    /// Whether this log currently has no retained rows.
302    #[must_use]
303    pub fn is_empty(&self) -> bool {
304        self.entries.is_empty()
305    }
306
307    /// Admit one row and evict the oldest if the count ceiling is reached.
308    ///
309    /// # Errors
310    /// Returns a typed refusal for oversized labels, regressed time or exhausted
311    /// sequence identities. Validation completes before any eviction or mutation.
312    pub fn append(&mut self, input: EntryInput<'_>) -> Result<u64, AppendError> {
313        self.validate_label(Label::Source, input.source)?;
314        if let Some(topic) = input.topic {
315            self.validate_label(Label::Topic, topic)?;
316        }
317        if let Some(previous) = self.last_created_at_secs
318            && input.created_at_secs < previous
319        {
320            return Err(AppendError::TimestampRegressed {
321                previous,
322                current: input.created_at_secs,
323            });
324        }
325        let sequence = self.next_sequence.ok_or(AppendError::SequenceExhausted)?;
326        let entry = Entry {
327            sequence,
328            created_at_secs: input.created_at_secs,
329            level: input.level,
330            source: String::from(input.source),
331            topic: input.topic.map(String::from),
332            message: bounded_message(input.message, self.limits.max_message_bytes),
333            truncated: input.message.len() > self.limits.max_message_bytes,
334        };
335        if self.entries.len() == self.limits.max_entries.get() {
336            self.entries.pop_front();
337        }
338        self.entries.push_back(entry);
339        self.next_sequence = sequence.checked_add(1);
340        self.last_created_at_secs = Some(input.created_at_secs);
341        Ok(sequence)
342    }
343
344    /// Remove at most `max_remove` entries strictly older than the age cutoff.
345    /// An entry remains available through its full configured age second.
346    pub fn prune_expired(
347        &mut self,
348        now_secs: u64,
349        max_age_secs: u64,
350        max_remove: usize,
351    ) -> RetentionBatch {
352        let cutoff = now_secs.saturating_sub(max_age_secs);
353        let mut removed = 0;
354        while removed < max_remove && self.expired_before(cutoff) {
355            self.entries.pop_front();
356            removed += 1;
357        }
358        RetentionBatch {
359            removed,
360            more_due: self.expired_before(cutoff),
361        }
362    }
363
364    /// First second at which the oldest row expires; `None` for empty history or overflow.
365    #[must_use]
366    pub fn next_expiry_secs(&self, max_age_secs: u64) -> Option<u64> {
367        self.entries
368            .front()?
369            .created_at_secs
370            .checked_add(max_age_secs)?
371            .checked_add(1)
372    }
373
374    /// Scan newest-first without cloning retained messages or counting every match.
375    /// Follow `next_before` with the same filters, including after an empty page.
376    #[must_use]
377    pub fn page(&self, filter: Filter<'_>, request: PageRequest) -> Page<'_> {
378        let mut cursor = request.before.map_or(self.entries.len(), |sequence| {
379            self.entries
380                .binary_search_by_key(&sequence, |entry| entry.sequence)
381                .unwrap_or_else(|index| index)
382        });
383        let mut entries = Vec::new();
384        let mut scanned = 0;
385        let mut last_scanned = None;
386        while cursor > 0
387            && scanned < request.max_scanned.get()
388            && entries.len() < request.max_results.get()
389        {
390            cursor -= 1;
391            let entry = &self.entries[cursor];
392            scanned += 1;
393            last_scanned = Some(entry.sequence);
394            if filter.matches(entry) {
395                entries.push(entry);
396            }
397        }
398        Page {
399            entries,
400            scanned,
401            next_before: if cursor > 0 { last_scanned } else { None },
402        }
403    }
404
405    const fn validate_label(&self, label: Label, text: &str) -> Result<(), AppendError> {
406        if text.len() > self.limits.max_label_bytes {
407            return Err(AppendError::LabelTooLong {
408                label,
409                actual_bytes: text.len(),
410                max_bytes: self.limits.max_label_bytes,
411            });
412        }
413        Ok(())
414    }
415
416    fn expired_before(&self, cutoff: u64) -> bool {
417        self.entries
418            .front()
419            .is_some_and(|entry| entry.created_at_secs < cutoff)
420    }
421}
422
423fn valid_retained_message(entry: &Entry, max_bytes: usize) -> bool {
424    if entry.message.len() > max_bytes {
425        return false;
426    }
427    if !entry.truncated {
428        return true;
429    }
430    let budget = if max_bytes > TRUNCATION_SUFFIX.len() {
431        if !entry.message.ends_with(TRUNCATION_SUFFIX) {
432            return false;
433        }
434        max_bytes - TRUNCATION_SUFFIX.len()
435    } else {
436        max_bytes
437    };
438    let prefix_len = if max_bytes > TRUNCATION_SUFFIX.len() {
439        entry.message.len() - TRUNCATION_SUFFIX.len()
440    } else {
441        entry.message.len()
442    };
443    budget.saturating_sub(prefix_len) < 4
444}
445
446fn bounded_message(message: &str, max_bytes: usize) -> String {
447    if message.len() <= max_bytes {
448        return String::from(message);
449    }
450    let suffix = if max_bytes > TRUNCATION_SUFFIX.len() {
451        TRUNCATION_SUFFIX
452    } else {
453        ""
454    };
455    let mut end = max_bytes - suffix.len();
456    while !message.is_char_boundary(end) {
457        end -= 1;
458    }
459    let mut retained = String::from(&message[..end]);
460    retained.push_str(suffix);
461    retained
462}
463
464#[cfg(test)]
465mod tests;
ic_logging/log/mod.rs

ic_logging/log/
mod.rs
