//! CLI-layer sync round driver (sync §3.7/§3.8).
//!
//! `sync.Client` is a sans-IO state machine: `buildRequest` composes a request
//! body and `consumeReply` folds a reply body back into the client and the
//! durable store, but neither touches a socket (§3.8 — the server is stateless,
//! only the client keeps state). This module is the production round loop that
//! pumps that machine to completion: build → exchange → consume, repeated until
//! §3.7 says stop, converging the local store against a peer.
//!
//! The transport is abstracted at comptime (`transport: anytype`) rather than
//! behind a vtable: the driver needs exactly one operation — `exchange(gpa,
//! request) -> ![]u8` — and both the real `transport.Transport` and a test's
//! in-process fake satisfy it structurally, so no interface type is minted and
//! `transport.Transport` stays concrete. All IO lives on the far side of that
//! `exchange` call; the driver itself is IO-free and deterministic, so the DST
//! (`sim/sync_sweep.zig`) drives the same protocol over a simulated network.
const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const assert = std.debug.assert;
/// The driver's own closed error causes, independent of the transport:
/// `buildRequest` → {OutOfMemory, RequestTooLarge} (§3.2/§3.8),
/// `consumeReply` → {OutOfMemory, FatalProtocolError} (§3.8; legal-but-
/// unsupported unversioned cards count into `Client.uv_skipped` instead of
/// erroring, #58),
/// the loop guard → RoundLimitExceeded (a runaway/misbehaving peer must not
/// loop forever; see `runSync` for what "runaway" is measured in). Folded
/// with the transport's own `exchange` error set by
/// `RunError`, so the CLI still switches the whole surface exhaustively.
pub const DriverError = error{
OutOfMemory,
RequestTooLarge,
FatalProtocolError,
RoundLimitExceeded,
};
/// The closed error set `runSync` returns for a given transport type: the
/// driver's own causes unioned with that transport's `exchange` error set.
/// Computed at comptime from the transport, so the union stays a named, closed
/// set (never `anyerror`) even though the transport is duck-typed.
pub fn RunError(comptime Transport: type) type {
return DriverError || ExchangeErrorSet(Transport);
}
/// Extract the error set of `Transport.exchange`, tolerating a pointer receiver
/// (`*Transport`, the usual case since the driver mutates the transport).
fn ExchangeErrorSet(comptime Transport: type) type {
const Base = switch (@typeInfo(Transport)) {
.pointer => |p| p.child,
else => Transport,
};
const exchange_fn = @TypeOf(@field(Base, "exchange"));
const return_type = @typeInfo(exchange_fn).@"fn".return_type.?;
return @typeInfo(return_type).error_union.error_set;
}
/// What one completed run reports to its caller. `rounds` is the number of
/// request/reply exchanges pumped — the caller's convergence-cost metric.
/// `stalled_peak` is the longest run of consecutive rounds that moved no bytes,
/// against the budget `runSync` was given: a run that finishes with this at or
/// near the budget converged, but only just.
pub const RunStats = struct {
rounds: usize,
stalled_peak: usize = 0,
/// Uncompressed protocol bytes moved over the run, summed per exchange —
/// request bytes out, reply bytes in. These are the bytes the sans-IO
/// layer sees; the transport's zlib container framing is hidden below
/// `exchange`, so they are labelled protocol bytes, never "wire" bytes.
sent_bytes: usize = 0,
received_bytes: usize = 0,
};
/// One completed round, as reported to a `RoundSink` — the live-progress seam
/// (#55). Totals are cumulative through this round.
pub const RoundInfo = struct {
round: usize,
reply_bytes: usize,
total_sent: usize,
total_received: usize,
delivered: bool,
};
/// A caller-supplied per-round observer. The driver stays sans-IO: whatever
/// printing the sink does is the caller's IO, on the caller's stream.
pub const RoundSink = struct {
ctx: *anyopaque,
emit: *const fn (ctx: *anyopaque, info: RoundInfo) void,
};
/// §3.7/§3.8 — drive `client` to a clean protocol terminal, converging `store`
/// against the peer reached through `transport`. Each round: compose the
/// request (`buildRequest`), carry it over the transport (`exchange`), and fold
/// the reply back in (`consumeReply`); repeat while the client's state machine
/// is mid-exchange. A peer that never lets §3.7 terminate yields
/// `error.RoundLimitExceeded` instead of hanging.
///
/// `max_stalled_rounds` bounds CONSECUTIVE BYTE-FREE ROUNDS, not total rounds.
/// A round that moves content (`sync.roundDeliveredContent`) resets the count;
/// only rounds that accomplish nothing accumulate. This is the whole guard —
/// there is deliberately no ceiling on total rounds.
///
/// A total-round cap cannot express what is being guarded against. Every
/// operation transfers a bounded amount per round (clone: at most
/// `clone_cfile_batch_max` cfiles), so N rounds caps the transfer at N batches
/// and the guard silently becomes a limit on repository SIZE. A large
/// repository then fails while converging perfectly — every round delivering a
/// full batch — and reports it as the peer's failure to converge, which is the
/// opposite of what happened. Counting stalls instead bounds unproductive
/// looping directly and lets a healthy transfer run as long as it needs to.
///
/// A legitimate round CAN move no bytes: §3.7's pending phases (config
/// exchange (f), private-content discovery (c), the unversioned three-round
/// grace window (d)(2)) each stall briefly by design, so the budget must clear
/// them with room to spare rather than sitting at 1.
///
/// Sans-IO: all IO is the transport's; the driver only moves `[]const u8` wire
/// bytes between the sans-IO client and the transport. Each round's reply is
/// owned by the caller (`exchange` contract) and freed here — no per-round leak.
/// A §3.8 r18 fatal surfaces as `consumeReply`'s `FatalProtocolError`, so the
/// loop never exits silently on an abort; it returns only on §3.7's clean stop.
pub fn runSync(
gpa: std.mem.Allocator,
client: *sync.Client,
store: repostate.ArtifactStore,
transport: anytype,
max_stalled_rounds: usize,
) RunError(@TypeOf(transport))!RunStats {
return runSyncProgress(gpa, client, store, transport, max_stalled_rounds, null);
}
/// `runSync` with a live per-round observer — the shape `zit clone`'s progress
/// reporting rides (#55). A null sink is byte-identical to `runSync`.
pub fn runSyncProgress(
gpa: std.mem.Allocator,
client: *sync.Client,
store: repostate.ArtifactStore,
transport: anytype,
max_stalled_rounds: usize,
sink: ?RoundSink,
) RunError(@TypeOf(transport))!RunStats {
// Preconditions (§3.8): a real budget and a startable, mid-exchange client
// (the caller seeds `initialState`; a terminal or `.init` client is a bug).
assert(max_stalled_rounds > 0);
assert(client.state == .exchange or client.state == .clone_round0);
var request_buf: std.ArrayListUnmanaged(u8) = .empty;
defer request_buf.deinit(gpa);
var rounds: usize = 0;
var stalled: usize = 0;
var stalled_peak: usize = 0;
var sent_bytes: usize = 0;
var received_bytes: usize = 0;
while (client.state == .exchange or client.state == .clone_round0) {
if (stalled == max_stalled_rounds) return error.RoundLimitExceeded;
rounds += 1;
const request = try client.buildRequest(gpa, store, &request_buf);
const reply = try transport.exchange(gpa, request);
defer gpa.free(reply);
sent_bytes += request.len;
received_bytes += reply.len;
const obs = try client.consumeReply(gpa, store, reply);
// §8.2 — a clone round carries the received sequence number into the
// next request; pull/push/sync never read it.
if (client.operation == .clone) client.clone_seqno = obs.clone_seqno_next;
// The guard's only input. A productive round clears the debt outright;
// it is CONSECUTIVE stalls that indicate a peer accomplishing nothing,
// so a slow-but-moving transfer never accumulates toward the bound.
const delivered = sync.roundDeliveredContent(obs);
if (delivered) {
stalled = 0;
} else {
stalled += 1;
stalled_peak = @max(stalled_peak, stalled);
}
if (sink) |s| s.emit(s.ctx, .{
.round = rounds,
.reply_bytes = reply.len,
.total_sent = sent_bytes,
.total_received = received_bytes,
.delivered = delivered,
});
}
// Postcondition (§3.7): the loop exits only on the clean terminal. A fatal
// abort (§3.8 r18) left through `consumeReply`'s error return above, so
// `.aborted` is never observed here.
assert(client.state == .terminated);
assert(rounds >= 1);
assert(stalled_peak <= max_stalled_rounds); // the guard never overshot
return .{
.rounds = rounds,
.stalled_peak = stalled_peak,
.sent_bytes = sent_bytes,
.received_bytes = received_bytes,
};
}