rocketmq_client_rust/
client_error.rs

1// Copyright 2026 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::error::Error as StdError;
16use std::fmt;
17use std::ops::Deref;
18use std::sync::Arc;
19
20use rocketmq_error::fields;
21use rocketmq_error::Error;
22use rocketmq_error::ErrorContext;
23use rocketmq_error::ErrorDescriptor;
24use rocketmq_error::SharedError;
25use rocketmq_error::ViewValueRef;
26
27/// Opaque client boundary error backed by one shared canonical error.
28///
29/// Client operations cross asynchronous callbacks and retry coordinators, so
30/// the canonical value is shared rather than copied or flattened into text.
31#[derive(Clone)]
32pub struct ClientError(SharedError);
33
34/// Result returned by RocketMQ client operations.
35pub type ClientResult<T> = std::result::Result<T, ClientError>;
36
37impl ClientError {
38    /// Promotes an owned canonical error into the client boundary.
39    #[must_use]
40    pub fn from_error(error: Error) -> Self {
41        Self(Arc::new(error))
42    }
43
44    /// Retains an already shared canonical error at the client boundary.
45    #[must_use]
46    pub const fn from_shared(error: SharedError) -> Self {
47        Self(error)
48    }
49
50    /// Returns the canonical error descriptor.
51    #[must_use]
52    pub fn descriptor(&self) -> &'static ErrorDescriptor {
53        self.0.descriptor()
54    }
55
56    /// Returns whether this error has the supplied canonical identity.
57    #[must_use]
58    pub fn is(&self, descriptor: &'static ErrorDescriptor) -> bool {
59        self.descriptor().code() == descriptor.code()
60    }
61
62    /// Returns the retained broker response code when this is a broker failure.
63    #[must_use]
64    pub fn broker_response_code(&self) -> Option<i32> {
65        self.diagnostic_i64(fields::BROKER_CODE.schema().name())
66            .and_then(|code| i32::try_from(code).ok())
67    }
68
69    /// Returns the bounded diagnostic broker address when present.
70    #[must_use]
71    pub fn broker_addr(&self) -> Option<&str> {
72        self.diagnostic_text(fields::BROKER_ADDR.schema().name())
73    }
74
75    /// Borrows the shared canonical error.
76    #[must_use]
77    pub const fn shared_error(&self) -> &SharedError {
78        &self.0
79    }
80
81    /// Finds a typed cause anywhere below the canonical client error.
82    #[must_use]
83    pub fn source_ref<T>(&self) -> Option<&T>
84    where
85        T: StdError + 'static,
86    {
87        let mut current: &(dyn StdError + 'static) = self.0.as_ref();
88        while let Some(source) = current.source() {
89            if let Some(found) = source.downcast_ref::<T>() {
90                return Some(found);
91            }
92            current = source;
93        }
94        None
95    }
96
97    /// Returns the shared canonical error.
98    #[must_use]
99    pub fn into_shared_error(self) -> SharedError {
100        self.0
101    }
102
103    /// Returns an owned canonical error while preserving the complete typed
104    /// source chain when the shared value has other owners.
105    #[must_use]
106    pub fn into_error(self) -> Error {
107        match Arc::try_unwrap(self.0) {
108            Ok(error) => error,
109            Err(shared) => {
110                let descriptor = shared.descriptor();
111                let context = shared.context().clone();
112                Error::caused_by(descriptor, ClientError(shared)).with_context(context)
113            }
114        }
115    }
116
117    /// Creates an invalid-argument failure without retaining arbitrary text.
118    #[must_use]
119    pub fn illegal_argument(message: impl Into<String>) -> Self {
120        let message = message.into();
121        let mut context = ErrorContext::new();
122        if !message.is_empty() {
123            context = context.with_secret_presence(fields::MESSAGE_PRESENT);
124        }
125        Self::from_error(Error::new(&rocketmq_error::CORE_ARGUMENT_INVALID).with_context(context))
126    }
127
128    /// Creates an invalid-argument failure with a typed source.
129    #[must_use]
130    pub fn illegal_argument_source(source: impl StdError + Send + Sync + 'static) -> Self {
131        Self::from_error(
132            Error::caused_by(&rocketmq_error::CORE_ARGUMENT_INVALID, source)
133                .with_context(ErrorContext::new().with_secret_presence(fields::MESSAGE_PRESENT)),
134        )
135    }
136
137    /// Creates a client-not-started lifecycle failure.
138    #[must_use]
139    pub fn not_started() -> Self {
140        Self::from_error(Error::new(&rocketmq_error::CLIENT_LIFECYCLE_NOT_STARTED))
141    }
142
143    /// Creates a client-already-started lifecycle failure.
144    #[must_use]
145    pub fn already_started() -> Self {
146        Self::from_error(Error::new(&rocketmq_error::CLIENT_LIFECYCLE_ALREADY_STARTED))
147    }
148
149    /// Creates a client-shutting-down lifecycle failure.
150    #[must_use]
151    pub fn shutting_down() -> Self {
152        Self::from_error(Error::new(&rocketmq_error::CLIENT_LIFECYCLE_SHUTTING_DOWN))
153    }
154
155    /// Creates an invalid client-state failure.
156    #[must_use]
157    pub fn invalid_state(expected: impl AsRef<str>, actual: impl AsRef<str>) -> Self {
158        let context = ErrorContext::new()
159            .with_text(fields::EXPECTED_STATE, expected)
160            .with_text(fields::ACTUAL_STATE, actual);
161        Self::from_error(Error::new(&rocketmq_error::CLIENT_LIFECYCLE_INVALID_STATE).with_context(context))
162    }
163
164    /// Creates an unavailable client-component failure.
165    #[must_use]
166    pub fn component_unavailable(role: impl AsRef<str>) -> Self {
167        let context = ErrorContext::new().with_text(fields::CLIENT_ROLE, role);
168        Self::from_error(Error::new(&rocketmq_error::CLIENT_COMPONENT_UNAVAILABLE).with_context(context))
169    }
170
171    /// Creates a component-not-initialized failure.
172    #[must_use]
173    pub fn not_initialized(component: impl AsRef<str>) -> Self {
174        let context = ErrorContext::new().with_text(fields::COMPONENT_NAME, component);
175        Self::from_error(Error::new(&rocketmq_error::CORE_LIFECYCLE_NOT_INITIALIZED).with_context(context))
176    }
177
178    /// Creates a configuration-missing failure.
179    #[must_use]
180    pub fn config_missing(key: impl AsRef<str>) -> Self {
181        let context = ErrorContext::new().with_text(fields::KEY, key);
182        Self::from_error(Error::new(&rocketmq_error::CORE_CONFIGURATION_MISSING).with_context(context))
183    }
184
185    /// Creates an invalid-configuration failure.
186    #[must_use]
187    pub fn config_invalid(key: impl AsRef<str>, value: impl AsRef<str>, reason: impl AsRef<str>) -> Self {
188        let mut context = ErrorContext::new().with_text(fields::KEY, key);
189        if !value.as_ref().is_empty() {
190            context = context.with_secret_presence(fields::VALUE_PRESENT);
191        }
192        if !reason.as_ref().is_empty() {
193            context = context.with_secret_presence(fields::REASON_PRESENT);
194        }
195        Self::from_error(Error::new(&rocketmq_error::CORE_CONFIGURATION_INVALID).with_context(context))
196    }
197
198    /// Creates an invalid-configuration failure with a typed source.
199    #[must_use]
200    pub fn config_invalid_source(
201        key: impl AsRef<str>,
202        value_present: bool,
203        source: impl StdError + Send + Sync + 'static,
204    ) -> Self {
205        let mut context = ErrorContext::new().with_text(fields::KEY, key);
206        if value_present {
207            context = context.with_secret_presence(fields::VALUE_PRESENT);
208        }
209        context = context.with_secret_presence(fields::REASON_PRESENT);
210        Self::from_error(Error::caused_by(&rocketmq_error::CORE_CONFIGURATION_INVALID, source).with_context(context))
211    }
212
213    /// Creates a configuration-parse failure.
214    #[must_use]
215    pub fn config_parse_failed(key: impl AsRef<str>, reason: impl AsRef<str>) -> Self {
216        let mut context = ErrorContext::new().with_text(fields::KEY, key);
217        if !reason.as_ref().is_empty() {
218            context = context.with_secret_presence(fields::REASON_PRESENT);
219        }
220        Self::from_error(Error::new(&rocketmq_error::CORE_CONFIGURATION_PARSE_FAILED).with_context(context))
221    }
222
223    /// Creates a configuration-parse failure with a typed source.
224    #[must_use]
225    pub fn config_parse_source(key: impl AsRef<str>, source: impl StdError + Send + Sync + 'static) -> Self {
226        let context = ErrorContext::new()
227            .with_text(fields::KEY, key)
228            .with_secret_presence(fields::REASON_PRESENT);
229        Self::from_error(
230            Error::caused_by(&rocketmq_error::CORE_CONFIGURATION_PARSE_FAILED, source).with_context(context),
231        )
232    }
233
234    /// Creates a response-processing failure.
235    #[must_use]
236    pub fn response_process_failed(operation: impl AsRef<str>, reason: impl AsRef<str>) -> Self {
237        let mut context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, operation);
238        if !reason.as_ref().is_empty() {
239            context = context.with_secret_presence(fields::REASON_PRESENT);
240        }
241        Self::from_error(Error::new(&rocketmq_error::PROTOCOL_RESPONSE_FAILED).with_context(context))
242    }
243
244    /// Creates a response-processing failure with a typed source.
245    #[must_use]
246    pub fn response_process_source(operation: impl AsRef<str>, source: impl StdError + Send + Sync + 'static) -> Self {
247        let context = ErrorContext::new()
248            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
249            .with_secret_presence(fields::REASON_PRESENT);
250        Self::from_error(Error::caused_by(&rocketmq_error::PROTOCOL_RESPONSE_FAILED, source).with_context(context))
251    }
252
253    /// Creates an operation-timeout failure.
254    #[must_use]
255    pub fn timeout(operation: impl AsRef<str>, timeout_ms: u64) -> Self {
256        let context = ErrorContext::new()
257            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
258            .with_u64(fields::TIMEOUT_MS, timeout_ms);
259        Self::from_error(Error::new(&rocketmq_error::CORE_OPERATION_TIMED_OUT).with_context(context))
260    }
261
262    /// Creates an internal failure while retaining its typed source.
263    #[must_use]
264    pub fn internal(operation: impl AsRef<str>, source: impl StdError + Send + Sync + 'static) -> Self {
265        let context = ErrorContext::new()
266            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
267            .with_secret_presence(fields::SOURCE_PRESENT);
268        Self::from_error(Error::caused_by(&rocketmq_error::CORE_INTERNAL_FAILURE, source).with_context(context))
269    }
270
271    /// Creates an invariant failure.
272    #[must_use]
273    pub fn invariant_violated(invariant: impl AsRef<str>) -> Self {
274        let context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, invariant);
275        Self::from_error(Error::new(&rocketmq_error::CORE_INTERNAL_FAILURE).with_context(context))
276    }
277
278    /// Creates a service lifecycle failure.
279    #[must_use]
280    pub fn service_failed(operation: impl AsRef<str>) -> Self {
281        let context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, operation);
282        Self::from_error(Error::new(&rocketmq_error::CORE_SERVICE_FAILED).with_context(context))
283    }
284
285    /// Creates a service lifecycle failure with a typed source.
286    #[must_use]
287    pub fn service_source(operation: impl AsRef<str>, source: impl StdError + Send + Sync + 'static) -> Self {
288        let context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, operation);
289        Self::from_error(Error::caused_by(&rocketmq_error::CORE_SERVICE_FAILED, source).with_context(context))
290    }
291
292    /// Creates an invalid-message-property failure.
293    #[must_use]
294    pub fn invalid_property(property: impl AsRef<str>) -> Self {
295        let context = ErrorContext::new().with_text(fields::PROPERTY, property);
296        Self::from_error(Error::new(&rocketmq_error::PROTOCOL_MESSAGE_PROPERTY_INVALID).with_context(context))
297    }
298
299    /// Creates a broker-lookup failure.
300    #[must_use]
301    pub fn broker_not_found(broker: impl AsRef<str>) -> Self {
302        let context = ErrorContext::new().with_text(fields::BROKER, broker);
303        Self::from_error(Error::new(&rocketmq_error::BROKER_LOOKUP_NOT_FOUND).with_context(context))
304    }
305
306    /// Creates a broker-operation failure.
307    #[must_use]
308    pub fn broker_operation_failed(operation: impl AsRef<str>, code: i32, message: impl AsRef<str>) -> Self {
309        let mut context = ErrorContext::new()
310            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
311            .with_i64(fields::BROKER_CODE, i64::from(code));
312        if !message.as_ref().is_empty() {
313            context = context.with_secret_presence(fields::MESSAGE_PRESENT);
314        }
315        Self::from_error(Error::new(&rocketmq_error::BROKER_OPERATION_FAILED).with_context(context))
316    }
317
318    /// Creates a broker-operation failure with an optional broker address.
319    #[must_use]
320    pub fn broker_operation_failed_at(
321        operation: impl AsRef<str>,
322        code: i32,
323        message: impl AsRef<str>,
324        broker_addr: Option<impl AsRef<str>>,
325    ) -> Self {
326        let error = Self::broker_operation_failed(operation, code, message);
327        match broker_addr {
328            Some(broker_addr) => error.with_broker_addr(broker_addr),
329            None => error,
330        }
331    }
332
333    /// Creates a broker-operation failure while retaining its typed source.
334    #[must_use]
335    pub fn broker_operation_source(
336        operation: impl AsRef<str>,
337        code: i32,
338        broker_addr: Option<impl AsRef<str>>,
339        source: impl StdError + Send + Sync + 'static,
340    ) -> Self {
341        let mut context = ErrorContext::new()
342            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
343            .with_i64(fields::BROKER_CODE, i64::from(code))
344            .with_secret_presence(fields::MESSAGE_PRESENT);
345        if let Some(broker_addr) = broker_addr {
346            context = context.with_text(fields::BROKER_ADDR, broker_addr);
347        }
348        Self::from_error(Error::caused_by(&rocketmq_error::BROKER_OPERATION_FAILED, source).with_context(context))
349    }
350
351    /// Adds a broker address to a newly created broker-operation failure.
352    #[must_use]
353    pub fn with_broker_addr(self, broker_addr: impl AsRef<str>) -> Self {
354        match Arc::try_unwrap(self.0) {
355            Ok(error) => {
356                let context = error.context().clone().with_text(fields::BROKER_ADDR, broker_addr);
357                Self::from_error(error.with_context(context))
358            }
359            Err(shared) => Self(shared),
360        }
361    }
362
363    /// Creates a route-not-found failure.
364    #[must_use]
365    pub fn route_not_found(topic: impl AsRef<str>) -> Self {
366        let context = ErrorContext::new().with_text(fields::TOPIC, topic);
367        Self::from_error(Error::new(&rocketmq_error::ROUTE_TOPIC_NOT_FOUND).with_context(context))
368    }
369
370    /// Creates a DNS-resolution failure without exposing endpoint details.
371    #[must_use]
372    pub fn dns_failed() -> Self {
373        Self::from_error(Error::new(&rocketmq_error::TRANSPORT_DNS_FAILED))
374    }
375
376    /// Creates a DNS-resolution failure while retaining its typed source.
377    #[must_use]
378    pub fn dns_failed_source(source: impl StdError + Send + Sync + 'static) -> Self {
379        let context = ErrorContext::new().with_secret_presence(fields::SOURCE_PRESENT);
380        Self::from_error(Error::caused_by(&rocketmq_error::TRANSPORT_DNS_FAILED, source).with_context(context))
381    }
382
383    fn diagnostic_i64(&self, name: &str) -> Option<i64> {
384        self.0
385            .diagnostic_view()
386            .ok()?
387            .fields()
388            .find_map(|field| match (field.name() == name, field.value()) {
389                (true, ViewValueRef::I64(value)) => Some(value),
390                _ => None,
391            })
392    }
393
394    fn diagnostic_text(&self, name: &str) -> Option<&str> {
395        self.0
396            .diagnostic_view()
397            .ok()?
398            .fields()
399            .find_map(|field| match (field.name() == name, field.value()) {
400                (true, ViewValueRef::Text(value)) => Some(value),
401                _ => None,
402            })
403    }
404}
405
406impl From<Error> for ClientError {
407    fn from(error: Error) -> Self {
408        Self::from_error(error)
409    }
410}
411
412impl From<SharedError> for ClientError {
413    fn from(error: SharedError) -> Self {
414        Self::from_shared(error)
415    }
416}
417
418impl From<ClientError> for SharedError {
419    fn from(error: ClientError) -> Self {
420        error.into_shared_error()
421    }
422}
423
424impl AsRef<Error> for ClientError {
425    fn as_ref(&self) -> &Error {
426        self.0.as_ref()
427    }
428}
429
430impl Deref for ClientError {
431    type Target = Error;
432
433    fn deref(&self) -> &Self::Target {
434        self.0.as_ref()
435    }
436}
437
438impl fmt::Display for ClientError {
439    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
440        self.0.fmt(formatter)
441    }
442}
443
444impl fmt::Debug for ClientError {
445    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
446        self.0.fmt(formatter)
447    }
448}
449
450impl StdError for ClientError {
451    fn source(&self) -> Option<&(dyn StdError + 'static)> {
452        Some(self.0.as_ref())
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459
460    #[test]
461    fn into_error_unwraps_sole_owner_preserving_descriptor_and_typed_cause() {
462        let io_error = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset");
463        let client_error = ClientError::response_process_source("some.operation", io_error);
464        let descriptor = client_error.descriptor();
465
466        let error = client_error.into_error();
467
468        assert_eq!(error.descriptor().code(), descriptor.code());
469        let cause = StdError::source(&error).expect("typed cause retained for sole owner");
470        let io_cause = cause.downcast_ref::<std::io::Error>().expect("io error cause retained");
471        assert_eq!(io_cause.kind(), std::io::ErrorKind::ConnectionReset);
472    }
473
474    #[test]
475    fn into_error_wraps_shared_owner_preserving_typed_cause_through_extra_layer() {
476        let io_error = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset");
477        let client_error = ClientError::response_process_source("some.operation", io_error);
478        let descriptor = client_error.descriptor();
479        let clone = client_error.clone();
480
481        let error = client_error.into_error();
482
483        assert_eq!(error.descriptor().code(), descriptor.code());
484        let outer_source = StdError::source(&error).expect("wrapper source present for shared owner");
485        let wrapped_client_error = outer_source
486            .downcast_ref::<ClientError>()
487            .expect("wrapped ClientError retained as the extra layer's source");
488        let io_cause = wrapped_client_error
489            .source_ref::<std::io::Error>()
490            .expect("original typed cause reachable through the extra wrapper");
491        assert_eq!(io_cause.kind(), std::io::ErrorKind::ConnectionReset);
492        drop(clone);
493    }
494
495    #[test]
496    fn into_error_leaves_retained_clone_usable_with_unchanged_shared_identity() {
497        let client_error = ClientError::illegal_argument("boom");
498        let clone = client_error.clone();
499        let original_shared_ptr = Arc::as_ptr(clone.shared_error());
500
501        let _ = client_error.into_error();
502
503        assert!(clone.is(&rocketmq_error::CORE_ARGUMENT_INVALID));
504        assert_eq!(Arc::as_ptr(clone.shared_error()), original_shared_ptr);
505    }
506
507    #[test]
508    fn from_shared_and_into_shared_error_round_trip_preserves_arc_identity() {
509        let canonical: SharedError = Arc::new(Error::new(&rocketmq_error::CORE_ARGUMENT_INVALID));
510        let client_error = ClientError::from_shared(Arc::clone(&canonical));
511
512        let shared = client_error.into_shared_error();
513
514        assert!(Arc::ptr_eq(&canonical, &shared));
515    }
516
517    #[test]
518    fn source_ref_returns_none_for_unrelated_source_type() {
519        let client_error = ClientError::illegal_argument("boom");
520        assert!(client_error.source_ref::<std::io::Error>().is_none());
521    }
522}
rocketmq_client_rust/client_error.rs

rocketmq_client_rust/
client_error.rs
