Skip to content
zit Artifact [18889a4f72]
Login

Artifact [18889a4f72]

Check-ins Using Download Hex

Artifact 18889a4f723c9f9926c119f5c3db5cf6c8bb3ed03ddd9fca86cc9a3ef4b27f72:

//! 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 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);
}