use std::fmt::{self, Write as _};
use snafu::Snafu;
use crate::{DiagnosticCode, DiagnosticSet, MssqlVersion, WriteBackend};
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum WritePhase {
SchemaPlanning,
WriterInitialization,
TargetMetadataValidation,
BatchWrite,
BatchSchemaValidation,
ValueConversion,
DirectEncoding,
PacketWrite,
Finish,
Finalize,
}
impl WritePhase {
pub const fn as_str(self) -> &'static str {
match self {
Self::SchemaPlanning => "schema_planning",
Self::WriterInitialization => "writer_initialization",
Self::TargetMetadataValidation => "target_metadata_validation",
Self::BatchWrite => "batch_write",
Self::BatchSchemaValidation => "batch_schema_validation",
Self::ValueConversion => "value_conversion",
Self::DirectEncoding => "direct_encoding",
Self::PacketWrite => "packet_write",
Self::Finish => "finish",
Self::Finalize => "finalize",
}
}
}
impl fmt::Display for WritePhase {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ErrorInfo<'a> {
phase: Option<WritePhase>,
kind: &'static str,
summary: &'static str,
diagnostics: Option<&'a DiagnosticSet>,
}
impl<'a> ErrorInfo<'a> {
pub const fn phase(&self) -> Option<WritePhase> {
self.phase
}
pub const fn kind(&self) -> &'static str {
self.kind
}
pub const fn summary(&self) -> &'static str {
self.summary
}
pub const fn diagnostics(&self) -> Option<&'a DiagnosticSet> {
self.diagnostics
}
pub fn diagnostic_codes(&self) -> String {
if let Some(diagnostics) = self.diagnostics {
diagnostic_codes(diagnostics)
} else if self.kind == "BackendUnavailable" {
format!("{:?}", DiagnosticCode::BackendUnavailable)
} else {
String::new()
}
}
}
#[derive(Debug, Snafu)]
#[non_exhaustive]
pub enum Error {
#[snafu(display("write phase {phase} failed"))]
WritePhaseContext {
phase: WritePhase,
source: Box<Error>,
},
#[snafu(display("invalid compatibility level {level}"))]
InvalidCompatibilityLevel {
level: u16,
},
#[snafu(display(
"{version:?} does not support database compatibility level {compatibility_level}"
))]
UnsupportedCompatibilityLevel {
version: MssqlVersion,
compatibility_level: u16,
},
#[snafu(display("invalid identifier: {reason}"))]
InvalidIdentifier {
reason: String,
},
#[snafu(display("write planning failed with {} diagnostic(s)", diagnostics.len()))]
Planning {
diagnostics: DiagnosticSet,
},
#[snafu(display(
"value conversion failed with {} diagnostic(s)",
diagnostics.len()
))]
ValueConversion {
diagnostics: DiagnosticSet,
},
#[snafu(display(
"direct encoding failed with {} diagnostic(s)",
diagnostics.len()
))]
DirectEncoding {
diagnostics: DiagnosticSet,
},
#[snafu(display("write backend {backend:?} is unavailable: {reason}"))]
BackendUnavailable {
backend: WriteBackend,
reason: String,
},
#[snafu(display("SQL Server connection string is invalid"))]
InvalidConnectionString,
#[snafu(display("TCP connection to SQL Server failed: {source}"))]
ConnectionTcpConnect {
source: std::io::Error,
},
#[snafu(display("SQL Server client setup failed: {source}"))]
ConnectionClientSetup {
source: tiberius::error::Error,
},
#[snafu(display("SQL Server table existence query failed: {source}"))]
TableExistsQuery {
source: tiberius::error::Error,
},
#[snafu(display("SQL Server table existence query returned an unexpected result: {reason}"))]
TableExistsUnexpectedResult {
reason: String,
},
#[snafu(display("SQL Server target row count query failed: {source}"))]
TargetRowCountQuery {
source: tiberius::error::Error,
},
#[snafu(display("SQL Server target row count query returned an unexpected result: {reason}"))]
TargetRowCountUnexpectedResult {
reason: String,
},
#[snafu(display("SQL Server statement execution failed: {source}"))]
SqlExecution {
source: tiberius::error::Error,
},
#[snafu(display("tiberius operation failed: {source}"))]
Tiberius {
source: tiberius::error::Error,
},
}
impl Error {
#[must_use]
pub fn with_write_phase(self, phase: WritePhase) -> Self {
Self::WritePhaseContext {
phase,
source: Box::new(self),
}
}
pub const fn write_phase(&self) -> Option<WritePhase> {
match self {
Self::WritePhaseContext { phase, .. } => Some(*phase),
Self::Planning { .. } => Some(WritePhase::SchemaPlanning),
Self::ValueConversion { .. } => Some(WritePhase::ValueConversion),
Self::DirectEncoding { .. } => Some(WritePhase::DirectEncoding),
_ => None,
}
}
pub fn without_write_phase(&self) -> &Self {
match self {
Self::WritePhaseContext { source, .. } => source.without_write_phase(),
_ => self,
}
}
pub fn safe_error_info(&self) -> ErrorInfo<'_> {
let inner = self.without_write_phase();
ErrorInfo {
phase: self.write_phase(),
kind: inner.kind(),
summary: inner.safe_summary(),
diagnostics: inner.diagnostics(),
}
}
fn kind(&self) -> &'static str {
match self {
Self::WritePhaseContext { source, .. } => source.kind(),
Self::InvalidCompatibilityLevel { .. } => "InvalidCompatibilityLevel",
Self::UnsupportedCompatibilityLevel { .. } => "UnsupportedCompatibilityLevel",
Self::InvalidIdentifier { .. } => "InvalidIdentifier",
Self::Planning { .. } => "Planning",
Self::ValueConversion { .. } => "ValueConversion",
Self::DirectEncoding { .. } => "DirectEncoding",
Self::BackendUnavailable { .. } => "BackendUnavailable",
Self::InvalidConnectionString => "InvalidConnectionString",
Self::ConnectionTcpConnect { .. } => "ConnectionTcpConnect",
Self::ConnectionClientSetup { .. } => "ConnectionClientSetup",
Self::TableExistsQuery { .. } => "TableExistsQuery",
Self::TableExistsUnexpectedResult { .. } => "TableExistsUnexpectedResult",
Self::TargetRowCountQuery { .. } => "TargetRowCountQuery",
Self::TargetRowCountUnexpectedResult { .. } => "TargetRowCountUnexpectedResult",
Self::SqlExecution { .. } => "SqlExecution",
Self::Tiberius { .. } => "Tiberius",
}
}
fn safe_summary(&self) -> &'static str {
match self {
Self::WritePhaseContext { source, .. } => source.safe_summary(),
Self::InvalidCompatibilityLevel { .. } => "invalid compatibility level",
Self::UnsupportedCompatibilityLevel { .. } => {
"compatibility level is not supported by SQL Server version"
}
Self::InvalidIdentifier { .. } => "invalid identifier",
Self::Planning { .. } => "planning failed with diagnostics",
Self::ValueConversion { .. } => "value conversion failed with diagnostics",
Self::DirectEncoding { .. } => "direct encoding failed with diagnostics",
Self::BackendUnavailable { .. } => "write backend unavailable",
Self::InvalidConnectionString => "invalid connection string",
Self::ConnectionTcpConnect { .. } => "TCP connection failed",
Self::ConnectionClientSetup { .. } => "SQL Server client setup failed",
Self::TableExistsQuery { .. } => "table existence query failed",
Self::TableExistsUnexpectedResult { .. } => {
"table existence query returned unexpected result"
}
Self::TargetRowCountQuery { .. } => "target row count query failed",
Self::TargetRowCountUnexpectedResult { .. } => {
"target row count query returned unexpected result"
}
Self::SqlExecution { .. } => "SQL statement execution failed",
Self::Tiberius { .. } => "tiberius operation failed",
}
}
fn diagnostics(&self) -> Option<&DiagnosticSet> {
match self {
Self::WritePhaseContext { source, .. } => source.diagnostics(),
Self::Planning { diagnostics }
| Self::ValueConversion { diagnostics }
| Self::DirectEncoding { diagnostics } => Some(diagnostics),
_ => None,
}
}
}
fn diagnostic_codes(diagnostics: &DiagnosticSet) -> String {
let mut codes = String::new();
for diagnostic in diagnostics.all() {
if !codes.is_empty() {
codes.push(',');
}
let _ = write!(codes, "{:?}", diagnostic.code());
}
codes
}
#[cfg(test)]
mod tests {
use std::{borrow::Cow, error::Error as StdError};
use crate::{
Diagnostic, DiagnosticCode, DiagnosticSet, Error, WritePhase,
observability::{
BATCH_SCHEMA_VALIDATION_PHASE, BATCH_WRITE_PHASE, DIRECT_ENCODING_PHASE,
FINALIZE_PHASE, FINISH_PHASE, PACKET_WRITE_PHASE, SCHEMA_PLANNING_PHASE,
TARGET_METADATA_VALIDATION_PHASE, VALUE_CONVERSION_PHASE, WRITER_INITIALIZATION_PHASE,
},
};
#[test]
fn planning_error_display_includes_diagnostic_count() {
let diagnostics = DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::UnsupportedArrowType,
"unsupported",
)]);
let err = Error::Planning { diagnostics };
assert_eq!(
err.to_string(),
"write planning failed with 1 diagnostic(s)"
);
}
#[test]
fn value_conversion_error_display_includes_diagnostic_count() {
let diagnostics = DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::ValueConversionUnsupported,
"unsupported conversion",
)]);
let err = Error::ValueConversion { diagnostics };
assert_eq!(
err.to_string(),
"value conversion failed with 1 diagnostic(s)"
);
}
#[test]
fn direct_encoding_error_display_includes_diagnostic_count() {
let diagnostics = DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::DirectEncodingInvalidPayload,
"invalid payload",
)]);
let err = Error::DirectEncoding { diagnostics };
assert_eq!(
err.to_string(),
"direct encoding failed with 1 diagnostic(s)"
);
}
#[test]
fn write_phase_names_match_tracing_vocabulary() {
assert_eq!(WritePhase::SchemaPlanning.as_str(), SCHEMA_PLANNING_PHASE);
assert_eq!(
WritePhase::WriterInitialization.as_str(),
WRITER_INITIALIZATION_PHASE
);
assert_eq!(
WritePhase::TargetMetadataValidation.as_str(),
TARGET_METADATA_VALIDATION_PHASE
);
assert_eq!(WritePhase::BatchWrite.as_str(), BATCH_WRITE_PHASE);
assert_eq!(
WritePhase::BatchSchemaValidation.as_str(),
BATCH_SCHEMA_VALIDATION_PHASE
);
assert_eq!(WritePhase::ValueConversion.as_str(), VALUE_CONVERSION_PHASE);
assert_eq!(WritePhase::DirectEncoding.as_str(), DIRECT_ENCODING_PHASE);
assert_eq!(WritePhase::PacketWrite.as_str(), PACKET_WRITE_PHASE);
assert_eq!(WritePhase::Finish.as_str(), FINISH_PHASE);
assert_eq!(WritePhase::Finalize.as_str(), FINALIZE_PHASE);
}
#[test]
fn write_phase_context_classifies_without_display_parsing() {
let err = Error::Tiberius {
source: tiberius::error::Error::BulkInput(Cow::Borrowed("packet failed")),
}
.with_write_phase(WritePhase::PacketWrite);
assert_eq!(err.write_phase(), Some(WritePhase::PacketWrite));
assert_eq!(err.to_string(), "write phase packet_write failed");
assert!(matches!(err.without_write_phase(), Error::Tiberius { .. }));
}
#[test]
fn write_phase_context_display_does_not_include_source_text() {
let err = Error::Tiberius {
source: tiberius::error::Error::BulkInput(Cow::Borrowed(
"server=tcp:sql.example.com;User ID=sa;password=secret",
)),
}
.with_write_phase(WritePhase::Finalize);
assert_eq!(err.to_string(), "write phase finalize failed");
assert!(!err.to_string().contains("password=secret"));
assert!(!err.to_string().contains("User ID=sa"));
assert!(!err.to_string().contains("sql.example.com"));
}
#[test]
fn safe_error_info_exposes_finalize_tiberius_cause_without_source_text() {
let err = Error::Tiberius {
source: tiberius::error::Error::BulkInput(Cow::Borrowed(
"server=tcp:sql.example.com;User ID=sa;password=secret",
)),
}
.with_write_phase(WritePhase::Finalize);
let info = err.safe_error_info();
assert_eq!(info.phase(), Some(WritePhase::Finalize));
assert_eq!(info.kind(), "Tiberius");
assert_eq!(info.summary(), "tiberius operation failed");
assert!(info.diagnostics().is_none());
assert_eq!(info.diagnostic_codes(), "");
assert!(!info.summary().contains("password=secret"));
assert!(!info.summary().contains("User ID=sa"));
assert!(!info.summary().contains("sql.example.com"));
}
#[test]
fn safe_error_info_exposes_nested_diagnostics() {
let err = Error::DirectEncoding {
diagnostics: DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::DirectEncodingInvalidPayload,
"invalid payload",
)]),
}
.with_write_phase(WritePhase::WriterInitialization);
let info = err.safe_error_info();
assert_eq!(info.phase(), Some(WritePhase::WriterInitialization));
assert_eq!(info.kind(), "DirectEncoding");
assert_eq!(info.summary(), "direct encoding failed with diagnostics");
assert_eq!(info.diagnostic_codes(), "DirectEncodingInvalidPayload");
assert_eq!(
info.diagnostics()
.and_then(|diagnostics| diagnostics.all().first())
.map(Diagnostic::code),
Some(DiagnosticCode::DirectEncodingInvalidPayload)
);
}
#[test]
fn unwrapped_diagnostic_errors_keep_default_phase_classification() {
let planning = Error::Planning {
diagnostics: DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::UnsupportedArrowType,
"unsupported",
)]),
};
let value_conversion = Error::ValueConversion {
diagnostics: DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::NonFiniteFloat,
"not finite",
)]),
};
let direct_encoding = Error::DirectEncoding {
diagnostics: DiagnosticSet::from(vec![Diagnostic::error(
DiagnosticCode::DirectEncodingInvalidPayload,
"invalid",
)]),
};
assert_eq!(planning.write_phase(), Some(WritePhase::SchemaPlanning));
assert_eq!(
value_conversion.write_phase(),
Some(WritePhase::ValueConversion)
);
assert_eq!(
direct_encoding.write_phase(),
Some(WritePhase::DirectEncoding)
);
}
#[test]
fn backend_unavailable_error_display_includes_backend() {
let err = Error::BackendUnavailable {
backend: crate::WriteBackend::DirectRawBulk,
reason: "not implemented".to_owned(),
};
assert_eq!(
err.to_string(),
"write backend DirectRawBulk is unavailable: not implemented"
);
}
#[test]
fn invalid_connection_string_error_display_is_redacted() {
let err = Error::InvalidConnectionString;
assert_eq!(err.to_string(), "SQL Server connection string is invalid");
}
#[test]
fn tiberius_error_display_includes_source_error() {
let err = Error::Tiberius {
source: tiberius::error::Error::Protocol(Cow::Borrowed("invalid token")),
};
assert_eq!(
err.to_string(),
"tiberius operation failed: Protocol error: invalid token"
);
}
#[test]
fn tiberius_error_preserves_source_error() {
let err = Error::Tiberius {
source: tiberius::error::Error::BulkInput(Cow::Borrowed("row payload is malformed")),
};
let source = StdError::source(&err).expect("tiberius source should be preserved");
assert_eq!(
source.to_string(),
"BULK UPLOAD input failure: row payload is malformed"
);
}
}