Artifact [e88aecb627]

Artifact e88aecb627641a80b092b529f9736982a66d5088e16c0cf7c28643289e53d5c1:

• File src/sync_driver_test.zig — part of check-in [9033c13f21] at 2026-09-17 17:04:25 on branch trunk — Remove issue-tracker numbers from source and documents

  The tracker is disposable and does not live with the source, so a number is a dangling pointer. Every zit#NN and #NN reference in code comments, test names, help text, smoke labels, identifiers, release notes and errata is replaced by enough context to stand alone, or by a short plain statement of the problem. No behaviour changed. (user: dmestas, size: 42200) [annotate] [blame] [check-ins using]

//! Unit tests for the sync round driver (`sync_driver.runSync`).
//!
//! The transport is an IN-PROCESS fake: its `exchange` runs the sans-IO
//! `sync.Server.respond` over a server-side `MemStore`, so a full round loop
//! runs through `runSync` exactly as the A3 sweep (`sim/sync_sweep.zig`) drives
//! `buildRequest → respond → consumeReply` — but here through the production
//! driver, with no simulated network. Expected values are the sync RFC's
//! §3.7 convergence / §3.8 termination invariants (mirroring the sweep's
//! `checkConvergence` / `checkTermination`), never observed from fossil.

const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const sync_driver = @import("sync_driver.zig");

const testing = std.testing;

/// A generous convergence bound: a clean bidirectional exchange resolves in a
/// handful of rounds (discovery is single-hop — a reply walks the peer store
/// once), so this is slack, not a tuned minimum. Mirrors the sweep's window.
const round_bound: usize = 64;

/// §9.1 — the client's own 16-hex repository identifiers, emitted in the
/// pull/push operation cards. Any well-formed pair works: the stateless server
/// echoes the peer's codes and never gates on a configured project code.
const servercode = "0123456789abcdef";
const projectcode = "fedcba9876543210";

/// A fresh sync-operation client carrying the identifiers above.
fn newClient() sync.Client {
    return .{
        .state = .exchange,
        .operation = .sync,
        .self_servercode = servercode,
        .project_code = projectcode,
    };
}

// ---------------------------------------------------------------------------
// In-process fakes satisfying the driver's `exchange(gpa, request) -> ![]u8`
// ---------------------------------------------------------------------------

/// The convergence fake: `exchange` answers each request with the sans-IO
/// `Server.respond` over a durable server store. `respond` writes into a reused
/// buffer, so the returned slice is duped into a fresh caller-owned allocation
/// (the driver frees it) — matching the real transport's "caller owns" contract.
const ServerTransport = struct {
    /// §6.5 — the driver hands every round's cookie to the transport;
    /// this double runs sans-HTTP, so the credential has no header to ride.
    pub fn setLoginCookie(self: *@This(), value: ?[]const u8) void {
        _ = self;
        _ = value;
    }

    server: sync.Server,
    store: repostate.ArtifactStore,
    reply_buf: std.ArrayListUnmanaged(u8) = .empty,

    pub const Error = sync.Server.RespondError; // {OutOfMemory, ResponseTooLarge}

    fn deinit(self: *ServerTransport, gpa: std.mem.Allocator) void {
        self.reply_buf.deinit(gpa);
    }

    pub fn exchange(self: *ServerTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        const reply = try self.server.respond(gpa, self.store, request, &self.reply_buf);
        return gpa.dupe(u8, reply);
    }
};

/// The round-limit fake: `exchange` announces a fresh phantom (`igot HASH`) on
/// every call, so §3.7(a) versioned-progress continuation holds every round and
/// the client never terminates — the driver's stall guard must catch it.
///
/// This is precisely the shape the guard has to recognise, and the reason it
/// cannot simply ask "did anything change": something changes every round here
/// (a new phantom), yet no content EVER arrives. Announcement is not delivery.
const NeverEndsTransport = struct {
    /// §6.5 — the driver hands every round's cookie to the transport;
    /// this double runs sans-HTTP, so the credential has no header to ride.
    pub fn setLoginCookie(self: *@This(), value: ?[]const u8) void {
        _ = self;
        _ = value;
    }

    counter: usize = 0,

    pub const Error = error{OutOfMemory};

    pub fn exchange(self: *NeverEndsTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        _ = request;
        self.counter += 1;
        // A distinct 64-hex hash per round → a brand-new client phantom each
        // round → §3.7(a) never stops (the content is never delivered).
        return std.fmt.allocPrint(gpa, "igot {x:0>64}\n", .{self.counter});
    }
};

