Skip to content
zit Artifact [d15e1efbc7]
Login

Artifact [d15e1efbc7]

Check-ins Using Download Hex

Artifact d15e1efbc7deed10bf383b6076c810d47d27e81f87f077baadb7719f513ea57f:

//! Two-phase deterministic simulation of the sync protocol (transport-simnet-
//! plan.md §A3) — the PRIMARY DST for sync. It drives the full round loop of an
//! A0 `Client`/`Server` pair (`buildRequest` → `respond` → `consumeReply`)
//! through the A1 `SimNet`, which injects the A2 scenario's drop/dup/corrupt/
//! clog faults, and asserts the sync RFC's convergence (§3.7), termination
//! (§3.7/§3.8), and private-content (§7.5) invariants.
//!
//! The seed is the whole universe: one `std.Random.DefaultPrng.init(seed)`
//! feeds the scenario generator AND the network, in a fixed draw order, so any
//! failing sweep replays exactly — dst.zig prints `reproduce with seed=N`.
//!
//! Runs are two-phase (safety → liveness), per the project DST discipline:
//!   Phase 1 (safety): bounded rounds under injected faults. Every round probes
//!   invariants that MUST always hold — no corrupt delivery is ever accepted as
//!   content, held content never regresses, the in-flight queue stays bounded,
//!   and a receive/exchange operation never aborts. A dropped/duplicated/
//!   corrupted packet may lose a round but can never break these.
//!   Phase 2 (liveness): the network is HEALED (all fault probabilities → 0) and
//!   a fresh sync session runs to completion within a BOUNDED convergence window
//!   (rounds, not wall time). The stores MUST converge to the direction-correct
//!   artifact sets, no phantom the peer holds intact may remain on the receiving
//!   side, and the client MUST reach `terminated`. Exceeding the window is a
//!   bug, never flakiness.
//!
//! Transport model (§4 / §A1): the sans-IO boundary is a whole request/reply
//! BODY. A `SimNet` corruption stands for a body a real transport's zlib
//! container decode (§4.1) would reject, so a `Delivery.corrupted` message is
//! treated as a lost round — it is NEVER handed to `respond`/`consumeReply`.
//! That is why "no corrupt content accepted" holds by construction and a
//! corrupted round degrades to a retry rather than a spurious abort.
//!
//! Crash-during-sync composition (TEST-HARDENING §2 H3) — the `runCrashResume`
//! entry point composes the two fault domains the disjoint `sync_sweep`/
//! `storage_sweep` streams never crossed: it backs the CLIENT store with a real
//! `LogStore` over a tearable `sim/FaultyFile` (the pairing the logstore fuzz
//! targets prove) instead of the pure `MemStore`, and — between phase-1 rounds,
//! with seeded probability — power-cuts that store mid-persist (`crashRestart`),
//! reopens it, reloads the recovered prefix, and RESUMES the exchange with a
//! fresh `Client` over the same durable store rather than aborting the run. The
//! composed oracle is the existing sweep's invariants PLUS the crash-boundary
//! bite: post-recovery client contents ⊆ the pre-crash held set (recovery never
//! fabricates or resurrects content the client never held), no corrupt frame is
//! ever served as truth across the boundary, and phase-2 liveness now heals BOTH
//! domains — net faults → 0 AND disk faults → 0 — before a fresh session must
//! converge within the bounded window. The client's identity space is the sync
//! wire's 64-hex name; the log keys frames by the 32-byte digest that name
//! decodes to (`max_hash_len`), hex-encoded back on reload — the single place
//! binary-log identity becomes wire identity, mirroring cmd/sync's
//! `loadContentStore`. ONE PRNG stream drives the WHOLE universe in a fixed draw
//! order: scenario generation, then per round the SimNet faults, the durable
//! persist's commit coin, and the crash coin — so a failing composed sweep
//! replays byte-identically from the shared `reproduce with seed=N` line.
//!
//! Attribution: the composed-fault (storage × network) simulation and the
//! liveness-under-faults discipline follow TigerBeetle's VOPR — one seeded
//! simulator driving every fault domain against a single deterministic replay.
//! The crash-tear storage model reuses this repo's `sim/FaultyFile`.

const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const logstore = zit.logstore;
const net = @import("net.zig");
const sync_gen = @import("sync_gen.zig");
const faulty_file = @import("faulty_file.zig");

const assert = std.debug.assert;

/// Closed error set: transport/protocol propagations that would signal a real
/// A0/A1/A2 bug if they ever fired here, plus the named invariant violations.
/// `FatalProtocolError`/`NotImplemented` do not appear — an abort is caught and
/// judged by the outer loop, and an unimplemented reply card is a server bug
/// surfaced as `UnexpectedNotImplemented`.
pub const Error = error{
    OutOfMemory,
    PayloadTooLarge,
    QueueFull,
    RequestTooLarge,
    ResponseTooLarge,
    // invariants
    CorruptContentAccepted,
    ContentRegressed,
    InFlightUnbounded,
    UnexpectedAbort,
    UnexpectedNotImplemented,
    RoundTickBudgetExceeded,
    StoresDiverged,
    ResidualPhantom,
    NoTermination,
    ConvergenceWindowExceeded,
    PrivateContentLeaked,
    PrivateStallsTermination,
    // H3 crash-during-sync composition invariants.
    /// A crash-reopen recovered a held hash the client did not hold before the
    /// crash — recovery fabricated or resurrected content (post ⊄ pre).
    RecoveryFabricated,
    /// Reopening the durable client store after a crash failed outright, or the
    /// recovered store served a corrupt frame at reload — a torn tail must
    /// recover to a clean prefix, never to an unreadable store.
    RecoveryOpenFailed,
};

