use std::borrow::Cow;
use std::error::Error as StdError;
use thiserror::Error;
use crate::capability::StorageCapability;
#[derive(Debug, Error)]
pub enum StorageError {
#[error("{capability:?} resource not found: {resource} ({key})")]
NotFound {
capability: StorageCapability,
resource: &'static str,
key: String,
},
#[error("{capability:?} resource already exists: {resource} ({key})")]
AlreadyExists {
capability: StorageCapability,
resource: &'static str,
key: String,
},
#[error("conflict in {capability:?} during {operation}: {message}")]
Conflict {
capability: StorageCapability,
operation: Cow<'static, str>,
message: String,
},
#[error("invalid input for {capability:?} during {operation}: {message}")]
InvalidInput {
capability: StorageCapability,
operation: Cow<'static, str>,
message: String,
},
#[error("unsupported operation for {capability:?}: {operation} ({message})")]
Unsupported {
capability: StorageCapability,
operation: Cow<'static, str>,
message: String,
},
#[error("pool failure during {operation}: {message}")]
Pool {
operation: Cow<'static, str>,
message: String,
},
#[error("timeout during {operation}")]
Timeout { operation: Cow<'static, str> },
#[error("sql transaction failure during {operation}: {message}")]
Transaction {
operation: Cow<'static, str>,
message: String,
},
#[error("serialization failure in {capability:?}: {message}")]
Serialization {
capability: StorageCapability,
message: String,
},
#[error("index maintenance failure in {capability:?}: {message}")]
IndexMaintenance {
capability: StorageCapability,
message: String,
},
#[error("backend driver error in {capability:?} during {operation}: {source}")]
Driver {
capability: StorageCapability,
operation: Cow<'static, str>,
#[source]
source: Box<dyn StdError + Send + Sync>,
},
#[error("write queue full: timed out after {timeout_ms}ms waiting for writer task capacity")]
WriteQueueFull { timeout_ms: u64 },
#[error("internal storage error: {0}")]
Internal(String),
#[error(
"KHIVE_WRITE_QUEUE=1 but no Tokio runtime context is available to spawn the writer task"
)]
WriterTaskNoRuntime,
#[error(
"refusing write on {capability:?} at {volume}: {available_bytes} bytes available, \
below the {floor_bytes}-byte floor"
)]
CapacityFloor {
capability: StorageCapability,
volume: String,
available_bytes: u64,
floor_bytes: u64,
},
}
impl StorageError {
pub fn driver(
capability: StorageCapability,
operation: impl Into<Cow<'static, str>>,
source: impl StdError + Send + Sync + 'static,
) -> Self {
Self::Driver {
capability,
operation: operation.into(),
source: Box::new(source),
}
}
pub fn capability(&self) -> Option<StorageCapability> {
match self {
Self::NotFound { capability, .. }
| Self::AlreadyExists { capability, .. }
| Self::Conflict { capability, .. }
| Self::InvalidInput { capability, .. }
| Self::Unsupported { capability, .. }
| Self::Serialization { capability, .. }
| Self::IndexMaintenance { capability, .. }
| Self::Driver { capability, .. }
| Self::CapacityFloor { capability, .. } => Some(*capability),
Self::Pool { .. }
| Self::Timeout { .. }
| Self::Transaction { .. }
| Self::WriteQueueFull { .. }
| Self::Internal(..)
| Self::WriterTaskNoRuntime => None,
}
}
pub fn is_retryable(&self) -> bool {
matches!(
self,
Self::Pool { .. }
| Self::Timeout { .. }
| Self::Transaction { .. }
| Self::WriteQueueFull { .. }
)
}
pub fn is_fts5_syntax_error(&self) -> bool {
let Self::Driver {
capability,
operation,
source,
} = self
else {
return false;
};
if *capability != StorageCapability::Text || operation.as_ref() != "fts_search" {
return false;
}
let msg = source.to_string();
msg.contains("fts5: syntax error")
|| msg.contains("fts5: parser stack overflow")
|| msg.contains("fts5: column queries are not supported")
|| msg.contains("fts5: phrase queries are not supported (detail")
|| msg.contains("fts5: NEAR queries are not supported (detail")
}
pub fn is_unique_constraint_violation(&self) -> bool {
let Self::Driver {
capability,
operation,
source,
} = self
else {
return false;
};
if *capability != StorageCapability::Sql {
return false;
}
if !matches!(
operation.as_ref(),
"execute" | "pool_writer.execute" | "tx.execute"
) {
return false;
}
source.to_string().contains("UNIQUE constraint failed")
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::fmt;
#[derive(Debug)]
struct FakeSource(String);
impl fmt::Display for FakeSource {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", self.0)
}
}
impl StdError for FakeSource {}
fn driver_err(operation: &'static str, message: &str) -> StorageError {
StorageError::driver(
StorageCapability::Text,
operation,
FakeSource(message.into()),
)
}
#[test]
fn fts5_syntax_error_at_fts_search_is_classified_as_syntax_error() {
let e = driver_err("fts_search", "fts5: syntax error near \"@\"");
assert!(e.is_fts5_syntax_error());
}
#[test]
fn fts5_parser_stack_overflow_is_classified_as_syntax_error() {
let e = driver_err("fts_search", "fts5: parser stack overflow");
assert!(e.is_fts5_syntax_error());
}
#[test]
fn fts5_unsupported_column_query_is_classified_as_syntax_error() {
let e = driver_err(
"fts_search",
"fts5: column queries are not supported (detail=none)",
);
assert!(e.is_fts5_syntax_error());
}
#[test]
fn timeout_is_not_classified_as_syntax_error() {
let e = StorageError::Timeout {
operation: "fts_search".into(),
};
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn pool_failure_is_not_classified_as_syntax_error() {
let e = StorageError::Pool {
operation: "fts_search".into(),
message: "pool exhausted".into(),
};
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn driver_error_at_non_search_operation_is_not_classified_as_syntax_error() {
let e = driver_err("open_fts_reader", "fts5: syntax error near \"@\"");
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn driver_error_with_unrelated_message_is_not_classified_as_syntax_error() {
let e = driver_err("fts_search", "disk I/O error");
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn fts5_phrase_detail_query_is_classified_as_syntax_error() {
let e = driver_err(
"fts_search",
"fts5: phrase queries are not supported (detail!=full)",
);
assert!(e.is_fts5_syntax_error());
}
#[test]
fn fts5_near_detail_query_is_classified_as_syntax_error() {
let e = driver_err(
"fts_search",
"fts5: NEAR queries are not supported (detail!=full)",
);
assert!(e.is_fts5_syntax_error());
}
#[test]
fn unprefixed_detail_message_is_not_classified_as_syntax_error() {
let e = driver_err(
"fts_search",
"phrase queries are not supported (detail!=full)",
);
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn fts5_shadow_table_corruption_is_not_classified_as_syntax_error() {
let e = driver_err(
"fts_search",
"fts5: error creating shadow table notes_content: no such table",
);
assert!(!e.is_fts5_syntax_error());
}
#[test]
fn non_text_capability_is_not_classified_as_syntax_error() {
let e = StorageError::Driver {
capability: StorageCapability::Vectors,
operation: "fts_search".into(),
source: Box::new(FakeSource("fts5: syntax error near \"@\"".into())),
};
assert!(!e.is_fts5_syntax_error());
}
fn driver_err_sql(operation: &'static str, message: &str) -> StorageError {
StorageError::driver(
StorageCapability::Sql,
operation,
FakeSource(message.into()),
)
}
#[test]
fn unique_constraint_failure_at_execute_sql_capability_is_classified() {
let e = driver_err_sql(
"execute",
"UNIQUE constraint failed: brain_serve_ledger.namespace, \
brain_serve_ledger.target_id, brain_serve_ledger.query_class, \
brain_serve_ledger.served_at",
);
assert!(e.is_unique_constraint_violation());
}
#[test]
fn unique_constraint_failure_at_pool_writer_execute_is_classified() {
let e = driver_err_sql("pool_writer.execute", "UNIQUE constraint failed: t.id");
assert!(e.is_unique_constraint_violation());
}
#[test]
fn unique_constraint_message_at_non_execute_operation_is_not_classified() {
let e = driver_err_sql("query_row", "UNIQUE constraint failed: t.id");
assert!(!e.is_unique_constraint_violation());
}
#[test]
fn non_unique_driver_error_at_execute_is_not_classified() {
let e = driver_err_sql("execute", "disk I/O error");
assert!(!e.is_unique_constraint_violation());
}
#[test]
fn non_sql_capability_is_not_classified_as_unique_violation() {
let e = driver_err("execute", "UNIQUE constraint failed: t.id");
assert!(!e.is_unique_constraint_violation());
}
#[test]
fn timeout_is_not_classified_as_unique_violation() {
let e = StorageError::Timeout {
operation: "execute".into(),
};
assert!(!e.is_unique_constraint_violation());
}
}