slatedb_mirror/
error.rs

1use std::error::Error;
2
3use object_store::path::Path;
4use slatedb_common::retry::NonRetryable;
5
6/// The `store` name on the [`object_store::Error::Generic`] that carries a
7/// [`MirrorError`].
8pub const MIRROR_STORE_NAME: &str = "ObjectStoreMirror";
9
10/// Errors raised by the mirror itself, as opposed to errors from the wrapped
11/// store.
12///
13/// The mirror returns these wrapped in [`object_store::Error::Generic`] with
14/// a [`NonRetryable`] wrapper around the `MirrorError` as the source.
15/// Use [`MirrorError::find`] to get the original error.
16#[non_exhaustive]
17#[derive(Debug, thiserror::Error)]
18pub enum MirrorError {
19    /// A `Local` route missed.
20    #[error("object is not in the local mirror. path=`{path}`")]
21    NotLocal { path: Path },
22
23    /// Local disk I/O failed. Nothing was written remotely.
24    #[error("local mirror I/O failed")]
25    Local { source: std::io::Error },
26
27    /// A write reached the remote store, but `observe` or local installation
28    /// failed afterwards. The remote object exists.
29    #[error("write reached the remote store but did not finish locally. path=`{path}`")]
30    WriteCommitted {
31        path: Path,
32        source: Box<dyn Error + Send + Sync>,
33    },
34
35    /// The policy returned an error from a route or `observe` call.
36    #[error("mirror policy failed")]
37    Policy {
38        source: Box<dyn Error + Send + Sync>,
39    },
40
41    /// The mirror doesn't support this operation: COPY or RENAME.
42    #[error("operation not supported by the mirror. operation=`{operation}`")]
43    Unsupported { operation: &'static str },
44
45    /// `ObjectStoreMirrorBuilder::build` was given an invalid option.
46    #[error("invalid mirror configuration. message=`{message}`")]
47    InvalidConfig { message: String },
48}
49
50impl MirrorError {
51    /// Returns the first `MirrorError` in `err`'s source chain, including `err`
52    /// itself.
53    pub fn find<'a>(err: &'a (dyn Error + 'static)) -> Option<&'a MirrorError> {
54        let mut current = Some(err);
55        while let Some(err) = current {
56            if let Some(mirror_err) = err.downcast_ref::<MirrorError>() {
57                return Some(mirror_err);
58            }
59            // `io::Error::source` skips the wrapped error and returns its
60            // source, so check the wrapped error directly.
61            if let Some(mirror_err) = err
62                .downcast_ref::<std::io::Error>()
63                .and_then(|err| err.get_ref())
64                .and_then(|inner| inner.downcast_ref::<MirrorError>())
65            {
66                return Some(mirror_err);
67            }
68            current = err.source();
69        }
70        None
71    }
72}
73
74impl From<MirrorError> for object_store::Error {
75    fn from(err: MirrorError) -> Self {
76        object_store::Error::Generic {
77            store: MIRROR_STORE_NAME,
78            source: Box::new(NonRetryable(Box::new(err))),
79        }
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use std::sync::Arc;
86
87    use super::*;
88
89    fn not_local() -> MirrorError {
90        MirrorError::NotLocal {
91            path: Path::from("a/b.sst"),
92        }
93    }
94
95    #[test]
96    fn should_find_mirror_error_itself() {
97        let err = not_local();
98        assert!(matches!(
99            MirrorError::find(&err),
100            Some(MirrorError::NotLocal { .. })
101        ));
102    }
103
104    #[test]
105    fn should_find_mirror_error_in_object_store_error() {
106        let err = object_store::Error::from(not_local());
107        assert!(matches!(
108            MirrorError::find(&err),
109            Some(MirrorError::NotLocal { .. })
110        ));
111    }
112
113    #[test]
114    fn should_find_mirror_error_through_arc_and_io_error() {
115        let err = Arc::new(object_store::Error::from(not_local()));
116        assert!(MirrorError::find(&err).is_some());
117
118        let err = std::io::Error::other(object_store::Error::from(not_local()));
119        assert!(MirrorError::find(&err).is_some());
120
121        let err = std::io::Error::other(not_local());
122        assert!(MirrorError::find(&err).is_some());
123    }
124
125    #[test]
126    fn should_not_find_mirror_error_in_other_errors() {
127        let err = object_store::Error::Generic {
128            store: "S3",
129            source: Box::new(std::io::Error::other("timeout")),
130        };
131        assert!(MirrorError::find(&err).is_none());
132    }
133}
slatedb_mirror/error.rs

slatedb_mirror/
error.rs
