//! Two-phase deterministic simulation of the sync protocol (transport-simnet-
//! plan.md §A3) — the PRIMARY DST for sync. It drives the full round loop of an
//! A0 `Client`/`Server` pair (`buildRequest` → `respond` → `consumeReply`)
//! through the A1 `SimNet`, which injects the A2 scenario's drop/dup/corrupt/
//! clog faults, and asserts the sync RFC's convergence (§3.7), termination
//! (§3.7/§3.8), and private-content (§7.5) invariants.
//!
//! The seed is the whole universe: one `std.Random.DefaultPrng.init(seed)`
//! feeds the scenario generator AND the network, in a fixed draw order, so any
//! failing sweep replays exactly — dst.zig prints `reproduce with seed=N`.
//!
//! Runs are two-phase (safety → liveness), per the project DST discipline:
//! Phase 1 (safety): bounded rounds under injected faults. Every round probes
//! invariants that MUST always hold — no corrupt delivery is ever accepted as
//! content, held content never regresses, the in-flight queue stays bounded,
//! and a receive/exchange operation never aborts. A dropped/duplicated/
//! corrupted packet may lose a round but can never break these.
//! Phase 2 (liveness): the network is HEALED (all fault probabilities → 0) and
//! a fresh sync session runs to completion within a BOUNDED convergence window
//! (rounds, not wall time). The stores MUST converge to the direction-correct
//! artifact sets, no phantom the peer holds intact may remain on the receiving
//! side, and the client MUST reach `terminated`. Exceeding the window is a
//! bug, never flakiness.
//!
//! Transport model (§4 / §A1): the sans-IO boundary is a whole request/reply
//! BODY. A `SimNet` corruption stands for a body a real transport's zlib
//! container decode (§4.1) would reject, so a `Delivery.corrupted` message is
//! treated as a lost round — it is NEVER handed to `respond`/`consumeReply`.
//! That is why "no corrupt content accepted" holds by construction and a
//! corrupted round degrades to a retry rather than a spurious abort.
const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const net = @import("net.zig");
const sync_gen = @import("sync_gen.zig");
const assert = std.debug.assert;
/// Closed error set: transport/protocol propagations that would signal a real
/// A0/A1/A2 bug if they ever fired here, plus the named invariant violations.
/// `FatalProtocolError`/`NotImplemented` do not appear — an abort is caught and
/// judged by the outer loop, and an unimplemented reply card is a server bug
/// surfaced as `UnexpectedNotImplemented`.
pub const Error = error{
OutOfMemory,
PayloadTooLarge,
QueueFull,
RequestTooLarge,
ResponseTooLarge,
// invariants
CorruptContentAccepted,
ContentRegressed,
InFlightUnbounded,
UnexpectedAbort,
UnexpectedNotImplemented,
RoundTickBudgetExceeded,
StoresDiverged,
ResidualPhantom,
NoTermination,
ConvergenceWindowExceeded,
PrivateContentLeaked,
PrivateStallsTermination,
};
/// Phase-1 fault-window length, in rounds. Bounded: no unbounded run.
const phase1_rounds: usize = 32;
/// Phase-2 convergence window, in rounds. A clean bidirectional exchange
/// converges in a handful of rounds (discovery is single-hop — a reply walks
/// the whole store once), so this is generous slack, not a tuned minimum.
const phase2_window: usize = 64;
/// Total round budget for one seed: an explicit upper bound on the whole run.
const total_budget: usize = phase1_rounds + phase2_window;
/// Per-round tick bound while draining one request/reply exchange to
/// quiescence. `max_delay_ticks` + `max_clog_ticks` (both ≤ 64) bound one
/// message; this comfortably bounds the few messages of one round.
const max_round_ticks: usize = 8192;
const Endpoint = net.Endpoint;
// ----------------------------------------------------------------------------
// Public entry point
// ----------------------------------------------------------------------------
/// Run the two-phase sweep for one seed, asserting every invariant. Returns on
/// success; a violation returns the named error. dst.zig calls this per seed in
/// its `zig build dst` gate loop, so `--seed N`/`--seeds K` select depth.
pub fn runSeed(gpa: std.mem.Allocator, seed: u64) Error!void {
_ = try simulate(gpa, seed, .{});
}
/// The observable result of one run: the client's terminal state and a stable
/// digest of both converged stores — two runs of the same seed hash equal
/// (the reproducibility oracle).
const Outcome = struct { state: sync.ClientState, digest: u64 };
// ----------------------------------------------------------------------------
// The two-phase driver
// ----------------------------------------------------------------------------
/// Generate the scenario, then run phase 1 (faults) and phase 2 (healed),
/// probing safety every round and asserting liveness + convergence at the end.
fn simulate(gpa: std.mem.Allocator, seed: u64, params: sync_gen.Params) Error!Outcome {
var prng = std.Random.DefaultPrng.init(seed);
const random = prng.random();
var scen = try sync_gen.generate(gpa, random, params);
defer scen.deinit();
// Scratch for the run's oracles: the content-integrity reference and the
// initial held sets. One arena, freed at return.
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
const arena = arena_state.allocator();
const ref = try buildReference(arena, &scen);
const c0 = try heldHashes(arena, scen.clientStore());
const s0 = try heldHashes(arena, scen.serverStore());
var netp = try net.SimNet.init(gpa, random);
defer netp.deinit();
netp.links[@intFromEnum(Endpoint.b)] = scen.link_to_b; // client → server
netp.links[@intFromEnum(Endpoint.a)] = scen.link_to_a; // server → client
// §6.4 — the server holds the same symmetric shared secret the client signs
// with, so a named login verifies (a mismatch would be an in-band error
// card, never a returned error).
// §8.2/§9.1 — the server carries the clone reply's SERVERCODE/PROJECTCODE.
// The project code matches the client's, so a clone's `push` card is a
// well-formed three-token card the peer accepts (an empty field would abort
// it); servercode is informational.
const server = sync.Server{
.version = scen.version,
.credential = scen.credential,
.self_servercode = scen.selfServercode(),
.project_code = scen.projectCode(),
};
var req_buf: std.ArrayListUnmanaged(u8) = .empty;
defer req_buf.deinit(gpa);
var rep_buf: std.ArrayListUnmanaged(u8) = .empty;
defer rep_buf.deinit(gpa);
var client = newClient(&scen);
defer client.deinit(gpa);
var prev_c: usize = c0.len;
var prev_s: usize = s0.len;
var phase: u8 = 1;
var phase2_start: usize = 0;
var round: usize = 0;
while (round < total_budget) : (round += 1) {
assert(round < total_budget); // explicit bound: no unbounded run
// Phase boundary: heal the wire and start a fresh session for the clean
// convergence run (a real re-invocation of `zit sync`; the durable
// stores carry phase-1 progress, phantoms re-derive from the store).
if (phase == 1 and round == phase1_rounds) {
netp.heal();
phase = 2;
phase2_start = round;
client.deinit(gpa);
client = newClient(&scen);
}
if (client.state == .terminated or client.state == .aborted) {
if (phase == 2) break; // the healed session reached its terminal
// A phase-1 session ended (possibly a packet-loss-induced early
// stop): re-invoke a fresh session over the same durable stores.
client.deinit(gpa);
client = newClient(&scen);
}
try runRound(gpa, &netp, &client, &server, &scen, &req_buf, &rep_buf);
try probeRoundSafety(&scen, ref, &netp, &client, &prev_c, &prev_s);
}
// Liveness: a terminal state must be reached within the phase-2 window.
assert(phase == 2);
if (client.state != .terminated and client.state != .aborted) return error.ConvergenceWindowExceeded;
assert(round - phase2_start <= phase2_window);
try checkTermination(&client, scen.operation);
try checkConvergence(arena, &scen, c0, s0);
return .{ .state = client.state, .digest = try outcomeDigest(arena, &scen, client.state) };
}
/// The per-round safety block, probed EVERY round of both phases: no corrupt
/// content accepted, held content never regresses, the in-flight queue stays
/// bounded, and (non-clone) the client never spuriously aborts. `prev_*` carry
/// the monotonic held counts across rounds.
fn probeRoundSafety(
scen: *sync_gen.Scenario,
ref: Reference,
netp: *const net.SimNet,
client: *const sync.Client,
prev_c: *usize,
prev_s: *usize,
) Error!void {
try probeIntegrity(scen.clientStore(), ref);
try probeIntegrity(scen.serverStore(), ref);
const ch = heldCount(scen.clientStore());
const sh = heldCount(scen.serverStore());
if (ch < prev_c.* or sh < prev_s.*) return error.ContentRegressed; // content never un-stored
prev_c.* = ch;
prev_s.* = sh;
if (netp.inFlight() > net.SimNet.max_inflight) return error.InFlightUnbounded;
try probeState(client, scen.operation);
}
/// §3.7/§3.8 termination: every operation, clone included, MUST reach
/// `terminated`. The server now answers §8 clone (paginated cfile batches with a
/// `clone_seqno` cursor to exhaustion), so a clone runs the round loop to a clean
/// §3.7 termination just like pull/push/sync — no operation aborts under
/// clean-or-filtered traffic.
fn checkTermination(client: *const sync.Client, op: sync.Operation) Error!void {
switch (op) {
.pull, .push, .sync, .clone => if (client.state != .terminated) return error.NoTermination,
}
}
/// A fresh client for `scen`'s operation: state seeded from §3.8's initial-state
/// map, identity/version/credential from the scenario. The phantom set and send
/// buffer default empty and re-derive from the durable store on each session.
fn newClient(scen: *const sync_gen.Scenario) sync.Client {
return .{
.state = sync.initialState(scen.operation),
.operation = scen.operation,
.self_servercode = scen.selfServercode(),
.project_code = scen.projectCode(),
.version = scen.version,
.credential = scen.credential,
};
}
/// One synchronous request/reply exchange, drained to quiescence: emit the
/// request, drive the net, `respond` to each intact request delivery, and
/// `consumeReply` the first intact reply (later duplicates are ignored — the
/// server is idempotent, §3.8). A corrupted delivery is a transport reject (a
/// lost round). Bounded by `max_round_ticks`.
fn runRound(
gpa: std.mem.Allocator,
netp: *net.SimNet,
client: *sync.Client,
server: *const sync.Server,
scen: *sync_gen.Scenario,
req_buf: *std.ArrayListUnmanaged(u8),
rep_buf: *std.ArrayListUnmanaged(u8),
) Error!void {
assert(client.state == .exchange or client.state == .clone_round0);
const client_store = scen.clientStore();
const server_store = scen.serverStore();
const req = try client.buildRequest(gpa, client_store, req_buf);
try netp.enqueue(.b, req);
var consumed = false;
var ticks: usize = 0;
while (netp.inFlight() > 0) {
while (netp.deliver()) |d| {
switch (d.dst) {
.b => {
// Request at the server. A corrupt body a real transport
// would reject before decode never reaches `respond`.
if (d.corrupted) continue;
const reply = try server.respond(gpa, server_store, d.payload, rep_buf);
try netp.enqueue(.a, reply);
},
.a => {
// Reply at the client.
if (d.corrupted) continue; // transport reject: a lost round
if (consumed) continue; // idempotent dup: consume exactly one
if (client.state != .exchange and client.state != .clone_round0) continue;
if (client.consumeReply(gpa, client_store, d.payload)) |obs| {
consumed = true;
// §8.2 — carry the received clone sequence number into
// the next request's cursor so clone pages to exhaustion.
if (client.operation == .clone) client.clone_seqno = obs.clone_seqno_next;
} else |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
// The server emits only full-content file/cfile cards; a
// NotImplemented (delta/uv form) reply would be a server
// bug, not wire noise.
error.NotImplemented => return error.UnexpectedNotImplemented,
// §3.8 r18 — the client aborted (`state == .aborted`).
// The outer loop judges whether that is legal for the op.
error.FatalProtocolError => consumed = true,
}
},
}
}
netp.tick();
ticks += 1;
if (ticks > max_round_ticks) return error.RoundTickBudgetExceeded;
}
}
/// Legal-state + no-spurious-abort probe. Any operation reaching `aborted` under
/// clean-or-filtered traffic implicates `consumeReply` or the A1 corruption
/// filter, so it is a hard failure — clone included now that the server serves
/// §8 clone and the clone client runs to a clean termination.
fn probeState(client: *const sync.Client, op: sync.Operation) Error!void {
_ = op;
assert(client.state != .init); // negative space: never regresses to pre-start
if (client.state == .aborted) return error.UnexpectedAbort;
}
// ----------------------------------------------------------------------------
// Safety oracle: content integrity
// ----------------------------------------------------------------------------
/// hash → the one true content for that hash. Every artifact is content-
/// addressed, so any bytes that appear under a hash must equal these.
const Reference = std.StringHashMapUnmanaged([]const u8);
/// Snapshot every held (hash, content) pair from both initial stores into
/// `arena`. Shared artifacts carry identical bytes on both sides (generator
/// contract), so the first sighting wins.
fn buildReference(arena: std.mem.Allocator, scen: *sync_gen.Scenario) Error!Reference {
var ref: Reference = .empty;
try indexReference(arena, scen.clientStore(), &ref);
try indexReference(arena, scen.serverStore(), &ref);
return ref;
}
fn indexReference(arena: std.mem.Allocator, store: repostate.ArtifactStore, ref: *Reference) Error!void {
const n = store.count();
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
if (store.content(rid)) |bytes| {
const h = store.hashOf(rid);
if (!ref.contains(h)) {
try ref.put(arena, try arena.dupe(u8, h), try arena.dupe(u8, bytes));
}
}
}
}
/// No corrupt/unknown content was ever interned as truth: every held entry's
/// bytes equal the content-addressed reference for its hash. A held hash absent
/// from the reference (a fabricated artifact) is equally a violation.
fn probeIntegrity(store: repostate.ArtifactStore, ref: Reference) Error!void {
const n = store.count();
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
if (store.content(rid)) |bytes| {
const want = ref.get(store.hashOf(rid)) orelse return error.CorruptContentAccepted;
if (!std.mem.eql(u8, bytes, want)) return error.CorruptContentAccepted;
}
}
}
// ----------------------------------------------------------------------------
// Convergence oracle (§3.7)
// ----------------------------------------------------------------------------
/// §3.7 — assert the healed run converged, per operation direction:
/// sync : both stores hold C₀ ∪ S₀, and neither retains a phantom.
/// pull : the client (receiver) holds C₀ ∪ S₀ with no phantom; the server's
/// content is unchanged (it keeps dangling phantoms for client-only
/// content it can never pull — expected, one-directional).
/// push : the server (receiver) holds C₀ ∪ S₀ with no phantom; the client
/// keeps at least its own content and no phantom.
/// §3.7(a): no residual phantom whose content the peer holds intact remains on
/// the receiving side(s) — asserted via `noPhantom`.
fn checkConvergence(
arena: std.mem.Allocator,
scen: *sync_gen.Scenario,
c0: []const []const u8,
s0: []const []const u8,
) Error!void {
const client_now = try heldHashes(arena, scen.clientStore());
const server_now = try heldHashes(arena, scen.serverStore());
const uni = try unionSorted(arena, c0, s0);
switch (scen.operation) {
.sync => {
if (!eqlSorted(client_now, uni) or !eqlSorted(server_now, uni)) return error.StoresDiverged;
if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
},
.pull => {
if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
if (!noPhantom(scen.clientStore())) return error.ResidualPhantom;
},
.push => {
if (!eqlSorted(server_now, uni)) return error.StoresDiverged;
if (!subsetSorted(c0, client_now)) return error.StoresDiverged;
if (!noPhantom(scen.serverStore()) or !noPhantom(scen.clientStore())) return error.ResidualPhantom;
},
.clone => {
// §8 — the client (which starts empty, generator contract) fetches
// the server's whole repository over paginated cfile batches, so it
// ends holding C₀ ∪ S₀ (= S₀) with no phantom; the server is
// read-only during clone and ends unchanged at S₀.
if (!eqlSorted(client_now, uni)) return error.StoresDiverged;
if (!eqlSorted(server_now, s0)) return error.StoresDiverged;
if (!noPhantom(scen.clientStore()) or !noPhantom(scen.serverStore())) return error.ResidualPhantom;
},
}
}
/// A stable digest of both stores' held sets plus the terminal state — the
/// reproducibility oracle.
fn outcomeDigest(arena: std.mem.Allocator, scen: *sync_gen.Scenario, state: sync.ClientState) Error!u64 {
var h = std.hash.Wyhash.init(0);
h.update(&.{@intFromEnum(state)});
for (try heldHashes(arena, scen.clientStore())) |x| h.update(x);
h.update("|");
for (try heldHashes(arena, scen.serverStore())) |x| h.update(x);
return h.final();
}
// ----------------------------------------------------------------------------
// Held-set helpers (sorted []const u8 sets)
// ----------------------------------------------------------------------------
fn heldCount(store: repostate.ArtifactStore) usize {
var count: usize = 0;
const n = store.count();
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
if (store.content(rid) != null) count += 1;
}
return count;
}
/// True iff no rid in `store` is a phantom (content-absent). §3.3 r24 — a
/// phantom is exactly `content(rid) == null`.
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;
}
/// The held hashes of `store`, duped into `arena` and sorted, so sets compare
/// across runs that assign rids differently.
fn heldHashes(arena: std.mem.Allocator, store: repostate.ArtifactStore) Error![][]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(arena, try arena.dupe(u8, store.hashOf(rid)));
}
const slice = try list.toOwnedSlice(arena);
std.mem.sort([]const u8, slice, {}, lessHash);
return slice;
}
fn lessHash(_: void, a: []const u8, b: []const u8) bool {
return std.mem.order(u8, a, b) == .lt;
}
/// Two sorted, duplicate-free sets are equal iff same length and elementwise.
fn eqlSorted(a: []const []const u8, b: []const []const u8) bool {
if (a.len != b.len) return false;
for (a, b) |x, y| if (!std.mem.eql(u8, x, y)) return false;
return true;
}
/// Every element of sorted `sub` appears in sorted `sup` (two-pointer merge).
fn subsetSorted(sub: []const []const u8, sup: []const []const u8) bool {
var i: usize = 0;
for (sup) |s| {
if (i == sub.len) break;
if (std.mem.eql(u8, sub[i], s)) i += 1;
}
return i == sub.len;
}
/// Merge two sorted, duplicate-free sets into their sorted union in `arena`.
fn unionSorted(arena: std.mem.Allocator, a: []const []const u8, b: []const []const u8) Error![][]const u8 {
var list: std.ArrayListUnmanaged([]const u8) = .empty;
var i: usize = 0;
var j: usize = 0;
while (i < a.len or j < b.len) {
if (j == b.len) {
try list.append(arena, a[i]);
i += 1;
} else if (i == a.len) {
try list.append(arena, b[j]);
j += 1;
} else switch (std.mem.order(u8, a[i], b[j])) {
.lt => {
try list.append(arena, a[i]);
i += 1;
},
.gt => {
try list.append(arena, b[j]);
j += 1;
},
.eq => {
try list.append(arena, a[i]);
i += 1;
j += 1;
},
}
}
return list.toOwnedSlice(arena);
}
// ----------------------------------------------------------------------------
// §7.5 private-content oracle
// ----------------------------------------------------------------------------
/// §7.5 — feed a client a server reply carrying a private artifact (a `private`
/// card then a full-content `file`) and report the receive-side outcome:
/// whether it was discarded, whether the store now holds it, and whether the
/// round still reached `terminated`. `granted` is the receiver's Private
/// capability (§7.5 r4/r5).
const PrivateOutcome = struct { discarded: bool, held: bool, terminated: bool };
fn drivePrivateReply(gpa: std.mem.Allocator, granted: bool) Error!PrivateOutcome {
var store = repostate.MemStore.init(gpa);
defer store.deinit();
var client = sync.Client{ .state = .exchange, .operation = .sync, .private_granted = granted };
defer client.deinit(gpa);
const hash = "aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00aa00";
const content = "private-artifact-bytes";
var reply: std.ArrayListUnmanaged(u8) = .empty;
defer reply.deinit(gpa);
try reply.appendSlice(gpa, "private\n"); // §7.5 L2359 — mark the next file private
const header = try std.fmt.allocPrint(gpa, "file {s} {d}\n", .{ hash, content.len });
defer gpa.free(header);
try reply.appendSlice(gpa, header);
try reply.appendSlice(gpa, content);
const obs = client.consumeReply(gpa, store.store(), reply.items) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
// A well-formed reply must parse; either error would itself be the §7.5
// leak/failure this oracle is here to catch.
error.FatalProtocolError, error.NotImplemented => return error.PrivateContentLeaked,
};
const rid = store.store().lookup(hash);
const held = rid != null and store.store().content(rid.?) != null;
return .{
.discarded = obs.private_content_discarded,
.held = held,
.terminated = client.state == .terminated,
};
}
// ----------------------------------------------------------------------------
// Tests — the invariants ARE the oracle: they encode §3.7 convergence/
// termination and §7.5 privacy. Seeds are literals; the seed reproduces a run.
// ----------------------------------------------------------------------------
const testing = std.testing;
/// A fixed smoke set: a handful of seeds through the full two-phase sweep.
/// Small mix of perfect-wire and heavy-fault scenarios (default `Params`), so
/// safety-under-faults AND liveness-after-heal are both exercised.
const smoke_seeds = [_]u64{ 1, 2, 3, 4, 5, 6, 7, 8 };
test "sync_sweep §3.7: the smoke sweep converges and terminates over every seed" {
for (smoke_seeds) |seed| try runSeed(testing.allocator, seed);
}
test "sync_sweep §3.7: a forced perfect wire converges and terminates" {
// No faults in either phase: the pure convergence path must still hold.
for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, .{ .perfect_wire_pct = 100 });
}
test "sync_sweep §3.7: a heavy-fault scenario converges once the wire heals" {
const heavy = sync_gen.Params{
.perfect_wire_pct = 0,
.max_drop_pct = 40,
.max_dup_pct = 40,
.max_corrupt_pct = 25,
.max_clog_pct = 25,
.max_mean_delay = 8,
};
for (smoke_seeds) |seed| _ = try simulate(testing.allocator, seed, heavy);
}
test "sync_sweep: the same seed reproduces an identical run" {
for (smoke_seeds) |seed| {
const a = try simulate(testing.allocator, seed, .{});
const b = try simulate(testing.allocator, seed, .{});
try testing.expectEqual(a.state, b.state);
try testing.expectEqual(a.digest, b.digest);
}
}
test "sync_sweep §7.5(a)(b): private content is not leaked and does not stall termination" {
// (a) L2383-2385 — without the Private capability the artifact is discarded,
// never stored, and (b) the discard sets no §3.7 continuation flag, so the
// round still terminates.
const denied = try drivePrivateReply(testing.allocator, false);
try testing.expect(denied.discarded);
try testing.expect(!denied.held); // §7.5(a): not interned — no leak into the store
try testing.expect(denied.terminated); // §7.5(b): discard did not stall §3.7 termination
// Negative space (L2381-2385) — with the capability the SAME reply IS stored
// and is not discarded, proving the gate is a real decision, not always-drop.
const allowed = try drivePrivateReply(testing.allocator, true);
try testing.expect(!allowed.discarded);
try testing.expect(allowed.held);
try testing.expect(allowed.terminated);
}