1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
//! Concurrent reader/writer test that pins the read-after-write cursor
//! invariant from `dev/design/engine.md` § Cursor contract and AC-059b /
//! REQ-013 / REQ-055.
//!
//! Invariant under test: the number of canonical rows visible to a
//! `search` MUST be ≤ the search's reported `projection_cursor`. Because
//! every committed write advances the engine cursor by exactly one and
//! inserts exactly one canonical row in this fixture, the row count the
//! search returned is a lower bound on "how many writes the reader's
//! snapshot saw"; that value can never exceed the cursor the reader
//! also returned without violating the cursor contract.
//!
//! Failure mode under test: the engine loads `next_cursor` from the
//! writer-side atomic before the reader connection runs its query; a
//! writer commit between those two events appears in the reader's WAL
//! snapshot but not in the reported cursor.
//!
//! Bounded operationally: 1,000 search calls, hard wall-clock cap of
//! 30 s, writer throttled.
//!
//! Runtime-budget category: long-run only (~1000-iteration race fixture).
//! `agent-verify.sh` skips this test entirely for runtime budget; the
//! ~1000-iteration race fixture is exercised only by `scripts/check.sh`
//! with `AGENT_LONG=1`. There is no smoke variant — the AC-059b evidence
//! comes exclusively from the long-run gate.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};
use fathomdb_engine::{Engine, PreparedWrite};
use tempfile::TempDir;
fn long_run_enabled() -> bool {
std::env::var_os("AGENT_LONG").is_some()
}
#[test]
fn projection_cursor_bounds_observed_row_count() {
if !long_run_enabled() {
return;
}
let dir = TempDir::new().unwrap();
let path = dir.path().join("cursor_race.sqlite");
let opened = Engine::open(&path).expect("open");
let engine = Arc::new(opened.engine);
let stop = Arc::new(AtomicBool::new(false));
let writer = {
let engine = Arc::clone(&engine);
let stop = Arc::clone(&stop);
thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
let _ = engine.write(&[PreparedWrite::Node {
kind: "doc".to_string(),
body: "needle".to_string(),
source_id: fathomdb_engine::SourceId::new("test:fixture")
.expect("test source id"),
logical_id: None,
state: fathomdb_engine::InitialState::Active,
reason: None,
valid_from: None,
valid_until: None,
}]);
thread::sleep(Duration::from_micros(50));
}
})
};
let iterations = 1000usize;
let mut violations = 0usize;
let started = Instant::now();
for i in 0..iterations {
if started.elapsed() > Duration::from_secs(30) {
stop.store(true, Ordering::Relaxed);
writer.join().expect("writer thread");
panic!("cursor invariant test exceeded 30 s wall clock at iteration {i}");
}
let result = engine.search("needle").expect("search");
// results[0] is the compiled SQL string from `compile_text_query`;
// every entry after that is a body row from canonical_nodes.
let row_count =
result.results.iter().filter(|h| h.body.as_str() == "needle").count() as u64;
if row_count > result.projection_cursor {
violations += 1;
if violations <= 5 {
eprintln!(
"iter {i}: rows_returned={row_count} > projection_cursor={}",
result.projection_cursor,
);
}
}
}
stop.store(true, Ordering::Relaxed);
writer.join().expect("writer thread");
assert_eq!(
violations, 0,
"cursor invariant violated {violations}/{iterations} times: search returned more \
matching rows than the reported projection_cursor permits — projection_cursor was \
derived from a writer-side atomic that races ahead of the reader's WAL snapshot",
);
}