krabka_log/log/
start_checkpoint.rs

1//! When the log start reaches `log-start-offset-checkpoint`.
2//!
3//! A start that no segment boundary witnesses is written before the call that
4//! moved it returns: a `DeleteRecords` or a trim that lands inside a segment, a
5//! tiered floor below every local segment, a truncation that lowers the start.
6//! Without the file, a reopened log serves those records again.
7//!
8//! A start that retention moved is different. Retention deletes whole
9//! segments, so the base offset of the first surviving segment already records
10//! the new start. Kafka writes that value on a timer,
11//! `log.flush.start.offset.checkpoint.interval.ms`
12//! (`LogManager.checkpointLogStartOffsets`), rather than on each
13//! `deleteOldSegments`. Here the log owes the write for at most
14//! [`LogConfig::start_offset_checkpoint_interval`](crate::LogConfig::start_offset_checkpoint_interval),
15//! and the broker's writer pays it from its maintenance pass, which
16//! [`Log::maintenance_delay`] wakes when the write falls due.
17//!
18//! A crash before the write leaves an older, lower start on disk. [`Log::open`]
19//! reads it as Kafka's `LogLoader` does: on a log without remote storage the
20//! start is never below the base offset of the first segment.
21
22use {
23    super::Log,
24    crate::{error::LogError, log_start_offset_checkpoint},
25    krabka_ids::Offset,
26    krabka_units::prelude::TimeExt as _,
27    std::{
28        sync::PoisonError,
29        time::{Duration, Instant},
30    },
31};
32
33impl Log {
34    /// Write `start` to the checkpoint durably, and record that the file holds
35    /// it, which pays any write retention owed.
36    pub(super) fn write_start_checkpoint(&mut self, start: Offset) -> Result<(), LogError> {
37        log_start_offset_checkpoint::write(&*self.io, &self.dir, start)?;
38        self.checkpointed_start = Some(start);
39        self.start_checkpoint_due = None;
40        Ok(())
41    }
42
43    /// Owe the checkpoint `start`, the start retention just moved the log to.
44    ///
45    /// The deadline is set by the first change the file has not seen, so a
46    /// log whose retention keeps moving the start still writes once an
47    /// interval rather than never.
48    pub(super) fn defer_start_checkpoint(&mut self, start: Offset, now: Instant) {
49        if self.checkpointed_start == Some(start) {
50            self.start_checkpoint_due = None;
51        } else if self.start_checkpoint_due.is_none() {
52            // Kafka's setting is an `Int` of milliseconds, so no larger
53            // interval can overflow the deadline.
54            let interval = self
55                .config
56                .read()
57                .unwrap_or_else(PoisonError::into_inner)
58                .start_offset_checkpoint_interval
59                .millis_i64_trunc()
60                .clamp(0, i64::from(i32::MAX));
61            let interval = Duration::from_millis(interval.unsigned_abs());
62            self.start_checkpoint_due = Some(now + interval);
63        }
64    }
65
66    /// Record that the checkpoint file is gone and nothing is owed to it.
67    pub(super) fn forget_start_checkpoint(&mut self) {
68        self.checkpointed_start = None;
69        self.start_checkpoint_due = None;
70    }
71
72    /// Write the log start retention moved, once it has waited
73    /// [`LogConfig::start_offset_checkpoint_interval`](crate::LogConfig::start_offset_checkpoint_interval).
74    /// [`Log::maintain`] and [`Log::maintain_deferring_deletions`] call this.
75    ///
76    /// # Errors
77    /// Returns [`LogError::Io`] when the checkpoint cannot be written. The
78    /// write stays owed.
79    pub fn checkpoint_log_start_if_due(&mut self) -> Result<(), LogError> {
80        self.checkpoint_log_start_at(Instant::now())
81    }
82
83    pub(super) fn checkpoint_log_start_at(&mut self, now: Instant) -> Result<(), LogError> {
84        if self.start_checkpoint_due.is_some_and(|due| due <= now) {
85            self.write_start_checkpoint(self.start_offset)?;
86        }
87        Ok(())
88    }
89
90    /// Write an owed log start now, whatever its deadline. [`Log::sync`] calls
91    /// this, and the broker calls it when a partition's writer stops, as
92    /// Kafka's `LogManager.shutdown` checkpoints every log start. A fenced log
93    /// is being removed, so it writes nothing.
94    ///
95    /// # Errors
96    /// Returns [`LogError::Io`] when the checkpoint cannot be written.
97    pub fn flush_log_start_checkpoint(&mut self) -> Result<(), LogError> {
98        if self.start_checkpoint_due.is_none() || self.fenced {
99            return Ok(());
100        }
101        self.write_start_checkpoint(self.start_offset)
102    }
103
104    /// Time until an owed log start falls due, or `None` when nothing is owed.
105    pub(super) fn start_checkpoint_delay(&self, now: Instant) -> Option<Duration> {
106        self.start_checkpoint_due
107            .map(|due| due.saturating_duration_since(now))
108    }
109}
110
111#[cfg(test)]
112mod tests;
krabka_log/log/start_checkpoint.rs

krabka_log/log/
start_checkpoint.rs