/// Phase-1 fault-window length, in rounds. Bounded: no unbounded run.
const phase1_rounds: usize = 32;
/// Phase-2 convergence window, in rounds. A clean bidirectional exchange
/// converges in a handful of rounds (discovery is single-hop — a reply walks
/// the whole store once), so this is generous slack, not a tuned minimum.
const phase2_window: usize = 64;
/// Total round budget for one seed: an explicit upper bound on the whole run.
const total_budget: usize = phase1_rounds + phase2_window;
/// §8.2 L1384 — the sweep server's per-reply clone batch cap, deliberately far
/// below the production default so a generated clone (1..max_server_only server
/// artifacts) routinely pages across MULTIPLE cfile batches. This makes the
/// primary sync DST exercise the §8.2 L1413 multi-batch continuation — the
/// client threading `clone_seqno NEXT` into successive `clone 3 NEXT` requests
/// to exhaustion — on every clone run, not just the single-batch path.
const clone_batch_max_sim: usize = 3;
/// Per-round tick bound while draining one request/reply exchange to
/// quiescence. `max_delay_ticks` + `max_clog_ticks` (both ≤ 64) bound one
/// message; this comfortably bounds the few messages of one round.
const max_round_ticks: usize = 8192;

const Endpoint = net.Endpoint;

/// Per-phase-1-round probability (percent) that the composed sweep power-cuts
/// the durable client store mid-persist and resumes. Chosen so a bounded run
/// (`phase1_rounds`) sees several crashes on average — enough to compose tears
/// with the network faults — without every round paying a reopen. The seed
/// still fixes exactly which rounds crash.
const crash_pct: u8 = 30;

/// The durable client store's crash-tear fault profile, engaged AFTER the
/// healthy initial seed lands (mirrors storage_sweep.run's post-open cfg). H3
/// composes TEARS (short/lost/torn writes) with the network faults; `crash_rot`
/// and `read_rot` stay 0 — latent bit rot on the read path is H8's dimension,
/// and keeping corruption off this store means a torn tail always recovers to a
/// clean prefix (`open`'s torn-tail truncation), never a `Corrupt` reopen.
const durable_faults: faulty_file.FaultCfg = .{
    .short_write_pct = 8,
    .lost_tail_sync_pct = 8,
    .crash_cut_at_durable_pct = 15,
    .sector_round = 512,
    .sector_round_pct = 25,
};

/// Debug-only defect injections for the prove-the-harness-can-fail discipline
/// (TEST-HARDENING §4.1). Never set on the gate path — the permanent
/// "oracle trips" test flips them to confirm the composed oracle fires.
const Inject = struct {
    /// Force a crash EVERY phase-1 round (bypass the seeded crash coin), so a
    /// defect that only bites at a crash boundary is reached deterministically.
    force_crash: bool = false,
    /// On reload, flip one byte of the first recovered held frame — model a
    /// recovery that ACCEPTS a torn frame (held hash, wrong bytes). The
    /// integrity oracle must then fire `CorruptContentAccepted`.
    torn_accept: bool = false,
};

/// The digest width the sync wire's 64-hex artifact name decodes to, and the
/// key width the client's durable `LogStore` frames by (`logstore.max_hash_len`).
const digest_len: usize = logstore.max_hash_len;

