tawasuyu/brahman

1//! End-to-end drift detector: spin up `run_server` against log A, run
2//! `check_against_socket` first against the same log (in-sync) and then
3//! against a divergent log B (drift detected, with the expected diff
4//! list).
5//!
6//! Same threading inversion as `tests/run.rs`: server on main thread
7//! (Executor is `!Send`), client on a worker thread.
8
9use std::io::{BufRead, BufReader, Write};
10use std::os::unix::net::UnixStream;
11use std::path::{Path, PathBuf};
12use std::thread;
13
14use nakui_core::drift::{DriftDiff, check_against_socket};
15use nakui_core::event_log::{EventLog, execute_and_log, seed_and_log};
16use nakui_core::executor::Executor;
17use nakui_core::run::run_server;
18use nakui_core::store::MemoryStore;
19use serde_json::json;
20use uuid::Uuid;
21
22fn workspace_root() -> PathBuf {
23    Path::new(env!("CARGO_MANIFEST_DIR"))
24        .parent()
25        .expect("workspace root above core/")
26        .to_path_buf()
27}
28
29fn treasury_module() -> PathBuf {
30    workspace_root().join("modules/treasury")
31}
32
33fn fresh_log_path() -> PathBuf {
34    std::env::temp_dir().join(format!("nakui_drift_log_{}.jsonl", Uuid::new_v4()))
35}
36
37fn fresh_socket_path() -> PathBuf {
38    std::env::temp_dir().join(format!("nakui_drift_{}.sock", Uuid::new_v4()))
39}
40
41/// Build a real WAL-formed log: two cajas seeded + one deposit.
42fn build_log_a(path: &Path, caja_a: Uuid, caja_b: Uuid) {
43    let executor = Executor::load_module(treasury_module()).expect("load");
44    let mut log = EventLog::open(path).expect("open log");
45    let mut store = MemoryStore::new();
46    seed_and_log(
47        &executor,
48        &mut store,
49        &mut log,
50        "Caja",
51        caja_a,
52        json!({"id": caja_a.to_string(), "name": "A", "saldo": 100_000_i64, "currency": "USD"}),
53    )
54    .unwrap();
55    seed_and_log(
56        &executor,
57        &mut store,
58        &mut log,
59        "Caja",
60        caja_b,
61        json!({"id": caja_b.to_string(), "name": "B", "saldo": 50_000_i64, "currency": "USD"}),
62    )
63    .unwrap();
64    execute_and_log(
65        &executor,
66        &mut store,
67        &mut log,
68        "register_cash_move",
69        &[("caja", caja_a)],
70        json!({
71            "monto": 5_000_i64,
72            "tipo": "in",
73            "timestamp": "2026-05-04T10:00:00Z",
74            "memo": "x",
75            "movimiento_id": Uuid::new_v4().to_string(),
76        }),
77    )
78    .unwrap();
79}
80
81/// Build a divergent log: only caja_a seeded, no deposit, no caja_b.
82/// Replaying B produces a different state than the server (which used A).
83fn build_log_b(path: &Path, caja_a: Uuid) {
84    let executor = Executor::load_module(treasury_module()).expect("load");
85    let mut log = EventLog::open(path).expect("open log b");
86    let mut store = MemoryStore::new();
87    seed_and_log(
88        &executor,
89        &mut store,
90        &mut log,
91        "Caja",
92        caja_a,
93        json!({"id": caja_a.to_string(), "name": "A", "saldo": 100_000_i64, "currency": "USD"}),
94    )
95    .unwrap();
96}
97
98/// Wait for the socket to exist and be connectable, then return a
99/// connected stream. Used by helpers that send raw requests bypassing
100/// `check_against_socket` (e.g. shutdown).
101fn connect_with_retry(path: &Path) -> UnixStream {
102    for _ in 0..100 {
103        if let Ok(s) = UnixStream::connect(path) {
104            return s;
105        }
106        thread::sleep(std::time::Duration::from_millis(20));
107    }
108    panic!("server never started accepting on {}", path.display());
109}
110
111fn send_shutdown(socket_path: &Path) {
112    let mut stream = connect_with_retry(socket_path);
113    stream.write_all(b"{\"op\":\"shutdown\"}\n").unwrap();
114    let mut reader = BufReader::new(stream.try_clone().unwrap());
115    let mut line = String::new();
116    reader.read_line(&mut line).unwrap();
117}
118
119#[test]
120fn drift_check_reports_in_sync_when_log_matches_server() {
121    let log_path = fresh_log_path();
122    let socket_path = fresh_socket_path();
123
124    let caja_a = Uuid::new_v4();
125    let caja_b = Uuid::new_v4();
126    build_log_a(&log_path, caja_a, caja_b);
127
128    let executor = Executor::load_module(treasury_module()).expect("load");
129    let log = EventLog::open(&log_path).expect("reopen");
130    let store = MemoryStore::new();
131
132    let socket_for_client = socket_path.clone();
133    let log_for_client = log_path.clone();
134    let client = thread::spawn(move || -> Result<(), String> {
135        let report = check_against_socket(&log_for_client, &socket_for_client)
136            .map_err(|e| format!("check failed: {}", e))?;
137        if !report.in_sync() {
138            return Err(format!(
139                "expected in_sync, got {} diffs: {:?}",
140                report.diffs.len(),
141                report.diffs
142            ));
143        }
144        if report.log_hash != report.server_hash {
145            return Err("hashes diverged with empty diff — invariant broken".into());
146        }
147        if report.log_records != report.server_records {
148            return Err(format!(
149                "record count diverged: log={} server={}",
150                report.log_records, report.server_records
151            ));
152        }
153        send_shutdown(&socket_for_client);
154        Ok(())
155    });
156
157    run_server(executor, log, store, None, &socket_path).expect("server clean exit");
158    client.join().unwrap().expect("client assertions");
159
160    let _ = std::fs::remove_file(&log_path);
161}
162
163#[test]
164fn drift_check_surfaces_expected_per_record_diffs() {
165    let log_a_path = fresh_log_path();
166    let log_b_path = fresh_log_path();
167    let socket_path = fresh_socket_path();
168
169    let caja_a = Uuid::new_v4();
170    let caja_b = Uuid::new_v4();
171    build_log_a(&log_a_path, caja_a, caja_b);
172    build_log_b(&log_b_path, caja_a);
173
174    let executor = Executor::load_module(treasury_module()).expect("load");
175    let log = EventLog::open(&log_a_path).expect("reopen");
176    let store = MemoryStore::new();
177
178    let socket_for_client = socket_path.clone();
179    let log_b_for_client = log_b_path.clone();
180    let client = thread::spawn(move || -> Result<(), String> {
181        // Server is running log A's state; we audit using log B's
182        // canonical view. Expected diffs:
183        //   - Caja caja_a: tampered (B says saldo=100_000, server has 105_000 from deposit)
184        //   - Caja caja_b: only_on_server (B never seeded it)
185        //   - Movimiento <some uuid>: only_on_server (B never executed the deposit)
186        let report = check_against_socket(&log_b_for_client, &socket_for_client)
187            .map_err(|e| format!("check failed: {}", e))?;
188        if report.in_sync() {
189            return Err("expected drift, got in_sync".into());
190        }
191
192        let mut tampered = 0;
193        let mut only_on_server = 0;
194        let mut only_in_log = 0;
195        let mut tampered_caja_a = false;
196        let mut server_extra_caja_b = false;
197        let mut server_extra_movimiento = false;
198
199        for d in &report.diffs {
200            match d {
201                DriftDiff::Tampered {
202                    entity,
203                    id,
204                    log_value,
205                    server_value,
206                } => {
207                    tampered += 1;
208                    if entity == "Caja" && *id == caja_a {
209                        tampered_caja_a = true;
210                        if log_value["saldo"] != json!(100_000_i64) {
211                            return Err(format!("log saldo wrong: {}", log_value));
212                        }
213                        if server_value["saldo"] != json!(105_000_i64) {
214                            return Err(format!("server saldo wrong: {}", server_value));
215                        }
216                    }
217                }
218                DriftDiff::OnlyOnServer { entity, id, .. } => {
219                    only_on_server += 1;
220                    if entity == "Caja" && *id == caja_b {
221                        server_extra_caja_b = true;
222                    }
223                    if entity == "Movimiento" {
224                        server_extra_movimiento = true;
225                    }
226                }
227                DriftDiff::OnlyInLog { .. } => only_in_log += 1,
228            }
229        }
230        if tampered != 1 {
231            return Err(format!("expected 1 tampered, got {}", tampered));
232        }
233        if only_on_server != 2 {
234            return Err(format!("expected 2 only_on_server, got {}", only_on_server));
235        }
236        if only_in_log != 0 {
237            return Err(format!("expected 0 only_in_log, got {}", only_in_log));
238        }
239        if !tampered_caja_a {
240            return Err("expected tampered diff for caja_a".into());
241        }
242        if !server_extra_caja_b {
243            return Err("expected only_on_server diff for caja_b".into());
244        }
245        if !server_extra_movimiento {
246            return Err("expected only_on_server diff for some Movimiento".into());
247        }
248
249        send_shutdown(&socket_for_client);
250        Ok(())
251    });
252
253    run_server(executor, log, store, None, &socket_path).expect("server clean exit");
254    client.join().unwrap().expect("client assertions");
255
256    let _ = std::fs::remove_file(&log_a_path);
257    let _ = std::fs::remove_file(&log_b_path);
258}