krabka_remote_storage/error.rs
1//! Error type shared by both tiered-storage SPIs.
2
3use {
4 crate::{
5 metadata::{
6 RemoteLogSegmentId, RemoteLogSegmentState, RemotePartitionDeleteState, TopicIdPartition,
7 },
8 worm::WormError,
9 },
10 std::io,
11};
12
13/// Errors raised by [`RemoteStorageManager`](crate::RemoteStorageManager)
14/// and [`RemoteLogMetadataManager`](crate::RemoteLogMetadataManager)
15/// implementations.
16#[derive(Debug, thiserror::Error)]
17pub enum RemoteStorageError {
18 /// An I/O failure in the underlying store (filesystem, object store, …).
19 #[error("remote storage I/O error: {0}")]
20 Io(#[from] io::Error),
21
22 /// A segment was referenced that the metadata store has never seen.
23 #[error("no remote log segment metadata for {0:?}")]
24 SegmentNotFound(RemoteLogSegmentId),
25
26 /// A caller gave `add_remote_log_segment_metadata` a starting state
27 /// other than [`RemoteLogSegmentState::CopySegmentStarted`], or a
28 /// segment id that already exists.
29 #[error("invalid add for {id:?}: {reason}")]
30 InvalidAdd {
31 /// The offending segment id.
32 id: RemoteLogSegmentId,
33 /// Why the add was rejected.
34 reason: String,
35 },
36
37 /// A caller requested a lifecycle transition that the state machine
38 /// forbids.
39 #[error("invalid segment state transition for {id:?}: {from:?} -> {to:?}")]
40 InvalidSegmentTransition {
41 /// The segment whose transition was rejected.
42 id: RemoteLogSegmentId,
43 /// Current state.
44 from: RemoteLogSegmentState,
45 /// Requested state.
46 to: RemoteLogSegmentState,
47 },
48
49 /// A caller requested a partition-delete lifecycle transition that the
50 /// state machine forbids.
51 #[error("invalid partition delete transition for {tp:?}: {from:?} -> {to:?}")]
52 InvalidPartitionDeleteTransition {
53 /// The partition whose transition was rejected.
54 tp: TopicIdPartition,
55 /// Current state (`None` when the partition was never marked).
56 from: Option<RemotePartitionDeleteState>,
57 /// Requested state.
58 to: RemotePartitionDeleteState,
59 },
60
61 /// Constructor / argument validation failed.
62 #[error("invalid argument: {0}")]
63 InvalidArgument(String),
64
65 /// A conditional create found an object already at the key.
66 ///
67 /// A write-once archive turns on `PutMode::Create` for every object it
68 /// writes, so this is how it learns that a key is already taken rather
69 /// than silently replacing what is there.
70 #[error("object already exists: {key}")]
71 ObjectExists {
72 /// Object-store key that already holds an object.
73 key: String,
74 },
75
76 /// A conditional write's precondition did not hold, for example an `ETag`
77 /// that no longer matches.
78 #[error("precondition failed for `{key}`: {reason}")]
79 PreconditionFailed {
80 /// Object-store key whose precondition was evaluated.
81 key: String,
82 /// Why the backend rejected the write.
83 reason: String,
84 },
85
86 /// A scan of a remote segment's records met a compressed batch with a record
87 /// whose body is larger than the caller's limit. This is Kafka trunk's
88 /// `max.decompressed.message.bytes`, which `DefaultRecord.readFrom` enforces
89 /// with an `InvalidRecordException`.
90 ///
91 /// The segment is fine and the answer is known: the lookup fails, and it
92 /// fails the same way on every retry until the limit is raised.
93 #[error("Invalid record size: {size} exceeds the configured maximum record size of {limit}.")]
94 RecordTooLarge {
95 /// The record's body size in bytes, Kafka's `sizeOfBodyInBytes`.
96 size: usize,
97 /// The configured maximum.
98 limit: usize,
99 },
100
101 /// The write-once (WORM) archive layer refused or could not complete the
102 /// operation.
103 #[error("write-once archive: {0}")]
104 Worm(#[from] WormError),
105
106 /// A backend, for example an object store, raised an error. The error
107 /// does not map cleanly to one of the structured variants above.
108 #[error("remote storage backend error: {0}")]
109 Backend(String),
110
111 /// This broker holds the metadata partition that would answer this
112 /// query, but its consumer has not yet caught up to the high-water mark
113 /// that the broker observed when it took the assignment. The answer is
114 /// unknown, not "no segment". Callers should retry rather than treat it
115 /// as a definitive miss. `Ok(None)` is reserved for "caught up, no
116 /// covering segment" and for partitions this broker does not consume at
117 /// all.
118 #[error("remote log metadata partition {partition} not ready (assigned but not caught up)")]
119 NotReady {
120 /// The `__remote_log_metadata` partition that is still catching up.
121 partition: i32,
122 },
123}
124
125impl From<krabka_object_store::ObjectStoreError> for RemoteStorageError {
126 fn from(err: krabka_object_store::ObjectStoreError) -> Self {
127 use krabka_object_store::ObjectStoreError as E;
128
129 match err {
130 E::Io(e) => Self::Io(e),
131 E::InvalidConfig(m) => Self::InvalidArgument(m),
132 // Engine methods with segment context match NotFound before converting.
133 E::NotFound(p) => Self::Backend(format!("not found: {p}")),
134 E::TooLarge {
135 key,
136 size,
137 max_bytes,
138 } => Self::Backend(format!(
139 "object `{key}` is {size} bytes, exceeds cap of {max_bytes} bytes"
140 )),
141 E::AlreadyExists(p) => Self::ObjectExists { key: p.to_string() },
142 E::Precondition { key, reason } => Self::PreconditionFailed {
143 key: key.to_string(),
144 reason,
145 },
146 E::Backend(m) => Self::Backend(m),
147 }
148 }
149}
150
151#[cfg(test)]
152mod tests {
153 use {
154 super::*,
155 assert2::{assert, check},
156 object_store::path::Path,
157 };
158
159 /// The write-once archive branches on these two, so each must arrive as
160 /// its own structured variant and not as an opaque `Backend` string.
161 #[test]
162 fn conditional_write_errors_convert_to_structured_variants() {
163 let already = RemoteStorageError::from(
164 krabka_object_store::ObjectStoreError::AlreadyExists(Path::from("worm/manifest")),
165 );
166 check!(
167 matches!(&already, RemoteStorageError::ObjectExists { key } if key == "worm/manifest")
168 );
169 check!(already.to_string() == "object already exists: worm/manifest");
170
171 let precondition =
172 RemoteStorageError::from(krabka_object_store::ObjectStoreError::Precondition {
173 key: Path::from("worm/manifest"),
174 reason: "etag mismatch".to_owned(),
175 });
176 check!(matches!(
177 &precondition,
178 RemoteStorageError::PreconditionFailed { key, reason }
179 if key == "worm/manifest" && reason == "etag mismatch"
180 ));
181 check!(
182 precondition.to_string() == "precondition failed for `worm/manifest`: etag mismatch"
183 );
184 }
185
186 /// `?` on a [`WormError`] must land in one variant that still renders the
187 /// underlying refusal, so a log line names the key that was protected.
188 #[test]
189 fn worm_errors_convert_and_keep_their_message() {
190 let err = RemoteStorageError::from(WormError::DeleteRefused {
191 key: "archive/orders-0/0.log".to_owned(),
192 });
193
194 check!(matches!(
195 err,
196 RemoteStorageError::Worm(WormError::DeleteRefused { .. })
197 ));
198 check!(
199 err.to_string()
200 == "write-once archive: delete refused: `archive/orders-0/0.log` is in a write-once archive"
201 );
202 }
203
204 #[test]
205 fn object_store_too_large_converts_to_backend_message() {
206 let err = krabka_object_store::ObjectStoreError::TooLarge {
207 key: Path::from("index/snapshot.json"),
208 size: 1000,
209 max_bytes: 256,
210 };
211
212 let got = RemoteStorageError::from(err);
213
214 assert!(matches!(
215 got,
216 RemoteStorageError::Backend(ref msg)
217 if msg == "object `index/snapshot.json` is 1000 bytes, exceeds cap of 256 bytes"
218 ));
219 }
220}