/// A durable, tearable backing for the CLIENT store: a `LogStore` over a
/// `FaultyFile`, kept as a SIDECAR to the live `Scenario.client` MemStore (which
/// stays the store the sans-IO protocol drives, so every existing helper works
/// unchanged). `persist` mirrors the live store into the log; a seeded crash
/// tears the unsynced tail, and `crashReopen` reloads the recovered prefix back
/// into a fresh `Scenario.client`. Must not move after `init` — the log borrows
/// `&self.faulty`.
const DurableClient = struct {
    faulty: faulty_file.FaultyFile,
    log: logstore.LogStore,

    /// Open an empty durable store on a HEALTHY disk (faults engage only after
    /// the initial seed is committed).
    fn init(self: *DurableClient, gpa: std.mem.Allocator, random: std.Random) Error!void {
        self.faulty = faulty_file.FaultyFile.init(gpa, random);
        self.log = logstore.LogStore.open(gpa, self.faulty.storageFile(), 0) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        };
    }

    fn deinit(self: *DurableClient, gpa: std.mem.Allocator) void {
        self.log.deinit(gpa);
        self.faulty.deinit();
        self.* = undefined;
    }

    /// Durably seed the initial client content on the healthy disk, commit it,
    /// then arm the crash-tear faults. C₀ now sits below `durable_len`, so no
    /// later tear can drop it — only content received DURING sync can tear.
    fn seedFrom(self: *DurableClient, store: repostate.ArtifactStore) Error!void {
        try self.appendHeld(store);
        self.log.commit() catch return error.RecoveryOpenFailed; // healthy disk never fails here
        self.faulty.cfg = durable_faults;
    }

    /// Append every held (content-bearing) entry of `store` into the log,
    /// idempotently (content-addressed: a repeat of an already-stored digest
    /// appends nothing). A torn/short write rolls back in the log — the same
    /// content re-persists on a later round. Phantoms are not durable (§C) and
    /// re-derive from the peer, so they are not written.
    fn appendHeld(self: *DurableClient, store: repostate.ArtifactStore) Error!void {
        const n = store.count();
        var rid: repostate.Rid = 1;
        while (rid <= n) : (rid += 1) {
            const bytes = store.content(rid) orelse continue; // phantom: not durable
            const digest = wireNameToDigest(store.hashOf(rid));
            _ = self.log.put(&digest, bytes) catch |e| switch (e) {
                error.OutOfMemory => return error.OutOfMemory,
                error.Storage => {}, // torn/short write rolled back; re-persists later
            };
        }
    }

    /// fsync the log (best-effort: a lying fsync leaves the tail tearable, which
    /// is the whole point of the crash test).
    fn commit(self: *DurableClient) void {
        self.log.commit() catch {};
    }

    /// Power-cut, reopen, and reload the recovered prefix into a FRESH
    /// `scen.client`. The crash tears the unsynced tail; `open` recovers the
    /// clean committed prefix; the reload re-derives the live MemStore the
    /// protocol drives. `scen.client`'s address is stable, so `clientStore()`
    /// stays valid across the swap.
    fn crashReopen(self: *DurableClient, gpa: std.mem.Allocator, scen: *sync_gen.Scenario, inject: Inject) Error!void {
        self.faulty.crashRestart();
        self.log.deinit(gpa);
        self.log = logstore.LogStore.open(gpa, self.faulty.storageFile(), 0) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        };
        scen.client.deinit();
        scen.client = repostate.MemStore.init(gpa);
        try loadInto(gpa, &self.log, &scen.client, inject);
    }

    /// Heal the disk for the phase-2 liveness window (net faults → 0 AND disk
    /// faults → 0).
    fn heal(self: *DurableClient) void {
        self.faulty.cfg = .{};
    }
};

/// Reload every recovered held frame of `log` into `ms`, keyed by the 64-hex
/// wire name its 32-byte digest decodes to. `ridsInPersistedOrder` skips
/// phantoms, so only content survives here. A reopened frame is CRC-validated by
/// `open`, so `content` must succeed; a `Corrupt`/`Storage` return would mean
/// recovery served an unreadable frame — a finding, surfaced as
/// `RecoveryOpenFailed`.
fn loadInto(gpa: std.mem.Allocator, log: *logstore.LogStore, ms: *repostate.MemStore, inject: Inject) Error!void {
    const rids = log.ridsInPersistedOrder(gpa) catch return error.OutOfMemory;
    defer gpa.free(rids);
    var injected = false;
    for (rids) |rid| {
        if (log.kindOf(rid) == .config) continue;
        const name = digestToWireName(log.hashOf(rid));
        const bytes = (log.content(gpa, rid) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => return error.RecoveryOpenFailed,
        }) orelse continue; // rids skip phantoms, so a null here cannot occur
        defer gpa.free(bytes);
        if (inject.torn_accept and !injected and bytes.len > 0) {
            bytes[0] +%= 1; // recovery "accepts" a torn frame: held hash, wrong bytes
            injected = true;
        }
        _ = ms.store().put(&name, bytes) catch return error.OutOfMemory;
    }
}

/// Decode a 64-hex sync wire name to its 32-byte log digest. Generator names are
/// well-formed lowercase hex by construction, so a decode failure is impossible.
fn wireNameToDigest(name: []const u8) [digest_len]u8 {
    assert(name.len == 2 * digest_len);
    var out: [digest_len]u8 = undefined;
    _ = std.fmt.hexToBytes(&out, name) catch unreachable;
    return out;
}

/// Encode a 32-byte log digest back to its 64-hex wire name.
fn digestToWireName(digest: []const u8) [2 * digest_len]u8 {
    assert(digest.len == digest_len);
    return std.fmt.bytesToHex(digest[0..digest_len].*, .lower);
}

/// The seeded phase-1 crash coin. Drawn every phase-1 round (durable path only),
/// so the stream stays fixed regardless of outcome.
fn drawCrash(random: std.Random) bool {
    return random.uintLessThan(u8, 100) < crash_pct;
}

// ----------------------------------------------------------------------------
// Public entry point
// ----------------------------------------------------------------------------

/// Run the two-phase sweep for one seed, asserting every invariant. Returns on
/// success; a violation returns the named error. dst.zig calls this per seed in
/// its `zig build dst` gate loop, so `--seed N`/`--seeds K` select depth.
pub fn runSeed(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try simulate(gpa, seed, .{});
}

/// H3 (TEST-HARDENING §2) — the crash-during-sync composition: the same
/// two-phase sweep, but with the client store backed by a real `LogStore` over a
/// tearable `FaultyFile` and seeded mid-persist crash+resume between phase-1
/// rounds. dst.zig calls this per seed in the same gate loop as `runSeed`, off
/// the shared reproduce line.
pub fn runCrashResume(gpa: std.mem.Allocator, seed: u64) Error!void {
    _ = try simulateImpl(gpa, seed, .{}, true, .{});
}

