Skip to content
zit Artifact [e7dce3a515]
Login

Artifact [e7dce3a515]

Check-ins Using Download Hex

Artifact e7dce3a51581fc24c08875d3aa64511dee9f11625327717dc4af5441df161ca1:

//! The shun+reclaim crash-recovery DST sweep. It drives the REAL rewrite —
//! `cmd/sync.zig`'s `rewriteAndSwap`, the body `zit pack` and `zit shun --force`
//! share — over a real directory whose every mutating operation goes through
//! `sim/faulty_dir.zig`, cuts the run at every mutating-operation boundary, reopens
//! the repository, and asserts that what survived is one of exactly two stores.
//!
//! WHAT IS AND IS NOT AN RFC PROPERTY HERE, stated first because the clean-room
//! rule makes it matter. The repository shun table is normative: sync-protocol
//! Section 3.3 L517-524 defines it as "a persistent, repository-local set keyed
//! by complete artifact name", which local administration "MAY add or remove
//! names" from. That a tombstone is keyed by a COMPLETE name, and that the set
//! is persistent rather than round-local, is what the oracle checks against the
//! drafts. Everything else this file pins — one temp file, one fsync, one
//! rename, and therefore one durable transition — is zit's own CLI plumbing.
//! No draft describes it, so it is asserted against the code's stated contract
//! (`rewriteAndSwap`'s doc comment), never dressed in RFC notation.
//!
//! Technique. Count the mutating operations a healthy rewrite performs, then
//! replay it once per cut and assert recovery from each: TigerBeetle's
//! storage-fault-injection lineage (the VOPR walks every disk-write boundary),
//! the same port `sim/checkout_sweep.zig` makes onto the POSIX directory
//! surface. One `std.Random.DefaultPrng.init(seed)` is the whole universe of a
//! run — corpus content, the shun selection, and every torn length — so a
//! failure replays verbatim from `reproduce with seed=N`.
//!
//! THE ORACLE, precisely, and no wider than it is true:
//!
//!   (i)   NO HYBRID. After any cut, `.zit` is byte-identical either to the
//!         pre-shun store or to the healthy post-shun store. This is a whole-
//!         file comparison against two images computed once per seed, not an
//!         enumeration of contents: it strictly catches truncation, a wrong-
//!         state swap, and a half-applied reclaim in one statement. It rests on
//!         BOTH images being reproducible byte-for-byte in a fresh directory —
//!         `packLog` a pure function of its source (pack.zig's own determinism
//!         claim) and `seedStore` a pure function of its corpus. `run`
//!         re-derives each image a second time in a second workspace and
//!         compares, once per seed, rather than assuming either.
//!
//!   (ii)  EVERY SURVIVING FRAME HASH-VERIFIES. Whichever image survived is
//!         reopened and every frame reconstructed through its (possibly
//!         re-based) delta chain and re-hashed against its own name, through the
//!         maintenance door so a tombstone cannot hide a mismatch.
//!
//!   (iii) THE TOMBSTONE AND THE RECLAIM ARE ONE TRANSITION. In the POST state
//!         every named digest has NO FRAME AT ALL — not a masked one — and its
//!         row is in the shun table. In the PRE state the table is EMPTY and
//!         every frame is still there.
//!
//!         This is deliberately narrower than "no shunned digest is readable in
//!         either state", which is false: in the pre state nothing has been
//!         shunned, so the digest reads normally, and even after a tombstone
//!         exists S1's mask is NAME-LEVEL — a masked artifact's plaintext can
//!         still be materialised as an intermediate while reconstructing a
//!         dependent that was deltified against it. Only the reclaim removes
//!         the bytes, and only in the post state.
//!
//!   (iv)  A CONCURRENT APPEND IS NEVER LOST. Driven separately (see
//!         `runConcurrentWriter`), because it is not a crash: a writer extends
//!         `.zit` in the window between the rewrite's last durable write and its
//!         rename, and the artifact it committed must still be in the store
//!         afterwards. This is what makes `rewriteAndSwap`'s fresh-stat tail
//!         re-check a live guard rather than a comment.
//!
//!   (v)   A CRASH NEVER WEDGES THE REPOSITORY. After every cut, a healthy
//!         rewrite is run again in the same directory and must succeed and land
//!         the POST image. This is §11's Recovery column for rows 1, 2, 3, 5
//!         and 7 ("re-run", "next run's truncate overwrites it") asserted rather
//!         than asserted-about, and it is what the cut enumeration actually
//!         pins — see the credit note below.
//!
//! §11 crash-matrix rows this sweep reaches, and the ones it declines. Reached
//! by the cut enumeration plus its recovery re-run: a crash before the temp
//! exists, mid-temp-write, at the temp fsync, before the rename, and at the
//! rename itself. Reached only because `faulty_dir` gained a durable watermark:
//! power loss AFTER a complete healthy rewrite (`runPowerLoss`) — without it,
//! unsynced bytes reach the real filesystem anyway and "fsync before rename" is
//! untestable. Reached only because `faulty_dir` gained an after-op hook: the
//! concurrent-writer row. DECLINED, with reasons, so the matrix is not padded:
//! rename atomicity is a POSIX platform axiom rather than a property of this
//! code; the repo-cache invalidation row is already pinned at
//! `src/repocache.zig:526`; the `serve` row is a documented limitation and an
//! e2e concern; the shun-row encoding row is an S2 unit test; ENOSPC collapses
//! into the mid-write row — same recovery path, same assertion — for a third
//! harness mode that would buy nothing; and the `--defer` torn-tail row is an S0
//! concern this sweep structurally cannot reach, because `--defer` appends a
//! `ShunRecord` to the LIVE `.zit` and never enters `rewriteAndSwap` at all,
//! which is the only function this sweep drives. Its recovery is the log's
//! existing `scan(recover_torn = true)`, pinned by the S0 unit tests; routing
//! `extra_shuns` through a rewrite, as this sweep does, exercises the OTHER
//! durability shape by construction.
//!
//! WHICH CLAUSES ARE COVERAGE AND WHICH ARE TRIPWIRES. A property no mutant can
//! kill is not coverage, so these are not credited equally:
//!
//!   (i)  IS coverage. Deleting `packLog`'s pre-rename `dst.commit()` fsync
//!        makes `runPowerLoss` report `HybridStore`, and — checked — no other
//!        test in the tree notices at all.
//!   (iv) IS coverage. Renaming before the tail re-check makes
//!        `runConcurrentWriter` report `ConcurrentAppendLost`. The existing unit
//!        "a run the concurrency guard aborts leaves .zit untouched and records
//!        NO tombstone" catches it too, so S3's fix is doubly pinned rather than
//!        newly pinned.
//!   (v)  IS coverage, and it is what gives the cut enumeration a reason to
//!        exist. Opening the temp `.exclusive` instead of under the advisory
//!        lock — the alternative `rewriteAndSwap`'s own doc rejects, because
//!        "whose leftover would block every future rewrite" — makes the recovery
//!        re-run report `RecoveryFailed` at the first cut. MEASURED: with that
//!        one change, `zig build test` reports 2306/2307 passing and the single
//!        failure is this sweep, so no other test in the tree sees it at all.
//!        The re-run also carries (ii) and (iii) into the POST state at every
//!        cut, which the cut's own oracle structurally cannot reach.
//!   (iii) POST half IS coverage: making `pack.NameSet.filter` answer
//!        `keep_all` — a tombstone recorded over content that was never dropped,
//!        the fail-OPEN the reclaim contract names — is invisible to the pack's
//!        own stats and is caught here by reading the store (`ReclaimIncomplete`).
//!        Its PRE half is NOT: an aborted rewrite that leaves residue changes the
//!        file, so (i) reports `HybridStore` first. It is kept as a cheap,
//!        specific diagnostic, not claimed as coverage.
//!   (ii) is a TRIPWIRE, not coverage. `packLog` reconstructs and re-hashes every
//!        packed artifact BEFORE its commit, and aborts the whole rewrite on a
//!        mismatch — so a single defect that would corrupt content also stops the
//!        store from ever reaching disk, and the sweep sees an untouched `.zit`
//!        and passes. No mutant makes this clause fire. It stays because it costs
//!        one pass over a five-frame store and would catch a future defect that
//!        removed that gate, but it is not evidence of anything today.
//!
//!   AND (i) AT A CUT IS A TRIPWIRE, NOT COVERAGE — stated because measuring it
//!        was the surprise of this slice. `rewriteAndSwap` writes only the temp
//!        and its LAST mutating operation is the rename, and this sweep reads the
//!        source through the healthy io on purpose, so for EVERY k in `[1, total]`
//!        the recovered `.zit` is the PRE image by construction: `classify` can
//!        return nothing else and `checkStore(.post)` is unreachable from a cut.
//!        Measured, not argued: with `runPowerLoss`, `runConcurrentWriter` and
//!        the recovery re-run all removed, deleting `packLog`'s fsync and
//!        inverting the drop predicate (so the reclaim keeps exactly the wrong
//!        artifacts) BOTH leave every cut green. What the cut's own oracle
//!        asserts is therefore "no crash inside the rewrite moves a byte of
//!        `.zit`" — a real regression tripwire against a future rewrite that
//!        mutated the source in place, and worth its cost, but not evidence
//!        about the swap itself. Both of those mutants ARE red today, and the
//!        credits above name which run catches each.
//!
//! EVERY RUN GUARDS ITS OWN PREMISE, because the credits above are worth exactly
//! as much as the guarantee that each run still does what it says. A sweep whose
//! machinery quietly stops engaging does not fail — it passes, faster, forever.
//! So each scenario asserts that it happened (`ScenarioNotReached`): a cut that
//! the crash schedule really fired, at the operation asked for; a `total` that
//! really is the number of operations a workspace performs, neither more
//! (`checkOpCount`) nor fewer; a power loss whose watermark model really had a
//! file to act on; a concurrent writer whose hook really ran. And each run
//! guards its inputs: the corpus reclaims something, both delta hops really
//! landed as deltas, the manifest really parses as one, and the same seed
//! re-derives the same corpus. MEASURED, one mutant each: switching
//! `lose_unsynced` off, aiming a cut past the last operation, under-reporting
//! `countOps` by one, emptying the drop set, corrupting the manifest's Z card,
//! truncating an F-card name, and drifting the corpus off its seed are all RED —
//! and every one of them was GREEN before these guards existed.

