Skip to content
zit Artifact [97d7205161]
Login

Artifact [97d7205161]

Check-ins Using Download Hex

Artifact 97d7205161011af689aba27ba80c5280c9c347c45ec19068b82850d357d2772a:

//! 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, pack.Filter.keep_all, &.{}) 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, pack.Filter.keep_all, &.{}) 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);
}