/// The observable result of one run: the client's terminal state and a stable
/// digest of both converged stores — two runs of the same seed hash equal
/// (the reproducibility oracle).
const Outcome = struct { state: sync.ClientState, digest: u64 };

// ----------------------------------------------------------------------------
// The two-phase driver
// ----------------------------------------------------------------------------

/// Generate the scenario, then run phase 1 (faults) and phase 2 (healed),
/// probing safety every round and asserting liveness + convergence at the end.
/// The pure-`MemStore` two-phase sweep; `simulateImpl` adds the H3 durable
/// composition without perturbing this path's PRNG stream.
fn simulate(gpa: std.mem.Allocator, seed: u64, params: sync_gen.Params) Error!Outcome {
    return simulateImpl(gpa, seed, params, false, .{});
}

/// The two-phase driver. `durable` backs the client store with a `LogStore` over
/// a tearable `FaultyFile` and injects seeded mid-persist crash+resume between
/// phase-1 rounds (H3); `inject` carries the prove-the-harness-can-fail defect
/// switches. On the non-durable path NO extra bit is drawn, so the existing gate
/// stream is byte-identical.
fn simulateImpl(gpa: std.mem.Allocator, seed: u64, params: sync_gen.Params, durable: bool, inject: Inject) Error!Outcome {
    var prng = std.Random.DefaultPrng.init(seed);
    const random = prng.random();

    var scen = try sync_gen.generate(gpa, random, params);
    defer scen.deinit();

    // Scratch for the run's oracles: the content-integrity reference and the
    // initial held sets. One arena, freed at return.
    var arena_state = std.heap.ArenaAllocator.init(gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const ref = try buildReference(arena, &scen);
    const c0 = try heldHashes(arena, scen.clientStore());
    const s0 = try heldHashes(arena, scen.serverStore());

    // H3: a durable, tearable sidecar for the client store, seeded from the t0
    // client content on a healthy disk, then armed with crash-tear faults. It
    // stays a sidecar to the live `scen.client`, so every helper below drives
    // the same store it always did.
    var dc: DurableClient = undefined;
    var dc_ptr: ?*DurableClient = null;
    if (durable) {
        try dc.init(gpa, random);
        try dc.seedFrom(scen.clientStore());
        dc_ptr = &dc;
    }
    defer if (dc_ptr) |d| d.deinit(gpa);

    var netp = try net.SimNet.init(gpa, random);
    defer netp.deinit();
    netp.links[@intFromEnum(Endpoint.b)] = scen.link_to_b; // client → server
    netp.links[@intFromEnum(Endpoint.a)] = scen.link_to_a; // server → client

    // §6.4 — the server holds the same symmetric shared secret the client signs
    // with, so a named login verifies (a mismatch would be an in-band error
    // card, never a returned error).
    // §8.2/§9.1 — the server carries the clone reply's SERVERCODE/PROJECTCODE.
    // The project code matches the client's, so a clone's `push` card is a
    // well-formed three-token card the peer accepts (an empty field would abort
    // it); servercode is informational.
    const server = sync.Server{
        .version = scen.version,
        .credential = scen.credential,
        .self_servercode = scen.selfServercode(),
        .project_code = scen.projectCode(),
        .clone_batch_max = clone_batch_max_sim,
    };

    var req_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer req_buf.deinit(gpa);
    var rep_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer rep_buf.deinit(gpa);

    var client = newClient(&scen);
    defer client.deinit(gpa);

    var prev_c: usize = c0.len;
    var prev_s: usize = s0.len;
    var phase: u8 = 1;
    var phase2_start: usize = 0;
    var round: usize = 0;
    while (round < total_budget) : (round += 1) {
        assert(round < total_budget); // explicit bound: no unbounded run

        // Phase boundary: heal the wire and start a fresh session for the clean
        // convergence run (a real re-invocation of `zit sync`; the durable
        // stores carry phase-1 progress, phantoms re-derive from the store).
        if (phase == 1 and round == phase1_rounds) {
            netp.heal();
            // H3 phase-2 heals BOTH domains: persist+commit the phase-1 progress
            // durably, heal the disk, then one clean crash+reopen so the healed
            // session starts from the fully-durable recovered store.
            if (dc_ptr) |d| {
                d.heal();
                try d.appendHeld(scen.clientStore());
                d.commit();
                try d.crashReopen(gpa, &scen, .{}); // committed: nothing to tear
                prev_c = heldCount(scen.clientStore());
            }
            phase = 2;
            phase2_start = round;
            client.deinit(gpa);
            client = newClient(&scen);
        }

        if (client.state == .terminated or client.state == .aborted) {
            if (phase == 2) break; // the healed session reached its terminal
            // A phase-1 session ended (possibly a packet-loss-induced early
            // stop): re-invoke a fresh session over the same durable stores.
            client.deinit(gpa);
            client = newClient(&scen);
        }

        try runRound(gpa, &netp, &client, &server, &scen, &req_buf, &rep_buf);

        // H3 crash-during-sync: persist what this round received into the durable
        // client store, then — with a seeded coin — power-cut mid-persist and
        // resume. Only phase 1 crashes; phase 2 is the clean convergence window.
        if (dc_ptr) |d| {
            if (phase == 1) {
                try d.appendHeld(scen.clientStore());
                if (random.boolean()) d.commit(); // leave an uncommitted tail to tear
                if (inject.force_crash or drawCrash(random)) {
                    const pre = try heldHashes(arena, scen.clientStore());
                    try d.crashReopen(gpa, &scen, inject);
                    const post = try heldHashes(arena, scen.clientStore());
                    // The crash-boundary bite: recovery holds no hash the client
                    // did not hold before the cut (post ⊆ pre — no fabrication)…
                    if (!subsetSorted(post, pre)) return error.RecoveryFabricated;
                    // …and serves no corrupt frame as content across the boundary.
                    try probeIntegrity(scen.clientStore(), ref);
                    // A crash legitimately drops the unsynced tail, so the
                    // monotonic held baseline resets to the recovered count (it
                    // stays monotonic WITHIN the resumed session).
                    prev_c = heldCount(scen.clientStore());
                    client.deinit(gpa);
                    client = newClient(&scen); // resume: fresh session, same store
                }
            }
        }

        try probeRoundSafety(&scen, ref, &netp, &client, &prev_c, &prev_s);
    }

    // Liveness: a terminal state must be reached within the phase-2 window.
    assert(phase == 2);
    if (client.state != .terminated and client.state != .aborted) return error.ConvergenceWindowExceeded;
    assert(round - phase2_start <= phase2_window);

    try checkTermination(&client, scen.operation);
    try checkConvergence(arena, &scen, c0, s0);
    return .{ .state = client.state, .digest = try outcomeDigest(arena, &scen, client.state) };
}

/// The per-round safety block, probed EVERY round of both phases: no corrupt
/// content accepted, held content never regresses, the in-flight queue stays
/// bounded, and (non-clone) the client never spuriously aborts. `prev_*` carry
/// the monotonic held counts across rounds.
fn probeRoundSafety(
    scen: *sync_gen.Scenario,
    ref: Reference,
    netp: *const net.SimNet,
    client: *const sync.Client,
    prev_c: *usize,
    prev_s: *usize,
) Error!void {
    try probeIntegrity(scen.clientStore(), ref);
    try probeIntegrity(scen.serverStore(), ref);
    const ch = heldCount(scen.clientStore());
    const sh = heldCount(scen.serverStore());
    if (ch < prev_c.* or sh < prev_s.*) return error.ContentRegressed; // content never un-stored
    prev_c.* = ch;
    prev_s.* = sh;
    if (netp.inFlight() > net.SimNet.max_inflight) return error.InFlightUnbounded;
    try probeState(client, scen.operation);
}

/// §3.7/§3.8 termination: every operation, clone included, MUST reach
/// `terminated`. The server now answers §8 clone (paginated cfile batches with a
/// `clone_seqno` cursor to exhaustion), so a clone runs the round loop to a clean
/// §3.7 termination just like pull/push/sync — no operation aborts under
/// clean-or-filtered traffic.
fn checkTermination(client: *const sync.Client, op: sync.Operation) Error!void {
    switch (op) {
        .pull, .push, .sync, .clone => if (client.state != .terminated) return error.NoTermination,
    }
}

/// A fresh client for `scen`'s operation: state seeded from §3.8's initial-state
/// map, identity/version/credential from the scenario. The phantom set and send
/// buffer default empty and re-derive from the durable store on each session.
fn newClient(scen: *const sync_gen.Scenario) sync.Client {
    return .{
        .state = sync.initialState(scen.operation),
        .operation = scen.operation,
        .self_servercode = scen.selfServercode(),
        .project_code = scen.projectCode(),
        .version = scen.version,
        .credential = scen.credential,
    };
}

/// One synchronous request/reply exchange, drained to quiescence: emit the
/// request, drive the net, `respond` to each intact request delivery, and
/// `consumeReply` the first intact reply (later duplicates are ignored — the
/// server is idempotent, §3.8). A corrupted delivery is a transport reject (a
/// lost round). Bounded by `max_round_ticks`.
fn runRound(
    gpa: std.mem.Allocator,
    netp: *net.SimNet,
    client: *sync.Client,
    server: *const sync.Server,
    scen: *sync_gen.Scenario,
    req_buf: *std.ArrayListUnmanaged(u8),
    rep_buf: *std.ArrayListUnmanaged(u8),
) Error!void {
    assert(client.state == .exchange or client.state == .clone_round0);
    const client_store = scen.clientStore();
    const server_store = scen.serverStore();

    const req = try client.buildRequest(gpa, client_store, req_buf);
    try netp.enqueue(.b, req);

    var consumed = false;
    var ticks: usize = 0;
    while (netp.inFlight() > 0) {
        while (netp.deliver()) |d| {
            switch (d.dst) {
                .b => {
                    // Request at the server. A corrupt body a real transport
                    // would reject before decode never reaches `respond`.
                    if (d.corrupted) continue;
                    const reply = try server.respond(gpa, server_store, d.payload, rep_buf);
                    try netp.enqueue(.a, reply);
                },
                .a => {
                    // Reply at the client.
                    if (d.corrupted) continue; // transport reject: a lost round
                    if (consumed) continue; // idempotent dup: consume exactly one
                    if (client.state != .exchange and client.state != .clone_round0) continue;
                    if (client.consumeReply(gpa, client_store, d.payload)) |obs| {
                        consumed = true;
                        // §8.2 — carry the received clone sequence number into
                        // the next request's cursor so clone pages to exhaustion.
                        if (client.operation == .clone) client.clone_seqno = obs.clone_seqno_next;
                    } else |e| switch (e) {
                        error.OutOfMemory => return error.OutOfMemory,
                        // The server emits only full-content file/cfile cards; a
                        // NotImplemented (delta/uv form) reply would be a server
                        // bug, not wire noise.
                        error.NotImplemented => return error.UnexpectedNotImplemented,
                        // §3.8 r18 — the client aborted (`state == .aborted`).
                        // The outer loop judges whether that is legal for the op.
                        error.FatalProtocolError => consumed = true,
                    }
                },
            }
        }
        netp.tick();
        ticks += 1;
        if (ticks > max_round_ticks) return error.RoundTickBudgetExceeded;
    }
}

/// Legal-state + no-spurious-abort probe. Any operation reaching `aborted` under
/// clean-or-filtered traffic implicates `consumeReply` or the A1 corruption
/// filter, so it is a hard failure — clone included now that the server serves
/// §8 clone and the clone client runs to a clean termination.
fn probeState(client: *const sync.Client, op: sync.Operation) Error!void {
    _ = op;
    assert(client.state != .init); // negative space: never regresses to pre-start
    if (client.state == .aborted) return error.UnexpectedAbort;
}

// ----------------------------------------------------------------------------
// Safety oracle: content integrity
// ----------------------------------------------------------------------------

/// hash → the one true content for that hash. Every artifact is content-
/// addressed, so any bytes that appear under a hash must equal these.
const Reference = std.StringHashMapUnmanaged([]const u8);

/// Snapshot every held (hash, content) pair from both initial stores into
/// `arena`. Shared artifacts carry identical bytes on both sides (generator
/// contract), so the first sighting wins.
fn buildReference(arena: std.mem.Allocator, scen: *sync_gen.Scenario) Error!Reference {
    var ref: Reference = .empty;
    try indexReference(arena, scen.clientStore(), &ref);
    try indexReference(arena, scen.serverStore(), &ref);
    return ref;
}

fn indexReference(arena: std.mem.Allocator, store: repostate.ArtifactStore, ref: *Reference) Error!void {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |bytes| {
            const h = store.hashOf(rid);
            if (!ref.contains(h)) {
                try ref.put(arena, try arena.dupe(u8, h), try arena.dupe(u8, bytes));
            }
        }
    }
}