const std = @import("std");
const zit = @import("zit");

/// The rewrite under test. Reached relatively because `cmd/sync.zig` lives
/// outside the public `zit` module, exactly as dst.zig reaches transport.zig
/// and serve.zig.
const cmd_sync = @import("../cmd/sync.zig");
const faulty_dir = @import("faulty_dir.zig");

const artifact = zit.artifact;
const logstore = zit.logstore;
const pack = zit.pack;
const storage = zit.storage;

const Dir = std.Io.Dir;
const Io = std.Io;

const repo_filename = zit.repo_filename;

/// The corpus is named with SHA3-256 (artifact-format Section 3.2 L347-349: the
/// 64-hexadecimal-digit form, `sha3-name = 64lhexdig` at L439), so every digest
/// is a COMPLETE name — the only width the shun table is keyed by (sync-protocol
/// Section 3.3 L518-519) and the only width `pack.Filter` will act on.
const digest_len = 32;

/// Read cap for a whole-store snapshot. The corpus is a few KiB; this is a
/// tripwire against a runaway read, not an expected size.
const max_store_bytes = 4 << 20;

pub const SweepError = error{
    OutOfMemory,
    /// A setup/backend filesystem or store operation failed unexpectedly — never
    /// a fault the sweep injected, which are all handled.
    Io,
    /// The CORPUS is not the shape the sweep's claims assume — a chain member
    /// landed whole instead of as a delta, a digest was not a complete name, a
    /// frame came back phantom. Distinct from `Io` on purpose: this is a
    /// harness/premise bug, and reporting it as a filesystem fault would send a
    /// reader to the disk for a problem that lives in this file.
    PremiseViolated,
    /// The scenario a run exists to create did not happen — the cut did not
    /// fire, it fired at a different operation than the one asked for, the
    /// after-op hook never ran, or the power-loss model never had a file to act
    /// on. A green run that silently stopped testing anything is the failure
    /// mode this names, and it must never be confused with a transient the
    /// operator would retry past.
    ScenarioNotReached,
    /// The same seed built a different corpus, so an archived `reproduce with
    /// seed=N` no longer reproduces what it captured. Distinct from
    /// `PackNondeterministic`, which is about the store the corpus produces:
    /// this one says the INPUT moved.
    CorpusNondeterministic,
    /// Invariant (v): after a crash, a healthy re-run did not recover the
    /// repository to the post-shun store.
    RecoveryFailed,
    /// Invariant (i): the recovered `.zit` matched neither image.
    HybridStore,
    /// Invariant (ii): a surviving frame's content does not hash to its name.
    ContentHashMismatch,
    /// Invariant (iii), post state: a reclaimed digest still has a frame, or its
    /// tombstone is missing.
    ReclaimIncomplete,
    /// Invariant (iii), pre state: a frame vanished, or a tombstone was recorded,
    /// without the rename having happened.
    AbortLeftResidue,
    /// Invariant (iv): the concurrency guard let a stale rewrite land over a
    /// concurrently committed artifact.
    ConcurrentAppendLost,
    /// The healthy reference rewrite itself failed — a harness fault, reported
    /// distinctly so it is never read as a crash-recovery finding.
    ReferenceRewriteFailed,
    /// Two healthy packs of the same source produced different bytes, so the
    /// two-image oracle has no ground to stand on.
    PackNondeterministic,
};

// ---------------------------------------------------------------------------
// The corpus
// ---------------------------------------------------------------------------

/// Fixed shape, seeded content. Five artifacts with real structure: a delta
/// CHAIN of depth two (`dep2` is encoded against `dep1`, which is encoded
/// against `base`), an unrelated standalone blob, and a genuine manifest whose F
/// cards name the blobs. The shape is fixed so the roles the shun selection must
/// be able to hit — a delta base, a mid-chain frame, a manifest — are always
/// present; the CONTENT varies with the seed so no run packs the same bytes.
const artifact_count = 5;
const idx_base = 0;
const idx_dep1 = 1;
const idx_dep2 = 2;
const idx_blob = 3;
const idx_manifest = 4;

