diff --git a/unii/docs/HANDOFF.md b/unii/docs/HANDOFF.md index 01e5163..419be1d 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 @@ -74,7 +74,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, adapter-result classification network/ REAL libcurl adapter (PR #75; its README covers TLS, logging, limits) @@ -132,14 +135,17 @@ The oracle's canonical-text function and the byte-hysteresis fixture were changed to the reconciled policy. The oracle remains a separate implementation. -**Phase 2 storage.** It is being built on a branch off this one. The -`host/storage.lua` interface is unchanged. The dispatch record is an -ordinary payload, so a new backend needs nothing extra. +**Phase 2 storage.** Content-addressed blobs, daily shards, checkpoints, +indexes and streaming replay are integrated behind the unchanged +`host/storage.lua` interface. Checkpoints preserve outstanding commands and +their dispatch intents, complete round/candidate state and provisional view +lines. An in-flight command restored from a checkpoint 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 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). @@ -199,6 +205,20 @@ 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. It also covers +hash-only `summary-uncertain` raw content and checkpoint restoration of the +round's best candidate and provisional line. + +`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 @@ -270,11 +290,9 @@ been reported upstream from this branch. * 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 @@ -287,13 +305,9 @@ 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. Run the chat-completions provider against a real endpoint with a key, +1. Run the chat-completions provider against a real endpoint with a key, and replace the provisional prompt with the plan's (Phase 3). -3. Add timers as host events (`TimerObserved`) for retry backoff. -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 +2. Add timers as host events (`TimerObserved`) for retry backoff. +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..e928996 --- /dev/null +++ b/unii/docs/STORAGE_PHASE2.md @@ -0,0 +1,55 @@ +# 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 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. + +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: + +* 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 6633b4a..ae22865 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 text, summary output and the raw leaf carried + by `summary-uncertain` are absent from physical frames. Their 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,11 @@ 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, pending commands and their dispatch intents. + A bad or stale checkpoint is renamed `.invalid-*`; recovery tries an older + one, then record one. ## Transaction payloads @@ -97,13 +119,23 @@ storage interface is unchanged: this is one more payload through `append`. ## 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. + +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 @@ -113,7 +145,10 @@ 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 (including summary + round candidates and provisional lines), view, pending commands, dispatch + intents 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 @@ -129,15 +164,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..884b83e 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,50 @@ 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 + 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 }) + 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 +138,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 +154,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 +194,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 +220,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..c7aa8f9 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,253 @@ 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 + +local function physicalize_content(content, blobs, refs, what) + 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(what .. " blob declaration mismatch", 0) end + blobs:put(text, hash) + content[4] = nil + refs[#refs + 1] = { hash = hash, bytes = bytes } + end +end + +local function hydrate_content(content, blobs) + if content and #content == 3 then + content[4] = codec.text(blobs:get(content[3].v, int(content[2]))) + end +end + +local function reference_content(content, refs) + if content and #content == 3 then + refs[#refs + 1] = { hash = content[3].v, bytes = int(content[2]) } + end +end + +-- Journal transactions omit the text field from message, summary and raw +-- uncertain-leaf content records. Their 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 + physicalize_content(content, blobs, refs, "message") + 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 } + elseif ev[1].v == "summary-uncertain" and ev[4] and ev[4].t == "list" then + for _, tagged in ipairs(ev[4].v) do + physicalize_content(tagged.v, blobs, refs, "uncertain raw") + end + 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 + hydrate_content(content, blobs) + elseif ev[1].v == "summary-completed" and #ev == 5 then + ev[6] = codec.text(blobs:get(ev[5].v, int(ev[4]))) + elseif ev[1].v == "summary-uncertain" and ev[4] and ev[4].t == "list" then + for _, tagged in ipairs(ev[4].v) do hydrate_content(tagged.v, blobs) end + 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 + reference_content(content, refs) + elseif ev[1].v == "summary-completed" and #ev == 5 then + refs[#refs + 1] = { hash = ev[5].v, bytes = int(ev[4]) } + elseif ev[1].v == "summary-uncertain" and ev[4] and ev[4].t == "list" then + for _, tagged in ipairs(ev[4].v) do reference_content(tagged.v, refs) end + 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 } + elseif ev and ev[1].v == "summary-uncertain" then + idx.jobs[ev[2].v] = { status = "uncertain", seq = seq } + end + if txn.v.kind and txn.v.kind.t == "sym" and txn.v.kind.v == "dispatch" + and txn.v.job and txn.v.job.t == "text" then + idx.jobs[txn.v.job.v] = { status = "dispatched", 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 } + elseif d.t == "list" and d.v[1] and d.v[1].v == "job-uncertain" then + idx.jobs[d.v[2].v] = { status = "uncertain", 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 +327,112 @@ 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) - 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 + 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 - 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:append(payload) +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 + +local function append_unprotected(self, 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: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 +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 1bc1dcd..c20e451 100644 --- a/unii/host/supervisor.lua +++ b/unii/host/supervisor.lua @@ -47,6 +47,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 = {}, intent = {}, 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 @@ -95,7 +96,83 @@ 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 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 + 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 + if not cp.v.intents or cp.v.intents.t ~= "list" then error("missing dispatch intents", 0) end + local intent = {} + for _, tagged in ipairs(cp.v.intents.v) do + if tagged.t ~= "text" or not outstanding[tagged.v] then + error("checkpoint has invalid dispatch intent", 0) + end + intent[tagged.v] = true + end + self.state, self.outstanding, self.started, self.intent = state, outstanding, {}, intent + 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_dispatch(rec, txn) @@ -137,7 +214,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" or ev._ == "summary-uncertain" then for cmd, c in pairs(self.outstanding) do @@ -159,9 +241,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, @@ -173,13 +260,50 @@ 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) + self:_maybe_checkpoint(seq) 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 intents = {} + for cmd in pairs(self.intent) do intents[#intents + 1] = codec.text(cmd) end + table.sort(intents, function(a, b) return a.v < b.v 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), + intents = codec.list(intents), + } + return self.store:write_checkpoint(seq, payload) +end + +function Sup:_maybe_checkpoint(seq) + if self.checkpoint_interval > 0 and seq % self.checkpoint_interval == 0 then + local ok, err = pcall(self.checkpoint, self, seq) + if not ok then self.info.checkpoint_error = tostring(err) end + end +end + local function by_cmd(a, b) return tonumber(a.cmd:sub(2)) < tonumber(b.cmd:sub(2)) end -- Outstanding commands a previous process started and never resolved. @@ -228,8 +352,11 @@ function Sup:dispatch_pending() end table.sort(cmds, by_cmd) for _, c in ipairs(cmds) do - self.store:append(txn_bytes { kind = codec.sym("dispatch"), cmd = codec.text(c.cmd), job = codec.text(c.job) }) + local seq = self.store:append(txn_bytes { + kind = codec.sym("dispatch"), cmd = codec.text(c.cmd), job = codec.text(c.job), + }) self.intent[c.cmd] = true + self:_maybe_checkpoint(seq) self.started[c.cmd] = true local source if c.input._ == "leaf-input" then 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 ae88a21..c5f32f0 100644 --- a/unii/test/run.lua +++ b/unii/test/run.lua @@ -31,7 +31,9 @@ end local files = { "test_codec", "test_boundary", "test_tree", "test_view", "test_transition", - "test_traces", "test_hysteresis", "test_oracle", "test_storage", "test_network_mock", "test_network_real", "test_milestone", + "test_traces", "test_hysteresis", "test_oracle", "test_storage", + "test_storage_phase2", "test_storage_faults", "test_network_mock", + "test_network_real", "test_milestone", } local passed, failed, skipped = 0, 0, 0 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")) diff --git a/unii/test/test_storage.lua b/unii/test/test_storage.lua index 050c6ea..97f9040 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 @@ -162,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() @@ -178,24 +184,27 @@ return { { "tampered records, foreign bundles and changed config are refused", function() local dir = T.tmpdir("sup") - local sup = open_sup(dir) + local sup = open_sup(dir, { checkpoint_interval = 0 }) 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..8595323 --- /dev/null +++ b/unii/test/test_storage_faults.lua @@ -0,0 +1,186 @@ +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 supervisor = require("unii.host.supervisor") +local mock = require("unii.host.mock.summarizer") + +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 }, + + { "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) + 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) + 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) + T.ok(info.tail_isolated) + reopened:close() + T.rm(dir) + end }, + + { "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) + 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) + 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" } + end + end) + 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) + 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..5af7150 --- /dev/null +++ b/unii/test/test_storage_phase2.lua @@ -0,0 +1,253 @@ +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 }, + + { "dispatch intent restored from checkpoint becomes uncertain and is never resent", function() + local dir = T.tmpdir("phase2-dispatch-checkpoint") + local first = mock.new { cap = 512 } + local sup = supervisor.open(dir, { + core = T.core(), provider = first, checkpoint_interval = 3, + config = schema.config { max_inflight = 1 }, + }) + sup:submit(T.msg(0, "user", ("in flight "):rep(100))) + T.eq(sup:dispatch_pending(), 1) + T.eq(first.calls, 1) + T.eq(#first.network.hits, 0, "provider has not been stepped") + T.eq(sup.store.next_seq - 1, 3, "init, event, dispatch") + T.eq(sup.info.checkpoint_error, nil) + sup:close() -- crash after the dispatch record/checkpoint, before any outcome + + local second = mock.new { cap = 512 } + sup = supervisor.open(dir, { + core = T.core(), provider = second, checkpoint_interval = 3, + }) + T.eq(sup.info.checkpoint_seq, 3) + T.eq(sup.info.replayed_records, 0) + T.eq(#sup:orphans(), 1, "dispatch intent survived checkpoint restore") + sup:pump() + T.eq(second.calls, 0, "uncertain command was not sent to the provider") + T.eq(#second.network.hits, 0) + T.eq(sup:status().uncertain, 1) + T.eq(sup:outstanding_count(), 0) + sup:close() + T.ok(read(dir .. "/indexes/jobs.idx"):find("\tuncertain\t", 1, true)) + T.rm(dir) + end }, + + { "uncertain leaf round state and provisional line restore from checkpoint", function() + local dir = T.tmpdir("phase2-provisional-checkpoint") + local raw = "raw provisional source " .. ("r"):rep(700) + local first = mock.new { + cap = 512, + fixture = { ["0/0"] = { + [1] = { text = "short candidate from try one" }, + [2] = { uncertain = true }, + } }, + } + local sup = supervisor.open(dir, { + core = T.core(), provider = first, checkpoint_interval = 0, + }) + sup:submit(T.msg(0, "user", raw)) + sup:pump() + T.eq(first.calls, 2) + T.eq(sup:status().uncertain, 1) + T.eq(sup:status().provisional, 1) + local state_hash, view, view_hash = sup:state_hash(), sup:view() + T.ok(view:find("short candidate from try one", 1, true), view) + local last = sup.store.next_seq - 1 + sup:checkpoint(last) + sup:close() + + local shard = read(dir .. "/journals/2026-10-09.uj") + T.ok(not shard:find(raw, 1, true), "uncertain raw text leaked into journal") + local saw_uncertain = false + for _, record in ipairs(storage.parse(shard, 2)) do + local txn = codec.decode(record.payload) + local ev = txn.v.event and txn.v.event.v + if ev and ev[1].v == "summary-uncertain" then + saw_uncertain = true + T.eq(#ev[4].v, 1) + T.eq(#ev[4].v[1].v, 3, "raw content is a hash-only blob reference") + end + end + T.ok(saw_uncertain) + + local second = mock.new { cap = 512 } + sup = supervisor.open(dir, { + core = T.core(), provider = second, checkpoint_interval = 0, + }) + T.eq(sup.info.checkpoint_seq, last) + T.eq(sup.info.replayed_records, 0) + T.eq(sup:state_hash(), state_hash) + local restored, restored_hash = sup:view() + T.eq(restored, view) + T.eq(restored_hash, view_hash) + T.eq(sup:status().uncertain, 1) + T.eq(sup:status().provisional, 1) + T.eq(second.calls, 0, "checkpointed uncertain job was not resent") + sup: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 }, +}