/// No corrupt/unknown content was ever interned as truth: every held entry's
/// bytes equal the content-addressed reference for its hash. A held hash absent
/// from the reference (a fabricated artifact) is equally a violation.
fn probeIntegrity(store: repostate.ArtifactStore, ref: Reference) Error!void {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |bytes| {
            const want = ref.get(store.hashOf(rid)) orelse return error.CorruptContentAccepted;
            if (!std.mem.eql(u8, bytes, want)) return error.CorruptContentAccepted;
        }
    }
}

// ----------------------------------------------------------------------------
// Convergence oracle (§3.7)
// ----------------------------------------------------------------------------

/// §3.7 — assert the healed run converged, per operation direction:
///   sync : both stores hold C₀ ∪ S₀, and neither retains a phantom.
///   pull : the client (receiver) holds C₀ ∪ S₀ with no phantom; the server's
///          content is unchanged (it keeps dangling phantoms for client-only
///          content it can never pull — expected, one-directional).
///   push : the server (receiver) holds C₀ ∪ S₀ with no phantom; the client
///          keeps at least its own content and no phantom.
/// §3.7(a): no residual phantom whose content the peer holds intact remains on
/// the receiving side(s) — asserted via `noPhantom`.
fn checkConvergence(
    arena: std.mem.Allocator,
    scen: *sync_gen.Scenario,
    c0: []const []const u8,
    s0: []const []const u8,
) Error!void {
    const client_now = try heldHashes(arena, scen.clientStore());
    const server_now = try heldHashes(arena, scen.serverStore());
    const uni = try unionSorted(arena, c0, s0);
    switch (scen.operation) {
        .sync => {
            if (!eqlSorted(client_now, uni) or !eqlSorted(server_now, uni)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
        },
        .pull => {
            if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
            if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore())) return error.ResidualPhantom;
        },
        .push => {
            if (!eqlSorted(server_now, uni)) return error.StoresDiverged;
            if (!subsetSorted(c0, client_now)) return error.StoresDiverged;
            if (!noPhantom(scen.serverStore()) or !noPhantom(scen.clientStore())) return error.ResidualPhantom;
        },
        .clone => {
            // §8 — the client (which starts empty, generator contract) fetches
            // the server's whole repository over paginated cfile batches, so it
            // ends holding C₀ ∪ S₀ (= S₀) with no phantom; the server is
            // read-only during clone and ends unchanged at S₀.
            if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
            if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
            if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
        },
    }
}

