use reifydb_core::{common::CommitVersion, interface::catalog::view::ViewKind};
use reifydb_value::{
error::{Diagnostic, Error, IntoDiagnostic},
fragment::Fragment,
};
use crate::dictionary::error::DictionaryError;
#[derive(Debug, thiserror::Error)]
pub enum TransactionError {
#[error("Transaction conflict detected")]
Conflict,
#[error("Transaction rolled back and cannot be committed")]
RolledBack,
#[error("Transaction contains too many writes and exceeds size limits")]
TooLarge,
#[error("Transaction open too long - conflict history has been evicted")]
TooOld,
#[error("Transaction was already committed")]
AlreadyCommitted,
#[error("Transaction was already rolled back")]
AlreadyRolledBack,
#[error("Key '{key}' is not in the transaction's declared key scope")]
KeyOutOfScope {
key: String,
},
#[error("Transaction was poisoned by a prior error")]
Poisoned {
cause: Box<Diagnostic>,
},
#[error("Snapshot version {} evicted by GC; cutoff is {}", version.0, cutoff.0)]
SnapshotVersionEvicted {
version: CommitVersion,
cutoff: CommitVersion,
},
#[error("Consumer overtaken: batch at version {} needs history below the reclaim cutoff {}", version.0, cutoff.0)]
ConsumerOvertaken {
version: CommitVersion,
cutoff: CommitVersion,
},
#[error("Database is shutting down; new transactions are rejected")]
ShuttingDown,
#[error("View '{view}' cannot be read: this transaction holds unprocessed changes to upstream {upstream:?}")]
ViewPendingUpstreamChanges {
view: String,
kind: ViewKind,
upstream: Vec<String>,
fragment: Fragment,
},
#[error(transparent)]
Dictionary(#[from] DictionaryError),
}
impl TransactionError {
pub const CONSUMER_OVERTAKEN: &'static str = "TXN_016";
pub const SHUTTING_DOWN: &'static str = "TXN_014";
pub const SNAPSHOT_EVICTED: &'static str = "TXN_012";
pub fn is_consumer_overtaken(error: &Error) -> bool {
error.0.code == Self::CONSUMER_OVERTAKEN
}
pub fn is_snapshot_evicted(error: &Error) -> bool {
error.0.code == Self::SNAPSHOT_EVICTED
}
pub fn is_shutting_down(error: &Error) -> bool {
error.0.code == Self::SHUTTING_DOWN
}
}
impl IntoDiagnostic for TransactionError {
fn into_diagnostic(self) -> Diagnostic {
match self {
TransactionError::Conflict => Diagnostic {
code: "TXN_001".to_string(),
rql: None,
message: "Transaction conflict detected - another transaction modified the same data".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Retry the transaction".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::RolledBack => Diagnostic {
code: "TXN_002".to_string(),
rql: None,
message: "Transaction rolled back and cannot be committed".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Start a new transaction".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::TooLarge => Diagnostic {
code: "TXN_003".to_string(),
rql: None,
message: "Transaction contains too many writes and exceeds size limits".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Split the transaction into smaller batches".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::TooOld => Diagnostic {
code: "TXN_004".to_string(),
rql: None,
message: "Transaction open too long - the conflict history for this read snapshot has been evicted".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Start a new transaction".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::AlreadyCommitted => Diagnostic {
code: "TXN_008".to_string(),
rql: None,
message: "Transaction was already committed".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Cannot use a transaction after it has been committed".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::AlreadyRolledBack => Diagnostic {
code: "TXN_009".to_string(),
rql: None,
message: "Transaction was already rolled back".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Cannot use a transaction after it has been rolled back".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::KeyOutOfScope { key } => Diagnostic {
code: "TXN_010".to_string(),
rql: None,
message: format!("Key '{}' is not in the transaction's declared key scope", key),
column: None,
fragment: Fragment::None,
label: None,
help: Some(
"Declare the key when beginning the transaction or use a different transaction scope"
.to_string(),
),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::Poisoned { cause } => Diagnostic {
code: "TXN_011".to_string(),
rql: None,
message: "Transaction was poisoned by a prior error".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("A previous statement failed, invalidating this transaction. Start a new transaction.".to_string()),
notes: vec![],
cause: Some(cause),
operator_chain: None,
},
TransactionError::SnapshotVersionEvicted { version, cutoff } => Diagnostic {
code: TransactionError::SNAPSHOT_EVICTED.to_string(),
rql: None,
message: format!(
"Snapshot version {} evicted by historical GC; current cutoff is {}",
version.0, cutoff.0
),
column: None,
fragment: Fragment::None,
label: None,
help: Some(
"Acquire the hydration lease against a more recent version, or subscribe with WITH { hydration: { enabled: false } }."
.to_string(),
),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::ConsumerOvertaken { version, cutoff } => Diagnostic {
code: TransactionError::CONSUMER_OVERTAKEN.to_string(),
rql: None,
message: format!(
"Consumer overtaken: batch at version {} needs history below the reclaim cutoff {}",
version.0, cutoff.0
),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The consumer lagged past retained MVCC history and must resync from a \
fresh snapshot instead of resuming its position"
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::ShuttingDown => Diagnostic {
code: TransactionError::SHUTTING_DOWN.to_string(),
rql: None,
message: "Database is shutting down; new transactions are rejected".to_string(),
column: None,
fragment: Fragment::None,
label: Some("shutdown in progress".to_string()),
help: Some("Retry once the instance has finished shutting down or restarted".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
},
TransactionError::ViewPendingUpstreamChanges { view, kind, upstream, fragment } => {
let (message, help) = match kind {
ViewKind::Transactional => (
format!(
"Transactional view '{}' cannot be read in this transaction: it is derived from {}, which this transaction has modified; those changes are applied to the view only at commit, so a read now would return stale contents",
view,
upstream.join(", ")
),
"Split the write and the read into separate requests, or read from the view's sources directly instead of the view",
),
ViewKind::Deferred => (
format!(
"Deferred view '{}' cannot be read in this transaction: it is derived from {}, which this transaction has modified; the view is updated asynchronously after commit, so a read now would return stale contents",
view,
upstream.join(", ")
),
"Deferred views update asynchronously after commit; read the view's sources directly, or consume the view through a subscription",
),
};
Diagnostic {
code: "TXN_015".to_string(),
rql: None,
message,
column: None,
fragment,
label: Some("view read after upstream write".to_string()),
help: Some(help.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
TransactionError::Dictionary(err) => err.into_diagnostic(),
}
}
}
impl From<TransactionError> for Error {
fn from(err: TransactionError) -> Self {
Error(Box::new(err.into_diagnostic()))
}
}
impl From<DictionaryError> for Error {
fn from(err: DictionaryError) -> Self {
TransactionError::from(err).into()
}
}