tawasuyu/brahman

1//! `nakui run` server: a long-lived process that holds an in-memory store
2//! reconstructed from the log, exposes a Unix Domain Socket, and serves
3//! line-delimited JSON requests to drive the kernel.
4//!
5//! Why UDS + line-JSON for V1:
6//!   - Multi-client without committing to a transport (HTTP/NATS later).
7//!   - Filesystem permissions gate access; no port exposure.
8//!   - Self-describing: `describe` returns the manifest's morphism specs
9//!     so an agent (human or LLM) can drive the server without external
10//!     docs.
11//!
12//! Concurrency: one connection at a time. Backed by `&mut Store`, the
13//! kernel is single-writer by design. Multiple clients queue in
14//! `accept()`. If/when we want concurrency, the right unit to parallelize
15//! is reads, not writes — that's a future refactor with locks at the
16//! right granularity.
17//!
18//! Recovery: every `execute` goes through `execute_and_log_with_recovery`
19//! so a transient apply failure auto-rebuilds the in-memory store from
20//! the log without taking the server down.
21
22use std::io::{BufRead, BufReader, Write};
23use std::os::unix::net::{UnixListener, UnixStream};
24use std::path::Path;
25
26use serde::Deserialize;
27use serde_json::{Value, json};
28use thiserror::Error;
29use uuid::Uuid;
30
31use crate::event_log::{
32    EventLog, RecoverableExecuteError, ReplayError, Snapshot, SnapshotMismatchError,
33    execute_and_log_with_recovery, replay_with_snapshot_into, verify_log,
34};
35use crate::executor::Executor;
36use crate::store::Store;
37
38#[derive(Debug, Error)]
39pub enum RunError {
40    #[error("io: {0}")]
41    Io(#[from] std::io::Error),
42    #[error("clear store on startup: {0}")]
43    Clear(#[source] crate::store::StoreError),
44    #[error("replay on startup: {0}")]
45    Replay(#[from] ReplayError),
46    #[error("log: {0}")]
47    Log(#[from] crate::event_log::LogError),
48    #[error("snapshot incompatible: {0}")]
49    SnapshotMismatch(#[from] SnapshotMismatchError),
50    #[error(
51        "snapshot/log gap: snapshot covers up to seq {snap_seq}, log's first remaining entry is seq {log_first_seq} (expected ≤ {expected})"
52    )]
53    SnapshotGap {
54        snap_seq: u64,
55        log_first_seq: u64,
56        expected: u64,
57    },
58}
59
60/// Run the server until a `shutdown` request is received or `accept`
61/// returns an unrecoverable error. On exit, removes the socket file.
62///
63/// Startup reconstruction:
64///   - With `Some(snapshot)`: validate its `schema_hash` against the
65///     executor, seed the store from the snapshot, replay only the log
66///     tail (entries with `seq > snapshot.seq`).
67///   - With `None`: full replay from seq 0. Slower for long logs.
68///
69/// In both cases the store is wiped first, so the server never serves
70/// requests against a state the log can't reproduce. This is true for
71/// `MemoryStore` and for persistent backends like `SurrealStore` —
72/// persistence is a durability property of the runtime cache, not a
73/// way to skip replay. (A future "skip replay if last_applied_seq
74/// matches" optimization would change that.)
75pub fn run_server<S: Store>(
76    executor: Executor,
77    mut log: EventLog,
78    mut store: S,
79    snapshot: Option<Snapshot>,
80    socket_path: &Path,
81) -> Result<(), RunError> {
82    startup_replay(&executor, &log, &mut store, snapshot.as_ref())?;
83
84    // Best-effort cleanup of stale sockets from a prior crashed run.
85    // Bind itself will fail if a live process is already listening.
86    let _ = std::fs::remove_file(socket_path);
87    let listener = UnixListener::bind(socket_path)?;
88
89    let result = accept_loop(&listener, &executor, &mut store, &mut log);
90    let _ = std::fs::remove_file(socket_path);
91    result
92}
93
94fn startup_replay<S: Store>(
95    executor: &Executor,
96    log: &EventLog,
97    store: &mut S,
98    snapshot: Option<&Snapshot>,
99) -> Result<(), RunError> {
100    // Snapshot validation runs first (cheap) so a bad snapshot is caught
101    // even when we'd otherwise take the skip-replay fast path.
102    if let Some(snap) = snapshot {
103        snap.ensure_compatible_with(executor)?;
104        let entries = log.entries()?;
105        if let Some(first) = entries.first() {
106            let expected = snap.seq.saturating_add(1);
107            if first.seq() > expected {
108                return Err(RunError::SnapshotGap {
109                    snap_seq: snap.seq,
110                    log_first_seq: first.seq(),
111                    expected,
112                });
113            }
114        }
115    }
116
117    // Fast path: persistent stores carry a `last_applied_seq` marker;
118    // when it matches the log's last seq, the store is verifiably in
119    // sync and we can skip the clear+replay entirely. Failures here
120    // (e.g. backend can't read meta) just fall through to full replay
121    // — never a correctness issue.
122    let log_last_seq = log.entries()?.last().map(|e| e.seq());
123    if let Ok(applied) = store.last_applied_seq() {
124        if applied == log_last_seq && applied.is_some() {
125            return Ok(());
126        }
127    }
128
129    store.clear().map_err(RunError::Clear)?;
130    replay_with_snapshot_into(log, snapshot, store)?;
131    Ok(())
132}
133
134fn accept_loop<S: Store>(
135    listener: &UnixListener,
136    executor: &Executor,
137    store: &mut S,
138    log: &mut EventLog,
139) -> Result<(), RunError> {
140    loop {
141        let (stream, _addr) = listener.accept()?;
142        let shutdown = handle_connection(stream, executor, store, log);
143        if shutdown {
144            return Ok(());
145        }
146    }
147}
148
149#[derive(Debug, Deserialize)]
150#[serde(tag = "op", rename_all = "snake_case")]
151enum Request {
152    Execute {
153        morphism: String,
154        #[serde(default)]
155        inputs: std::collections::BTreeMap<String, Uuid>,
156        #[serde(default)]
157        params: Value,
158    },
159    Load {
160        entity: String,
161        id: Uuid,
162    },
163    Describe,
164    Verify,
165    /// Return the SHA-256 of the live store's full state plus a record
166    /// count. Used by the drift detector as the cheap fast-path check
167    /// before asking for the full record dump.
168    HashState,
169    /// Return every record on the server in canonical order. Used after
170    /// a hash mismatch to compute the per-record diff. Response can be
171    /// large — the operator opts into it.
172    DumpRecords,
173    Shutdown,
174}
175
176/// Process one connection. Returns `true` if the client requested
177/// shutdown — the caller should stop the accept loop after the response
178/// has been flushed.
179///
180/// IO errors on a single connection don't kill the server: we log to
181/// stderr and move on. Only a request-level shutdown ends the loop.
182fn handle_connection<S: Store>(
183    stream: UnixStream,
184    executor: &Executor,
185    store: &mut S,
186    log: &mut EventLog,
187) -> bool {
188    let mut writer = match stream.try_clone() {
189        Ok(s) => s,
190        Err(e) => {
191            eprintln!("nakui run: clone stream: {}", e);
192            return false;
193        }
194    };
195    let reader = BufReader::new(stream);
196    for line in reader.lines() {
197        let line = match line {
198            Ok(l) => l,
199            Err(e) => {
200                eprintln!("nakui run: read: {}", e);
201                return false;
202            }
203        };
204        if line.trim().is_empty() {
205            continue;
206        }
207        let (response, shutdown) = dispatch(&line, executor, store, log);
208        let bytes = serde_json::to_vec(&response).expect("response serializes");
209        if let Err(e) = writer.write_all(&bytes).and_then(|_| writer.write_all(b"\n")) {
210            eprintln!("nakui run: write: {}", e);
211            return false;
212        }
213        if shutdown {
214            let _ = writer.flush();
215            return true;
216        }
217    }
218    false
219}
220
221fn dispatch<S: Store>(
222    line: &str,
223    executor: &Executor,
224    store: &mut S,
225    log: &mut EventLog,
226) -> (Value, bool) {
227    let req: Request = match serde_json::from_str(line) {
228        Ok(r) => r,
229        Err(e) => return (error_response(&format!("bad request: {}", e)), false),
230    };
231    match req {
232        Request::Execute {
233            morphism,
234            inputs,
235            params,
236        } => {
237            let inputs_vec: Vec<(&str, Uuid)> =
238                inputs.iter().map(|(k, v)| (k.as_str(), *v)).collect();
239            match execute_and_log_with_recovery(
240                executor,
241                store,
242                log,
243                &morphism,
244                &inputs_vec,
245                params,
246            ) {
247                Ok(ops) => (
248                    json!({
249                        "ok": true,
250                        "seq": log.next_seq().saturating_sub(1),
251                        "ops": ops,
252                        "schema_hash": executor.schema_hash(&morphism).map(|h| hex_encode(&h)),
253                    }),
254                    false,
255                ),
256                Err(RecoverableExecuteError::PreLog(e)) => (
257                    json!({"ok": false, "stage": "pre_log", "error": e.to_string()}),
258                    false,
259                ),
260                Err(RecoverableExecuteError::LogAppend(e)) => (
261                    json!({"ok": false, "stage": "log_append", "error": e.to_string()}),
262                    false,
263                ),
264                Err(e @ RecoverableExecuteError::Unrecoverable { .. }) => (
265                    json!({"ok": false, "stage": "unrecoverable", "error": e.to_string()}),
266                    false,
267                ),
268            }
269        }
270        Request::Load { entity, id } => {
271            let value = store.load(&entity, id);
272            (json!({"ok": true, "value": value}), false)
273        }
274        Request::Describe => {
275            let hashes: std::collections::BTreeMap<String, String> = executor
276                .schema_hashes
277                .iter()
278                .map(|(k, v)| (k.clone(), hex_encode(v)))
279                .collect();
280            (
281                json!({
282                    "ok": true,
283                    "protocol": 1,
284                    "module": executor.manifest.module,
285                    "schemas": executor.manifest.effective_schemas(),
286                    "morphisms": executor.manifest.morphisms,
287                    "schema_hashes": hashes,
288                }),
289                false,
290            )
291        }
292        Request::Verify => match verify_log(log, executor) {
293            Ok(()) => {
294                let entries = log
295                    .entries()
296                    .map(|es| es.len())
297                    .unwrap_or(0);
298                (json!({"ok": true, "entries": entries}), false)
299            }
300            Err(e) => (
301                json!({"ok": false, "error": e.to_string()}),
302                false,
303            ),
304        },
305        Request::HashState => {
306            let records: Vec<_> = match store.iter() {
307                Ok(it) => it.collect(),
308                Err(e) => return (json!({"ok": false, "error": e.to_string()}), false),
309            };
310            let count = records.len();
311            let hash = match store.hash_state() {
312                Ok(h) => h,
313                Err(e) => return (json!({"ok": false, "error": e.to_string()}), false),
314            };
315            (
316                json!({
317                    "ok": true,
318                    "hash": hex_encode(&hash),
319                    "records": count,
320                }),
321                false,
322            )
323        }
324        Request::DumpRecords => match store.iter() {
325            Ok(it) => {
326                let records: Vec<Value> = it
327                    .map(|(entity, id, value)| {
328                        json!({"entity": entity, "id": id, "value": value})
329                    })
330                    .collect();
331                (json!({"ok": true, "records": records}), false)
332            }
333            Err(e) => (json!({"ok": false, "error": e.to_string()}), false),
334        },
335        Request::Shutdown => (json!({"ok": true, "shutdown": true}), true),
336    }
337}
338
339fn error_response(msg: &str) -> Value {
340    json!({"ok": false, "error": msg})
341}
342
343fn hex_encode(bytes: &[u8]) -> String {
344    const HEX: &[u8; 16] = b"0123456789abcdef";
345    let mut out = String::with_capacity(bytes.len() * 2);
346    for &b in bytes {
347        out.push(HEX[(b >> 4) as usize] as char);
348        out.push(HEX[(b & 0x0f) as usize] as char);
349    }
350    out
351}
352