/// The many-rounds fake: a peer with more to send than fits in one round, which
/// delivers steadily and terminates. Each round announces one new artifact and
/// delivers the one announced last round, so every round after the first moves
/// bytes; after `announce_rounds` it stops announcing, drains the last delivery,
/// and lets the phantom table empty so §3.7(a) stops continuing.
///
/// This is the shape a large real repository has: converging perfectly, one
/// bounded batch at a time, for as many rounds as its size requires. A guard
/// that counted total rounds would abort it partway through and blame the peer
/// for failing to converge — the exact failure a 1,166,689-artifact store hit
/// against the old flat cap of 1024 rounds.
const ManyRoundsTransport = struct {
    /// §6.5 — the driver hands every round's cookie to the transport;
    /// this double runs sans-HTTP, so the credential has no header to ride.
    pub fn setLoginCookie(self: *@This(), value: ?[]const u8) void {
        _ = self;
        _ = value;
    }

    announce_rounds: usize,
    round: usize = 0,

    pub const Error = error{OutOfMemory};

    pub fn exchange(self: *ManyRoundsTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        _ = request;
        self.round += 1;
        var out: std.ArrayListUnmanaged(u8) = .empty;
        errdefer out.deinit(gpa);
        if (self.round <= self.announce_rounds) {
            const h = hashN(self.round);
            try out.appendSlice(gpa, "igot ");
            try out.appendSlice(gpa, &h);
            try out.append(gpa, '\n');
        }
        // Deliver what the previous round announced (§3.3 content delivery).
        if (self.round > 1 and self.round <= self.announce_rounds + 1) {
            const prev = hashN(self.round - 1);
            try out.appendSlice(gpa, "file ");
            try out.appendSlice(gpa, &prev);
            try out.appendSlice(gpa, " 7\ncontent");
        }
        return out.toOwnedSlice(gpa);
    }
};

// ---------------------------------------------------------------------------
// Store seeding + held-set oracle (arbitrary but distinct 64-hex hashes: the
// protocol treats a hash as an opaque token at this layer, as the sweep does)
// ---------------------------------------------------------------------------

/// A distinct lowercase-64-hex hash for a small id — no crypto needed; the
/// stores only require both sides to agree on the hash→content mapping.
fn hashN(id: usize) [64]u8 {
    var buf: [64]u8 = undefined;
    @memset(&buf, 'a');
    const hexd = "0123456789abcdef";
    buf[0] = hexd[id & 0xf];
    buf[1] = hexd[(id >> 4) & 0xf];
    return buf;
}

/// `hashN` varies only two nibbles, so it yields 256 distinct names — fine for
/// the small fixtures above, but not for a corpus that has to outnumber the
/// round cap this file also tests. Four nibbles carry 65,536.
fn hashWide(id: usize) [64]u8 {
    var buf: [64]u8 = undefined;
    @memset(&buf, 'a');
    const hexd = "0123456789abcdef";
    for (0..4) |i| buf[i] = hexd[(id >> @intCast(4 * i)) & 0xf];
    return buf;
}

/// The sorted set of held (content-bearing) hashes in `store`, duped into `a`.
fn heldHashes(a: std.mem.Allocator, store: repostate.ArtifactStore) ![][]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(a, try a.dupe(u8, store.hashOf(rid)));
    }
    const slice = try list.toOwnedSlice(a);
    std.mem.sort([]const u8, slice, {}, lessHash);
    return slice;
}

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

/// §3.3 r24 — a phantom is exactly a content-absent rid. True iff none remain.
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;
}

