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::Bincode(_)
160            | UniversalIoError::BytemuckCast(_)
161            | UniversalIoError::ZerocopySize(_)
162            | UniversalIoError::IoUringNotSupported(_)
163            | UniversalIoError::NotFound { .. }
164            | UniversalIoError::OutOfBounds { .. }
165            | UniversalIoError::InvalidFileIndex { .. }
166            | UniversalIoError::Uninitialized { .. }
167            | UniversalIoError::QueueIsFull
168            | UniversalIoError::S3(_)
169            | UniversalIoError::S3Config { .. }
170            | UniversalIoError::TaskPanicked(_) => Self::service_error(err.to_string()),
171        }
172    }
173}
174
175impl<Src, Dst: ?Sized> From<zerocopy::SizeError<Src, Dst>> for OperationError
176where
177    zerocopy::SizeError<Src, Dst>: std::fmt::Display,
178{
179    fn from(err: zerocopy::SizeError<Src, Dst>) -> Self {
180        Self::service_error(format!("Zerocopy size error: {err}"))
181    }
182}
183
184impl<A, S, V> From<zerocopy::ConvertError<A, S, V>> for OperationError
185where
186    zerocopy::ConvertError<A, S, V>: std::fmt::Display,
187{
188    fn from(err: zerocopy::ConvertError<A, S, V>) -> Self {
189        Self::service_error(format!("Zerocopy convert error: {err}"))
190    }
191}
192
193impl From<serde_cbor::Error> for OperationError {
194    fn from(err: serde_cbor::Error) -> Self {
195        Self::service_error(format!("Failed to parse data: {err}"))
196    }
197}
198
199impl<E> From<AtomicIoError<E>> for OperationError {
200    fn from(err: AtomicIoError<E>) -> Self {
201        match err {
202            AtomicIoError::Internal(io_err) => Self::from(io_err),
203            AtomicIoError::User(_user_err) => Self::service_error("Unknown atomic write error"),
204        }
205    }
206}
207
208impl From<IoError> for OperationError {
209    fn from(err: IoError) -> Self {
210        #[expect(clippy::wildcard_enum_match_arm, reason = "error handling")]
211        match err.kind() {
212            ErrorKind::OutOfMemory => {
213                let free_memory = Mem::new().available_memory_bytes();
214                Self::OutOfMemory {
215                    description: format!("IO Error: {err}"),
216                    free: free_memory,
217                }
218            }
219            _ => Self::service_error(format!("IO Error: {err}")),
220        }
221    }
222}
223
224impl From<serde_json::Error> for OperationError {
225    fn from(err: serde_json::Error) -> Self {
226        Self::service_error(format!("Json error: {err}"))
227    }
228}
229
230impl From<fs_extra::error::Error> for OperationError {
231    fn from(err: fs_extra::error::Error) -> Self {
232        Self::service_error(format!("File system error: {err}"))
233    }
234}
235
236impl From<geohash::GeohashError> for OperationError {
237    fn from(err: geohash::GeohashError) -> Self {
238        Self::service_error(format!("Geohash error: {err}"))
239    }
240}
241
242impl From<crate::quantization::EncodingError> for OperationError {
243    fn from(err: crate::quantization::EncodingError) -> Self {
244        match err {
245            crate::quantization::EncodingError::IOError(err)
246            | crate::quantization::EncodingError::EncodingError(err)
247            | crate::quantization::EncodingError::ArgumentsError(err) => {
248                Self::service_error(format!("Quantization encoding error: {err}"))
249            }
250            crate::quantization::EncodingError::Stopped => {
251                Self::cancelled(PROCESS_CANCELLED_BY_SERVICE_MESSAGE)
252            }
253        }
254    }
255}
256
257impl From<TryReserveError> for OperationError {
258    fn from(err: TryReserveError) -> Self {
259        let free_memory = Mem::new().available_memory_bytes();
260        Self::OutOfMemory {
261            description: format!("Failed to reserve memory: {err}"),
262            free: free_memory,
263        }
264    }
265}
266
267impl From<GridstoreError> for OperationError {
268    fn from(err: GridstoreError) -> Self {
269        match err {
270            GridstoreError::ServiceError { description } => {
271                Self::service_error(format!("Gridstore error: {description}"))
272            }
273            GridstoreError::FlushCancelled => Self::cancelled("Gridstore flushing was cancelled"),
274            GridstoreError::Io(_) | GridstoreError::Mmap(_) | GridstoreError::SerdeJson(_) => {
275                Self::service_error(err.to_string())
276            }
277            GridstoreError::ValidationError { message } => Self::validation_error(message),
278            GridstoreError::UniversalIo(err) => {
279                Self::service_error(format!("Gridstore IO error: {err}"))
280            }
281            GridstoreError::PageNotFound { .. } => Self::service_error(err.to_string()),
282            GridstoreError::ValueNotFound { .. } => Self::service_error(err.to_string()),
283        }
284    }
285}
286
287#[cfg(feature = "gpu")]
288impl From<gpu::GpuError> for OperationError {
289    fn from(err: gpu::GpuError) -> Self {
290        Self::service_error(format!("GPU error: {err:?}"))
291    }
292}
293
294pub type OperationResult<T> = Result<T, OperationError>;
295
296pub fn get_service_error<T>(err: &OperationResult<T>) -> Option<OperationError> {
297    match err {
298        Ok(_) => None,
299        #[expect(clippy::wildcard_enum_match_arm, reason = "error handling")]
300        Err(error) => match error {
301            OperationError::ServiceError { .. } => Some(error.clone()),
302            _ => None,
303        },
304    }
305}
306
307#[derive(Debug, Copy, Clone)]
308pub struct CancelledError;
309
310pub type CancellableResult<T> = Result<T, CancelledError>;
311
312impl From<CancelledError> for OperationError {
313    fn from(CancelledError: CancelledError) -> Self {
314        Self::cancelled(PROCESS_CANCELLED_BY_SERVICE_MESSAGE)
315    }
316}
317
318pub fn check_process_stopped(stopped: &AtomicBool) -> CancellableResult<()> {
319    if stopped.load(Ordering::Relaxed) {
320        return Err(CancelledError);
321    }
322    Ok(())
323}
324
325#[cfg(test)]
326mod tests {
327    use std::time::Duration;
328
329    use super::*;
330
331    #[test]
332    fn test_timeout_error_formatting() {
333        // Test sub-second timeout (500ms)
334        let timeout = Duration::from_millis(500);
335        let error = OperationError::timeout(timeout, "test operation");
336        let error_msg = error.to_string();
337        assert!(
338            error_msg.contains("500ms"),
339            "Expected '500ms' but got: {error_msg}"
340        );
341
342        // Test exact second timeout (1000ms = 1s)
343        let timeout = Duration::from_millis(1000);
344        let error = OperationError::timeout(timeout, "test operation");
345        let error_msg = error.to_string();
346        assert!(
347            error_msg.contains("1s"),
348            "Expected '1s' but got: {error_msg}"
349        );
350
351        // Test multi-second timeout with sub-second precision (2500ms = 2.5s)
352        let timeout = Duration::from_millis(2500);
353        let error = OperationError::timeout(timeout, "test operation");
354        let error_msg = error.to_string();
355        assert!(
356            error_msg.contains("2.5s"),
357            "Expected '2.5s' but got: {error_msg}"
358        );
359
360        // Test large timeout (60000ms = 60s)
361        let timeout = Duration::from_millis(60000);
362        let error = OperationError::timeout(timeout, "test operation");
363        let error_msg = error.to_string();
364        assert!(
365            error_msg.contains("60s"),
366            "Expected '60s' but got: {error_msg}"
367        );
368    }
369}
qdrant_edge/segment/common/operation_error.rs

qdrant_edge/segment/common/
operation_error.rs
