krabka_backup/
capture.rs

1//! Finding the two restore inputs that live on a node's disk, and naming the
2//! capture they are written into.
3//!
4//! Neither file has a stable name. The RLMM snapshot does — it is always
5//! `<log.dir>/remote-log-metadata/snapshot` — but the controller metadata
6//! checkpoint is `<end-offset>-<epoch>.checkpoint` and there are several of
7//! them, in one of two directories under the metadata log directory,
8//! depending on whether the node runs a controller. Picking the right one is
9//! this module's whole job, and it is a pure function over directory listings
10//! so a test can pin the choice without a broker.
11
12use {
13    crate::manifest::CAPTURE_ROOT,
14    krabka_metadata::MetadataRecord,
15    std::{
16        fs,
17        path::{Path, PathBuf},
18    },
19    uuid::Uuid,
20};
21
22/// The RLMM snapshot's path under a log directory.
23pub const RLMM_SNAPSHOT_RELATIVE: &str = "remote-log-metadata/snapshot";
24
25/// Where a node that runs a controller keeps its metadata checkpoints, under
26/// the metadata log directory: Kafka's metadata partition directory.
27pub const CONTROLLER_CHECKPOINT_DIR: &str = "__cluster_metadata-0";
28
29/// Where a broker-only node keeps the checkpoints its metadata observer wrote,
30/// under the metadata log directory.
31///
32/// They sit in a subdirectory of `__cluster_metadata-0` and never directly in
33/// it, because a controller must not load one.
34pub const OBSERVER_CHECKPOINT_DIR: &str = "__cluster_metadata-0/observer";
35
36/// Suffix of a KIP-630 checkpoint artifact.
37const CHECKPOINT_SUFFIX: &str = ".checkpoint";
38
39/// Width of the end-offset field in a checkpoint name.
40const OFFSET_WIDTH: usize = 20;
41
42/// Width of the epoch field in a checkpoint name.
43const EPOCH_WIDTH: usize = 10;
44
45/// The identity of a metadata checkpoint, as its file name encodes it.
46///
47/// The derived order is the checkpoint order: end offset first, then epoch.
48#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
49pub struct CheckpointId {
50    /// The offset the checkpoint ends at.
51    pub end_offset: u64,
52    /// The epoch of the controller that wrote it.
53    pub epoch: u64,
54}
55
56/// The [`CheckpointId`] a checkpoint file name encodes, or `None` when the
57/// name is not a checkpoint.
58///
59/// The encoding is `<end_offset:020>-<epoch:010>.checkpoint`, which
60/// `krabka-raft`'s `checkpoint_name` writes. A name that parses to numbers but
61/// does not re-encode to itself is rejected, so a hand-made `1-0.checkpoint`
62/// is not mistaken for an artifact the controller wrote.
63#[must_use]
64pub fn checkpoint_id(name: &str) -> Option<CheckpointId> {
65    let stem = name.strip_suffix(CHECKPOINT_SUFFIX)?;
66    let (offset, epoch) = stem.split_once('-')?;
67    if offset.len() != OFFSET_WIDTH || epoch.len() != EPOCH_WIDTH {
68        return None;
69    }
70    Some(CheckpointId {
71        end_offset: offset.parse().ok()?,
72        epoch: epoch.parse().ok()?,
73    })
74}
75
76/// The newest checkpoint among `names`, by `(end_offset, epoch)`.
77///
78/// The end offset orders first: a checkpoint taken further along the metadata
79/// log holds more of it, whatever epoch wrote it.
80#[must_use]
81pub fn newest_checkpoint(names: &[String]) -> Option<String> {
82    names
83        .iter()
84        .filter_map(|name| checkpoint_id(name).map(|id| (id, name)))
85        .max_by_key(|(id, _)| *id)
86        .map(|(_, name)| name.clone())
87}
88
89/// Every file name directly inside `dir`. An absent or unreadable directory is
90/// an empty list: a node can lack either checkpoint directory, and that is not
91/// a failure.
92#[must_use]
93pub fn file_names(dir: &Path) -> Vec<String> {
94    let Ok(entries) = fs::read_dir(dir) else {
95        return Vec::new();
96    };
97    entries
98        .flatten()
99        .filter_map(|entry| entry.file_name().to_str().map(ToOwned::to_owned))
100        .collect()
101}
102
103impl CheckpointId {
104    /// Whether this is the checkpoint `krabka format` writes, at end offset
105    /// zero.
106    ///
107    /// A plain format puts the bootstrap records only in it, and never a
108    /// topic. A format that `krabka-restore` runs also seeds the restored
109    /// topics into it, so [`holds_a_topic`] tells the two apart. The controller
110    /// writes its first checkpoint past it only after
111    /// `metadata_max_snapshot_interval`, or once enough metadata is committed,
112    /// so until then it is the newest checkpoint on the node.
113    #[must_use]
114    pub const fn is_format_time(self) -> bool { self.end_offset == 0 }
115}
116
117/// Whether a checkpoint's decoded records hold a topic.
118///
119/// The format-time checkpoint of a plain `krabka format` holds none. One that
120/// `krabka-restore` wrote holds the restored topics, and a controller's
121/// periodic checkpoint holds every topic the cluster had.
122#[must_use]
123pub fn holds_a_topic(records: &[MetadataRecord]) -> bool {
124    records
125        .iter()
126        .any(|record| matches!(record, MetadataRecord::V1Topic(_)))
127}
128
129/// A metadata checkpoint file and the id its name encodes.
130#[derive(Debug, Clone, PartialEq, Eq)]
131pub struct MetadataCheckpoint {
132    /// The id the file name encodes.
133    pub id: CheckpointId,
134    /// The file.
135    pub path: PathBuf,
136}
137
138/// The newest metadata checkpoint under `log_dir`, the node's metadata log
139/// directory, looking in the controller's directory and then in the
140/// observer's.
141///
142/// A node that runs a controller has both; the controller's is authoritative
143/// and is preferred at an equal id, because the observer records a placeholder
144/// epoch in a checkpoint it wrote itself.
145#[must_use]
146pub fn newest_metadata_checkpoint(log_dir: &Path) -> Option<MetadataCheckpoint> {
147    let mut best: Option<MetadataCheckpoint> = None;
148    for subdir in [CONTROLLER_CHECKPOINT_DIR, OBSERVER_CHECKPOINT_DIR] {
149        let dir = log_dir.join(subdir);
150        // `newest_checkpoint` selects only names that `checkpoint_id` parsed.
151        let Some((id, name)) = newest_checkpoint(&file_names(&dir))
152            .and_then(|name| Some((checkpoint_id(&name)?, name)))
153        else {
154            continue;
155        };
156        if best.as_ref().is_none_or(|seen| id > seen.id) {
157            best = Some(MetadataCheckpoint {
158                id,
159                path: dir.join(name),
160            });
161        }
162    }
163    best
164}
165
166/// The id a capture taken at `now_ms` carries, which is also its directory
167/// name inside the archive.
168///
169/// The zero-padded epoch millisecond orders captures across timestamps; a random UUID suffix distinguishes concurrent captures and repeated clock values. Legacy timestamp-only names retain their position in the same listing, and `--capture latest` still needs only a lexicographic maximum.
170#[must_use]
171pub fn capture_id(now_ms: u64) -> String { format!("{now_ms:016}-{}", Uuid::new_v4().simple()) }
172
173/// The archive-relative key of one artifact inside one capture.
174#[must_use]
175pub fn capture_key(capture_id: &str, artifact: &str) -> String {
176    format!("{CAPTURE_ROOT}/{capture_id}/{artifact}")
177}
178
179#[cfg(test)]
180mod tests {
181    use {
182        super::{
183            CheckpointId, MetadataCheckpoint, capture_id, capture_key, checkpoint_id,
184            newest_checkpoint, newest_metadata_checkpoint,
185        },
186        assert2::check,
187        std::fs,
188    };
189
190    #[yare::parameterized(
191        regular = {
192            "00000000000000123456-0000000042.checkpoint",
193            Some(CheckpointId { end_offset: 123_456, epoch: 42 }),
194        },
195        zero = {
196            "00000000000000000000-0000000000.checkpoint",
197            Some(CheckpointId { end_offset: 0, epoch: 0 }),
198        },
199        short_both_fields = { "123456-42.checkpoint", None },
200        short_epoch = { "00000000000000123456-42.checkpoint", None },
201        snapshot = { "snapshot", None },
202        high_watermark = { "high-watermark.checkpoint", None },
203        suffix = { "00000000000000123456-0000000042.checkpoint.tmp", None },
204    )]
205    fn only_the_canonical_fixed_width_name_is_a_checkpoint(
206        name: &str,
207        expected: Option<CheckpointId>,
208    ) {
209        check!(checkpoint_id(name) == expected, "for {name}");
210    }
211
212    #[yare::parameterized(
213        format_time = { 0, 0, true },
214        format_time_later_epoch = { 0, 3, true },
215        first_controller_checkpoint = { 1, 1, false },
216        far_along = { 123_456, 42, false },
217    )]
218    fn only_end_offset_zero_is_the_format_time_checkpoint(
219        end_offset: u64,
220        epoch: u64,
221        expected: bool,
222    ) {
223        check!(CheckpointId { end_offset, epoch }.is_format_time() == expected);
224    }
225
226    #[test]
227    fn the_newest_checkpoint_orders_by_end_offset_before_epoch() {
228        let names = vec![
229            "00000000000000000010-0000000009.checkpoint".to_owned(),
230            "00000000000000000200-0000000001.checkpoint".to_owned(),
231            "00000000000000000200-0000000002.checkpoint".to_owned(),
232            "snapshot".to_owned(),
233        ];
234        check!(
235            newest_checkpoint(&names)
236                == Some("00000000000000000200-0000000002.checkpoint".to_owned())
237        );
238    }
239
240    #[test]
241    fn a_directory_with_no_checkpoint_has_no_newest_one() {
242        check!(newest_checkpoint(&["snapshot".to_owned()]) == None);
243        check!(newest_checkpoint(&[]) == None);
244    }
245
246    #[test]
247    fn an_observer_checkpoint_is_found_when_there_is_no_controller_directory() {
248        let log_dir = tempfile::tempdir().expect("log dir");
249        let observer = log_dir.path().join("__cluster_metadata-0/observer");
250        fs::create_dir_all(&observer).expect("create the observer dir");
251        let newest = observer.join("00000000000000000042-0000000000.checkpoint");
252        fs::write(&newest, b"bytes").expect("write a checkpoint");
253        fs::write(
254            observer.join("00000000000000000009-0000000000.checkpoint"),
255            b"older",
256        )
257        .expect("write an older checkpoint");
258
259        check!(
260            newest_metadata_checkpoint(log_dir.path())
261                == Some(MetadataCheckpoint {
262                    id: CheckpointId {
263                        end_offset: 42,
264                        epoch: 0
265                    },
266                    path: newest,
267                })
268        );
269    }
270
271    #[test]
272    fn the_controller_directory_wins_when_both_hold_the_same_id() {
273        let log_dir = tempfile::tempdir().expect("log dir");
274        let name = "00000000000000000042-0000000000.checkpoint";
275        let controller = log_dir.path().join("__cluster_metadata-0");
276        let observer = controller.join("observer");
277        fs::create_dir_all(&controller).expect("create the controller checkpoint dir");
278        fs::create_dir_all(&observer).expect("create the observer checkpoint dir");
279        fs::write(controller.join(name), b"bytes").expect("write a checkpoint");
280        fs::write(observer.join(name), b"bytes").expect("write a checkpoint");
281
282        check!(
283            newest_metadata_checkpoint(log_dir.path())
284                == Some(MetadataCheckpoint {
285                    id: CheckpointId {
286                        end_offset: 42,
287                        epoch: 0
288                    },
289                    path: log_dir.path().join("__cluster_metadata-0").join(name),
290                })
291        );
292    }
293
294    #[test]
295    fn a_log_directory_with_no_checkpoint_yields_none() {
296        let log_dir = tempfile::tempdir().expect("log dir");
297        check!(newest_metadata_checkpoint(log_dir.path()) == None);
298    }
299
300    #[test]
301    fn capture_ids_sort_lexicographically_in_time_order() {
302        let early = capture_id(1_700_000_000_000);
303        let late = capture_id(1_700_000_000_001);
304        check!(early < late);
305        check!(capture_id(0) < early);
306        check!(
307            capture_key(&early, "manifest.json") == format!("restore-inputs/{early}/manifest.json")
308        );
309        check!(early.starts_with("0001700000000000-"));
310        check!(early != capture_id(1_700_000_000_000));
311        check!("0001700000000000" < early.as_str());
312        check!(early.as_str() < "0001700000000001");
313    }
314}
krabka_backup/capture.rs

krabka_backup/
capture.rs