/// Build a client/server store pair: `shared` artifacts on both, `client_only`
/// held by the client and interned as a server-side phantom, `server_only` held
/// by the server and interned as a client-side phantom. This is the sweep's
/// three-way composition — the general convergence shape, not a special case.
fn seedStores(
    client_store: repostate.ArtifactStore,
    server_store: repostate.ArtifactStore,
    shared: usize,
    client_only: usize,
    server_only: usize,
) !void {
    var id: usize = 0;
    var k: usize = 0;
    while (k < shared) : (k += 1) {
        const h = hashN(id);
        _ = try client_store.put(&h, "shared-content");
        _ = try server_store.put(&h, "shared-content");
        id += 1;
    }
    k = 0;
    while (k < client_only) : (k += 1) {
        const h = hashN(id);
        _ = try client_store.put(&h, "client-only-content");
        _ = try server_store.intern(&h); // server knows it only as a phantom
        id += 1;
    }
    k = 0;
    while (k < server_only) : (k += 1) {
        const h = hashN(id);
        _ = try server_store.put(&h, "server-only-content");
        _ = try client_store.intern(&h); // client knows it only as a phantom
        id += 1;
    }
}

/// A fresh clone-operation client: begins in `clone_round0` (§3.8), with no
/// credential or codes — the first clone round sends no login card (§8.3) and
/// the clone operation emits no push/pull operation card (§8.2).
fn newCloneClient() sync.Client {
    return .{ .state = .clone_round0, .operation = .clone };
}

/// A fresh pull-operation client carrying the §9.1 identifiers: begins mid-
/// exchange (`initialState(.pull)`), announces its holdings via igot, and
/// obtains but never uploads content (§3.3 r24). The receive-only counterpart of
/// `newClient` (which is bidirectional sync).
fn newPullClient() sync.Client {
    return .{
        .state = .exchange,
        .operation = .pull,
        .self_servercode = servercode,
        .project_code = projectcode,
    };
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

test "sync_driver §8.2: a clone needing more rounds than the old flat cap allowed still converges" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // The failure this reproduces, in miniature. A real 1,166,689-artifact store
    // could not be cloned at all: `max_sync_rounds` (1024) multiplied against
    // `clone_cfile_batch_max` (1024) capped any clone at ~1,048,576 artifacts,
    // and the client aborted with `RoundLimitExceeded` while the peer was
    // converging perfectly — delivering a full batch every single round.
    //
    // Reaching that ceiling by artifact count would need a million-artifact
    // fixture. The ceiling is a PRODUCT, so shrinking the batch to 1 reaches the
    // same place: this clone needs more rounds than the old cap permitted, at a
    // corpus size a unit test can hold.
    const old_round_cap: usize = 1024;
    const total: usize = old_round_cap + 100;
    for (0..total) |id| {
        const h = hashWide(id);
        _ = try server_ms.store().put(&h, "c");
    }

    var fake = ServerTransport{
        .server = .{
            .version = .{ .version = 0 },
            .self_servercode = "s",
            .project_code = "p",
            .clone_batch_max = 1, // one artifact per round: rounds track the corpus
        },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = newCloneClient();
    defer client.deinit(gpa);

    // A tight stall budget deliberately: every round delivers an artifact, so a
    // converging clone never accumulates a stall no matter how long it runs.
    // Passing 4 here proves the guard bounds unproductive looping and NOT the
    // length of a healthy transfer — the whole point of the change.
    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, 4);

    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // Ran past the old ceiling rather than aborting at it.
    try testing.expect(stats.rounds > old_round_cap);
    // And never accumulated a stall run while doing so, which is why it was
    // allowed to. Exactly ONE byte-free round is expected, not zero: the clone
    // ends on a terminal round that carries no cfile, so even a flawless
    // transfer of any length finishes one stall deep. That is the concrete
    // reason the budget cannot be 1 — a healthy clone would trip it — and the
    // count does not grow with the corpus, which is the property under test.
    try testing.expectEqual(@as(usize, 1), stats.stalled_peak);

    // Converged, not merely survived: every artifact arrived.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    try testing.expectEqual(total, (try heldHashes(arena.allocator(), client_ms.store())).len);
}

