use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use fathomdb_engine::lifecycle::{
Event, EventCategory, EventSource, Phase, ProfileRecord, ProjectionStatus, SlowStatement,
StressFailureContext, Subscriber,
};
use fathomdb_engine::{CounterSnapshot, Engine, EngineOpenError, PreparedWrite};
use tempfile::TempDir;
#[path = "support/corruption.rs"]
mod corruption;
fn fixture() -> (TempDir, Engine) {
let dir = TempDir::new().unwrap();
let engine = Engine::open(dir.path().join("observability.sqlite")).expect("engine open").engine;
(dir, engine)
}
#[derive(Default)]
struct CapturingSubscriber {
events: Mutex<Vec<Event>>,
profile_records: Mutex<Vec<ProfileRecord>>,
slow_statements: Mutex<Vec<SlowStatement>>,
stress_failures: Mutex<Vec<StressFailureContext>>,
}
impl Subscriber for CapturingSubscriber {
fn on_event(&self, event: &Event) {
self.events.lock().unwrap().push(event.clone());
}
fn on_profile(&self, record: &ProfileRecord) {
self.profile_records.lock().unwrap().push(*record);
}
fn on_slow_statement(&self, signal: &SlowStatement) {
self.slow_statements.lock().unwrap().push(signal.clone());
}
fn on_stress_failure(&self, context: &StressFailureContext) {
self.stress_failures.lock().unwrap().push(context.clone());
}
}
#[test]
fn ac_001_phase_enum_has_five_typed_variants() {
let variants = [Phase::Started, Phase::Slow, Phase::Heartbeat, Phase::Finished, Phase::Failed];
for phase in variants {
match phase {
Phase::Started | Phase::Slow | Phase::Heartbeat | Phase::Finished | Phase::Failed => {}
}
}
assert_ne!(Phase::Started, Phase::Slow);
assert_ne!(Phase::Started, Phase::Heartbeat);
assert_ne!(Phase::Started, Phase::Finished);
assert_ne!(Phase::Started, Phase::Failed);
assert_ne!(Phase::Finished, Phase::Failed);
assert_ne!(Phase::Slow, Phase::Heartbeat);
}
#[test]
fn ac_001_event_struct_carries_typed_source_and_category() {
let event = Event {
phase: Phase::Started,
source: EventSource::Engine,
category: EventCategory::Writer,
code: None,
};
assert_eq!(event.phase, Phase::Started);
assert_eq!(event.source, EventSource::Engine);
assert_eq!(event.category, EventCategory::Writer);
assert_eq!(event.code, None);
}
#[test]
fn ac_002_no_log_files_without_subscriber() {
use std::collections::BTreeSet;
use std::path::PathBuf;
fn walk(root: &std::path::Path, out: &mut BTreeSet<PathBuf>, depth: u32) {
if depth > 6 {
return;
}
let Ok(entries) = std::fs::read_dir(root) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
let Ok(meta) = entry.metadata() else { continue };
out.insert(path.clone());
if meta.is_dir() && !meta.file_type().is_symlink() {
walk(&path, out, depth + 1);
}
}
}
fn snapshot(roots: &[PathBuf]) -> BTreeSet<PathBuf> {
let mut out = BTreeSet::new();
for root in roots {
walk(root, &mut out, 0);
}
out
}
let dir = TempDir::new().unwrap();
let db_path = dir.path().join("nolog.sqlite");
let db_parent = db_path.parent().expect("db parent").to_path_buf();
let mut roots: Vec<PathBuf> = Vec::new();
if let Ok(p) = std::env::current_dir() {
roots.push(p);
}
if let Some(home) = std::env::var_os("HOME") {
roots.push(PathBuf::from(home));
}
for var in ["XDG_CONFIG_HOME", "XDG_DATA_HOME", "XDG_CACHE_HOME", "XDG_STATE_HOME"] {
if let Some(v) = std::env::var_os(var) {
roots.push(PathBuf::from(v));
}
}
roots.push(std::env::temp_dir());
let before = snapshot(&roots);
let opened = Engine::open(&db_path).expect("open");
opened
.engine
.write(&[PreparedWrite::Node {
kind: "doc".to_string(),
body: "hello".to_string(),
source_id: None,
}])
.expect("write");
let _ = opened.engine.search("hello").expect("search");
opened.engine.close().expect("close");
let after = snapshot(&roots);
let new: Vec<&PathBuf> = after.difference(&before).collect();
let allowed_names = [
"nolog.sqlite",
"nolog.sqlite.lock",
"nolog.sqlite-wal",
"nolog.sqlite-shm",
"nolog.sqlite-journal",
];
for path in &new {
if path.starts_with(&db_parent) {
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
if *path == &db_parent {
continue;
}
assert!(
allowed_names.contains(&name),
"engine created unexpected artifact inside DB parent: {}",
path.display(),
);
continue;
}
let lossy = path.to_string_lossy().to_lowercase();
assert!(
!lossy.contains("fathomdb") && !lossy.contains("fathom_"),
"engine created a fathomdb-named artifact outside the DB path: {}",
path.display(),
);
}
}
#[test]
fn ac_003a_writer_events_flow_to_subscriber() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
let _ = engine.write(&[PreparedWrite::Node {
kind: "doc".to_string(),
body: "hello".to_string(),
source_id: None,
}]);
let captured = sink.events.lock().unwrap();
assert!(captured
.iter()
.any(|e| e.source == EventSource::Engine && e.category == EventCategory::Writer));
}
#[test]
fn ac_003b_search_events_flow_to_subscriber() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
let _ = engine.search("hello");
let captured = sink.events.lock().unwrap();
assert!(captured
.iter()
.any(|e| e.source == EventSource::Engine && e.category == EventCategory::Search));
}
#[test]
fn ac_003c_admin_events_flow_to_subscriber() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
let _ = engine.write(&[PreparedWrite::AdminSchema {
name: "things".to_string(),
kind: "latest_state".to_string(),
schema_json: "{}".to_string(),
retention_json: "{}".to_string(),
}]);
let captured = sink.events.lock().unwrap();
assert!(captured
.iter()
.any(|e| e.source == EventSource::Engine && e.category == EventCategory::Admin));
}
#[test]
fn ac_003d_error_events_flow_to_subscriber() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
let _ = engine.write(&[]); let captured = sink.events.lock().unwrap();
assert!(captured
.iter()
.any(|e| e.source == EventSource::Engine && e.category == EventCategory::Error));
}
#[test]
fn ac_004a_counter_snapshot_key_set() {
let (_dir, engine) = fixture();
let snapshot = engine.counters();
assert_eq!(snapshot.queries, 0);
assert_eq!(snapshot.writes, 0);
assert_eq!(snapshot.write_rows, 0);
assert_eq!(snapshot.admin_ops, 0);
assert_eq!(snapshot.cache_hit, 0);
assert_eq!(snapshot.cache_miss, 0);
assert!(snapshot.errors_by_code.is_empty());
let _: BTreeMap<String, u64> = snapshot.errors_by_code.clone();
}
#[test]
fn ac_004b_counter_delta_exact_over_mixed_ops() {
let (_dir, engine) = fixture();
let s0 = engine.counters();
for _ in 0..400 {
engine
.write(&[PreparedWrite::Node {
kind: "doc".to_string(),
body: "hello".to_string(),
source_id: None,
}])
.expect("write");
}
for _ in 0..400 {
let _ = engine.search("hello").expect("search");
}
for i in 0..200 {
engine
.write(&[PreparedWrite::AdminSchema {
name: format!("things_{}", i % 4),
kind: "latest_state".to_string(),
schema_json: "{}".to_string(),
retention_json: "{}".to_string(),
}])
.expect("admin");
}
let s1 = engine.counters();
assert_eq!(s1.writes - s0.writes, 400, "writes");
assert_eq!(s1.write_rows - s0.write_rows, 400, "write_rows");
assert_eq!(s1.queries - s0.queries, 400, "queries");
assert_eq!(s1.admin_ops - s0.admin_ops, 200, "admin_ops");
assert!(s1.cache_hit >= s0.cache_hit, "cache_hit monotonic non-decreasing");
assert!(s1.cache_miss >= s0.cache_miss, "cache_miss monotonic non-decreasing");
}
#[test]
fn ac_004c_counter_snapshot_does_not_perturb() {
let (_dir, engine) = fixture();
let s0 = engine.counters();
let s1 = engine.counters();
assert_eq!(s0, s1);
}
#[test]
fn ac_005a_profiling_toggleable_at_runtime() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.set_profiling(false).expect("disable profiling");
let _ = engine.search("hello").expect("search");
assert_eq!(
sink.profile_records.lock().unwrap().len(),
0,
"no profile records expected while profiling disabled"
);
engine.set_profiling(true).expect("enable profiling");
let _ = engine.search("hello").expect("search");
let after = sink.profile_records.lock().unwrap().len();
assert!(after >= 1, "expected ≥ 1 profile record after enabling profiling, saw {after}");
engine.set_profiling(false).expect("disable profiling again");
let frozen = sink.profile_records.lock().unwrap().len();
let _ = engine.search("hello").expect("search");
assert_eq!(sink.profile_records.lock().unwrap().len(), frozen);
}
#[test]
fn ac_005b_profile_record_typed_numeric_fields() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.set_profiling(true).expect("enable profiling");
let _ = engine.search("hello").expect("search");
let records = sink.profile_records.lock().unwrap();
let record = records.first().expect("at least one profile record");
let _: u64 = record.wall_clock_ms;
let _: u64 = record.step_count;
let _: i64 = record.cache_delta;
}
#[test]
fn ac_006_sqlite_internal_events_typed_source() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("corrupt-open.sqlite");
let opened = Engine::open(&path).expect("seed database");
opened.engine.close().expect("close before corruption");
corruption::corrupt_database_header(&path);
let sink = Arc::new(CapturingSubscriber::default());
let err = Engine::open_with_subscriber_for_test(&path, sink.clone())
.expect_err("corrupted database must fail open");
assert!(matches!(err, EngineOpenError::Corruption(_)));
let captured = sink.events.lock().unwrap();
assert!(captured.iter().any(|e| {
e.source == EventSource::SqliteInternal
&& e.category == EventCategory::Corruption
&& e.code == Some("SQLITE_NOTADB")
}));
}
#[test]
fn ac_006_interior_page_corruption_emits_sqlite_corrupt() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("corrupt-interior.sqlite");
let opened = Engine::open(&path).expect("seed database");
opened
.engine
.write(&[PreparedWrite::Node {
kind: "doc".to_string(),
body: "interior".to_string(),
source_id: None,
}])
.expect("seed write");
opened.engine.close().expect("close before corruption");
corruption::corrupt_interior_page_byte(&path, 0, 100, 0xFF);
let sink = Arc::new(CapturingSubscriber::default());
let err = Engine::open_with_subscriber_for_test(&path, sink.clone())
.expect_err("corrupted interior page must fail open");
assert!(matches!(err, EngineOpenError::Corruption(_)));
let captured = sink.events.lock().unwrap();
assert!(
captured.iter().any(|e| {
e.source == EventSource::SqliteInternal
&& e.category == EventCategory::Corruption
&& e.code == Some("SQLITE_CORRUPT")
}),
"expected (SqliteInternal, Corruption, SQLITE_CORRUPT) event, saw: {:?}",
captured.iter().map(|e| (e.source, e.category, e.code)).collect::<Vec<_>>(),
);
}
#[test]
fn ac_007a_slow_statement_event_at_default_threshold() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.execute_for_test(SLOW_CTE).expect("slow cte");
let signals = sink.slow_statements.lock().unwrap();
assert_eq!(
signals.len(),
1,
"expected exactly one slow-statement signal at default threshold, saw {}",
signals.len(),
);
assert!(
signals[0].statement.contains("RECURSIVE"),
"slow signal must identify the statement; got: {:?}",
signals[0].statement,
);
assert!(
signals[0].wall_clock_ms >= 100,
"slow signal wall_clock_ms must be ≥ 100 ms (default threshold); got {} ms",
signals[0].wall_clock_ms,
);
}
#[test]
fn ac_007b_slow_threshold_reconfigurable() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.set_slow_threshold_ms(500).expect("set threshold");
engine.execute_for_test(FAST_CTE).expect("fast cte");
assert_eq!(
sink.slow_statements.lock().unwrap().len(),
0,
"fast fixture must not emit a slow-statement signal at threshold=500 ms"
);
engine.execute_for_test(SLOW_CTE).expect("slow cte");
let signals = sink.slow_statements.lock().unwrap();
assert_eq!(
signals.len(),
1,
"slow fixture must emit exactly one slow-statement signal at threshold=500 ms"
);
assert!(signals[0].wall_clock_ms >= 500);
}
const FAST_CTE: &str = "WITH RECURSIVE c(x) AS (VALUES(1) UNION ALL \
SELECT x + 1 FROM c WHERE x < 100000) \
SELECT count(*) FROM c";
const SLOW_CTE: &str = "WITH RECURSIVE c(x) AS (VALUES(1) UNION ALL \
SELECT x + 1 FROM c WHERE x < 1000000) \
SELECT count(*) FROM c";
#[test]
fn ac_008_slow_signal_feeds_lifecycle() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.execute_for_test(SLOW_CTE).expect("slow cte");
let signals = sink.slow_statements.lock().unwrap();
assert!(!signals.is_empty(), "expected at least one slow-statement signal");
let events = sink.events.lock().unwrap();
assert!(
events.iter().any(|e| e.phase == Phase::Slow),
"expected at least one lifecycle event with phase == Slow"
);
}
#[test]
fn ac_009_poison_thread_emits_full_stress_failure_context() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.run_one_thread_poison_for_test().expect("poison fixture should run");
let captured = sink.stress_failures.lock().unwrap();
let ctx = captured.first().expect("one-thread poison must emit a stress failure context");
assert_ne!(ctx.thread_group_id, 0, "thread_group_id must be non-zero");
assert_eq!(ctx.op_kind, "write", "op_kind must be the documented op label");
assert!(
ctx.last_error_chain.len() >= 2,
"last_error_chain must include stable_code + ≥ 1 causal-chain segment, got {:?}",
ctx.last_error_chain,
);
assert_eq!(
ctx.last_error_chain[0], "WriteValidationError",
"first chain entry must equal EngineError::stable_code"
);
assert!(
ctx.last_error_chain.iter().skip(1).any(|s| !s.is_empty()),
"at least one causal-chain segment after stable_code must be non-empty"
);
assert!(
matches!(ctx.projection_state.as_str(), "Pending" | "Failed" | "UpToDate"),
"projection_state must be a stringified ProjectionStatus variant, got {:?}",
ctx.projection_state,
);
}
#[test]
fn ac_009_stress_failure_context_constructs() {
let (_dir, engine) = fixture();
let sink = Arc::new(CapturingSubscriber::default());
let _sub = engine.subscribe(sink.clone());
engine.run_one_thread_poison_for_test().expect("poison fixture should emit");
let captured = sink.stress_failures.lock().unwrap();
let ctx = captured.first().expect("one-thread poison must emit a stress failure context");
let _: u64 = ctx.thread_group_id;
let _: String = ctx.op_kind.clone();
let _: Vec<String> = ctx.last_error_chain.clone();
let _: String = ctx.projection_state.clone();
assert!(!ctx.op_kind.is_empty());
assert!(!ctx.last_error_chain.is_empty());
assert!(!ctx.projection_state.is_empty());
}
#[test]
fn ac_010_projection_status_enum_three_values() {
let variants =
[ProjectionStatus::Pending, ProjectionStatus::Failed, ProjectionStatus::UpToDate];
for status in variants {
match status {
ProjectionStatus::Pending | ProjectionStatus::Failed | ProjectionStatus::UpToDate => {}
}
}
assert_ne!(ProjectionStatus::Pending, ProjectionStatus::Failed);
assert_ne!(ProjectionStatus::Pending, ProjectionStatus::UpToDate);
assert_ne!(ProjectionStatus::Failed, ProjectionStatus::UpToDate);
}
#[test]
fn counter_snapshot_default_is_zero() {
let s = CounterSnapshot::default();
assert_eq!(s.queries, 0);
assert_eq!(s.writes, 0);
assert_eq!(s.write_rows, 0);
assert_eq!(s.admin_ops, 0);
assert_eq!(s.cache_hit, 0);
assert_eq!(s.cache_miss, 0);
assert!(s.errors_by_code.is_empty());
}