⎇ d3649bfd1a80153652c2a028315369bb4b184024
crates/modules/nakui/core/src/drift.rs
Historia
Quién tocó qué
crudo
chore: absorbe nakui (ERP matemático) en modules/nakui
1//! Drift detection: compare two snapshots of store state and surface
2//! the records that differ.
3//!
4//! "Drift" here means the live store has departed from what the log can
5//! reproduce. The `Store::hash_state` contract makes the binary check
6//! cheap (32 bytes); when those disagree, `compare_states` walks both
7//! enumerations and produces a diff list the operator can act on.
8//!
9//! No IO in this module. The wire bits (asking a `nakui run` server for
10//! its hash and records) live in the CLI; this is the pure comparison
11//! used by both the CLI and any future automated drift-watcher.
12
13use serde::Serialize;
14use serde_json::Value;
15use std::collections::HashMap;
16use std::io::{BufRead, BufReader, Write};
17use std::os::unix::net::UnixStream;
18use std::path::Path;
19use thiserror::Error;
20use uuid::Uuid;
21
22use crate::event_log::{EventLog, replay};
23use crate::store::Store;
24
25/// A single record-level difference between two snapshots. Variants are
26/// labeled from the perspective of the operator running the check: the
27/// "log" side is the canonical state (what the log replays to), the
28/// "server" side is the live state being audited.
29#[derive(Debug, Clone, PartialEq, Serialize)]
30#[serde(tag = "kind", rename_all = "snake_case")]
31pub enum DriftDiff {
32 /// Server has a record the log doesn't know about. Phantom data —
33 /// either an out-of-band write, or a successful op that never
34 /// reached the WAL (which would itself be a kernel bug).
35 OnlyOnServer {
36 entity: String,
37 id: Uuid,
38 value: Value,
39 },
40 /// Log expects a record the server lost. Either the server's apply
41 /// rolled back without a reconcile, or someone deleted a record
42 /// out-of-band.
43 OnlyInLog {
44 entity: String,
45 id: Uuid,
46 value: Value,
47 },
48 /// Same (entity, id) on both sides but the values differ — the most
49 /// dangerous case, because it means a logged event was overwritten
50 /// or a field was tampered with.
51 Tampered {
52 entity: String,
53 id: Uuid,
54 log_value: Value,
55 server_value: Value,
56 },
57}
58
59#[derive(Debug, Clone, Serialize)]
60pub struct DriftReport {
61 pub log_hash: [u8; 32],
62 pub server_hash: [u8; 32],
63 pub log_records: usize,
64 pub server_records: usize,
65 /// Empty iff the two snapshots are byte-identical. Sorted by
66 /// (entity, id_bytes) so two runs against the same drift produce
67 /// the same report.
68 pub diffs: Vec<DriftDiff>,
69}
70
71impl DriftReport {
72 pub fn in_sync(&self) -> bool {
73 self.log_hash == self.server_hash && self.diffs.is_empty()
74 }
75}
76
77/// Pure comparison: take two canonical-order enumerations (as returned
78/// by `Store::iter`) plus their hashes, and return the diff list.
79///
80/// Inputs need not be pre-sorted — we re-key by (entity, id) and walk
81/// the union — but if the iterators were produced via `Store::iter`,
82/// they're already in canonical order and the report's `diffs` will be
83/// emitted in that same order.
84pub fn compare_states(
85 log_records: Vec<(String, Uuid, Value)>,
86 log_hash: [u8; 32],
87 server_records: Vec<(String, Uuid, Value)>,
88 server_hash: [u8; 32],
89) -> DriftReport {
90 let log_count = log_records.len();
91 let server_count = server_records.len();
92
93 let mut log_map: HashMap<(String, Uuid), Value> = log_records
94 .into_iter()
95 .map(|(e, id, v)| ((e, id), v))
96 .collect();
97 let server_map: HashMap<(String, Uuid), Value> = server_records
98 .into_iter()
99 .map(|(e, id, v)| ((e, id), v))
100 .collect();
101
102 let mut diffs: Vec<DriftDiff> = Vec::new();
103 for ((entity, id), server_value) in &server_map {
104 match log_map.remove(&(entity.clone(), *id)) {
105 None => diffs.push(DriftDiff::OnlyOnServer {
106 entity: entity.clone(),
107 id: *id,
108 value: server_value.clone(),
109 }),
110 Some(log_value) => {
111 if log_value != *server_value {
112 diffs.push(DriftDiff::Tampered {
113 entity: entity.clone(),
114 id: *id,
115 log_value,
116 server_value: server_value.clone(),
117 });
118 }
119 }
120 }
121 }
122 // Whatever is left in log_map is missing on the server.
123 for ((entity, id), value) in log_map {
124 diffs.push(DriftDiff::OnlyInLog { entity, id, value });
125 }
126
127 // Canonical sort: (entity, id_bytes), then by variant kind so
128 // diff ordering is fully deterministic even when the same key
129 // appears (which it can't here, but defensively).
130 diffs.sort_by(|a, b| {
131 let (ea, ia) = key(a);
132 let (eb, ib) = key(b);
133 ea.cmp(eb)
134 .then_with(|| ia.as_bytes().cmp(ib.as_bytes()))
135 .then_with(|| variant_order(a).cmp(&variant_order(b)))
136 });
137
138 DriftReport {
139 log_hash,
140 server_hash,
141 log_records: log_count,
142 server_records: server_count,
143 diffs,
144 }
145}
146
147fn key(d: &DriftDiff) -> (&str, &Uuid) {
148 match d {
149 DriftDiff::OnlyOnServer { entity, id, .. }
150 | DriftDiff::OnlyInLog { entity, id, .. }
151 | DriftDiff::Tampered { entity, id, .. } => (entity.as_str(), id),
152 }
153}
154
155fn variant_order(d: &DriftDiff) -> u8 {
156 match d {
157 DriftDiff::OnlyInLog { .. } => 0,
158 DriftDiff::Tampered { .. } => 1,
159 DriftDiff::OnlyOnServer { .. } => 2,
160 }
161}
162
163#[derive(Debug, Error)]
164pub enum DriftError {
165 #[error("open log: {0}")]
166 Log(#[from] crate::event_log::LogError),
167 #[error("replay log: {0}")]
168 Replay(#[from] crate::event_log::ReplayError),
169 #[error("store: {0}")]
170 Store(#[from] crate::store::StoreError),
171 #[error("connect to server socket: {0}")]
172 Connect(#[source] std::io::Error),
173 #[error("server io: {0}")]
174 Io(#[from] std::io::Error),
175 #[error("server response not json: {0}")]
176 Parse(#[from] serde_json::Error),
177 #[error("server returned error for `{op}`: {msg}")]
178 Server { op: String, msg: String },
179 #[error("server response missing field `{field}` for op `{op}`")]
180 MissingField { op: String, field: String },
181 #[error("server hash `{0}` is not 32 hex bytes")]
182 BadHash(String),
183}
184
185/// Audit a live `nakui run` server against the canonical state derived
186/// from a log file.
187///
188/// Cheap path: ask the server for `hash_state`, replay the log locally,
189/// hash that. If the hashes match, we return immediately with an empty
190/// diff list — no large `dump_records` round-trip.
191///
192/// Expensive path: hashes differ. Pull the full record dump from the
193/// server, run `compare_states`, return the structured report.
194pub fn check_against_socket(
195 log_path: &Path,
196 socket_path: &Path,
197) -> Result<DriftReport, DriftError> {
198 // Local: replay log → MemoryStore, snapshot.
199 let log = EventLog::open(log_path)?;
200 let local_store = replay(&log)?;
201 let local_records: Vec<(String, Uuid, Value)> = local_store.iter()?.collect();
202 let local_hash = local_store.hash_state()?;
203
204 // Wire: open the connection once and reuse it for both requests.
205 let stream = UnixStream::connect(socket_path).map_err(DriftError::Connect)?;
206 let mut conn = SocketClient::new(stream)?;
207
208 // Cheap path.
209 let hash_resp = conn.exchange(serde_json::json!({"op": "hash_state"}))?;
210 require_ok(&hash_resp, "hash_state")?;
211 let server_hash = parse_hash(&hash_resp, "hash_state")?;
212 let server_count = hash_resp
213 .get("records")
214 .and_then(Value::as_u64)
215 .ok_or_else(|| DriftError::MissingField {
216 op: "hash_state".into(),
217 field: "records".into(),
218 })? as usize;
219
220 if server_hash == local_hash {
221 return Ok(DriftReport {
222 log_hash: local_hash,
223 server_hash,
224 log_records: local_records.len(),
225 server_records: server_count,
226 diffs: Vec::new(),
227 });
228 }
229
230 // Expensive path: pull the full server snapshot.
231 let dump_resp = conn.exchange(serde_json::json!({"op": "dump_records"}))?;
232 require_ok(&dump_resp, "dump_records")?;
233 let server_records = parse_records(&dump_resp)?;
234
235 Ok(compare_states(
236 local_records,
237 local_hash,
238 server_records,
239 server_hash,
240 ))
241}
242
243struct SocketClient {
244 writer: UnixStream,
245 reader: BufReader<UnixStream>,
246}
247
248impl SocketClient {
249 fn new(stream: UnixStream) -> Result<Self, DriftError> {
250 let reader_stream = stream.try_clone()?;
251 Ok(Self {
252 writer: stream,
253 reader: BufReader::new(reader_stream),
254 })
255 }
256
257 fn exchange(&mut self, req: Value) -> Result<Value, DriftError> {
258 let mut bytes = serde_json::to_vec(&req).expect("request serializes");
259 bytes.push(b'\n');
260 self.writer.write_all(&bytes)?;
261 let mut line = String::new();
262 let n = self.reader.read_line(&mut line)?;
263 if n == 0 {
264 return Err(DriftError::Io(std::io::Error::new(
265 std::io::ErrorKind::UnexpectedEof,
266 "server closed connection without responding",
267 )));
268 }
269 Ok(serde_json::from_str(line.trim())?)
270 }
271}
272
273fn require_ok(resp: &Value, op: &str) -> Result<(), DriftError> {
274 if resp.get("ok").and_then(Value::as_bool) == Some(true) {
275 Ok(())
276 } else {
277 Err(DriftError::Server {
278 op: op.into(),
279 msg: resp
280 .get("error")
281 .and_then(Value::as_str)
282 .unwrap_or("(no error message)")
283 .to_string(),
284 })
285 }
286}
287
288fn parse_hash(resp: &Value, op: &str) -> Result<[u8; 32], DriftError> {
289 let s = resp
290 .get("hash")
291 .and_then(Value::as_str)
292 .ok_or_else(|| DriftError::MissingField {
293 op: op.into(),
294 field: "hash".into(),
295 })?;
296 if s.len() != 64 {
297 return Err(DriftError::BadHash(s.into()));
298 }
299 let mut out = [0u8; 32];
300 for (i, byte) in out.iter_mut().enumerate() {
301 let hi = hex_nibble(s.as_bytes()[i * 2]).ok_or_else(|| DriftError::BadHash(s.into()))?;
302 let lo =
303 hex_nibble(s.as_bytes()[i * 2 + 1]).ok_or_else(|| DriftError::BadHash(s.into()))?;
304 *byte = (hi << 4) | lo;
305 }
306 Ok(out)
307}
308
309fn hex_nibble(c: u8) -> Option<u8> {
310 match c {
311 b'0'..=b'9' => Some(c - b'0'),
312 b'a'..=b'f' => Some(c - b'a' + 10),
313 b'A'..=b'F' => Some(c - b'A' + 10),
314 _ => None,
315 }
316}
317
318fn parse_records(resp: &Value) -> Result<Vec<(String, Uuid, Value)>, DriftError> {
319 let arr = resp
320 .get("records")
321 .and_then(Value::as_array)
322 .ok_or_else(|| DriftError::MissingField {
323 op: "dump_records".into(),
324 field: "records".into(),
325 })?;
326 let mut out: Vec<(String, Uuid, Value)> = Vec::with_capacity(arr.len());
327 for item in arr {
328 let entity = item
329 .get("entity")
330 .and_then(Value::as_str)
331 .ok_or_else(|| DriftError::MissingField {
332 op: "dump_records".into(),
333 field: "records[].entity".into(),
334 })?
335 .to_string();
336 let id_str = item
337 .get("id")
338 .and_then(Value::as_str)
339 .ok_or_else(|| DriftError::MissingField {
340 op: "dump_records".into(),
341 field: "records[].id".into(),
342 })?;
343 let id = Uuid::parse_str(id_str).map_err(|_| DriftError::MissingField {
344 op: "dump_records".into(),
345 field: format!("records[].id (not uuid: {})", id_str),
346 })?;
347 let value = item
348 .get("value")
349 .cloned()
350 .ok_or_else(|| DriftError::MissingField {
351 op: "dump_records".into(),
352 field: "records[].value".into(),
353 })?;
354 out.push((entity, id, value));
355 }
356 Ok(out)
357}
358
359#[cfg(test)]
360mod tests {
361 use super::*;
362 use serde_json::json;
363
364 fn h(byte: u8) -> [u8; 32] {
365 [byte; 32]
366 }
367
368 #[test]
369 fn empty_inputs_yield_no_diffs() {
370 let report = compare_states(Vec::new(), h(0), Vec::new(), h(0));
371 assert!(report.in_sync());
372 assert!(report.diffs.is_empty());
373 }
374
375 #[test]
376 fn equal_records_yield_no_diffs_even_if_hashes_were_lied_to() {
377 // The function compares records, not hashes — hash equality is
378 // the operator's fast-path, but the report's truth is the diffs.
379 let a = Uuid::new_v4();
380 let log = vec![(
381 "Caja".to_string(),
382 a,
383 json!({"saldo": 100}),
384 )];
385 let server = vec![(
386 "Caja".to_string(),
387 a,
388 json!({"saldo": 100}),
389 )];
390 let report = compare_states(log, h(1), server, h(2));
391 assert!(report.diffs.is_empty(), "records equal → no diffs");
392 }
393
394 #[test]
395 fn detects_only_on_server() {
396 let a = Uuid::new_v4();
397 let b = Uuid::new_v4();
398 let log = vec![(
399 "Caja".to_string(),
400 a,
401 json!({"saldo": 100}),
402 )];
403 let server = vec![
404 ("Caja".to_string(), a, json!({"saldo": 100})),
405 ("Caja".to_string(), b, json!({"saldo": 999})),
406 ];
407 let report = compare_states(log, h(0), server, h(1));
408 assert_eq!(report.diffs.len(), 1);
409 match &report.diffs[0] {
410 DriftDiff::OnlyOnServer { entity, id, .. } => {
411 assert_eq!(entity, "Caja");
412 assert_eq!(*id, b);
413 }
414 other => panic!("expected OnlyOnServer, got {:?}", other),
415 }
416 }
417
418 #[test]
419 fn detects_only_in_log() {
420 let a = Uuid::new_v4();
421 let log = vec![("Caja".to_string(), a, json!({"saldo": 100}))];
422 let server = vec![];
423 let report = compare_states(log, h(0), server, h(1));
424 assert_eq!(report.diffs.len(), 1);
425 match &report.diffs[0] {
426 DriftDiff::OnlyInLog { id, .. } => assert_eq!(*id, a),
427 other => panic!("expected OnlyInLog, got {:?}", other),
428 }
429 }
430
431 #[test]
432 fn detects_tampered() {
433 let a = Uuid::new_v4();
434 let log = vec![("Caja".to_string(), a, json!({"saldo": 100}))];
435 let server = vec![("Caja".to_string(), a, json!({"saldo": 999}))];
436 let report = compare_states(log, h(0), server, h(1));
437 assert_eq!(report.diffs.len(), 1);
438 match &report.diffs[0] {
439 DriftDiff::Tampered {
440 id,
441 log_value,
442 server_value,
443 ..
444 } => {
445 assert_eq!(*id, a);
446 assert_eq!(log_value["saldo"], json!(100));
447 assert_eq!(server_value["saldo"], json!(999));
448 }
449 other => panic!("expected Tampered, got {:?}", other),
450 }
451 }
452
453 #[test]
454 fn diffs_emerge_in_canonical_order() {
455 // Two entities, mixed drift kinds. Result must be sorted by
456 // (entity, id_bytes) so two runs produce the same report.
457 let id_caja = Uuid::nil(); // sorts first byte-wise
458 let id_mov = Uuid::from_u128(u128::MAX);
459
460 let log = vec![
461 ("Movimiento".to_string(), id_mov, json!({"x": 1})),
462 ];
463 let server = vec![
464 ("Caja".to_string(), id_caja, json!({"saldo": 0})),
465 ];
466 let report = compare_states(log, h(0), server, h(1));
467 assert_eq!(report.diffs.len(), 2);
468 // Caja sorts before Movimiento.
469 match (&report.diffs[0], &report.diffs[1]) {
470 (DriftDiff::OnlyOnServer { entity: e1, .. }, DriftDiff::OnlyInLog { entity: e2, .. }) => {
471 assert_eq!(e1, "Caja");
472 assert_eq!(e2, "Movimiento");
473 }
474 other => panic!("unexpected order: {:?}", other),
475 }
476 }
477
478 #[test]
479 fn in_sync_requires_both_hashes_and_no_diffs() {
480 // Defensive: if hashes match but somehow diffs is non-empty
481 // (caller mismatch), in_sync says no.
482 let report = DriftReport {
483 log_hash: h(0),
484 server_hash: h(0),
485 log_records: 1,
486 server_records: 1,
487 diffs: vec![DriftDiff::Tampered {
488 entity: "x".into(),
489 id: Uuid::nil(),
490 log_value: json!(1),
491 server_value: json!(2),
492 }],
493 };
494 assert!(!report.in_sync());
495 }
496}