test "sync_driver: a round sink sees every round with cumulative byte totals matching RunStats" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    var id: usize = 0;
    while (id < 5) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "clone-artifact-content");
    }
    var fake = ServerTransport{
        .server = .{ .version = .{ .version = 0 }, .self_servercode = "s", .project_code = "p", .clone_batch_max = 2 },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);
    var client = newCloneClient();
    defer client.deinit(gpa);

    const Counter = struct {
        calls: usize = 0,
        last_round: usize = 0,
        last_total_received: usize = 0,
        monotone: bool = true,
        fn emit(ctx: *anyopaque, info: sync_driver.RoundInfo) void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            self.calls += 1;
            // Rounds arrive in order, 1-based; totals never decrease.
            if (info.round != self.last_round + 1) self.monotone = false;
            if (info.total_received < self.last_total_received) self.monotone = false;
            self.last_round = info.round;
            self.last_total_received = info.total_received;
        }
    };
    var counter: Counter = .{};
    const stats = try sync_driver.runSyncProgress(
        gpa,
        &client,
        client_ms.store(),
        &fake,
        round_bound,
        .{ .ctx = &counter, .emit = Counter.emit },
        null,
    );

    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // The sink saw exactly the rounds the stats report, in order, and the
    // final cumulative total is the RunStats total — one accounting, two views.
    try testing.expectEqual(stats.rounds, counter.calls);
    try testing.expect(counter.monotone);
    try testing.expectEqual(stats.received_bytes, counter.last_total_received);
    // A clone that moved five artifacts moved bytes in both directions.
    try testing.expect(stats.received_bytes > 0);
    try testing.expect(stats.sent_bytes > 0);
}

test "sync_driver §8.2: a clone spanning multiple cfile batches pulls every artifact and terminates on clone_seqno 0" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // A server holding more artifacts than one small batch carries: with a
    // batch cap of 2, five artifacts page across three cfile batches (§8.2
    // L1384), so the client MUST thread the `clone_seqno NEXT` cursor into
    // follow-up `clone 3 NEXT` requests to pull past the first batch.
    const total: usize = 5;
    var id: usize = 0;
    while (id < total) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "clone-artifact-content");
    }

    var fake = ServerTransport{
        .server = .{ .version = .{ .version = 0 }, .self_servercode = "s", .project_code = "p", .clone_batch_max = 2 },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = newCloneClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the clone reached the clean terminal, well within the round bound.
    // Three cfile batches force at least three rounds: a client truncating at
    // batch one (the pre-fix bug) would stop at round two with content missing.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 3 and stats.rounds <= round_bound);

    // §8.2 — the whole repository cloned: every server artifact is now held by
    // the client (not just the first batch), and no phantom remains.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(total, held.len);
    try testing.expect(noPhantom(client_ms.store()));
    try testing.expectEqualDeep(try heldHashes(a, server_ms.store()), held);
}

test "sync_driver §8.2: the byte budget pages a clone alone — count backstop never binds, cursor resumes, every card lands" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // Five artifacts against a one-byte budget: the batch closes after every
    // card (the crossing card lands WHOLE — the budget is a target, not a
    // ceiling), so the byte bound alone forces one artifact per round while
    // the count backstop (default, never reached) stays inert. Liveness:
    // even the tiniest budget always carries the first card, so the
    // clone_seqno cursor advances every round.
    const total: usize = 5;
    var id: usize = 0;
    while (id < total) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "clone-artifact-content");
    }

    var fake = ServerTransport{
        .server = .{ .version = .{ .version = 0 }, .self_servercode = "s", .project_code = "p", .clone_batch_bytes = 1 },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = newCloneClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= total and stats.rounds <= round_bound);

    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(total, held.len);
    try testing.expect(noPhantom(client_ms.store()));
    try testing.expectEqualDeep(try heldHashes(a, server_ms.store()), held);
}