/// A stable digest of both stores' held sets plus the terminal state — the
/// reproducibility oracle.
fn outcomeDigest(arena: std.mem.Allocator, scen: *sync_gen.Scenario, state: sync.ClientState) Error!u64 {
    var h = std.hash.Wyhash.init(0);
    h.update(&.{@intFromEnum(state)});
    for (try heldHashes(arena, scen.clientStore())) |x| h.update(x);
    h.update("|");
    for (try heldHashes(arena, scen.serverStore())) |x| h.update(x);
    return h.final();
}

// ----------------------------------------------------------------------------
// Held-set helpers (sorted []const u8 sets)
// ----------------------------------------------------------------------------

fn heldCount(store: repostate.ArtifactStore) usize {
    var count: usize = 0;
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid) != null) count += 1;
    }
    return count;
}

/// True iff no rid in `store` is a phantom (content-absent). §3.3 r24 — a
/// phantom is exactly `content(rid) == null`.
fn noPhantom(store: repostate.ArtifactStore) bool {
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid) == null) return false;
    }
    return true;
}

/// The held hashes of `store`, duped into `arena` and sorted, so sets compare
/// across runs that assign rids differently.
fn heldHashes(arena: std.mem.Allocator, store: repostate.ArtifactStore) Error![][]const u8 {
    var list: std.ArrayListUnmanaged([]const u8) = .empty;
    const n = store.count();
    var rid: repostate.Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (store.content(rid)) |_| try list.append(arena, try arena.dupe(u8, store.hashOf(rid)));
    }
    const slice = try list.toOwnedSlice(arena);
    std.mem.sort([]const u8, slice, {}, lessHash);
    return slice;
}

