krabka_restore/error.rs
1//! The restore error taxonomy and the process exit codes it maps to.
2//!
3//! The library returns [`RestoreError`]. Only [`run`](crate::run) and the
4//! binary turn one into an exit code and a line on stderr, so a caller that
5//! embeds the restore keeps the structured error.
6
7use {
8 krabka_log::LogError,
9 krabka_object_store::ObjectStoreError,
10 krabka_protocol::records::RecordsError,
11 krabka_remote_storage::RemoteStorageError,
12 std::{io::Error, path::PathBuf},
13 uuid::Uuid,
14};
15
16/// Exit codes:
17/// - 0: success
18/// - 2: bad arguments
19/// - 3: target log directory not empty
20/// - 4: archive unreadable or empty
21/// - 5: integrity failure
22/// - 6: materialization failure
23pub const EXIT_OK: i32 = 0;
24/// See [`EXIT_OK`].
25pub const EXIT_BAD_ARGUMENTS: i32 = 2;
26/// See [`EXIT_OK`].
27pub const EXIT_DIRTY_LOG_DIR: i32 = 3;
28/// See [`EXIT_OK`].
29pub const EXIT_ARCHIVE_UNREADABLE: i32 = 4;
30/// See [`EXIT_OK`].
31pub const EXIT_INTEGRITY: i32 = 5;
32/// See [`EXIT_OK`].
33pub const EXIT_MATERIALIZE: i32 = 6;
34
35/// Every way a point-in-time restore stops.
36///
37/// The variants keep the object key, the byte position, and the partition that
38/// identify the failure, because an operator who runs a restore during an
39/// incident has to decide whether to re-run with `--continue-on-corrupt` or to
40/// go back to the archive. A message alone does not support that decision.
41#[derive(Debug, thiserror::Error)]
42#[non_exhaustive]
43pub enum RestoreError {
44 /// A flag combination the parser accepts but the restore cannot act on.
45 #[error("invalid argument: {0}")]
46 InvalidArgument(String),
47
48 /// `--log-dir` names a directory that already holds entries. A restore
49 /// writes a complete cluster, so it never merges into existing state.
50 #[error("refusing to restore into non-empty log directory {0}")]
51 LogDirNotEmpty(PathBuf),
52
53 /// The scan reached the archive but found no segment for any selected
54 /// topic. A wrong `--archive-prefix` reads exactly like an empty bucket,
55 /// so the message carries the prefix that was searched.
56 #[error("archive holds no segment under prefix {prefix:?}")]
57 EmptyArchive {
58 /// The key prefix the scan searched, empty when the whole bucket was
59 /// scanned.
60 prefix: String,
61 },
62
63 /// A batch header's CRC does not match the bytes that follow it.
64 #[error(
65 "checksum mismatch in {key} at byte {position}: header records {expected:#010x}, \
66 body computes {computed:#010x}"
67 )]
68 ChecksumMismatch {
69 /// The object key of the artifact that holds the corrupt batch.
70 key: String,
71 /// Byte position of the batch header within that object.
72 position: u64,
73 /// The CRC the batch header declares.
74 expected: u32,
75 /// The CRC re-computed over the batch body.
76 computed: u32,
77 },
78
79 /// A segment ends part way through a batch. The archived copy is short of
80 /// the length its last batch header declares.
81 #[error(
82 "segment {key} ends inside a batch at byte {position}: {declared} bytes declared, \
83 {available} available"
84 )]
85 TruncatedSegment {
86 /// The object key of the short artifact.
87 key: String,
88 /// Byte position of the batch header that overruns the object.
89 position: u64,
90 /// The batch length that header declares.
91 declared: u64,
92 /// The bytes that remain in the object from `position`.
93 available: u64,
94 },
95
96 /// A segment copy stopped part way: the archive holds some of the
97 /// segment's artifacts and not the rest.
98 #[error("torn copy of segment {segment_id} of {topic}-{partition}: {artifact} is absent")]
99 TornCopy {
100 /// Topic of the incomplete segment.
101 topic: String,
102 /// Partition index of the incomplete segment.
103 partition: i32,
104 /// The segment id the archive names the copy with.
105 segment_id: Uuid,
106 /// The artifact that is absent, such as `.timeindex`.
107 artifact: String,
108 },
109
110 /// Trusted WORM evidence did not authenticate an archive object or chain.
111 #[error("WORM authentication failed: {reason}")]
112 Authenticity {
113 /// The failed trust check.
114 reason: String,
115 },
116
117 /// Captured diskless-WAL state or a required object is inconsistent.
118 #[error("integrity failure: {0}")]
119 Integrity(String),
120
121 /// The bucket scan and the `--rlmm-snapshot` do not agree about a
122 /// partition. One of the two is stale, and a restore must not pick for the
123 /// operator.
124 #[error(
125 "metadata disagreement for {topic}-{partition}: the bucket scan reports {scanned}, \
126 the RLMM snapshot reports {snapshot}"
127 )]
128 MetadataDisagreement {
129 /// Topic the two sources disagree about.
130 topic: String,
131 /// Partition index the two sources disagree about.
132 partition: i32,
133 /// What the bucket scan found.
134 scanned: String,
135 /// What the RLMM snapshot states.
136 snapshot: String,
137 },
138
139 /// A controller metadata snapshot could not be decoded.
140 #[error("cannot read metadata snapshot {path}: {reason}")]
141 MetadataSnapshot {
142 /// Snapshot path supplied by `--metadata-snapshot`.
143 path: PathBuf,
144 /// Decoder failure.
145 reason: String,
146 },
147
148 /// A topic configuration belongs to a different incarnation of an
149 /// archived topic, or cannot be tied to one at all.
150 #[error(
151 "metadata snapshot topic id mismatch for {topic}: archive has {archive_topic_id}, \
152 snapshot has {snapshot_topic_id}"
153 )]
154 MetadataSnapshotTopicMismatch {
155 /// Topic name shared by the archive and snapshot.
156 topic: String,
157 /// Topic id carried by the archived partition paths.
158 archive_topic_id: Uuid,
159 /// Topic id in the snapshot, or `absent` when its topic record is missing.
160 snapshot_topic_id: String,
161 },
162
163 /// A bound names a topic partition the archive does not hold. Left
164 /// unreported this silently restores the partition whole, which is the
165 /// opposite of what the operator asked for.
166 #[error("bound names {topic}-{partition}, which the archive does not hold")]
167 UnknownPartition {
168 /// Topic the bound names.
169 topic: String,
170 /// Partition index the bound names.
171 partition: i32,
172 },
173
174 /// The formatter refused to format the target log directory.
175 #[error("formatting the target log directory failed with exit code {code}")]
176 Format {
177 /// The exit code the formatter returned.
178 code: i32,
179 },
180
181 /// The tiered-storage layer failed.
182 #[error("remote storage: {0}")]
183 RemoteStorage(#[from] RemoteStorageError),
184
185 /// The object store failed, or its configuration was rejected.
186 #[error("object store: {0}")]
187 ObjectStore(#[from] ObjectStoreError),
188
189 /// The target log failed to accept a rehydrated segment.
190 #[error("log: {0}")]
191 Log(#[from] LogError),
192
193 /// A record batch would not decode or re-encode.
194 #[error("records: {0}")]
195 Records(#[from] RecordsError),
196
197 /// A local filesystem operation failed.
198 #[error("I/O: {0}")]
199 Io(#[from] Error),
200}
201
202impl RestoreError {
203 /// The process exit code this error reports.
204 ///
205 /// The grouping is what an operator acts on: an integrity failure means
206 /// the archive is damaged and a re-run needs `--continue-on-corrupt`,
207 /// while a materialization failure means the target is at fault and the
208 /// archive is intact.
209 #[must_use]
210 pub fn exit_code(&self) -> i32 {
211 match self {
212 Self::InvalidArgument(_) | Self::UnknownPartition { .. } => EXIT_BAD_ARGUMENTS,
213 Self::LogDirNotEmpty(_) => EXIT_DIRTY_LOG_DIR,
214 Self::EmptyArchive { .. } | Self::ObjectStore(_) | Self::RemoteStorage(_) => {
215 EXIT_ARCHIVE_UNREADABLE
216 }
217 Self::ChecksumMismatch { .. }
218 | Self::TruncatedSegment { .. }
219 | Self::TornCopy { .. }
220 | Self::Authenticity { .. }
221 | Self::Integrity(_)
222 | Self::MetadataDisagreement { .. }
223 | Self::MetadataSnapshot { .. }
224 | Self::MetadataSnapshotTopicMismatch { .. }
225 | Self::Records(_) => EXIT_INTEGRITY,
226 Self::Format { .. } | Self::Log(_) | Self::Io(_) => EXIT_MATERIALIZE,
227 }
228 }
229}
230
231#[cfg(test)]
232mod tests {
233 use {super::*, assert2::check};
234
235 #[test]
236 fn exit_codes_are_distinct_and_documented() {
237 let codes = [
238 EXIT_OK,
239 EXIT_BAD_ARGUMENTS,
240 EXIT_DIRTY_LOG_DIR,
241 EXIT_ARCHIVE_UNREADABLE,
242 EXIT_INTEGRITY,
243 EXIT_MATERIALIZE,
244 ];
245 let mut seen = codes.to_vec();
246 seen.sort_unstable();
247 seen.dedup();
248 check!(seen.len() == codes.len());
249 check!(codes == [0, 2, 3, 4, 5, 6]);
250 }
251
252 #[test]
253 fn each_failure_class_maps_to_its_exit_code() {
254 let cases = vec![
255 (
256 RestoreError::InvalidArgument("bad".into()),
257 EXIT_BAD_ARGUMENTS,
258 ),
259 (
260 RestoreError::UnknownPartition {
261 topic: "orders".into(),
262 partition: 3,
263 },
264 EXIT_BAD_ARGUMENTS,
265 ),
266 (
267 RestoreError::LogDirNotEmpty(PathBuf::from("/var/lib/krabka")),
268 EXIT_DIRTY_LOG_DIR,
269 ),
270 (
271 RestoreError::EmptyArchive {
272 prefix: "tier".into(),
273 },
274 EXIT_ARCHIVE_UNREADABLE,
275 ),
276 (
277 RestoreError::ChecksumMismatch {
278 key: "orders-0-abc/000.log".into(),
279 position: 4096,
280 expected: 0xdead_beef,
281 computed: 0x0bad_f00d,
282 },
283 EXIT_INTEGRITY,
284 ),
285 (
286 RestoreError::TruncatedSegment {
287 key: "orders-0-abc/000.log".into(),
288 position: 8192,
289 declared: 512,
290 available: 40,
291 },
292 EXIT_INTEGRITY,
293 ),
294 (
295 RestoreError::TornCopy {
296 topic: "orders".into(),
297 partition: 0,
298 segment_id: Uuid::nil(),
299 artifact: ".timeindex".into(),
300 },
301 EXIT_INTEGRITY,
302 ),
303 (
304 RestoreError::MetadataDisagreement {
305 topic: "orders".into(),
306 partition: 0,
307 scanned: "4 segments".into(),
308 snapshot: "3 segments".into(),
309 },
310 EXIT_INTEGRITY,
311 ),
312 (
313 RestoreError::MetadataSnapshot {
314 path: "metadata.checkpoint".into(),
315 reason: "bad frame".into(),
316 },
317 EXIT_INTEGRITY,
318 ),
319 (
320 RestoreError::MetadataSnapshotTopicMismatch {
321 topic: "orders".into(),
322 archive_topic_id: Uuid::nil(),
323 snapshot_topic_id: "absent".into(),
324 },
325 EXIT_INTEGRITY,
326 ),
327 (RestoreError::Format { code: 3 }, EXIT_MATERIALIZE),
328 (
329 RestoreError::Io(Error::other("disk full")),
330 EXIT_MATERIALIZE,
331 ),
332 ];
333 for (error, expected) in cases {
334 check!(error.exit_code() == expected, "{error}");
335 }
336 }
337
338 #[test]
339 fn checksum_mismatch_names_the_key_and_the_byte_position() {
340 let error = RestoreError::ChecksumMismatch {
341 key: "tier/orders-0-abc/00000000000000000000-xyz.log".into(),
342 position: 65_536,
343 expected: 0x1234_5678,
344 computed: 0x8765_4321,
345 };
346 let rendered = error.to_string();
347 check!(rendered.contains("tier/orders-0-abc/00000000000000000000-xyz.log"));
348 check!(rendered.contains("65536"));
349 check!(rendered.contains("0x12345678"));
350 check!(rendered.contains("0x87654321"));
351 }
352
353 #[test]
354 fn io_errors_convert_without_a_string_hop() {
355 let error: RestoreError = Error::other("boom").into();
356 check!(matches!(error, RestoreError::Io(_)));
357 }
358}