Artifact 78e8a877e22afe537d29d289273d739a18f220497584b925492ed57331af27d9:
- File
src/sim/storage_sweep.zig
— part of check-in
[fde61b03b2]
at
2026-07-20 03:29:09
on branch h8-readrot
— Add online read-fault injection to the storage DST sweep
FaultyFile injected corruption only at crash/restart boundaries. Add a read_rot_pct knob: a read of an intact byte interns one offset in its span as permanently corrupt (payload-derived mask, never self-heals) -- a latent sector error surfacing at content() time on a store that opened clean, independent of any crash. A healthy disk draws nothing, so existing sweeps keep a byte-identical PRNG stream.
New storage_sweep.readRotSweep joins the dst gate: after a healthy open, every content-bearing hash reads back either its exact bytes or a declared corruption error -- wrong bytes presented as success is SilentCorruption. The append open path re-runs each frame's whole-frame CRC per read, so latent rot is detected and the gate stays green. The verified-on-open read path (openReadOnly) trusts the scan-time CRC on an immutable-length assumption that bit rot violates; a test drives that path to prove the oracle fires SilentCorruption.
Read-fault model follows TigerBeetle's storage-fault-injection lineage. (user: dmestas, size: 71296) [annotate] [blame] [check-ins using] [more...]
//! The DST sweep that drives the real `LogStore` (logstore.zig) through //! `FaultyFile`, asserting the four crash-recovery invariants //! (storage-track-plan.md §B). `run` is exercised by `zig build dst`, not //! a `zig build test` unit — the sweep IS the gate. //! //! The model tracks WHAT WAS WRITTEN (hash, bytes, and the store's own //! `size()` right after each successful append) rather than re-deriving //! frame-layout byte offsets itself: LogStore owns that arithmetic, and an //! independent oracle should observe outcomes through the public //! hash-addressed API, not reimplement wire format math that could share //! the same bug. const std = @import("std"); const storage = @import("zit").storage; const faulty_file = @import("faulty_file.zig"); const logstore = @import("zit").logstore; const delta = @import("zit").delta; const artifact = @import("zit").artifact; const container = @import("zit").container; const pack = @import("zit").pack; const hash_len = 32; // sha256 digest const Frame = struct { hash: [hash_len]u8, /// Owned UNCOMPRESSED content -- what `content()` must reconstruct for /// this frame regardless of its on-disk encoding (raw/zlib/delta). bytes: []u8, /// `store.size()` immediately before this put's frame was appended. frame_start: u64, /// `store.size()` immediately after -- the structural boundary a /// clean recovery of this frame implies. `frame_end - frame_start` /// includes trailing zero-pad; the pad is excluded from both crcs /// (logstore.zig's wire format), so it's inert and never read back -- /// a corrupt offset landing there is harmless, not silent corruption. frame_end: u64, /// On-disk payload length (`payloadLenOf`): the container/wire byte count, /// which differs from `bytes.len` for zlib/delta frames. Bounds the /// crc-covered span in the silent-corruption check. disk_payload_len: u32, /// Delta-chain depth: 0 for raw/zlib, base.depth+1 for a delta. Kept below /// `max_delta_chain` so every created chain stays resolvable. depth: u8, }; /// One config frame the sweep wrote, in append order. Stores the LOGICAL /// value (the config analog of a content Frame's `bytes`) so recovery is /// checked field-wise through `readConfig`, never by re-deriving wire bytes. const ConfigRec = struct { frame_start: u64, frame_end: u64, hash_policy: u8, branch: []u8, pid: []u8, }; /// Ground-truth bookkeeping: every content-bearing put in append order, /// how many of them are known-durable, every intern-only (phantom) hash /// (which must never appear on disk), and every config frame written (a /// distinguished singleton kind, not interned). const Model = struct { frames: std.ArrayList(Frame) = .empty, acked_count: usize = 0, phantoms: std.ArrayList([hash_len]u8) = .empty, configs: std.ArrayList(ConfigRec) = .empty, acked_config_count: usize = 0, fn deinit(self: *Model, gpa: std.mem.Allocator) void { for (self.frames.items) |f| gpa.free(f.bytes); self.frames.deinit(gpa); self.phantoms.deinit(gpa); for (self.configs.items) |c| { gpa.free(c.branch); gpa.free(c.pid); } self.configs.deinit(gpa); self.* = undefined; } fn recordConfig(self: *Model, gpa: std.mem.Allocator, frame_start: u64, frame_end: u64, cfg: logstore.Config) error{OutOfMemory}!void { const branch = try gpa.dupe(u8, cfg.default_branch); errdefer gpa.free(branch); const pid = try gpa.dupe(u8, cfg.project_id); errdefer gpa.free(pid); try self.configs.append(gpa, .{ .frame_start = frame_start, .frame_end = frame_end, .hash_policy = cfg.hash_policy, .branch = branch, .pid = pid }); } /// Drops every config frame past `n`, matching what a fresh recovery /// proved survived, and clamps the acked-config watermark to it. fn truncateConfigsTo(self: *Model, gpa: std.mem.Allocator, n: usize) void { std.debug.assert(n <= self.configs.items.len); for (self.configs.items[n..]) |c| { gpa.free(c.branch); gpa.free(c.pid); } self.configs.shrinkRetainingCapacity(n); self.acked_config_count = @min(self.acked_config_count, n); } fn recordPut(self: *Model, gpa: std.mem.Allocator, hash: [hash_len]u8, bytes: []const u8, frame_start: u64, frame_end: u64, disk_payload_len: u32, depth: u8) error{OutOfMemory}!void { const owned = try gpa.dupe(u8, bytes); errdefer gpa.free(owned); try self.frames.append(gpa, .{ .hash = hash, .bytes = owned, .frame_start = frame_start, .frame_end = frame_end, .disk_payload_len = disk_payload_len, .depth = depth }); } fn recordPhantom(self: *Model, gpa: std.mem.Allocator, hash: [hash_len]u8) error{OutOfMemory}!void { try self.phantoms.append(gpa, hash); } /// Drops the phantom at `idx` once it has gained content (order among /// phantoms is immaterial, so a swap-remove is exact and O(1)). fn removePhantomAt(self: *Model, idx: usize) void { std.debug.assert(idx < self.phantoms.items.len); _ = self.phantoms.swapRemove(idx); } /// Acked frames only grow, and only past a frame end the disk has /// actually promised durable. Both lists are append-ordered by /// `frame_end`, so each watermark advances independently against the same /// durable promise — this is the pre-recovery acked snapshot the config /// durability check reads (captured before `open` runs). fn recomputeAcked(self: *Model, durable_len: u64) void { while (self.acked_count < self.frames.items.len and self.frames.items[self.acked_count].frame_end <= durable_len) { self.acked_count += 1; } while (self.acked_config_count < self.configs.items.len and self.configs.items[self.acked_config_count].frame_end <= durable_len) { self.acked_config_count += 1; } } /// Drops every frame the log didn't actually keep, matching what a /// fresh recovery just proved survived. fn truncateTo(self: *Model, gpa: std.mem.Allocator, survived: usize) void { std.debug.assert(survived <= self.frames.items.len); for (self.frames.items[survived..]) |f| gpa.free(f.bytes); self.frames.shrinkRetainingCapacity(survived); self.acked_count = @min(self.acked_count, survived); } }; pub const SweepError = error{ OutOfMemory, Io, /// Invariant 1: recovery lost a frame the model had proof was durable. MissingAcknowledged, /// Invariant 1/2: a frame beyond the determined survived boundary is /// nonetheless present -- not a clean prefix -- or a phantom-only /// hash was found durable (phantoms must never touch disk). PhantomFrame, /// Invariant 2: the post-recovery structural size disagrees with the /// model's own record of where the last surviving frame ended. IndexMismatch, /// Invariant 1/4: a surviving frame's content or hash disagrees with /// what was written. HashMismatch, /// Invariant 4: a corrupt offset lies inside the recovered span -- /// validated data that isn't actually trustworthy. SilentCorruption, /// Config inv: recovery restored the wrong live config (or one at all /// when none survived, or none when one did). ConfigMismatch, }; fn sha256(payload: []const u8) [hash_len]u8 { var out: [hash_len]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(payload, &out, .{}); return out; } fn putOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { var buf: [64]u8 = undefined; const n = 1 + random.uintLessThan(usize, buf.len - 1); random.bytes(buf[0..n]); const payload = buf[0..n]; std.debug.assert(payload.len <= logstore.max_payload / 1000); // sweep payloads stay well under the 1 GiB bound const hash = sha256(payload); const before = store.size(); if (store.put(&hash, payload)) |rid| { const after = store.size(); // LogStore.put is idempotent for an already-content-bearing hash: // it appends nothing and returns the existing rid (self.end // unchanged). The sweep's small payload space (1..63 random // bytes) makes that collision real within a single run, so only // a put that actually landed a frame grows the model -- a // same-hash repeat is already covered by the frame recorded the // first time, and recording it again would fabricate a second // model frame the log never wrote. if (after != before) { try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0); } } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => {}, // torn/short write: LogStore already rolled back } } /// A zlib-encoded put: the store compresses `payload` into a fossil zlib /// container, but `content()` must still round-trip to the uncompressed /// bytes, so the model records `payload` (uncompressed) and the on-disk /// container length separately. fn zlibOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { var buf: [64]u8 = undefined; const n = 1 + random.uintLessThan(usize, buf.len - 1); random.bytes(buf[0..n]); const payload = buf[0..n]; std.debug.assert(payload.len <= logstore.max_payload / 1000); const hash = sha256(payload); const before = store.size(); if (store.putZlib(&hash, payload)) |rid| { const after = store.size(); if (after != before) { try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0); } } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => {}, } } /// A delta-encoded put over an existing content-bearing base whose chain has /// room (`depth < max_delta_chain`). The target shares a run with the base so /// the encoder emits a copy; `content()` must resolve the chain back to the /// target. Falls back to a raw put when no eligible base exists yet. fn deltaOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { if (model.frames.items.len == 0) return putOnce(gpa, store, model, random); const start = random.uintLessThan(usize, model.frames.items.len); var base_idx: ?usize = null; var k: usize = 0; while (k < model.frames.items.len) : (k += 1) { const idx = (start + k) % model.frames.items.len; if (model.frames.items[idx].depth < logstore.max_delta_chain) { base_idx = idx; break; } } const bi = base_idx orelse return putOnce(gpa, store, model, random); const base = &model.frames.items[bi]; const base_rid = store.lookup(&base.hash) orelse return; // recorded frame must be live const bn = base.bytes.len; std.debug.assert(bn >= 1); var tbuf: [64]u8 = undefined; @memcpy(tbuf[0..bn], base.bytes); tbuf[random.uintLessThan(usize, bn)] +%= 1; // one-byte mutation keeps a shared run const target = tbuf[0..bn]; const raw = delta.create(gpa, base.bytes, target) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.TargetTooLarge => unreachable, // bn <= 63, far below 2^32 }; defer gpa.free(raw); const hash = sha256(target); const depth = base.depth + 1; const before = store.size(); if (store.putDelta(&hash, base_rid, raw)) |rid| { const after = store.size(); if (after != before) { try model.recordPut(gpa, hash, target, before, after, store.payloadLenOf(rid), depth); } } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => {}, } } fn internOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { var seed_bytes: [16]u8 = undefined; random.bytes(&seed_bytes); const hash = sha256(&seed_bytes); _ = try store.intern(&hash); try model.recordPhantom(gpa, hash); } /// Content arrives for a hash the store already knows only as a phantom: the /// core ArtifactStore lifecycle (intern a hash, put its bytes later). Picks a /// live phantom so `put` takes the phantom-upgrade branch, then folds the /// upgraded frame into the durability model exactly like any other put. fn upgradeOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { std.debug.assert(model.phantoms.items.len > 0); // caller-guarded supply const idx = random.uintLessThan(usize, model.phantoms.items.len); const hash = model.phantoms.items[idx]; // live phantom -> hits put's upgrade branch var buf: [64]u8 = undefined; const n = 1 + random.uintLessThan(usize, buf.len - 1); random.bytes(buf[0..n]); const payload = buf[0..n]; std.debug.assert(payload.len <= logstore.max_payload / 1000); // Content-addressing is the caller's contract; the sweep only requires that // content() round-trips whatever bytes were put under this rid, so the // payload need not hash to `hash`. const before = store.size(); if (store.put(&hash, payload)) |rid| { const after = store.size(); std.debug.assert(after != before); // a phantom upgrade always appends its first frame try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0); model.removePhantomAt(idx); } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => {}, // torn write rolled back: entry stays a phantom in store and model } } /// Writes one bounded random config frame. A config write always appends a /// fresh frame (never idempotent), so a success must grow `store.size()`; a /// torn write rolls back in the store AND leaves the model unchanged, exactly /// like a content put. fn configOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void { var bbuf: [8]u8 = undefined; var pbuf: [8]u8 = undefined; const bn = random.uintAtMost(usize, bbuf.len); const pn = random.uintAtMost(usize, pbuf.len); random.bytes(bbuf[0..bn]); random.bytes(pbuf[0..pn]); const cfg: logstore.Config = .{ .hash_policy = random.int(u8), .default_branch = bbuf[0..bn], .project_id = pbuf[0..pn] }; const before = store.size(); if (store.putConfig(cfg)) |_| { const after = store.size(); std.debug.assert(after != before); // a config write is never idempotent try model.recordConfig(gpa, before, after, cfg); } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => {}, // torn write rolled back: config_offset unchanged in store and model } } /// Config-frame recovery checks, returning how many config frames survived. /// Recovery keeps a contiguous log prefix, and every frame with frame_end <= /// size validated during the scan that produced this store, so a surviving /// config is by construction intact. Two guarantees: (inv1 analog) the /// survivor count never falls below the pre-recovery acked-config snapshot, /// and (equality) the latest survivor is exactly what `readConfig` returns /// (or null when none survived), field-for-field. fn checkConfig(gpa: std.mem.Allocator, model: *const Model, store: *logstore.LogStore) SweepError!usize { var cfg_survived: usize = 0; while (cfg_survived < model.configs.items.len and model.configs.items[cfg_survived].frame_end <= store.size()) : (cfg_survived += 1) {} std.debug.assert(cfg_survived <= model.configs.items.len); if (cfg_survived < model.acked_config_count) return error.MissingAcknowledged; var read = store.readConfig(gpa) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => return error.ConfigMismatch, }; if (read) |*owned| { defer owned.deinit(gpa); if (cfg_survived == 0) return error.ConfigMismatch; // a config survived that shouldn't have const want = &model.configs.items[cfg_survived - 1]; if (owned.config.hash_policy != want.hash_policy or !std.mem.eql(u8, owned.config.default_branch, want.branch) or !std.mem.eql(u8, owned.config.project_id, want.pid)) { return error.ConfigMismatch; } } else if (cfg_survived != 0) { return error.ConfigMismatch; // a durable config was expected but none is live } return cfg_survived; } /// Determines the survived prefix (recovery only ever drops a suffix), /// then checks all four invariants against `model` and advances it to /// match reality. fn checkInvariants( gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model, store: *logstore.LogStore, guaranteed: usize, ) SweepError!void { const total = model.frames.items.len; var survived: usize = 0; while (survived < total and store.lookup(&model.frames.items[survived].hash) != null) : (survived += 1) {} if (survived < guaranteed) return error.MissingAcknowledged; var i: usize = survived; while (i < total) : (i += 1) { if (store.lookup(&model.frames.items[i].hash) != null) return error.PhantomFrame; } i = 0; while (i < survived) : (i += 1) { const f = &model.frames.items[i]; const rid = store.lookup(&f.hash) orelse return error.MissingAcknowledged; if (!std.mem.eql(u8, store.hashOf(rid), &f.hash)) return error.HashMismatch; const bytes = store.content(gpa, rid) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => return error.HashMismatch, } orelse return error.HashMismatch; defer gpa.free(bytes); if (!std.mem.eql(u8, bytes, f.bytes)) return error.HashMismatch; } for (model.phantoms.items) |h| { if (store.lookup(&h) != null) return error.PhantomFrame; } // A reopen wiped every in-memory phantom (phantoms never touch disk, §C), // so the live phantom set resets each crash. model.phantoms.clearRetainingCapacity(); const cfg_survived = try checkConfig(gpa, model, store); // Size: whichever kind holds the last surviving frame ends exactly at // store.size(); the other end is <= it, so @max recovers the true size. // With no configs, cfg_end == header_len and this reduces to the content // size exactly. const content_end: u64 = if (survived == 0) logstore.header_len else model.frames.items[survived - 1].frame_end; const cfg_end: u64 = if (cfg_survived == 0) logstore.header_len else model.configs.items[cfg_survived - 1].frame_end; const want_size: u64 = @max(content_end, cfg_end); if (store.size() != want_size) return error.IndexMismatch; // No corrupt offset may lie within a surviving frame's MEANINGFUL span // (its crc-covered bytes, 22+hash_len+payload_len from frame_start) -- // that would mean validated data isn't actually trustworthy. Trailing // zero-pad up to the 8-byte boundary is explicitly excluded from both // crcs (logstore.zig) and never read back, so corruption landing only // in pad is inert, not silent corruption. var j: usize = 0; while (j < survived) : (j += 1) { const f = &model.frames.items[j]; // The crc-covered span is the fixed header plus the ON-DISK payload // (container length for zlib/delta), not the uncompressed content. const meaningful_end = f.frame_start + 22 + @as(u64, f.hash.len) + f.disk_payload_len; for (faulty.corrupt.items) |off| { if (off >= f.frame_start and off < meaningful_end) return error.SilentCorruption; } } model.truncateTo(gpa, survived); model.truncateConfigsTo(gpa, cfg_survived); } /// A corrupt offset is only ever the payload-derived bitflip fault /// (storage-track-plan.md §B) -- nothing else in the simulator flips /// bits. Repair by truncating below the earliest one: a header-region hit /// (< header_len) makes the whole log untrustworthy (nothing downstream /// of a bad header can be validated), so that wipes to empty; a /// frame-region hit keeps every frame that ended before it. fn repairAfterCorrupt(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model) SweepError!void { std.debug.assert(faulty.corrupt.items.len > 0); var min_off: u64 = faulty.corrupt.items[0]; for (faulty.corrupt.items) |o| { if (o < min_off) min_off = o; } if (min_off < logstore.header_len) { faulty.storageFile().truncate(0) catch return error.Io; model.truncateTo(gpa, 0); model.truncateConfigsTo(gpa, 0); return; } var safe: usize = 0; while (safe < model.frames.items.len and model.frames.items[safe].frame_end <= min_off) : (safe += 1) {} const cut: u64 = if (safe == 0) logstore.header_len else model.frames.items[safe - 1].frame_end; faulty.storageFile().truncate(cut) catch return error.Io; model.truncateTo(gpa, safe); var cfg_safe: usize = 0; while (cfg_safe < model.configs.items.len and model.configs.items[cfg_safe].frame_end <= cut) : (cfg_safe += 1) {} model.truncateConfigsTo(gpa, cfg_safe); } /// Simulates power loss, reopens, and checks the four invariants, /// repairing (and rechecking) once if the open itself reports corruption. /// Returns the freshly opened store, which the caller now owns. fn crashCheckOnce(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model) SweepError!logstore.LogStore { faulty.crashRestart(); model.recomputeAcked(faulty.durable_len); var opened = logstore.LogStore.open(gpa, faulty.storageFile(), 0); if (opened) |*store| { checkInvariants(gpa, faulty, model, store, model.acked_count) catch |e| { store.deinit(gpa); return e; }; return store.*; } else |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => {}, } std.debug.assert(faulty.corrupt.items.len > 0); try repairAfterCorrupt(gpa, faulty, model); var store2 = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, }; checkInvariants(gpa, faulty, model, &store2, model.acked_count) catch |e| { store2.deinit(gpa); return e; }; return store2; } /// The two-phase sweep (mirrors dst.zig's repostate phases): phase 1 /// drives random puts/interns/syncs/crashes with faults live, checking /// the four recovery invariants after every crash; phase 2 heals the /// disk and proves one final clean crash+recover converges on the full /// model. pub fn run(gpa: std.mem.Allocator, seed: u64, steps: usize) SweepError!void { var prng = std.Random.DefaultPrng.init(seed); const random = prng.random(); var faulty = faulty_file.FaultyFile.init(gpa, random); defer faulty.deinit(); var model: Model = .{}; defer model.deinit(gpa); // The very first open runs against a clean disk (cfg is still the // all-zero, fault-free default) -- the four invariants are about // *recovering* from a crash, not about whether the log can even be // created under a lossy first write, and LogStore.open has no // retry/tolerance path for that (nor should it: a fresh empty file // is never "acknowledged" data to lose). Faults engage only once the // store exists. var store = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, }; faulty.cfg = .{ .short_write_pct = 8, .lost_tail_sync_pct = 8, .crash_cut_at_durable_pct = 15, .crash_rot_pct = 6, .sector_round = 512, .sector_round_pct = 25, }; const crash_every: usize = 41; var step: usize = 0; while (step < steps) : (step += 1) { switch (random.uintLessThan(u8, 7)) { 0, 1 => putOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, 2 => zlibOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, 3 => deltaOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, 4 => internOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, // Upgrade a live phantom when one exists, else keep the phantom // supply flowing so the branch has fuel next time. 5 => if (model.phantoms.items.len > 0) upgradeOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; } else internOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, 6 => configOnce(gpa, &store, &model, random) catch |e| { store.deinit(gpa); return e; }, else => unreachable, // uintLessThan(u8, 7) yields 0..6 } if (random.boolean()) { store.commit() catch {}; model.recomputeAcked(faulty.durable_len); } if (step % crash_every == crash_every - 1) { store.deinit(gpa); store = crashCheckOnce(gpa, &faulty, &model) catch |e| return e; } } store.deinit(gpa); // Phase 2 (liveness): faults healed, everything becomes durable, and a // harmless crash (nothing left unsynced) must recover the full model. faulty.cfg = .{}; { var healed = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, }; healed.commit() catch {}; healed.deinit(gpa); } model.recomputeAcked(faulty.durable_len); std.debug.assert(model.acked_count == model.frames.items.len); std.debug.assert(model.acked_config_count == model.configs.items.len); var final_store = try crashCheckOnce(gpa, &faulty, &model); final_store.deinit(gpa); std.debug.assert(model.acked_count == model.frames.items.len); std.debug.assert(model.acked_config_count == model.configs.items.len); } /// The most artifacts one equivalence run generates. Bounds the corpus (and /// so every allocation and loop below) to a fixed window regardless of seed. const eq_max_artifacts: usize = 64; /// Base-artifact length floor. `>= delta_min_source` (logstore, 50) so a base /// is never rejected by pickBase's cheap length early-out, and comfortably /// larger than `eq_stamp` so the shared prefix a derived artifact keeps is /// always non-trivial — a profitable (<75%) delta is then guaranteed. const eq_min_len: usize = 64; /// Base-artifact length ceiling. const eq_max_len: usize = 256; /// Trailing bytes each artifact stamps with its own index: makes every /// artifact's content (and therefore its hash) distinct, and is the ONLY /// region a derived artifact differs from its base in — the guaranteed-tiny /// delta the non-vacuity assertion relies on. const eq_stamp: usize = 8; /// One artifact in an equivalence corpus: its content-addressed hash, owned /// uncompressed bytes, the index of the earlier artifact it was derived from /// (`null` ⇒ a fresh random base), and the rid it landed at in each store. const EqArtifact = struct { hash: [hash_len]u8, bytes: []u8, base: ?usize, naive_rid: logstore.Rid = 0, pick_rid: logstore.Rid = 0, }; /// A fault-free in-memory backend + an opened store on it — the healthy-disk /// precondition both equivalence stores depend on (no fault ever perturbs a /// pure decode comparison). fn eqOpen(gpa: std.mem.Allocator, backend: *faulty_file.FaultyFile) SweepError!logstore.LogStore { std.debug.assert(std.meta.eql(backend.cfg, faulty_file.FaultCfg{})); return logstore.LogStore.open(gpa, backend.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage, error.Corrupt => return error.Io, }; } /// Step-2 (§5) write-path equivalence: the delta-base policy is a pure /// compaction choice, never a content change. The SAME seeded artifact stream /// is stored two ways — a NAIVE store (every artifact full-form via `putZlib`, /// depth 0) and a pickBase store (`putArtifact`'s best-of-N window, which /// stores derived artifacts as deltas) — and every hash must decode /// byte-identically in both, hence identical to the bytes that went in. Because /// every repostate view (leaf set, branch-of, current-trunk hash) is a pure /// function of decoded artifact bytes, byte-identical decode across the two /// stores is the strictly STRONGER condition that SUBSUMES repostate-view /// identity: identical decoded manifests derive an identical trunk hash — so /// this asserts the content half directly and the view half by construction. /// Seeded off `seed` with its own PRNG stream (disjoint from `run`'s), so a /// failure replays from the shared `reproduce with seed=N` line. /// /// The pack-redeltify form of this property (naive store vs a re-deltified /// COPY, and repostate views recomputed from both) belongs to the pack pass /// in step 3 — `packLog` does not yet route through `pickBase`. This is that /// property's write-path analog and exercises the identical `pickBase` policy /// module, so step 3 extends rather than replaces it. pub fn pickBaseEquivalence(gpa: std.mem.Allocator, seed: u64) SweepError!void { // Disjoint from run()'s stream (which inits straight off `seed`) so the two // subsystems never share entropy — the same discipline run() uses. var prng = std.Random.DefaultPrng.init(seed +% 0x9E37_79B9_7F4A_7C15); const random = prng.random(); var items: std.ArrayList(EqArtifact) = .empty; defer { for (items.items) |it| gpa.free(it.bytes); items.deinit(gpa); } // Corpus: 2..eq_max_artifacts artifacts. Once at least one exists, ~half // are derived from an earlier one (base content + a fresh trailing stamp) // so the best-of-N delta path is actually taken; the rest are fresh random // bases. The trailing stamp makes every artifact distinct. const count = 2 + random.uintLessThan(usize, eq_max_artifacts - 1); std.debug.assert(count >= 2 and count <= eq_max_artifacts); var i: usize = 0; while (i < count) : (i += 1) { const base: ?usize = if (i > 0 and random.boolean()) random.uintLessThan(usize, i) else null; // A derived artifact copies its base's length so the two differ ONLY in // the trailing stamp (guaranteed tiny delta); a base draws a random // length in [eq_min_len, eq_max_len]. const len: usize = if (base) |b| items.items[b].bytes.len else eq_min_len + random.uintLessThan(usize, eq_max_len - eq_min_len + 1); std.debug.assert(len >= eq_min_len and len <= eq_max_len); const bytes = try gpa.alloc(u8, len); errdefer gpa.free(bytes); if (base) |b| { @memcpy(bytes, items.items[b].bytes); } else { random.bytes(bytes); } // Unique per-artifact tail ⇒ distinct content ⇒ distinct sha256 hash, // so no put ever hits the idempotent short-circuit for a DIFFERENT // artifact. `count <= 64` fits u64; `len > eq_stamp` by the floor above. std.mem.writeInt(u64, bytes[len - eq_stamp ..][0..eq_stamp], @intCast(i), .little); try items.append(gpa, .{ .hash = sha256(bytes), .bytes = bytes, .base = base }); } var naive_backend = faulty_file.FaultyFile.init(gpa, random); defer naive_backend.deinit(); var naive = try eqOpen(gpa, &naive_backend); defer naive.deinit(gpa); var pick_backend = faulty_file.FaultyFile.init(gpa, random); defer pick_backend.deinit(); var pick = try eqOpen(gpa, &pick_backend); defer pick.deinit(gpa); // Store the identical stream into both. Naive: always full form. Pick: the // lineage hint (a derived artifact's base) plus putArtifact's own chrono // window drive the best-of-N choice. for (items.items) |*it| { it.naive_rid = naive.putZlib(&it.hash, it.bytes) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, }; const hint: ?logstore.Rid = if (it.base) |b| items.items[b].pick_rid else null; it.pick_rid = pick.putArtifact(&it.hash, it.bytes, hint) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, }; // Non-vacuity: a derived artifact differs from its base only in the // trailing stamp, so its delta is far under the 75% bar against a // content-bearing, under-cap base — the write path MUST store it as a // delta (depth grows), while the naive store keeps every frame full. if (it.base != null) { std.debug.assert(pick.depthOf(it.pick_rid) >= 1); std.debug.assert(naive.depthOf(it.naive_rid) == 0); } } // The invariant: storage is a cache of the log. Every artifact decodes to // exactly its input bytes in BOTH stores, whatever encoding each chose. for (items.items) |*it| { const nb = eqDecode(gpa, &naive, it.naive_rid) catch |e| return e; defer gpa.free(nb); const pb = eqDecode(gpa, &pick, it.pick_rid) catch |e| return e; defer gpa.free(pb); if (!std.mem.eql(u8, nb, it.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, pb, it.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, nb, pb)) return error.HashMismatch; } } /// `content` for the equivalence sweep with its error set mapped into /// `SweepError`: a null (phantom) or corrupt decode of a frame we just wrote /// is itself the failure the property is asserting against. fn eqDecode(gpa: std.mem.Allocator, store: *logstore.LogStore, rid: logstore.Rid) SweepError![]u8 { return (store.content(gpa, rid) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => return error.HashMismatch, }) orelse error.HashMismatch; } test "storage §5: pickBase vs naive at-rest decode equivalence (smoke)" { // A fixed seed so the smoke case replays exactly under `zig build test`; // `zig build dst` sweeps this across the gate seeds. try pickBaseEquivalence(std.testing.allocator, 0x5E_5EED); } // --------------------------------------------------------------------------- // Step-3 (§6.1/§7.2-§7.3) pack-redeltify equivalence: the REAL `zit.pack.packLog` // transform, swept across seeds. A source store built the production ingest way // (verbatim wire deltas over deliberately POOR bases via `putDeltaContainer`, // plus check-in manifests carrying F/P lineage) is packed; the packed copy must // decode byte-identically to a naive full-form store of the same corpus for // EVERY artifact — the strictly stronger condition that subsumes repostate-view // identity (leaf set, branch-of, current-trunk hash) since every such view is a // pure function of decoded artifact bytes, and the trunk tip is one of those // artifacts. Unlike `pickBaseEquivalence` (the write-path analog) this drives // the whole-store rewrite, exercising `deriveLineageHints`, the wire-delta // re-basing, the §3 post-write verify sweep, and config carry-over. // --------------------------------------------------------------------------- /// The most check-ins one pack-equivalence run generates (bounds every loop and /// allocation below regardless of seed). const eq_pack_max_checkins: usize = 8; /// Unrelated "noise" full-form blobs seeded first, used as the DELIBERATELY poor /// bases a source wire delta names — the sender's suboptimal choice pack must /// discard in favor of the same-path predecessor. const eq_pack_noise_blobs: usize = 2; /// One artifact in a pack-equivalence corpus: content-addressed sha3-256 key and /// owned uncompressed bytes (the ground truth every store must decode to). const EqPackArt = struct { key: [hash_len]u8, bytes: []u8, }; /// sha3-256 of `payload` — the digest `artifact.verify` recomputes from a 64-hex /// name (`nameAlgo(64) == .sha3_256`), so every corpus artifact MUST be keyed by /// it for `packLog`'s §7.7 verify to pass. fn sha3(payload: []const u8) [hash_len]u8 { var out: [hash_len]u8 = undefined; std.crypto.hash.sha3.Sha3_256.hash(payload, &out, .{}); return out; } /// Serialize a one-file check-in manifest (`file_name` → `blob_name64`, optional /// primary parent `parent_name`, root carries a propagating `branch=trunk` T /// card) — the sim analog of main.zig's `packSeedCheckin`, so `deriveLineageHints` /// re-parses it identically. Returns owned bytes; only OOM escapes. fn eqPackCheckin( gpa: std.mem.Allocator, ts: []const u8, parent_name: ?[]const u8, file_name: []const u8, blob_name64: []const u8, ) SweepError![]u8 { var pbuf: [1]artifact.PCard = undefined; var p_slice: []const artifact.PCard = &.{}; if (parent_name) |pn| { pbuf[0] = .{ .parents = pn, .count = 1 }; p_slice = pbuf[0..1]; } const fcards = [_]artifact.FCard{ .{ .filename = file_name, .hash = blob_name64, .perm = null, .old_name = null, .mode = .regular }, }; const root_tags = [_]artifact.TCard{ .{ .sign = .propagate, .name = "branch", .target = null, .value = "trunk" }, }; var m: artifact.Manifest = .{ .arena_state = std.heap.ArenaAllocator.init(gpa), .kind = .manifest, .a = null, .b = null, .c = .{ .comment = "c" }, .d = .{ .timestamp = ts }, .e = null, .g = null, .h = null, .i = null, .k = null, .l = null, .n = null, .r = null, .u = .{ .login = "u" }, .z = null, .w = null, .f = &fcards, .j = &.{}, .m = &.{}, .p = p_slice, .q = &.{}, .t = if (parent_name == null) &root_tags else &.{}, }; defer m.deinit(); return artifact.create(&m, gpa) catch |e| switch (e) { error.OutOfMemory => error.OutOfMemory, }; } /// A near-identical evolving blob: a fixed 128-line body (shared verbatim across /// the whole chain, so consecutive versions delta to a fraction) plus a /// per-`version` trailer that makes every version's content — and therefore its /// sha3 key — distinct. Caller owns the result. fn eqPackEvolvingBlob(gpa: std.mem.Allocator, version: usize) SweepError![]u8 { var buf: std.ArrayList(u8) = .empty; errdefer buf.deinit(gpa); var i: usize = 0; while (i < 128) : (i += 1) buf.print(gpa, "line {d} of the tracked file\n", .{i}) catch return error.OutOfMemory; buf.print(gpa, "trailer version {d}\n", .{version}) catch return error.OutOfMemory; return buf.toOwnedSlice(gpa) catch return error.OutOfMemory; } /// Store `bytes` full-form (zlib) under its sha3 key into `store`, record it in /// `arts` as ground truth (transferring ownership of `bytes`), and return the /// rid. On any failure `bytes` is freed so the caller never double-owns it. fn eqPackKeepFull( gpa: std.mem.Allocator, store: *logstore.LogStore, arts: *std.ArrayList(EqPackArt), key: [hash_len]u8, bytes: []u8, ) SweepError!logstore.Rid { errdefer gpa.free(bytes); const rid = store.putZlib(&key, bytes) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, }; try arts.append(gpa, .{ .key = key, .bytes = bytes }); return rid; } /// Store `bytes` full-form under its sha3 key into `store` WITHOUT recording it /// (the corpus ground truth is recorded once, when the same content lands in the /// naive store). Returns the rid. Used for the source store's full-form frames. fn eqPackPutFull(store: *logstore.LogStore, key: [hash_len]u8, bytes: []const u8) SweepError!logstore.Rid { return store.putZlib(&key, bytes) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, }; } /// Store `bytes` into the SOURCE store as a verbatim wire delta against the poor /// base `poor_base_rid` (`putDeltaContainer`, the retained peer form pack must /// re-base). `bytes` is left owned by the caller. Only memory/storage escapes. fn eqPackPutWireDelta( gpa: std.mem.Allocator, store: *logstore.LogStore, key: [hash_len]u8, bytes: []const u8, poor_base_bytes: []const u8, poor_base_rid: logstore.Rid, ) SweepError!void { const raw = delta.create(gpa, poor_base_bytes, bytes) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.TargetTooLarge => unreachable, // corpus blobs are far below 2^32 }; defer gpa.free(raw); const cont = container.encode(gpa, raw, .default) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.InputTooLarge => unreachable, // a bounded corpus delta always encodes }; defer gpa.free(cont); _ = store.putDeltaContainer(&key, poor_base_rid, cont) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, }; } /// Step-3 (§6.1/§7.2-§7.3) pack-redeltify equivalence. Builds one seeded /// production-shaped repository — noise blobs, then a check-in chain over a /// single evolving file whose blobs are stored in a per-seed mix of full form, /// local deltas, and verbatim wire deltas over the poor noise bases — into both /// a NAIVE full-form store (the independent oracle + ground truth) and a SOURCE /// store, then packs the source with the REAL `zit.pack.packLog`. Every artifact /// (blobs AND manifests, including the trunk tip) must decode byte-identically /// across naive, source, and packed. Disjoint PRNG stream off `seed`, so a /// failure replays from the shared `reproduce with seed=N` line. pub fn packRedeltifyEquivalence(gpa: std.mem.Allocator, seed: u64) SweepError!void { // Disjoint from run()/pickBaseEquivalence()'s streams (different odd offset). var prng = std.Random.DefaultPrng.init(seed +% 0xD1B5_4A32_D192_ED03); const random = prng.random(); var arts: std.ArrayList(EqPackArt) = .empty; defer { for (arts.items) |a| gpa.free(a.bytes); arts.deinit(gpa); } var naive_backend = faulty_file.FaultyFile.init(gpa, random); defer naive_backend.deinit(); var naive = try eqOpen(gpa, &naive_backend); defer naive.deinit(gpa); var src_backend = faulty_file.FaultyFile.init(gpa, random); defer src_backend.deinit(); var src = try eqOpen(gpa, &src_backend); defer src.deinit(gpa); // The live config survives pack as exactly one frame; put it into both. const cfg: logstore.Config = .{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" }; naive.putConfig(cfg) catch |e| return cfgPutErr(e); src.putConfig(cfg) catch |e| return cfgPutErr(e); // Noise blobs first (lowest offset), full-form in both — the poor bases a // source wire delta will name. Track each one's SOURCE rid + bytes. var noise_rid: [eq_pack_noise_blobs]logstore.Rid = undefined; var noise_idx: [eq_pack_noise_blobs]usize = undefined; // index into `arts` { var n: usize = 0; while (n < eq_pack_noise_blobs) : (n += 1) { var nb: std.ArrayList(u8) = .empty; errdefer nb.deinit(gpa); var j: usize = 0; const rows = 60 + random.uintLessThan(usize, 80); while (j < rows) : (j += 1) nb.print(gpa, "noise {d} row {d}\n", .{ n, j }) catch return error.OutOfMemory; const nbytes = nb.toOwnedSlice(gpa) catch return error.OutOfMemory; const key = sha3(nbytes); noise_rid[n] = try eqPackPutFull(&src, key, nbytes); noise_idx[n] = arts.items.len; _ = try eqPackKeepFull(gpa, &naive, &arts, key, nbytes); } } // A check-in chain over one evolving file. Each version's blob lands in the // source in a per-seed encoding; i==1 is forced to a wire delta so every run // exercises the re-basing path. Manifests are always full form in both. const nc = 2 + random.uintLessThan(usize, eq_pack_max_checkins - 1); std.debug.assert(nc >= 2 and nc <= eq_pack_max_checkins); var prev_name: [64]u8 = undefined; var have_prev = false; var prev_blob_src_rid: logstore.Rid = 0; var i: usize = 0; while (i < nc) : (i += 1) { // Blob: recorded as ground truth in `arts` via the naive store. const bbytes = try eqPackEvolvingBlob(gpa, i); const bkey = sha3(bbytes); var bname64: [64]u8 = undefined; _ = artifact.artifactName(.sha3_256, bbytes, &bname64); _ = try eqPackKeepFull(gpa, &naive, &arts, bkey, bbytes); // `arts` now owns bbytes // Source encoding of the blob (variant forced to wire-delta at i==1). const variant: u8 = if (i == 1) 0 else random.uintLessThan(u8, 3); var blob_src_rid: logstore.Rid = undefined; if (i == 0) { blob_src_rid = try eqPackPutFull(&src, bkey, bbytes); } else switch (variant) { 0 => { // A verbatim wire delta over a poor noise base (same blob for // bytes AND rid, so the delta reconstructs). Pack must discard // this base in favor of the same-path predecessor blob. const pick = random.uintLessThan(usize, eq_pack_noise_blobs); try eqPackPutWireDelta(gpa, &src, bkey, bbytes, arts.items[noise_idx[pick]].bytes, noise_rid[pick]); blob_src_rid = src.lookup(&bkey).?; }, 1 => { // Local delta against the previous version's blob (the ideal base). blob_src_rid = src.putArtifact(&bkey, bbytes, prev_blob_src_rid) catch |e| return cfgPutErr(e); }, else => { blob_src_rid = try eqPackPutFull(&src, bkey, bbytes); }, } prev_blob_src_rid = blob_src_rid; // Manifest snapshotting f.txt -> this blob, parented to the previous. var ts_buf: [32]u8 = undefined; const ts = std.fmt.bufPrint(&ts_buf, "2020-01-{d:0>2}T00:00:00", .{i + 1}) catch unreachable; const parent: ?[]const u8 = if (have_prev) prev_name[0..] else null; const mbytes = try eqPackCheckin(gpa, ts, parent, "f.txt", &bname64); const mkey = sha3(mbytes); _ = try eqPackPutFull(&src, mkey, mbytes); _ = try eqPackKeepFull(gpa, &naive, &arts, mkey, mbytes); // `arts` now owns mbytes _ = artifact.artifactName(.sha3_256, mbytes, &prev_name); have_prev = true; } // `prev_name` now names the trunk tip check-in: the current-trunk hash. const trunk_key = sha3Name(&prev_name); naive.commit() catch return error.Io; src.commit() catch return error.Io; // Pack the source with the REAL transform into a fresh fault-free store. var packed_backend = faulty_file.FaultyFile.init(gpa, random); defer packed_backend.deinit(); var packed_store = try eqOpen(gpa, &packed_backend); defer packed_store.deinit(gpa); const stats = pack.packLog(gpa, &src, &packed_store) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => return error.SilentCorruption, // a pack that fails §7.7 round-trip }; // Every non-phantom source frame is repacked (no drop, no fabrication). if (stats.artifacts != arts.items.len) return error.IndexMismatch; // The invariant: storage is a cache of the log. Every artifact decodes to // exactly its input bytes in the naive oracle, the source, AND the packed // copy — so the pack-re-based chains reconstruct byte-identical content. for (arts.items) |a| { const nb = try eqDecodeKey(gpa, &naive, &a.key); defer gpa.free(nb); const sb = try eqDecodeKey(gpa, &src, &a.key); defer gpa.free(sb); const pb = try eqDecodeKey(gpa, &packed_store, &a.key); defer gpa.free(pb); if (!std.mem.eql(u8, nb, a.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, sb, a.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, pb, a.bytes)) return error.HashMismatch; } // Current-trunk hash identity, made concrete: the trunk tip check-in is // present and decodes identically in the naive oracle and the packed copy — // and since every manifest in the corpus decodes byte-identically (asserted // above), any repostate view derived over the packed store equals the one // over the naive store by construction. A missing tip is a dropped trunk. if (naive.lookup(&trunk_key) == null) return error.MissingAcknowledged; if (packed_store.lookup(&trunk_key) == null) return error.MissingAcknowledged; // Config carried through the pack, field-for-field. var soc = (src.readConfig(gpa) catch return error.ConfigMismatch) orelse return error.ConfigMismatch; defer soc.deinit(gpa); var doc = (packed_store.readConfig(gpa) catch return error.ConfigMismatch) orelse return error.ConfigMismatch; defer doc.deinit(gpa); if (soc.config.hash_policy != doc.config.hash_policy or !std.mem.eql(u8, soc.config.default_branch, doc.config.default_branch) or !std.mem.eql(u8, soc.config.project_id, doc.config.project_id)) { return error.ConfigMismatch; } } /// The 32-byte binary digest a 64-hex artifact name decodes to (the log-frame /// key). The name is `artifactName` output, always valid 64-hex, so the decode /// never fails — an assert guards the invariant. fn sha3Name(name: *const [64]u8) [hash_len]u8 { var key: [hash_len]u8 = undefined; _ = std.fmt.hexToBytes(&key, name) catch unreachable; return key; } /// `content` by hash key with pack's decode error set mapped into `SweepError`: /// a null (phantom) or corrupt decode of a frame we just wrote is itself the /// failure the property asserts against. fn eqDecodeKey(gpa: std.mem.Allocator, store: *logstore.LogStore, key: []const u8) SweepError![]u8 { const rid = store.lookup(key) orelse return error.MissingAcknowledged; return eqDecode(gpa, store, rid); } /// `putConfig`/`putArtifact` failure narrowed to the sweep's error set. fn cfgPutErr(e: logstore.PutError) SweepError { return switch (e) { error.OutOfMemory => error.OutOfMemory, error.Storage => error.Io, }; } test "storage §6.1: pack-redeltify vs naive at-rest decode equivalence + trunk preserved (smoke)" { // Fixed seed for `zig build test`; `zig build dst` sweeps it across the gate. try packRedeltifyEquivalence(std.testing.allocator, 0x9AC5_5EED); } // --------------------------------------------------------------------------- // Step-4 (§6.2/§7.2-§7.3) same-path pack equivalence: `packRedeltifyEquivalence`'s // SINGLE evolving file makes a blob's same-path predecessor coincide with its // chronological neighbor, so the §6.2 manifest-walk lane runs but is never the // only path to the ideal base. This property builds MULTIPLE tracked files whose // versions are interleaved by round-robin single-file check-ins: between two // versions of one file sit the other files' check-ins (≥ 2·(files-1) frames), // pushing the earlier version OUTSIDE the bounded chronological window of the // later one. Pack can then reach the ideal base ONLY through `deriveCandidateHints`'s // same-path walk — the exact §6.2 mechanism. The equivalence bar is unchanged: // every artifact (blobs AND manifests, incl. the trunk tip) decodes byte-identical // across a naive full-form store, the mixed-encoding source, and the repacked copy // — the strictly stronger condition that subsumes repostate-view identity. // --------------------------------------------------------------------------- /// Tracked-file / versions-per-file bounds for one same-path equivalence run /// (bounds every loop and allocation regardless of seed). `min_files` is above /// `max_base_candidates`+1 so a file's earlier version is reliably beyond the /// chronological window of its later one — the out-of-window interleave the /// property exists to stress. const eq_sp_min_files: usize = 4; const eq_sp_max_files: usize = 6; const eq_sp_min_versions: usize = 2; const eq_sp_max_versions: usize = 4; /// A near-identical evolving blob for tracked file `file` at `version`: a /// file-tagged first line (so distinct files never alias to one content hash) + /// a fixed 128-line shared body (so consecutive versions of ONE file delta to a /// fraction) + a per-(file,version) trailer (so every version's content — and /// its sha3 key — is distinct). Caller owns the result. fn eqPackFileBlob(gpa: std.mem.Allocator, file: usize, version: usize) SweepError![]u8 { var buf: std.ArrayList(u8) = .empty; errdefer buf.deinit(gpa); buf.print(gpa, "tracked file {d}\n", .{file}) catch return error.OutOfMemory; var i: usize = 0; while (i < 128) : (i += 1) buf.print(gpa, "line {d} of the tracked file\n", .{i}) catch return error.OutOfMemory; buf.print(gpa, "file {d} trailer version {d}\n", .{ file, version }) catch return error.OutOfMemory; return buf.toOwnedSlice(gpa) catch return error.OutOfMemory; } /// Step-4 (§6.2/§7.2-§7.3) same-path pack-redeltify equivalence. Builds one /// seeded multi-file repository — noise blobs (poor bases), then `files` tracked /// files each evolving over `versions` interleaved single-file check-ins whose /// blobs land in a per-seed mix of full form, local deltas, and verbatim wire /// deltas over the poor noise bases — into both a NAIVE full-form store (oracle + /// ground truth) and a SOURCE store, then packs the source with the REAL /// `zit.pack.packLog`. Every artifact must decode byte-identically across naive, /// source, and packed. Disjoint PRNG stream off `seed`, so a failure replays from /// the shared `reproduce with seed=N` line. pub fn packSamePathEquivalence(gpa: std.mem.Allocator, seed: u64) SweepError!void { // A fourth distinct odd offset — disjoint from run()/pickBaseEquivalence()/ // packRedeltifyEquivalence()'s streams, so no subsystem shares entropy. var prng = std.Random.DefaultPrng.init(seed +% 0x2545_F491_4F6C_DD1D); const random = prng.random(); var arts: std.ArrayList(EqPackArt) = .empty; defer { for (arts.items) |a| gpa.free(a.bytes); arts.deinit(gpa); } var naive_backend = faulty_file.FaultyFile.init(gpa, random); defer naive_backend.deinit(); var naive = try eqOpen(gpa, &naive_backend); defer naive.deinit(gpa); var src_backend = faulty_file.FaultyFile.init(gpa, random); defer src_backend.deinit(); var src = try eqOpen(gpa, &src_backend); defer src.deinit(gpa); // The live config survives pack as exactly one frame; put it into both. const cfg: logstore.Config = .{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" }; naive.putConfig(cfg) catch |e| return cfgPutErr(e); src.putConfig(cfg) catch |e| return cfgPutErr(e); // Poor bases first (lowest offset), full-form in both — the bases a source // wire delta names and pack must discard for the same-path predecessor. var noise_rid: [eq_pack_noise_blobs]logstore.Rid = undefined; var noise_idx: [eq_pack_noise_blobs]usize = undefined; // index into `arts` { var n: usize = 0; while (n < eq_pack_noise_blobs) : (n += 1) { var nb: std.ArrayList(u8) = .empty; errdefer nb.deinit(gpa); var j: usize = 0; const rows = 60 + random.uintLessThan(usize, 80); while (j < rows) : (j += 1) nb.print(gpa, "noise {d} row {d}\n", .{ n, j }) catch return error.OutOfMemory; const nbytes = nb.toOwnedSlice(gpa) catch return error.OutOfMemory; const key = sha3(nbytes); noise_rid[n] = try eqPackPutFull(&src, key, nbytes); noise_idx[n] = arts.items.len; _ = try eqPackKeepFull(gpa, &naive, &arts, key, nbytes); } } const nfiles = eq_sp_min_files + random.uintLessThan(usize, eq_sp_max_files - eq_sp_min_files + 1); const nvers = eq_sp_min_versions + random.uintLessThan(usize, eq_sp_max_versions - eq_sp_min_versions + 1); std.debug.assert(nfiles >= eq_sp_min_files and nfiles <= eq_sp_max_files); std.debug.assert(nvers >= eq_sp_min_versions and nvers <= eq_sp_max_versions); // Per-file bookkeeping: the tracked filename bytes and the source rid of the // last stored version (the ideal local-delta base). One linear check-in chain // spans ALL files round-robin, so `prev_name` threads the whole history and a // file's versions are separated by the other files' check-ins. var fname_buf: [eq_sp_max_files][16]u8 = undefined; var prev_blob_rid: [eq_sp_max_files]logstore.Rid = undefined; var prev_name: [64]u8 = undefined; var have_prev = false; var ci: usize = 0; // global check-in index — a valid, monotone, distinct day var forced_wire_done = false; var version: usize = 0; while (version < nvers) : (version += 1) { var file: usize = 0; while (file < nfiles) : (file += 1) { const fname = std.fmt.bufPrint(&fname_buf[file], "file{d}.txt", .{file}) catch unreachable; // Blob ground truth (naive owns bbytes once recorded). const bbytes = try eqPackFileBlob(gpa, file, version); const bkey = sha3(bbytes); var bname64: [64]u8 = undefined; _ = artifact.artifactName(.sha3_256, bbytes, &bname64); _ = try eqPackKeepFull(gpa, &naive, &arts, bkey, bbytes); // Source encoding. Version 0 of each file is FULL (a clean re-delta // target); a later version takes a per-seed form, and file 0's first // later version is forced to a wire delta so every run exercises the // re-basing path over a poor base. var blob_src_rid: logstore.Rid = undefined; if (version == 0) { blob_src_rid = try eqPackPutFull(&src, bkey, bbytes); } else { const force_wire = !forced_wire_done and file == 0; const variant: u8 = if (force_wire) 0 else random.uintLessThan(u8, 3); if (force_wire) forced_wire_done = true; switch (variant) { 0 => { const pick = random.uintLessThan(usize, eq_pack_noise_blobs); try eqPackPutWireDelta(gpa, &src, bkey, bbytes, arts.items[noise_idx[pick]].bytes, noise_rid[pick]); blob_src_rid = src.lookup(&bkey).?; }, 1 => blob_src_rid = src.putArtifact(&bkey, bbytes, prev_blob_rid[file]) catch |e| return cfgPutErr(e), else => blob_src_rid = try eqPackPutFull(&src, bkey, bbytes), } } prev_blob_rid[file] = blob_src_rid; // Manifest snapshotting `fname` -> this blob, parented to the previous // check-in in the shared linear chain. Full form in both stores. var ts_buf: [32]u8 = undefined; const ts = std.fmt.bufPrint(&ts_buf, "2020-01-{d:0>2}T00:00:00", .{ci + 1}) catch unreachable; const parent: ?[]const u8 = if (have_prev) prev_name[0..] else null; const mbytes = try eqPackCheckin(gpa, ts, parent, fname, &bname64); const mkey = sha3(mbytes); _ = try eqPackPutFull(&src, mkey, mbytes); _ = try eqPackKeepFull(gpa, &naive, &arts, mkey, mbytes); _ = artifact.artifactName(.sha3_256, mbytes, &prev_name); have_prev = true; ci += 1; // `ci <= eq_sp_max_files * eq_sp_max_versions == 24` fits a two-digit // day (01..24), so the timestamp is always a valid distinct date. std.debug.assert(ci <= eq_sp_max_files * eq_sp_max_versions); } } // `prev_name` now names the trunk tip check-in: the current-trunk hash. const trunk_key = sha3Name(&prev_name); naive.commit() catch return error.Io; src.commit() catch return error.Io; // Pack the source with the REAL transform into a fresh fault-free store. var packed_backend = faulty_file.FaultyFile.init(gpa, random); defer packed_backend.deinit(); var packed_store = try eqOpen(gpa, &packed_backend); defer packed_store.deinit(gpa); const stats = pack.packLog(gpa, &src, &packed_store) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => return error.Io, error.Corrupt => return error.SilentCorruption, // a pack that fails §7.7 round-trip }; // Every non-phantom source frame is repacked (no drop, no fabrication). if (stats.artifacts != arts.items.len) return error.IndexMismatch; // The invariant: storage is a cache of the log. Every artifact decodes to // exactly its input bytes in the naive oracle, the source, AND the packed // copy — so the pack-re-based (same-path-chosen) chains reconstruct // byte-identical content. for (arts.items) |a| { const nb = try eqDecodeKey(gpa, &naive, &a.key); defer gpa.free(nb); const sb = try eqDecodeKey(gpa, &src, &a.key); defer gpa.free(sb); const pb = try eqDecodeKey(gpa, &packed_store, &a.key); defer gpa.free(pb); if (!std.mem.eql(u8, nb, a.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, sb, a.bytes)) return error.HashMismatch; if (!std.mem.eql(u8, pb, a.bytes)) return error.HashMismatch; } // Current-trunk hash identity: the trunk tip check-in is present and decodes // identically in the naive oracle and the packed copy (asserted above), and // since every manifest decodes byte-identically, any repostate view over the // packed store equals the one over the naive store by construction. if (naive.lookup(&trunk_key) == null) return error.MissingAcknowledged; if (packed_store.lookup(&trunk_key) == null) return error.MissingAcknowledged; // Config carried through the pack, field-for-field. var soc = (src.readConfig(gpa) catch return error.ConfigMismatch) orelse return error.ConfigMismatch; defer soc.deinit(gpa); var doc = (packed_store.readConfig(gpa) catch return error.ConfigMismatch) orelse return error.ConfigMismatch; defer doc.deinit(gpa); if (soc.config.hash_policy != doc.config.hash_policy or !std.mem.eql(u8, soc.config.default_branch, doc.config.default_branch) or !std.mem.eql(u8, soc.config.project_id, doc.config.project_id)) { return error.ConfigMismatch; } } test "storage §6.2: same-path (multi-file interleaved) pack-redeltify vs naive decode equivalence + trunk preserved (smoke)" { // Fixed seed for `zig build test`; `zig build dst` sweeps it across the gate. try packSamePathEquivalence(std.testing.allocator, 0x5A3E_5EED); } // --------------------------------------------------------------------------- // H8 (TEST-HARDENING §2 H8) online read-fault injection: latent sector errors // (`read_rot_pct`) surfacing at content()-time on a store that OPENED CLEAN — // the read-path dimension FaultyFile's crash-boundary faults never reached. // Follows TigerBeetle's storage-fault-injection lineage: a read of rotted data // must be DETECTED, never silently served. The oracle: after a healthy open, // every content-bearing hash reads back either exactly the model's bytes or a // DECLARED corruption error — wrong bytes presented as success is // `SilentCorruption`. The append open path (`open`, verified_on_open == false) // re-runs each frame's whole-frame CRC on every read, so it detects the rot and // the gate stays green; the read-only path (`openReadOnly`, verified_on_open == // true) trusts the scan-time CRC on the immutable-length assumption latent rot // violates — the documented trust boundary the prove-it-can-fail test drives // via `trust_on_read`. // --------------------------------------------------------------------------- /// The most content frames one read-rot corpus holds (bounds every loop and /// allocation regardless of seed). const read_rot_max_frames: usize = 24; /// The armed latent-rot probability the sweep reads under. Moderate on purpose: /// high enough that meaningful-span rot lands often across the corpus, low /// enough that a frame's header CRC often survives so the payload read is the /// byte actually under test. const read_rot_read_pct: u8 = 30; /// Builds a bounded healthy corpus (raw/zlib/delta content frames, no faults) /// into a fresh store, commits it durable, reopens it CLEAN, then arms /// `read_rot_pct` and reads every known hash. `trust_on_read` selects the reopen /// path: `open` (append, re-verifies each frame per read) vs `openReadOnly` /// (trusts the scan-time CRC). Disjoint PRNG stream off `seed`, so a failure /// replays from the shared reproduce line. fn readRotCorpus(gpa: std.mem.Allocator, seed: u64, trust_on_read: bool) SweepError!void { // A fifth distinct odd offset — disjoint from run()/pickBase/packRedeltify/ // packSamePath's streams, so no subsystem shares entropy. var prng = std.Random.DefaultPrng.init(seed +% 0x8EBC_6AF0_9C88_C6E1); const random = prng.random(); var faulty = faulty_file.FaultyFile.init(gpa, random); // cfg = {} : healthy disk defer faulty.deinit(); var model: Model = .{}; defer model.deinit(gpa); // Healthy corpus: `open` writes the header, every put runs fault-free, so // the store is byte-clean when committed durable. { var store = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, }; // A raw/zlib/delta mix exercises every content() decode path. Raw frames // are the silent vector under trust-on-read (no checksum inside the // payload), so the draw weights them at half. const count = 2 + random.uintLessThan(usize, read_rot_max_frames - 1); var i: usize = 0; while (i < count) : (i += 1) { (switch (random.uintLessThan(u8, 4)) { 0, 1 => putOnce(gpa, &store, &model, random), 2 => zlibOnce(gpa, &store, &model, random), 3 => deltaOnce(gpa, &store, &model, random), else => unreachable, }) catch |e| { store.deinit(gpa); return e; }; } store.commit() catch { store.deinit(gpa); return error.Io; }; store.deinit(gpa); } // Reopen CLEAN (faults still off, so the open-time scan is fault-free), then // arm the latent sector error: bit rot that surfaces only now, after the // store proved healthy at open. var store = if (trust_on_read) logstore.LogStore.openReadOnly(gpa, faulty.storageFile()) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, } else logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, else => return error.Io, }; defer store.deinit(gpa); faulty.cfg.read_rot_pct = read_rot_read_pct; // The oracle: every content-bearing hash reads back either its exact model // bytes or a DECLARED corruption error. Wrong bytes presented as success is // silent corruption — the read path served rot it should have caught. for (model.frames.items) |*f| { const rid = store.lookup(&f.hash) orelse return error.MissingAcknowledged; const got = store.content(gpa, rid) catch |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.Storage => continue, // a declared device error is honest error.Corrupt => continue, // detection fired — the product } orelse return error.MissingAcknowledged; // a recorded frame is never phantom defer gpa.free(got); if (!std.mem.eql(u8, got, f.bytes)) return error.SilentCorruption; } } /// H8 gate sweep: reads through the append open path, which re-verifies each /// frame's whole-frame CRC on every read — so latent read-rot is always detected /// (`error.Corrupt`), never served as wrong bytes. Joins `zig build dst`. pub fn readRotSweep(gpa: std.mem.Allocator, seed: u64) SweepError!void { return readRotCorpus(gpa, seed, false); } test "storage §B(H8): read-rot on the append read path is detected, never served (smoke)" { // Fixed seed for `zig build test`; `zig build dst` sweeps it across the gate. try readRotSweep(std.testing.allocator, 0x5EED_5A70); } test "storage §B(H8): prove-it-can-fail — the verified-on-open read path serves latent rot silently" { // Skipping the per-read whole-frame CRC (what `openReadOnly` does, on the // immutable-length assumption latent bit rot violates) makes the read path // trust rotted bytes. Across a fixed seed range at least one corpus lands a // payload rot on a raw frame whose header survived — content() then returns // wrong bytes as success, which the oracle catches as SilentCorruption. The // seed sweep keeps the demonstration deterministic and non-flaky. var seed: u64 = 1; var fired = false; while (seed <= 256) : (seed += 1) { readRotCorpus(std.testing.allocator, seed, true) catch |e| { if (e == error.SilentCorruption) { fired = true; break; } // Any other error is not the demonstration; keep scanning. }; } try std.testing.expect(fired); }