krabka_log/
retention.rs

1//! Retention that `Log::tick` applies. These are free functions, so tests can
2//! check the policy apart from `Log`'s mutable state.
3
4use {
5    crate::{
6        error::LogError,
7        io::{IoTarget, LogIo},
8        name,
9    },
10    krabka_ids::Offset,
11    std::{
12        io::ErrorKind,
13        path::{Path, PathBuf},
14        sync::atomic::{AtomicU64, Ordering},
15        time::{Duration, SystemTime, UNIX_EPOCH},
16    },
17    tracing::instrument,
18};
19
20/// Suffix a segment file wears between the moment retention claims it and the
21/// moment it is unlinked, mirroring Kafka's `.deleted` rename.
22pub const DELETED_SUFFIX: &str = "deleted";
23
24/// The tombstone name for `path`: the same name with `.deleted` appended.
25pub fn deleted_path(path: &Path) -> PathBuf {
26    let mut name = path.as_os_str().to_os_string();
27    name.push(".");
28    name.push(DELETED_SUFFIX);
29    PathBuf::from(name)
30}
31
32/// `now` in epoch milliseconds. An instant before the epoch reads as `0`, and
33/// one past `i64::MAX` milliseconds saturates there.
34#[must_use]
35pub fn now_ms(now: SystemTime) -> i64 {
36    let millis = now
37        .duration_since(UNIX_EPOCH)
38        .unwrap_or(Duration::ZERO)
39        .as_millis();
40    i64::try_from(millis).unwrap_or(i64::MAX)
41}
42
43/// Remove one segment's whole file set: the `.log`, both sparse indexes, and
44/// the optional `.txnindex`, `.stampindex` and `.snapshot` sidecars.
45///
46/// The producer snapshot is written at this segment's base offset on every
47/// roll (see `Log::roll_active_segment`), so it belongs to this segment the
48/// same way its `.txnindex` does. Kafka's `UnifiedLog.deleteSegments` deletes
49/// it alongside the segment (`deleteProducerSnapshots`); leaving it behind
50/// piles up `.snapshot` files for the life of the partition, and every
51/// `Log::open` lists and sorts all of them.
52///
53/// Every file is first renamed to a `<name>.deleted` tombstone and only then
54/// unlinked, the way Kafka's `deleteSegments` renames before it deletes. A
55/// failure part-way through therefore leaves names that say what they are: a
56/// tombstone nothing reads, or a file still under its live name. Neither is a
57/// segment whose `.log` is gone and whose sidecars are invisible to the
58/// `.log`-keyed directory scan in [`crate::Log::open`], and
59/// [`crate::recovery::deleted_orphan_recover`] reclaims both on the next open.
60///
61/// # Errors
62/// Returns the first rename or unlink error. A missing file is not one: a
63/// segment need not carry any of the optional sidecars, and a retried
64/// deletion finds part of the set already gone.
65#[instrument(level = "debug", skip_all, fields(dir = %dir.display(), base_offset = base_offset.0), err)]
66pub fn delete_segment_files(
67    io: &dyn LogIo,
68    dir: &Path,
69    base_offset: Offset,
70) -> Result<(), LogError> {
71    for tombstone in retire_segment_files(io, dir, base_offset)? {
72        remove_optional(io, &tombstone)?;
73    }
74    Ok(())
75}
76
77/// Rename a removed segment out of the live namespace, retaining its files for open readers.
78pub fn retire_segment_files(
79    io: &dyn LogIo,
80    dir: &Path,
81    base_offset: Offset,
82) -> Result<Vec<PathBuf>, LogError> {
83    let mut tombstones = Vec::with_capacity(6);
84    for path in [
85        name::log_path(dir, base_offset.0),
86        name::index_path(dir, base_offset.0),
87        name::timeindex_path(dir, base_offset.0),
88        name::txnindex_path(dir, base_offset.0),
89        name::stampindex_path(dir, base_offset.0),
90        name::producer_snapshot_path(dir, base_offset.0),
91    ] {
92        let tombstone = deleted_path(&path);
93        match io.rename(IoTarget::SegmentDeletion, &path, &tombstone) {
94            Ok(()) => tombstones.push(tombstone),
95            // Absent under its live name: either this segment never had the
96            // sidecar, or an earlier attempt already renamed it.
97            Err(error) if error.kind() == ErrorKind::NotFound => {
98                if tombstone.exists() {
99                    tombstones.push(tombstone);
100                }
101            }
102            Err(error) => return Err(LogError::Io(error)),
103        }
104    }
105    Ok(tombstones)
106}
107
108/// When [`retire_files`] makes its renames durable.
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110pub enum DirSync {
111    /// Sync the directory before returning.
112    Now,
113    /// Leave the sync to the caller, which syncs the directory once after a
114    /// run of retirements and before anything depends on them.
115    Deferred,
116}
117
118/// Retire a file set under unique names, so repeated compactions of the same
119/// base cannot replace a file still retained for readers. Failed retirement restores
120/// successful renames, and syncs the directory, before returning the error.
121pub fn retire_files(
122    io: &dyn LogIo,
123    dir: &Path,
124    base: Offset,
125    extensions: &[&str],
126    retired: &mut Vec<PathBuf>,
127    sync: DirSync,
128) -> Result<(), LogError> {
129    static GENERATION: AtomicU64 = AtomicU64::new(0);
130    let generation = GENERATION.fetch_add(1, Ordering::Relaxed);
131    let mut renamed = Vec::new();
132    let result = (|| {
133        for extension in extensions {
134            let path = dir.join(format!("{}.{extension}", name::format_base_offset(base.0)));
135            let tombstone = dir.join(format!(
136                "{}.{extension}.{generation}.deleted",
137                name::format_base_offset(base.0)
138            ));
139            match io.rename(IoTarget::SegmentDeletion, &path, &tombstone) {
140                Ok(()) => renamed.push((path, tombstone)),
141                Err(error) if error.kind() == ErrorKind::NotFound => {}
142                Err(error) => return Err(LogError::Io(error)),
143            }
144        }
145        if sync == DirSync::Now {
146            io.sync_dir(dir)?;
147        }
148        Ok(())
149    })();
150    if let Err(error) = result {
151        // The caller still owns these segments until retirement succeeds.
152        // Restore their live names rather than queueing referenced files for deletion.
153        for (path, tombstone) in renamed.iter().rev() {
154            io.rename(IoTarget::SegmentDeletion, tombstone, path)?;
155        }
156        io.sync_dir(dir)?;
157        return Err(error);
158    }
159    retired.extend(renamed.into_iter().map(|(_, tombstone)| tombstone));
160    Ok(())
161}
162
163pub fn remove_optional(io: &dyn LogIo, path: &Path) -> Result<(), LogError> {
164    remove_optional_counted(io, path).map(|_| ())
165}
166
167/// Remove an optional file and report whether it existed.
168pub fn remove_optional_counted(io: &dyn LogIo, path: &Path) -> Result<bool, LogError> {
169    match io.remove_file(IoTarget::SegmentDeletion, path) {
170        Ok(()) => Ok(true),
171        Err(error) if error.kind() == ErrorKind::NotFound => Ok(false),
172        Err(error) => Err(LogError::Io(error)),
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use {
179        super::*,
180        crate::io::FileIo,
181        assert2::check,
182        std::{
183            fs,
184            io::{self, ErrorKind},
185            path::PathBuf,
186        },
187        tempfile::tempdir,
188    };
189
190    /// A temporary directory and the segment files written into it.
191    struct SegmentFiles {
192        dir: tempfile::TempDir,
193        paths: Vec<PathBuf>,
194    }
195
196    fn segment_files(base: Offset, extensions: &[&str], contents: &[u8]) -> SegmentFiles {
197        let dir = tempdir().unwrap();
198        let paths = extensions
199            .iter()
200            .map(|extension| {
201                dir.path()
202                    .join(format!("{}.{extension}", name::format_base_offset(base.0)))
203            })
204            .collect::<Vec<_>>();
205        for path in &paths {
206            fs::write(path, contents).unwrap();
207        }
208        SegmentFiles { dir, paths }
209    }
210
211    #[test]
212    fn delete_segment_files_removes_required_and_optional_sidecars() {
213        let base = Offset(7);
214        let SegmentFiles { dir, paths } = segment_files(
215            base,
216            &[
217                "log",
218                "index",
219                "timeindex",
220                "txnindex",
221                "stampindex",
222                "snapshot",
223            ],
224            &[],
225        );
226
227        delete_segment_files(&FileIo, dir.path(), base).unwrap();
228
229        assert2::assert!(paths.iter().all(|path| !path.exists()));
230    }
231
232    #[test]
233    fn delete_segment_files_accepts_missing_optional_sidecars() {
234        let base = Offset(8);
235        let SegmentFiles { dir, .. } = segment_files(base, &["log", "index", "timeindex"], &[]);
236
237        delete_segment_files(&FileIo, dir.path(), base).unwrap();
238    }
239
240    /// Every file is renamed to its `.deleted` tombstone before any of them is
241    /// unlinked, so a failure part-way through leaves names that say what they
242    /// are rather than a `.log` that is gone and sidecars nothing can see.
243    #[test]
244    fn deletion_renames_every_file_to_a_tombstone_before_it_unlinks_any() {
245        /// Lets every rename through and refuses every unlink.
246        #[derive(Debug)]
247        struct NoUnlink;
248
249        impl LogIo for NoUnlink {
250            fn remove_file(&self, _target: IoTarget, _path: &Path) -> io::Result<()> {
251                Err(ErrorKind::PermissionDenied.into())
252            }
253        }
254
255        let base = Offset(9);
256        let SegmentFiles { dir, paths: live } =
257            segment_files(base, &["log", "index", "timeindex", "stampindex"], b"bytes");
258
259        let error = delete_segment_files(&NoUnlink, dir.path(), base)
260            .expect_err("the refused unlink must be reported");
261
262        check!(matches!(error, LogError::Io(_)));
263        for path in &live {
264            check!(!path.exists(), "renamed away: {}", path.display());
265            check!(
266                deleted_path(path).exists(),
267                "tombstoned: {}",
268                deleted_path(path).display()
269            );
270        }
271
272        // The retry finds the live names gone and the tombstones present, and
273        // finishes the job from there.
274        delete_segment_files(&FileIo, dir.path(), base).unwrap();
275        for path in &live {
276            check!(!deleted_path(path).exists());
277        }
278    }
279
280    #[test]
281    fn remove_optional_propagates_non_missing_errors() {
282        let dir = tempdir().unwrap();
283        let path = dir.path().join("not-a-file");
284        fs::create_dir(&path).unwrap();
285
286        assert2::assert!(let Err(LogError::Io(_)) = remove_optional(&FileIo, &path));
287    }
288
289    #[test]
290    fn remove_optional_ignores_missing_file() {
291        let dir = tempdir().unwrap();
292        let path = dir.path().join("nonexistent-file");
293        assert2::assert!(remove_optional(&FileIo, &path).is_ok());
294    }
295}
krabka_log/retention.rs

krabka_log/
retention.rs