fn lessHash(_: void, a: []const u8, b: []const u8) bool {
    return std.mem.order(u8, a, b) == .lt;
}

/// Two sorted, duplicate-free sets are equal iff same length and elementwise.
fn eqlSorted(a: []const []const u8, b: []const []const u8) bool {
    if (a.len != b.len) return false;
    for (a, b) |x, y| if (!std.mem.eql(u8, x, y)) return false;
    return true;
}

/// Every element of sorted `sub` appears in sorted `sup` (two-pointer merge).
fn subsetSorted(sub: []const []const u8, sup: []const []const u8) bool {
    var i: usize = 0;
    for (sup) |s| {
        if (i == sub.len) break;
        if (std.mem.eql(u8, sub[i], s)) i += 1;
    }
    return i == sub.len;
}

/// Merge two sorted, duplicate-free sets into their sorted union in `arena`.
fn unionSorted(arena: std.mem.Allocator, a: []const []const u8, b: []const []const u8) Error![][]const u8 {
    var list: std.ArrayListUnmanaged([]const u8) = .empty;
    var i: usize = 0;
    var j: usize = 0;
    while (i < a.len or j < b.len) {
        if (j == b.len) {
            try list.append(arena, a[i]);
            i += 1;
        } else if (i == a.len) {
            try list.append(arena, b[j]);
            j += 1;
        } else switch (std.mem.order(u8, a[i], b[j])) {
            .lt => {
                try list.append(arena, a[i]);
                i += 1;
            },
            .gt => {
                try list.append(arena, b[j]);
                j += 1;
            },
            .eq => {
                try list.append(arena, a[i]);
                i += 1;
                j += 1;
            },
        }
    }
    return list.toOwnedSlice(arena);
}

// ----------------------------------------------------------------------------
// §7.5 private-content oracle
// ----------------------------------------------------------------------------

/// §7.5 — feed a client a server reply carrying a private artifact (a `private`
/// card then a full-content `file`) and report the receive-side outcome:
/// whether it was discarded, whether the store now holds it, and whether the
/// round still reached `terminated`. `granted` is the receiver's Private
/// capability (§7.5 r4/r5).
const PrivateOutcome = struct { discarded: bool, held: bool, terminated: bool };

fn drivePrivateReply(gpa: std.mem.Allocator, granted: bool) Error!PrivateOutcome {
    var store = repostate.MemStore.init(gpa);
    defer store.deinit();
    var client = sync.Client{ .state = .exchange, .operation = .sync, .private_granted = granted };
    defer client.deinit(gpa);

    const hash = "aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00";
    const content = "private-artifact-bytes";
    var reply: std.ArrayListUnmanaged(u8) = .empty;
    defer reply.deinit(gpa);
    try reply.appendSlice(gpa, "private\n"); // §7.5 L2359 — mark the next file private
    const header = try std.fmt.allocPrint(gpa, "file {s} {d}\n", .{ hash, content.len });
    defer gpa.free(header);
    try reply.appendSlice(gpa, header);
    try reply.appendSlice(gpa, content);

    const obs = client.consumeReply(gpa, store.store(), reply.items) catch |e| switch (e) {
        error.OutOfMemory => return error.OutOfMemory,
        // A well-formed reply must parse; either error would itself be the §7.5
        // leak/failure this oracle is here to catch.
        error.FatalProtocolError, error.NotImplemented => return error.PrivateContentLeaked,
    };
    const rid = store.store().lookup(hash);
    const held = rid != null and store.store().content(rid.?) != null;
    return .{
        .discarded = obs.private_content_discarded,
        .held = held,
        .terminated = client.state == .terminated,
    };
}

// ----------------------------------------------------------------------------
// Tests — the invariants ARE the oracle: they encode §3.7 convergence/
// termination and §7.5 privacy. Seeds are literals; the seed reproduces a run.
// ----------------------------------------------------------------------------

const testing = std.testing;

/// A fixed smoke set: a handful of seeds through the full two-phase sweep.
/// Small mix of perfect-wire and heavy-fault scenarios (default `Params`), so
/// safety-under-faults AND liveness-after-heal are both exercised.
const smoke_seeds = [_]u64{ 1, 2, 3, 4, 5, 6, 7, 8 };