const Corpus = struct {
    bodies: [artifact_count][]u8,
    digests: [artifact_count][digest_len]u8,
    /// Which artifacts this run reclaims. At least one is always set.
    shunned: [artifact_count]bool,
    epoch: i64,

    fn deinit(self: *Corpus, gpa: std.mem.Allocator) void {
        for (self.bodies) |b| gpa.free(b);
        self.* = undefined;
    }

    fn shunCount(self: *const Corpus) usize {
        var n: usize = 0;
        for (self.shunned) |s| n += @intFromBool(s);
        return n;
    }
};

fn sha3(bytes: []const u8) [digest_len]u8 {
    var out: [digest_len]u8 = undefined;
    std.crypto.hash.sha3.Sha3_256.hash(bytes, &out, .{});
    return out;
}

/// Incompressible seeded bytes. Deltas between near-identical incompressible
/// bodies are tiny, which is what makes the store store them AS deltas — a
/// compressible body would land whole and the "delta chain" premise would be
/// quietly false.
fn noiseBody(gpa: std.mem.Allocator, random: std.Random, len: usize) SweepError![]u8 {
    const buf = gpa.alloc(u8, len) catch return error.OutOfMemory;
    random.bytes(buf);
    return buf;
}

fn buildCorpus(gpa: std.mem.Allocator, random: std.Random, seed: u64) SweepError!Corpus {
    var self: Corpus = .{
        .bodies = @splat(&.{}),
        .digests = @splat(@splat(0)),
        .shunned = @splat(false),
        .epoch = 1753000000 + @as(i64, @intCast(random.uintLessThan(u32, 1_000_000))),
    };
    errdefer for (self.bodies) |b| gpa.free(b);

    const base = try noiseBody(gpa, random, 1024);
    self.bodies[idx_base] = base;

    const dep1 = gpa.dupe(u8, base) catch return error.OutOfMemory;
    dep1[100] ^= 0xff;
    self.bodies[idx_dep1] = dep1;

    const dep2 = gpa.dupe(u8, dep1) catch return error.OutOfMemory;
    dep2[900] ^= 0x0f;
    self.bodies[idx_dep2] = dep2;

    self.bodies[idx_blob] = try noiseBody(gpa, random, 700);

    self.bodies[idx_manifest] = try buildManifest(gpa, self.bodies[0..idx_manifest]);

    for (self.bodies, 0..) |b, i| self.digests[i] = sha3(b);

    // THE SELECTION. One FORCED member cycling over the three structural roles,
    // so a delta base, a mid-chain frame and a manifest each get reclaimed on a
    // known fraction of seeds rather than by luck; then every other artifact
    // joins with probability 1/4, so the size and shape of the drop set still
    // vary.
    //
    // INDEXED BY THE SEED, NOT DRAWN FROM THE PRNG, and that is the whole value
    // of it. An independent 1-in-3 draw per seed guarantees nothing: measured on
    // the draw order this function had, seeds 1, 2 and 3 forced base, manifest
    // and manifest, so a three-seed gate never forced the mid-chain frame at all
    // and the coverage was back to the luck the forcing exists to remove. `seed
    // % roles.len` makes any three consecutive seeds cover all three roles, by
    // construction and independently of how many draws precede it.
    const roles = [_]usize{ idx_base, idx_manifest, idx_dep1 };
    const forced = roles[@intCast(seed % roles.len)];
    self.shunned[forced] = true;
    for (&self.shunned, 0..) |*s, i| {
        if (i == forced) continue;
        s.* = random.uintLessThan(u8, 4) == 0;
    }
    return self;
}

/// A real manifest: a D card, one F card per blob naming it by its 64-hex
/// SHA3-256 name (artifact-format Section 3.2 L347-349), and the trailing
/// self-describing Z card.
///
/// WHAT THE MANIFEST DOES AND DOES NOT BUY, stated so it is not read as more
/// than it is. `packLog` is content-agnostic and `rewriteAndSwap` never inspects
/// artifact type, so on the path under test a manifest is a 32-byte name in a
/// `NameSet` exactly like a blob: nothing here parses a D/F/Z card, and no
/// clause distinguishes reclaiming `idx_manifest` from reclaiming `idx_blob`.
/// What it DOES buy is the reference structure `checkStore`'s post branch then
/// checks by name: reclaiming the manifest must not orphan the blobs it names
/// (they are all still required present), and reclaiming a named blob must
/// leave the manifest itself present and hash-verifying with its now-dangling F
/// card. Both are asserted; neither would be expressible over five anonymous
/// blobs.
fn buildManifest(gpa: std.mem.Allocator, blobs: []const []u8) SweepError![]u8 {
    var body: std.ArrayList(u8) = .empty;
    defer body.deinit(gpa);
    body.appendSlice(gpa, "D 2020-01-01T00:00:00\n") catch return error.OutOfMemory;
    // Filename-sorted F cards — one per blob, f0 through f3 — the canonical
    // order, which the fixed `{d}` numbering gives for free below ten.
    for (blobs, 0..) |b, i| {
        var name: [64]u8 = undefined;
        const id = artifact.artifactName(.sha3_256, b, &name);
        body.print(gpa, "F f{d}.bin {s}\n", .{ i, id }) catch return error.OutOfMemory;
    }
    var ck: [32]u8 = undefined;
    const checksum = artifact.selfChecksum(body.items, &ck);
    return std.fmt.allocPrint(gpa, "{s}Z {s}\n", .{ body.items, checksum }) catch return error.OutOfMemory;
}

/// The MANIFEST premise, checked rather than assumed, for the same reason
/// `seedStore` checks the delta-chain premise: the path under test is content-
/// agnostic, so nothing downstream would ever notice a malformed D/F/Z card and
/// the "genuine manifest" this sweep's shape claims would be decorative — a
/// 32-byte name dressed up as an artifact.
///
/// Framing gates first (artifact-format Section 4.1, via `artifact.frame`),
/// then the Z self-checksum stated explicitly, because `frame` also passes a
/// Z-ABSENT artifact and this one must have one; then every card parsed by the
/// same single-card parsers the assembler uses, with each F card required to
/// name its blob by the exact 64-hex SHA3-256 name the corpus stored it under
/// (artifact-format Section 3.2 L347-349).
fn validateManifest(body: []const u8, blobs: []const []const u8) SweepError!void {
    const framed = artifact.frame(body) orelse return error.PremiseViolated;
    if (artifact.verifyZCard(framed) != .match) return error.PremiseViolated;

    var off: usize = 0;
    var f_seen: usize = 0;
    var d_seen: usize = 0;
    var z_seen: usize = 0;
    while (off < framed.len) {
        const nl = std.mem.indexOfScalarPos(u8, framed, off, '\n') orelse return error.PremiseViolated;
        const line = framed[off .. nl + 1];
        off = nl + 1;
        const card = artifact.tokenizeCardLine(line) orelse return error.PremiseViolated;
        switch (card.letter) {
            'D' => {
                _ = artifact.parseDCard(card) orelse return error.PremiseViolated;
                d_seen += 1;
            },
            'F' => {
                // Sized by the parser's stated precondition: filename plus
                // old-name, and these cards carry no old name.
                var decode_buf: [256]u8 = undefined;
                if (line.len > decode_buf.len) return error.PremiseViolated;
                const fc = artifact.parseFCard(card, &decode_buf) orelse return error.PremiseViolated;
                if (f_seen >= blobs.len) return error.PremiseViolated;
                var want_name: [16]u8 = undefined;
                const want = std.fmt.bufPrint(&want_name, "f{d}.bin", .{f_seen}) catch return error.PremiseViolated;
                if (!std.mem.eql(u8, fc.filename, want)) return error.PremiseViolated;
                var id_buf: [64]u8 = undefined;
                const id = artifact.artifactName(.sha3_256, blobs[f_seen], &id_buf);
                const hash = fc.hash orelse return error.PremiseViolated;
                if (!std.mem.eql(u8, hash, id)) return error.PremiseViolated;
                f_seen += 1;
            },
            'Z' => z_seen += 1,
            else => return error.PremiseViolated,
        }
    }
    if (d_seen != 1 or z_seen != 1 or f_seen != blobs.len) return error.PremiseViolated;
}

