1use alloc::{collections::VecDeque, string::String, vec::Vec};
6use core::{num::NonZeroUsize, ops::ControlFlow};
7
8const TRUNCATION_SUFFIX: &str = "...[truncated]";
9
10#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
12pub enum Level {
13 Debug,
15 Info,
17 Ok,
19 Warn,
21 Error,
23}
24
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
27pub struct Limits {
28 pub max_entries: NonZeroUsize,
30 pub max_message_bytes: usize,
32 pub max_label_bytes: usize,
34 pub max_text_bytes: usize,
37}
38
39#[derive(Clone, Copy, Debug)]
41pub struct EntryInput<'a> {
42 pub created_at_secs: u64,
44 pub level: Level,
46 pub source: &'a str,
48 pub topic: Option<&'a str>,
50 pub message: &'a str,
52}
53
54#[derive(Clone, Debug, Eq, PartialEq)]
56pub struct Entry {
57 pub sequence: u64,
59 pub created_at_secs: u64,
61 pub level: Level,
63 pub source: String,
65 pub topic: Option<String>,
67 pub message: String,
69 pub truncated: bool,
71}
72
73#[derive(Clone, Copy, Debug, Eq, PartialEq)]
75pub enum Label {
76 Source,
78 Topic,
80}
81
82#[derive(Clone, Copy, Debug, Eq, PartialEq)]
84pub enum AppendError {
85 LabelTooLong {
87 label: Label,
89 actual_bytes: usize,
91 max_bytes: usize,
93 },
94 TimestampRegressed {
96 previous: u64,
98 current: u64,
100 },
101 SequenceExhausted,
103 TextBudgetExceeded {
105 required_bytes: u128,
107 max_bytes: usize,
109 },
110}
111
112impl core::fmt::Display for AppendError {
113 fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
114 match self {
115 Self::LabelTooLong {
116 label,
117 actual_bytes,
118 max_bytes,
119 } => {
120 let label = match label {
121 Label::Source => "source",
122 Label::Topic => "topic",
123 };
124 write!(
125 formatter,
126 "log {label} label exceeds {max_bytes} bytes ({actual_bytes} supplied)"
127 )
128 }
129 Self::TimestampRegressed { previous, current } => write!(
130 formatter,
131 "log timestamp regressed from {previous} to {current} seconds"
132 ),
133 Self::SequenceExhausted => formatter.write_str("log sequence identities are exhausted"),
134 Self::TextBudgetExceeded {
135 required_bytes,
136 max_bytes,
137 } => write!(
138 formatter,
139 "log row requires {required_bytes} text bytes (limit {max_bytes})"
140 ),
141 }
142 }
143}
144
145impl core::error::Error for AppendError {}
146
147#[derive(Clone, Copy, Debug, Eq, PartialEq)]
149pub struct AppendReceipt {
150 pub sequence: u64,
152 pub truncated: bool,
154 pub evicted: usize,
156}
157
158#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
160pub enum TopicMatch<'a> {
161 #[default]
163 Any,
164 Absent,
166 Exact(&'a str),
168}
169
170#[derive(Clone, Copy, Debug, Default)]
172pub struct Filter<'a> {
173 pub source: Option<&'a str>,
175 pub topic: TopicMatch<'a>,
177 pub min_level: Option<Level>,
179}
180
181impl Filter<'_> {
182 fn matches(self, entry: &Entry) -> bool {
183 let source_matches = self.source.is_none_or(|source| source == entry.source);
184 let topic_matches = match self.topic {
185 TopicMatch::Any => true,
186 TopicMatch::Absent => entry.topic.is_none(),
187 TopicMatch::Exact(topic) => entry.topic.as_deref() == Some(topic),
188 };
189 let level_matches = self.min_level.is_none_or(|level| entry.level >= level);
190 source_matches && topic_matches && level_matches
191 }
192}
193
194#[derive(Clone, Copy, Debug)]
196pub struct PageRequest {
197 pub before: Option<u64>,
199 pub max_results: NonZeroUsize,
201 pub max_scanned: NonZeroUsize,
203}
204
205#[derive(Debug)]
207pub struct Page<'a> {
208 pub entries: Vec<&'a Entry>,
210 pub scanned: usize,
212 pub next_before: Option<u64>,
214}
215
216#[derive(Clone, Copy, Debug, Eq, PartialEq)]
218pub struct PageScan {
219 pub matched: usize,
221 pub scanned: usize,
223 pub next_before: Option<u64>,
225}
226
227#[derive(Clone, Copy, Debug)]
230pub struct LogCheckpoint<'a> {
231 limits: Limits,
232 entries: &'a VecDeque<Entry>,
233 next_sequence: Option<u64>,
234 last_created_at_secs: Option<u64>,
235}
236
237impl<'a> LogCheckpoint<'a> {
238 #[must_use]
240 pub const fn limits(&self) -> Limits {
241 self.limits
242 }
243
244 pub fn entries(&self) -> impl ExactSizeIterator<Item = &'a Entry> + DoubleEndedIterator + 'a {
246 self.entries.iter()
247 }
248
249 #[must_use]
251 pub const fn next_sequence(&self) -> Option<u64> {
252 self.next_sequence
253 }
254
255 #[must_use]
257 pub const fn last_created_at_secs(&self) -> Option<u64> {
258 self.last_created_at_secs
259 }
260}
261
262#[derive(Clone, Copy, Debug, Eq, PartialEq)]
264pub struct RetentionBatch {
265 pub removed: usize,
267 pub more_due: bool,
269}
270
271#[derive(Clone, Debug, Eq, PartialEq)]
276pub struct LogSnapshot {
277 pub limits: Limits,
279 pub entries: Vec<Entry>,
281 pub next_sequence: Option<u64>,
283 pub last_created_at_secs: Option<u64>,
285}
286
287#[derive(Clone, Copy, Debug, Eq, PartialEq)]
289pub enum RestoreError {
290 TooManyEntries,
292 InvalidSequence,
294 InvalidTimestamp,
296 InvalidEntry,
298 TextBudgetExceeded,
300}
301
302impl core::fmt::Display for RestoreError {
303 fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
304 formatter.write_str(match self {
305 Self::TooManyEntries => "log checkpoint exceeds its entry limit",
306 Self::InvalidSequence => "invalid log checkpoint sequence",
307 Self::InvalidTimestamp => "invalid log checkpoint timestamp",
308 Self::InvalidEntry => "invalid bounded log checkpoint entry",
309 Self::TextBudgetExceeded => "log checkpoint exceeds its text budget",
310 })
311 }
312}
313
314impl core::error::Error for RestoreError {}
315
316#[derive(Debug)]
318pub struct Log {
319 limits: Limits,
320 entries: VecDeque<Entry>,
321 retained_text_bytes: usize,
322 next_sequence: Option<u64>,
323 last_created_at_secs: Option<u64>,
324}
325
326impl Log {
327 #[must_use]
329 pub const fn new(limits: Limits) -> Self {
330 Self {
331 limits,
332 entries: VecDeque::new(),
333 retained_text_bytes: 0,
334 next_sequence: Some(0),
335 last_created_at_secs: None,
336 }
337 }
338
339 #[must_use]
341 pub const fn limits(&self) -> Limits {
342 self.limits
343 }
344
345 #[must_use]
347 pub const fn retained_text_bytes(&self) -> usize {
348 self.retained_text_bytes
349 }
350
351 #[must_use]
353 pub const fn checkpoint(&self) -> LogCheckpoint<'_> {
354 LogCheckpoint {
355 limits: self.limits,
356 entries: &self.entries,
357 next_sequence: self.next_sequence,
358 last_created_at_secs: self.last_created_at_secs,
359 }
360 }
361
362 #[must_use]
365 pub fn snapshot(&self) -> LogSnapshot {
366 LogSnapshot {
367 limits: self.limits,
368 entries: self.entries.iter().cloned().collect(),
369 next_sequence: self.next_sequence,
370 last_created_at_secs: self.last_created_at_secs,
371 }
372 }
373
374 #[must_use]
378 pub fn into_snapshot(self) -> LogSnapshot {
379 LogSnapshot {
380 limits: self.limits,
381 entries: self.entries.into(),
382 next_sequence: self.next_sequence,
383 last_created_at_secs: self.last_created_at_secs,
384 }
385 }
386
387 pub fn restore(snapshot: LogSnapshot) -> Result<Self, RestoreError> {
397 let retained_text_bytes = validate_snapshot(&snapshot)?;
398 let LogSnapshot {
399 limits,
400 entries,
401 next_sequence,
402 last_created_at_secs,
403 } = snapshot;
404 Ok(Self {
405 limits,
406 entries: entries.into(),
407 retained_text_bytes,
408 next_sequence,
409 last_created_at_secs,
410 })
411 }
412
413 #[must_use]
415 pub fn len(&self) -> usize {
416 self.entries.len()
417 }
418
419 #[must_use]
421 pub fn is_empty(&self) -> bool {
422 self.entries.is_empty()
423 }
424
425 pub fn append(&mut self, input: EntryInput<'_>) -> Result<AppendReceipt, AppendError> {
433 self.validate_label(Label::Source, input.source)?;
434 if let Some(topic) = input.topic {
435 self.validate_label(Label::Topic, topic)?;
436 }
437 if let Some(previous) = self.last_created_at_secs
438 && input.created_at_secs < previous
439 {
440 return Err(AppendError::TimestampRegressed {
441 previous,
442 current: input.created_at_secs,
443 });
444 }
445 let sequence = self.next_sequence.ok_or(AppendError::SequenceExhausted)?;
446 let (prefix, suffix) = message_parts(input.message, self.limits.max_message_bytes);
447 let required_bytes = input.source.len() as u128
448 + input.topic.map_or(0, str::len) as u128
449 + prefix.len() as u128
450 + suffix.len() as u128;
451 let text_bytes = usize::try_from(required_bytes)
452 .ok()
453 .filter(|bytes| *bytes <= self.limits.max_text_bytes)
454 .ok_or(AppendError::TextBudgetExceeded {
455 required_bytes,
456 max_bytes: self.limits.max_text_bytes,
457 })?;
458 let truncated = input.message.len() > self.limits.max_message_bytes;
459 let entry = Entry {
460 sequence,
461 created_at_secs: input.created_at_secs,
462 level: input.level,
463 source: String::from(input.source),
464 topic: input.topic.map(String::from),
465 message: bounded_message(prefix, suffix),
466 truncated,
467 };
468 let mut evicted = 0;
469 while (self.entries.len() >= self.limits.max_entries.get()
470 || self.retained_text_bytes > self.limits.max_text_bytes - text_bytes)
471 && self.remove_oldest()
472 {
473 evicted += 1;
474 }
475 self.entries.push_back(entry);
476 self.retained_text_bytes += text_bytes;
477 self.next_sequence = sequence.checked_add(1);
478 self.last_created_at_secs = Some(input.created_at_secs);
479 Ok(AppendReceipt {
480 sequence,
481 truncated,
482 evicted,
483 })
484 }
485
486 pub fn prune_expired(
489 &mut self,
490 now_secs: u64,
491 max_age_secs: u64,
492 max_remove: usize,
493 ) -> RetentionBatch {
494 let cutoff = now_secs.saturating_sub(max_age_secs);
495 let mut removed = 0;
496 while removed < max_remove && self.expired_before(cutoff) {
497 self.remove_oldest();
498 removed += 1;
499 }
500 RetentionBatch {
501 removed,
502 more_due: self.expired_before(cutoff),
503 }
504 }
505
506 #[must_use]
508 pub fn next_expiry_secs(&self, max_age_secs: u64) -> Option<u64> {
509 self.entries
510 .front()?
511 .created_at_secs
512 .checked_add(max_age_secs)?
513 .checked_add(1)
514 }
515
516 #[must_use]
519 pub fn page(&self, filter: Filter<'_>, request: PageRequest) -> Page<'_> {
520 let mut entries = Vec::new();
521 let scan = self.visit_page(filter, request, |entry| {
522 entries.push(entry);
523 ControlFlow::Continue(())
524 });
525 Page {
526 entries,
527 scanned: scan.scanned,
528 next_before: scan.next_before,
529 }
530 }
531
532 pub fn visit_page<'a>(
537 &'a self,
538 filter: Filter<'_>,
539 request: PageRequest,
540 mut visit: impl FnMut(&'a Entry) -> ControlFlow<()>,
541 ) -> PageScan {
542 let mut cursor = request.before.map_or(self.entries.len(), |sequence| {
543 self.entries
544 .binary_search_by_key(&sequence, |entry| entry.sequence)
545 .unwrap_or_else(|index| index)
546 });
547 let mut matched = 0;
548 let mut scanned = 0;
549 let mut last_scanned = None;
550 while cursor > 0
551 && scanned < request.max_scanned.get()
552 && matched < request.max_results.get()
553 {
554 cursor -= 1;
555 let entry = &self.entries[cursor];
556 scanned += 1;
557 last_scanned = Some(entry.sequence);
558 if filter.matches(entry) {
559 matched += 1;
560 if visit(entry).is_break() {
561 break;
562 }
563 }
564 }
565 PageScan {
566 matched,
567 scanned,
568 next_before: if cursor > 0 { last_scanned } else { None },
569 }
570 }
571
572 const fn validate_label(&self, label: Label, text: &str) -> Result<(), AppendError> {
573 if text.len() > self.limits.max_label_bytes {
574 return Err(AppendError::LabelTooLong {
575 label,
576 actual_bytes: text.len(),
577 max_bytes: self.limits.max_label_bytes,
578 });
579 }
580 Ok(())
581 }
582
583 fn expired_before(&self, cutoff: u64) -> bool {
584 self.entries
585 .front()
586 .is_some_and(|entry| entry.created_at_secs < cutoff)
587 }
588
589 fn remove_oldest(&mut self) -> bool {
590 let Some(entry) = self.entries.pop_front() else {
591 return false;
592 };
593 self.retained_text_bytes -= entry.source.len();
595 self.retained_text_bytes -= entry.topic.as_ref().map_or(0, String::len);
596 self.retained_text_bytes -= entry.message.len();
597 true
598 }
599}
600
601fn validate_snapshot(snapshot: &LogSnapshot) -> Result<usize, RestoreError> {
602 let LogSnapshot {
603 limits,
604 entries,
605 next_sequence,
606 last_created_at_secs,
607 } = snapshot;
608 let (next_sequence, last_created_at_secs) = (*next_sequence, *last_created_at_secs);
609 if entries.len() > limits.max_entries.get() {
610 return Err(RestoreError::TooManyEntries);
611 }
612 if (next_sequence == Some(0)) != last_created_at_secs.is_none() {
613 return Err(RestoreError::InvalidTimestamp);
614 }
615 if entries.is_empty() && last_created_at_secs == Some(u64::MAX) {
618 return Err(RestoreError::InvalidTimestamp);
619 }
620 let mut previous: Option<&Entry> = None;
621 let mut retained_text_bytes = 0_usize;
622 for entry in entries {
623 if entry.source.len() > limits.max_label_bytes
624 || entry
625 .topic
626 .as_ref()
627 .is_some_and(|topic| topic.len() > limits.max_label_bytes)
628 || !valid_retained_message(entry, limits.max_message_bytes)
629 {
630 return Err(RestoreError::InvalidEntry);
631 }
632 for bytes in [
633 entry.source.len(),
634 entry.topic.as_ref().map_or(0, String::len),
635 entry.message.len(),
636 ] {
637 retained_text_bytes = retained_text_bytes
638 .checked_add(bytes)
639 .filter(|total| *total <= limits.max_text_bytes)
640 .ok_or(RestoreError::TextBudgetExceeded)?;
641 }
642 if let Some(previous) = previous {
643 if previous.sequence.checked_add(1) != Some(entry.sequence) {
644 return Err(RestoreError::InvalidSequence);
645 }
646 if entry.created_at_secs < previous.created_at_secs {
647 return Err(RestoreError::InvalidTimestamp);
648 }
649 }
650 previous = Some(entry);
651 }
652 if let Some(last) = previous {
653 if last.sequence.checked_add(1) != next_sequence {
654 return Err(RestoreError::InvalidSequence);
655 }
656 if Some(last.created_at_secs) != last_created_at_secs {
657 return Err(RestoreError::InvalidTimestamp);
658 }
659 }
660 Ok(retained_text_bytes)
661}
662
663const fn truncation_suffix(max_bytes: usize) -> &'static str {
664 if max_bytes > TRUNCATION_SUFFIX.len() {
665 TRUNCATION_SUFFIX
666 } else {
667 ""
668 }
669}
670
671fn valid_retained_message(entry: &Entry, max_bytes: usize) -> bool {
672 if entry.message.len() > max_bytes {
673 return false;
674 }
675 if !entry.truncated {
676 return true;
677 }
678 let suffix = truncation_suffix(max_bytes);
679 let Some(prefix) = entry.message.strip_suffix(suffix) else {
680 return false;
681 };
682 let budget = max_bytes - suffix.len();
683 budget.saturating_sub(prefix.len()) < 4
684}
685
686fn bounded_message(prefix: &str, suffix: &str) -> String {
687 let mut retained = String::with_capacity(prefix.len() + suffix.len());
688 retained.push_str(prefix);
689 retained.push_str(suffix);
690 retained
691}
692
693fn message_parts(message: &str, max_bytes: usize) -> (&str, &'static str) {
694 if message.len() <= max_bytes {
695 return (message, "");
696 }
697 let suffix = truncation_suffix(max_bytes);
698 let mut end = max_bytes - suffix.len();
699 while !message.is_char_boundary(end) {
700 end -= 1;
701 }
702 (&message[..end], suffix)
703}
704
705#[cfg(test)]
706mod tests;