zeppelin_embed/wal/
replay.rs

1//! Checked WAL prefix replay.
2
3use super::LogSeq;
4use super::header::{WAL_HEADER_LEN, WalHeaderError, decode_header};
5use super::record::{RecordError, WalRecord, decode_record};
6
7/// Whether a corrupt record is the final observable record or precedes a
8/// checksum-valid successor.
9#[derive(Clone, Copy, Debug, Eq, PartialEq)]
10pub enum CorruptionLocation {
11    /// No checksum-valid successor was found at the declared next boundary.
12    Tail,
13    /// A checksum-valid record follows the corrupt record.
14    Middle,
15}
16
17/// Specific corruption signal that ended replay.
18#[derive(Clone, Copy, Debug, Eq, PartialEq)]
19pub enum CorruptionReason {
20    /// Record framing or checksum validation failed.
21    Record {
22        /// Tail versus proven middle corruption.
23        location: CorruptionLocation,
24        /// Exact record validation failure.
25        error: RecordError,
26    },
27    /// The first valid record disagreed with the sequence declared by the file header.
28    FirstSequenceMismatch {
29        /// Sequence declared by the WAL file header.
30        expected: LogSeq,
31        /// Sequence observed in the first checked record.
32        actual: LogSeq,
33        /// Tail versus proven middle corruption.
34        location: CorruptionLocation,
35    },
36    /// A valid record skipped one or more required sequence numbers.
37    SequenceGap {
38        /// Required next sequence.
39        expected: LogSeq,
40        /// Sequence observed in the checked record.
41        actual: LogSeq,
42        /// Tail versus proven middle corruption.
43        location: CorruptionLocation,
44    },
45    /// A valid record repeated or regressed below the required sequence.
46    SequenceRegression {
47        /// Required next sequence.
48        expected: LogSeq,
49        /// Sequence observed in the checked record.
50        actual: LogSeq,
51        /// Tail versus proven middle corruption.
52        location: CorruptionLocation,
53    },
54}
55
56/// Why replay stopped after its trusted prefix.
57#[derive(Clone, Copy, Debug, Eq, PartialEq)]
58pub enum ReplayTerminator {
59    /// WAL file-header validation failed before record replay began.
60    InvalidHeader(WalHeaderError),
61    /// All bytes ended exactly after a valid record, or the log was empty.
62    CleanEnd,
63    /// Replay stopped before consuming the named invalid record.
64    CorruptAt {
65        /// Byte offset of the first invalid record.
66        offset: usize,
67        /// Specific validation or sequence failure.
68        reason: CorruptionReason,
69    },
70}
71
72/// A checksum- and sequence-validated prefix plus its explicit terminator.
73#[derive(Clone, Debug, Eq, PartialEq)]
74pub struct ReplayResult<'a> {
75    /// Records preceding the first invalid record.
76    pub records: Vec<WalRecord<'a>>,
77    /// Clean end versus typed corruption boundary.
78    pub terminator: ReplayTerminator,
79}
80
81/// Replays the largest trusted prefix of an append-only WAL.
82///
83/// Every record considered for return is checksum-checked. Replay never skips
84/// an invalid record. If a checksum-valid record begins at the invalid
85/// record's declared successor boundary, the reason is classified as
86/// [`CorruptionLocation::Middle`], but that successor is not returned. A torn
87/// tail or corruption without such a successor is classified as `Tail`.
88/// The file header declares the first required sequence, and every subsequent
89/// record must increment by exactly one; gaps, duplicates, and regressions
90/// terminate the trusted prefix.
91#[must_use]
92pub fn replay(bytes: &[u8]) -> ReplayResult<'_> {
93    let mut records = Vec::new();
94    let header = match decode_header(bytes) {
95        Ok(header) => header,
96        Err(error) => {
97            return ReplayResult {
98                records,
99                terminator: ReplayTerminator::InvalidHeader(error),
100            };
101        }
102    };
103    let mut offset = WAL_HEADER_LEN;
104    let mut expected = header.first_seq;
105    while let Some(remaining) = bytes
106        .get(offset..)
107        .filter(|remaining| !remaining.is_empty())
108    {
109        let decoded = match decode_record(remaining) {
110            Ok(decoded) => decoded,
111            Err(error) => {
112                let location = error
113                    .framed_length()
114                    .and_then(|length| offset.checked_add(length))
115                    .map_or(CorruptionLocation::Tail, |next| {
116                        successor_location(bytes, next)
117                    });
118                return ReplayResult {
119                    records,
120                    terminator: ReplayTerminator::CorruptAt {
121                        offset,
122                        reason: CorruptionReason::Record { location, error },
123                    },
124                };
125            }
126        };
127        let next = offset.saturating_add(decoded.encoded_len);
128        if decoded.record.seq != expected {
129            let location = successor_location(bytes, next);
130            let reason = if records.is_empty() {
131                CorruptionReason::FirstSequenceMismatch {
132                    expected,
133                    actual: decoded.record.seq,
134                    location,
135                }
136            } else if decoded.record.seq > expected {
137                CorruptionReason::SequenceGap {
138                    expected,
139                    actual: decoded.record.seq,
140                    location,
141                }
142            } else {
143                CorruptionReason::SequenceRegression {
144                    expected,
145                    actual: decoded.record.seq,
146                    location,
147                }
148            };
149            return ReplayResult {
150                records,
151                terminator: ReplayTerminator::CorruptAt { offset, reason },
152            };
153        }
154        records.push(decoded.record);
155        offset = next;
156        expected = LogSeq::new(expected.get().saturating_add(1));
157    }
158    ReplayResult {
159        records,
160        terminator: ReplayTerminator::CleanEnd,
161    }
162}
163
164fn successor_location(bytes: &[u8], offset: usize) -> CorruptionLocation {
165    bytes
166        .get(offset..)
167        .filter(|successor| !successor.is_empty())
168        .and_then(|successor| decode_record(successor).ok())
169        .map_or(CorruptionLocation::Tail, |_| CorruptionLocation::Middle)
170}
zeppelin_embed/wal/replay.rs

zeppelin_embed/wal/
replay.rs