// ---------------------------------------------------------------------------
// Workspaces
// ---------------------------------------------------------------------------

/// A throwaway directory holding one `.zit`. Every name is derived from the run
/// seed, so two runs can never contend for one directory even though each is
/// deleted as soon as it is finished with. ONE PER CUT, never shared: the
/// rewrite's temp has a FIXED name held under an exclusive advisory lock, so two
/// cuts sharing a directory would contend for it and the second would abort with
/// `RewriteBusy` instead of being cut where the sweep asked.
const Workspace = struct {
    name: []u8,
    dir: Dir,
    root: Dir,
    io: Io,

    fn open(gpa: std.mem.Allocator, io: Io, root: Dir, nonce: u64) SweepError!Workspace {
        const name = std.fmt.allocPrint(gpa, "shun-{x}", .{nonce}) catch return error.OutOfMemory;
        errdefer gpa.free(name);
        root.deleteTree(io, name) catch {}; // a prior aborted run may have left it
        root.createDir(io, name, .default_dir) catch return error.Io;
        // The directory now exists and only `close` removes it, so a failure
        // between here and the return would leak it into the SHARED process
        // workspace root (dst.zig hands the checkout sweep's root to this one),
        // where it would outlive the whole run.
        errdefer root.deleteTree(io, name) catch {};
        const dir = root.openDir(io, name, .{ .access_sub_paths = true, .follow_symlinks = false }) catch return error.Io;
        return .{ .name = name, .dir = dir, .root = root, .io = io };
    }

    /// LOUD, not best-effort. `open`'s error path already guards this leak with
    /// an `errdefer`; dropping the same guarantee on the success path is how a
    /// systematic failure accumulates a directory per cut plus five per seed in
    /// the SHARED process workspace root, across a whole `--seeds` run, with no
    /// signal. `close` is called from `defer`, so it cannot return an error and
    /// there is nothing above it to report to: nothing else in the process holds
    /// this directory, so a failure to delete it is a harness bug, and harness
    /// bugs in this sweep stop in every build mode (`faulty_dir` sets the
    /// precedent).
    fn close(self: *Workspace, gpa: std.mem.Allocator) void {
        self.dir.close(self.io);
        self.root.deleteTree(self.io, self.name) catch |e| {
            std.debug.panic("shun_sweep: workspace {s} leaked: {s}", .{ self.name, @errorName(e) });
        };
        gpa.free(self.name);
        self.* = undefined;
    }
};

/// Write the corpus into a fresh `.zit`. Byte-identical every call for a given
/// corpus: the puts are in fixed order and the content is already decided.
fn seedStore(gpa: std.mem.Allocator, io: Io, dir: Dir, corpus: *const Corpus) SweepError!void {
    try validateManifest(corpus.bodies[idx_manifest], corpus.bodies[0..idx_manifest]);

    var file = dir.createFile(io, repo_filename, .{ .read = true, .truncate = true }) catch return error.Io;
    defer file.close(io);
    var fs = storage.FileStorage.init(io, file, 0);
    var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch return error.Io;
    defer log.deinit(gpa);

    const base_rid = log.putArtifact(&corpus.digests[idx_base], corpus.bodies[idx_base], null) catch return error.Io;
    const dep1_rid = log.putArtifact(&corpus.digests[idx_dep1], corpus.bodies[idx_dep1], base_rid) catch return error.Io;
    const dep2_rid = log.putArtifact(&corpus.digests[idx_dep2], corpus.bodies[idx_dep2], dep1_rid) catch return error.Io;
    _ = log.putArtifact(&corpus.digests[idx_blob], corpus.bodies[idx_blob], null) catch return error.Io;
    _ = log.putArtifact(&corpus.digests[idx_manifest], corpus.bodies[idx_manifest], null) catch return error.Io;
    log.commit() catch return error.Io;

    // The delta-chain PREMISE, checked rather than assumed: if these ever landed
    // whole, every "reclaiming a delta base" claim below would be vacuous. BOTH
    // hops are checked — `dep2` deltified against `dep1` is what makes the chain
    // depth two rather than one, and it is exactly as easy to check as to assume.
    if (log.kindOf(dep1_rid) != .delta) return error.PremiseViolated;
    if (log.kindOf(dep2_rid) != .delta) return error.PremiseViolated;
}

fn snapshot(gpa: std.mem.Allocator, io: Io, dir: Dir) SweepError![]u8 {
    return dir.readFileAlloc(io, repo_filename, gpa, .limited(max_store_bytes)) catch |e| switch (e) {
        error.OutOfMemory => error.OutOfMemory,
        else => error.Io,
    };
}

// ---------------------------------------------------------------------------
// Driving the real rewrite
// ---------------------------------------------------------------------------

/// The drop set and tombstone rows for one run, derived from ONE object exactly
/// as `runShun` derives them, so a name cannot be recorded without being dropped.
const Plan = struct {
    set: pack.NameSet = .{},
    rows: [artifact_count]logstore.ShunRecord = undefined,
    row_count: usize = 0,

    fn init(gpa: std.mem.Allocator, corpus: *const Corpus) SweepError!Plan {
        var self: Plan = .{};
        errdefer self.set.deinit(gpa);
        for (corpus.shunned, 0..) |s, i| {
            if (!s) continue;
            // `BadKey` is a CHECKED error in `pack.NameSet.add` — it guards the
            // fail-OPEN case where a wrong-width key silently matches nothing —
            // so it is reported as the corpus bug it would be, never relabelled
            // as an allocation failure.
            self.set.add(gpa, &corpus.digests[i]) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.BadKey => return error.PremiseViolated,
            };
            self.rows[self.row_count] = .{
                .hash = &corpus.digests[i],
                .epoch = corpus.epoch,
                .reason = "dst reclaim",
            };
            self.row_count += 1;
        }
        // THE PREMISE THE WHOLE RUN RESTS ON, and the one thing this loop cannot
        // trivially satisfy: the corpus must reclaim SOMETHING. An empty drop
        // set would hand `rewriteAndSwap` a no-op — pre and post would coincide,
        // and every clause about "what the reclaim removed" would pass by not
        // having anything to say. A returned error rather than an assert on
        // purpose: `zig build dst` compiles with `standardOptimizeOption`, so a
        // `std.debug.assert` here is nothing at all under `-Doptimize=ReleaseFast`
        // — precisely the build where a vacuous sweep would go unnoticed.
        //
        // (`row_count == corpus.shunCount()` is NOT checked: `row_count` is
        // incremented once per true entry of the same array `shunCount` counts,
        // so the two cannot disagree. An assertion that restates its own loop
        // carries no information.)
        if (self.row_count == 0) return error.PremiseViolated;
        return self;
    }

    fn deinit(self: *Plan, gpa: std.mem.Allocator) void {
        self.set.deinit(gpa);
        self.* = undefined;
    }
};

