Skip to content
zit Artifact [12aef56f07]
Login

Artifact [12aef56f07]

Check-ins Using Download Hex

Artifact 12aef56f077c99c4edbd64ebdbc64f0d10fb9b1dc814d6fce5651b5d231101c0:

//! 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);
        // §6.5 (#74) — the round's credential rides the transport, not the
        // body, once the peer gate selects the cookie channel; null clears a
        // prior round's cookie so an unauthenticated round can never replay
        // one. Set BEFORE the exchange the cookie signs.
        transport.setLoginCookie(client.loginCookie());
        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,
    };
}