test "sync_driver §3.7: runSync converges the client store against a peer and terminates" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // A mix with both directions and shared ancestry, plus phantoms on each
    // side — the general reconcile, exercising pull, push, and dephantomize.
    try seedStores(client_ms.store(), server_ms.store(), 3, 2, 2);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);

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

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the client reached the clean terminal within the round bound.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);

    // §3.7(a) — the client store now holds the union of intact content and
    // retains no phantom the peer could have satisfied.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(@as(usize, 3 + 2 + 2), held.len); // C₀ ∪ S₀
    try testing.expect(noPhantom(client_ms.store()));
    // Sync is symmetric: the server (also a receiver) converged to the union too.
    try testing.expect(noPhantom(server_ms.store()));
    try testing.expectEqual(@as(usize, 7), (try heldHashes(a, server_ms.store())).len);
}

test "sync_driver §3.7: an already-converged pair terminates in a single round with no phantom churn" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 4, 0, 0); // identical stores

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // Nothing to reconcile: one round observes no continuation condition (§3.7).
    try testing.expectEqual(@as(usize, 1), stats.rounds);
    try testing.expect(noPhantom(client_ms.store()));
}

test "sync_driver §3.2/§3.3: pull fetches the artifacts a subset repo lacks and converges" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    // The client holds a strict SUBSET of the server: three artifacts on both,
    // two more only on the server. The client is unaware of the server-only pair
    // until the peer announces them (no pre-seeded phantom) — exactly the state
    // `loadRepoHoldings` leaves a repository in before a real pull.
    const shared: usize = 3;
    const server_only: usize = 2;
    var id: usize = 0;
    while (id < shared) : (id += 1) {
        const h = hashN(id);
        _ = try client_ms.store().put(&h, "shared-content");
        _ = try server_ms.store().put(&h, "shared-content");
    }
    while (id < shared + server_only) : (id += 1) {
        const h = hashN(id);
        _ = try server_ms.store().put(&h, "server-only-content");
    }

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newPullClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.8 — the pull reached the clean terminal within the round bound.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);

    // §3.7(a) — the client converged to the server's full set: every server
    // artifact is now held by the client and no phantom remains.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held = try heldHashes(a, client_ms.store());
    try testing.expectEqual(shared + server_only, held.len);
    try testing.expect(noPhantom(client_ms.store()));
    try testing.expectEqualDeep(try heldHashes(a, server_ms.store()), held);
}

test "sync_driver §3.7: a pull against an already-current repo receives nothing and terminates in one round" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // Identical stores: the client already holds everything the server does.
    var id: usize = 0;
    while (id < 4) : (id += 1) {
        const h = hashN(id);
        _ = try client_ms.store().put(&h, "content");
        _ = try server_ms.store().put(&h, "content");
    }

    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const before = (try heldHashes(a, client_ms.store())).len;

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newPullClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // Nothing to fetch: one round observes no continuation condition (§3.7), and
    // the held set is unchanged — the "already up to date" case `runPull` reports.
    try testing.expectEqual(@as(usize, 1), stats.rounds);
    try testing.expectEqual(before, (try heldHashes(a, client_ms.store())).len);
    try testing.expect(noPhantom(client_ms.store()));
}

test "sync_driver: a peer that announces forever without delivering hits the stall guard, no leak" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var fake = NeverEndsTransport{};
    var client = newClient();
    defer client.deinit(gpa);

    const max_stalled: usize = 4;
    const err = sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_stalled);
    try testing.expectError(error.RoundLimitExceeded, err);
    // Every round stalls (a phantom is created, but no content is delivered),
    // so the consecutive count never resets and the guard fires at the bound:
    // one exchange per permitted round.
    try testing.expectEqual(max_stalled, fake.counter);
    // The client is still mid-exchange (never a terminal): the driver refused
    // to spin, it did not fabricate a stop.
    try testing.expectEqual(sync.ClientState.exchange, client.state);
}

test "sync_driver: a transfer needing many more rounds than the stall budget still converges" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();

    // Far more rounds than the budget: under a total-round cap this is the
    // failure mode, since the cap would have to exceed the transfer's length.
    const announce_rounds: usize = 200;
    const max_stalled: usize = 4;
    var fake = ManyRoundsTransport{ .announce_rounds = announce_rounds };
    var client = newClient();
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_stalled);

    try testing.expectEqual(sync.ClientState.terminated, client.state);
    // The point of the test: it ran an order of magnitude past the bound,
    // because the bound counts stalls and this peer never stalled twice.
    try testing.expect(stats.rounds > announce_rounds);
    try testing.expect(stats.stalled_peak <= max_stalled);
    // And it actually converged rather than merely surviving: every announced
    // artifact was delivered and stored.
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    try testing.expectEqual(announce_rounds, (try heldHashes(arena.allocator(), client_ms.store())).len);
}