/// Open `.zit`, run the plan through `rewriteAndSwap` under `io`, close. The
/// rewrite's own error is RETURNED, never swallowed: the crash cuts expect one
/// and the reference runs expect none.
fn rewrite(
    gpa: std.mem.Allocator,
    healthy: Io,
    rewrite_io: Io,
    dir: Dir,
    plan: *const Plan,
) SweepError!cmd_sync.RewriteError!void {
    // The SOURCE is read through the healthy io on purpose: this sweep is about
    // what the rewrite WRITES, and a source read that lies would only mask that.
    var src_file = dir.openFile(healthy, repo_filename, .{}) catch return error.Io;
    defer src_file.close(healthy);
    const st = src_file.stat(healthy) catch return error.Io;
    var src_fs = storage.FileStorage.init(healthy, src_file, st.size);
    var src = logstore.LogStore.openReadOnly(gpa, src_fs.storageFile()) catch return error.Io;
    defer src.deinit(gpa);
    const src_key = src.logKey() catch return error.Io;

    if (cmd_sync.rewriteAndSwap(
        gpa,
        rewrite_io,
        dir,
        &src,
        src_file,
        src_key,
        plan.set.filter(),
        plan.rows[0..plan.row_count],
    )) |stats| {
        // On the healthy path the pack's own observation of the new generation
        // must agree with the row list it was handed. DELIBERATELY ONLY THE ROW
        // COUNT: whether the content was actually reclaimed is decided by
        // reading the resulting store (`checkStore`), never by asking the pack
        // whether it did its job — a stats field is the wrong witness for the
        // one failure mode the reclaim contract calls fail-OPEN.
        if (stats.shuns_recorded != plan.row_count) return error.ReferenceRewriteFailed;
        return {};
    } else |e| {
        const wrapped: cmd_sync.RewriteError!void = e;
        return wrapped;
    }
}

/// Count the mutating operations one healthy rewrite performs — the cut range
/// `[1, total]`, whose last member is the rename.
fn countOps(gpa: std.mem.Allocator, io: Io, root: Dir, corpus: *const Corpus, plan: *const Plan, seed: u64) SweepError!u64 {
    var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "count-ops"));
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);

    var fprng = std.Random.DefaultPrng.init(0xC0FFEE);
    var faulty = faulty_dir.FaultyDir.init(io, fprng.random(), .{}); // no crash: count only
    faulty.install();
    defer faulty_dir.FaultyDir.uninstall();
    const outcome = try rewrite(gpa, io, faulty.io(), ws.dir, plan);
    outcome catch return error.ReferenceRewriteFailed;
    return faulty.op_index;
}

/// The healthy post-shun store's bytes — one of the oracle's two images.
fn referenceImage(gpa: std.mem.Allocator, io: Io, root: Dir, corpus: *const Corpus, plan: *const Plan, nonce: u64) SweepError![]u8 {
    var ws = try Workspace.open(gpa, io, root, nonce);
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);
    const outcome = try rewrite(gpa, io, io, ws.dir, plan);
    outcome catch return error.ReferenceRewriteFailed;
    return snapshot(gpa, io, ws.dir);
}

// ---------------------------------------------------------------------------
// The oracle
// ---------------------------------------------------------------------------

const State = enum { pre, post };

/// Invariant (i): the surviving `.zit` is one of the two images, whole.
fn classify(bytes: []const u8, pre: []const u8, post: []const u8) SweepError!State {
    if (std.mem.eql(u8, bytes, pre)) return .pre;
    if (std.mem.eql(u8, bytes, post)) return .post;
    return error.HybridStore;
}

/// Invariants (ii) and (iii) over the store that survived. Reopening is itself
/// part of the check: `openReadOnly` whole-frame-CRC-verifies every frame, so a
/// store that will not open is already a verdict.
fn checkStore(gpa: std.mem.Allocator, io: Io, dir: Dir, corpus: *const Corpus, state: State) SweepError!void {
    var file = dir.openFile(io, repo_filename, .{}) catch return error.Io;
    defer file.close(io);
    const st = file.stat(io) catch return error.Io;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch return switch (state) {
        .pre => error.AbortLeftResidue,
        .post => error.ReclaimIncomplete,
    };
    defer log.deinit(gpa);

    // (ii) every surviving frame reconstructs and re-hashes to its own name.
    // Through the MAINTENANCE door: a masked frame must be honest too, or a
    // tombstone could hide a corrupt reclaim survivor.
    const rids = log.ridsInPersistedOrder(gpa) catch return error.OutOfMemory;
    defer gpa.free(rids);
    for (rids) |rid| {
        // EVERY frame, with no vacuous skip. `contentUnmasked` answers null for
        // exactly one reason — a phantom entry — and this corpus creates none,
        // so a null here means the store is not the one the sweep built and the
        // clause would otherwise pass by not looking.
        const maybe = log.contentUnmasked(gpa, rid, null) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            else => return error.Io,
        };
        const content = maybe orelse return error.PremiseViolated;
        defer gpa.free(content);
        if (!std.mem.eql(u8, &sha3(content), log.hashOf(rid))) return error.ContentHashMismatch;
    }

    // (iii) the tombstone and the reclaim are one transition.
    switch (state) {
        .pre => {
            // Nothing landed: the table is empty and every frame is still here.
            if (log.shunList().len != 0) return error.AbortLeftResidue;
            for (&corpus.digests) |*dg| {
                if (log.lookup(dg) == null) return error.AbortLeftResidue;
            }
        },
        .post => {
            if (log.shunList().len != corpus.shunCount()) return error.ReclaimIncomplete;
            for (&corpus.digests, corpus.shunned) |*dg, was_shunned| {
                if (was_shunned) {
                    // NO FRAME AT ALL, not merely masked: the bytes are gone, so
                    // even the maintenance door finds nothing to reconstruct.
                    if (frameExists(&log, rids, dg)) return error.ReclaimIncomplete;
                    if (!log.isShunned(dg)) return error.ReclaimIncomplete;
                } else {
                    if (log.lookup(dg) == null) return error.ReclaimIncomplete;
                }
            }
        },
    }
}

/// Does `log` hold a frame named `digest`, ignoring the shun mask? `lookup`
/// answers null for a merely-tombstoned name too, which would let a reclaim that
/// kept the bytes pass; the persisted-order walk cannot be fooled that way.
fn frameExists(log: *const logstore.LogStore, rids: []const logstore.Rid, digest: []const u8) bool {
    for (rids) |rid| {
        if (std.mem.eql(u8, log.hashOf(rid), digest)) return true;
    }
    return false;
}

// ---------------------------------------------------------------------------
// The enumeration
// ---------------------------------------------------------------------------

