use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use crate::pool::{Connection, QueryRows};
use crate::value::Value;
#[derive(Debug, Clone)]
pub struct ShadowComparison {
pub sql: String,
pub orm_duration: Duration,
pub raw_duration: Duration,
pub orm_rows: usize,
pub raw_rows: usize,
pub consistent: bool,
pub mismatch: Option<String>,
}
impl ShadowComparison {
pub fn latency_ratio(&self) -> f64 {
if self.raw_duration.is_zero() {
1.0
} else {
self.orm_duration.as_secs_f64() / self.raw_duration.as_secs_f64()
}
}
}
#[derive(Debug, Default)]
pub struct ShadowStats {
pub comparisons: AtomicU64,
pub mismatches: AtomicU64,
pub orm_total_us: AtomicU64,
pub raw_total_us: AtomicU64,
}
impl ShadowStats {
pub fn avg_orm_us(&self) -> u64 {
let n = self.comparisons.load(Ordering::Relaxed);
self.orm_total_us
.load(Ordering::Relaxed)
.checked_div(n)
.unwrap_or(0)
}
pub fn avg_raw_us(&self) -> u64 {
let n = self.comparisons.load(Ordering::Relaxed);
self.raw_total_us
.load(Ordering::Relaxed)
.checked_div(n)
.unwrap_or(0)
}
pub fn mismatch_rate(&self) -> f64 {
let n = self.comparisons.load(Ordering::Relaxed);
if n == 0 {
0.0
} else {
self.mismatches.load(Ordering::Relaxed) as f64 / n as f64
}
}
}
#[derive(Debug, Clone)]
pub struct ShadowConfig {
pub timeout: Duration,
pub row_count_only: bool,
pub max_compare_rows: usize,
}
impl Default for ShadowConfig {
fn default() -> Self {
Self {
timeout: Duration::from_secs(3),
row_count_only: false,
max_compare_rows: 10_000,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MismatchAction {
Record,
Error,
Panic,
}
pub struct ShadowConnection<C: Connection> {
orm_conn: C,
raw_conn: C,
config: ShadowConfig,
on_mismatch: MismatchAction,
stats: ShadowStats,
}
impl<C: Connection> ShadowConnection<C> {
pub fn new(orm_conn: C, raw_conn: C, config: ShadowConfig) -> Self {
Self {
orm_conn,
raw_conn,
config,
on_mismatch: MismatchAction::Record,
stats: ShadowStats::default(),
}
}
pub fn with_mismatch_action(mut self, action: MismatchAction) -> Self {
self.on_mismatch = action;
self
}
pub fn stats(&self) -> &ShadowStats {
&self.stats
}
pub async fn query_shadow(&mut self, sql: &str) -> Result<QueryRows, crate::DbError> {
let orm_fut = self.orm_conn.query(sql);
let raw_fut = self.raw_conn.query(sql);
let orm_start = Instant::now();
let orm_result = tokio::time::timeout(self.config.timeout, orm_fut).await;
let orm_duration = orm_start.elapsed();
let raw_start = Instant::now();
let raw_result = tokio::time::timeout(self.config.timeout, raw_fut).await;
let raw_duration = raw_start.elapsed();
self.stats
.orm_total_us
.fetch_add(orm_duration.as_micros() as u64, Ordering::Relaxed);
self.stats
.raw_total_us
.fetch_add(raw_duration.as_micros() as u64, Ordering::Relaxed);
self.stats.comparisons.fetch_add(1, Ordering::Relaxed);
let orm_rows = match orm_result {
Ok(Ok(rows)) => rows,
Ok(Err(e)) => {
self.stats.mismatches.fetch_add(1, Ordering::Relaxed);
self.handle_mismatch(ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: 0,
raw_rows: 0,
consistent: false,
mismatch: Some(format!("ORM error: {}", e)),
});
return Err(e);
}
Err(_) => {
self.stats.mismatches.fetch_add(1, Ordering::Relaxed);
return Err(crate::DbError::ConnectionTimeout(format!(
"ORM shadow path timeout after {:?}",
self.config.timeout
)));
}
};
let raw_rows = match raw_result {
Ok(Ok(rows)) => rows,
Ok(Err(e)) => {
self.stats.mismatches.fetch_add(1, Ordering::Relaxed);
self.handle_mismatch(ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: 0,
consistent: false,
mismatch: Some(format!("Raw path error: {}", e)),
});
return Ok(orm_rows);
}
Err(_) => {
self.stats.mismatches.fetch_add(1, Ordering::Relaxed);
self.handle_mismatch(ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: 0,
consistent: false,
mismatch: Some(format!("Raw path timeout after {:?}", self.config.timeout)),
});
return Ok(orm_rows);
}
};
let comparison = self.compare_rows(sql, orm_duration, raw_duration, &orm_rows, &raw_rows);
if !comparison.consistent {
self.stats.mismatches.fetch_add(1, Ordering::Relaxed);
self.handle_mismatch(comparison);
}
Ok(orm_rows)
}
fn compare_rows(
&self,
sql: &str,
orm_duration: Duration,
raw_duration: Duration,
orm_rows: &[HashMap<String, Value>],
raw_rows: &[HashMap<String, Value>],
) -> ShadowComparison {
if orm_rows.len() != raw_rows.len() {
return ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: false,
mismatch: Some(format!(
"Row count mismatch: ORM={} vs Raw={}",
orm_rows.len(),
raw_rows.len()
)),
};
}
if self.config.row_count_only {
return ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: true,
mismatch: None,
};
}
let limit = self.config.max_compare_rows.min(orm_rows.len());
for (i, (orm_row, raw_row)) in orm_rows[..limit]
.iter()
.zip(raw_rows[..limit].iter())
.enumerate()
{
if orm_row.len() != raw_row.len() {
return ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: false,
mismatch: Some(format!(
"Row {} column count mismatch: ORM={} vs Raw={}",
i,
orm_row.len(),
raw_row.len()
)),
};
}
for (key, orm_val) in orm_row {
match raw_row.get(key) {
Some(raw_val) if !values_equal(orm_val, raw_val) => {
return ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: false,
mismatch: Some(format!(
"Row {} column '{}' value mismatch: ORM={:?} vs Raw={:?}",
i, key, orm_val, raw_val
)),
};
}
None => {
return ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: false,
mismatch: Some(format!(
"Row {} column '{}' missing in raw result",
i, key
)),
};
}
_ => {}
}
}
}
ShadowComparison {
sql: sql.to_string(),
orm_duration,
raw_duration,
orm_rows: orm_rows.len(),
raw_rows: raw_rows.len(),
consistent: true,
mismatch: None,
}
}
fn handle_mismatch(&self, comparison: ShadowComparison) {
tracing::warn!(
target: "sz_orm::shadow",
sql = %comparison.sql,
orm_rows = comparison.orm_rows,
raw_rows = comparison.raw_rows,
orm_us = ?comparison.orm_duration,
raw_us = ?comparison.raw_duration,
mismatch = ?comparison.mismatch,
"Shadow traffic mismatch detected"
);
match self.on_mismatch {
MismatchAction::Record => { }
MismatchAction::Error => { }
MismatchAction::Panic => {
panic!(
"Shadow traffic mismatch: SQL={:?} mismatch={:?}",
comparison.sql, comparison.mismatch
);
}
}
}
}
fn values_equal(a: &Value, b: &Value) -> bool {
use Value::*;
match (a, b) {
(Null, Null) => true,
(Bool(x), Bool(y)) => x == y,
(I8(x), I8(y)) => x == y,
(I8(x), I16(y)) => (*x as i16) == *y,
(I16(x), I8(y)) => *x == (*y as i16),
(I16(x), I16(y)) => x == y,
(I16(x), I32(y)) => (*x as i32) == *y,
(I32(x), I16(y)) => *x == (*y as i32),
(I32(x), I32(y)) => x == y,
(I32(x), I64(y)) => (*x as i64) == *y,
(I64(x), I32(y)) => *x == (*y as i64),
(I64(x), I64(y)) => x == y,
(U8(x), U8(y)) => x == y,
(U16(x), U16(y)) => x == y,
(U32(x), U32(y)) => x == y,
(U64(x), U64(y)) => x == y,
(F32(x), F32(y)) => (x - y).abs() < 1e-6,
(F32(x), F64(y)) => ((*x as f64) - y).abs() < 1e-6,
(F64(x), F32(y)) => (x - (*y as f64)).abs() < 1e-6,
(F64(x), F64(y)) => (x - y).abs() < 1e-9,
(String(x), String(y)) => x == y,
(Bytes(x), Bytes(y)) => x == y,
_ => a == b,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_values_equal_numeric_cross_type() {
assert!(values_equal(&Value::I32(42), &Value::I64(42)));
assert!(values_equal(&Value::I8(1), &Value::I16(1)));
assert!(!values_equal(&Value::I32(42), &Value::I64(43)));
}
#[test]
fn test_values_equal_float_tolerance() {
assert!(values_equal(&Value::F32(1.0), &Value::F64(1.0)));
assert!(!values_equal(&Value::F32(1.0), &Value::F64(2.0)));
}
#[test]
fn test_shadow_stats_mismatch_rate() {
let stats = ShadowStats::default();
assert_eq!(stats.mismatch_rate(), 0.0);
stats.comparisons.store(10, Ordering::Relaxed);
stats.mismatches.store(1, Ordering::Relaxed);
assert!((stats.mismatch_rate() - 0.1).abs() < 1e-9);
}
}