test "sync §3.2/§3.3: a sync run both creates phantoms from peer igot and uploads held content" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // A both-directions gap: two artifacts only the client holds (it must UPLOAD
    // them) and two only the server holds (it must RECEIVE them). A `.sync`
    // client drives both on ONE §3.7 session — not pull-then-push over two.
    const client_only: usize = 2;
    const server_only: usize = 2;
    try seedStores(client_ms.store(), server_ms.store(), 1, client_only, server_only);

    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const a = arena.allocator();
    const held_before = (try heldHashes(a, client_ms.store())).len;

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient(); // operation = .sync
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    // §3.2/§3.3 — one session both uploaded the client-only content (§3.2
    // both-directions, counted by `content_sent`) and dephantomized the
    // server-only content into the client store (§3.3 held-set growth). Both
    // strictly positive: sync is bidirectional.
    const received = (try heldHashes(a, client_ms.store())).len - held_before;
    try testing.expect(client.content_sent > 0);
    try testing.expect(received > 0);
    try testing.expectEqual(client_only, client.content_sent);
    try testing.expectEqual(server_only, received);
}

test "sync §3.7: sync drives to a clean protocol terminal in bounded rounds" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 2, 1, 1);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient(); // operation = .sync
    defer client.deinit(gpa);

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    // §3.7/§3.8 — a bidirectional sync always reaches the clean terminal (never
    // `aborted`, never spinning) within a bounded number of rounds.
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 1 and stats.rounds <= round_bound);
}

/// One full convergence run over `gpa`, no assertions — the body under
/// `checkAllAllocationFailures`, which drives every allocation-failure point and
/// checks each returns `error.OutOfMemory` leak-free.
fn convergeOnce(gpa: std.mem.Allocator) !void {
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    try seedStores(client_ms.store(), server_ms.store(), 2, 1, 1);

    var fake = ServerTransport{ .server = .{ .version = .{ .version = 0 } }, .store = server_ms.store() };
    defer fake.deinit(gpa);
    var client = newClient();
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
}

test "sync_driver OOM: every allocation-failure point along a full run is handled leak-free" {
    try testing.checkAllAllocationFailures(testing.allocator, convergeOnce, .{});
}

// ---------------------------------------------------------------------------
// draft-fossil-submanifests-00 §7.2 — the peer-version gate, driven end to end
// ---------------------------------------------------------------------------

/// The tree manifest the gate tests upload: a check-in manifest carrying an S
/// run, which is §7.2's second shape ("a manifest carrying an S card").
const gate_tree = "C t\nD 2026-07-27T12:00:00.000\nS a.txt " ++ ("1" ** 64) ++ "\nU u\n";
/// A flat check-in manifest — everything that predates this draft — carried
/// alongside so every assertion can say what the gate left alone.
const gate_flat = "C m\nD 2026-07-27T12:00:00.000\nF a.txt " ++ ("2" ** 64) ++ "\nU u\n";

/// One §7.2 upload run through the PRODUCTION driver, with nothing hand-seeded:
/// the client learns the peer's version only from the wire, exactly as `zit
/// push`/`zit sync` do. `client_version` is the pragma the CLI would announce.
/// Returns the client after the run (caller deinits).
const GateRun = struct {
    refused: usize,
    peer_version: u32,
    server_holds_tree: bool,
    server_holds_flat: bool,
    /// §3.8 — the state the run ended in. A refusal must still reach the CLEAN
    /// terminal: a peer that keeps re-`gimme`ing what this side keeps refusing
    /// is the shape that spins to `RoundLimitExceeded`.
    state: sync.ClientState,
};