/// THE ONE ENTRY POINT. One seed: build the corpus, check that the seed
/// reproduces it, compute the two images, then cut — every reachable cut, each
/// with its recovery re-run, plus the power-loss and concurrent-writer runs.
/// There is no gate/soak split because there is nothing to split: the rewrite's
/// mutating-operation count is a single digit, so "every boundary" IS the fast
/// path.
///
/// EVERY cut is replayed — there is no sampling knob and no soak/gate split.
/// `rewriteAndSwap` performs a single-digit number of GATED operations on this
/// corpus — nine, measured: the temp's create, the pack's positional writes and
/// its fsync, and the rename (`setLength` and `close` are deliberately not
/// counted; see `faulty_dir`) — so exhaustive enumeration costs less than the
/// arithmetic that would sample it.
/// An earlier draft carried a `max_cuts = 12` cap; with `total = 9` the cap
/// never bound, the even-spacing formula degenerated to `k = 1 + j`, and the
/// knob was an interface pretending to have a user.
pub fn run(gpa: std.mem.Allocator, io: Io, root: Dir, seed: u64) SweepError!void {
    var prng = std.Random.DefaultPrng.init(seed);
    const random = prng.random();

    var corpus = try buildCorpus(gpa, random, seed);
    defer corpus.deinit(gpa);
    var plan = try Plan.init(gpa, &corpus);
    defer plan.deinit(gpa);

    // THE INPUT IS RE-DERIVED TOO, not just the two output images below. A
    // failure report is only worth archiving if `reproduce with seed=N` still
    // reproduces it, and the thing a future edit is most likely to perturb is
    // `buildCorpus`'s PRNG CONSUMPTION ORDER — the epoch draw, then three
    // `noiseBody` fills, then the per-artifact admission draws. Shift any one of
    // them and the same seed silently reclaims a different set of artifacts,
    // which is a captured failure that no longer replays rather than a test that
    // fails.
    {
        var again_prng = std.Random.DefaultPrng.init(seed);
        var again = try buildCorpus(gpa, again_prng.random(), seed);
        defer again.deinit(gpa);
        if (again.epoch != corpus.epoch) return error.CorpusNondeterministic;
        for (&again.digests, &corpus.digests) |*a, *b| {
            if (!std.mem.eql(u8, a, b)) return error.CorpusNondeterministic;
        }
        for (again.shunned, corpus.shunned) |a, b| {
            if (a != b) return error.CorpusNondeterministic;
        }
    }

    // Image 1: the pre-shun store, exactly as `seedStore` writes it. Computed
    // TWICE in different workspaces for the same reason image 2 is: the oracle
    // compares whole files across directories, so `seedStore` being a pure
    // function of its corpus is half the ground it stands on, and it was the
    // half that used to be assumed.
    const pre = blk: {
        var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "pre-image"));
        defer ws.close(gpa);
        try seedStore(gpa, io, ws.dir, &corpus);
        break :blk try snapshot(gpa, io, ws.dir);
    };
    defer gpa.free(pre);
    {
        var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "pre-image-again"));
        defer ws.close(gpa);
        try seedStore(gpa, io, ws.dir, &corpus);
        const again = try snapshot(gpa, io, ws.dir);
        defer gpa.free(again);
        if (!std.mem.eql(u8, pre, again)) return error.PackNondeterministic;
    }

    // Image 2: the healthy post-shun store. Computed TWICE in different
    // workspaces, because the two-image oracle is only sound if `packLog` is a
    // pure function of its source — pack.zig claims it, so the claim is checked
    // once per seed rather than trusted.
    const post = try referenceImage(gpa, io, root, &corpus, &plan, std.hash.Wyhash.hash(seed, "post-image"));
    defer gpa.free(post);
    {
        const again = try referenceImage(gpa, io, root, &corpus, &plan, std.hash.Wyhash.hash(seed, "post-image-again"));
        defer gpa.free(again);
        if (!std.mem.eql(u8, post, again)) return error.PackNondeterministic;
    }
    if (std.mem.eql(u8, pre, post)) return error.ReferenceRewriteFailed; // the run must do something

    const total = try countOps(gpa, io, root, &corpus, &plan, seed);
    if (total == 0) return error.ReferenceRewriteFailed;
    try checkOpCount(gpa, io, root, &corpus, &plan, post, seed, total);

    var k: u64 = 1;
    while (k <= total) : (k += 1) {
        try sweepCut(gpa, io, root, &corpus, &plan, pre, post, seed, k);
    }

    // §11 row 4, and the only cut that is not a cut: a COMPLETE healthy rewrite
    // followed by power loss. Under `lose_unsynced` the temp keeps only what it
    // fsynced, so this is where "fsync the new generation before renaming it" is
    // a tested property instead of a comment.
    try runPowerLoss(gpa, io, root, &corpus, &plan, pre, post, seed);

    // §11 row 6: the concurrency guard, driven at the one instant it can matter.
    try runConcurrentWriter(gpa, io, root, &corpus, &plan, pre, seed, total);
}

/// `total` is measured by `countOps` in its OWN workspace, and everything the
/// enumeration claims rests on it also being the count a CUT's workspace
/// performs: the cut range `[1, total]` covers every boundary only if there is
/// no op `total + 1`, and `runConcurrentWriter`'s `after_op = total - 1` names
/// the pre-rename window only if the rename really is the last op.
///
/// Each cut asserts `crash_op_index == k`, which pins the numbering up to
/// `total` — a workspace performing FEWER ops fails to reach its cut and says
/// so. It cannot see one op too MANY, though: cuts 1..total would all still fire
/// exactly where asked while the tail went untested and the sweep stayed green.
/// So that half is measured directly, once per seed, by arming a cut at
/// `total + 1` and requiring it never to fire.
fn checkOpCount(
    gpa: std.mem.Allocator,
    io: Io,
    root: Dir,
    corpus: *const Corpus,
    plan: *const Plan,
    post: []const u8,
    seed: u64,
    total: u64,
) SweepError!void {
    var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "op-count"));
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);

    var fprng = std.Random.DefaultPrng.init(seed *% 0x165667B1);
    var faulty = faulty_dir.FaultyDir.init(io, fprng.random(), .{ .crash_at_op = total + 1 });
    faulty.install();
    defer faulty_dir.FaultyDir.uninstall(); // see `sweepCut` on why this is deferred
    const outcome = try rewrite(gpa, io, faulty.io(), ws.dir, plan);

    // THE OP COUNT IS READ BEFORE THE OUTCOME, deliberately. If there IS an op
    // `total + 1`, the cut fires on it and the rewrite fails — and reporting
    // that as `ReferenceRewriteFailed` would send an operator hunting a
    // filesystem fault for what is actually a stale op count, which is exactly
    // the misattribution this ordering exists to avoid.
    if (faulty.crashed or faulty.op_index != total) return error.ScenarioNotReached;
    outcome catch return error.ReferenceRewriteFailed;
    const after = try snapshot(gpa, io, ws.dir);
    defer gpa.free(after);
    if (!std.mem.eql(u8, after, post)) return error.ReferenceRewriteFailed;
}

