krabka_log/error.rs
1//! Errors returned by `Log` and `Segment`.
2
3use {
4 krabka_ids::Offset,
5 krabka_protocol::records::RecordsError,
6 std::{io, path::PathBuf},
7 thiserror::Error,
8};
9
10/// Errors returned by [`Log`](crate::Log) and [`Segment`](crate::Segment).
11#[derive(Debug, Error)]
12pub enum LogError {
13 /// An I/O operation against the filesystem failed.
14 #[error("I/O: {0}")]
15 Io(#[from] io::Error),
16
17 /// Recovery found a partial batch at the tail of a `.log` file. The log
18 /// truncates the trailing bytes back to the last cleanly decoded batch.
19 #[error("partial batch at offset {file_offset} in segment {segment}: truncating")]
20 PartialBatch {
21 /// Absolute base offset of the segment containing the partial batch.
22 segment: Offset,
23 /// Byte position within the `.log` file where the partial batch starts.
24 file_offset: u64,
25 },
26
27 /// A batch's stored CRC did not match the one computed over its bytes.
28 #[error(
29 "CRC mismatch at offset {file_offset} in segment {segment}: \
30 expected {expected:#x}, computed {computed:#x}"
31 )]
32 CrcMismatch {
33 /// Absolute base offset of the segment.
34 segment: Offset,
35 /// Byte position within the `.log` file where the corrupt batch starts.
36 file_offset: u64,
37 /// CRC value embedded in the batch header.
38 expected: u32,
39 /// CRC value re-computed by the reader.
40 computed: u32,
41 },
42
43 /// A caller requested an offset below [`Log::log_start_offset`](crate::Log::log_start_offset).
44 #[error("offset {requested} below log start {log_start}")]
45 OffsetTooLow {
46 /// Offset the caller asked for.
47 requested: Offset,
48 /// Current log start.
49 log_start: Offset,
50 },
51
52 /// A caller requested an offset that is at or above
53 /// [`Log::log_start_offset`](crate::Log::log_start_offset) but below
54 /// [`Log::local_log_start_offset`](crate::Log::local_log_start_offset).
55 ///
56 /// KIP-405: the records are in the remote tier and no local segment holds
57 /// them. The broker's fetch path answers `OFFSET_OUT_OF_RANGE` and then
58 /// falls through to the remote reader, which is where these offsets are
59 /// served from.
60 #[error("offset {requested} is tiered: below local log start {local_log_start}")]
61 OffsetBelowLocalStart {
62 /// Offset the caller asked for.
63 requested: Offset,
64 /// First offset the local segments still hold.
65 local_log_start: Offset,
66 },
67
68 /// The encode or decode of a `RecordBatch` failed.
69 #[error("records: {0}")]
70 Records(#[from] RecordsError),
71
72 /// A segment filename would not parse. For example, it has the wrong
73 /// length, or it is not all digits.
74 #[error("invalid segment filename: {0}")]
75 BadSegmentName(String),
76
77 /// A caller supplied an explicit offset to [`Log::append_at`](crate::Log::append_at)
78 /// that was below the log's current end offset. Replication paths use
79 /// this to detect a duplicate or a divergence between leader-assigned offsets and the local
80 /// log's expected next offset.
81 #[error("offset mismatch: expected {expected}, got {actual}")]
82 OffsetMismatch {
83 /// The lowest offset the log takes, that is, its current end offset.
84 expected: Offset,
85 /// The offset the caller actually supplied.
86 actual: Offset,
87 },
88
89 /// A log file such as `.txnindex` is corrupt. It has the wrong size, a
90 /// bad checksum, or a similar defect.
91 #[error("corrupt log: {0}")]
92 Corrupt(String),
93
94 /// A persisted sidecar or checkpoint names a format version this build
95 /// does not read. A future 1.x broker wrote it, or it is not the artifact
96 /// its name says.
97 #[error("unrecognized version {found} of the {artifact} file {}", path.display())]
98 UnsupportedFormatVersion {
99 /// The artifact kind, such as `stampindex` or `leader-epoch-checkpoint`.
100 artifact: &'static str,
101 /// The file that carries the version.
102 path: PathBuf,
103 /// The version the file carries.
104 found: i64,
105 },
106
107 /// A persisted sidecar or checkpoint has no version marker. Only a broker
108 /// before 1.0 wrote that layout, and 1.x reads no 0.x data.
109 #[error(
110 "the {artifact} file {} has no version marker: it predates krabka 1.0, so the node \
111 must be reformatted",
112 path.display()
113 )]
114 MissingFormatVersion {
115 /// The artifact kind, such as `stampindex` or `leader-epoch-checkpoint`.
116 artifact: &'static str,
117 /// The file that lacks the marker.
118 path: PathBuf,
119 },
120
121 /// A control record's key or end-transaction-marker value is too short to
122 /// hold its fields. Kafka's `ControlRecordType.parseTypeId` throws
123 /// `InvalidRecordException` for a short key, and
124 /// `EndTransactionMarker.deserializeValue` cannot read a short value.
125 #[error(
126 "Invalid value size found for {record}. Must have at least {needed} bytes, but found \
127 only {found}"
128 )]
129 InvalidControlRecordSize {
130 /// `end control record key` or `end transaction marker value`.
131 record: &'static str,
132 /// The bytes the record must hold.
133 needed: usize,
134 /// The bytes it holds; a missing key or value counts as zero.
135 found: usize,
136 },
137
138 /// A control record names a negative key version or end-transaction-marker
139 /// value version. Kafka's `ControlRecordType.parseTypeId` and
140 /// `EndTransactionMarker.deserializeValue` throw `InvalidRecordException`
141 /// for one: the bytes are corrupt. A version above the current one is not
142 /// this error; Kafka parses it as the current version.
143 #[error("Invalid version found for {record}: {version}. May indicate data corruption")]
144 InvalidControlRecordVersion {
145 /// `control record` for the key, `end transaction marker` for the value.
146 record: &'static str,
147 /// The negative version the record carries.
148 version: i16,
149 },
150
151 /// A caller supplied an invalid argument.
152 #[error("invalid argument: {0}")]
153 InvalidArgument(String),
154
155 /// Kafka trunk's `max.decompressed.message.bytes`: a compressed batch that a
156 /// lookup or a compaction pass had to decompress holds a record whose body
157 /// is larger than [`LogConfig::max_decompressed_record`](crate::LogConfig::max_decompressed_record).
158 ///
159 /// Kafka throws `InvalidRecordException` from `DefaultRecord.readFrom`, and
160 /// the message is its own. The broker answers `INVALID_RECORD` (87) to a
161 /// `ListOffsets`, and the cleaner leaves the partition uncleanable until a
162 /// pass succeeds.
163 #[error("Invalid record size: {size} exceeds the configured maximum record size of {limit}.")]
164 RecordTooLarge {
165 /// The record's body size in bytes, Kafka's `sizeOfBodyInBytes`.
166 size: usize,
167 /// The configured maximum.
168 limit: usize,
169 },
170
171 /// KFC-1 `delivery.schedule.monotonic`: the appended batch's delivery time
172 /// precedes a delivery time the partition already holds, so the batch
173 /// would make the partition's schedule run backwards.
174 ///
175 /// The log raises it because the log is what serializes appends: the test
176 /// and the write it guards are one critical section here and nowhere
177 /// above. The broker answers it with `INVALID_TIMESTAMP` (32).
178 #[error("delivery time {delivery_ms} precedes a delivery time already in the partition")]
179 ScheduleRunsBackwards {
180 /// The refused batch's delivery time, which is its `max_timestamp`.
181 delivery_ms: i64,
182 },
183
184 /// The log was fenced with [`Log::fence`](crate::Log::fence) because its
185 /// partition is being removed, and takes no more appends.
186 #[error("the log is fenced: its partition is being removed")]
187 Fenced,
188}
189
190#[cfg(test)]
191mod tests {
192 use super::*;
193
194 #[test]
195 fn display_partial_batch() {
196 let e = LogError::PartialBatch {
197 segment: Offset(0),
198 file_offset: 1024,
199 };
200 assert2::assert!(e.to_string() == "partial batch at offset 1024 in segment 0: truncating");
201 }
202}