fn runGateUpload(gpa: std.mem.Allocator, client_version: sync.VersionInfo) !GateRun {
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();

    var tree_name: [64]u8 = undefined;
    var flat_name: [64]u8 = undefined;
    _ = zit.artifact.artifactName(.sha3_256, gate_tree, &tree_name);
    _ = zit.artifact.artifactName(.sha3_256, gate_flat, &flat_name);
    _ = try client_ms.store().put(&tree_name, gate_tree);
    _ = try client_ms.store().put(&flat_name, gate_flat);

    // The server is a plain zit `serve`: a build identity carrying DATE/TIME so
    // §10.1 r11/r12 CAN answer, and §8.1's emission setting off (there is no
    // operator surface for it). What it announces is `announcedVersion`'s job.
    var fake = ServerTransport{
        .server = .{
            .version = .{ .version = 22800, .datetime = .{ .date = 20260715, .time = 0 } },
            .self_servercode = "s",
            .project_code = projectcode,
        },
        .store = server_ms.store(),
    };
    defer fake.deinit(gpa);

    var client = sync.Client{
        .state = .exchange,
        .operation = .sync,
        .self_servercode = servercode,
        .project_code = projectcode,
        .version = client_version,
        // §8.1's durable indicator: this repository holds a tree, so its send
        // gate is armed even though the emission setting is off.
        .seen_submanifest = true,
        // .peer is DELIBERATELY not seeded — the whole point of the run.
    };
    defer client.deinit(gpa);

    _ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);

    return .{
        .refused = client.submanifests_refused,
        .peer_version = client.peer.version,
        .state = client.state,
        .server_holds_tree = if (server_ms.store().lookup(&tree_name)) |r| server_ms.store().content(r) != null else false,
        .server_holds_flat = if (server_ms.store().lookup(&flat_name)) |r| server_ms.store().content(r) != null else false,
    };
}

test "sync_driver submanifests §7.2/§10.1: an armed client uploads a tree because the peer's announcement reached it" {
    // THE LIVELOCK THIS PINS. §7.2 gates every upload on "the remote peer's
    // announced version", and a client learns that version from exactly one
    // place: the `pragma server-version` reply. §10.1 r11/r12 make that reply
    // conditional on the request's client-version carrying DATE and TIME. An
    // undated request therefore leaves `Client.peer.version` at 0 — below every
    // threshold — for the whole session, so an armed client refuses its own
    // trees to a server that would have announced support if only it had been
    // asked: trees pullable, never pushable, and `submanifests_refused` fails
    // every run of the feature's happy path.
    //
    // Nothing here is hand-seeded. The client's `.peer` is left at its default
    // and the announcement travels the real request/reply loop through the
    // production driver, so the test cannot pass on unwired code.
    const gpa = testing.allocator;

    const dated = try runGateUpload(gpa, .{ .version = 22700, .datetime = .{ .date = 20260715, .time = 0 } });
    // The server's announcement arrived and was recorded (§10.1 r13)…
    try testing.expectEqual(sync.submanifest_min_peer_version, dated.peer_version);
    try testing.expectEqual(sync.ClientState.terminated, dated.state);
    // …so the gate opened and the tree was uploaded, not withheld.
    try testing.expectEqual(@as(usize, 0), dated.refused);
    try testing.expect(dated.server_holds_tree);
    try testing.expect(dated.server_holds_flat);

    // Drop DATE/TIME from the client-version pragma — the CLI's behaviour before
    // this was wired — and the server stays silent about its version (§10.1
    // r12), so the peer record never leaves 0 and the tree is refused to a peer
    // that has, from the client's side of the wire, announced nothing at all.
    const undated = try runGateUpload(gpa, .{ .version = 22700 });
    try testing.expectEqual(@as(u32, 0), undated.peer_version);
    try testing.expectEqual(@as(usize, 1), undated.refused);
    // …and the refusal SETTLES: the peer re-gimmes the tree it never received,
    // and a client that answered by re-queueing it would spin to the driver's
    // round limit instead of stopping with one counted refusal.
    try testing.expectEqual(sync.ClientState.terminated, undated.state);
    try testing.expect(!undated.server_holds_tree);
    // …and the refusal is discriminating: the flat manifest still lands.
    try testing.expect(undated.server_holds_flat);
}