/// One cut: a fresh workspace at the pre-shun store, the rewrite faulted at op
/// `k`, the oracle, and then — the part that earns the enumeration its cost —
/// the RECOVERY: a healthy rewrite in the same directory, which must succeed and
/// land the post-shun store.
///
/// Why the recovery re-run is the load-bearing half. `rewriteAndSwap` writes
/// only the temp and its last mutating operation is the rename, so no cut can
/// leave `.zit` as anything but the pre image; the oracle at a cut is a tripwire
/// against a future rewrite that mutated the source in place, not a statement
/// about the swap (the module header says so at length). The re-run is what
/// asserts §11's Recovery column — "re-run", "next run's truncate overwrites
/// it" — and it is falsifiable: opening the temp `.exclusive` rather than under
/// the advisory lock, the alternative `rewriteAndSwap`'s own doc rejects because
/// "whose leftover would block every future rewrite", leaves a crashed run's
/// temp fatal to the next one and turns this green red.
fn sweepCut(
    gpa: std.mem.Allocator,
    io: Io,
    root: Dir,
    corpus: *const Corpus,
    plan: *const Plan,
    pre: []const u8,
    post: []const u8,
    seed: u64,
    k: u64,
) SweepError!void {
    var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, std.mem.asBytes(&k)));
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);

    {
        var fprng = std.Random.DefaultPrng.init(seed *% 0x9E37_79B9 +% k);
        var faulty = faulty_dir.FaultyDir.init(io, fprng.random(), .{
            .crash_at_op = k,
            .torn_write = true,
            .lose_unsynced = true,
        });
        faulty.install();
        // DEFERRED, not trailing: `rewrite` can fail with a SETUP error before
        // the crash it was armed for, and leaving the module-global decorator
        // pointing at this dead stack frame would turn a diagnosable harness
        // error into the next `install`'s assert — or, in a release build where
        // that assert is elided, into a dangling pointer.
        defer faulty_dir.FaultyDir.uninstall();
        // The faulted rewrite is EXPECTED to fail — that is the crash. Its error
        // is not the oracle; the disk is.
        const outcome = try rewrite(gpa, io, faulty.io(), ws.dir, plan);
        outcome catch {};

        // THE CUT HAS TO HAVE FIRED, AT THE OPERATION IT WAS AIMED AT. Without
        // this the oracle below is blind twice over. First, a cut that never
        // fires leaves a store reached by a completely HEALTHY rewrite, and
        // `classify`/`checkStore(.post)` pass on it happily — so the enumeration
        // can silently stop testing boundaries while staying green. Second,
        // `total` was measured by `countOps` in a DIFFERENT workspace; nothing
        // otherwise ties it to the operation count this one performs, so if this
        // rewrite ever performed more operations, the tail boundaries — the
        // rename at op `total` above all — would never be generated and would
        // silently go untested.
        //
        // `crash_op_index == k` closes both at once: it proves the cut fired,
        // and it proves this workspace's op numbering is the numbering
        // `countOps` measured. That second half is also what `runConcurrentWriter`
        // depends on, since its `after_op = total - 1` only names the pre-rename
        // window while the rename really is op `total`.
        if (!faulty.crashed or faulty.crash_op_index != k) return error.ScenarioNotReached;
    }

    try oracle(gpa, io, ws.dir, corpus, pre, post);

    // (v) RECOVERY. Whatever the crash left beside `.zit`, the next run must
    // work — through the healthy io, so nothing is injected and the only way to
    // fail is for the crashed run's residue to have wedged the repository.
    const recovered = rewrite(gpa, io, io, ws.dir, plan) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory, // never a verdict about the code
        else => return error.RecoveryFailed,
    };
    recovered catch return error.RecoveryFailed;
    const after = try snapshot(gpa, io, ws.dir);
    defer gpa.free(after);
    if (!std.mem.eql(u8, after, post)) return error.RecoveryFailed;
    try checkStore(gpa, io, ws.dir, corpus, .post);
}

/// A complete healthy rewrite, then power loss: anything written but never
/// fsynced is discarded at close. The result must be the POST image — the
/// rewrite fsynced the new generation before renaming it, so there is nothing to
/// lose.
fn runPowerLoss(
    gpa: std.mem.Allocator,
    io: Io,
    root: Dir,
    corpus: *const Corpus,
    plan: *const Plan,
    pre: []const u8,
    post: []const u8,
    seed: u64,
) SweepError!void {
    var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "power-loss"));
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);

    {
        var fprng = std.Random.DefaultPrng.init(seed *% 0x85EB_CA6B);
        var faulty = faulty_dir.FaultyDir.init(io, fprng.random(), .{ .lose_unsynced = true });
        faulty.install();
        defer faulty_dir.FaultyDir.uninstall(); // see `sweepCut` on why this is deferred
        const outcome = try rewrite(gpa, io, faulty.io(), ws.dir, plan);
        outcome catch return error.ReferenceRewriteFailed;

        // THE POWER-LOSS MODEL HAS TO HAVE HAD SOMETHING TO ACT ON, and this is
        // the run that carries invariant (i)'s coverage credit, so a silent
        // degradation here costs the slice its only real pin.
        //
        // The assertion is NOT "a truncation happened": on the healthy path
        // `dirty == durable` at close and nothing is lost, which is the whole
        // point of the property. It is that the machinery SAW the temp —
        // `dirty_closes` counts files closed with a tracked slot and bytes in
        // it. With the model switched off (`lose_unsynced = false`, the exact
        // end-state a silently-dead watermark produces) this run degenerates
        // into "a healthy rewrite produces the post image", which `run`
        // already established twice per seed via `referenceImage`, and it would
        // assert nothing new while staying green. `runConcurrentWriter` guards
        // its own premise the same way, and this run was the one it was missing
        // from.
        if (faulty.dirty_closes == 0) return error.ScenarioNotReached;
    }

    try oracle(gpa, io, ws.dir, corpus, pre, post);
}

/// Invariants (i)-(iii) over whatever survived in `dir`.
fn oracle(
    gpa: std.mem.Allocator,
    io: Io,
    dir: Dir,
    corpus: *const Corpus,
    pre: []const u8,
    post: []const u8,
) SweepError!void {
    const after = try snapshot(gpa, io, dir);
    defer gpa.free(after);
    const state = try classify(after, pre, post);
    try checkStore(gpa, io, dir, corpus, state);
}

// ---------------------------------------------------------------------------
// The concurrent writer (§11 row 6)
// ---------------------------------------------------------------------------

/// Context for the after-op hook: everything the concurrent writer needs, plus
/// a place to report its own failure (the hook cannot return an error).
const Appender = struct {
    gpa: std.mem.Allocator,
    io: Io,
    dir: Dir,
    body: []const u8,
    digest: [digest_len]u8,
    fired: bool = false,
    failed: bool = false,

    /// Append one artifact to the live `.zit` through the HEALTHY io — driving
    /// the decorated io here would re-enter the gate and shift every later
    /// operation index, which is exactly the nondeterminism this sweep exists to
    /// not have.
    fn hook(ctx: ?*anyopaque) void {
        const self: *Appender = @ptrCast(@alignCast(ctx.?));
        self.fired = true;
        self.append() catch {
            self.failed = true;
        };
    }

    fn append(self: *Appender) !void {
        var f = try self.dir.openFile(self.io, repo_filename, .{ .mode = .read_write });
        defer f.close(self.io);
        const st = try f.stat(self.io);
        var fs = storage.FileStorage.init(self.io, f, st.size);
        var log = try logstore.LogStore.open(self.gpa, fs.storageFile(), 0);
        defer log.deinit(self.gpa);
        _ = try log.putArtifact(&self.digest, self.body, null);
        try log.commit();
    }
};