test "sync_sweep §3.7: the smoke sweep converges and terminates over every seed" {
    for (smoke_seeds) |seed| try runSeed(testing.allocator, seed);
}

test "sync_sweep §3.7: a forced perfect wire converges and terminates" {
    // No faults in either phase: the pure convergence path must still hold.
    for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, .{ .perfect_wire_pct = 100 });
}

test "sync_sweep §3.7: a heavy-fault scenario converges once the wire heals" {
    const heavy = sync_gen.Params{
        .perfect_wire_pct = 0,
        .max_drop_pct = 40,
        .max_dup_pct = 40,
        .max_corrupt_pct = 25,
        .max_clog_pct = 25,
        .max_mean_delay = 8,
    };
    for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, heavy);
}

test "sync_sweep: the same seed reproduces an identical run" {
    for (smoke_seeds) |seed| {
        const a = try simulate(testing.allocator, seed, .{});
        const b = try simulate(testing.allocator, seed, .{});
        try testing.expectEqual(a.state, b.state);
        try testing.expectEqual(a.digest, b.digest);
    }
}

test "sync_sweep §8.2 L1384-1413: a clone whose repository spans multiple cfile batches converges in full" {
    const gpa = testing.allocator;
    // The sweep server pages clone at `clone_batch_max_sim`. Find seeds whose
    // generated clone holds MORE server artifacts than one batch carries — the
    // multi-batch path — then drive the full two-phase sweep on that same seed.
    // `simulate` regenerates the identical scenario (deterministic from the
    // seed), so a classified multi-batch seed runs the very run classified.
    var multi_batch_clones: usize = 0;
    var seed: u64 = 0;
    while (seed < 96) : (seed += 1) {
        var prng = std.Random.DefaultPrng.init(seed);
        var scen = try sync_gen.generate(gpa, prng.random(), .{});
        const is_multi = scen.operation == .clone and scen.composition.serverHeld() > clone_batch_max_sim;
        scen.deinit();
        if (!is_multi) continue;
        multi_batch_clones += 1;
        // §8.2 L1413 / §3.7(e) — the whole repository clones across successive
        // batches and reaches a clean terminal; a truncating client would leave
        // the store short of S₀ and fail `checkConvergence`.
        _ = try simulate(gpa, seed, .{});
    }
    // The case must actually be covered: at least one multi-batch clone in range.
    try testing.expect(multi_batch_clones >= 1);
}

test "sync_sweep §7.5(a)(b): private content is not leaked and does not stall termination" {
    // (a) L2383-2385 — without the Private capability the artifact is discarded,
    // never stored, and (b) the discard sets no §3.7 continuation flag, so the
    // round still terminates.
    const denied = try drivePrivateReply(testing.allocator, false);
    try testing.expect(denied.discarded);
    try testing.expect(!denied.held); // §7.5(a): not interned — no leak into the store
    try testing.expect(denied.terminated); // §7.5(b): discard did not stall §3.7 termination

    // Negative space (L2381-2385) — with the capability the SAME reply IS stored
    // and is not discarded, proving the gate is a real decision, not always-drop.
    const allowed = try drivePrivateReply(testing.allocator, true);
    try testing.expect(!allowed.discarded);
    try testing.expect(allowed.held);
    try testing.expect(allowed.terminated);
}

// ----------------------------------------------------------------------------
// H3 (TEST-HARDENING §2) — crash-during-sync composition (storage × network).
// The composed sweep survives every existing invariant AND the crash-boundary
// bite; the seed still replays a run exactly.
// ----------------------------------------------------------------------------

test "sync_sweep H3: the composed storage×network sweep converges across crashes" {
    // The client store is a real LogStore over a tearable FaultyFile, power-cut
    // mid-persist between phase-1 rounds; every seed must still converge and
    // terminate once BOTH domains heal.
    for (smoke_seeds) |seed| try runCrashResume(testing.allocator, seed);
}

test "sync_sweep H3: the same seed reproduces an identical composed run" {
    // One PRNG universe: generation, network faults, the persist commit coin, and
    // the crash coin all draw from it in a fixed order, so a composed run is
    // byte-reproducible from its seed.
    for (smoke_seeds) |seed| {
        const a = try simulateImpl(testing.allocator, seed, .{}, true, .{});
        const b = try simulateImpl(testing.allocator, seed, .{}, true, .{});
        try testing.expectEqual(a.state, b.state);
        try testing.expectEqual(a.digest, b.digest);
    }
}

test "sync_sweep H3: the oracle catches a recovery that accepts a torn frame" {
    // Prove-the-harness-can-fail (TEST-HARDENING §4.1): under the debug switches,
    // every phase-1 round crashes and the reload flips one byte of a recovered
    // held frame — modelling a recovery that ACCEPTS a torn frame (held hash,
    // wrong bytes). The composed oracle MUST fire; a green run here would mean the
    // integrity/convergence check is vacuous. Scan a few seeds so at least one has
    // client content to corrupt at a crash boundary.
    var caught = false;
    var seed: u64 = 1;
    while (seed <= 12 and !caught) : (seed += 1) {
        if (simulateImpl(testing.allocator, seed, .{}, true, .{ .force_crash = true, .torn_accept = true })) |_| {
            // Converged despite the injected torn frame: the oracle missed it.
        } else |_| {
            caught = true; // an Error fired — the oracle is live
        }
    }
    try testing.expect(caught);
}