Skip to content
zit Artifact [c20a7b08fb]
Login

Artifact [c20a7b08fb]

Check-ins Using Download Hex

Artifact c20a7b08fba551294060fbc1f100f7dba4e929568c400d0a181c90d4f22bc5d8:

//! 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; 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,
    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.
/// `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,
};

/// §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 {
    // 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;
    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);
        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.
        if (sync.roundDeliveredContent(obs)) {
            stalled = 0;
        } else {
            stalled += 1;
            stalled_peak = @max(stalled_peak, stalled);
        }
    }

    // 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 };
}