use tracing::warn;
use crate::catalog::DataFileId;
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum Error {
#[error("commit conflict: {0}")]
CommitConflict(String),
#[error("retry budget exhausted: {0}")]
RetryBudgetExhausted(String),
#[error("corruption: {0}")]
Corruption(String),
#[error("not found: {0}")]
NotFound(String),
#[error("already exists: {0}")]
AlreadyExists(String),
#[error("constraint violation: {0}")]
Constraint(String),
#[error("unsupported: {0}")]
Unsupported(String),
#[error("snapshot expired: {0}")]
SnapshotExpired(String),
#[error("interrupted: {0}")]
Interrupted(String),
#[error("migration required: {0}")]
Migration(String),
#[error("index building: {0}")]
IndexBuilding(String),
#[error("configuration: {0}")]
Configuration(String),
#[error("writer fenced: {0}")]
Fenced(String),
#[error("open raced: {0}")]
OpenRaced(String),
#[error("store error")]
Store(#[source] Box<slatedb::Error>),
#[error("concurrent modification")]
ConcurrentModification,
#[error("row {row_id} could not be positioned against data file {data_file_id:?}: {reason}")]
RowPosition {
row_id: u64,
data_file_id: Option<DataFileId>,
reason: String,
},
}
pub(crate) const MANIFEST_VERSION_EXISTS: &str =
"transactional object (e.g. manifest) version already exists";
pub(crate) fn fenced() -> Error {
warn!("another process attached this catalog read-write; this writer is fenced");
Error::Fenced(
"another process attached this catalog read-write and took over as \
the writer; this handle can no longer commit — re-attach to write"
.to_string(),
)
}
impl From<slatedb::Error> for Error {
fn from(err: slatedb::Error) -> Self {
if err.kind() == slatedb::ErrorKind::Closed(slatedb::CloseReason::Fenced) {
return fenced();
}
if err.kind() == slatedb::ErrorKind::Data
&& err.to_string().contains(MANIFEST_VERSION_EXISTS)
{
warn!("another process created this store first; nothing was written");
return Self::OpenRaced(
"another process created this store while this open was creating it; \
nothing was written — open again to adopt the store it created"
.to_string(),
);
}
Self::Store(Box::new(err))
}
}
pub type Result<T> = std::result::Result<T, Error>;
#[cfg(test)]
mod tests {
use super::Error;
const RETRY_SUBSTRINGS: [&str; 4] = ["conflict", "concurrent", "unique", "primary key"];
#[test]
fn only_commit_conflict_carries_a_retry_substring() {
let sample = "index \"unique_by_primary key\" saw a concurrent conflict";
let non_retryable = [
Error::RetryBudgetExhausted(sample.into()),
Error::Corruption(sample.into()),
Error::NotFound(sample.into()),
Error::AlreadyExists(sample.into()),
Error::Constraint(sample.into()),
Error::IndexBuilding(sample.into()),
Error::Configuration(sample.into()),
Error::Fenced(sample.into()),
Error::OpenRaced(sample.into()),
Error::Unsupported(sample.into()),
Error::SnapshotExpired(sample.into()),
Error::Interrupted(sample.into()),
Error::Migration(sample.into()),
Error::RowPosition {
row_id: 1,
data_file_id: None,
reason: sample.into(),
},
];
for err in non_retryable {
let rendered = err.to_string();
let prefix = rendered
.strip_suffix(sample)
.expect("every variant renders as `<prefix>{sample}`")
.to_lowercase();
for needle in RETRY_SUBSTRINGS {
assert!(
!prefix.contains(needle),
"{prefix:?} carries retry substring {needle:?}"
);
}
}
assert!(
Error::CommitConflict(String::new())
.to_string()
.contains("conflict"),
"CommitConflict must stay retryable"
);
}
}