nodedb_wal/segment/
continuity.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Cross-segment LSN continuity for multi-segment replay.
4//!
5//! Within a segment, a stop point with committed records behind it is caught by
6//! [`crate::torn_tail::verify_committed_prefix`]. The same class of silent data
7//! loss exists one level up: replay concatenates the records of every segment
8//! in the directory, so a segment file that has gone missing from the middle of
9//! the log produces a record stream with a hole in it and no error anywhere.
10//!
11//! Segments are written in strictly increasing LSN order, so the boundary is
12//! checkable: the next segment's declared first LSN must not be above the LSN
13//! that follows the last record of the previous one.
14//!
15//! Two things are deliberately *not* holes:
16//!
17//! - **A missing prefix.** Checkpoint truncation deletes whole segments below
18//!   the checkpoint by design, so the first surviving segment may start
19//!   anywhere. Only boundaries between surviving segments are checked.
20//! - **A segment with no records.** A rollover that was never written to says
21//!   nothing about continuity, so the boundary carries over from the last
22//!   segment that actually held records.
23
24use std::path::PathBuf;
25
26use crate::error::{Result, WalError};
27
28use super::meta::SegmentMeta;
29
30/// Running LSN boundary between the segments replayed so far.
31#[derive(Debug, Default)]
32pub struct SegmentContinuity {
33    /// Path and highest LSN of the most recent segment that was read to
34    /// completion and contained at least one record.
35    previous: Option<(PathBuf, u64)>,
36}
37
38impl SegmentContinuity {
39    /// Start a fresh check with no established boundary.
40    pub fn new() -> Self {
41        Self::default()
42    }
43
44    /// Verify that `segment` continues the log where the previous completed
45    /// segment left off. Call this before reading `segment`.
46    pub fn check(&self, segment: &SegmentMeta) -> Result<()> {
47        let Some((previous_path, previous_last_lsn)) = &self.previous else {
48            return Ok(());
49        };
50
51        let expected_lsn = previous_last_lsn.saturating_add(1);
52        if segment.first_lsn > expected_lsn {
53            let err = WalError::SegmentLsnGap {
54                path: segment.path.display().to_string(),
55                previous_path: previous_path.display().to_string(),
56                previous_last_lsn: *previous_last_lsn,
57                expected_lsn,
58                found_lsn: segment.first_lsn,
59            };
60            // The boundary check is the only place that can see the hole: a
61            // single segment says nothing about the one that should precede it.
62            crate::diag::segment_lsn_gap(
63                &err,
64                &segment.path,
65                previous_path,
66                *previous_last_lsn,
67                expected_lsn,
68                segment.first_lsn,
69            );
70            return Err(err);
71        }
72
73        Ok(())
74    }
75
76    /// Record that `segment` was read all the way through, ending at
77    /// `last_lsn`. Call this only after the segment has been fully consumed —
78    /// a replay that stopped early (a record limit) has learned nothing about
79    /// where the segment ends.
80    ///
81    /// `last_lsn` must be the highest LSN the segment *contains*, not the
82    /// highest one the caller kept: LSN filtering changes what is returned, not
83    /// what is on disk. A `last_lsn` of 0 means the segment held no records.
84    pub fn completed(&mut self, segment: &SegmentMeta, last_lsn: u64) {
85        if last_lsn != 0 {
86            self.previous = Some((segment.path.clone(), last_lsn));
87        }
88    }
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94    use std::path::Path;
95
96    fn meta(first_lsn: u64) -> SegmentMeta {
97        SegmentMeta {
98            path: Path::new("/wal").join(super::super::segment_filename(first_lsn)),
99            first_lsn,
100            file_size: 0,
101        }
102    }
103
104    #[test]
105    fn first_segment_may_start_anywhere() {
106        let continuity = SegmentContinuity::new();
107        assert!(continuity.check(&meta(5000)).is_ok());
108    }
109
110    #[test]
111    fn contiguous_segments_pass() {
112        let mut continuity = SegmentContinuity::new();
113        let first = meta(1);
114        continuity.check(&first).unwrap();
115        continuity.completed(&first, 42);
116        assert!(continuity.check(&meta(43)).is_ok());
117    }
118
119    #[test]
120    fn gap_between_segments_is_rejected() {
121        let mut continuity = SegmentContinuity::new();
122        let first = meta(1);
123        continuity.completed(&first, 42);
124        match continuity.check(&meta(90)) {
125            Err(WalError::SegmentLsnGap {
126                expected_lsn,
127                found_lsn,
128                ..
129            }) => {
130                assert_eq!(expected_lsn, 43);
131                assert_eq!(found_lsn, 90);
132            }
133            other => panic!("expected a segment LSN gap, got {other:?}"),
134        }
135    }
136
137    #[test]
138    fn empty_segment_keeps_the_previous_boundary() {
139        let mut continuity = SegmentContinuity::new();
140        let first = meta(1);
141        continuity.completed(&first, 42);
142        let empty = meta(43);
143        continuity.check(&empty).unwrap();
144        continuity.completed(&empty, 0);
145        assert!(continuity.check(&meta(43)).is_ok());
146        assert!(continuity.check(&meta(90)).is_err());
147    }
148}
nodedb_wal/segment/continuity.rs

nodedb_wal/segment/
continuity.rs