/// The credentialed, gated peer: replies come from a Server holding a
/// credential and announcing a cookie-gated version, and every round records
/// (a) whether the request body carried a `login` card and (b) the cookie the
/// driver installed. Auth is REAL: the recorded cookie feeds `respondCookie`,
/// so a missing or wrong per-round credential makes the server answer with an
/// error card and the driver surface `FatalProtocolError` — the failure
/// mode, structurally unable to pass.
const AuthRecordingTransport = struct {
    server: sync.Server,
    store: repostate.ArtifactStore,
    reply_buf: std.ArrayListUnmanaged(u8) = .empty,
    cookie: ?[]u8 = null,
    /// Per round: did the BODY carry a login card / did the driver set a cookie?
    body_logins: std.ArrayListUnmanaged(bool) = .empty,
    cookies: std.ArrayListUnmanaged(bool) = .empty,
    gpa: std.mem.Allocator,

    pub const Error = sync.Server.RespondError;

    fn deinit(self: *AuthRecordingTransport) void {
        self.reply_buf.deinit(self.gpa);
        if (self.cookie) |c| self.gpa.free(c);
        self.body_logins.deinit(self.gpa);
        self.cookies.deinit(self.gpa);
    }

    pub fn setLoginCookie(self: *AuthRecordingTransport, value: ?[]const u8) void {
        if (self.cookie) |c| self.gpa.free(c);
        self.cookie = if (value) |v| (self.gpa.dupe(u8, v) catch @panic("oom")) else null;
    }

    pub fn exchange(self: *AuthRecordingTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
        try self.body_logins.append(self.gpa, std.mem.indexOf(u8, request, "\nlogin ") != null or std.mem.startsWith(u8, request, "login "));
        try self.cookies.append(self.gpa, self.cookie != null);
        const reply = try self.server.respondCookie(gpa, self.store, request, self.cookie, &self.reply_buf);
        return gpa.dupe(u8, reply);
    }
};

test "sync_driver §6.5: an authenticated push carries a credential on EVERY round — body login before the gate is known, cookie after" {
    const gpa = testing.allocator;
    var client_ms = repostate.MemStore.init(gpa);
    defer client_ms.deinit();
    var server_ms = repostate.MemStore.init(gpa);
    defer server_ms.deinit();
    // Push-heavy: several client-only artifacts, so the exchange runs multiple
    // rounds and the artifact-carrying rounds are exactly the post-gate ones
    // that were being sent anonymous.
    try seedStores(client_ms.store(), server_ms.store(), 2, 4, 0);

    const secret = sync.deriveSharedSecret(projectcode, "dan", "pw");
    const cred: sync.Credential = .{ .user = "dan", .secret_hex = secret };
    var fake = AuthRecordingTransport{
        .gpa = gpa,
        .server = .{
            // Past the §6.5 gate on both conjuncts, so the client's second
            // round switches channels.
            .version = .{ .version = 22900, .datetime = .{ .date = 20260715, .time = 120000 } },
            .credential = cred,
        },
        .store = server_ms.store(),
    };
    defer fake.deinit();

    var client = newClient();
    defer client.deinit(gpa);
    client.credential = cred;
    // A DATED client-version, as the production client sends (§9.1
    // L1854-1859): the server announces its own version only to a dated
    // pragma, and that announcement is what flips the §6.5 gate.
    client.version = .{ .version = 22700, .datetime = .{ .date = 20260101, .time = 0 } };

    const stats = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
    try testing.expectEqual(sync.ClientState.terminated, client.state);
    try testing.expect(stats.rounds >= 2); // the regression needs a round past the first

    // The invariant: every round authenticated through exactly one channel.
    // Round 1 built before any reply recorded the peer's version: body login,
    // no cookie. Every later round: cookie, no body login.
    for (fake.body_logins.items, fake.cookies.items, 0..) |body, cookie, i| {
        if (i == 0) {
            try testing.expect(body);
            try testing.expect(!cookie);
        } else {
            try testing.expect(!body);
            try testing.expect(cookie);
        }
    }
    // And the push landed: the server holds the client-only artifacts.
    try testing.expect(noPhantom(server_ms.store()));
}