krabka_raft/error.rs
1use {
2 crate::types::NodeId,
3 krabka_protocol::records::RecordsError,
4 std::{io, net::SocketAddr, path::PathBuf},
5 thiserror::Error,
6 wincode::error::{ReadError, WriteError},
7};
8
9/// What is wrong with a node-local file whose layout carries a version
10/// marker. [`RaftError::PersistedFormat`] names the file.
11#[derive(Debug, Clone, PartialEq, Eq, Error)]
12pub enum PersistedFormatError {
13 /// The file has the layout a krabka build before 1.0 wrote, with no
14 /// version marker. 1.0 reads no 0.x data, so the node is reformatted.
15 #[error(
16 "has no version marker, so a krabka build before 1.0 wrote it; reformat this node \
17 with krabka-format"
18 )]
19 MissingVersion,
20 /// The version marker names a version this build does not read, such as
21 /// one a later build wrote.
22 #[error("has version {found}, and this build reads versions {min} to {max}")]
23 UnsupportedVersion { found: i64, min: i64, max: i64 },
24 /// The file does not parse at a version this build reads.
25 #[error("does not parse: {0}")]
26 Malformed(String),
27}
28
29/// A committed metadata log record that a controller cannot replay.
30///
31/// It stops the controller, as Kafka's `QuorumController` hands a record it
32/// cannot replay to `SharedServer.fatalQuorumControllerFaultHandler`, and a
33/// controller-role node's `MetadataLoader` hands one it cannot load to
34/// `SharedServer.metadataLoaderFaultHandler`, which is fatal on a node whose
35/// `process.roles` include `controller`. Both handlers halt the process.
36#[derive(Debug, PartialEq, Eq, Error)]
37pub enum MetadataReplayError {
38 /// The record's bytes are not a metadata record this build reads: an
39 /// unknown apiKey, a value version above the highest it supports, a frame
40 /// version other than 1, or a krabka-private record that does not decode.
41 #[error("the metadata record at offset {offset} does not decode: {error}")]
42 UndecodableRecord {
43 /// The record's offset in the metadata log.
44 offset: i64,
45 /// Why it does not decode.
46 error: krabka_metadata::TranslateError,
47 },
48 /// A KIP-853 control record that does not decode, or that names a
49 /// `kraft.version` or voter set Kafka would refuse. Kafka's
50 /// `KRaftControlRecordStateMachine` throws while it reads such a record,
51 /// and `KafkaRaftClientDriver` hands the throw to
52 /// `SharedServer.raftManagerFaultHandler`, which is fatal on every role.
53 #[error("the KRaft control record at offset {offset} is invalid: {reason}")]
54 InvalidControlRecord {
55 /// The record's offset in the metadata log.
56 offset: i64,
57 /// Why it is invalid.
58 reason: String,
59 },
60}
61
62#[derive(Debug, Error)]
63#[non_exhaustive]
64pub enum RaftError {
65 #[error("storage: {0}")]
66 Storage(#[from] krabka_log::LogError),
67
68 #[error("network: {0}")]
69 Network(#[from] krabka_client_core::ClientError),
70
71 #[error("protocol: {0}")]
72 Protocol(#[from] krabka_protocol::ProtocolError),
73
74 #[error("controller admin dispatch: {0}")]
75 ControllerAdmin(String),
76
77 #[error("records: {0}")]
78 Records(#[from] RecordsError),
79
80 /// A record of a control batch lacks its key or value, or has an empty
81 /// one. The message is the text of Kafka's
82 /// `RecordsIterator.decodeControlRecord`.
83 #[error("{0}")]
84 MalformedControlRecord(&'static str),
85
86 #[error("metadata: {0}")]
87 Metadata(#[from] krabka_metadata::MetadataError),
88
89 #[error("unsupported: {0}")]
90 Unsupported(&'static str),
91
92 #[error("not leader; current leader: {current_leader:?}")]
93 NotLeader { current_leader: Option<NodeId> },
94
95 #[error("leader unknown (election in progress)")]
96 LeaderUnknown,
97
98 #[error("change rejected: {0}")]
99 ChangeRejected(String),
100
101 /// A leader-local mutation was planned from an image that has since changed.
102 /// The caller must read a fresh image and repeat its admission checks.
103 #[error("the metadata image changed before the mutation was submitted")]
104 StaleMetadataImage,
105
106 /// The leader refused a compare-and-set because its log holds records
107 /// that are not committed yet.
108 ///
109 /// A break-glass consume and a topic-freeze replacement check the
110 /// committed image, so the leader decides them only when its whole log is
111 /// committed. A newly elected leader is in this state until it commits a
112 /// record from its own epoch. Kafka's controller is not active in the same
113 /// window, and it answers `NOT_CONTROLLER`. The refusal clears when the
114 /// tail commits, so a caller can retry.
115 ///
116 /// A diskless offset reservation gets the same refusal until the leader
117 /// has committed its own epoch. The leader reserves no offsets before
118 /// that.
119 #[error("the controller leader has uncommitted metadata records")]
120 UncommittedTail,
121
122 /// The controller refused a request because the principal of the
123 /// connection lacks the cluster operation that the request needs.
124 #[error("the controller denied the cluster operation to this node's principal")]
125 ClusterAuthorizationFailed,
126
127 #[error("reconfiguration rejected: {0}")]
128 ReconfigRejected(String),
129
130 #[error("a reconfiguration is already in progress")]
131 ReconfigInProgress,
132
133 #[error("voter {0} already exists")]
134 DuplicateVoter(NodeId),
135
136 #[error("voter {0} was not found")]
137 VoterNotFound(NodeId),
138
139 #[error("invalid voter update: {0}")]
140 InvalidVoterUpdate(String),
141
142 #[error("kraft.version {0} does not permit dynamic voter changes")]
143 UnsupportedKraftVersion(u16),
144
145 #[error("voter {id} is not a caught-up observer (lag {lag})")]
146 VoterNotCaughtUp { id: NodeId, lag: u64 },
147
148 #[error("serialization: {0}")]
149 SerdeFailed(#[from] WriteError),
150
151 #[error("deserialization: {0}")]
152 SerdeFailedDecode(#[from] ReadError),
153
154 /// A node-local file in the metadata partition directory has a version
155 /// this build does not read, or does not parse. The node does not start
156 /// on it.
157 #[error("{artifact} file {}: {problem}", .path.display())]
158 PersistedFormat {
159 /// The file's name, such as `quorum-state`.
160 artifact: &'static str,
161 path: PathBuf,
162 problem: PersistedFormatError,
163 },
164
165 /// A peer answered a krabka-private controller RPC with
166 /// `UNSUPPORTED_VERSION`: it does not implement `version` of `api_key`.
167 /// A mixed-version cluster meets it during a rolling upgrade.
168 #[error("the peer does not implement version {version} of krabka-private api {api_key}")]
169 UnsupportedPrivateVersion { api_key: i16, version: i16 },
170
171 /// A committed metadata log record that this controller cannot replay.
172 /// The controller stops.
173 #[error("metadata replay: {0}")]
174 MetadataReplay(#[from] MetadataReplayError),
175
176 #[error("startup misconfiguration: {0}")]
177 Startup(String),
178
179 /// The controller listener could not bind its address. Kafka's
180 /// `SocketServer` fails its start the same way, with `Socket server failed
181 /// to bind to` and the address. It is a network failure, not a storage
182 /// one.
183 #[error("unable to bind the controller listener to {addr}: {source}")]
184 ControllerListenerBind {
185 /// The address the listener tried to bind.
186 addr: SocketAddr,
187 /// Why the bind failed, such as `Address already in use`.
188 source: io::Error,
189 },
190
191 /// The controller met a fault that Kafka's `ProcessTerminatingFaultHandler`
192 /// halts the process over, such as a replay of a `FeatureLevelRecord`
193 /// above the range this controller supports. The text is the
194 /// `Encountered fatal fault: ` prefix of that handler's log line, then the
195 /// message of the exception that caused the fault. Kafka's own
196 /// `failureMessage` between the two, such as `Unable to apply
197 /// FeatureLevelRecord record at offset N on active controller, ...`, is
198 /// not part of it.
199 #[error("Encountered fatal fault: {0}")]
200 FatalFault(String),
201
202 #[error("controller shut down")]
203 Shutdown,
204}
205
206#[cfg(test)]
207mod tests {
208
209 use super::*;
210
211 #[test]
212 fn display_not_leader_with_id() {
213 let e = RaftError::NotLeader {
214 current_leader: Some(NodeId(7)),
215 };
216 assert2::assert!(e.to_string().contains("Some(NodeId(7))"));
217 }
218
219 #[test]
220 fn display_not_leader_without_id() {
221 let e = RaftError::NotLeader {
222 current_leader: None,
223 };
224 assert2::assert!(e.to_string().contains("None"));
225 }
226}