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
102impl core::fmt::Display for AppendError {
103    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
104        match self {
105            Self::LabelTooLong {
106                label,
107                actual_bytes,
108                max_bytes,
109            } => {
110                let label = match label {
111                    Label::Source => "source",
112                    Label::Topic => "topic",
113                };
114                write!(
115                    formatter,
116                    "log {label} label exceeds {max_bytes} bytes ({actual_bytes} supplied)"
117                )
118            }
119            Self::TimestampRegressed { previous, current } => write!(
120                formatter,
121                "log timestamp regressed from {previous} to {current} seconds"
122            ),
123            Self::SequenceExhausted => formatter.write_str("log sequence identities are exhausted"),
124        }
125    }
126}
127
128impl core::error::Error for AppendError {}
129
130/// Optional exact label filters and a minimum severity; all supplied filters apply.
131#[derive(Clone, Copy, Debug, Default)]
132pub struct Filter<'a> {
133    /// Match this exact source when present.
134    pub source: Option<&'a str>,
135    /// Match this exact topic when present.
136    pub topic: Option<&'a str>,
137    /// Include this severity and higher when present.
138    pub min_level: Option<Level>,
139}
140
141impl Filter<'_> {
142    fn matches(self, entry: &Entry) -> bool {
143        let source_matches = self.source.is_none_or(|source| source == entry.source);
144        let topic_matches = self
145            .topic
146            .is_none_or(|topic| Some(topic) == entry.topic.as_deref());
147        let level_matches = self.min_level.is_none_or(|level| entry.level >= level);
148        source_matches && topic_matches && level_matches
149    }
150}
151
152/// One newest-first scan using an exclusive sequence cursor.
153#[derive(Clone, Copy, Debug)]
154pub struct PageRequest {
155    /// Scan entries older than this sequence, or begin at the newest entry.
156    pub before: Option<u64>,
157    /// Maximum matching rows to return.
158    pub max_results: NonZeroUsize,
159    /// Maximum candidate rows to examine after binary-search positioning.
160    pub max_scanned: NonZeroUsize,
161}
162
163/// Borrowed page; a continuation can be present even when no row matched.
164#[derive(Debug)]
165pub struct Page<'a> {
166    /// Matching rows, newest first.
167    pub entries: Vec<&'a Entry>,
168    /// Number of candidate rows examined, not the total number of matches.
169    pub scanned: usize,
170    /// Exclusive cursor for the next older scan, if retained candidates remain.
171    pub next_before: Option<u64>,
172}
173
174/// Result of a bounded age-retention call.
175#[derive(Clone, Copy, Debug, Eq, PartialEq)]
176pub struct RetentionBatch {
177    /// Number of rows removed by this call.
178    pub removed: usize,
179    /// Whether the oldest remaining row is also expired.
180    pub more_due: bool,
181}
182
183/// Complete passive checkpoint, with rows ordered oldest first.
184///
185/// This is not a byte format or an instance identity. Restore only through
186/// [`Log::restore`]; consumers own durable encoding, identity and allocation.
187#[derive(Clone, Debug, Eq, PartialEq)]
188pub struct LogSnapshot {
189    /// Fixed bounds of the original log.
190    pub limits: Limits,
191    /// Retained contiguous suffix of issued identities, oldest first.
192    pub entries: Vec<Entry>,
193    /// Next unissued identity; `None` means permanent sequence exhaustion.
194    pub next_sequence: Option<u64>,
195    /// Last admitted timestamp, retained even when no rows remain.
196    pub last_created_at_secs: Option<u64>,
197}
198
199/// Refusal of a checkpoint that the maintained engine cannot produce.
200#[derive(Clone, Copy, Debug, Eq, PartialEq)]
201pub enum RestoreError {
202    /// Retained row count exceeds the fixed limit.
203    TooManyEntries,
204    /// Rows are not contiguous or the next identity contradicts the retained tail.
205    InvalidSequence,
206    /// Row times regress or the timestamp watermark is inconsistent.
207    InvalidTimestamp,
208    /// A label/message exceeds its bound or truncated text is inconsistent.
209    InvalidEntry,
210}
211
212impl core::fmt::Display for RestoreError {
213    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
214        formatter.write_str(match self {
215            Self::TooManyEntries => "log checkpoint exceeds its entry limit",
216            Self::InvalidSequence => "invalid log checkpoint sequence",
217            Self::InvalidTimestamp => "invalid log checkpoint timestamp",
218            Self::InvalidEntry => "invalid bounded log checkpoint entry",
219        })
220    }
221}
222
223impl core::error::Error for RestoreError {}
224
225/// Caller-owned heap history with fixed limits and monotonic append watermarks.
226#[derive(Debug)]
227pub struct Log {
228    limits: Limits,
229    entries: VecDeque<Entry>,
230    next_sequence: Option<u64>,
231    last_created_at_secs: Option<u64>,
232}
233
234impl Log {
235    /// Construct an empty log without preallocating the configured maximum.
236    #[must_use]
237    pub const fn new(limits: Limits) -> Self {
238        Self {
239            limits,
240            entries: VecDeque::new(),
241            next_sequence: Some(0),
242            last_created_at_secs: None,
243        }
244    }
245
246    /// Return the fixed bounds without copying retained messages.
247    #[must_use]
248    pub const fn limits(&self) -> Limits {
249        self.limits
250    }
251
252    /// Copy all retained rows and both append watermarks for explicit checkpointing.
253    /// Work and allocation scale with retained history, not a cursor-page limit.
254    #[must_use]
255    pub fn snapshot(&self) -> LogSnapshot {
256        LogSnapshot {
257            limits: self.limits,
258            entries: self.entries.iter().cloned().collect(),
259            next_sequence: self.next_sequence,
260            last_created_at_secs: self.last_created_at_secs,
261        }
262    }
263
264    /// Consume the log into a checkpoint without cloning retained entries.
265    /// Fixed limits, row order and both append watermarks are preserved, including
266    /// empty histories and permanent sequence exhaustion.
267    #[must_use]
268    pub fn into_snapshot(self) -> LogSnapshot {
269        LogSnapshot {
270            limits: self.limits,
271            entries: self.entries.into(),
272            next_sequence: self.next_sequence,
273            last_created_at_secs: self.last_created_at_secs,
274        }
275    }
276
277    /// Admit a complete checkpoint without replaying entries or reissuing identities.
278    ///
279    /// Empty retained history preserves its original append watermarks. Admission
280    /// proves internal consistency, not source authenticity or host continuity.
281    /// Byte decoders must bound allocation before constructing the checkpoint.
282    ///
283    /// # Errors
284    /// Rejects bounds, row ordering, watermarks and truncation states that cannot
285    /// result from this engine's append/retention operations.
286    pub fn restore(snapshot: LogSnapshot) -> Result<Self, RestoreError> {
287        let LogSnapshot {
288            limits,
289            entries,
290            next_sequence,
291            last_created_at_secs,
292        } = snapshot;
293        if entries.len() > limits.max_entries.get() {
294            return Err(RestoreError::TooManyEntries);
295        }
296        if (next_sequence == Some(0)) != last_created_at_secs.is_none() {
297            return Err(RestoreError::InvalidTimestamp);
298        }
299        // The last row cannot expire at the final representable timestamp, and
300        // count eviction always leaves a newer row in this nonzero-capacity log.
301        if entries.is_empty() && last_created_at_secs == Some(u64::MAX) {
302            return Err(RestoreError::InvalidTimestamp);
303        }
304        let mut previous: Option<&Entry> = None;
305        for entry in &entries {
306            if entry.source.len() > limits.max_label_bytes
307                || entry
308                    .topic
309                    .as_ref()
310                    .is_some_and(|topic| topic.len() > limits.max_label_bytes)
311                || !valid_retained_message(entry, limits.max_message_bytes)
312            {
313                return Err(RestoreError::InvalidEntry);
314            }
315            if let Some(previous) = previous {
316                if previous.sequence.checked_add(1) != Some(entry.sequence) {
317                    return Err(RestoreError::InvalidSequence);
318                }
319                if entry.created_at_secs < previous.created_at_secs {
320                    return Err(RestoreError::InvalidTimestamp);
321                }
322            }
323            previous = Some(entry);
324        }
325        if let Some(last) = previous {
326            if last.sequence.checked_add(1) != next_sequence {
327                return Err(RestoreError::InvalidSequence);
328            }
329            if Some(last.created_at_secs) != last_created_at_secs {
330                return Err(RestoreError::InvalidTimestamp);
331            }
332        }
333        Ok(Self {
334            limits,
335            entries: entries.into(),
336            next_sequence,
337            last_created_at_secs,
338        })
339    }
340
341    /// Return the retained row count.
342    #[must_use]
343    pub fn len(&self) -> usize {
344        self.entries.len()
345    }
346
347    /// Whether this log currently has no retained rows.
348    #[must_use]
349    pub fn is_empty(&self) -> bool {
350        self.entries.is_empty()
351    }
352
353    /// Admit one row and evict the oldest if the count ceiling is reached.
354    ///
355    /// # Errors
356    /// Returns a typed refusal for oversized labels, regressed time or exhausted
357    /// sequence identities. Validation completes before any eviction or mutation.
358    pub fn append(&mut self, input: EntryInput<'_>) -> Result<u64, AppendError> {
359        self.validate_label(Label::Source, input.source)?;
360        if let Some(topic) = input.topic {
361            self.validate_label(Label::Topic, topic)?;
362        }
363        if let Some(previous) = self.last_created_at_secs
364            && input.created_at_secs < previous
365        {
366            return Err(AppendError::TimestampRegressed {
367                previous,
368                current: input.created_at_secs,
369            });
370        }
371        let sequence = self.next_sequence.ok_or(AppendError::SequenceExhausted)?;
372        let entry = Entry {
373            sequence,
374            created_at_secs: input.created_at_secs,
375            level: input.level,
376            source: String::from(input.source),
377            topic: input.topic.map(String::from),
378            message: bounded_message(input.message, self.limits.max_message_bytes),
379            truncated: input.message.len() > self.limits.max_message_bytes,
380        };
381        if self.entries.len() == self.limits.max_entries.get() {
382            self.entries.pop_front();
383        }
384        self.entries.push_back(entry);
385        self.next_sequence = sequence.checked_add(1);
386        self.last_created_at_secs = Some(input.created_at_secs);
387        Ok(sequence)
388    }
389
390    /// Remove at most `max_remove` entries strictly older than the age cutoff.
391    /// An entry remains available through its full configured age second.
392    pub fn prune_expired(
393        &mut self,
394        now_secs: u64,
395        max_age_secs: u64,
396        max_remove: usize,
397    ) -> RetentionBatch {
398        let cutoff = now_secs.saturating_sub(max_age_secs);
399        let mut removed = 0;
400        while removed < max_remove && self.expired_before(cutoff) {
401            self.entries.pop_front();
402            removed += 1;
403        }
404        RetentionBatch {
405            removed,
406            more_due: self.expired_before(cutoff),
407        }
408    }
409
410    /// First second at which the oldest row expires; `None` for empty history or overflow.
411    #[must_use]
412    pub fn next_expiry_secs(&self, max_age_secs: u64) -> Option<u64> {
413        self.entries
414            .front()?
415            .created_at_secs
416            .checked_add(max_age_secs)?
417            .checked_add(1)
418    }
419
420    /// Scan newest-first without cloning retained messages or counting every match.
421    /// Follow `next_before` with the same filters, including after an empty page.
422    #[must_use]
423    pub fn page(&self, filter: Filter<'_>, request: PageRequest) -> Page<'_> {
424        let mut cursor = request.before.map_or(self.entries.len(), |sequence| {
425            self.entries
426                .binary_search_by_key(&sequence, |entry| entry.sequence)
427                .unwrap_or_else(|index| index)
428        });
429        let mut entries = Vec::new();
430        let mut scanned = 0;
431        let mut last_scanned = None;
432        while cursor > 0
433            && scanned < request.max_scanned.get()
434            && entries.len() < request.max_results.get()
435        {
436            cursor -= 1;
437            let entry = &self.entries[cursor];
438            scanned += 1;
439            last_scanned = Some(entry.sequence);
440            if filter.matches(entry) {
441                entries.push(entry);
442            }
443        }
444        Page {
445            entries,
446            scanned,
447            next_before: if cursor > 0 { last_scanned } else { None },
448        }
449    }
450
451    const fn validate_label(&self, label: Label, text: &str) -> Result<(), AppendError> {
452        if text.len() > self.limits.max_label_bytes {
453            return Err(AppendError::LabelTooLong {
454                label,
455                actual_bytes: text.len(),
456                max_bytes: self.limits.max_label_bytes,
457            });
458        }
459        Ok(())
460    }
461
462    fn expired_before(&self, cutoff: u64) -> bool {
463        self.entries
464            .front()
465            .is_some_and(|entry| entry.created_at_secs < cutoff)
466    }
467}
468
469const fn truncation_suffix(max_bytes: usize) -> &'static str {
470    if max_bytes > TRUNCATION_SUFFIX.len() {
471        TRUNCATION_SUFFIX
472    } else {
473        ""
474    }
475}
476
477fn valid_retained_message(entry: &Entry, max_bytes: usize) -> bool {
478    if entry.message.len() > max_bytes {
479        return false;
480    }
481    if !entry.truncated {
482        return true;
483    }
484    let suffix = truncation_suffix(max_bytes);
485    let Some(prefix) = entry.message.strip_suffix(suffix) else {
486        return false;
487    };
488    let budget = max_bytes - suffix.len();
489    budget.saturating_sub(prefix.len()) < 4
490}
491
492fn bounded_message(message: &str, max_bytes: usize) -> String {
493    if message.len() <= max_bytes {
494        return String::from(message);
495    }
496    let suffix = truncation_suffix(max_bytes);
497    let mut end = max_bytes - suffix.len();
498    while !message.is_char_boundary(end) {
499        end -= 1;
500    }
501    // Reserve the final payload once, including the suffix, to avoid growth.
502    let mut retained = String::with_capacity(end + suffix.len());
503    retained.push_str(&message[..end]);
504    retained.push_str(suffix);
505    retained
506}
507
508#[cfg(test)]
509mod tests;
ic_logging/log/mod.rs

ic_logging/log/
mod.rs
