From 98c4f63366f128c127fd671237a90a3c21c951fb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 08:47:36 +0000 Subject: [PATCH 1/7] unii: implement Phase 2 durable storage and replay Co-authored-by: pyrex41 --- unii/docs/HANDOFF.md | 39 ++-- unii/docs/STORAGE_PHASE2.md | 50 +++++ unii/docs/contracts/storage.md | 83 ++++--- unii/host/posix.lua | 100 ++++++++- unii/host/storage.lua | 349 ++++++++++++++++++++++++++---- unii/host/storage_blobs.lua | 57 +++++ unii/host/storage_checkpoint.lua | 93 ++++++++ unii/host/storage_index.lua | 54 +++++ unii/host/supervisor.lua | 103 ++++++++- unii/manifest.lua | 2 +- unii/test/run.lua | 3 +- unii/test/test_storage.lua | 25 ++- unii/test/test_storage_faults.lua | 129 +++++++++++ unii/test/test_storage_phase2.lua | 166 ++++++++++++++ 14 files changed, 1150 insertions(+), 103 deletions(-) create mode 100644 unii/docs/STORAGE_PHASE2.md create mode 100644 unii/host/storage_blobs.lua create mode 100644 unii/host/storage_checkpoint.lua create mode 100644 unii/host/storage_index.lua create mode 100644 unii/test/test_storage_faults.lua create mode 100644 unii/test/test_storage_phase2.lua diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index 3943e96..45ddddc 100644 --- a/unii/docs/HANDOFF.md +++ b/unii/docs/HANDOFF.md @@ -1,4 +1,4 @@ -# unii handoff: Phase 0 and Phase 1 +# unii handoff: Phase 0 through Phase 2 unii is a provisional name. It is the memory engine for an endless conversation agent, built from the plan in this repository's task, *Shen and @@ -57,7 +57,10 @@ unii/ core.lua boots shen-lua, checks the build stamp, calls the typed core supervisor.lua single writer: validate, transition, journal and fsync, adopt; replay storage.lua framed journal adapter (interface in docs/contracts/storage.md) - posix.lua FFI open/write-all/fsync/flock/ftruncate/rename/atomic write + posix.lua FFI durability operations and fault boundaries + storage_blobs.lua content-addressed message/summary bodies + storage_checkpoint.lua durable checkpoint framing and fallback + storage_index.lua rebuildable journal/message/blob/node/job indexes sha256.lua pure-Lua SHA-256 models.lua provider interface, failure classification mock/network.lua MOCK transport (contract in docs/contracts/network.md) @@ -153,6 +156,18 @@ same property at the API level, plus crash tails at every byte offset, corruption refusal, lock contention across processes, tampered transactions, foreign bundles and configuration changes. +### Phase 2 + +`test_storage_phase2` proves that journals hold blob hashes rather than +message/summary bodies, sequence order crosses daily shards, indexes rebuild, +valid checkpoints skip earlier records with identical state/view hashes, +stale checkpoints fall back, bad blobs refuse without journal mutation, and +recovery does not read whole journal files into RAM. + +`test_storage_faults` injects process death at every observed write, sync and +rename boundary, and separately covers commit ordering, torn writes and +ENOSPC. See `docs/STORAGE_PHASE2.md` for the outcome matrix. + ## Decisions worth knowing * **Leaf cap.** The 512-byte cap applies to `kind ": " text`. A user @@ -209,11 +224,9 @@ been reported upstream from this branch. raw-context recovery mode, policy epoch change, message chunking (events longer than `chunk_max` are rejected), compaction context, and a search fallback. -* **Storage.** One journal file holds everything. It is read entirely into - RAM, replay starts from record 1, and message text is held in RAM. There - are no blobs, checkpoints, indexes, day shards or backup/restore. Fault - injection covers only a failed append. On macOS, `fsync` has weaker - guarantees, and macOS has not been tested. +* **Storage limits.** Backup/restore and blob garbage collection do not + exist. Index metadata is proportional to history. Linux x86_64 is tested; + macOS uses `F_FULLFSYNC` for regular files but is not exercised in CI. * **Complexity.** View, live-node and job collections are Shen lists, so each transition is O(view + live + jobs). That is a few hundred lines at the default thresholds. The one-million-message gate in Phase 7 has not @@ -226,14 +239,10 @@ been reported upstream from this branch. ## Suggested next steps -1. Phase 2 storage: content-addressed blobs for message text, checkpoints - (view, state, journal position) so replay does not start at record 1, - rebuildable indexes, and fault injection around every write, fsync and - rename. -2. Plug in the real network adapter and one real summarizer provider +1. Plug in the real network adapter and one real summarizer provider (Phase 3), keeping the same tests running against the mock. -3. Add timers as host events (`TimerObserved`) for retry backoff, and an +2. Add timers as host events (`TimerObserved`) for retry backoff, and an operator retry for blocked jobs. -4. Add message chunking on UTF-8 boundaries, upstream of `message-appended`. -5. Turns and prompts (Phase 4) on top of `unii.view-ready?` and the +3. Add message chunking on UTF-8 boundaries, upstream of `message-appended`. +4. Turns and prompts (Phase 4) on top of `unii.view-ready?` and the rendered view. diff --git a/unii/docs/STORAGE_PHASE2.md b/unii/docs/STORAGE_PHASE2.md new file mode 100644 index 0000000..dd366a9 --- /dev/null +++ b/unii/docs/STORAGE_PHASE2.md @@ -0,0 +1,50 @@ +# Phase 2 storage and recovery + +The journal remains the sole authority. Blobs hold message and accepted +summary text; checkpoints and indexes can always be discarded and rebuilt. + +## Fault outcomes + +| Fault | Defined outcome | Automated evidence | +|---|---|---| +| Crash before/during blob write, sync or rename | No journal record; any temp/orphan is ignored | `test_storage_faults` boundary sweep | +| Crash after blob durability, before journal completion | Blob may be orphaned; incomplete journal bytes are set aside; no record is fabricated | boundary sweep and torn-write case | +| Crash after a complete frame write, before/after sync | On the test filesystem the checksum-valid frame recovers; callers must still treat a failed append as unacknowledged | boundary sweep | +| Incomplete final frame at every byte | Preserve `journal.tail-*`, truncate to the committed prefix, continue | `test_storage` exhaustive cut test | +| Bad header/checksum/separator/sequence in committed data | Refuse with `repair required`; do not rewrite the shard | `test_storage` corruption cases | +| Missing or corrupted committed blob | Refuse; journal remains byte-identical | `test_storage_phase2` | +| Corrupt or stale checkpoint | Rename `.invalid-*`, try an older valid checkpoint, otherwise replay from record one | `test_storage_phase2` | +| Stale/corrupt/missing index | Rebuild from validated journal and blobs | indexes are replaced on open | +| ENOSPC before blob commit | Refuse append; no journal or supervisor state mutation | `test_storage_faults` and `test_storage` | +| ENOSPC/torn journal append | Refuse append; isolate partial final frame on restart | `test_storage_faults` | +| Second owner | Refuse nonblocking OS `flock`; no PID-file inference | `test_storage` same- and cross-process cases | + +The boundary sweep discovers every write, file/directory sync and rename +callback exercised by blob, shard-manifest, journal and index commits, then +repeats the append with a simulated process death at each callback. Recovery +must produce either zero records or the one complete transaction, never a +partial or invented transaction. + +## Checkpoint proof + +A checkpoint binds: + +* physical journal sequence and physical-payload hash; +* rule bundle and complete policy configuration; +* canonical serialized Shen state and state hash; +* exact rendered view hash and revision; +* pending summary commands. + +Recovery validates all fields, restores the state, and starts transition +replay at `checkpoint.seq + 1`. Tests compare final state and view bytes/hashes +with uninterrupted execution and assert the number of replayed records. + +## Remaining limits + +* Blob garbage collection and backup/restore are not implemented. +* Indexes are periodically/close-time replaced rather than incrementally + paged; their in-memory metadata is proportional to record/message count. +* Checkpoint files retain the newest four manifest entries; invalid and + orphaned diagnostic files require operator cleanup. +* Linux x86_64 is tested. macOS uses `F_FULLFSYNC` for regular files, but + filesystem/device durability still needs platform testing. diff --git a/unii/docs/contracts/storage.md b/unii/docs/contracts/storage.md index 96f0cb9..40d1e43 100644 --- a/unii/docs/contracts/storage.md +++ b/unii/docs/contracts/storage.md @@ -1,8 +1,9 @@ -# Storage (storage_format 1) +# Storage (storage_format 2) -A chat is stored in one directory, which one process owns at a time. Phase 1 -has one storage adapter: a framed append journal (`host/storage.lua`) with -POSIX helpers in `host/posix.lua`, which use the LuaJIT FFI. +A chat is stored in one directory, which one process owns at a time. Existing +callers keep the same storage interface, while Phase 2 adds content-addressed +blobs, daily journal shards, checkpoints and rebuildable indexes. POSIX +durability operations remain isolated in `host/posix.lua`. ## Interface @@ -13,19 +14,29 @@ store:append(bytes) -> seq -- write all bytes, then fsync, before r store:close() -- release the lock ``` -Another backend (with blobs, checkpoints or shards) can replace this one -without changing the supervisor, as long as it keeps the same contract: -records come back in the order they were appended, `append` is durable -when it returns, and the store has a single owner. +`records()` remains compatible, but its payload fields are lazy: one framed +record is read and hydrated at a time. `iter_records(after_seq)` is the +streaming replay path. `append` does not return until blob, journal and +directory durability boundaries have succeeded. ## Directory layout | File | Purpose | |---|---| | `LOCK` | holds the `flock(LOCK_EX \| LOCK_NB)` lock, open for the lifetime of the store. A second owner, in this process or another, gets `chat is owned by another process` | -| `journal.uj` | the journal | +| `journals/MANIFEST` | ordered `USM1` list of daily shards | +| `journals/YYYY-MM-DD.uj` | framed records; sequence numbers remain global across shards | +| `blobs/` | immutable message and accepted-summary UTF-8 bytes | +| `checkpoints/checkpoint--.uc` | exact serialized core state and recovery metadata | +| `checkpoints/MANIFEST` | newest four checkpoint candidates | +| `indexes/{journal,messages,blobs,nodes,jobs}.idx` | rebuildable projections, never authority | | `journal.tail--.bin` | uncommitted crash tail kept for diagnostics: the `n` bytes that followed committed data at `offset` | +The init record uses the `0000-00-00` shard. A `message-appended` host event +selects its recorded UTC day; records without a date remain on the active +shard. Creating a shard syncs the empty file, journal directory and manifest +before a frame can be appended. + ## Frame ``` @@ -40,7 +51,13 @@ when it returns, and the store has a single owner. When the journal is opened: -* **Complete frames:** read in order. +* **Complete frames:** stream in manifest order. Only one bounded frame is + resident. Global sequence continuity and every frame checksum are checked. +* **Blob references:** message and summary event text is absent from physical + frames. The existing byte count and SHA-256 fields name the blob. Every + committed reference is size- and hash-verified before replay. Missing or + corrupted blobs refuse open with `repair required`; journal and blob bytes + are not changed. * **Incomplete final frame:** treated as an uncommitted crash tail. This covers a partial header shorter than 64 bytes and a frame that ends before its checksum line. The tail bytes are written to a `journal.tail-*` file @@ -52,6 +69,10 @@ When the journal is opened: (repair required)`, and the file is left unchanged. This covers a bad header, a checksum mismatch, a missing separator or a sequence gap anywhere before the end. No history is discarded. +* **Checkpoints:** newest to oldest, verify framing, journal sequence and + physical-record anchor, bundle/configuration, serialized state hash, + rendered view hash/revision and pending commands. A bad or stale checkpoint + is renamed `.invalid-*`; recovery tries an older one, then record one. ## Transaction payloads @@ -81,13 +102,18 @@ Records 2 and later are `event` transactions: ## Commit order (supervisor) -1. The boundary validates the event. -2. The pure transition runs. +1. The boundary validates the event and the pure transition runs. +2. Each new message or summary blob is written to a temporary file, checked, + file-synced, atomically renamed and its directory synced. 3. The view is rendered and its bytes are checked against the core's accounting. The view and the state are hashed. -4. The transaction is appended, using a write-all loop followed by fsync. +4. The hash-only transaction is appended with a write-all loop, the shard is + synced, then the journal directory is synced. 5. Only after that does the supervisor adopt the new state and dispatch commands. If step 4 fails, the previous state is kept. +6. Checkpoints and indexes are written by temp/write/sync/rename/directory + sync. Their failure cannot undo or compete with an authoritative journal + commit. ## Replay @@ -97,7 +123,9 @@ When the store is opened: unsupported format version, or a configuration different from the one passed in. Changing the policy epoch needs an epoch event that does not exist yet. -* Every transaction is replayed through the same transition, and the +* The latest valid checkpoint restores exact core state, view, pending + commands and journal position. Only later transactions are replayed. +* Every replayed transaction runs through the same transition, and the commands, decisions, view hash and state hash must match exactly. Otherwise the open fails with `replay divergence at seq N: `. * The view is never re-fitted from the current policy. It comes only from @@ -106,15 +134,18 @@ When the store is opened: ## Durability and limits -* Durability depends on `fsync` of the journal and, when the journal is - created, of the directory. On macOS, `fsync` does not flush the drive - cache (`F_FULLFSYNC` is not used), so its guarantee is weaker. Only Linux - x86_64 has been verified. -* The whole journal is read into memory on open, and replay starts from - record 1. Checkpoints, indexes, blobs, day shards and backup/restore do - not exist yet. -* Message text is stored only inside journal transactions, and the - supervisor keeps a map from message id to text in RAM. The plan requires - bounded memory and paging, which is still to do. -* Fault injection exists only for a failed append (`test_storage.lua`). It - does not yet cover every write, sync and rename boundary. +* Linux uses `fsync`. macOS regular files use `fcntl(F_FULLFSYNC)` because + macOS `fsync` may not flush volatile drive caches; directories still use + `fsync`. Actual guarantees remain subject to filesystem, device firmware + and mount settings. Linux x86_64 is the verified platform; macOS behavior + is implemented but not exercised in CI. +* Recovery stores bounded frame metadata and rebuildable indexes in memory, + not journal payloads. Core state still contains its bounded active view and + queues. Message source text is loaded from blobs on demand. +* Checkpoints are taken every 32 records by default and explicitly on demand. + At most four are named by the checkpoint manifest. +* `test_storage_faults.lua` injects process death at every observed write, + sync and rename boundary, plus torn writes and ENOSPC. Complete durable + frames recover; incomplete final frames are set aside; committed + corruption/missing blobs refuse without rewriting authority. +* Backup/restore is not implemented. diff --git a/unii/host/posix.lua b/unii/host/posix.lua index 1b8bc9a..e7d31c3 100644 --- a/unii/host/posix.lua +++ b/unii/host/posix.lua @@ -1,7 +1,7 @@ -- unii/host/posix.lua -- the only module that touches POSIX file APIs. --- LuaJIT FFI bindings for open/write/fsync/flock/ftruncate/rename and a --- directory fsync. Flag values are per platform; only Linux x86_64 has been --- exercised by the tests (see docs/contracts/storage.md for guarantees). +-- All durability boundaries pass through the fault hook below. Tests use it +-- to emulate process death, short/torn writes and ENOSPC without weakening +-- production error handling. local ffi = require("ffi") ffi.cdef [[ @@ -15,6 +15,7 @@ ffi.cdef [[ int rename(const char *from, const char *to); int mkdir(const char *path, unsigned int mode); long lseek(int fd, long offset, int whence); + int fcntl(int fd, int cmd, ...); char *strerror(int errnum); ]] @@ -34,6 +35,28 @@ M.verified_platform = (M.platform == "Linux/x64") local LOCK_EX, LOCK_NB, LOCK_UN = 2, 4, 8 local EINTR, EWOULDBLOCK = 4, (jit.os == "OSX") and 35 or 11 +local F_FULLFSYNC = 51 +local fault_hook +local fd_paths = {} + +function M.set_fault_hook(fn) + local old = fault_hook + fault_hook = fn + return old +end + +local function hit(op, detail) + if not fault_hook then return nil end + return fault_hook(op, detail or {}) +end + +local function injected(action, op) + if action == "crash" then error("injected crash at " .. op, 0) end + if type(action) == "table" and action.error then + return nil, action.error .. " (injected at " .. op .. ")" + end + return true +end local function errstr(what, path) local e = ffi.errno() @@ -49,11 +72,13 @@ function M.open(path, mode) else error("posix.open: unknown mode " .. tostring(mode)) end local fd = C.open(path, flags, ffi.new("int", 420)) -- 0644 if fd < 0 then return nil, errstr("open", path) end + fd_paths[tonumber(fd)] = path return fd end function M.close(fd) if C.close(fd) ~= 0 then return nil, errstr("close", fd) end + fd_paths[tonumber(fd)] = nil return true end @@ -62,18 +87,48 @@ function M.write_all(fd, s) local buf = ffi.cast("const char *", s) local off, n = 0, #s while off < n do - local w = tonumber(C.write(fd, buf + off, n - off)) + local op = "write" + local action = hit(op .. ".before", { fd = fd, path = fd_paths[tonumber(fd)], offset = off, bytes = n - off }) + local ok, err = injected(action, op .. ".before") + if not ok then return nil, err end + local want = n - off + if type(action) == "table" and action.short then want = math.min(want, action.short) end + local w = tonumber(C.write(fd, buf + off, want)) if w < 0 then if ffi.errno() ~= EINTR then return nil, errstr("write", fd) end else off = off + w + action = hit(op .. ".after", { fd = fd, path = fd_paths[tonumber(fd)], offset = off, bytes = w }) + ok, err = injected(action, op .. ".after") + if not ok then return nil, err end end end return true end function M.fsync(fd) + local action = hit("fsync.before", { fd = fd, path = fd_paths[tonumber(fd)] }) + local ok, err = injected(action, "fsync.before") + if not ok then return nil, err end if C.fsync(fd) ~= 0 then return nil, errstr("fsync", fd) end + action = hit("fsync.after", { fd = fd, path = fd_paths[tonumber(fd)] }) + ok, err = injected(action, "fsync.after") + if not ok then return nil, err end + return true +end + +-- macOS fsync does not request a drive-cache flush. F_FULLFSYNC does. It is +-- used for regular files when available; directory descriptors still use +-- fsync because F_FULLFSYNC is not defined for them. +function M.sync_file(fd) + if jit.os ~= "OSX" then return M.fsync(fd) end + local action = hit("fullfsync.before", { fd = fd, path = fd_paths[tonumber(fd)] }) + local ok, err = injected(action, "fullfsync.before") + if not ok then return nil, err end + if C.fcntl(fd, F_FULLFSYNC) ~= 0 then return nil, errstr("fcntl(F_FULLFSYNC)", fd) end + action = hit("fullfsync.after", { fd = fd, path = fd_paths[tonumber(fd)] }) + ok, err = injected(action, "fullfsync.after") + if not ok then return nil, err end return true end @@ -81,7 +136,7 @@ function M.fsync_dir(path) local fd, err = M.open(path, "dir") if not fd then return nil, err end local ok, e2 = M.fsync(fd) - C.close(fd) + M.close(fd) return ok, e2 end @@ -97,12 +152,24 @@ end function M.unlock(fd) return C.flock(fd, LOCK_UN) == 0 end function M.ftruncate(fd, len) + local action = hit("truncate.before", { fd = fd, length = len }) + local ok, err = injected(action, "truncate.before") + if not ok then return nil, err end if C.ftruncate(fd, len) ~= 0 then return nil, errstr("ftruncate", fd) end + action = hit("truncate.after", { fd = fd, length = len }) + ok, err = injected(action, "truncate.after") + if not ok then return nil, err end return true end function M.rename(from, to) + local action = hit("rename.before", { from = from, to = to }) + local ok, err = injected(action, "rename.before") + if not ok then return nil, err end if C.rename(from, to) ~= 0 then return nil, errstr("rename", from) end + action = hit("rename.after", { from = from, to = to }) + ok, err = injected(action, "rename.after") + if not ok then return nil, err end return true end @@ -125,6 +192,25 @@ function M.read_file(path) return s end +function M.file_size(path) + local f = io.open(path, "rb") + if not f then return nil end + local n = f:seek("end") + f:close() + return n +end + +function M.read_range(path, offset, len) + local f, err = io.open(path, "rb") + if not f then return nil, err end + local ok = f:seek("set", offset) + if not ok then f:close(); return nil, "seek failed for " .. path end + local s = f:read(len) + f:close() + if not s or #s ~= len then return nil, "short read from " .. path end + return s +end + -- Write a whole file durably: temp file, write-all, fsync, rename, fsync dir. function M.write_file_atomic(dir, name, bytes) local tmp = dir .. "/." .. name .. ".tmp" @@ -132,8 +218,8 @@ function M.write_file_atomic(dir, name, bytes) if not fd then return nil, err end local ok, e = M.ftruncate(fd, 0) if ok then ok, e = M.write_all(fd, bytes) end - if ok then ok, e = M.fsync(fd) end - C.close(fd) + if ok then ok, e = M.sync_file(fd) end + M.close(fd) if not ok then return nil, e end ok, e = M.rename(tmp, dir .. "/" .. name) if not ok then return nil, e end diff --git a/unii/host/storage.lua b/unii/host/storage.lua index 4f25aaa..02038fb 100644 --- a/unii/host/storage.lua +++ b/unii/host/storage.lua @@ -1,5 +1,5 @@ --- unii/host/storage.lua -- storage adapter interface and its first --- implementation: a single framed append journal per chat directory. +-- unii/host/storage.lua -- Phase 2 day-sharded journal, content-addressed +-- blobs, checkpoints and rebuildable indexes. -- -- Interface (docs/contracts/storage.md): -- local store, info = storage.open(dir) -- exclusive owner; recovers tail @@ -9,16 +9,18 @@ -- -- Frame: "UJ1 \n" "\n" "\n" -- --- Not yet implemented (Phase 2): content-addressed blobs, checkpoints, --- rebuildable indexes, day-sharded journals, backup/restore, and fault --- injection at every storage boundary. local posix = require("unii.host.posix") local sha256 = require("unii.host.sha256") +local codec = require("unii.host.codec") +local Blobs = require("unii.host.storage_blobs") +local Checkpoints = require("unii.host.storage_checkpoint") +local Index = require("unii.host.storage_index") local M = {} M.MAX_RECORD = 16 * 1024 * 1024 -M.JOURNAL = "journal.uj" +M.JOURNAL = "journals/0000-00-00.uj" M.LOCK = "LOCK" +M.SHARDS = "journals/MANIFEST" local Store = {} Store.__index = Store @@ -32,7 +34,8 @@ M.frame = frame -- Parse journal bytes. Returns records, the byte offset where committed -- data ends, and either nil (clean), "tail" (incomplete final frame) or -- raises on corruption that is not a crash tail. -function M.parse(bytes) +function M.parse(bytes, first_seq) + first_seq = first_seq or 1 local records, off, n = {}, 1, #bytes while off <= n do local nl = bytes:find("\n", off, true) @@ -58,8 +61,9 @@ function M.parse(bytes) if sep1 ~= "\n" or sep2 ~= "\n" or sum ~= sha256.hex(header .. payload) then error(("journal corrupt at byte %d (seq %d): checksum mismatch (repair required)"):format(off - 1, seq), 0) end - if seq ~= #records + 1 then - error(("journal corrupt at byte %d: sequence %d, expected %d"):format(off - 1, seq, #records + 1), 0) + local expected = first_seq + #records + if seq ~= expected then + error(("journal corrupt at byte %d: sequence %d, expected %d"):format(off - 1, seq, expected), 0) end records[#records + 1] = { seq = seq, payload = payload } off = off + total @@ -67,6 +71,222 @@ function M.parse(bytes) return records, n, nil end +local function int(v) return codec.to_number(v) end + +local function event_of(txn) + return txn.t == "map" and txn.v.event and txn.v.event.t == "list" and txn.v.event.v or nil +end + +-- Journal transactions omit the text field from message and summary events. +-- Their existing byte count and SHA-256 fields are the durable blob reference. +local function physicalize(payload, blobs) + local decoded, txn = pcall(codec.decode, payload) + if not decoded then return payload, {}, nil end + local ev = event_of(txn) + local refs, day = {} + if ev and ev[1] and ev[1].t == "sym" then + if ev[1].v == "message-appended" then + local content = ev[5] and ev[5].v + if content and #content == 4 then + local hash, bytes, text = content[3].v, int(content[2]), content[4].v + if #text ~= bytes or sha256.hex(text) ~= hash then error("message blob declaration mismatch", 0) end + blobs:put(text, hash) + content[4] = nil + refs[#refs + 1] = { hash = hash, bytes = bytes } + end + local date = ev[4] and ev[4].v + day = date and date:match("^(%d%d%d%d%-%d%d%-%d%d)") or nil + elseif ev[1].v == "summary-completed" and #ev == 6 then + local hash, bytes, text = ev[5].v, int(ev[4]), ev[6].v + if #text ~= bytes or sha256.hex(text) ~= hash then error("summary blob declaration mismatch", 0) end + blobs:put(text, hash) + ev[6] = nil + refs[#refs + 1] = { hash = hash, bytes = bytes } + end + end + return codec.encode(txn), refs, day +end + +local function hydrate(physical, blobs) + local decoded, txn = pcall(codec.decode, physical) + if not decoded then return physical end + local ev = event_of(txn) + if ev and ev[1] and ev[1].t == "sym" then + if ev[1].v == "message-appended" then + local content = ev[5] and ev[5].v + if content and #content == 3 then + content[4] = codec.text(blobs:get(content[3].v, int(content[2]))) + end + elseif ev[1].v == "summary-completed" and #ev == 5 then + ev[6] = codec.text(blobs:get(ev[5].v, int(ev[4]))) + end + end + return codec.encode(txn), txn +end + +local function refs_of(physical) + local decoded, txn = pcall(codec.decode, physical) + if not decoded then return nil, {} end + local refs = {} + local ev = event_of(txn) + if ev and ev[1] and ev[1].t == "sym" then + if ev[1].v == "message-appended" then + local content = ev[5] and ev[5].v + if content and #content == 3 then refs[1] = { hash = content[3].v, bytes = int(content[2]) } end + elseif ev[1].v == "summary-completed" and #ev == 5 then + refs[1] = { hash = ev[5].v, bytes = int(ev[4]) } + end + end + return txn, refs +end + +local function read_manifest(path) + local bytes = posix.read_file(path) + if not bytes then return {} end + if bytes:sub(1, 5) ~= "USM1\n" then error("journal shard manifest is corrupt (repair required)", 0) end + local out, seen = {}, {} + for day in bytes:sub(6):gmatch("([^\n]+)\n") do + if not day:match("^%d%d%d%d%-%d%d%-%d%d$") or seen[day] then + error("journal shard manifest is corrupt (repair required)", 0) + end + seen[day], out[#out + 1] = true, day + end + return out +end + +local function index_txn(idx, txn, seq, refs) + for _, ref in ipairs(refs) do + local b = idx.blobs[ref.hash] or { bytes = ref.bytes, seqs = {} } + b.seqs[#b.seqs + 1], idx.blobs[ref.hash] = seq, b + end + if not txn then return end + local ev = event_of(txn) + if ev and ev[1].v == "message-appended" then + local content = ev[5].v + idx.messages[tostring(int(ev[2]))] = { + seq = seq, hash = content[3].v, bytes = int(content[2]), kind = ev[3].v, date = ev[4].v, + } + elseif ev and (ev[1].v == "summary-completed" or ev[1].v == "summary-failed") then + idx.jobs[ev[2].v] = { status = ev[1].v == "summary-completed" and "completed" or "failed", seq = seq } + end + local decisions = txn.v.decisions + if decisions and decisions.t == "list" then + for _, d in ipairs(decisions.v) do + if d.t == "list" and d.v[1] and d.v[1].v == "node-committed" then + local key = d.v[2].v + idx.nodes[int(key[2]) .. "/" .. int(key[3])] = seq + elseif d.t == "list" and d.v[1] and d.v[1].v == "job-blocked" then + idx.jobs[d.v[2].v] = { status = "blocked", seq = seq } + end + end + end + local commands = txn.v.commands + if commands and commands.t == "list" then + for _, c in ipairs(commands.v) do + if c.t == "list" and c.v[1] and c.v[1].v == "submit-summary" then + idx.jobs[c.v[3].v] = { status = "pending", seq = seq } + end + end + end +end + +local function lazy_record(store, meta) + return setmetatable({ seq = meta.seq }, { + __index = function(_, key) + if key ~= "payload" then return nil end + local physical, err = posix.read_range(meta.path, meta.payload_offset, meta.length) + if not physical then error(err, 0) end + return (hydrate(physical, store.blobs)) + end, + }) +end + +local function recover_tail(store, path, good, size) + local tail = posix.read_range(path, good, size - good) or "" + local name = ("journal.tail-%d-%d.bin"):format(good, #tail) + assert(posix.write_file_atomic(store.dir, name, tail)) + local fd = assert(posix.open(path, "append")) + assert(posix.ftruncate(fd, good)) + assert(posix.sync_file(fd)) + posix.close(fd) + assert(posix.fsync_dir(store.journal_dir)) + store.info.tail_isolated = store.dir .. "/" .. name +end + +-- Stream one frame at a time. Only bounded record bytes, never a whole shard, +-- are resident. Record payloads returned through records() are lazy. +local function scan_shard(store, day, expected, is_last) + local path = store.journal_dir .. "/" .. day .. ".uj" + local f = io.open(path, "rb") + if not f then error("journal shard " .. day .. " is missing (repair required)", 0) end + local size, off = f:seek("end"), 0 + f:seek("set", 0) + while off < size do + local start = off + local header_line = f:read("*l") + if not header_line then break end + off = f:seek() + local header_had_newline = off - start == #header_line + 1 + local seqs, lens = header_line:match("^UJ1 (%d+) (%d+)$") + local seq, len = tonumber(seqs), tonumber(lens) + if not seq or not len or len > M.MAX_RECORD then + f:close() + if is_last and not header_had_newline then + recover_tail(store, path, start, size) + return expected + end + error(("journal corrupt at byte %d in shard %s: bad frame header (repair required)"):format(start, day), 0) + end + local payload_offset = off + local physical, sep = f:read(len), f:read(1) + local sum_start = f:seek() + local sumline = f:read("*l") + off = f:seek() or size + local sum_had_newline = sumline and off - sum_start == #sumline + 1 + if not physical or #physical ~= len or sep ~= "\n" or not sumline or #sumline ~= 64 or not sum_had_newline then + f:close() + local expected_end = start + #header_line + 1 + len + 1 + 64 + 1 + if is_last and size < expected_end then recover_tail(store, path, start, size); return expected end + error(("journal corrupt at byte %d in non-final shard %s (repair required)"):format(start, day), 0) + end + local header = header_line .. "\n" + if sumline ~= sha256.hex(header .. physical) then + f:close() + error(("journal corrupt at byte %d (seq %d): checksum mismatch (repair required)"):format(start, seq), 0) + end + if seq ~= expected then + f:close() + error(("journal corrupt at byte %d: sequence %d, expected %d"):format(start, seq, expected), 0) + end + local txn, refs = refs_of(physical) + for _, ref in ipairs(refs) do store.blobs:verify(ref.hash, ref.bytes) end + local meta = { seq = seq, day = day, path = path, offset = start, + payload_offset = payload_offset, length = len, physical_hash = sha256.hex(physical) } + store.meta[#store.meta + 1] = meta + store.recs[#store.recs + 1] = lazy_record(store, meta) + store.idx.records[#store.idx.records + 1] = meta + index_txn(store.idx, txn, seq, refs) + expected = expected + 1 + end + f:close() + return expected +end + +local function ensure_shard(store, day) + day = day or store.active_day or "0000-00-00" + if store.shard_set[day] then store.active_day = day; return end + local path = store.journal_dir .. "/" .. day .. ".uj" + local fd, err = posix.open(path, "append") + if not fd then error(err, 0) end + assert(posix.sync_file(fd)) + posix.close(fd) + assert(posix.fsync_dir(store.journal_dir)) + store.shards[#store.shards + 1], store.shard_set[day] = day, true + assert(posix.write_file_atomic(store.journal_dir, "MANIFEST", + "USM1\n" .. table.concat(store.shards, "\n") .. "\n")) + store.active_day = day +end + function M.open(dir) assert(posix.mkdir_p(dir)) local lockfd, err = posix.open(dir .. "/" .. M.LOCK, "rw") @@ -76,50 +296,97 @@ function M.open(dir) posix.close(lockfd) error(busy and ("chat " .. dir .. " is owned by another process") or lerr, 0) end - - local path = dir .. "/" .. M.JOURNAL - local existed = posix.read_file(path) ~= nil - local bytes = posix.read_file(path) or "" - local pok, records, good, state = pcall(M.parse, bytes) - if not pok then - posix.close(lockfd) - error(records, 0) + local function build() + local journal_dir = dir .. "/journals" + assert(posix.mkdir_p(journal_dir)) + assert(posix.fsync_dir(dir)) + local info = {} + local store = setmetatable({ + dir = dir, journal_dir = journal_dir, lockfd = lockfd, info = info, + blobs = Blobs.new(dir), checkpoints = Checkpoints.new(dir), + recs = {}, meta = {}, idx = { records = {}, messages = {}, blobs = {}, nodes = {}, jobs = {} }, + }, Store) + store.shards = read_manifest(dir .. "/" .. M.SHARDS) + store.shard_set = {} + for _, day in ipairs(store.shards) do store.shard_set[day] = true end + local expected = 1 + for i, day in ipairs(store.shards) do expected = scan_shard(store, day, expected, i == #store.shards) end + store.next_seq, store.active_day = expected, store.shards[#store.shards] + info.records = #store.recs + local iok, ierr = pcall(Index.write, dir, store.idx) + if not iok then info.index_error = tostring(ierr) end + store.index_dirty = false + return store, info end - local info = { records = #records } - - local fd, oerr = posix.open(path, "append") - if not fd then posix.close(lockfd); error(oerr, 0) end - if state == "tail" then - -- Isolate the uncommitted crash tail for diagnostics, then drop it. - local tail = bytes:sub(good + 1) - local name = ("journal.tail-%d-%d.bin"):format(good, #tail) - assert(posix.write_file_atomic(dir, name, tail)) - assert(posix.ftruncate(fd, good)) - assert(posix.fsync(fd)) - info.tail_isolated = dir .. "/" .. name - end - if not existed then assert(posix.fsync(fd)); assert(posix.fsync_dir(dir)) end - - return setmetatable({ dir = dir, fd = fd, lockfd = lockfd, recs = records, - next_seq = #records + 1 }, Store), info + local bok, store, info = pcall(build) + if not bok then posix.close(lockfd); error(store, 0) end + return store, info end function Store:records() return self.recs end +function Store:iter_records(after) + local i = (after or 0) + 1 + return function() + local record = self.recs[i] + i = i + 1 + return record + end +end + function Store:append(payload) if #payload > M.MAX_RECORD then error("storage: record over " .. M.MAX_RECORD .. " bytes", 0) end - local seq = self.next_seq - local ok, err = posix.write_all(self.fd, frame(seq, payload)) - if not ok then error("storage append: " .. err, 0) end - ok, err = posix.fsync(self.fd) - if not ok then error("storage sync: " .. err, 0) end - self.recs[#self.recs + 1] = { seq = seq, payload = payload } + local physical, refs, day = physicalize(payload, self.blobs) + ensure_shard(self, day) + local seq, path = self.next_seq, self.journal_dir .. "/" .. self.active_day .. ".uj" + local bytes = frame(seq, physical) + local offset = posix.file_size(path) or 0 + local fd, err = posix.open(path, "append") + if not fd then error("storage append: " .. tostring(err), 0) end + local ok + ok, err = posix.write_all(fd, bytes) + if ok then ok, err = posix.sync_file(fd) end + posix.close(fd) + if not ok then error("storage append: " .. tostring(err), 0) end + ok, err = posix.fsync_dir(self.journal_dir) + if not ok then error("storage directory sync: " .. tostring(err), 0) end + local header = ("UJ1 %d %d\n"):format(seq, #physical) + local meta = { seq = seq, day = self.active_day, path = path, offset = offset, + payload_offset = offset + #header, length = #physical, physical_hash = sha256.hex(physical) } + self.meta[#self.meta + 1] = meta + self.recs[#self.recs + 1] = lazy_record(self, meta) + self.idx.records[#self.idx.records + 1] = meta + local decoded, txn = pcall(codec.decode, physical) + index_txn(self.idx, decoded and txn or nil, seq, refs) self.next_seq = seq + 1 + self.index_dirty = true + if seq % 64 == 0 then + local iok, ierr = pcall(self.rebuild_indexes, self) + if not iok then self.info.index_error = tostring(ierr) end + end return seq end +function Store:record_anchor(seq) + local meta = self.meta[seq] + return meta and meta.physical_hash or nil +end + +function Store:message_index() return self.idx.messages end +function Store:get_blob(hash, bytes) return self.blobs:get(hash, bytes) end +function Store:write_checkpoint(seq, payload) return self.checkpoints:write(seq, payload) end +function Store:checkpoint_candidates() return self.checkpoints:candidates() end +function Store:set_aside_checkpoint(name, reason) return self.checkpoints:set_aside(name, reason) end +function Store:rebuild_indexes() + Index.write(self.dir, self.idx) + self.index_dirty = false +end + function Store:close() - if self.fd then posix.close(self.fd); self.fd = nil end + if self.index_dirty then + local ok, err = pcall(self.rebuild_indexes, self) + if not ok then self.info.index_error = tostring(err) end + end if self.lockfd then posix.unlock(self.lockfd); posix.close(self.lockfd); self.lockfd = nil end end diff --git a/unii/host/storage_blobs.lua b/unii/host/storage_blobs.lua new file mode 100644 index 0000000..c3a2c9a --- /dev/null +++ b/unii/host/storage_blobs.lua @@ -0,0 +1,57 @@ +-- Content-addressed immutable blobs. A blob is durable before its hash can +-- appear in a journal frame: temp write, file sync, rename, directory sync. +local posix = require("unii.host.posix") +local sha256 = require("unii.host.sha256") + +local M = {} + +local function assert_ok(ok, err) + if not ok then error("blob storage: " .. tostring(err), 0) end +end + +function M.new(root) + local dir = root .. "/blobs" + assert_ok(posix.mkdir_p(dir)) + assert_ok(posix.fsync_dir(root)) + return setmetatable({ root = root, dir = dir }, { __index = M }) +end + +function M:path(hash) + return self.dir .. "/" .. hash +end + +function M:put(bytes, expected) + local hash = sha256.hex(bytes) + if expected and hash ~= expected then error("blob hash does not match journal reference", 0) end + local path = self:path(hash) + local old = posix.read_file(path) + if old then + if #old ~= #bytes or sha256.hex(old) ~= hash then + error("corrupted existing blob " .. hash .. "; refusing to overwrite", 0) + end + return hash + end + assert_ok(posix.write_file_atomic(self.dir, hash, bytes)) + return hash +end + +function M:get(hash, expected_size) + local path = self:path(hash) + local bytes = posix.read_file(path) + if not bytes then error("missing committed blob " .. hash .. " (repair required)", 0) end + if expected_size and #bytes ~= expected_size then + error(("corrupted blob %s: size %d, expected %d (repair required)") + :format(hash, #bytes, expected_size), 0) + end + if sha256.hex(bytes) ~= hash then + error("corrupted blob " .. hash .. ": checksum mismatch (repair required)", 0) + end + return bytes +end + +function M:verify(hash, expected_size) + self:get(hash, expected_size) + return true +end + +return M diff --git a/unii/host/storage_checkpoint.lua b/unii/host/storage_checkpoint.lua new file mode 100644 index 0000000..6ebf675 --- /dev/null +++ b/unii/host/storage_checkpoint.lua @@ -0,0 +1,93 @@ +-- Durable checkpoint files and their small manifest. Checkpoints are +-- accelerators; a bad or stale one is set aside and an older candidate is +-- tried before falling back to full journal replay. +local posix = require("unii.host.posix") +local sha256 = require("unii.host.sha256") + +local M = {} +local MAGIC = "UC1 " + +local function frame(payload) + local header = MAGIC .. #payload .. "\n" + return header .. payload .. "\n" .. sha256.hex(header .. payload) .. "\n" +end + +local function parse(bytes) + local nl = bytes:find("\n", 1, true) + if not nl then return nil, "missing header" end + local len = tonumber(bytes:sub(1, nl):match("^UC1 (%d+)\n$")) + if not len then return nil, "bad header" end + local payload = bytes:sub(nl + 1, nl + len) + local sep = bytes:sub(nl + len + 1, nl + len + 1) + local sum = bytes:sub(nl + len + 2, nl + len + 65) + local final = bytes:sub(nl + len + 66) + if #payload ~= len or sep ~= "\n" or #sum ~= 64 or final ~= "\n" then + return nil, "truncated" + end + if sum ~= sha256.hex(bytes:sub(1, nl) .. payload) then return nil, "checksum mismatch" end + return payload +end + +local Checkpoints = {} +Checkpoints.__index = Checkpoints + +function M.new(root) + local dir = root .. "/checkpoints" + assert(posix.mkdir_p(dir)) + assert(posix.fsync_dir(root)) + return setmetatable({ root = root, dir = dir, manifest = dir .. "/MANIFEST" }, Checkpoints) +end + +function Checkpoints:names() + local bytes = posix.read_file(self.manifest) + if not bytes then return {} end + if bytes:sub(1, 5) ~= "UCM1\n" then return {} end + local out = {} + for name in bytes:sub(6):gmatch("([^\n]+)\n") do + if name:match("^checkpoint%-%d+%-%x+%.uc$") then out[#out + 1] = name end + end + return out +end + +function Checkpoints:write(seq, payload) + local digest = sha256.hex(payload) + local name = ("checkpoint-%d-%s.uc"):format(seq, digest:sub(1, 16)) + assert(posix.write_file_atomic(self.dir, name, frame(payload))) + local names, next_names = self:names(), { name } + for _, old in ipairs(names) do + if old ~= name and #next_names < 4 then next_names[#next_names + 1] = old end + end + assert(posix.write_file_atomic(self.dir, "MANIFEST", "UCM1\n" .. table.concat(next_names, "\n") .. "\n")) + return name +end + +function Checkpoints:set_aside(name, reason) + local suffix = sha256.hex(reason):sub(1, 8) + local ok = posix.rename(self.dir .. "/" .. name, self.dir .. "/" .. name .. ".invalid-" .. suffix) + if ok then + assert(posix.fsync_dir(self.dir)) + local kept = {} + for _, candidate in ipairs(self:names()) do + if candidate ~= name then kept[#kept + 1] = candidate end + end + assert(posix.write_file_atomic(self.dir, "MANIFEST", "UCM1\n" .. table.concat(kept, "\n") + .. (#kept > 0 and "\n" or ""))) + end +end + +function Checkpoints:candidates() + local out = {} + for _, name in ipairs(self:names()) do + local bytes = posix.read_file(self.dir .. "/" .. name) + if bytes then + local payload, err = parse(bytes) + out[#out + 1] = { name = name, payload = payload, error = err } + else + out[#out + 1] = { name = name, error = "missing checkpoint file" } + end + end + return out +end + +M.frame, M.parse = frame, parse +return M diff --git a/unii/host/storage_index.lua b/unii/host/storage_index.lua new file mode 100644 index 0000000..19f60f5 --- /dev/null +++ b/unii/host/storage_index.lua @@ -0,0 +1,54 @@ +-- Rebuildable projections. These files are never authoritative: startup +-- validates the journal and blobs first, then replaces all indexes. +local posix = require("unii.host.posix") + +local M = {} + +local function sorted_keys(t) + local out = {} + for k in pairs(t) do out[#out + 1] = k end + table.sort(out, function(a, b) return tostring(a) < tostring(b) end) + return out +end + +function M.write(root, idx) + local dir = root .. "/indexes" + assert(posix.mkdir_p(dir)) + assert(posix.fsync_dir(root)) + + local journal = { "UIX1\n" } + for _, r in ipairs(idx.records) do + journal[#journal + 1] = ("%d\t%s\t%d\t%d\n"):format(r.seq, r.day, r.offset, r.length) + end + assert(posix.write_file_atomic(dir, "journal.idx", table.concat(journal))) + + local messages = { "UMX1\n" } + for _, id in ipairs(sorted_keys(idx.messages)) do + local m = idx.messages[id] + messages[#messages + 1] = ("%s\t%d\t%s\t%d\t%s\t%s\n") + :format(id, m.seq, m.hash, m.bytes, m.kind, m.date) + end + assert(posix.write_file_atomic(dir, "messages.idx", table.concat(messages))) + + local blobs = { "UBX1\n" } + for _, hash in ipairs(sorted_keys(idx.blobs)) do + local b = idx.blobs[hash] + blobs[#blobs + 1] = ("%s\t%d\t%s\n"):format(hash, b.bytes, table.concat(b.seqs, ",")) + end + assert(posix.write_file_atomic(dir, "blobs.idx", table.concat(blobs))) + + local nodes = { "UNX1\n" } + for _, key in ipairs(sorted_keys(idx.nodes)) do + nodes[#nodes + 1] = ("%s\t%d\n"):format(key, idx.nodes[key]) + end + assert(posix.write_file_atomic(dir, "nodes.idx", table.concat(nodes))) + + local jobs = { "UJX1\n" } + for _, job in ipairs(sorted_keys(idx.jobs)) do + local j = idx.jobs[job] + jobs[#jobs + 1] = ("%s\t%s\t%d\n"):format(job, j.status, j.seq) + end + assert(posix.write_file_atomic(dir, "jobs.idx", table.concat(jobs))) +end + +return M diff --git a/unii/host/supervisor.lua b/unii/host/supervisor.lua index 0f3987e..ed24c01 100644 --- a/unii/host/supervisor.lua +++ b/unii/host/supervisor.lua @@ -40,6 +40,7 @@ function M.open(dir, opts) dir = dir, core = core, store = store, info = info, provider = opts.provider, on_client_event = opts.on_client_event or function() end, outstanding = {}, started = {}, inbox = {}, messages = {}, view_cache = nil, + checkpoint_interval = opts.checkpoint_interval == nil and 32 or opts.checkpoint_interval, }, Sup) local ok, err = pcall(self._load, self, opts) if not ok then store:close(); error(err, 0) end @@ -88,7 +89,71 @@ function Sup:_load(opts) error("replay divergence at seq 1: initial state hash", 0) end self:_refresh_view() - for i = 2, #recs do self:_replay(recs[i]) end + -- Message text remains in blobs. The in-memory table contains only small + -- metadata records with a lazy text accessor. + for id, m in pairs(self.store:message_index()) do + local meta = m + self.messages[tonumber(id)] = setmetatable({ kind = meta.kind, date = meta.date }, { + __index = function(_, k) + if k == "text" then return self.store:get_blob(meta.hash, meta.bytes) end + end, + }) + end + local start = self:_restore_checkpoint() or 1 + self.info.replay_start_seq = start + 1 + self.info.replayed_records = 0 + for rec in self.store:iter_records(start) do + self:_replay(rec) + self.info.replayed_records = self.info.replayed_records + 1 + end +end + +local function checkpoint_int(cp, name) + local v = cp.v[name] + if not v or v.t ~= "int" then error("checkpoint lacks int field " .. name, 0) end + return codec.to_number(v) +end + +-- Try newest to oldest. Invalid/stale checkpoints are accelerators, not +-- authority: set them aside and continue from an older one or record one. +function Sup:_restore_checkpoint() + for _, candidate in ipairs(self.store:checkpoint_candidates()) do + local ok, seq_or_error = pcall(function() + if candidate.error then error(candidate.error, 0) end + local cp = codec.decode(candidate.payload) + if cp.t ~= "map" or not cp.v.kind or cp.v.kind.v ~= "checkpoint" then + error("not a checkpoint record", 0) + end + local seq = checkpoint_int(cp, "seq") + if seq < 1 or seq >= self.store.next_seq then error("stale journal position", 0) end + if text_field(cp, "anchor") ~= self.store:record_anchor(seq) then error("stale journal anchor", 0) end + if text_field(cp, "bundle") ~= self.core.bundle_hash then error("different rule bundle", 0) end + if codec.encode(cp.v.config) ~= codec.encode(schema.encode("config", self.config)) then + error("different configuration", 0) + end + local state = codec.to_shen(cp.v.state) + if self.core:state_hash(state) ~= text_field(cp, "state_hash") then error("state hash mismatch", 0) end + local rendered = self.core:render(state) + if sha256.hex(rendered) ~= text_field(cp, "view_hash") then error("view hash mismatch", 0) end + if self.core:status(state).rev ~= checkpoint_int(cp, "view_rev") then error("view revision mismatch", 0) end + local outstanding = {} + if not cp.v.outstanding or cp.v.outstanding.t ~= "list" then error("missing outstanding commands", 0) end + for _, tagged in ipairs(cp.v.outstanding.v) do + local command = schema.decode("command", tagged) + if command._ == "submit-summary" then outstanding[command.cmd] = command end + end + self.state, self.outstanding, self.started = state, outstanding, {} + self.view_cache = nil + self:_refresh_view() + self.info.checkpoint = candidate.name + self.info.checkpoint_seq = seq + return seq + end) + if ok then return seq_or_error end + self.store:set_aside_checkpoint(candidate.name, tostring(seq_or_error)) + self.info.checkpoint_rejected = tostring(seq_or_error) + end + return nil end function Sup:_replay(rec) @@ -120,7 +185,12 @@ function Sup:_absorb(ev, out, replaying) local rejected = false for _, d in ipairs(out.decisions) do if d._ == "event-rejected" then rejected = true end end if ev._ == "message-appended" and not rejected then - self.messages[ev.id] = { kind = ev.kind, text = ev.content.text, date = ev.date } + local hash, bytes = ev.content.sha256, ev.content.bytes + self.messages[ev.id] = setmetatable({ kind = ev.kind, date = ev.date }, { + __index = function(_, k) + if k == "text" then return self.store:get_blob(hash, bytes) end + end, + }) end if ev._ == "summary-completed" or ev._ == "summary-failed" then for cmd, c in pairs(self.outstanding) do @@ -160,9 +230,38 @@ function Sup:submit(event) error(seq, 0) end self:_absorb(schema.decode("event", tagged), out, false) + if self.checkpoint_interval > 0 and seq % self.checkpoint_interval == 0 then + local cok, cerr = pcall(self.checkpoint, self, seq) + if not cok then self.info.checkpoint_error = tostring(cerr) end + end return out, seq end +function Sup:checkpoint(seq) + seq = seq or (self.store.next_seq - 1) + if seq < 1 then error("cannot checkpoint an empty journal", 0) end + local commands = {} + for _, command in pairs(self.outstanding) do + commands[#commands + 1] = schema.encode("command", command) + end + table.sort(commands, function(a, b) + return codec.encode(a) < codec.encode(b) + end) + local payload = txn_bytes { + kind = codec.sym("checkpoint"), + seq = codec.int_of(seq), + anchor = codec.text(assert(self.store:record_anchor(seq))), + bundle = codec.text(self.core.bundle_hash), + config = schema.encode("config", self.config), + state = codec.decode(self.core:state_bytes(self.state)), + state_hash = codec.text(self:state_hash()), + view_hash = codec.text(self.view_hash), + view_rev = codec.int_of(self:status().rev), + outstanding = codec.list(commands), + } + return self.store:write_checkpoint(seq, payload) +end + -- Start every outstanding summary command not yet started in this process -- (after a restart this re-dispatches commands whose outcome was never -- journaled; summaries are read-only, so a duplicate costs only spend). diff --git a/unii/manifest.lua b/unii/manifest.lua index 1c816c1..3edb472 100644 --- a/unii/manifest.lua +++ b/unii/manifest.lua @@ -13,7 +13,7 @@ return { "jobs.shen", "transition.shen", "invariants.shen", }, - storage_format = 1, -- docs/contracts/storage.md + storage_format = 2, -- docs/contracts/storage.md codec_version = 1, -- docs/contracts/codec.md shen_lua = { diff --git a/unii/test/run.lua b/unii/test/run.lua index c66b26e..79c1035 100644 --- a/unii/test/run.lua +++ b/unii/test/run.lua @@ -31,7 +31,8 @@ end local files = { "test_codec", "test_boundary", "test_tree", "test_view", "test_transition", - "test_traces", "test_hysteresis", "test_storage", "test_network_mock", "test_milestone", + "test_traces", "test_hysteresis", "test_storage", "test_storage_phase2", + "test_storage_faults", "test_network_mock", "test_milestone", } local passed, failed, skipped = 0, 0, 0 diff --git a/unii/test/test_storage.lua b/unii/test/test_storage.lua index 050c6ea..a6533c5 100644 --- a/unii/test/test_storage.lua +++ b/unii/test/test_storage.lua @@ -40,17 +40,19 @@ end -- Rewrite record `seq` of a journal through `edit(decoded_payload)`. local function tamper(dir, seq, edit) - local path = dir .. "/" .. storage.JOURNAL - local recs = storage.parse(read(path)) + local first = seq == 1 and 1 or 2 + local path = seq == 1 and (dir .. "/" .. storage.JOURNAL) + or (dir .. "/journals/2026-10-09.uj") + local recs = storage.parse(read(path), first) local out = {} - for i, r in ipairs(recs) do + for _, r in ipairs(recs) do local payload = r.payload - if i == seq then + if r.seq == seq then local v = codec.decode(payload) edit(v) payload = codec.encode(v) end - out[#out + 1] = storage.frame(i, payload) + out[#out + 1] = storage.frame(r.seq, payload) end write(path, table.concat(out)) end @@ -182,20 +184,23 @@ return { feed(sup, 0, 6) sup:close() local path = dir .. "/" .. storage.JOURNAL - local clean = read(path) - local n = #storage.parse(clean) + local day_path = dir .. "/journals/2026-10-09.uj" + local clean, clean_day = read(path), read(day_path) + local inspected = storage.open(dir) + local n = #inspected:records() + inspected:close() tamper(dir, n, function(v) v.v.view_hash = codec.text(("0"):rep(64)) end) T.raises(function() open_sup(dir) end, "replay divergence at seq " .. n .. ": view hash") - write(path, clean) + write(path, clean); write(day_path, clean_day) tamper(dir, 3, function(v) v.v.decisions = codec.list {} end) T.raises(function() open_sup(dir) end, "replay divergence at seq 3: decisions") - write(path, clean) + write(path, clean); write(day_path, clean_day) tamper(dir, 1, function(v) v.v.bundle = codec.text(("f"):rep(64)) end) T.raises(function() open_sup(dir) end, "Refusing to replay under different rules") - write(path, clean) + write(path, clean); write(day_path, clean_day) T.raises(function() open_sup(dir, { config = schema.config { high = 99999 } }) end, "configuration differs") diff --git a/unii/test/test_storage_faults.lua b/unii/test/test_storage_faults.lua new file mode 100644 index 0000000..625c4e3 --- /dev/null +++ b/unii/test/test_storage_faults.lua @@ -0,0 +1,129 @@ +local T = require("unii.test.lib") +local storage = require("unii.host.storage") +local posix = require("unii.host.posix") +local codec = require("unii.host.codec") +local schema = require("unii.host.schema") + +local function transaction(text) + return codec.encode(codec.map { + kind = codec.sym("event"), + event = schema.encode("event", T.msg(0, "user", text, "2026-10-11T00:00:00Z")), + commands = codec.list {}, + decisions = codec.list {}, + }) +end + +local function with_hook(fn, body) + posix.set_fault_hook(fn) + local ok, a, b = pcall(body) + posix.set_fault_hook(nil) + if not ok then error(a, 0) end + return a, b +end + +return { + { "blob, journal and directory durability boundaries occur in commit order", function() + local dir = T.tmpdir("fault-order") + local store = storage.open(dir) + local trace = {} + with_hook(function(op, detail) + trace[#trace + 1] = { op = op, path = detail.path, from = detail.from, to = detail.to } + end, function() store:append(transaction("ordered blob")) end) + store:close() + + local blob_sync, journal_write, journal_sync, directory_sync + for i, e in ipairs(trace) do + if e.op == "fsync.after" and e.path and e.path:find("/blobs/.", 1, true) then blob_sync = blob_sync or i end + if e.op == "write.before" and e.path and e.path:match("%.uj$") then journal_write = journal_write or i end + if journal_write and i > journal_write and e.op == "fsync.after" + and e.path and e.path:match("%.uj$") then journal_sync = journal_sync or i end + if journal_sync and i > journal_sync and e.op == "fsync.after" + and e.path and e.path:match("/journals$") then directory_sync = directory_sync or i end + end + T.ok(blob_sync and journal_write and journal_sync and directory_sync, + "missing durability boundary in trace") + T.ok(blob_sync < journal_write and journal_write < journal_sync and journal_sync < directory_sync, + "commit order was not blob -> journal -> directory") + T.rm(dir) + end }, + + { "crash injection at every observed write, sync and rename boundary recovers a prefix", function() + local payload = transaction("boundary-" .. ("x"):rep(700)) + local probe = T.tmpdir("fault-probe") + local probe_store = storage.open(probe) + local events = {} + with_hook(function(op, detail) + if op:find("write", 1, true) or op:find("fsync", 1, true) or op:find("rename", 1, true) then + events[#events + 1] = { op = op, path = detail.path, from = detail.from, to = detail.to } + end + end, function() + probe_store:append(payload) + probe_store:rebuild_indexes() + end) + probe_store:close() + T.rm(probe) + T.ok(#events > 20, "fault harness did not observe all storage layers") + + for target = 1, #events do + local dir = T.tmpdir("fault-each") + local store = storage.open(dir) + local seen = 0 + posix.set_fault_hook(function(op) + if op:find("write", 1, true) or op:find("fsync", 1, true) or op:find("rename", 1, true) then + seen = seen + 1 + if seen == target then return "crash" end + end + end) + pcall(function() + store:append(payload) + store:rebuild_indexes() + end) + posix.set_fault_hook(nil) + store:close() + local reopened = storage.open(dir) + local records = reopened:records() + T.ok(#records == 0 or #records == 1, "non-prefix recovery at boundary " .. target) + if #records == 1 then T.eq(records[1].payload, payload, "committed record at boundary " .. target) end + reopened:close() + T.rm(dir) + end + end }, + + { "torn journal write is isolated and never becomes a committed record", function() + local dir = T.tmpdir("fault-torn") + local store = storage.open(dir) + local wrote = false + posix.set_fault_hook(function(op, detail) + if detail.path and detail.path:match("%.uj$") then + if op == "write.before" and not wrote then wrote = true; return { short = 7 } end + if op == "write.after" and wrote then return { error = "EIO torn write" } end + end + end) + T.raises(function() store:append(transaction("torn")) end, "EIO torn write") + posix.set_fault_hook(nil) + store:close() + local reopened, info = storage.open(dir) + T.eq(#reopened:records(), 0) + T.ok(info.tail_isolated) + reopened:close() + T.rm(dir) + end }, + + { "full disk before blob commit refuses without journal or in-memory mutation", function() + local dir = T.tmpdir("fault-full") + local store = storage.open(dir) + posix.set_fault_hook(function(op, detail) + if op == "write.before" and detail.path and detail.path:find("/blobs/", 1, true) then + return { error = "ENOSPC full disk" } + end + end) + T.raises(function() store:append(transaction("full disk")) end, "ENOSPC") + posix.set_fault_hook(nil) + T.eq(#store:records(), 0) + store:close() + local reopened = storage.open(dir) + T.eq(#reopened:records(), 0) + reopened:close() + T.rm(dir) + end }, +} diff --git a/unii/test/test_storage_phase2.lua b/unii/test/test_storage_phase2.lua new file mode 100644 index 0000000..edad686 --- /dev/null +++ b/unii/test/test_storage_phase2.lua @@ -0,0 +1,166 @@ +local T = require("unii.test.lib") +local storage = require("unii.host.storage") +local supervisor = require("unii.host.supervisor") +local codec = require("unii.host.codec") +local schema = require("unii.host.schema") +local sha256 = require("unii.host.sha256") +local posix = require("unii.host.posix") +local mock = require("unii.host.mock.summarizer") + +local function read(path) + local f = assert(io.open(path, "rb")) + local bytes = f:read("*a") + f:close() + return bytes +end + +local function write(path, bytes) + local f = assert(io.open(path, "wb")) + f:write(bytes) + f:close() +end + +local function open_sup(dir, extra) + local opts = { core = T.core(), provider = mock.new { cap = 512 } } + for k, v in pairs(extra or {}) do opts[k] = v end + return supervisor.open(dir, opts) +end + +return { + { "message and summary bodies are blobs; physical journal contains only hashes", function() + local dir = T.tmpdir("phase2-blobs") + local original = "private-body-" .. ("x"):rep(900) + local sup = open_sup(dir, { checkpoint_interval = 0 }) + sup:submit(T.msg(0, "user", original, "2026-10-09T23:59:59Z")) + sup:pump() + local hash = sha256.hex(original) + local final_hash = sup:state_hash() + sup:close() + + T.eq(read(dir .. "/blobs/" .. hash), original) + local shard = read(dir .. "/journals/2026-10-09.uj") + T.ok(not shard:find(original, 1, true), "message text leaked into the journal") + local records = storage.parse(shard, 2) + local saw_message, saw_summary = false, false + for _, record in ipairs(records) do + local txn = codec.decode(record.payload) + local ev = txn.v.event and txn.v.event.v + if ev and ev[1].v == "message-appended" then + saw_message = true + T.eq(#ev[5].v, 3) + T.eq(ev[5].v[3].v, hash) + elseif ev and ev[1].v == "summary-completed" then + saw_summary = true + T.eq(#ev, 5) + T.ok(read(dir .. "/blobs/" .. ev[5].v) ~= nil) + end + end + T.ok(saw_message and saw_summary) + local again = open_sup(dir, { checkpoint_interval = 0 }) + T.eq(again:state_hash(), final_hash) + again:close() + T.rm(dir) + end }, + + { "daily shards retain global order and rebuild all indexes", function() + local dir = T.tmpdir("phase2-shards") + local sup = open_sup(dir, { checkpoint_interval = 0 }) + sup:submit(T.msg(0, "user", "day one", "2026-10-09T23:59:59Z")) + sup:submit(T.msg(1, "assistant", "day two", "2026-10-10T00:00:01Z")) + local state_hash = sup:state_hash() + sup:close() + T.eq(read(dir .. "/journals/MANIFEST"), + "USM1\n0000-00-00\n2026-10-09\n2026-10-10\n") + for _, name in ipairs { "journal.idx", "messages.idx", "blobs.idx", "nodes.idx", "jobs.idx" } do + T.ok(#read(dir .. "/indexes/" .. name) >= 5, name) + end + T.ok(read(dir .. "/indexes/messages.idx"):find("0\t2\t", 1, true)) + T.ok(read(dir .. "/indexes/messages.idx"):find("1\t3\t", 1, true)) + local again = open_sup(dir, { checkpoint_interval = 0 }) + T.eq(again:state_hash(), state_hash) + T.eq(again.info.replayed_records, 2) + again:close() + T.rm(dir) + end }, + + { "latest valid checkpoint skips old records and preserves exact hashes", function() + local dir = T.tmpdir("phase2-checkpoint") + local sup = open_sup(dir, { checkpoint_interval = 4 }) + for i = 0, 11 do sup:submit(T.msg(i, "user", "checkpoint " .. i)) end + sup:checkpoint() + local state_hash, view, view_hash = sup:state_hash(), sup:view() + local last = sup.store.next_seq - 1 + sup:close() + + local again = open_sup(dir, { checkpoint_interval = 4 }) + T.eq(again.info.checkpoint_seq, last) + T.eq(again.info.replayed_records, 0) + T.eq(again:state_hash(), state_hash) + local view2, hash2 = again:view() + T.eq(view2, view); T.eq(hash2, view_hash) + again:close() + T.rm(dir) + end }, + + { "stale checkpoint is set aside and an older valid checkpoint is used", function() + local dir = T.tmpdir("phase2-stale") + local sup = open_sup(dir, { checkpoint_interval = 4 }) + for i = 0, 8 do sup:submit(T.msg(i, "user", "stale " .. i)) end + local expected = sup:state_hash() + sup:close() + + local store = storage.open(dir) + local candidates = store:checkpoint_candidates() + T.ok(#candidates > 0) + local cp = codec.decode(candidates[1].payload) + cp.v.anchor = codec.text(("0"):rep(64)) + store:write_checkpoint(store.next_seq - 1, codec.encode(cp)) + store:close() + + local again = open_sup(dir, { checkpoint_interval = 4 }) + T.ok(again.info.checkpoint_rejected and again.info.checkpoint_rejected:find("stale journal anchor", 1, true)) + T.ok(again.info.checkpoint_seq, "older checkpoint was not used") + T.eq(again:state_hash(), expected) + again:close() + T.rm(dir) + end }, + + { "missing or corrupted committed blobs refuse recovery without changing journal", function() + local dir = T.tmpdir("phase2-badblob") + local text = "blob source" + local hash = sha256.hex(text) + local sup = open_sup(dir, { checkpoint_interval = 0 }) + sup:submit(T.msg(0, "user", text)) + sup:close() + local journal_path = dir .. "/journals/2026-10-09.uj" + local journal = read(journal_path) + local blob_path = dir .. "/blobs/" .. hash + assert(os.rename(blob_path, blob_path .. ".missing")) + T.raises(function() storage.open(dir) end, "missing committed blob") + T.eq(read(journal_path), journal) + assert(os.rename(blob_path .. ".missing", blob_path)) + write(blob_path, "corrupt") + T.raises(function() storage.open(dir) end, "corrupted blob") + T.eq(read(journal_path), journal) + T.rm(dir) + end }, + + { "recovery streams shards and lazy records instead of reading journal files whole", function() + local dir = T.tmpdir("phase2-stream") + local store = storage.open(dir) + for i = 1, 200 do store:append(("record-%04d-"):format(i) .. ("z"):rep(2000)) end + store:close() + local real = posix.read_file + posix.read_file = function(path) + if path:match("%.uj$") then error("whole journal read attempted: " .. path, 0) end + return real(path) + end + local ok, reopened = pcall(storage.open, dir) + posix.read_file = real + if not ok then error(reopened, 0) end + T.eq(#reopened:records(), 200) + T.eq(reopened:records()[200].payload:sub(1, 11), "record-0200") + reopened:close() + T.rm(dir) + end }, +} From 32e9348358e06b65cbfe33f8631c947880eb0893 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 08:49:37 +0000 Subject: [PATCH 2/7] unii: record Phase 2 verification results Co-authored-by: pyrex41 --- unii/docs/HANDOFF.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index 45ddddc..82b45aa 100644 --- a/unii/docs/HANDOFF.md +++ b/unii/docs/HANDOFF.md @@ -103,7 +103,7 @@ and stops reporting SKIP. ## Acceptance gates and how each was verified Every row is an automated test in `luajit unii/test/run.lua`. The last full -run passed 72 tests, failed 0 and skipped 1. The skip is the oracle +run passed 82 tests, failed 0 and skipped 1. The skip is the oracle fixtures, which do not exist yet. That count includes the upstream port specs (1,088 checks across 22 specs) and the kernel suite (134 of 134). The suite passes on both LuaJIT builds listed in `MANIFEST.md`. From 4ff73ce549f26b41bc5c7b9e01df08b2d94cd6ee Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 08:51:57 +0000 Subject: [PATCH 3/7] unii: exercise checkpoint durability boundaries Co-authored-by: pyrex41 --- unii/docs/HANDOFF.md | 2 +- unii/docs/STORAGE_PHASE2.md | 11 ++++--- unii/test/test_storage_faults.lua | 53 ++++++++++++++++++++++++++++++- 3 files changed, 59 insertions(+), 7 deletions(-) diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index 82b45aa..9a4b555 100644 --- a/unii/docs/HANDOFF.md +++ b/unii/docs/HANDOFF.md @@ -103,7 +103,7 @@ and stops reporting SKIP. ## Acceptance gates and how each was verified Every row is an automated test in `luajit unii/test/run.lua`. The last full -run passed 82 tests, failed 0 and skipped 1. The skip is the oracle +run passed 83 tests, failed 0 and skipped 1. The skip is the oracle fixtures, which do not exist yet. That count includes the upstream port specs (1,088 checks across 22 specs) and the kernel suite (134 of 134). The suite passes on both LuaJIT builds listed in `MANIFEST.md`. diff --git a/unii/docs/STORAGE_PHASE2.md b/unii/docs/STORAGE_PHASE2.md index dd366a9..e238762 100644 --- a/unii/docs/STORAGE_PHASE2.md +++ b/unii/docs/STORAGE_PHASE2.md @@ -19,11 +19,12 @@ summary text; checkpoints and indexes can always be discarded and rebuilt. | ENOSPC/torn journal append | Refuse append; isolate partial final frame on restart | `test_storage_faults` | | Second owner | Refuse nonblocking OS `flock`; no PID-file inference | `test_storage` same- and cross-process cases | -The boundary sweep discovers every write, file/directory sync and rename -callback exercised by blob, shard-manifest, journal and index commits, then -repeats the append with a simulated process death at each callback. Recovery -must produce either zero records or the one complete transaction, never a -partial or invented transaction. +The boundary sweeps discover every write, file/directory sync and rename +callback exercised by blob, shard-manifest, journal, index and checkpoint +commits, then repeat the operation with a simulated process death at each +callback. Journal recovery must produce either zero records or the one +complete transaction, never a partial or invented transaction; checkpoint +recovery must preserve the journal-derived state. ## Checkpoint proof diff --git a/unii/test/test_storage_faults.lua b/unii/test/test_storage_faults.lua index 625c4e3..e8afc19 100644 --- a/unii/test/test_storage_faults.lua +++ b/unii/test/test_storage_faults.lua @@ -3,6 +3,8 @@ local storage = require("unii.host.storage") local posix = require("unii.host.posix") local codec = require("unii.host.codec") local schema = require("unii.host.schema") +local supervisor = require("unii.host.supervisor") +local mock = require("unii.host.mock.summarizer") local function transaction(text) return codec.encode(codec.map { @@ -89,6 +91,47 @@ return { end end }, + { "checkpoint crash injection at every boundary falls back to journal replay", function() + local function open_sup(dir) + return supervisor.open(dir, { + core = T.core(), provider = mock.new { cap = 512 }, checkpoint_interval = 0, + }) + end + local probe = T.tmpdir("fault-cp-probe") + local probe_sup = open_sup(probe) + probe_sup:submit(T.msg(0, "user", "checkpoint boundary")) + local events = {} + with_hook(function(op) + if op:find("write", 1, true) or op:find("fsync", 1, true) or op:find("rename", 1, true) then + events[#events + 1] = op + end + end, function() probe_sup:checkpoint() end) + probe_sup:close() + T.rm(probe) + T.ok(#events >= 8) + + for target = 1, #events do + local dir = T.tmpdir("fault-cp") + local sup = open_sup(dir) + sup:submit(T.msg(0, "user", "checkpoint boundary")) + local expected = sup:state_hash() + local seen = 0 + posix.set_fault_hook(function(op) + if op:find("write", 1, true) or op:find("fsync", 1, true) or op:find("rename", 1, true) then + seen = seen + 1 + if seen == target then return "crash" end + end + end) + pcall(sup.checkpoint, sup) + posix.set_fault_hook(nil) + sup:close() + local again = open_sup(dir) + T.eq(again:state_hash(), expected, "checkpoint boundary " .. target) + again:close() + T.rm(dir) + end + end }, + { "torn journal write is isolated and never becomes a committed record", function() local dir = T.tmpdir("fault-torn") local store = storage.open(dir) @@ -109,7 +152,7 @@ return { T.rm(dir) end }, - { "full disk before blob commit refuses without journal or in-memory mutation", function() + { "full disk during blob or journal commit refuses without in-memory mutation", function() local dir = T.tmpdir("fault-full") local store = storage.open(dir) posix.set_fault_hook(function(op, detail) @@ -120,6 +163,14 @@ return { T.raises(function() store:append(transaction("full disk")) end, "ENOSPC") posix.set_fault_hook(nil) T.eq(#store:records(), 0) + posix.set_fault_hook(function(op, detail) + if op == "write.before" and detail.path and detail.path:match("%.uj$") then + return { error = "ENOSPC full disk" } + end + end) + T.raises(function() store:append(transaction("journal full disk")) end, "ENOSPC") + posix.set_fault_hook(nil) + T.eq(#store:records(), 0) store:close() local reopened = storage.open(dir) T.eq(#reopened:records(), 0) From e28d1fc78ef06aec986b198e35b97cef47fd21a1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 08:54:55 +0000 Subject: [PATCH 4/7] unii: poison storage after uncertain append failures Co-authored-by: pyrex41 --- unii/docs/STORAGE_PHASE2.md | 4 ++++ unii/docs/contracts/storage.md | 5 +++++ unii/host/posix.lua | 2 ++ unii/host/storage.lua | 17 ++++++++++++++++- unii/host/supervisor.lua | 13 +++++++++++-- unii/test/test_storage.lua | 4 ++++ unii/test/test_storage_faults.lua | 6 ++++++ 7 files changed, 48 insertions(+), 3 deletions(-) diff --git a/unii/docs/STORAGE_PHASE2.md b/unii/docs/STORAGE_PHASE2.md index e238762..e928996 100644 --- a/unii/docs/STORAGE_PHASE2.md +++ b/unii/docs/STORAGE_PHASE2.md @@ -26,6 +26,10 @@ callback. Journal recovery must produce either zero records or the one complete transaction, never a partial or invented transaction; checkpoint recovery must preserve the journal-derived state. +After any mutating append error, the store handle is poisoned and refuses +later appends until it is closed and recovery reopens it. This prevents +sequence reuse or appending after an unknown partial frame. + ## Checkpoint proof A checkpoint binds: diff --git a/unii/docs/contracts/storage.md b/unii/docs/contracts/storage.md index 40d1e43..4ceb95e 100644 --- a/unii/docs/contracts/storage.md +++ b/unii/docs/contracts/storage.md @@ -115,6 +115,11 @@ Records 2 and later are `event` transactions: sync. Their failure cannot undo or compete with an authoritative journal commit. +Any failed mutating append poisons that store handle. It refuses every later +append until the process closes and reopens the chat, because the failure may +have left either a torn frame or a complete frame whose sync result is +uncertain. + ## Replay When the store is opened: diff --git a/unii/host/posix.lua b/unii/host/posix.lua index e7d31c3..884b83e 100644 --- a/unii/host/posix.lua +++ b/unii/host/posix.lua @@ -96,6 +96,8 @@ function M.write_all(fd, s) local w = tonumber(C.write(fd, buf + off, want)) if w < 0 then if ffi.errno() ~= EINTR then return nil, errstr("write", fd) end + elseif w == 0 then + return nil, "write(" .. tostring(fd) .. "): returned zero bytes" else off = off + w action = hit(op .. ".after", { fd = fd, path = fd_paths[tonumber(fd)], offset = off, bytes = w }) diff --git a/unii/host/storage.lua b/unii/host/storage.lua index 02038fb..381e692 100644 --- a/unii/host/storage.lua +++ b/unii/host/storage.lua @@ -334,7 +334,7 @@ function Store:iter_records(after) end end -function Store:append(payload) +local function append_unprotected(self, payload) if #payload > M.MAX_RECORD then error("storage: record over " .. M.MAX_RECORD .. " bytes", 0) end local physical, refs, day = physicalize(payload, self.blobs) ensure_shard(self, day) @@ -367,6 +367,21 @@ function Store:append(payload) return seq end +function Store:append(payload) + if self.poisoned then + error("storage is poisoned after a failed mutation; close and reopen before appending", 0) + end + local ok, result = pcall(append_unprotected, self, payload) + if not ok then + -- The failure may have happened after any prefix of a frame, or after a + -- complete frame reached the kernel but before its sync was confirmed. + -- Continuing could append after a torn frame or reuse a committed seq. + self.poisoned = true + error(result, 0) + end + return result +end + function Store:record_anchor(seq) local meta = self.meta[seq] return meta and meta.physical_hash or nil diff --git a/unii/host/supervisor.lua b/unii/host/supervisor.lua index ed24c01..f22e8b9 100644 --- a/unii/host/supervisor.lua +++ b/unii/host/supervisor.lua @@ -133,6 +133,10 @@ function Sup:_restore_checkpoint() end local state = codec.to_shen(cp.v.state) if self.core:state_hash(state) ~= text_field(cp, "state_hash") then error("state hash mismatch", 0) end + local invariant_errors = self.core:invariant_errors(state) + if #invariant_errors > 0 then + error("state invariants failed: " .. table.concat(invariant_errors, "; "), 0) + end local rendered = self.core:render(state) if sha256.hex(rendered) ~= text_field(cp, "view_hash") then error("view hash mismatch", 0) end if self.core:status(state).rev ~= checkpoint_int(cp, "view_rev") then error("view revision mismatch", 0) end @@ -212,9 +216,14 @@ function Sup:submit(event) local tagged = schema.encode("event", event) local state, out = self.core:transition(self.state, tagged) local prev_state, prev_cache = self.state, self.view_cache + local prev_text, prev_hash = self.view_text, self.view_hash self.state = state local ok, err = pcall(self._refresh_view, self) - if not ok then self.state, self.view_cache = prev_state, prev_cache; error(err, 0) end + if not ok then + self.state, self.view_cache = prev_state, prev_cache + self.view_text, self.view_hash = prev_text, prev_hash + error(err, 0) + end local aok, seq = pcall(self.store.append, self.store, txn_bytes { kind = codec.sym("event"), event = tagged, @@ -226,7 +235,7 @@ function Sup:submit(event) }) if not aok then self.state, self.view_cache = prev_state, prev_cache - self:_refresh_view() + self.view_text, self.view_hash = prev_text, prev_hash error(seq, 0) end self:_absorb(schema.decode("event", tagged), out, false) diff --git a/unii/test/test_storage.lua b/unii/test/test_storage.lua index a6533c5..07c5d1b 100644 --- a/unii/test/test_storage.lua +++ b/unii/test/test_storage.lua @@ -164,10 +164,14 @@ return { local sup = open_sup(dir) feed(sup, 0, 3) local before = sup:state_hash() + local before_view, before_view_hash = sup:view() local real = sup.store.append sup.store.append = function() error("storage append: disk full (injected)", 0) end T.raises(function() sup:submit(T.msg(3, "user", "lost")) end, "disk full") T.eq(sup:state_hash(), before) + local after_view, after_view_hash = sup:view() + T.eq(after_view, before_view) + T.eq(after_view_hash, before_view_hash) sup.store.append = real sup:submit(T.msg(3, "user", "kept")) local h = sup:state_hash() diff --git a/unii/test/test_storage_faults.lua b/unii/test/test_storage_faults.lua index e8afc19..8595323 100644 --- a/unii/test/test_storage_faults.lua +++ b/unii/test/test_storage_faults.lua @@ -144,6 +144,7 @@ return { end) T.raises(function() store:append(transaction("torn")) end, "EIO torn write") posix.set_fault_hook(nil) + T.raises(function() store:append(transaction("must reopen")) end, "storage is poisoned") store:close() local reopened, info = storage.open(dir) T.eq(#reopened:records(), 0) @@ -163,6 +164,10 @@ return { T.raises(function() store:append(transaction("full disk")) end, "ENOSPC") posix.set_fault_hook(nil) T.eq(#store:records(), 0) + T.raises(function() store:append(transaction("must reopen")) end, "storage is poisoned") + store:close() + + store = storage.open(dir) posix.set_fault_hook(function(op, detail) if op == "write.before" and detail.path and detail.path:match("%.uj$") then return { error = "ENOSPC full disk" } @@ -171,6 +176,7 @@ return { T.raises(function() store:append(transaction("journal full disk")) end, "ENOSPC") posix.set_fault_hook(nil) T.eq(#store:records(), 0) + T.raises(function() store:append(transaction("must reopen")) end, "storage is poisoned") store:close() local reopened = storage.open(dir) T.eq(#reopened:records(), 0) From 973cef0128ef2df8011e35372e38a115b3ceba71 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 10:17:22 +0000 Subject: [PATCH 5/7] unii: inspect secrets across sharded journals Co-authored-by: pyrex41 --- unii/test/test_network_real.lua | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/unii/test/test_network_real.lua b/unii/test/test_network_real.lua index bb25479..ab5b701 100644 --- a/unii/test/test_network_real.lua +++ b/unii/test/test_network_real.lua @@ -10,6 +10,19 @@ local chat = require("unii.host.providers.chat_completions") local server +local function journal_bytes(dir) + local manifest = assert(io.open(dir .. "/journals/MANIFEST", "rb")) + local names = manifest:read("*a") + manifest:close() + local out = {} + for day in names:gmatch("(%d%d%d%d%-%d%d%-%d%d)\n") do + local file = assert(io.open(dir .. "/journals/" .. day .. ".uj", "rb")) + out[#out + 1] = file:read("*a") + file:close() + end + return table.concat(out) +end + local function setup() if server then return server end local ok, network = pcall(require, "unii.host.network") @@ -115,7 +128,7 @@ return { out = run(("status --dir %q"):format(dir)) local job = out:match("stuck uncertain%s+(%S+)") T.ok(job, out) - local journal = io.open(dir .. "/journal.uj", "rb"):read("*a") + local journal = journal_bytes(dir) T.ok(not journal:find("test-key-not-secret", 1, true), "key not journaled") out, ok = run(("retry --dir %q --job %s --network real --url %q --model m") :format(dir, job, s.base .. "/v1/chat/completions")) From 575e217844a8870e1e3a44092af37b16f7c81e6d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 10:18:55 +0000 Subject: [PATCH 6/7] unii: record merged Phase 2 verification Co-authored-by: pyrex41 --- unii/docs/HANDOFF.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index f7c31d8..170997f 100644 --- a/unii/docs/HANDOFF.md +++ b/unii/docs/HANDOFF.md @@ -142,7 +142,7 @@ becomes uncertain and is not resent. ## Acceptance gates and how each was verified Every row is an automated test in `luajit unii/test/run.lua`. The last run -passed 84 tests, with 0 failed and 0 skipped. `--with-upstream` adds the +passed 98 tests, with 0 failed and 0 skipped. `--with-upstream` adds the pinned port specs (1,088 checks across 22 specs) and the kernel suite (134 of 134). From c9514936241154fc7be151792669fb140e80a87e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 9 Oct 2026 10:37:56 +0000 Subject: [PATCH 7/7] unii: record R1-R6 merged verification Co-authored-by: pyrex41 --- unii/docs/HANDOFF.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index f1d44a5..419be1d 100644 --- a/unii/docs/HANDOFF.md +++ b/unii/docs/HANDOFF.md @@ -145,7 +145,7 @@ is not resent. ## Acceptance gates and how each was verified Every row is an automated test in `luajit unii/test/run.lua`. The last run -passed 93 tests, with 0 failed and 0 skipped. `--with-upstream` adds the +passed 108 tests, with 0 failed and 0 skipped. `--with-upstream` adds the pinned port specs (1,088 checks across 22 specs) and the kernel suite (134 of 134).