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::{EventLog, seed_and_log};
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::{Value, json};
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 =
236            SurrealStore::new_persistent(&store_path).expect("reopen for poison");
237        store.seed(
238            "Caja",
239            caja,
240            json!({
241                "id": caja.to_string(),
242                "name": "A",
243                "saldo": 999_999_i64, // poison
244                "currency": "USD",
245            }),
246        );
247        // The marker we set during the WAL flow stays intact — seed()
248        // alone does not bump it.
249    }
250
251    // Cycle 2: run_server with the poisoned store. Marker == log_last
252    // (still 0 from the seed) → skip path → poison saldo survives.
253    let executor = Executor::load_module(treasury_module()).expect("load");
254    let log = EventLog::open(&log_path).expect("reopen log");
255    let store = SurrealStore::new_persistent(&store_path).expect("reopen final");
256
257    let socket_for_client = socket_path1.clone();
258    let client = thread::spawn(move || -> Result<(), String> {
259        let mut conn = connect_with_retry(&socket_for_client);
260        let resp = exchange(
261            &mut conn,
262            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
263        );
264        let saldo = resp["value"]["saldo"].as_i64();
265        let _ = exchange(&mut conn, json!({"op": "shutdown"}));
266        if saldo != Some(999_999) {
267            return Err(format!(
268                "skip-replay should preserve out-of-band saldo (999_999), got {:?}",
269                saldo
270            ));
271        }
272        Ok(())
273    });
274
275    run_server(executor, log, store, None, &socket_path1).expect("server clean exit");
276    client.join().unwrap().expect("client assertions");
277
278    // Cycle 3: explicitly invalidate the marker (simulating a backend
279    // that lost track) and confirm full replay restores log-canonical
280    // state — wiping the poison.
281    {
282        let mut store = SurrealStore::new_persistent(&store_path).expect("reopen for marker reset");
283        // Force the marker into the "uninitialized" state by clearing
284        // and reseeding the legitimate record without bumping it. The
285        // simplest way is to clear() then re-seed; clear nukes
286        // last_applied_seq.
287        store.clear().expect("clear");
288        store.seed(
289            "Caja",
290            caja,
291            json!({
292                "id": caja.to_string(),
293                "name": "A",
294                "saldo": 999_999_i64, // poison still present
295                "currency": "USD",
296            }),
297        );
298        // last_applied_seq is now None → mismatch with log_last → full replay path.
299    }
300
301    let executor = Executor::load_module(treasury_module()).expect("load");
302    let log = EventLog::open(&log_path).expect("reopen");
303    let store = SurrealStore::new_persistent(&store_path).expect("reopen");
304
305    let socket_for_client = socket_path2.clone();
306    let client = thread::spawn(move || -> Result<(), String> {
307        let mut conn = connect_with_retry(&socket_for_client);
308        let resp = exchange(
309            &mut conn,
310            json!({"op": "load", "entity": "Caja", "id": caja.to_string()}),
311        );
312        let saldo = resp["value"]["saldo"].as_i64();
313        let _ = exchange(&mut conn, json!({"op": "shutdown"}));
314        if saldo != Some(100_000) {
315            return Err(format!(
316                "full replay should restore canonical saldo (100_000), got {:?}",
317                saldo
318            ));
319        }
320        Ok(())
321    });
322
323    run_server(executor, log, store, None, &socket_path2).expect("server clean exit");
324    client.join().unwrap().expect("client assertions");
325
326    let _ = std::fs::remove_file(&log_path);
327    let _ = std::fs::remove_dir_all(&store_path);
328}