//! 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, NotImplemented} (§3.8),
/// the loop guard → RoundLimitExceeded (a runaway/misbehaving peer must not
/// loop forever). 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,
NotImplemented,
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.
pub const RunStats = struct {
rounds: usize,
};
/// §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. `max_rounds` is an explicit upper bound — a peer that never
/// lets §3.7 terminate yields `error.RoundLimitExceeded` instead of hanging.
///
/// 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_rounds: usize,
) 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_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;
while (client.state == .exchange or client.state == .clone_round0) {
if (rounds == max_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);
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;
}
// 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 and rounds <= max_rounds);
return .{ .rounds = rounds };
}