qdrant_edge/segment/common/
operation_error.rs

1use std::backtrace::Backtrace;
2use std::collections::TryReserveError;
3use std::io::{Error as IoError, ErrorKind};
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::Duration;
6
7use atomicwrites::Error as AtomicIoError;
8use crate::common::mmap::Error as MmapError;
9use crate::common::universal_io::UniversalIoError;
10use crate::gridstore::error::GridstoreError;
11use rayon::ThreadPoolBuildError;
12use thiserror::Error;
13
14use crate::segment::types::{PayloadKeyType, PointIdType, SeqNumberType, VectorNameBuf};
15use crate::segment::utils::mem::Mem;
16
17pub const PROCESS_CANCELLED_BY_SERVICE_MESSAGE: &str = "process cancelled by service";
18
19#[derive(Error, Debug, Clone, PartialEq)]
20#[error("{0}")]
21pub enum OperationError {
22    #[error("Vector dimension error: expected dim: {expected_dim}, got {received_dim}")]
23    WrongVectorDimension {
24        expected_dim: usize,
25        received_dim: usize,
26    },
27    #[error("Not existing vector name error: {received_name}")]
28    VectorNameNotExists { received_name: VectorNameBuf },
29    #[error("No point with id {missed_point_id}")]
30    PointIdError { missed_point_id: PointIdType },
31    #[error(
32        "Payload type does not match with previously given for field {field_name}. Expected: {expected_type}"
33    )]
34    TypeError {
35        field_name: PayloadKeyType,
36        expected_type: String,
37    },
38    #[error("Unable to infer type for the field '{field_name}'. Please specify `field_type`")]
39    TypeInferenceError { field_name: PayloadKeyType },
40    /// Service Error prevents further update of the collection until it is fixed.
41    /// Should only be used for hardware, data corruption, IO, or other unexpected internal errors.
42    #[error("Service runtime error: {description}")]
43    ServiceError {
44        description: String,
45        backtrace: Option<String>,
46    },
47    #[error("Inconsistent storage: {description}")]
48    InconsistentStorage { description: String },
49    #[error("Out of memory, free: {free}, {description}")]
50    OutOfMemory { description: String, free: u64 },
51    #[error("Operation cancelled: {description}")]
52    Cancelled { description: String },
53    #[error("Timeout error: {description}")]
54    Timeout { description: String },
55    #[error("Validation failed: {description}")]
56    ValidationError { description: String },
57    #[error("Wrong usage of sparse vectors")]
58    WrongSparse,
59    #[error("Wrong usage of multi vectors")]
60    WrongMulti,
61    #[error(
62        "No range index for `order_by` key: `{key}`. Please create one to use `order_by`. Check https://qdrant.tech/documentation/concepts/indexing/#payload-index to see which payload schemas support Range conditions"
63    )]
64    MissingRangeIndexForOrderBy { key: String },
65    #[error(
66        "No appropriate index for faceting: `{key}`. Please create one to facet on this field. Check https://qdrant.tech/documentation/concepts/indexing/#payload-index to see which payload schemas support Match conditions"
67    )]
68    MissingMapIndexForFacet { key: String },
69    #[error(
70        "Expected {expected_type} value for {field_name} in the payload and/or in the formula defaults. Error: {description}"
71    )]
72    VariableTypeError {
73        field_name: PayloadKeyType,
74        expected_type: String,
75        description: String,
76    },
77    #[error("The expression {expression} produced a non-finite number")]
78    NonFiniteNumber { expression: String },
79}
80
81impl OperationError {
82    /// Create a new service error with a description and a backtrace
83    /// Warning: capturing a backtrace can be an expensive operation on some platforms, so this should be used with caution in performance-sensitive parts of code.
84    pub fn service_error(description: impl Into<String>) -> Self {
85        Self::ServiceError {
86            description: description.into(),
87            backtrace: Some(Backtrace::force_capture().to_string()),
88        }
89    }
90
91    /// Create a new service error with a description and no backtrace
92    pub fn service_error_light(description: impl Into<String>) -> Self {
93        Self::ServiceError {
94            description: description.into(),
95            backtrace: None,
96        }
97    }
98
99    pub fn validation_error(description: impl Into<String>) -> Self {
100        Self::ValidationError {
101            description: description.into(),
102        }
103    }
104
105    pub fn inconsistent_storage(description: impl Into<String>) -> Self {
106        Self::InconsistentStorage {
107            description: description.into(),
108        }
109    }
110
111    pub fn cancelled(description: impl Into<String>) -> Self {
112        Self::Cancelled {
113            description: description.into(),
114        }
115    }
116
117    pub fn vector_name_not_exists(vector_name: impl Into<String>) -> Self {
118        Self::VectorNameNotExists {
119            received_name: vector_name.into(),
120        }
121    }
122
123    pub fn timeout(timeout: Duration, operation: impl Into<String>) -> Self {
124        Self::Timeout {
125            description: format!(
126                "Operation '{}' timed out after {timeout:?}",
127                operation.into(),
128            ),
129        }
130    }
131}
132
133/// Contains information regarding last operation error, which should be fixed before next operation could be processed
134#[derive(Debug, Clone)]
135pub struct SegmentFailedState {
136    pub version: SeqNumberType,
137    pub point_id: Option<PointIdType>,
138    pub error: OperationError,
139}
140
141impl From<ThreadPoolBuildError> for OperationError {
142    fn from(error: ThreadPoolBuildError) -> Self {
143        Self::service_error(error.to_string())
144    }
145}
146
147impl From<MmapError> for OperationError {
148    fn from(err: MmapError) -> Self {
149        Self::service_error(err.to_string())
150    }
151}
152
153impl From<UniversalIoError> for OperationError {
154    fn from(err: UniversalIoError) -> Self {
155        match err {
156            UniversalIoError::Io(err) => Self::from(err),
157            UniversalIoError::Mmap(err) => Self::from(err),
158
159            UniversalIoError::BytemuckCast(_)
160            | UniversalIoError::ZerocopySize(_)
161            | UniversalIoError::IoUringNotSupported(_)
162            | UniversalIoError::NotFound { .. }
163            | UniversalIoError::OutOfBounds { .. }
164            | UniversalIoError::InvalidFileIndex { .. }
165            | UniversalIoError::Uninitialized { .. }
166            | UniversalIoError::QueueIsFull
167            | UniversalIoError::S3(_)
168            | UniversalIoError::S3Config { .. }
169            | UniversalIoError::TaskPanicked(_) => Self::service_error(err.to_string()),
170        }
171    }
172}
173
174impl<Src, Dst: ?Sized> From<zerocopy::SizeError<Src, Dst>> for OperationError
175where
176    zerocopy::SizeError<Src, Dst>: std::fmt::Display,
177{
178    fn from(err: zerocopy::SizeError<Src, Dst>) -> Self {
179        Self::service_error(format!("Zerocopy size error: {err}"))
180    }
181}
182
183impl<A, S, V> From<zerocopy::ConvertError<A, S, V>> for OperationError
184where
185    zerocopy::ConvertError<A, S, V>: std::fmt::Display,
186{
187    fn from(err: zerocopy::ConvertError<A, S, V>) -> Self {
188        Self::service_error(format!("Zerocopy convert error: {err}"))
189    }
190}
191
192impl From<serde_cbor::Error> for OperationError {
193    fn from(err: serde_cbor::Error) -> Self {
194        Self::service_error(format!("Failed to parse data: {err}"))
195    }
196}
197
198impl<E> From<AtomicIoError<E>> for OperationError {
199    fn from(err: AtomicIoError<E>) -> Self {
200        match err {
201            AtomicIoError::Internal(io_err) => Self::from(io_err),
202            AtomicIoError::User(_user_err) => Self::service_error("Unknown atomic write error"),
203        }
204    }
205}
206
207impl From<IoError> for OperationError {
208    fn from(err: IoError) -> Self {
209        #[expect(clippy::wildcard_enum_match_arm, reason = "error handling")]
210        match err.kind() {
211            ErrorKind::OutOfMemory => {
212                let free_memory = Mem::new().available_memory_bytes();
213                Self::OutOfMemory {
214                    description: format!("IO Error: {err}"),
215                    free: free_memory,
216                }
217            }
218            _ => Self::service_error(format!("IO Error: {err}")),
219        }
220    }
221}
222
223impl From<serde_json::Error> for OperationError {
224    fn from(err: serde_json::Error) -> Self {
225        Self::service_error(format!("Json error: {err}"))
226    }
227}
228
229impl From<fs_extra::error::Error> for OperationError {
230    fn from(err: fs_extra::error::Error) -> Self {
231        Self::service_error(format!("File system error: {err}"))
232    }
233}
234
235impl From<geohash::GeohashError> for OperationError {
236    fn from(err: geohash::GeohashError) -> Self {
237        Self::service_error(format!("Geohash error: {err}"))
238    }
239}
240
241impl From<crate::quantization::EncodingError> for OperationError {
242    fn from(err: crate::quantization::EncodingError) -> Self {
243        match err {
244            crate::quantization::EncodingError::IOError(err)
245            | crate::quantization::EncodingError::EncodingError(err)
246            | crate::quantization::EncodingError::ArgumentsError(err) => {
247                Self::service_error(format!("Quantization encoding error: {err}"))
248            }
249            crate::quantization::EncodingError::Stopped => {
250                Self::cancelled(PROCESS_CANCELLED_BY_SERVICE_MESSAGE)
251            }
252        }
253    }
254}
255
256impl From<TryReserveError> for OperationError {
257    fn from(err: TryReserveError) -> Self {
258        let free_memory = Mem::new().available_memory_bytes();
259        Self::OutOfMemory {
260            description: format!("Failed to reserve memory: {err}"),
261            free: free_memory,
262        }
263    }
264}
265
266impl From<GridstoreError> for OperationError {
267    fn from(err: GridstoreError) -> Self {
268        match err {
269            GridstoreError::ServiceError { description } => {
270                Self::service_error(format!("Gridstore error: {description}"))
271            }
272            GridstoreError::FlushCancelled => Self::cancelled("Gridstore flushing was cancelled"),
273            GridstoreError::Io(_) | GridstoreError::Mmap(_) | GridstoreError::SerdeJson(_) => {
274                Self::service_error(err.to_string())
275            }
276            GridstoreError::ValidationError { message } => Self::validation_error(message),
277            GridstoreError::UniversalIo(err) => {
278                Self::service_error(format!("Gridstore IO error: {err}"))
279            }
280            GridstoreError::PageNotFound { .. } => Self::service_error(err.to_string()),
281            GridstoreError::ValueNotFound { .. } => Self::service_error(err.to_string()),
282        }
283    }
284}
285
286#[cfg(feature = "gpu")]
287impl From<gpu::GpuError> for OperationError {
288    fn from(err: gpu::GpuError) -> Self {
289        Self::service_error(format!("GPU error: {err:?}"))
290    }
291}
292
293pub type OperationResult<T> = Result<T, OperationError>;
294
295pub fn get_service_error<T>(err: &OperationResult<T>) -> Option<OperationError> {
296    match err {
297        Ok(_) => None,
298        #[expect(clippy::wildcard_enum_match_arm, reason = "error handling")]
299        Err(error) => match error {
300            OperationError::ServiceError { .. } => Some(error.clone()),
301            _ => None,
302        },
303    }
304}
305
306#[derive(Debug, Copy, Clone)]
307pub struct CancelledError;
308
309pub type CancellableResult<T> = Result<T, CancelledError>;
310
311impl From<CancelledError> for OperationError {
312    fn from(CancelledError: CancelledError) -> Self {
313        Self::cancelled(PROCESS_CANCELLED_BY_SERVICE_MESSAGE)
314    }
315}
316
317pub fn check_process_stopped(stopped: &AtomicBool) -> CancellableResult<()> {
318    if stopped.load(Ordering::Relaxed) {
319        return Err(CancelledError);
320    }
321    Ok(())
322}
323
324#[cfg(test)]
325mod tests {
326    use std::time::Duration;
327
328    use super::*;
329
330    #[test]
331    fn test_timeout_error_formatting() {
332        // Test sub-second timeout (500ms)
333        let timeout = Duration::from_millis(500);
334        let error = OperationError::timeout(timeout, "test operation");
335        let error_msg = error.to_string();
336        assert!(
337            error_msg.contains("500ms"),
338            "Expected '500ms' but got: {error_msg}"
339        );
340
341        // Test exact second timeout (1000ms = 1s)
342        let timeout = Duration::from_millis(1000);
343        let error = OperationError::timeout(timeout, "test operation");
344        let error_msg = error.to_string();
345        assert!(
346            error_msg.contains("1s"),
347            "Expected '1s' but got: {error_msg}"
348        );
349
350        // Test multi-second timeout with sub-second precision (2500ms = 2.5s)
351        let timeout = Duration::from_millis(2500);
352        let error = OperationError::timeout(timeout, "test operation");
353        let error_msg = error.to_string();
354        assert!(
355            error_msg.contains("2.5s"),
356            "Expected '2.5s' but got: {error_msg}"
357        );
358
359        // Test large timeout (60000ms = 60s)
360        let timeout = Duration::from_millis(60000);
361        let error = OperationError::timeout(timeout, "test operation");
362        let error_msg = error.to_string();
363        assert!(
364            error_msg.contains("60s"),
365            "Expected '60s' but got: {error_msg}"
366        );
367    }
368}
qdrant_edge/segment/common/operation_error.rs

qdrant_edge/segment/common/
operation_error.rs
