tawasuyu/brahman

1//! End-to-end tests for the snapshot lifecycle: capture, compact, and
2//! boot from snapshot. Plus the schema-hash binding that ties a snapshot
3//! to the bundle that produced it.
4
5use std::io::{BufRead, BufReader, Write};
6use std::os::unix::net::UnixStream;
7use std::path::{Path, PathBuf};
8use std::thread;
9use std::time::Duration;
10
11use nakui_core::event_log::{
12    EventLog, Snapshot, SnapshotMismatchError, execute_and_log, replay, seed_and_log,
13};
14use nakui_core::executor::Executor;
15use nakui_core::run::run_server;
16use nakui_core::store::{MemoryStore, Store};
17use serde_json::{Value, json};
18use uuid::Uuid;
19
20fn workspace_root() -> PathBuf {
21    Path::new(env!("CARGO_MANIFEST_DIR"))
22        .parent()
23        .expect("workspace root above core/")
24        .to_path_buf()
25}
26
27fn treasury_module() -> PathBuf {
28    workspace_root().join("modules/treasury")
29}
30
31fn fresh_log_path() -> PathBuf {
32    std::env::temp_dir().join(format!("nakui_snap_log_{}.jsonl", Uuid::new_v4()))
33}
34
35fn fresh_snap_path() -> PathBuf {
36    std::env::temp_dir().join(format!("nakui_snap_{}.json", Uuid::new_v4()))
37}
38
39fn fresh_socket_path() -> PathBuf {
40    std::env::temp_dir().join(format!("nakui_snap_run_{}.sock", Uuid::new_v4()))
41}
42
43fn seed_caja(exec: &Executor, store: &mut MemoryStore, log: &mut EventLog, id: Uuid, saldo: i64) {
44    seed_and_log(
45        exec,
46        store,
47        log,
48        "Caja",
49        id,
50        json!({"id": id.to_string(), "name": "A", "saldo": saldo, "currency": "USD"}),
51    )
52    .unwrap();
53}
54
55fn deposit(exec: &Executor, store: &mut MemoryStore, log: &mut EventLog, caja: Uuid, monto: i64) {
56    execute_and_log(
57        exec,
58        store,
59        log,
60        "register_cash_move",
61        &[("caja", caja)],
62        json!({
63            "monto": monto,
64            "tipo": "in",
65            "timestamp": "2026-05-04T10:00:00Z",
66            "memo": "x",
67            "movimiento_id": Uuid::new_v4().to_string(),
68        }),
69    )
70    .unwrap();
71}
72
73#[test]
74fn module_schema_hash_is_stable_and_independent_of_load_order() {
75    let exec1 = Executor::load_module(treasury_module()).expect("load 1");
76    let exec2 = Executor::load_module(treasury_module()).expect("load 2");
77    assert_eq!(
78        exec1.module_schema_hash(),
79        exec2.module_schema_hash(),
80        "two clean loads of the same module → identical module hash"
81    );
82}
83
84#[test]
85fn capture_records_executor_hash_legacy_does_not() {
86    let exec = Executor::load_module(treasury_module()).expect("load");
87    let mut store = MemoryStore::new();
88    store.seed("Caja", Uuid::new_v4(), json!({"x": 1}));
89
90    let captured = Snapshot::capture(&store, 0, &exec);
91    assert_eq!(captured.schema_hash, Some(exec.module_schema_hash()));
92
93    let legacy = Snapshot::from_memory_store(&store, 0);
94    assert_eq!(legacy.schema_hash, None, "legacy constructor opts out");
95
96    captured
97        .ensure_compatible_with(&exec)
98        .expect("captured snapshot is compatible with the executor that built it");
99    legacy
100        .ensure_compatible_with(&exec)
101        .expect("legacy snapshot has no hash → no check, passes");
102}
103
104#[test]
105fn ensure_compatible_with_rejects_mismatched_hash() {
106    let exec = Executor::load_module(treasury_module()).expect("load");
107    let mut snap = Snapshot::capture(&MemoryStore::new(), 0, &exec);
108    // Tamper with the hash to simulate a snapshot from a different bundle.
109    snap.schema_hash = Some([0xAB; 32]);
110    match snap.ensure_compatible_with(&exec) {
111        Err(SnapshotMismatchError::SchemaMismatch { .. }) => {}
112        other => panic!("expected SchemaMismatch, got {:?}", other),
113    }
114}
115
116#[test]
117fn snapshot_then_compact_then_run_server_resumes_correctly() {
118    // The full operator workflow:
119    //   1. Run a series of WAL-validated ops.
120    //   2. Capture a snapshot covering the last seq.
121    //   3. Compact the log so it only retains entries past snap.seq.
122    //   4. Start a server pointing at the (compacted) log + snapshot.
123    //   5. Confirm the server's state is correct via the load op.
124    //
125    // After step 3 the log alone can't reconstruct the state — the
126    // snapshot is the only thing that proves the server isn't lying.
127    let log_path = fresh_log_path();
128    let snap_path = fresh_snap_path();
129    let socket_path = fresh_socket_path();
130
131    let caja = Uuid::new_v4();
132    let snap_seq;
133    let captured_module_hash;
134    {
135        let exec = Executor::load_module(treasury_module()).expect("load");
136        captured_module_hash = exec.module_schema_hash();
137        let mut log = EventLog::open(&log_path).unwrap();
138        let mut store = MemoryStore::new();
139        seed_caja(&exec, &mut store, &mut log, caja, 100_000);
140        deposit(&exec, &mut store, &mut log, caja, 5_000);
141        deposit(&exec, &mut store, &mut log, caja, 7_500);
142
143        snap_seq = log.next_seq() - 1;
144        let snap = Snapshot::capture(&store, snap_seq, &exec);
145        snap.write(&snap_path).unwrap();
146        log.compact_through(snap_seq).unwrap();
147
148        // Sanity: after compaction the log has no surviving entries.
149        let surviving = log.entries().unwrap();
150        assert_eq!(surviving.len(), 0);
151        // But next_seq is preserved, so future appends keep monotonicity.
152        assert_eq!(log.next_seq(), snap_seq + 1);
153    }
154
155    // Verify the snapshot file carries the captured hash (resilient
156    // through write+read).
157    let reloaded = Snapshot::load(&snap_path).unwrap().unwrap();
158    assert_eq!(reloaded.schema_hash, Some(captured_module_hash));
159    assert_eq!(reloaded.seq, snap_seq);
160
161    // Boot the server with snapshot + compacted log.
162    let executor = Executor::load_module(treasury_module()).expect("reload");
163    let log = EventLog::open(&log_path).unwrap();
164    let store = MemoryStore::new();
165
166    let socket_for_client = socket_path.clone();
167    let client = thread::spawn(move || -> Result<(), String> {
168        let mut conn = connect_with_retry(&socket_for_client);
169        let resp = exchange(&mut conn, json!({"op": "load", "entity": "Caja", "id": caja.to_string()}));
170        if resp["value"]["saldo"].as_i64() != Some(112_500) {
171            return Err(format!(
172                "expected saldo 112_500 (100k seed + 5k + 7.5k from snapshot), got {}",
173                resp
174            ));
175        }
176        // Append a new op via the live server and load it back —
177        // confirms the WAL still works on top of a snapshot-loaded state.
178        let resp = exchange(
179            &mut conn,
180            json!({
181                "op": "execute",
182                "morphism": "register_cash_move",
183                "inputs": {"caja": caja.to_string()},
184                "params": {
185                    "monto": 1_000_i64,
186                    "tipo": "in",
187                    "timestamp": "2026-05-04T11:00:00Z",
188                    "memo": "post-snap",
189                    "movimiento_id": Uuid::new_v4().to_string(),
190                }
191            }),
192        );
193        if resp["ok"] != json!(true) {
194            return Err(format!("execute on snapshot-booted server failed: {}", resp));
195        }
196        let resp = exchange(&mut conn, json!({"op": "load", "entity": "Caja", "id": caja.to_string()}));
197        if resp["value"]["saldo"].as_i64() != Some(113_500) {
198            return Err(format!("post-execute saldo wrong: {}", resp));
199        }
200        send_shutdown(&mut conn);
201        Ok(())
202    });
203
204    run_server(executor, log, store, Some(reloaded), &socket_path).expect("server clean exit");
205    client.join().unwrap().expect("client assertions");
206
207    let _ = std::fs::remove_file(&log_path);
208    let _ = std::fs::remove_file(&snap_path);
209}
210
211#[test]
212fn run_server_refuses_snapshot_with_wrong_schema_hash() {
213    let log_path = fresh_log_path();
214    let socket_path = fresh_socket_path();
215
216    let caja = Uuid::new_v4();
217    {
218        let exec = Executor::load_module(treasury_module()).expect("load");
219        let mut log = EventLog::open(&log_path).unwrap();
220        let mut store = MemoryStore::new();
221        seed_caja(&exec, &mut store, &mut log, caja, 100_000);
222        deposit(&exec, &mut store, &mut log, caja, 5_000);
223    }
224
225    // Build a snapshot with a fabricated hash — simulates "snapshot
226    // taken under module A, loaded against module B."
227    let exec = Executor::load_module(treasury_module()).expect("reload");
228    let log = EventLog::open(&log_path).unwrap();
229    let snap_state = replay(&log).unwrap();
230    let last_seq = log.entries().unwrap().last().unwrap().seq();
231    let mut bad_snap = Snapshot::capture(&snap_state, last_seq, &exec);
232    bad_snap.schema_hash = Some([0xAB; 32]);
233
234    let store = MemoryStore::new();
235    let result = run_server(exec, log, store, Some(bad_snap), &socket_path);
236    assert!(
237        matches!(
238            result,
239            Err(nakui_core::run::RunError::SnapshotMismatch(_))
240        ),
241        "expected SnapshotMismatch, got {:?}",
242        result
243    );
244    // Socket must not have been bound.
245    assert!(!socket_path.exists());
246
247    let _ = std::fs::remove_file(&log_path);
248}
249
250#[test]
251fn run_server_detects_gap_between_snapshot_and_compacted_log() {
252    // Snapshot says it covers up to seq K. Log was compacted further,
253    // so its first remaining entry is K+5 — entries K+1..=K+4 are
254    // gone. run_server must refuse rather than silently fabricate a
255    // state that drops events.
256    let log_path = fresh_log_path();
257    let socket_path = fresh_socket_path();
258
259    let caja = Uuid::new_v4();
260    let exec = Executor::load_module(treasury_module()).expect("load");
261    {
262        let mut log = EventLog::open(&log_path).unwrap();
263        let mut store = MemoryStore::new();
264        seed_caja(&exec, &mut store, &mut log, caja, 100_000);
265        deposit(&exec, &mut store, &mut log, caja, 1_000);
266        deposit(&exec, &mut store, &mut log, caja, 1_000);
267        deposit(&exec, &mut store, &mut log, caja, 1_000);
268        deposit(&exec, &mut store, &mut log, caja, 1_000);
269        deposit(&exec, &mut store, &mut log, caja, 1_000);
270    }
271
272    // Snapshot at seq 0 (only the seed).
273    let mut log = EventLog::open(&log_path).unwrap();
274    let mut state = MemoryStore::new();
275    nakui_core::event_log::replay_with_snapshot_into(&log, None, &mut state).unwrap();
276    let snap = Snapshot::capture(&state, 0, &exec);
277
278    // Compact the log past the snapshot — drop seqs 0..=3, leaving
279    // entries from seq 4 onward. The snapshot can't reconstruct the
280    // missing tail.
281    log.compact_through(3).unwrap();
282    drop(log);
283
284    let exec = Executor::load_module(treasury_module()).expect("reload");
285    let log = EventLog::open(&log_path).unwrap();
286    let store = MemoryStore::new();
287    let result = run_server(exec, log, store, Some(snap), &socket_path);
288    match result {
289        Err(nakui_core::run::RunError::SnapshotGap {
290            snap_seq,
291            log_first_seq,
292            expected,
293        }) => {
294            assert_eq!(snap_seq, 0);
295            assert_eq!(expected, 1);
296            assert!(
297                log_first_seq >= 4,
298                "log's first surviving entry should be ≥ 4, got {}",
299                log_first_seq
300            );
301        }
302        other => panic!("expected SnapshotGap, got {:?}", other),
303    }
304
305    let _ = std::fs::remove_file(&log_path);
306}
307
308#[test]
309fn snapshot_write_overwrites_existing_atomically() {
310    // Two snapshots at different seqs written to the same path. The
311    // second must completely replace the first; load() returns the
312    // newer one.
313    let snap_path = fresh_snap_path();
314    let exec = Executor::load_module(treasury_module()).expect("load");
315
316    let s1 = Snapshot::capture(&MemoryStore::new(), 0, &exec);
317    s1.write(&snap_path).expect("write first");
318    let loaded = Snapshot::load(&snap_path).unwrap().unwrap();
319    assert_eq!(loaded.seq, 0);
320
321    // Now write a different snapshot to the same path.
322    let mut store = MemoryStore::new();
323    let id = Uuid::new_v4();
324    store.seed("Caja", id, json!({"id": id.to_string(), "saldo": 7}));
325    let s2 = Snapshot::capture(&store, 42, &exec);
326    s2.write(&snap_path).expect("overwrite");
327    let loaded = Snapshot::load(&snap_path).unwrap().unwrap();
328    assert_eq!(loaded.seq, 42, "second write must replace the first");
329    assert!(loaded.records.contains_key("Caja"));
330
331    // No leftover tempfile.
332    let writing_path = snap_path.with_extension("writing");
333    assert!(
334        !writing_path.exists(),
335        "tempfile must be renamed, not left behind"
336    );
337
338    let _ = std::fs::remove_file(&snap_path);
339}
340
341#[test]
342fn snapshot_write_recovers_from_stale_tempfile() {
343    // A prior write crashed after creating .writing but before rename.
344    // The next write must succeed regardless — File::create truncates
345    // the stale tempfile.
346    let snap_path = fresh_snap_path();
347    let writing_path = snap_path.with_extension("writing");
348
349    // Plant a stale tempfile with garbage content.
350    std::fs::write(&writing_path, b"junk from a prior crashed write").unwrap();
351    assert!(writing_path.exists());
352
353    let exec = Executor::load_module(treasury_module()).expect("load");
354    let snap = Snapshot::capture(&MemoryStore::new(), 0, &exec);
355    snap.write(&snap_path).expect("write despite stale tempfile");
356
357    // Tempfile should be renamed (not orphaned), so it's gone.
358    assert!(
359        !writing_path.exists(),
360        "stale tempfile must be consumed by the rename"
361    );
362    let loaded = Snapshot::load(&snap_path).unwrap().unwrap();
363    assert_eq!(loaded.seq, 0);
364
365    let _ = std::fs::remove_file(&snap_path);
366}
367
368// === helpers shared with the run-server protocol tests ===
369
370struct Conn {
371    writer: UnixStream,
372    reader: BufReader<UnixStream>,
373}
374
375fn connect_with_retry(path: &Path) -> Conn {
376    for _ in 0..200 {
377        if let Ok(stream) = UnixStream::connect(path) {
378            let reader_stream = stream.try_clone().expect("clone");
379            return Conn {
380                writer: stream,
381                reader: BufReader::new(reader_stream),
382            };
383        }
384        thread::sleep(Duration::from_millis(20));
385    }
386    panic!("server never started accepting on {}", path.display());
387}
388
389fn exchange(conn: &mut Conn, req: Value) -> Value {
390    let mut bytes = serde_json::to_vec(&req).unwrap();
391    bytes.push(b'\n');
392    conn.writer.write_all(&bytes).unwrap();
393    let mut line = String::new();
394    conn.reader.read_line(&mut line).unwrap();
395    serde_json::from_str(line.trim()).unwrap()
396}
397
398fn send_shutdown(conn: &mut Conn) {
399    let _ = exchange(conn, json!({"op": "shutdown"}));
400}