//! 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()));
}
/// Replays fixed reply bodies, one per round; past the end it answers with an
/// empty body.
const ScriptedTransport = struct {
pub fn setLoginCookie(self: *@This(), value: ?[]const u8) void {
_ = self;
_ = value;
}
replies: []const []const u8,
next: usize = 0,
pub const Error = error{OutOfMemory};
pub fn exchange(self: *ScriptedTransport, gpa: std.mem.Allocator, request: []const u8) Error![]u8 {
_ = request;
const reply = if (self.next < self.replies.len) self.replies[self.next] else "";
self.next += 1;
return gpa.dupe(u8, reply);
}
};
test "sync_driver §5.1/§10.1: a later clone round cut short at its first card ends the run with the truncation counted" {
const gpa = testing.allocator;
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
const hash_a = "a" ** 40;
const hash_b = "b" ** 40;
// Round 0 delivers one artifact and a cursor; round 1 is cut short inside
// its first card, so its clone_seqno card never arrives. Continuation
// finds no cursor and the exchange ends. Only the counter tells the
// caller the clone is incomplete.
var fake = ScriptedTransport{ .replies = &.{
"file " ++ hash_a ++ " 5\nhelloclone_seqno 2\n",
"file " ++ hash_b ++ " 100\nshort",
} };
var client = newCloneClient();
defer client.deinit(gpa);
_ = try sync_driver.runSync(gpa, &client, client_ms.store(), &fake, round_bound);
try testing.expectEqual(@as(usize, 2), fake.next);
try testing.expectEqual(@as(u64, 1), client.payloads_truncated);
try testing.expect(client_ms.store().lookup(hash_a) != null);
try testing.expect(client_ms.store().lookup(hash_b) == null);
try testing.expectEqual(sync.ClientState.terminated, client.state);
}