//! 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 {
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 `max_rounds` guard must catch it.
const NeverEndsTransport = struct {
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});
}
};
// ---------------------------------------------------------------------------
// 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;
}
/// 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 };
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
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 §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: a peer that never terminates hits max_rounds cleanly, 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_rounds: usize = 4;
const err = sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_rounds);
try testing.expectError(error.RoundLimitExceeded, err);
// The guard fired exactly at the bound: one exchange per permitted round.
try testing.expectEqual(max_rounds, 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);
}
/// 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, .{});
}