mqtt5_protocol/
error_classification.rs

1use crate::error::MqttError;
2use crate::protocol::v5::reason_codes::ReasonCode;
3
4/// A transient fault for which retrying the *same* operation after a delay can
5/// succeed on its own, with no state change or user intervention.
6///
7/// This is the retry/backoff machinery's notion of "recoverable", not a general
8/// judgement about whether the failure is the caller's fault. An error is
9/// recoverable here only if the fix is "wait, then try again": the operation's
10/// preconditions still hold and something outside the caller's control is
11/// expected to clear. Faults whose remedy is a state transition (re-establishing
12/// a connection), a different request (smaller payload), or human action
13/// (fixing credentials) are deliberately *not* recoverable, because a naive
14/// backoff loop on them would spin forever.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
16pub enum RecoverableError {
17    /// Transient network fault (timeout, reset, unreachable). Retry after backoff.
18    NetworkError,
19    /// Server temporarily cannot serve the request. Retry after backoff.
20    ServerUnavailable,
21    /// Server-side quota hit. Retry after a longer backoff (see `base_delay_multiplier`).
22    QuotaExceeded,
23    /// No packet identifiers currently free; one frees as in-flight messages ack.
24    PacketIdExhausted,
25    /// Receive-maximum flow-control limit reached; capacity returns as acks arrive.
26    FlowControlLimited,
27    /// Session was taken over by another client using the same client id.
28    SessionTakenOver,
29    /// Server is shutting down; reconnect will land on a fresh instance.
30    ServerShuttingDown,
31    /// Recoverable `MQoQ` flow condition (idle/cancelled/refused flow).
32    MqoqFlowRecoverable,
33}
34
35impl RecoverableError {
36    #[must_use]
37    pub fn base_delay_multiplier(&self) -> u32 {
38        match self {
39            Self::QuotaExceeded => 10,
40            Self::MqoqFlowRecoverable => 3,
41            Self::FlowControlLimited => 2,
42            _ => 1,
43        }
44    }
45
46    #[must_use]
47    pub fn default_set() -> [Self; 6] {
48        [
49            Self::NetworkError,
50            Self::ServerUnavailable,
51            Self::QuotaExceeded,
52            Self::PacketIdExhausted,
53            Self::FlowControlLimited,
54            Self::MqoqFlowRecoverable,
55        ]
56    }
57}
58
59impl MqttError {
60    /// Classify this error for the retry/backoff layer.
61    ///
62    /// Returns `Some(kind)` when retrying the *same* operation after a delay can
63    /// succeed on its own (see [`RecoverableError`]), and `None` when it cannot.
64    ///
65    /// `None` covers three distinct cases, all of which a blind backoff loop
66    /// would handle wrong:
67    ///
68    /// - **Precondition errors** such as [`MqttError::NotConnected`]. These are
69    ///   returned whenever there is no active connection, which covers both a
70    ///   dropped link and calling `publish`/`subscribe` before `connect` or
71    ///   after `disconnect`. The remedy is re-establishing the connection, a
72    ///   state transition owned by the reconnect layer, not re-issuing the same
73    ///   packet. For a `QoS` 1 publish that failed because the link dropped, the
74    ///   intended recovery is reconnect plus resend of unacked messages on the
75    ///   next session, not treating the raw `NotConnected` as retryable here.
76    /// - **Caller/permanent errors** such as authentication failures, bad
77    ///   credentials, authorization denials, and protocol errors. Retrying is
78    ///   futile until the caller changes the request or their configuration.
79    /// - **Connection-limit signals** carried in [`MqttError::ConnectionError`]
80    ///   strings (see [`MqttError::is_aws_iot_connection_limit`]). These look
81    ///   like network resets but indicate the account/client limit was hit;
82    ///   immediate retry makes it worse, so they are excluded on purpose.
83    ///
84    /// This is intentionally conservative: a variant only classifies as
85    /// recoverable when re-issuing the identical operation is the correct
86    /// response. Anything requiring a reconnect, a different request, or human
87    /// intervention returns `None`.
88    #[must_use]
89    pub fn classify(&self) -> Option<RecoverableError> {
90        match self {
91            Self::ConnectionError(msg) => classify_connection_error(msg),
92            Self::ConnectionRefused(reason) => classify_connection_refused(*reason),
93            Self::PacketIdExhausted => Some(RecoverableError::PacketIdExhausted),
94            Self::FlowControlExceeded => Some(RecoverableError::FlowControlLimited),
95            Self::Timeout => Some(RecoverableError::NetworkError),
96            Self::ServerUnavailable | Self::ServerBusy => Some(RecoverableError::ServerUnavailable),
97            Self::QuotaExceeded => Some(RecoverableError::QuotaExceeded),
98            Self::ServerShuttingDown => Some(RecoverableError::ServerShuttingDown),
99            Self::SessionExpired => Some(RecoverableError::SessionTakenOver),
100            _ => None,
101        }
102    }
103
104    #[must_use]
105    pub fn is_aws_iot_connection_limit(&self) -> bool {
106        match self {
107            Self::ConnectionError(msg) => is_aws_iot_limit_error(msg),
108            _ => false,
109        }
110    }
111}
112
113fn classify_connection_error(msg: &str) -> Option<RecoverableError> {
114    if is_aws_iot_limit_error(msg) {
115        return None;
116    }
117
118    if msg.contains("temporarily unavailable")
119        || msg.contains("Connection refused")
120        || msg.contains("Network is unreachable")
121        || msg.contains("connection reset")
122        || msg.contains("broken pipe")
123        || msg.contains("timed out")
124    {
125        return Some(RecoverableError::NetworkError);
126    }
127
128    None
129}
130
131fn is_aws_iot_limit_error(msg: &str) -> bool {
132    msg.contains("Connection reset by peer")
133        || msg.contains("RST")
134        || msg.contains("TCP RST")
135        || msg.contains("reset by peer")
136        || msg.contains("connection limit")
137        || msg.contains("client limit")
138}
139
140fn classify_connection_refused(reason: ReasonCode) -> Option<RecoverableError> {
141    match reason {
142        ReasonCode::ServerUnavailable | ReasonCode::ServerBusy => {
143            Some(RecoverableError::ServerUnavailable)
144        }
145        ReasonCode::QuotaExceeded => Some(RecoverableError::QuotaExceeded),
146        ReasonCode::SessionTakenOver => Some(RecoverableError::SessionTakenOver),
147        ReasonCode::ServerShuttingDown => Some(RecoverableError::ServerShuttingDown),
148        ReasonCode::MqoqIncompletePacket
149        | ReasonCode::MqoqFlowOpenIdle
150        | ReasonCode::MqoqFlowCancelled
151        | ReasonCode::MqoqFlowPacketCancelled
152        | ReasonCode::MqoqFlowRefused => Some(RecoverableError::MqoqFlowRecoverable),
153        _ => None,
154    }
155}
156
157#[cfg(test)]
158mod tests {
159    use super::*;
160
161    #[test]
162    fn test_connection_error_classification() {
163        let error = MqttError::ConnectionError("Connection refused".to_string());
164        assert_eq!(error.classify(), Some(RecoverableError::NetworkError));
165
166        let error = MqttError::ConnectionError("Network is unreachable".to_string());
167        assert_eq!(error.classify(), Some(RecoverableError::NetworkError));
168
169        let error = MqttError::ConnectionError("temporarily unavailable".to_string());
170        assert_eq!(error.classify(), Some(RecoverableError::NetworkError));
171    }
172
173    #[test]
174    fn test_aws_iot_limit_not_recoverable() {
175        let error = MqttError::ConnectionError("Connection reset by peer".to_string());
176        assert_eq!(error.classify(), None);
177        assert!(error.is_aws_iot_connection_limit());
178
179        let error = MqttError::ConnectionError("TCP RST received".to_string());
180        assert_eq!(error.classify(), None);
181        assert!(error.is_aws_iot_connection_limit());
182
183        let error = MqttError::ConnectionError("client limit exceeded".to_string());
184        assert_eq!(error.classify(), None);
185        assert!(error.is_aws_iot_connection_limit());
186    }
187
188    #[test]
189    fn test_connection_refused_classification() {
190        let error = MqttError::ConnectionRefused(ReasonCode::ServerUnavailable);
191        assert_eq!(error.classify(), Some(RecoverableError::ServerUnavailable));
192
193        let error = MqttError::ConnectionRefused(ReasonCode::QuotaExceeded);
194        assert_eq!(error.classify(), Some(RecoverableError::QuotaExceeded));
195
196        let error = MqttError::ConnectionRefused(ReasonCode::SessionTakenOver);
197        assert_eq!(error.classify(), Some(RecoverableError::SessionTakenOver));
198
199        let error = MqttError::ConnectionRefused(ReasonCode::BadUsernameOrPassword);
200        assert_eq!(error.classify(), None);
201    }
202
203    #[test]
204    fn test_mqoq_classification() {
205        let error = MqttError::ConnectionRefused(ReasonCode::MqoqIncompletePacket);
206        assert_eq!(
207            error.classify(),
208            Some(RecoverableError::MqoqFlowRecoverable)
209        );
210
211        let error = MqttError::ConnectionRefused(ReasonCode::MqoqFlowOpenIdle);
212        assert_eq!(
213            error.classify(),
214            Some(RecoverableError::MqoqFlowRecoverable)
215        );
216
217        let error = MqttError::ConnectionRefused(ReasonCode::MqoqFlowCancelled);
218        assert_eq!(
219            error.classify(),
220            Some(RecoverableError::MqoqFlowRecoverable)
221        );
222
223        let error = MqttError::ConnectionRefused(ReasonCode::MqoqNoFlowState);
224        assert_eq!(error.classify(), None);
225    }
226
227    #[test]
228    fn test_direct_error_classification() {
229        assert_eq!(
230            MqttError::PacketIdExhausted.classify(),
231            Some(RecoverableError::PacketIdExhausted)
232        );
233        assert_eq!(
234            MqttError::FlowControlExceeded.classify(),
235            Some(RecoverableError::FlowControlLimited)
236        );
237        assert_eq!(
238            MqttError::Timeout.classify(),
239            Some(RecoverableError::NetworkError)
240        );
241        assert_eq!(
242            MqttError::ServerUnavailable.classify(),
243            Some(RecoverableError::ServerUnavailable)
244        );
245        assert_eq!(
246            MqttError::ServerBusy.classify(),
247            Some(RecoverableError::ServerUnavailable)
248        );
249        assert_eq!(
250            MqttError::QuotaExceeded.classify(),
251            Some(RecoverableError::QuotaExceeded)
252        );
253        assert_eq!(
254            MqttError::ServerShuttingDown.classify(),
255            Some(RecoverableError::ServerShuttingDown)
256        );
257    }
258
259    #[test]
260    fn test_non_recoverable_errors() {
261        assert_eq!(MqttError::NotConnected.classify(), None);
262        assert_eq!(MqttError::AlreadyConnected.classify(), None);
263        assert_eq!(MqttError::AuthenticationFailed.classify(), None);
264        assert_eq!(MqttError::NotAuthorized.classify(), None);
265        assert_eq!(MqttError::BadUsernameOrPassword.classify(), None);
266        assert_eq!(
267            MqttError::ProtocolError("test".to_string()).classify(),
268            None
269        );
270    }
271
272    #[test]
273    fn test_base_delay_multiplier() {
274        assert_eq!(RecoverableError::NetworkError.base_delay_multiplier(), 1);
275        assert_eq!(
276            RecoverableError::ServerUnavailable.base_delay_multiplier(),
277            1
278        );
279        assert_eq!(RecoverableError::QuotaExceeded.base_delay_multiplier(), 10);
280        assert_eq!(
281            RecoverableError::FlowControlLimited.base_delay_multiplier(),
282            2
283        );
284        assert_eq!(
285            RecoverableError::MqoqFlowRecoverable.base_delay_multiplier(),
286            3
287        );
288    }
289
290    #[test]
291    fn test_default_set() {
292        let defaults = RecoverableError::default_set();
293        assert_eq!(defaults.len(), 6);
294        assert!(defaults.contains(&RecoverableError::NetworkError));
295        assert!(defaults.contains(&RecoverableError::ServerUnavailable));
296        assert!(defaults.contains(&RecoverableError::QuotaExceeded));
297        assert!(defaults.contains(&RecoverableError::PacketIdExhausted));
298        assert!(defaults.contains(&RecoverableError::FlowControlLimited));
299        assert!(defaults.contains(&RecoverableError::MqoqFlowRecoverable));
300        assert!(!defaults.contains(&RecoverableError::SessionTakenOver));
301        assert!(!defaults.contains(&RecoverableError::ServerShuttingDown));
302    }
303}
mqtt5_protocol/error_classification.rs

mqtt5_protocol/
error_classification.rs
