tawasuyu/brahman

1//! Smoke test for the persistent backend wired into `nakui run`.
2//!
3//! Gated behind `--features persistent` because SurrealKV pulls in a
4//! ~5 min cold native build. Run with:
5//!     cargo test --features persistent --test run_persistent
6//!
7//! What this proves:
8//!   1. `run_server` accepts a `SurrealStore` and serves the standard
9//!      protocol (execute/load/shutdown round-trip).
10//!   2. After shutdown, reopening the same backing store path reveals
11//!      the records were actually written through to disk — i.e., the
12//!      runtime wasn't just hitting an in-memory façade.
13//!
14//! What this does NOT prove (covered elsewhere or deferred):
15//!   - That startup skips replay when the persistent state is current.
16//!     V1 always replays from log, even with a persistent store; the
17//!     persistent layer is durability for the runtime cache, not a
18//!     replay shortcut. A future `last_applied_seq` tracker would
19//!     change that.
20//!   - Cross-backend hash equality (Memory vs Surreal). Different
21//!     concern — round-trip parity of serde_json::Value through the
22//!     SurrealDB driver.
23
24#![cfg(feature = "persistent")]
25
26use std::io::{BufRead, BufReader, Write};
27use std::os::unix::net::UnixStream;
28use std::path::{Path, PathBuf};
29use std::thread;
30use std::time::Duration;
31
32use nakui_core::event_log::{seed_and_log, EventLog};
33use nakui_core::executor::Executor;
34use nakui_core::run::run_server;
35use nakui_core::store::Store;
36use nakui_core::surreal_store::SurrealStore;
37use serde_json::{json, Value};
38use uuid::Uuid;
39
40fn workspace_root() -> PathBuf {
41    Path::new(env!("CARGO_MANIFEST_DIR"))
42        .parent()
43        .expect("workspace root above core/")
44        .to_path_buf()
45}
46
47fn treasury_module() -> PathBuf {
48    workspace_root().join("modules/treasury")
49}
50
51fn fresh_log_path() -> PathBuf {
52    std::env::temp_dir().join(format!("nakui_runp_log_{}.jsonl", Uuid::new_v4()))
53}
54
55fn fresh_store_path() -> PathBuf {
56    std::env::temp_dir().join(format!("nakui_runp_store_{}", Uuid::new_v4()))
57}
58
59fn fresh_socket_path() -> PathBuf {
60    std::env::temp_dir().join(format!("nakui_runp_{}.sock", Uuid::new_v4()))
61}
62
63struct Conn {
64    writer: UnixStream,
65    reader: BufReader<UnixStream>,
66}
67
68fn connect_with_retry(path: &Path) -> Conn {
69    for _ in 0..200 {
70        if let Ok(stream) = UnixStream::connect(path) {
71            let reader_stream = stream.try_clone().expect("clone");
72            return Conn {
73                writer: stream,
74                reader: BufReader::new(reader_stream),
75            };
76        }
77        thread::sleep(Duration::from_millis(20));
78    }
79    panic!("server never started accepting on {}", path.display());
80}
81
82fn exchange(conn: &mut Conn, req: Value) -> Value {
83    let mut bytes = serde_json::to_vec(&req).unwrap();
84    bytes.push(b'\n');
85    conn.writer.write_all(&bytes).unwrap();
86    let mut line = String::new();
87    conn.reader.read_line(&mut line).unwrap();
88    serde_json::from_str(line.trim()).unwrap()
89}
90
91#[test]
92fn run_server_with_persistent_surreal_serves_protocol_and_writes_to_disk() {
93    let log_path = fresh_log_path();
94    let store_path = fresh_store_path();
95    let socket_path = fresh_socket_path();
96
97    // Pre-seed via the WAL so the log has a record the server can
98    // replay into the persistent store on startup.
99    let caja = Uuid::new_v4();
100    {
101        let mut store = SurrealStore::new_persistent(&store_path).expect("open persistent");
102        let mut log = EventLog::open(&log_path).expect("open log");
103        seed_and_log(
104            &executor,
105            &mut store,
106            &mut log,
107            "Caja",
108            caja,
109            json!({
110                "id": caja.to_string(),
111                "name": "A",
112                "saldo": 100_000_i64,
113                "currency": "USD",
114            }),
115        )
116        .expect("seed");
117    }
118
119    // Start the server with the same persistent store path.
120    let executor = Executor::load_module(treasury_module()).expect("load module");
121    let log = EventLog::open(&log_path).expect("reopen log");
122    let store = SurrealStore::new_persistent(&store_path).expect("reopen persistent");
123
124    let socket_for_client = socket_path.clone();
125    let client = thread::spawn(move || -> Result<(), String> {
126        let mut conn = connect_with_retry(&socket_for_client);
127
128        // Initial load picks up the seed (replayed at startup into the
129        // persistent store).
130        let resp = exchange(
131            &mut conn,
132            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
133        );
134        if resp["value"]["saldo"].as_i64() != Some(100_000) {
135            return Err(format!("startup replay didn't land seed: {}", resp));
136        }
137
138        // Drive a deposit through the server — this writes through the
139        // log AND the persistent store.
140        let resp = exchange(
141            &mut conn,
142            json!({
143                "op": "execute",
144                "morphism": "register_cash_move",
145                "inputs": {"caja": caja.to_string()},
146                "params": {
147                    "monto": 7_500_i64,
148                    "tipo": "in",
149                    "timestamp": "2026-05-04T10:00:00Z",
150                    "memo": "persisted",
151                    "movimiento_id": Uuid::new_v4().to_string(),
152                }
153            }),
154        );
155        if resp["ok"] != json!(true) {
156            return Err(format!("execute failed: {}", resp));
157        }
158
159        let resp = exchange(
160            &mut conn,
161            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
162        );
163        if resp["value"]["saldo"].as_i64() != Some(107_500) {
164            return Err(format!("post-execute saldo wrong: {}", resp));
165        }
166
167        let _ = exchange(&mut conn, json!({"op": "shutdown"}));
168        Ok(())
169    });
170
171    run_server(executor, log, store, None, &socket_path).expect("server clean exit");
172    client.join().unwrap().expect("client assertions");
173
174    // Now the server is gone. Open a fresh handle to the SAME persistent
175    // store path — the records must be there without any replay. This
176    // is what proves "persistent backend" beyond the unit-level tests
177    // in surreal_persist.rs: the runtime actually wrote through.
178    let store_again = SurrealStore::new_persistent(&store_path).expect("reopen final");
179    let v = store_again
180        .load("Caja", caja)
181        .expect("Caja persisted across server shutdown");
182    assert_eq!(
183        v.get("saldo").and_then(Value::as_i64),
184        Some(107_500),
185        "deposit landed in persistent store"
186    );
187
188    let _ = std::fs::remove_file(&log_path);
189    let _ = std::fs::remove_dir_all(&store_path);
190}
191
192#[test]
193fn run_server_skips_replay_when_persistent_store_is_in_sync() {
194    // The optimization: when the persistent store's `last_applied_seq`
195    // matches the log's last seq, startup_replay must skip the
196    // clear+replay entirely. We prove that by mutating the store
197    // out-of-band between cycles — if skip happens, the mutation
198    // survives; if full replay runs (clear+replay), it'd be wiped.
199    let log_path = fresh_log_path();
200    let store_path = fresh_store_path();
201    let socket_path1 = fresh_socket_path();
202    let socket_path2 = fresh_socket_path();
203    let caja = Uuid::new_v4();
204
205    // Cycle 1: drive a deposit through the server. After shutdown the
206    // persistent store's marker should equal the log's last seq.
207    {
208        let executor = Executor::load_module(treasury_module()).expect("load");
209        let mut log = EventLog::open(&log_path).expect("open log");
210        let mut store = SurrealStore::new_persistent(&store_path).expect("open persistent");
211        seed_and_log(
212            &executor,
213            &mut store,
214            &mut log,
215            "Caja",
216            caja,
217            json!({
218                "id": caja.to_string(),
219                "name": "A",
220                "saldo": 100_000_i64,
221                "currency": "USD",
222            }),
223        )
224        .expect("seed");
225        // We end the WAL flow without running run_server in this cycle —
226        // the next cycle is the one that exercises the skip path.
227        drop(store);
228        drop(log);
229        drop(executor);
230    }
231
232    // Out-of-band mutation: open the persistent store directly and
233    // change the saldo. Marker stays at the same seq.
234    {
235        let mut store = SurrealStore::new_persistent(&store_path).expect("reopen for poison");
236        store.seed(
237            "Caja",
238            caja,
239            json!({
240                "id": caja.to_string(),
241                "name": "A",
242                "saldo": 999_999_i64, // poison
243                "currency": "USD",
244            }),
245        );
246        // The marker we set during the WAL flow stays intact — seed()
247        // alone does not bump it.
248    }
249
250    // Cycle 2: run_server with the poisoned store. Marker == log_last
251    // (still 0 from the seed) → skip path → poison saldo survives.
252    let executor = Executor::load_module(treasury_module()).expect("load");
253    let log = EventLog::open(&log_path).expect("reopen log");
254    let store = SurrealStore::new_persistent(&store_path).expect("reopen final");
255
256    let socket_for_client = socket_path1.clone();
257    let client = thread::spawn(move || -> Result<(), String> {
258        let mut conn = connect_with_retry(&socket_for_client);
259        let resp = exchange(
260            &mut conn,
261            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
262        );
263        let saldo = resp["value"]["saldo"].as_i64();
264        let _ = exchange(&mut conn, json!({"op": "shutdown"}));
265        if saldo != Some(999_999) {
266            return Err(format!(
267                "skip-replay should preserve out-of-band saldo (999_999), got {:?}",
268                saldo
269            ));
270        }
271        Ok(())
272    });
273
274    run_server(executor, log, store, None, &socket_path1).expect("server clean exit");
275    client.join().unwrap().expect("client assertions");
276
277    // Cycle 3: explicitly invalidate the marker (simulating a backend
278    // that lost track) and confirm full replay restores log-canonical
279    // state — wiping the poison.
280    {
281        let mut store = SurrealStore::new_persistent(&store_path).expect("reopen for marker reset");
282        // Force the marker into the "uninitialized" state by clearing
283        // and reseeding the legitimate record without bumping it. The
284        // simplest way is to clear() then re-seed; clear nukes
285        // last_applied_seq.
286        store.clear().expect("clear");
287        store.seed(
288            "Caja",
289            caja,
290            json!({
291                "id": caja.to_string(),
292                "name": "A",
293                "saldo": 999_999_i64, // poison still present
294                "currency": "USD",
295            }),
296        );
297        // last_applied_seq is now None → mismatch with log_last → full replay path.
298    }
299
300    let executor = Executor::load_module(treasury_module()).expect("load");
301    let log = EventLog::open(&log_path).expect("reopen");
302    let store = SurrealStore::new_persistent(&store_path).expect("reopen");
303
304    let socket_for_client = socket_path2.clone();
305    let client = thread::spawn(move || -> Result<(), String> {
306        let mut conn = connect_with_retry(&socket_for_client);
307        let resp = exchange(
308            &mut conn,
309            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
310        );
311        let saldo = resp["value"]["saldo"].as_i64();
312        let _ = exchange(&mut conn, json!({"op": "shutdown"}));
313        if saldo != Some(100_000) {
314            return Err(format!(
315                "full replay should restore canonical saldo (100_000), got {:?}",
316                saldo
317            ));
318        }
319        Ok(())
320    });
321
322    run_server(executor, log, store, None, &socket_path2).expect("server clean exit");
323    client.join().unwrap().expect("client assertions");
324
325    let _ = std::fs::remove_file(&log_path);
326    let _ = std::fs::remove_dir_all(&store_path);
327}