Struct Log 

pub struct Log { /* private fields */ }

Caller-owned heap history with fixed limits and monotonic append watermarks.

Implementations

impl Log

pub const fn new(limits: Limits) -> Self

Construct an empty log without preallocating the configured maximum.

Examples found in repository

examples/byte-budgeted-export.rs (lines 78-83)

fn history() -> Log {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(3).unwrap(),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 128,
    });
    for (time, message) in ["a", "é", "🙂"].into_iter().enumerate() {
        log.append(EntryInput {
            created_at_secs: u64::try_from(time).unwrap(),
            source: "worker",
            topic: None,
            level: Level::Info,
            message,
        })
        .unwrap();
    }
    log
}

examples/worker.rs (lines 7-12)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

pub const fn limits(&self) -> Limits

Return the fixed bounds without copying retained messages.

pub const fn retained_text_bytes(&self) -> usize

Combined retained text bytes; excludes metadata, spare capacity and allocator overhead.

pub const fn checkpoint(&self) -> LogCheckpoint<'_>

Borrow rows and append watermarks together, without copying or consuming the log.

Examples found in repository

examples/worker.rs (line 44)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

pub fn snapshot(&self) -> LogSnapshot

Copy all retained rows and both append watermarks for explicit checkpointing. Work and allocation scale with retained history, not a cursor-page limit.

pub fn into_snapshot(self) -> LogSnapshot

Consume the log into a checkpoint without cloning retained entries. Fixed limits, row order and both append watermarks are preserved, including empty histories and permanent sequence exhaustion.

pub fn restore(snapshot: LogSnapshot) -> Result<Self, RestoreError>

Admit a complete checkpoint without replaying entries or reissuing identities.

Empty retained history preserves its original append watermarks. Admission proves internal consistency, not source authenticity or host continuity. Byte decoders must bound allocation before constructing the checkpoint.

Errors

Rejects bounds, row ordering, watermarks and truncation states that cannot result from this engine’s append/retention operations.

pub fn len(&self) -> usize

Return the retained row count.

pub fn is_empty(&self) -> bool

Whether this log currently has no retained rows.

pub fn append( &mut self, input: EntryInput<'_>, ) -> Result<AppendReceipt, AppendError>

Admit one row, evicting the oldest rows until both count and text ceilings fit.

Errors

Returns a typed refusal for oversized labels, regressed time or exhausted sequence identities, or a row that cannot fit the text budget by itself. Validation completes before any eviction or mutation. Error priority is source, topic, regressed time, sequence exhaustion, then total text budget.

Examples found in repository

examples/byte-budgeted-export.rs (lines 85-91)

fn history() -> Log {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(3).unwrap(),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 128,
    });
    for (time, message) in ["a", "é", "🙂"].into_iter().enumerate() {
        log.append(EntryInput {
            created_at_secs: u64::try_from(time).unwrap(),
            source: "worker",
            topic: None,
            level: Level::Info,
            message,
        })
        .unwrap();
    }
    log
}

examples/worker.rs (lines 22-28)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

pub fn prune_expired( &mut self, now_secs: u64, max_age_secs: u64, max_remove: usize, ) -> RetentionBatch

Remove at most max_remove entries strictly older than the age cutoff. An entry remains available through its full configured age second.

Examples found in repository

examples/worker.rs (line 45)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

pub fn next_expiry_secs(&self, max_age_secs: u64) -> Option<u64>

First second at which the oldest row expires; None for empty history or overflow.

Examples found in repository

examples/worker.rs (line 47)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

pub fn page(&self, filter: Filter<'_>, request: PageRequest) -> Page<'_>

Scan newest-first without cloning retained messages or counting every match. Follow next_before with the same filters, including after an empty page.

pub fn visit_page<'a>( &'a self, filter: Filter<'_>, request: PageRequest, visit: impl FnMut(&'a Entry) -> ControlFlow<()>, ) -> PageScan

Visit matching rows without allocating a page collection or cloning text. Break stops after consuming that matched row; the returned cursor resumes at the next older candidate. Empty filtered scans can still have a cursor. Callback work/allocation belongs to the caller; filter/limits match page.

Examples found in repository

examples/worker.rs (lines 31-42)

fn main() {
    let mut log = Log::new(Limits {
        max_entries: NonZeroUsize::new(8).expect("positive capacity"),
        max_message_bytes: 32,
        max_label_bytes: 16,
        max_text_bytes: 512,
    });
    for (created_at_secs, level, message) in [
        (10, Level::Info, "delivery accepted"),
        (
            11,
            Level::Warn,
            "provider reply is still unknown; reconcile before retry",
        ),
        (12, Level::Ok, "original delivery reconciled"),
    ] {
        log.append(EntryInput {
            created_at_secs,
            level,
            source: "worker",
            topic: Some("delivery"),
            message,
        })
        .expect("valid bounded entry");
    }
    let scan = log.visit_page(
        Filter::default(),
        PageRequest {
            before: None,
            max_results: NonZeroUsize::new(8).expect("positive results"),
            max_scanned: NonZeroUsize::new(8).expect("positive scan"),
        },
        |entry| {
            println!("{} {:?}: {}", entry.sequence, entry.level, entry.message);
            ControlFlow::Continue(())
        },
    );
    assert_eq!(scan.matched, 3);
    assert_eq!(log.checkpoint().next_sequence(), Some(3));
    let batch = log.prune_expired(16, 5, 1);
    assert_eq!(batch.removed, 1);
    assert_eq!(log.next_expiry_secs(5), Some(17));
}

examples/byte-budgeted-export.rs (lines 35-53)

fn export(
    log: &Log,
    filter: Filter<'_>,
    request: PageRequest,
    max_bytes: usize,
) -> Result<ExportPage, ExportError> {
    let mut body = String::new();
    let mut emitted = 0;
    let mut last_emitted = None;
    let mut stopped_for_budget = Ok(false);
    let scan = log.visit_page(filter, request, |entry| {
        let encoded = format!("{}: {}\n", entry.sequence, entry.message);
        if encoded.len() > max_bytes.saturating_sub(body.len()) {
            stopped_for_budget = if emitted == 0 {
                Err(ExportError::EntryTooLarge {
                    sequence: entry.sequence,
                    required_bytes: encoded.len(),
                    max_bytes,
                })
            } else {
                Ok(true)
            };
            return ControlFlow::Break(());
        }
        body.push_str(&encoded);
        emitted += 1;
        last_emitted = Some(entry.sequence);
        ControlFlow::Continue(())
    });
    let stopped_for_budget = stopped_for_budget?;
    Ok(ExportPage {
        body,
        emitted,
        scanned: scan.scanned,
        // Break consumes its candidate in the engine. The candidate that did
        // not fit has not been emitted, so resume from the last delivered row.
        next_before: if stopped_for_budget {
            last_emitted
        } else {
            scan.next_before
        },
    })
}

Trait Implementations

impl Debug for Log

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.