/// A writer commits an artifact into `.zit` in the window between the rewrite's
/// last durable write and its rename. The rewrite is working from a snapshot
/// that predates that artifact, so swapping its temp in would DELETE a committed
/// artifact. `after_op = total - 1` names that window exactly: the rename is by
/// construction op `total`, so op `total - 1` is the last mutating operation
/// before it, and the fresh-stat tail re-check sits between them.
///
/// The oracle is deliberately about the STORE, not the return value: a rewrite
/// that renames first and validates afterwards still reports `TailChanged`,
/// while having already thrown the concurrent artifact away.
fn runConcurrentWriter(
    gpa: std.mem.Allocator,
    io: Io,
    root: Dir,
    corpus: *const Corpus,
    plan: *const Plan,
    pre: []const u8,
    seed: u64,
    total: u64,
) SweepError!void {
    if (total < 2) return error.ReferenceRewriteFailed;

    var ws = try Workspace.open(gpa, io, root, std.hash.Wyhash.hash(seed, "concurrent"));
    defer ws.close(gpa);
    try seedStore(gpa, io, ws.dir, corpus);

    var late_prng = std.Random.DefaultPrng.init(seed *% 0xC2B2_AE35);
    const late_body = try noiseBody(gpa, late_prng.random(), 512);
    defer gpa.free(late_body);
    var appender: Appender = .{
        .gpa = gpa,
        .io = io,
        .dir = ws.dir,
        .body = late_body,
        .digest = sha3(late_body),
    };

    {
        var fprng = std.Random.DefaultPrng.init(seed *% 0x27D4_EB2F);
        var faulty = faulty_dir.FaultyDir.init(io, fprng.random(), .{
            .after_op = total - 1,
            .hook = Appender.hook,
            .hook_ctx = &appender,
        });
        faulty.install();
        defer faulty_dir.FaultyDir.uninstall(); // see `sweepCut` on why this is deferred
        const outcome = try rewrite(gpa, io, faulty.io(), ws.dir, plan);
        outcome catch {};
    }

    // The scenario has to have HAPPENED, or a green here means nothing. Named
    // as its own error rather than `Io`: if a future change to `rewriteAndSwap`
    // moved the rename off the last position, `after_op = total - 1` would stop
    // naming the pre-rename window, this row would silently stop being tested,
    // and an operator reading `shun-dst: Io` would take it for flaky
    // infrastructure and retry past a coverage regression.
    if (!appender.fired or appender.failed) return error.ScenarioNotReached;

    // (iv) THE PIN: the concurrently committed artifact is still in the store,
    // and the aborted rewrite recorded no tombstone.
    var file = ws.dir.openFile(io, repo_filename, .{}) catch return error.Io;
    defer file.close(io);
    const st = file.stat(io) catch return error.Io;
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch return error.ConcurrentAppendLost;
    defer log.deinit(gpa);
    if (log.lookup(&appender.digest) == null) return error.ConcurrentAppendLost;
    if (log.shunList().len != 0) return error.AbortLeftResidue;

    // ...and the rest of the store is the pre-shun image with exactly that one
    // frame appended: the stale rewrite moved nothing.
    const after = try snapshot(gpa, io, ws.dir);
    defer gpa.free(after);
    if (after.len <= pre.len or !std.mem.eql(u8, after[0..pre.len], pre)) return error.ConcurrentAppendLost;
}

// ---------------------------------------------------------------------------
// Inline tests (join `zig build test` through dst.zig's aggregation).
// ---------------------------------------------------------------------------

test "shun_sweep: a shun+reclaim rewrite recovers to one of two stores at every cut" {
    const t = std.testing;
    var tmp = t.tmpDir(.{});
    defer tmp.cleanup();
    // Three CONSECUTIVE seeds, which is what makes the coverage claim true:
    // `buildCorpus` indexes the forced role by `seed % 3`, so seeds 1, 2 and 3
    // force the manifest, the mid-chain frame and the delta base respectively.
    // All three structural reclaims are therefore covered by `zig build test`
    // itself, not merely somewhere inside the DST's seed range.
    try run(t.allocator, t.io, tmp.dir, 1);
    try run(t.allocator, t.io, tmp.dir, 2);
    try run(t.allocator, t.io, tmp.dir, 3);
}

test "shun_sweep: packLog is byte-deterministic, so the two-image oracle has ground to stand on" {
    // The oracle compares whole files. That is only sound because a healthy pack
    // of a fixed source produces fixed bytes — no hash-map iteration order and no
    // ambient input leaks into the delta-base choice. `run` re-checks this
    // per seed; this test states it as its own claim.
    const t = std.testing;
    var tmp = t.tmpDir(.{});
    defer tmp.cleanup();
    const gpa = t.allocator;

    var prng = std.Random.DefaultPrng.init(99);
    var corpus = try buildCorpus(gpa, prng.random(), 99);
    defer corpus.deinit(gpa);
    var plan = try Plan.init(gpa, &corpus);
    defer plan.deinit(gpa);

    const a = try referenceImage(gpa, t.io, tmp.dir, &corpus, &plan, 1);
    defer gpa.free(a);
    const b = try referenceImage(gpa, t.io, tmp.dir, &corpus, &plan, 2);
    defer gpa.free(b);
    try t.expectEqualSlices(u8, a, b);
}

test "shun_sweep: the oracle rejects a store that is neither image" {
    // AN ORACLE NEVER OBSERVED FAILING IS NOT AN ORACLE. Three deliberate
    // defects, one per invariant the sweep claims, each shown to be caught.
    const t = std.testing;
    var tmp = t.tmpDir(.{});
    defer tmp.cleanup();
    const gpa = t.allocator;

    var prng = std.Random.DefaultPrng.init(7);
    var corpus = try buildCorpus(gpa, prng.random(), 7);
    defer corpus.deinit(gpa);
    var plan = try Plan.init(gpa, &corpus);
    defer plan.deinit(gpa);

    var ws = try Workspace.open(gpa, t.io, tmp.dir, 0xDEAD);
    defer ws.close(gpa);
    try seedStore(gpa, t.io, ws.dir, &corpus);
    const pre = try snapshot(gpa, t.io, ws.dir);
    defer gpa.free(pre);
    const post = try referenceImage(gpa, t.io, tmp.dir, &corpus, &plan, 0xF00D);
    defer gpa.free(post);

    // (i) a truncated store — the shape a missing pre-rename fsync leaves.
    {
        var f = try ws.dir.openFile(t.io, repo_filename, .{ .mode = .read_write });
        defer f.close(t.io);
        try f.setLength(t.io, pre.len / 2);
    }
    try t.expectError(error.HybridStore, oracle(gpa, t.io, ws.dir, &corpus, pre, post));

    // (ii) a full image, but one whose classification is a lie: the PRE bytes
    // judged as the POST state must fail, because the reclaim plainly did not
    // happen in them.
    {
        var f = try ws.dir.createFile(t.io, repo_filename, .{ .truncate = true });
        defer f.close(t.io);
        try f.writePositionalAll(t.io, pre, 0);
    }
    try t.expectError(error.ReclaimIncomplete, checkStore(gpa, t.io, ws.dir, &corpus, .post));

    // (iii) and symmetrically, the POST bytes judged as the PRE state: the
    // tombstones are there and the frames are not.
    {
        var f = try ws.dir.createFile(t.io, repo_filename, .{ .truncate = true });
        defer f.close(t.io);
        try f.writePositionalAll(t.io, post, 0);
    }
    try t.expectError(error.AbortLeftResidue, checkStore(gpa, t.io, ws.dir, &corpus, .pre));
}