Compare commits
3 Commits
loops/cont
...
loops/host
| Author | SHA1 | Date | |
|---|---|---|---|
| 95e981eb03 | |||
| c6c2cebf98 | |||
| 65f274c573 |
@@ -571,9 +571,12 @@ and cek_run_with_io state =
|
|||||||
Hashtbl.replace d "descent" (Number desc);
|
Hashtbl.replace d "descent" (Number desc);
|
||||||
Dict d
|
Dict d
|
||||||
| _ ->
|
| _ ->
|
||||||
let args = let a = Sx_runtime.get_val request (String "args") in
|
let argsv = Sx_runtime.get_val request (String "args") in
|
||||||
(match a with List l -> l | _ -> [a]) in
|
(match Sx_persist_store.handle_op op argsv with
|
||||||
io_request op args
|
| Some resp -> resp
|
||||||
|
| None ->
|
||||||
|
let args = (match argsv with List l -> l | _ -> [argsv]) in
|
||||||
|
io_request op args)
|
||||||
in
|
in
|
||||||
s := Sx_ref.cek_resume !s response;
|
s := Sx_ref.cek_resume !s response;
|
||||||
loop ()
|
loop ()
|
||||||
@@ -1540,7 +1543,12 @@ let rec dispatch env cmd =
|
|||||||
| Some path -> load_library_file path | None -> ());
|
| Some path -> load_library_file path | None -> ());
|
||||||
Nil
|
Nil
|
||||||
end
|
end
|
||||||
end else Nil (* non-import IO: resume with nil *) in
|
end else
|
||||||
|
(* durable-storage ops: service against on-disk store *)
|
||||||
|
let args = Sx_runtime.get_val request (String "args") in
|
||||||
|
(match Sx_persist_store.handle_op op args with
|
||||||
|
| Some resp -> resp
|
||||||
|
| None -> Nil (* non-import IO: resume with nil *)) in
|
||||||
s := Sx_ref.cek_resume !s response
|
s := Sx_ref.cek_resume !s response
|
||||||
done;
|
done;
|
||||||
Sx_ref.cek_value !s
|
Sx_ref.cek_value !s
|
||||||
@@ -3893,7 +3901,10 @@ let http_mode port =
|
|||||||
Dict d
|
Dict d
|
||||||
| "io-sleep" | "sleep" -> Nil
|
| "io-sleep" | "sleep" -> Nil
|
||||||
| "import" -> Nil
|
| "import" -> Nil
|
||||||
| _ -> Nil);
|
| _ ->
|
||||||
|
(match Sx_persist_store.handle_op op args with
|
||||||
|
| Some resp -> resp
|
||||||
|
| None -> Nil));
|
||||||
(* Response cache — path → full HTTP response string.
|
(* Response cache — path → full HTTP response string.
|
||||||
Populated during pre-warm, serves cached responses in <0.1ms.
|
Populated during pre-warm, serves cached responses in <0.1ms.
|
||||||
Thread-safe: reads are lock-free (Hashtbl.find_opt is atomic for
|
Thread-safe: reads are lock-free (Hashtbl.find_opt is atomic for
|
||||||
|
|||||||
293
hosts/ocaml/lib/sx_persist_store.ml
Normal file
293
hosts/ocaml/lib/sx_persist_store.ml
Normal file
@@ -0,0 +1,293 @@
|
|||||||
|
(* sx_persist_store — host durable-storage adapter for lib/persist.
|
||||||
|
Production twin of `persist/serve` (lib/persist/durable.sx): it answers the
|
||||||
|
same `persist/...` IO ops, but backs them with real on-disk storage so writes
|
||||||
|
survive a process restart. Stateless-on-disk: every op reads/writes the
|
||||||
|
filesystem directly, so a fresh process recovers state with no warm-up — the
|
||||||
|
log on disk IS the state.
|
||||||
|
|
||||||
|
On-disk layout under the root dir (default ./persist-data, or $SX_PERSIST_DIR):
|
||||||
|
streams/<hex(stream)>.log append-only, one SX-serialized event per line
|
||||||
|
streams/<hex(stream)>.seq per-stream monotonic high-water counter (int)
|
||||||
|
kv/<hex(key)> one SX-serialized value per key
|
||||||
|
|
||||||
|
Invariants honoured (see plans/persist-on-sx.md Blocker spec):
|
||||||
|
1. last-seq is a per-stream monotonic counter stored in .seq, SEPARATE from
|
||||||
|
the rows — it keeps climbing across truncate, so a compacted stream never
|
||||||
|
reassigns a seq.
|
||||||
|
2. append never renumbers — the event already carries its :seq (log.sx does
|
||||||
|
last-seq+1); the host only bumps the high-water mark to max(hw, seq).
|
||||||
|
3. read returns surviving events in append order with :seq intact.
|
||||||
|
4. streams is the set of streams that ever had an append — keyed off the .seq
|
||||||
|
file, which truncate never deletes, so it survives full compaction.
|
||||||
|
5. values round-trip structurally via the SX serializer/parser. *)
|
||||||
|
|
||||||
|
open Sx_types
|
||||||
|
|
||||||
|
(* ---- root dir ---------------------------------------------------------- *)
|
||||||
|
|
||||||
|
let _root : string option ref = ref None
|
||||||
|
|
||||||
|
let set_root dir = _root := Some dir
|
||||||
|
|
||||||
|
let root_dir () =
|
||||||
|
match !_root with
|
||||||
|
| Some d -> d
|
||||||
|
| None -> (try Sys.getenv "SX_PERSIST_DIR" with Not_found -> "persist-data")
|
||||||
|
|
||||||
|
(* ---- filesystem helpers ------------------------------------------------ *)
|
||||||
|
|
||||||
|
let rec ensure_dir dir =
|
||||||
|
if dir = "" || dir = "." || dir = "/" || Sys.file_exists dir then ()
|
||||||
|
else begin
|
||||||
|
ensure_dir (Filename.dirname dir);
|
||||||
|
(try Unix.mkdir dir 0o755 with Unix.Unix_error (Unix.EEXIST, _, _) -> ())
|
||||||
|
end
|
||||||
|
|
||||||
|
let streams_dir () = Filename.concat (root_dir ()) "streams"
|
||||||
|
let kv_dir () = Filename.concat (root_dir ()) "kv"
|
||||||
|
let blobs_dir () = Filename.concat (root_dir ()) "blobs"
|
||||||
|
|
||||||
|
let read_file path =
|
||||||
|
let ic = open_in_bin path in
|
||||||
|
let n = in_channel_length ic in
|
||||||
|
let s = really_input_string ic n in
|
||||||
|
close_in ic;
|
||||||
|
s
|
||||||
|
|
||||||
|
(* Atomic write: temp file in the same dir then rename over the target. *)
|
||||||
|
let write_file_atomic path contents =
|
||||||
|
ensure_dir (Filename.dirname path);
|
||||||
|
let tmp = path ^ ".tmp" in
|
||||||
|
let oc = open_out_bin tmp in
|
||||||
|
output_string oc contents;
|
||||||
|
flush oc;
|
||||||
|
close_out oc;
|
||||||
|
Sys.rename tmp path
|
||||||
|
|
||||||
|
let append_line path line =
|
||||||
|
ensure_dir (Filename.dirname path);
|
||||||
|
let oc = open_out_gen [Open_append; Open_creat; Open_wronly] 0o644 path in
|
||||||
|
output_string oc line;
|
||||||
|
output_char oc '\n';
|
||||||
|
close_out oc
|
||||||
|
|
||||||
|
(* ---- name <-> filename (hex, reversible, fs-safe) ---------------------- *)
|
||||||
|
|
||||||
|
let hex_encode s =
|
||||||
|
let b = Buffer.create (String.length s * 2) in
|
||||||
|
String.iter (fun c -> Buffer.add_string b (Printf.sprintf "%02x" (Char.code c))) s;
|
||||||
|
Buffer.contents b
|
||||||
|
|
||||||
|
let hex_decode s =
|
||||||
|
let n = String.length s / 2 in
|
||||||
|
String.init n (fun i -> Char.chr (int_of_string ("0x" ^ String.sub s (i * 2) 2)))
|
||||||
|
|
||||||
|
let stream_log stream = Filename.concat (streams_dir ()) (hex_encode stream ^ ".log")
|
||||||
|
let stream_seq stream = Filename.concat (streams_dir ()) (hex_encode stream ^ ".seq")
|
||||||
|
let kv_path key = Filename.concat (kv_dir ()) (hex_encode key)
|
||||||
|
|
||||||
|
(* ---- value <-> SX text (round-trips through Sx_parser) ----------------- *)
|
||||||
|
|
||||||
|
let escape_str s =
|
||||||
|
let len = String.length s in
|
||||||
|
let buf = Buffer.create (len + 16) in
|
||||||
|
for i = 0 to len - 1 do
|
||||||
|
match s.[i] with
|
||||||
|
| '"' -> Buffer.add_string buf "\\\""
|
||||||
|
| '\\' -> Buffer.add_string buf "\\\\"
|
||||||
|
| '\n' -> Buffer.add_string buf "\\n"
|
||||||
|
| '\r' -> Buffer.add_string buf "\\r"
|
||||||
|
| '\t' -> Buffer.add_string buf "\\t"
|
||||||
|
| c -> Buffer.add_char buf c
|
||||||
|
done;
|
||||||
|
Buffer.contents buf
|
||||||
|
|
||||||
|
let rec serialize = function
|
||||||
|
| Nil -> "nil"
|
||||||
|
| Bool true -> "true"
|
||||||
|
| Bool false -> "false"
|
||||||
|
| Integer n -> string_of_int n
|
||||||
|
| Number n -> format_number n
|
||||||
|
| String s -> "\"" ^ escape_str s ^ "\""
|
||||||
|
| Symbol s -> "(quote " ^ s ^ ")"
|
||||||
|
| Keyword k -> ":" ^ k
|
||||||
|
| List items | ListRef { contents = items } ->
|
||||||
|
"(list" ^ (List.fold_left (fun acc v -> acc ^ " " ^ serialize v) "" items) ^ ")"
|
||||||
|
| Dict d ->
|
||||||
|
let pairs = Hashtbl.fold (fun k v acc ->
|
||||||
|
(Printf.sprintf ":%s %s" k (serialize v)) :: acc) d [] in
|
||||||
|
"{" ^ String.concat " " (List.sort String.compare pairs) ^ "}"
|
||||||
|
| _ -> "nil"
|
||||||
|
|
||||||
|
(* Parse one serialized value back. Empty / blank -> Nil. *)
|
||||||
|
let rec deserialize line =
|
||||||
|
let line = String.trim line in
|
||||||
|
if line = "" then Nil
|
||||||
|
else match Sx_parser.parse_all line with
|
||||||
|
| v :: _ -> eval_quote_lists v
|
||||||
|
| [] -> Nil
|
||||||
|
|
||||||
|
(* serialize emits lists as `(list ...)` and symbols as `(quote s)` so the
|
||||||
|
parser yields data, not a call — but the parser leaves those as AST. Walk
|
||||||
|
the parsed AST and collapse `(list ...)`/`(quote s)` back to values. *)
|
||||||
|
and eval_quote_lists v =
|
||||||
|
match v with
|
||||||
|
| List (Symbol "quote" :: x :: []) -> x
|
||||||
|
| List (Symbol "list" :: rest) -> List (List.map eval_quote_lists rest)
|
||||||
|
| List items -> List (List.map eval_quote_lists items)
|
||||||
|
| ListRef { contents = items } -> List (List.map eval_quote_lists items)
|
||||||
|
| Dict d ->
|
||||||
|
let d' = Hashtbl.create (Hashtbl.length d) in
|
||||||
|
Hashtbl.iter (fun k v -> Hashtbl.replace d' k (eval_quote_lists v)) d;
|
||||||
|
Dict d'
|
||||||
|
| other -> other
|
||||||
|
|
||||||
|
(* ---- seq counter ------------------------------------------------------- *)
|
||||||
|
|
||||||
|
let read_seq stream =
|
||||||
|
let p = stream_seq stream in
|
||||||
|
if Sys.file_exists p then (try int_of_string (String.trim (read_file p)) with _ -> 0)
|
||||||
|
else 0
|
||||||
|
|
||||||
|
let write_seq stream n = write_file_atomic (stream_seq stream) (string_of_int n)
|
||||||
|
|
||||||
|
let value_to_int = function
|
||||||
|
| Integer n -> n
|
||||||
|
| Number n -> int_of_float n
|
||||||
|
| _ -> 0
|
||||||
|
|
||||||
|
let event_seq ev =
|
||||||
|
match ev with
|
||||||
|
| Dict d -> (match Hashtbl.find_opt d "seq" with Some v -> value_to_int v | None -> 0)
|
||||||
|
| _ -> 0
|
||||||
|
|
||||||
|
(* ---- ops --------------------------------------------------------------- *)
|
||||||
|
|
||||||
|
let do_append stream ev =
|
||||||
|
ensure_dir (streams_dir ());
|
||||||
|
(* bump the monotonic high-water mark; create .seq on first append so the
|
||||||
|
stream shows up in `streams` and survives later truncation. *)
|
||||||
|
let hw = read_seq stream in
|
||||||
|
let s = event_seq ev in
|
||||||
|
write_seq stream (max hw s);
|
||||||
|
append_line (stream_log stream) (serialize ev)
|
||||||
|
|
||||||
|
let do_read stream =
|
||||||
|
let p = stream_log stream in
|
||||||
|
if not (Sys.file_exists p) then List []
|
||||||
|
else begin
|
||||||
|
let content = read_file p in
|
||||||
|
let lines = String.split_on_char '\n' content in
|
||||||
|
let evs = List.filter_map (fun l ->
|
||||||
|
if String.trim l = "" then None else Some (deserialize l)) lines in
|
||||||
|
List evs
|
||||||
|
end
|
||||||
|
|
||||||
|
let do_last_seq stream = Number (float_of_int (read_seq stream))
|
||||||
|
|
||||||
|
let list_dir_suffix dir suffix =
|
||||||
|
if not (Sys.file_exists dir) then []
|
||||||
|
else
|
||||||
|
Array.to_list (Sys.readdir dir)
|
||||||
|
|> List.filter (fun f -> Filename.check_suffix f suffix)
|
||||||
|
|> List.map (fun f -> hex_decode (Filename.chop_suffix f suffix))
|
||||||
|
|> List.sort String.compare
|
||||||
|
|
||||||
|
let do_streams () = List (List.map (fun s -> String s) (list_dir_suffix (streams_dir ()) ".seq"))
|
||||||
|
|
||||||
|
(* drop events with seq <= n; the .seq high-water counter is untouched. *)
|
||||||
|
let do_truncate stream n =
|
||||||
|
let p = stream_log stream in
|
||||||
|
if Sys.file_exists p then begin
|
||||||
|
let evs = match do_read stream with List l -> l | _ -> [] in
|
||||||
|
let kept = List.filter (fun ev -> event_seq ev > n) evs in
|
||||||
|
let body = String.concat "" (List.map (fun ev -> serialize ev ^ "\n") kept) in
|
||||||
|
write_file_atomic p body
|
||||||
|
end
|
||||||
|
|
||||||
|
let do_kv_get key =
|
||||||
|
let p = kv_path key in
|
||||||
|
if Sys.file_exists p then deserialize (read_file p) else Nil
|
||||||
|
|
||||||
|
let do_kv_put key v =
|
||||||
|
ensure_dir (kv_dir ());
|
||||||
|
write_file_atomic (kv_path key) (serialize v)
|
||||||
|
|
||||||
|
let do_kv_delete key =
|
||||||
|
let p = kv_path key in
|
||||||
|
if Sys.file_exists p then (try Sys.remove p with _ -> ())
|
||||||
|
|
||||||
|
let do_kv_has key = Bool (Sys.file_exists (kv_path key))
|
||||||
|
|
||||||
|
let do_kv_keys () =
|
||||||
|
if not (Sys.file_exists (kv_dir ())) then List []
|
||||||
|
else
|
||||||
|
List (
|
||||||
|
Array.to_list (Sys.readdir (kv_dir ()))
|
||||||
|
|> List.map hex_decode
|
||||||
|
|> List.sort String.compare
|
||||||
|
|> List.map (fun s -> String s))
|
||||||
|
|
||||||
|
(* ---- blob store (content-addressed) ------------------------------------ *)
|
||||||
|
(* Same pattern as the persist ops, but a SEPARATE adapter: large objects live
|
||||||
|
in a content-addressed directory keyed by a CIDv1 (raw codec, sha2-256).
|
||||||
|
persist only ever stores the returned ref ({:cid :size :mime}), never bytes.
|
||||||
|
blob/put is idempotent — identical bytes hash to the same cid + same file. *)
|
||||||
|
|
||||||
|
let codec_raw = 0x55
|
||||||
|
|
||||||
|
let blob_cid bytes =
|
||||||
|
let digest = Sx_cid.unhex (Sx_sha2.sha256_hex bytes) in
|
||||||
|
Sx_cid.cidv1 codec_raw (Sx_cid.multihash Sx_cid.mh_sha2_256 digest)
|
||||||
|
|
||||||
|
let blob_path cid = Filename.concat (blobs_dir ()) cid
|
||||||
|
|
||||||
|
let do_blob_put bytes =
|
||||||
|
let cid = blob_cid bytes in
|
||||||
|
let p = blob_path cid in
|
||||||
|
if not (Sys.file_exists p) then write_file_atomic p bytes;
|
||||||
|
String cid
|
||||||
|
|
||||||
|
let do_blob_get cid =
|
||||||
|
let p = blob_path cid in
|
||||||
|
if Sys.file_exists p then String (read_file p) else Nil
|
||||||
|
|
||||||
|
let do_blob_has cid = Bool (Sys.file_exists (blob_path cid))
|
||||||
|
|
||||||
|
(* ---- dispatch ---------------------------------------------------------- *)
|
||||||
|
|
||||||
|
let arglist = function
|
||||||
|
| List l | ListRef { contents = l } -> l
|
||||||
|
| Nil -> []
|
||||||
|
| v -> [v]
|
||||||
|
|
||||||
|
(* Returns Some response if op is a persist op this store owns, None otherwise. *)
|
||||||
|
let handle_op op args =
|
||||||
|
let a = arglist args in
|
||||||
|
let str = function String s -> s | v -> value_to_string v in
|
||||||
|
match op with
|
||||||
|
| "persist/append" ->
|
||||||
|
(match a with stream :: ev :: _ -> do_append (str stream) ev | _ -> ()); Some Nil
|
||||||
|
| "persist/read" ->
|
||||||
|
(match a with stream :: _ -> Some (do_read (str stream)) | _ -> Some (List []))
|
||||||
|
| "persist/last-seq" ->
|
||||||
|
(match a with stream :: _ -> Some (do_last_seq (str stream)) | _ -> Some (Number 0.0))
|
||||||
|
| "persist/streams" -> Some (do_streams ())
|
||||||
|
| "persist/truncate" ->
|
||||||
|
(match a with stream :: n :: _ -> do_truncate (str stream) (value_to_int n) | _ -> ()); Some Nil
|
||||||
|
| "persist/kv-get" ->
|
||||||
|
(match a with key :: _ -> Some (do_kv_get (str key)) | _ -> Some Nil)
|
||||||
|
| "persist/kv-put" ->
|
||||||
|
(match a with key :: v :: _ -> do_kv_put (str key) v | _ -> ()); Some Nil
|
||||||
|
| "persist/kv-delete" ->
|
||||||
|
(match a with key :: _ -> do_kv_delete (str key) | _ -> ()); Some Nil
|
||||||
|
| "persist/kv-has?" ->
|
||||||
|
(match a with key :: _ -> Some (do_kv_has (str key)) | _ -> Some (Bool false))
|
||||||
|
| "persist/kv-keys" -> Some (do_kv_keys ())
|
||||||
|
| "blob/put" ->
|
||||||
|
(match a with bytes :: _ -> Some (do_blob_put (str bytes)) | _ -> Some Nil)
|
||||||
|
| "blob/get" ->
|
||||||
|
(match a with cid :: _ -> Some (do_blob_get (str cid)) | _ -> Some Nil)
|
||||||
|
| "blob/has?" ->
|
||||||
|
(match a with cid :: _ -> Some (do_blob_has (str cid)) | _ -> Some (Bool false))
|
||||||
|
| _ -> None
|
||||||
144
hosts/ocaml/test/persist_durable_test.sh
Executable file
144
hosts/ocaml/test/persist_durable_test.sh
Executable file
@@ -0,0 +1,144 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# hosts/ocaml/test/persist_durable_test.sh
|
||||||
|
# Acceptance test for the host durable-storage adapter (Sx_persist_store).
|
||||||
|
#
|
||||||
|
# Exercises `persist/durable-backend` (REAL `perform`, not the mock) under the
|
||||||
|
# WORKTREE-built sx_server.exe, and asserts:
|
||||||
|
# 1. durable: writes land on disk and read back (the silent-data-loss repro
|
||||||
|
# from plans/persist-on-sx.md now returns correct values).
|
||||||
|
# 2. last-seq is monotonic across truncate (compaction never reassigns a seq).
|
||||||
|
# 3. kv ops round-trip and delete.
|
||||||
|
# 4. recovery: a REAL process restart (write, exit, fresh process, replay)
|
||||||
|
# recovers state from disk.
|
||||||
|
#
|
||||||
|
# Run from repo root or anywhere; locates the worktree binary relative to itself.
|
||||||
|
set -uo pipefail
|
||||||
|
|
||||||
|
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||||
|
ROOT="$(cd "$HERE/../../.." && pwd)" # repo/worktree root
|
||||||
|
cd "$ROOT"
|
||||||
|
|
||||||
|
SX="hosts/ocaml/_build/default/bin/sx_server.exe"
|
||||||
|
if [ ! -x "$SX" ]; then
|
||||||
|
echo "ERROR: worktree binary not found at $SX — build it first:" >&2
|
||||||
|
echo " (cd hosts/ocaml && dune build bin/sx_server.exe)" >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
|
||||||
|
DATADIR="$(mktemp -d)"
|
||||||
|
trap 'rm -rf "$DATADIR"' EXIT
|
||||||
|
|
||||||
|
PASS=0
|
||||||
|
FAIL=0
|
||||||
|
check() { # check <label> <got> <expected>
|
||||||
|
if [ "$2" = "$3" ]; then
|
||||||
|
PASS=$((PASS + 1)); printf ' ok %-40s => %s\n' "$1" "$2"
|
||||||
|
else
|
||||||
|
FAIL=$((FAIL + 1)); printf ' FAIL %-40s got [%s] want [%s]\n' "$1" "$2" "$3"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
PRELUDE='(epoch 1)
|
||||||
|
(load "spec/stdlib.sx")
|
||||||
|
(load "lib/r7rs.sx")
|
||||||
|
(load "lib/persist/event.sx")
|
||||||
|
(load "lib/persist/backend.sx")
|
||||||
|
(load "lib/persist/log.sx")
|
||||||
|
(load "lib/persist/kv.sx")
|
||||||
|
(load "lib/persist/durable.sx")
|
||||||
|
(load "lib/persist/blob.sx")
|
||||||
|
(epoch 2)'
|
||||||
|
|
||||||
|
# run_eval <sx-expr-string>: prints the final (ok-len 2 ...) payload line.
|
||||||
|
run_eval() {
|
||||||
|
local expr="$1"
|
||||||
|
printf '%s\n(eval %s)\n' "$PRELUDE" "$expr" \
|
||||||
|
| SX_PERSIST_DIR="$DATADIR" timeout 60 "$SX" 2>/dev/null \
|
||||||
|
| awk '/^\(ok-len 2 / {getline; print; exit}'
|
||||||
|
}
|
||||||
|
|
||||||
|
# escape an SX program into a single-line double-quoted SX string literal for
|
||||||
|
# (eval "..."). The REPL reads one command per physical line, so newlines in the
|
||||||
|
# program are collapsed to spaces.
|
||||||
|
q() { printf '"%s"' "$(printf '%s' "$1" | tr '\n' ' ' | sed 's/\\/\\\\/g; s/"/\\"/g')"; }
|
||||||
|
|
||||||
|
echo "== durable: append/read/last-seq round-trip on disk =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(begin
|
||||||
|
(persist/append b "s" "x" 0 {:v 1})
|
||||||
|
(persist/append b "s" "x" 0 {:v 2})
|
||||||
|
(list (persist/event-seq (persist/append b "s" "x" 0 {:v 3}))
|
||||||
|
(persist/count b "s")
|
||||||
|
(len (persist/read b "s")))))')")
|
||||||
|
check "append/count/read" "$GOT" "(3 3 3)"
|
||||||
|
|
||||||
|
echo "== last-seq monotonic across truncate =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(begin
|
||||||
|
(persist/append b "t" "x" 0 {})
|
||||||
|
(persist/append b "t" "x" 0 {})
|
||||||
|
(persist/append b "t" "x" 0 {})
|
||||||
|
(persist/truncate b "t" 2)
|
||||||
|
(list (persist/last-seq b "t") (persist/count b "t"))))')")
|
||||||
|
check "last-seq survives truncate" "$GOT" "(3 1)"
|
||||||
|
|
||||||
|
echo "== streams set survives compaction =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(sort ((get b "streams"))))')")
|
||||||
|
check "streams" "$GOT" '("s" "t")'
|
||||||
|
|
||||||
|
echo "== kv round-trip + delete =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(begin
|
||||||
|
(persist/kv-put b "k" {:a 1 :b "two"})
|
||||||
|
(persist/kv-put b "gone" 9)
|
||||||
|
(persist/kv-delete b "gone")
|
||||||
|
(list (get (persist/kv-get b "k") :b)
|
||||||
|
(persist/kv-has? b "k")
|
||||||
|
(persist/kv-has? b "gone"))))')")
|
||||||
|
check "kv get/has/delete" "$GOT" '("two" true false)'
|
||||||
|
|
||||||
|
echo "== recovery: state survives a REAL process restart =="
|
||||||
|
# write in process A then let it exit; the next run is a brand-new process.
|
||||||
|
run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(begin
|
||||||
|
(persist/append b "r" "ev" 0 {:n 1})
|
||||||
|
(persist/append b "r" "ev" 0 {:n 2})
|
||||||
|
(persist/kv-put b "survive" "yes")
|
||||||
|
(persist/count b "r")))')" >/dev/null
|
||||||
|
# fresh process, same SX_PERSIST_DIR — must replay from disk.
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
|
||||||
|
(list (persist/count b "r")
|
||||||
|
(persist/last-seq b "r")
|
||||||
|
(get (get (nth (persist/read b "r") 1) :data) :n)
|
||||||
|
(persist/kv-get b "survive")))')")
|
||||||
|
check "recovered after restart" "$GOT" '(2 2 2 "yes")'
|
||||||
|
|
||||||
|
echo "== blob: content-addressed put/get/has? round-trip =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
|
||||||
|
(let ((r (persist/blob-store bs "hello world" "text/plain")))
|
||||||
|
(list (persist/blob-size r)
|
||||||
|
(persist/blob-mime r)
|
||||||
|
(persist/blob-fetch bs r)
|
||||||
|
(persist/blob-exists? bs r))))')")
|
||||||
|
check "blob size/mime/fetch/exists" "$GOT" '(11 "text/plain" "hello world" true)'
|
||||||
|
|
||||||
|
echo "== blob: put is content-addressed (idempotent cid) =="
|
||||||
|
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
|
||||||
|
(equal? (persist/blob-cid (persist/blob-store bs "same bytes" "x"))
|
||||||
|
(persist/blob-cid (persist/blob-store bs "same bytes" "x"))))')")
|
||||||
|
check "same bytes -> same cid" "$GOT" "true"
|
||||||
|
|
||||||
|
echo "== blob: bytes + ref-in-kv survive a REAL restart =="
|
||||||
|
# process A: store a blob, keep only its ref in the durable kv.
|
||||||
|
run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
|
||||||
|
(begin (persist/kv-put b "logo" (persist/blob-store bs "PNGDATA" "image/png")) nil))')" >/dev/null
|
||||||
|
# fresh process: read the ref from kv, fetch the bytes from the blob store.
|
||||||
|
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
|
||||||
|
(let ((r (persist/kv-get b "logo")))
|
||||||
|
(list (persist/blob-fetch bs r) (persist/blob-exists? bs r) (persist/blob-mime r))))')")
|
||||||
|
check "blob recovered via ref after restart" "$GOT" '("PNGDATA" true "image/png")'
|
||||||
|
|
||||||
|
echo
|
||||||
|
echo "durable adapter: $PASS passed, $FAIL failed"
|
||||||
|
[ "$FAIL" -eq 0 ]
|
||||||
94
plans/agent-briefings/host-persist-loop.md
Normal file
94
plans/agent-briefings/host-persist-loop.md
Normal file
@@ -0,0 +1,94 @@
|
|||||||
|
# host-persist loop agent (single agent, builds the durable storage host)
|
||||||
|
|
||||||
|
Role: make `lib/persist`'s durable backend **actually durable**. The persist
|
||||||
|
substrate (`lib/persist/**`, 201/201 tests) performs `{:op "persist/..." :args}`
|
||||||
|
IO requests for every storage op; under `sx_server.exe` today nothing services
|
||||||
|
them, so writes silently vanish. You build the **host-side adapter** that answers
|
||||||
|
those ops against real on-disk storage — the one piece standing between persist
|
||||||
|
and "all subsystems share a durable substrate."
|
||||||
|
|
||||||
|
```
|
||||||
|
worktree: /root/rose-ash-loops/host-persist
|
||||||
|
branch: loops/host-persist (push origin/loops/host-persist; NEVER main/architecture)
|
||||||
|
```
|
||||||
|
|
||||||
|
## The authoritative contract — read this first, every restart
|
||||||
|
|
||||||
|
`plans/persist-on-sx.md` → **Blockers → "OPEN — host durable-storage adapter"**.
|
||||||
|
That entry is the spec: the silent-data-loss repro, the full op contract table,
|
||||||
|
the hard invariants (monotonic `last-seq`, etc.), the blob adapter shape, where
|
||||||
|
to register in `sx_server.ml`, and the acceptance test. Do not restate it here —
|
||||||
|
read it there and implement it. The reference implementation to mirror is
|
||||||
|
`persist/serve` in `lib/persist/durable.sx` (same op names, same shapes).
|
||||||
|
|
||||||
|
## Restart baseline — check before iterating
|
||||||
|
|
||||||
|
1. Read the Blocker spec (above) + this briefing.
|
||||||
|
2. `git log --oneline -8` on `loops/host-persist` to see what's done.
|
||||||
|
3. Is there a worktree-local build? `ls hosts/ocaml/_build/default/bin/sx_server.exe`.
|
||||||
|
Fresh worktrees have none — the first build is the first task.
|
||||||
|
4. If an acceptance suite exists (e.g. `hosts/ocaml/test/persist_durable_*` or a
|
||||||
|
`lib/persist/tests/durable-real.sx`), run it against the **worktree-built**
|
||||||
|
binary. Green before new work.
|
||||||
|
|
||||||
|
## The queue (phases)
|
||||||
|
|
||||||
|
- **Phase 0 — reproduce.** Confirm the silent-data-loss repro from the spec under
|
||||||
|
this worktree. Builds your mental model; costs one short run.
|
||||||
|
- **Phase 1 — storage module.** A new OCaml module under `hosts/ocaml/` that
|
||||||
|
implements the op contract over **real persistent storage**. Start simple and
|
||||||
|
correct: a filesystem-backed store (one append-only file per stream + a kv
|
||||||
|
file + a per-stream seq high-water file), or SQLite if the toolchain has it.
|
||||||
|
Honour every invariant in the spec — especially: `last-seq` is a monotonic
|
||||||
|
counter stored separately from rows so it survives `truncate`; values
|
||||||
|
round-trip structurally.
|
||||||
|
- **Phase 2 — register.** Wire a `"persist/..."` arm into the kernel's IO
|
||||||
|
resolver (`Sx_types._cek_io_resolver`, ~line 3864 of `hosts/ocaml/bin/sx_server.ml`)
|
||||||
|
and/or the `cek_run_with_io` bridge path (~528–576), dispatching to the storage
|
||||||
|
module. Op names are the contract — do not rename.
|
||||||
|
- **Phase 3 — acceptance.** New tests that use `persist/durable-backend` (REAL
|
||||||
|
`perform`, not the mock) run under the freshly-built worktree binary: the
|
||||||
|
`durable` + `recovery` semantics must pass, and a **real process restart**
|
||||||
|
(start the built server, write, stop it, start again, replay) must recover
|
||||||
|
state from disk. Put host-owned tests under a host path (e.g.
|
||||||
|
`hosts/ocaml/test/`) — do not churn persist's existing suites.
|
||||||
|
- **Phase 4 — blob adapter.** Same pattern for `blob/put|get|has?` backed by a
|
||||||
|
content-addressed directory; persist stores only the ref.
|
||||||
|
|
||||||
|
Every iteration: implement → build → test → commit (short factual message) →
|
||||||
|
push → update `plans/persist-on-sx.md` (tick the Blocker toward CLOSED, append a
|
||||||
|
dated Progress-log line, newest first) → next.
|
||||||
|
|
||||||
|
## Ground rules (hard)
|
||||||
|
|
||||||
|
- **Build is your job** (unlike the persist loop). But build **only in this
|
||||||
|
worktree's `_build`** via `dune` from `/root/rose-ash-loops/host-persist`.
|
||||||
|
**NEVER overwrite the shared binary** at
|
||||||
|
`/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe` — every sibling
|
||||||
|
loop uses it; clobbering it breaks them all. Point acceptance tests at the
|
||||||
|
worktree binary (`hosts/ocaml/_build/default/bin/sx_server.exe` *inside this
|
||||||
|
worktree*).
|
||||||
|
- **First build is slow** (full OCaml). The `sx_build` MCP tool has a ~600s
|
||||||
|
watchdog that may kill it — prefer `dune build bin/sx_server.exe` (or `@all`)
|
||||||
|
run via Bash with `run_in_background: true` and a long timeout, then poll.
|
||||||
|
- **NEVER `pkill sx_server`** — siblings share the process/binary. Start your own
|
||||||
|
server on a throwaway path/port for restart tests and stop only that PID; bound
|
||||||
|
every run with `timeout`.
|
||||||
|
- **Scope:** `hosts/**`, host-owned test files, and the Blocker entry +
|
||||||
|
Progress log in `plans/persist-on-sx.md`. Do **not** modify `lib/persist/**`
|
||||||
|
source (the persist loop owns it; its API is your contract, not your code) —
|
||||||
|
if you need an upstream change, leave a note in the Blocker entry.
|
||||||
|
- **Determinism:** replay from disk must equal the in-memory semantics; same log
|
||||||
|
→ same state.
|
||||||
|
- **Commits:** one feature per commit; push to `origin/loops/host-persist`.
|
||||||
|
- **SX files:** `sx-tree` MCP tools ONLY, `file:` not `path:`, `sx_validate`
|
||||||
|
after edits. (Most of your work is OCaml — edit those with normal tools.)
|
||||||
|
|
||||||
|
## Definition of done
|
||||||
|
|
||||||
|
The Blocker entry flips to **CLOSED**: `persist/durable-backend` writes land on
|
||||||
|
disk, survive a real server restart, and the durable + recovery acceptance suites
|
||||||
|
are green against the worktree-built binary. At that point a subsystem migrated
|
||||||
|
per `lib/persist/examples/acl.sx` is genuinely durable.
|
||||||
|
|
||||||
|
Go. Read the Blocker spec; reproduce the gap; build the storage module.
|
||||||
@@ -197,6 +197,30 @@ durable backend. Other subsystem loops copy this pattern; it does not touch the
|
|||||||
real `lib/acl`.
|
real `lib/acl`.
|
||||||
|
|
||||||
## Progress log
|
## Progress log
|
||||||
|
- **Host durable-storage adapter — blob adapter LIVE, Blocker CLOSED (8/8).**
|
||||||
|
Added content-addressed `blob/put|get|has?` to `sx_persist_store.ml`: bytes
|
||||||
|
land in `<root>/blobs/<cid>` keyed by a CIDv1 (raw codec, sha2-256 via
|
||||||
|
`Sx_cid`/`Sx_sha2`); `put` is idempotent (identical bytes → same cid → same
|
||||||
|
file); persist stores only the `{:cid :size :mime}` ref, never the bytes. The
|
||||||
|
`(eval ...)`/bridge/resolver fall-through already routes any unowned op to the
|
||||||
|
store, so no new wiring was needed. `persist_durable_test.sh` extended:
|
||||||
|
blob round-trip, content-address idempotency, and bytes+ref-in-kv surviving a
|
||||||
|
real process restart — all green. Existing mock blob suite 14/0 against the
|
||||||
|
worktree binary. The host durable-storage Blocker is now CLOSED.
|
||||||
|
- **Host durable-storage adapter — durable+recovery LIVE (5/5 acceptance).**
|
||||||
|
`hosts/ocaml/lib/sx_persist_store.ml` services every `persist/*` IO op against
|
||||||
|
real on-disk storage (append-only log + separate monotonic `.seq` high-water
|
||||||
|
per stream + per-key kv files; SX-serialized, structurally round-tripping).
|
||||||
|
Wired into all three kernel IO sites in `hosts/ocaml/bin/sx_server.ml`: the
|
||||||
|
`(eval ...)` suspension loop (was the silent-no-op), the `cek_run_with_io`
|
||||||
|
bridge, and the in-process `_cek_io_resolver`. The spec's data-loss repro now
|
||||||
|
returns `(3 3 3)` instead of `(1 0 nil)`. `last-seq` climbs across `truncate`;
|
||||||
|
`streams` survives compaction. New `hosts/ocaml/test/persist_durable_test.sh`
|
||||||
|
drives `persist/durable-backend` (REAL `perform`) under the worktree binary —
|
||||||
|
durable, monotonic-seq, streams, kv, and a **real process restart** (write,
|
||||||
|
exit, fresh process, replay from `$SX_PERSIST_DIR`) all green. Existing mock
|
||||||
|
suites unaffected (durable 15/0, recovery 6/0 against worktree binary).
|
||||||
|
Remaining: blob adapter (`blob/put|get|has?`).
|
||||||
- **Reference migration: acl grants (201/201).** `lib/persist/examples/acl.sx` —
|
- **Reference migration: acl grants (201/201).** `lib/persist/examples/acl.sx` —
|
||||||
a worked, in-scope template migrating an ACL-grants store from a hand-rolled
|
a worked, in-scope template migrating an ACL-grants store from a hand-rolled
|
||||||
ephemeral map to persist: grants/revokes as events, current set as a
|
ephemeral map to persist: grants/revokes as events, current set as a
|
||||||
@@ -309,11 +333,26 @@ real `lib/acl`.
|
|||||||
|
|
||||||
## Blockers
|
## Blockers
|
||||||
|
|
||||||
### OPEN — host durable-storage adapter (the only gap to real durability)
|
### CLOSED — host durable-storage adapter (real durability is live)
|
||||||
|
|
||||||
**Owner:** a `hosts/` loop (NOT this one — `lib/persist/**` is the scope fence,
|
**Owner:** the `loops/host-persist` loop (`hosts/**` scope; `lib/persist/**` is
|
||||||
and `sx_build` is forbidden here). **Without it, durable persistence silently
|
its contract, not its code).
|
||||||
drops all writes.**
|
|
||||||
|
**Resolution (2026-06-06):** `hosts/ocaml/lib/sx_persist_store.ml` services every
|
||||||
|
`persist/*` AND `blob/*` IO op against real on-disk storage, wired into all three
|
||||||
|
IO sites of `hosts/ocaml/bin/sx_server.ml` (the `(eval ...)` suspension loop, the
|
||||||
|
`cek_run_with_io` bridge, and the in-process `_cek_io_resolver`). The
|
||||||
|
silent-data-loss repro below now returns the correct values instead of
|
||||||
|
`(1 0 nil)`. Acceptance + real-process-restart recovery green
|
||||||
|
(`hosts/ocaml/test/persist_durable_test.sh`, 8/8): durable round-trip,
|
||||||
|
monotonic-seq-across-truncate, streams-survive-compaction, kv, blob
|
||||||
|
content-addressing/idempotency, and bytes+refs surviving an actual process
|
||||||
|
restart. Existing mock suites unaffected (blob 14/0, durable 10/0, recovery,
|
||||||
|
against the worktree binary). A subsystem migrated per
|
||||||
|
`lib/persist/examples/acl.sx` is now genuinely durable. Original spec retained
|
||||||
|
below for reference.
|
||||||
|
|
||||||
|
**Was:** without it, durable persistence silently dropped all writes.
|
||||||
|
|
||||||
**Symptom / minimal repro.** `persist/durable-backend` performs
|
**Symptom / minimal repro.** `persist/durable-backend` performs
|
||||||
`{:op "persist/..." :args (...)}` for every storage op. Under `sx_server.exe`
|
`{:op "persist/..." :args (...)}` for every storage op. Under `sx_server.exe`
|
||||||
|
|||||||
Reference in New Issue
Block a user