//! Sync clone benchmark harness (sync-performance-plan.md §M1).
//!
//! `zig build bench -- sync [--size small|medium|large|all] [--artifacts N]
//! [--blob B] [--delta-frac F] [--seed S] [--reps R]`. For each preset repo
//! size it generates a seeded synthetic server repository, clones it entirely
//! IN-PROCESS over the production sync seams, and prints a per-phase wall-time
//! breakdown plus protocol/byte counters. It MEASURES and REPORTS; it makes no
//! performance claim and encodes no target — the deliverable is attribution:
//! turning "clone is slower" into "X% rounds/exchange, Y% persist+fsync, Z%
//! decompress/reconstruct".
//!
//! ZERO measurement contamination of the core (Ousterhout): no timing or
//! counter code lives in sync.zig, sync_driver.zig, transport.zig, serve.zig,
//! logstore.zig, or repostate.zig. All instrumentation is here, at the harness
//! boundary — a bench-owned round loop times each production seam call
//! (`Client.buildRequest` → `transport.frame` → the server phase →
//! `transport.unframe` → `Client.consumeReply`), and `store.put` is sub-timed
//! by a `TimingStore` DECORATOR that wraps the client's `ArtifactStore` vtable
//! without touching repostate.zig. The bench round loop duplicates the ~12
//! driver lines of `sync_driver.runSync`; that copy is neutralized by a
//! mandatory control cross-check that reruns the real `runSync` on the same
//! seed and asserts equal round count and equal final store digest.
//!
//! The server phase is not timed as one opaque `serve.handleXfer` span, because
//! that conflates three costs a clone-speed decision must tell apart: the
//! inbound `zlibContainerDecode` of a per-round igot flood the clone re-sends
//! and then ignores, the `respondWithMedia` parse+store-walk, and the per-cfile
//! `zlibContainerEncode` that is the real artifact compression. So the phase is
//! decomposed over the SAME pub steps `handleXfer` composes — decode, respond,
//! then the outer reframe (encode for an ordinary reply, a copy for the §8.1
//! uncompressed clone reply) — each timed at the boundary. The decomposition is
//! a reconstruction, so every round its composed reply is cross-checked
//! byte-identical to the real `serve.handleXfer` (`error.ControlDivergence`
//! otherwise), exactly as the round loop is cross-checked against `runSync`.
//! `scanRequest` and `emitCloneReply` are file-private, so parse cannot be
//! timed apart from the store-walk+compress inside `respond` without pub-ifying
//! core (project rule: never). The compression share is instead attributed by a
//! standalone `zlibContainerEncode` micro-measurement over the corpus — faithful
//! because a clone emits each artifact as exactly one full-content cfile via the
//! same `writeCfile`→`zlibContainerEncode(.default)` — and decode+parse is the
//! reported remainder of the server phase.
//!
//! Determinism: the seed is the whole universe — corpus bytes, sizes, and the
//! delta-chain shape all derive from one `--seed`. The ONE deliberately-injected
//! ambient effect is the monotonic clock (`Io.Timestamp.now(io, .awake)`) plus
//! the persist-phase temp file, both reached only through the injected `std.Io`
//! (same as logstore_benchmark). Counters are pure functions of the seed: two
//! runs at one seed print byte-identical counter blocks (asserted at runtime,
//! `error.NondeterministicRun` otherwise); timings vary and are never asserted.
const std = @import("std");
const Io = std.Io;
const zit = @import("zit");
const sync = zit.sync;
const repostate = zit.repostate;
const storage = @import("zit_storage");
const logstore = @import("zit_logstore");
const serve = @import("serve.zig");
const transport = @import("transport.zig");
const sync_driver = @import("sync_driver.zig");
/// Closed harness error set. Seam errors (build/frame/server/unframe/consume,
/// logstore, storage) fold in explicitly; `ControlDivergence` fires when the
/// in-process loop disagrees with real `runSync`, `NondeterministicRun` when a
/// rep's counters differ from the first rep, `CounterScanFailed` when the
/// harness cannot walk its own card stream, `BadArgs` on a rejected flag.
/// Never `anyerror`. `RunError` adds the writer's own error set for reporting.
pub const Error = error{
OutOfMemory,
BadArgs,
Io,
Storage,
Corrupt,
ControlDivergence,
NondeterministicRun,
CounterScanFailed,
RequestTooLarge,
RequestMalformed,
ResponseTooLarge,
FatalProtocolError,
NotImplemented,
RoundLimitExceeded,
};
pub const RunError = Error || Io.Writer.Error;
// ---------------------------------------------------------------------------
// Explicit fixed caps (TigerStyle bounds); each overflow -> BadArgs.
// ---------------------------------------------------------------------------
const max_artifacts: u64 = 1_000_000;
const min_blob: u64 = 8; // 8-byte index prefix guarantees distinct content.
const max_blob: u64 = 1 << 20; // 1 MiB.
const max_delta_frac: u64 = 100;
const max_reps: u64 = 100;
/// A clone pages the repository in `clone_batch_max`-sized cfile batches, so
/// rounds ≈ artifacts / batch + a small constant; this bound is the runaway
/// backstop, far above any real clone's round count (§3.7 termination).
const max_bench_rounds: usize = 1 << 20;
/// §9.1 — well-formed, whitespace-free 16-hex identifiers so a clone reply's
/// `push SERVERCODE PROJECTCODE` card carries three tokens. Informational for
/// the benchmark (both endpoints are synthetic); reused across every size.
const servercode: []const u8 = "0000000000000000";
const projectcode: []const u8 = "0000000000000000";
// ---------------------------------------------------------------------------
// Presets and configuration (all inputs arrive from argv only).
// ---------------------------------------------------------------------------
const Size = enum { small, medium, large };
/// One synthetic repository shape. `delta_frac` is the percent chance an
/// artifact is a seeded mutation of its predecessor (a delta-able chain); it
/// makes server-side delta cfiles POSSIBLE but never forces them — the cfile
/// counter observes whether the server actually emits them.
const Preset = struct { artifacts: u64, blob: u64, delta_frac: u64 };
fn presetFor(s: Size) Preset {
return switch (s) {
.small => .{ .artifacts = 100, .blob = 1024, .delta_frac = 60 },
.medium => .{ .artifacts = 5000, .blob = 4096, .delta_frac = 60 },
.large => .{ .artifacts = 50_000, .blob = 8192, .delta_frac = 60 },
};
}
/// Which repositories to run: every preset, a single named preset, or one
/// custom shape assembled from `--artifacts/--blob/--delta-frac` overrides.
const Selection = union(enum) { all, one: Size, custom: Preset };
const Config = struct {
seed: u64 = 42,
reps: u64 = 3,
selection: Selection = .all,
};
// ---------------------------------------------------------------------------
// Entry point + argument parsing.
// ---------------------------------------------------------------------------
/// Dispatched from logstore_benchmark's `run` when argv[1] == "sync". `args`
/// is the whole process argv; the sync flags begin at index 2.
pub fn run(gpa: std.mem.Allocator, io: Io, args: []const []const u8, out: *Io.Writer) RunError!void {
const cfg = try parseArgs(args);
switch (cfg.selection) {
.all => {
try runSize(gpa, io, cfg, .small, out);
try runSize(gpa, io, cfg, .medium, out);
try runSize(gpa, io, cfg, .large, out);
},
.one => |s| try runSize(gpa, io, cfg, s, out),
.custom => |p| try runCustom(gpa, io, cfg, p, out),
}
try out.flush();
}
fn parseArgs(args: []const []const u8) Error!Config {
var cfg = Config{};
var custom: ?Preset = null;
var i: usize = 2; // argv[0]=exe, argv[1]="sync".
while (i < args.len) : (i += 2) {
if (i + 1 >= args.len) return error.BadArgs; // flag with no value
const flag = args[i];
const val = args[i + 1];
if (std.mem.eql(u8, flag, "--size")) {
cfg.selection = try parseSize(val);
} else if (std.mem.eql(u8, flag, "--seed")) {
cfg.seed = std.fmt.parseInt(u64, val, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--reps")) {
cfg.reps = std.fmt.parseInt(u64, val, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--artifacts")) {
var p = custom orelse presetFor(.medium);
p.artifacts = std.fmt.parseInt(u64, val, 0) catch return error.BadArgs;
custom = p;
} else if (std.mem.eql(u8, flag, "--blob")) {
var p = custom orelse presetFor(.medium);
p.blob = std.fmt.parseInt(u64, val, 0) catch return error.BadArgs;
custom = p;
} else if (std.mem.eql(u8, flag, "--delta-frac")) {
var p = custom orelse presetFor(.medium);
p.delta_frac = std.fmt.parseInt(u64, val, 0) catch return error.BadArgs;
custom = p;
} else return error.BadArgs;
}
if (custom) |p| cfg.selection = .{ .custom = p };
if (cfg.reps < 1 or cfg.reps > max_reps) return error.BadArgs;
return cfg;
}
fn parseSize(s: []const u8) Error!Selection {
if (std.mem.eql(u8, s, "all")) return .all;
if (std.mem.eql(u8, s, "small")) return .{ .one = .small };
if (std.mem.eql(u8, s, "medium")) return .{ .one = .medium };
if (std.mem.eql(u8, s, "large")) return .{ .one = .large };
return error.BadArgs;
}
fn validatePreset(p: Preset) Error!void {
if (p.artifacts < 1 or p.artifacts > max_artifacts) return error.BadArgs;
if (p.blob < min_blob or p.blob > max_blob) return error.BadArgs;
if (p.delta_frac > max_delta_frac) return error.BadArgs;
}
// ---------------------------------------------------------------------------
// Deterministic corpus (pure — no clock, no file, no ambient entropy).
// ---------------------------------------------------------------------------
/// Seed a server-side `MemStore` with `p.artifacts` content-bearing entries.
/// Artifact 0 is fresh random bytes; each later artifact is, with probability
/// `p.delta_frac`, a small seeded mutation of its predecessor (a delta-able
/// chain), else fresh random bytes. Every content buffer carries its index in
/// the first 8 bytes, so all hashes are distinct and `count == artifacts`
/// holds. All randomness derives from `seed` alone via one `DefaultPrng` with a
/// fixed draw order, so a run replays exactly.
fn generateCorpus(gpa: std.mem.Allocator, ms: *repostate.MemStore, seed: u64, p: Preset) Error!void {
try validatePreset(p);
const n: usize = @intCast(p.artifacts);
const blob: usize = @intCast(p.blob);
var prng = std.Random.DefaultPrng.init(seed);
const rand = prng.random();
const cur = gpa.alloc(u8, blob) catch return error.OutOfMemory;
defer gpa.free(cur);
const prev = gpa.alloc(u8, blob) catch return error.OutOfMemory;
defer gpa.free(prev);
var i: usize = 0;
while (i < n) : (i += 1) {
const chained = i != 0 and rand.intRangeLessThan(u64, 0, 100) < p.delta_frac;
if (chained) {
@memcpy(cur, prev);
const muts = 1 + rand.uintLessThan(usize, 8);
var m: usize = 0;
while (m < muts) : (m += 1) cur[rand.uintLessThan(usize, blob)] = rand.int(u8);
} else {
rand.bytes(cur);
}
std.mem.writeInt(u64, cur[0..8], i, .little); // distinct content -> distinct hash
const hash = hexHash(cur);
_ = ms.store().put(&hash, cur) catch return error.OutOfMemory;
@memcpy(prev, cur);
}
std.debug.assert(ms.store().count() == n);
}
/// Lowercase 64-hex SHA-256 of `content` — the wire-real artifact name.
fn hexHash(content: []const u8) [64]u8 {
var h: [32]u8 = undefined;
std.crypto.hash.sha2.Sha256.hash(content, &h, .{});
return std.fmt.bytesToHex(h, .lower);
}
/// Raw 32-byte SHA-256 of `content` — the LogStore content address (the wire
/// NAME is the hex of this).
fn rawHash(content: []const u8) [32]u8 {
var h: [32]u8 = undefined;
std.crypto.hash.sha2.Sha256.hash(content, &h, .{});
return h;
}
// ---------------------------------------------------------------------------
// TimingStore — the measurement DECORATOR over the client's ArtifactStore.
// ---------------------------------------------------------------------------
/// An `ArtifactStore` vtable wrapper that forwards every call to an inner
/// store while accumulating wall time and call counts for `put` and `content`
/// (the two seams `consumeReply` drives during ingest and delta-basis lookup).
/// This is the whole reason no timer enters sync.zig: the consume phase's
/// store cost is attributed here, at the vtable boundary, leaving repostate.zig
/// and sync.zig byte-identical.
const TimingStore = struct {
inner: repostate.ArtifactStore,
io: Io,
put_ns: u64 = 0,
content_ns: u64 = 0,
put_calls: u64 = 0,
content_calls: u64 = 0,
fn store(self: *TimingStore) repostate.ArtifactStore {
return .{ .context = self, .vtable = &vtable };
}
/// The store time charged to a round for the store_put sub-attribution:
/// put plus the content lookups a delta reconstruction performs.
fn storeNs(self: *const TimingStore) u64 {
return self.put_ns + self.content_ns;
}
const vtable = repostate.ArtifactStore.VTable{
.intern = internImpl,
.put = putImpl,
.content = contentImpl,
.hashOf = hashOfImpl,
.lookup = lookupImpl,
.count = countImpl,
};
fn putImpl(ctx: *anyopaque, hash: []const u8, content: []const u8) error{OutOfMemory}!repostate.Rid {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
const t0 = now(self.io);
const rid = try self.inner.put(hash, content);
self.put_ns += elapsed(t0, now(self.io));
self.put_calls += 1;
return rid;
}
fn contentImpl(ctx: *anyopaque, rid: repostate.Rid) ?[]const u8 {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
const t0 = now(self.io);
const r = self.inner.content(rid);
self.content_ns += elapsed(t0, now(self.io));
self.content_calls += 1;
return r;
}
fn internImpl(ctx: *anyopaque, hash: []const u8) error{OutOfMemory}!repostate.Rid {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
return self.inner.intern(hash);
}
fn hashOfImpl(ctx: *anyopaque, rid: repostate.Rid) []const u8 {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
return self.inner.hashOf(rid);
}
fn lookupImpl(ctx: *anyopaque, hash: []const u8) ?repostate.Rid {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
return self.inner.lookup(hash);
}
fn countImpl(ctx: *anyopaque) usize {
const self: *TimingStore = @ptrCast(@alignCast(ctx));
return self.inner.count();
}
};
// ---------------------------------------------------------------------------
// Counters (all collected OUTSIDE timed spans).
// ---------------------------------------------------------------------------
const Counters = struct {
rounds: u64 = 0,
// request cards
req_igot: u64 = 0,
req_gimme: u64 = 0,
req_file: u64 = 0,
req_clone: u64 = 0,
req_pull: u64 = 0,
req_push: u64 = 0,
// reply cards
reply_cfile_full: u64 = 0,
reply_cfile_delta: u64 = 0,
reply_file: u64 = 0,
reply_igot: u64 = 0,
reply_push: u64 = 0,
reply_clone_seqno: u64 = 0,
// bytes both ways (plaintext = post/pre framing, wire = framed container)
req_plaintext: u64 = 0,
req_wire: u64 = 0,
reply_plaintext: u64 = 0,
reply_wire: u64 = 0,
// convergence yield
artifacts_received: u64 = 0,
phantom_peak: u64 = 0,
store_put_calls: u64 = 0,
// persist phase
fsync_count: u64 = 0,
disk_bytes: u64 = 0,
content_bytes: u64 = 0,
};
/// Walk an LF-delimited card stream, tallying cards and skipping byte-counted
/// content payloads exactly as `consumeReply` does. `is_reply` routes the
/// shared `igot`/`push` operators to the correct direction. A parse failure on
/// the harness's own well-formed stream is a bug, surfaced as CounterScanFailed.
fn scanCards(body: []const u8, c: *Counters, is_reply: bool) Error!void {
var cursor: usize = 0;
while (cursor < body.len) {
const line = sync.readLine(body, cursor) catch return error.CounterScanFailed;
var next = line.next;
if (!sync.isCommentLine(line.bytes)) {
const cc = sync.classifyCard(line.bytes);
if (cc.operator) |op| switch (op) {
.igot => if (is_reply) {
c.reply_igot += 1;
} else {
c.req_igot += 1;
},
.gimme => c.req_gimme += 1,
.clone => c.req_clone += 1,
.pull => c.req_pull += 1,
.push => if (is_reply) {
c.reply_push += 1;
} else {
c.req_push += 1;
},
.clone_seqno => c.reply_clone_seqno += 1,
.file => {
const fc = sync.parseFileCard(cc.tokens) catch return error.CounterScanFailed;
if (is_reply) {
c.reply_file += 1;
c.artifacts_received += 1;
} else c.req_file += 1;
next = (sync.consumeContentPayload(body, line.next, fc.size) catch return error.CounterScanFailed).next;
},
.cfile => {
const cf = sync.parseCfileCard(cc.tokens) catch return error.CounterScanFailed;
if (cf.deltasrc != null) c.reply_cfile_delta += 1 else c.reply_cfile_full += 1;
c.artifacts_received += 1;
next = (sync.consumeContentPayload(body, line.next, cf.compressed_size) catch return error.CounterScanFailed).next;
},
else => {}, // pragma/login/cookie/config/error/message: not counted
};
}
std.debug.assert(next > cursor);
cursor = next;
}
std.debug.assert(cursor == body.len);
}
/// §3.3 r24 — a phantom is a content-absent rid. Counted through the INNER
/// store so the scan is untimed and never charged to store_ns.
fn countPhantoms(store: repostate.ArtifactStore) u64 {
const n = store.count();
var c: u64 = 0;
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
if (store.content(rid) == null) c += 1;
}
return c;
}
// ---------------------------------------------------------------------------
// The measured path — a bench-owned round loop over the production seams.
// ---------------------------------------------------------------------------
/// Per-round wall-time samples, one growable series per timed phase.
const RoundSamples = struct {
build: std.ArrayListUnmanaged(u64) = .empty,
frame: std.ArrayListUnmanaged(u64) = .empty,
server: std.ArrayListUnmanaged(u64) = .empty, // = decode + respond + reframe
decode: std.ArrayListUnmanaged(u64) = .empty, // inbound zlibContainerDecode
respond: std.ArrayListUnmanaged(u64) = .empty, // respondWithMedia: parse+walk+compress
reframe: std.ArrayListUnmanaged(u64) = .empty, // outer encode / clone-reply copy
unframe: std.ArrayListUnmanaged(u64) = .empty,
consume: std.ArrayListUnmanaged(u64) = .empty,
store: std.ArrayListUnmanaged(u64) = .empty,
fn deinit(self: *RoundSamples, gpa: std.mem.Allocator) void {
self.build.deinit(gpa);
self.frame.deinit(gpa);
self.server.deinit(gpa);
self.decode.deinit(gpa);
self.respond.deinit(gpa);
self.reframe.deinit(gpa);
self.unframe.deinit(gpa);
self.consume.deinit(gpa);
self.store.deinit(gpa);
}
};
/// Drive one full in-process clone against `server_store`, timing each seam at
/// the boundary and tallying counters. Mirrors `sync_driver.runSync` exactly —
/// same build → frame → server → deframe → consume sequence, same clone_seqno
/// carry, same `.exchange`/`.clone_round0` loop condition — decomposed so a
/// timestamp falls between each seam. Content ingest is charged to store_ns by
/// the `TimingStore` the caller passes. Returns the round count.
fn runClone(
gpa: std.mem.Allocator,
io: Io,
server_store: repostate.ArtifactStore,
ts: *TimingStore,
rs: *RoundSamples,
c: *Counters,
) Error!usize {
const store = ts.store();
var server = sync.Server{ .version = serve.server_version, .self_servercode = servercode, .project_code = projectcode };
var client = sync.Client{ .state = .clone_round0, .operation = .clone };
defer client.deinit(gpa);
var req_buf: std.ArrayListUnmanaged(u8) = .empty;
defer req_buf.deinit(gpa);
var rounds: usize = 0;
while (client.state == .exchange or client.state == .clone_round0) {
if (rounds == max_bench_rounds) return error.RoundLimitExceeded;
rounds += 1;
const t0 = now(io);
const request = client.buildRequest(gpa, store, &req_buf) catch |e| return mapBuild(e);
const t1 = now(io);
const container = transport.frame(gpa, request) catch |e| return mapFrame(e);
defer gpa.free(container);
const t2 = now(io);
// The server phase, decomposed over handleXfer's own pub steps so decode
// (inbound igot-flood decompression), respond (parse+walk+compress), and
// reframe are timed apart; `st` carries the three sub-spans.
const sp = try serverPhase(gpa, io, &server, server_store, container);
const reply = sp.reply;
defer gpa.free(reply.body);
const st = sp.times;
const t3 = now(io);
// The composed reply MUST equal the real monolithic handleXfer, byte for
// byte — the decomposition's cross-check, run OUTSIDE the timed span (the
// clone never mutates server_store, so a second answer is safe). A drift
// errors rather than reports a breakdown for bytes production never sends.
{
const oracle = serve.handleXfer(gpa, &server, server_store, container) catch |e| return mapXfer(e);
defer gpa.free(oracle.body);
if (!std.mem.eql(u8, oracle.body, reply.body) or
!std.mem.eql(u8, oracle.content_type, reply.content_type))
return error.ControlDivergence;
}
// §4.1 — an ordinary reply is the compressed container (deframe it); a
// §8.1 clone reply is x-fossil-uncompressed and passes through (its
// decompression cost lives in the cfile payloads, inside consume).
const compressed = std.mem.eql(u8, reply.content_type, serve.xfer_media_type);
const plaintext = if (compressed) (transport.unframe(gpa, reply.body) catch |e| return mapUnframe(e)) else reply.body;
defer if (compressed) gpa.free(plaintext);
const t4 = now(io);
const store_before = ts.storeNs();
const obs = client.consumeReply(gpa, store, plaintext) catch |e| return mapConsume(e);
const t5 = now(io);
const store_after = ts.storeNs();
if (client.operation == .clone) client.clone_seqno = obs.clone_seqno_next;
try recordRound(gpa, rs, t0, t1, t2, t3, t4, t5, st, store_after - store_before);
// Counters and the phantom scan sit OUTSIDE every timed span.
c.req_plaintext += request.len;
c.req_wire += container.len;
c.reply_plaintext += plaintext.len;
c.reply_wire += reply.body.len;
try scanCards(request, c, false);
try scanCards(plaintext, c, true);
const phantoms = countPhantoms(ts.inner);
if (phantoms > c.phantom_peak) c.phantom_peak = phantoms;
}
// §3.8 — the loop exits only on the clean terminal (a fatal abort left
// through consumeReply's error return above).
std.debug.assert(client.state == .terminated);
c.rounds = rounds;
c.store_put_calls = ts.put_calls;
return rounds;
}
fn recordRound(
gpa: std.mem.Allocator,
rs: *RoundSamples,
t0: Io.Timestamp,
t1: Io.Timestamp,
t2: Io.Timestamp,
t3: Io.Timestamp,
t4: Io.Timestamp,
t5: Io.Timestamp,
st: ServerTimes,
store_ns: u64,
) Error!void {
// t2/t3 bracket the server phase for frame/unframe; the server span itself is
// `st` (its three sub-timings, measured inside `serverPhase`).
rs.build.append(gpa, elapsed(t0, t1)) catch return error.OutOfMemory;
rs.frame.append(gpa, elapsed(t1, t2)) catch return error.OutOfMemory;
rs.decode.append(gpa, st.decode) catch return error.OutOfMemory;
rs.respond.append(gpa, st.respond) catch return error.OutOfMemory;
rs.reframe.append(gpa, st.reframe) catch return error.OutOfMemory;
// The parent server sample is the exact sum of its measured parts, so the
// sub-lines partition it without slack from the untimed control flow between.
rs.server.append(gpa, st.decode + st.respond + st.reframe) catch return error.OutOfMemory;
rs.unframe.append(gpa, elapsed(t3, t4)) catch return error.OutOfMemory;
rs.consume.append(gpa, elapsed(t4, t5)) catch return error.OutOfMemory;
rs.store.append(gpa, store_ns) catch return error.OutOfMemory;
}
// ---------------------------------------------------------------------------
// The decomposed server phase — handleXfer, reconstructed from its pub steps so
// each is timed, and byte-verified against the real handleXfer by the caller.
// ---------------------------------------------------------------------------
/// The three timed sub-spans of one server answer: `decode` is the inbound §4.1
/// `zlibContainerDecode` (the growing igot-flood the clone re-announces every
/// round), `respond` is `respondWithMedia` (parse + store walk + per-cfile
/// compression), `reframe` is the outer framing (`zlibContainerEncode` for an
/// ordinary reply, a copy for the §8.1 uncompressed clone reply).
const ServerTimes = struct { decode: u64, respond: u64, reframe: u64 };
/// A composed server answer plus its sub-span timings. `reply.body` is
/// caller-owned; the caller cross-checks `reply` against `serve.handleXfer`.
const ServerPhase = struct { reply: serve.XferReply, times: ServerTimes };
/// Mirror `serve.handleXfer` over the pub steps it composes — the §4 L1148-1149
/// bound, then decode → respond → reframe — timing each of the three. Identical
/// to `handleXfer` by construction (same steps, same media decision), so the
/// caller's per-round byte-identity guard catches any drift; the whole reason
/// this is not one `handleXfer` call is to place a timestamp between the steps.
fn serverPhase(
gpa: std.mem.Allocator,
io: Io,
server: *const sync.Server,
store: repostate.ArtifactStore,
container: []const u8,
) Error!ServerPhase {
if (container.len > serve.max_request_body) return error.RequestTooLarge;
const d0 = now(io);
const request = sync.zlibContainerDecode(gpa, container, serve.max_request_body) catch |e| return mapDecode(e);
defer gpa.free(request);
const d1 = now(io);
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(gpa);
const reply = server.respondWithMedia(gpa, store, request, &out) catch |e| return mapRespond(e);
const d2 = now(io);
// §4.1 — an ordinary reply is re-framed into a container; a §8.1 clone reply
// passes through uncompressed (its cfiles are already stored compressed).
const props = sync.media_props.get(reply.media);
const body = if (props.outer_compressed)
(sync.zlibContainerEncode(gpa, reply.body, .default) catch |e| return mapEncode(e))
else
(gpa.dupe(u8, reply.body) catch return error.OutOfMemory);
const d3 = now(io);
return .{
.reply = .{ .body = body, .content_type = props.wire },
.times = .{ .decode = elapsed(d0, d1), .respond = elapsed(d1, d2), .reframe = elapsed(d2, d3) },
};
}
/// The clone's total cfile-compression cost, measured standalone: time one
/// `zlibContainerEncode(.default)` per corpus artifact — exactly what
/// `emitCloneReply`→`writeCfile` does, and each artifact is emitted as one
/// full-content cfile across the whole clone, so this sum equals the server
/// phase's real compression work. It attributes compression apart from the
/// igot decode+parse that `respondWithMedia` bundles them with (scanRequest /
/// emitCloneReply being file-private, that split is not reachable at the pub
/// boundary). Reported as a diagnostic, not summed into wall.
fn compressionMicro(gpa: std.mem.Allocator, io: Io, store: repostate.ArtifactStore) Error!u64 {
const n = store.count();
var total: u64 = 0;
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
const bytes = store.content(rid) orelse continue; // phantoms carry no stored bytes
const t0 = now(io);
const c = sync.zlibContainerEncode(gpa, bytes, .default) catch |e| return mapEncode(e);
total += elapsed(t0, now(io));
gpa.free(c);
}
return total;
}
// ---------------------------------------------------------------------------
// Persist phase — mirrors main.zig persistStore + storeClonedProjectCode.
// ---------------------------------------------------------------------------
const PersistTimes = struct { persist_ns: u64, fsync_ns: u64, reopen_ns: u64 };
/// After convergence, write every content-bearing artifact into a fresh
/// LogStore temp file (persist), commit it (one fsync), then reopen the written
/// file (the production project-code write reopens and rescans every frame —
/// hypothesis 4). Fills the disk/content/fsync counters and returns the three
/// phase times. Restated over the public logstore API (~12 lines); the mirrored
/// production path is main.zig's persistStore.
fn persistPhase(gpa: std.mem.Allocator, io: Io, client: repostate.ArtifactStore, c: *Counters) Error!PersistTimes {
var temp = try Temp.create(io);
defer temp.cleanup(io);
var fs = storage.FileStorage.init(io, temp.file, 0);
var ls = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| return mapOpen(e);
const n = client.count();
var content_bytes: u64 = 0;
const t0 = now(io);
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
const bytes = client.content(rid) orelse continue; // phantoms (none post-clone) skipped
// §7.7 — the LogStore is addressed by the RAW digest (<=32 bytes), not
// the 64-hex wire NAME the sync store keys on; the name is hex of this.
_ = ls.put(&rawHash(bytes), bytes) catch |e| return mapPut(e);
content_bytes += bytes.len;
}
const t1 = now(io);
ls.commit() catch return error.Storage; // one fsync over the whole clone
const t2 = now(io);
const disk = ls.size();
ls.deinit(gpa);
var fs2 = storage.FileStorage.init(io, temp.file, disk);
const t3 = now(io);
var ls2 = logstore.LogStore.open(gpa, fs2.storageFile(), 0) catch |e| return mapOpen(e);
const t4 = now(io);
ls2.deinit(gpa);
c.disk_bytes = disk;
c.content_bytes = content_bytes;
c.fsync_count = 1;
return .{ .persist_ns = elapsed(t0, t1), .fsync_ns = elapsed(t1, t2), .reopen_ns = elapsed(t3, t4) };
}
// ---------------------------------------------------------------------------
// Control cross-check — the D1 duplication mitigation.
// ---------------------------------------------------------------------------
/// The in-process convergence fake driving the REAL `sync_driver.runSync`:
/// `exchange` answers each request with the sans-IO `Server.respond` over a
/// server store, duped to a caller-owned slice. Structurally identical to the
/// sync_driver_test fake — this is the production driver's own transport shape.
const ServerTransport = struct {
server: sync.Server,
store: repostate.ArtifactStore,
reply_buf: std.ArrayListUnmanaged(u8) = .empty,
pub const Error = sync.Server.RespondError; // {OutOfMemory, ResponseTooLarge}
fn deinit(self: *ServerTransport, gpa: std.mem.Allocator) void {
self.reply_buf.deinit(gpa);
}
pub fn exchange(self: *ServerTransport, gpa: std.mem.Allocator, request: []const u8) ServerTransport.Error![]u8 {
const reply = try self.server.respond(gpa, self.store, request, &self.reply_buf);
return gpa.dupe(u8, reply);
}
};
/// Regenerate the identical scenario from `seed` and clone it through the real
/// `sync_driver.runSync` (not a second copy of the bench loop). The bench's
/// numbers are printed only if this agrees on both the round count and the
/// final client-store digest — a drifted bench loop errors out here rather than
/// reporting a breakdown for behavior production never exhibits.
fn crossCheck(gpa: std.mem.Allocator, seed: u64, p: Preset, bench_rounds: usize, bench_digest: [32]u8) Error!void {
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, seed, p);
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var fake = ServerTransport{
.server = .{ .version = serve.server_version, .self_servercode = servercode, .project_code = projectcode },
.store = server_ms.store(),
};
defer fake.deinit(gpa);
var client = sync.Client{ .state = .clone_round0, .operation = .clone };
defer client.deinit(gpa);
const stats = sync_driver.runSync(gpa, &client, client_ms.store(), &fake, max_bench_rounds) catch |e| return mapRun(e);
if (stats.rounds != bench_rounds) return error.ControlDivergence;
const digest = try storeDigest(gpa, client_ms.store());
if (!std.mem.eql(u8, &digest, &bench_digest)) return error.ControlDivergence;
}
/// A SHA-256 over the store's held (content-bearing) artifacts in hash-sorted
/// order, so it is insertion-order-independent: two stores holding the same
/// content-addressed set digest identically regardless of clone-round ordering.
fn storeDigest(gpa: std.mem.Allocator, store: repostate.ArtifactStore) Error![32]u8 {
const n = store.count();
var held: std.ArrayListUnmanaged(repostate.Rid) = .empty;
defer held.deinit(gpa);
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
if (store.content(rid) != null) held.append(gpa, rid) catch return error.OutOfMemory;
}
const Ctx = struct {
s: repostate.ArtifactStore,
fn less(self: @This(), a: repostate.Rid, b: repostate.Rid) bool {
return std.mem.order(u8, self.s.hashOf(a), self.s.hashOf(b)) == .lt;
}
};
std.mem.sort(repostate.Rid, held.items, Ctx{ .s = store }, Ctx.less);
var h = std.crypto.hash.sha2.Sha256.init(.{});
for (held.items) |r| {
h.update(store.hashOf(r));
h.update(store.content(r).?);
}
return h.finalResult();
}
// ---------------------------------------------------------------------------
// One size: reps, determinism guard, cross-check, report.
// ---------------------------------------------------------------------------
/// Median-of-reps totals for the once-per-rep and summed-per-round phases,
/// plus the last rep's per-round samples for percentiles.
const SizeResult = struct {
build: u64,
frame: u64,
server: u64,
decode: u64,
respond: u64,
reframe: u64,
compress: u64, // standalone cfile-compression micro-measurement (not in wall)
unframe: u64,
consume: u64,
store: u64,
persist: u64,
fsync: u64,
reopen: u64,
counters: Counters,
rounds: usize,
last: RoundSamples,
fn deinit(self: *SizeResult, gpa: std.mem.Allocator) void {
self.last.deinit(gpa);
}
};
fn runSize(gpa: std.mem.Allocator, io: Io, cfg: Config, s: Size, out: *Io.Writer) RunError!void {
var result = try measure(gpa, io, cfg, presetFor(s));
defer result.deinit(gpa);
try report(out, cfg, @tagName(s), presetFor(s), &result);
}
fn runCustom(gpa: std.mem.Allocator, io: Io, cfg: Config, p: Preset, out: *Io.Writer) RunError!void {
var result = try measure(gpa, io, cfg, p);
defer result.deinit(gpa);
try report(out, cfg, "custom", p, &result);
}
/// One rep's outcome: the per-phase totals (summed over rounds for the round
/// phases, single-shot for persist/fsync/reopen), the counters, round count,
/// final store digest, and the per-round samples (kept only for the last rep).
const RepOutcome = struct {
build: u64,
frame: u64,
server: u64,
decode: u64,
respond: u64,
reframe: u64,
compress: u64,
unframe: u64,
consume: u64,
store: u64,
persist: u64,
fsync: u64,
reopen: u64,
counters: Counters,
rounds: usize,
digest: [32]u8,
samples: RoundSamples,
};
/// Per-rep phase totals across the reps, on bounded fixed storage (reps is
/// validated <= max_reps), so the median is computed without any allocation.
const RepTotals = struct {
build: [max_reps]u64 = undefined,
frame: [max_reps]u64 = undefined,
server: [max_reps]u64 = undefined,
decode: [max_reps]u64 = undefined,
respond: [max_reps]u64 = undefined,
reframe: [max_reps]u64 = undefined,
compress: [max_reps]u64 = undefined,
unframe: [max_reps]u64 = undefined,
consume: [max_reps]u64 = undefined,
store: [max_reps]u64 = undefined,
persist: [max_reps]u64 = undefined,
fsync: [max_reps]u64 = undefined,
reopen: [max_reps]u64 = undefined,
fn record(self: *RepTotals, rep: usize, o: RepOutcome) void {
self.build[rep] = o.build;
self.frame[rep] = o.frame;
self.server[rep] = o.server;
self.decode[rep] = o.decode;
self.respond[rep] = o.respond;
self.reframe[rep] = o.reframe;
self.compress[rep] = o.compress;
self.unframe[rep] = o.unframe;
self.consume[rep] = o.consume;
self.store[rep] = o.store;
self.persist[rep] = o.persist;
self.fsync[rep] = o.fsync;
self.reopen[rep] = o.reopen;
}
};
/// Drive one full timed clone on fresh client state over `server_store`, then
/// its persist phase, returning the rep's phase totals + counters + digest +
/// per-round samples (caller owns `samples`).
fn runRep(gpa: std.mem.Allocator, io: Io, server_store: repostate.ArtifactStore) Error!RepOutcome {
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var ts = TimingStore{ .inner = client_ms.store(), .io = io };
var rs: RoundSamples = .{};
errdefer rs.deinit(gpa);
var c = Counters{};
const rounds = try runClone(gpa, io, server_store, &ts, &rs, &c);
const pt = try persistPhase(gpa, io, ts.inner, &c);
// The standalone cfile-compression cost, over the server corpus (untimed by
// the round loop) — the attribution of compression apart from igot decode+parse.
const compress = try compressionMicro(gpa, io, server_store);
const digest = try storeDigest(gpa, ts.inner);
return .{
.build = sum(rs.build.items),
.frame = sum(rs.frame.items),
.server = sum(rs.server.items),
.decode = sum(rs.decode.items),
.respond = sum(rs.respond.items),
.reframe = sum(rs.reframe.items),
.compress = compress,
.unframe = sum(rs.unframe.items),
.consume = sum(rs.consume.items),
.store = sum(rs.store.items),
.persist = pt.persist_ns,
.fsync = pt.fsync_ns,
.reopen = pt.reopen_ns,
.counters = c,
.rounds = rounds,
.digest = digest,
.samples = rs,
};
}
/// Run `cfg.reps` timed clones over one shared server corpus, asserting counter
/// determinism across reps, then cross-check the first rep against real
/// `runSync`. Returns median phase totals + the last rep's per-round samples.
fn measure(gpa: std.mem.Allocator, io: Io, cfg: Config, p: Preset) Error!SizeResult {
try validatePreset(p);
const reps: usize = @intCast(cfg.reps);
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, cfg.seed, p); // clone never mutates the server -> generate once
var totals = RepTotals{};
var base: ?Counters = null;
var last: RoundSamples = .{};
errdefer last.deinit(gpa);
var first = RepOutcome{ .build = 0, .frame = 0, .server = 0, .decode = 0, .respond = 0, .reframe = 0, .compress = 0, .unframe = 0, .consume = 0, .store = 0, .persist = 0, .fsync = 0, .reopen = 0, .counters = .{}, .rounds = 0, .digest = undefined, .samples = .{} };
var rep: usize = 0;
while (rep < reps) : (rep += 1) {
var o = try runRep(gpa, io, server_ms.store());
if (base) |b| {
if (!std.meta.eql(b, o.counters)) {
o.samples.deinit(gpa);
return error.NondeterministicRun;
}
} else {
base = o.counters;
first = o;
}
totals.record(rep, o);
if (rep == reps - 1) last = o.samples else o.samples.deinit(gpa);
}
// The mandatory D1 mitigation — never report numbers for a loop that does
// not match production behavior.
try crossCheck(gpa, cfg.seed, p, first.rounds, first.digest);
return .{
.build = median(totals.build[0..reps]),
.frame = median(totals.frame[0..reps]),
.server = median(totals.server[0..reps]),
.decode = median(totals.decode[0..reps]),
.respond = median(totals.respond[0..reps]),
.reframe = median(totals.reframe[0..reps]),
.compress = median(totals.compress[0..reps]),
.unframe = median(totals.unframe[0..reps]),
.consume = median(totals.consume[0..reps]),
.store = median(totals.store[0..reps]),
.persist = median(totals.persist[0..reps]),
.fsync = median(totals.fsync[0..reps]),
.reopen = median(totals.reopen[0..reps]),
.counters = base.?,
.rounds = first.rounds,
.last = last,
};
}
// ---------------------------------------------------------------------------
// Report — a human phase table + counter block + a machine-parseable line.
// ---------------------------------------------------------------------------
fn report(out: *Io.Writer, cfg: Config, size: []const u8, p: Preset, r: *SizeResult) RunError!void {
// wall = every phase; consume already contains store_put + parse+recon, so
// those sub-lines are not re-summed.
const wall = r.build + r.frame + r.server + r.unframe + r.consume + r.persist + r.fsync + r.reopen;
const store_t = if (r.store > r.consume) r.consume else r.store;
const parse_recon = r.consume - store_t;
try out.print(
"\n# sync bench: in-process clone, seed={d}, reps={d} (medians) reproduce with seed={d}\n",
.{ cfg.seed, cfg.reps, cfg.seed },
);
try out.print("size={s} artifacts={d} blob={d} delta_frac={d}\n", .{ size, p.artifacts, p.blob, p.delta_frac });
try out.writeAll("phase total_ms %wall p50_us p99_us\n");
// Server attribution. `server = decode + respond + reframe` (measured); the
// two `=` lines partition it into the clone-speed decision's real question —
// walk+compress (≈ the standalone cfile-compression micro) vs decode+parse
// (the igot flood the clone re-sends and ignores), the remainder. respond
// bundles parse with walk+compress (scanRequest/emitCloneReply are private),
// so the split is derived from `compress_micro`, not timed inside respond.
const compress_est = if (r.compress > r.server) r.server else r.compress;
const decode_parse = r.server - compress_est;
try phaseLine(out, "build", r.build, wall, r.last.build.items);
try phaseLine(out, "frame", r.frame, wall, r.last.frame.items);
try phaseLine(out, "server", r.server, wall, r.last.server.items);
try phaseLine(out, " decode", r.decode, wall, r.last.decode.items);
try phaseLine(out, " respond", r.respond, wall, r.last.respond.items);
try phaseLine(out, " reframe", r.reframe, wall, r.last.reframe.items);
try subLine(out, " =decode+parse", decode_parse, wall); // server - compress_est
try subLine(out, " =walk+compress", compress_est, wall); // ≈ compress_micro
try subLine(out, " compress_micro", r.compress, wall); // standalone, not in wall
try phaseLine(out, "unframe", r.unframe, wall, r.last.unframe.items);
try phaseLine(out, "consume", r.consume, wall, r.last.consume.items);
try phaseLine(out, " store_put", store_t, wall, r.last.store.items);
try subLine(out, " parse+recon", parse_recon, wall); // derived = consume - store_put
try subLine(out, "persist", r.persist, wall);
try out.print("fsync {d:>10.3} {d:>5.1}% (n={d})\n", .{ nsToMs(r.fsync), pct(r.fsync, wall), r.counters.fsync_count });
try subLine(out, "reopen", r.reopen, wall);
try out.print("wall {d:>10.3} 100.0%\n", .{nsToMs(wall)});
try reportCounters(out, r.counters);
try reportMachine(out, size, cfg.seed, r, wall, store_t, parse_recon);
}
/// A per-round phase line: median total, %wall, and p50/p99 over the last rep's
/// per-round samples.
fn phaseLine(out: *Io.Writer, name: []const u8, total: u64, wall: u64, samples: []u64) RunError!void {
std.sort.pdq(u64, samples, {}, std.sort.asc(u64));
const p50 = percentile(samples, 50);
const p99 = percentile(samples, 99);
try out.print("{s:<16} {d:>10.3} {d:>5.1}% {d:>10.2} {d:>10.2}\n", .{ name, nsToMs(total), pct(total, wall), nsToUs(p50), nsToUs(p99) });
}
/// A once-per-rep (or derived) phase line: no per-round percentile column.
fn subLine(out: *Io.Writer, name: []const u8, total: u64, wall: u64) RunError!void {
try out.print("{s:<16} {d:>10.3} {d:>5.1}%\n", .{ name, nsToMs(total), pct(total, wall) });
}
fn reportCounters(out: *Io.Writer, c: Counters) RunError!void {
const amp = if (c.content_bytes == 0) 0.0 else @as(f64, @floatFromInt(c.disk_bytes)) / @as(f64, @floatFromInt(c.content_bytes));
try out.print(
"rounds={d} req_bytes={d} reply_bytes={d} req_wire={d} reply_wire={d} igot={d} gimme={d} file={d} cfile_full={d} cfile_delta={d} artifacts={d} phantom_peak={d} store_puts={d} fsync_count={d} disk_bytes={d} content_bytes={d} amplification={d:.2}x\n",
.{
c.rounds, c.req_plaintext, c.reply_plaintext, c.req_wire,
c.reply_wire, c.req_igot, c.req_gimme, c.reply_file,
c.reply_cfile_full, c.reply_cfile_delta, c.artifacts_received, c.phantom_peak,
c.store_put_calls, c.fsync_count, c.disk_bytes, c.content_bytes,
amp,
},
);
}
fn reportMachine(out: *Io.Writer, size: []const u8, seed: u64, r: *SizeResult, wall: u64, store_t: u64, parse_recon: u64) RunError!void {
const c = r.counters;
// Split across two prints: Zig's format caps a single call at 32 arguments.
// The line is one logical `#m` record (no newline until the second print).
try out.print(
"#m sync size={s} seed={d} rounds={d} build_ns={d} frame_ns={d} server_ns={d} decode_ns={d} respond_ns={d} reframe_ns={d} compress_ns={d} unframe_ns={d} consume_ns={d} store_put_ns={d} parse_recon_ns={d} persist_ns={d} fsync_ns={d} reopen_ns={d} wall_ns={d}",
.{ size, seed, r.rounds, r.build, r.frame, r.server, r.decode, r.respond, r.reframe, r.compress, r.unframe, r.consume, store_t, parse_recon, r.persist, r.fsync, r.reopen, wall },
);
try out.print(
" req_bytes={d} reply_bytes={d} req_wire={d} reply_wire={d} req_igot={d} req_gimme={d} req_clone={d} reply_cfile_full={d} reply_cfile_delta={d} reply_file={d} reply_igot={d} reply_push={d} reply_clone_seqno={d} artifacts={d} phantom_peak={d} store_puts={d} fsync_count={d} disk_bytes={d} content_bytes={d}\n",
.{
c.req_plaintext, c.reply_plaintext, c.req_wire, c.reply_wire,
c.req_igot, c.req_gimme, c.req_clone, c.reply_cfile_full,
c.reply_cfile_delta, c.reply_file, c.reply_igot, c.reply_push,
c.reply_clone_seqno, c.artifacts_received, c.phantom_peak, c.store_put_calls,
c.fsync_count, c.disk_bytes, c.content_bytes,
},
);
// Belt-and-braces anti-DCE for ReleaseFast: the accumulators feed output,
// but pin them so the optimizer cannot elide the timed work.
std.mem.doNotOptimizeAway(wall);
std.mem.doNotOptimizeAway(c.artifacts_received);
}
// ---------------------------------------------------------------------------
// Temp backing file (real file, reached only through injected Io).
// ---------------------------------------------------------------------------
/// A throwaway /tmp file named from the monotonic clock, unlinked on cleanup.
const Temp = struct {
file: Io.File,
path_buf: [64]u8,
path_len: usize,
fn create(io: Io) Error!Temp {
var t: Temp = undefined;
const nonce = Io.Clock.Timestamp.now(io, .awake).raw.nanoseconds;
const p = std.fmt.bufPrint(&t.path_buf, "/tmp/zit-syncbench-{d}.log", .{nonce}) catch return error.Io;
t.path_len = p.len;
t.file = Io.Dir.createFileAbsolute(io, t.path_buf[0..t.path_len], .{ .read = true }) catch return error.Io;
return t;
}
fn path(self: *const Temp) []const u8 {
return self.path_buf[0..self.path_len];
}
fn cleanup(self: *Temp, io: Io) void {
self.file.close(io);
Io.Dir.deleteFileAbsolute(io, self.path()) catch {};
}
};
// ---------------------------------------------------------------------------
// Timing / statistics helpers.
// ---------------------------------------------------------------------------
fn now(io: Io) Io.Timestamp {
return Io.Timestamp.now(io, .awake);
}
fn elapsed(t0: Io.Timestamp, t1: Io.Timestamp) u64 {
const ns = t0.durationTo(t1).nanoseconds; // i96, monotonic so >= 0
return if (ns < 0) 0 else @intCast(ns);
}
fn sum(xs: []const u64) u64 {
var total: u64 = 0;
for (xs) |x| total += x;
return total;
}
/// Median of per-rep totals; sorts `xs` in place (order is no longer needed).
fn median(xs: []u64) u64 {
std.debug.assert(xs.len >= 1);
std.sort.pdq(u64, xs, {}, std.sort.asc(u64));
return xs[xs.len / 2];
}
/// p-th percentile of an already-sorted slice; empty -> 0 (a zero-round clone
/// is impossible, but the report must not divide by an empty sample set).
fn percentile(sorted: []const u64, p: u64) u64 {
std.debug.assert(p <= 100);
if (sorted.len == 0) return 0;
var idx: usize = @intCast(p * sorted.len / 100);
if (idx >= sorted.len) idx = sorted.len - 1;
return sorted[idx];
}
fn pct(part: u64, whole: u64) f64 {
if (whole == 0) return 0;
return @as(f64, @floatFromInt(part)) * 100.0 / @as(f64, @floatFromInt(whole));
}
fn nsToMs(ns: u64) f64 {
return @as(f64, @floatFromInt(ns)) / 1_000_000.0;
}
fn nsToUs(ns: u64) f64 {
return @as(f64, @floatFromInt(ns)) / 1000.0;
}
// ---------------------------------------------------------------------------
// Error mapping — every seam's closed set folds explicitly into Error.
// ---------------------------------------------------------------------------
fn mapBuild(e: sync.Client.BuildError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.RequestTooLarge => error.RequestTooLarge,
};
}
fn mapFrame(e: transport.TransportError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.RequestTooLarge => error.RequestTooLarge,
else => error.RequestMalformed, // frame only emits the two above
};
}
fn mapUnframe(e: transport.TransportError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.ReplyTooLarge => error.ResponseTooLarge,
error.ReplyTruncated, error.ReplyMalformed => error.Corrupt,
else => error.Corrupt,
};
}
fn mapXfer(e: serve.XferError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.RequestTooLarge => error.RequestTooLarge,
error.RequestMalformed => error.RequestMalformed,
error.ResponseTooLarge => error.ResponseTooLarge,
};
}
fn mapConsume(e: sync.Client.ConsumeError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.FatalProtocolError => error.FatalProtocolError,
error.NotImplemented => error.NotImplemented,
};
}
/// The decomposed server phase's inbound decode/respond/encode fold into the
/// same `Error` variants `mapXfer` produces, so the reconstruction and the real
/// `handleXfer` surface identical causes.
fn mapDecode(e: sync.ContainerDecodeError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.TruncatedHeader, error.InvalidStream => error.RequestMalformed,
error.OutputTooLarge => error.RequestTooLarge,
};
}
fn mapRespond(e: sync.Server.RespondError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.ResponseTooLarge => error.ResponseTooLarge,
};
}
fn mapEncode(e: sync.ContainerEncodeError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.InputTooLarge => error.ResponseTooLarge,
};
}
fn mapRun(e: sync_driver.RunError(*ServerTransport)) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.RequestTooLarge => error.RequestTooLarge,
error.FatalProtocolError => error.FatalProtocolError,
error.NotImplemented => error.NotImplemented,
error.RoundLimitExceeded => error.RoundLimitExceeded,
error.ResponseTooLarge => error.ResponseTooLarge,
};
}
fn mapOpen(e: logstore.OpenError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.Storage => error.Storage,
error.Corrupt => error.Corrupt,
};
}
fn mapPut(e: logstore.PutError) Error {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.Storage => error.Storage,
};
}
// ===========================================================================
// Tests — a benchmark's only oracle is correctness + determinism, never timing.
// Debug, wired into `zig build test`; each uses the injected `std.testing.io`
// for the clock/temp file and asserts convergence + counter determinism. No
// multi-rep timing/percentile machinery runs under test.
// ===========================================================================
const testing = std.testing;
const test_preset = Preset{ .artifacts = 32, .blob = 256, .delta_frac = 60 };
test "sync bench corpus: one seed reproduces an identical store digest; distinct seeds differ; delta_frac honored" {
const gpa = testing.allocator;
var a = repostate.MemStore.init(gpa);
defer a.deinit();
var b = repostate.MemStore.init(gpa);
defer b.deinit();
try generateCorpus(gpa, &a, 42, test_preset);
try generateCorpus(gpa, &b, 42, test_preset);
try testing.expectEqual(@as(usize, 32), a.store().count());
const da = try storeDigest(gpa, a.store());
const db = try storeDigest(gpa, b.store());
try testing.expectEqualSlices(u8, &da, &db);
// A different seed reaches the corpus (distinct content).
var c = repostate.MemStore.init(gpa);
defer c.deinit();
try generateCorpus(gpa, &c, 43, test_preset);
const dc = try storeDigest(gpa, c.store());
try testing.expect(!std.mem.eql(u8, &da, &dc));
// Bounds are enforced.
var d = repostate.MemStore.init(gpa);
defer d.deinit();
try testing.expectError(error.BadArgs, generateCorpus(gpa, &d, 1, .{ .artifacts = 1, .blob = 4, .delta_frac = 0 }));
try testing.expectError(error.BadArgs, generateCorpus(gpa, &d, 1, .{ .artifacts = 0, .blob = 64, .delta_frac = 0 }));
try testing.expectError(error.BadArgs, generateCorpus(gpa, &d, 1, .{ .artifacts = 1, .blob = 64, .delta_frac = 101 }));
}
test "sync bench: the in-process clone converges to the server's content, no phantom remains" {
const gpa = testing.allocator;
const io = testing.io;
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, 7, test_preset);
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var ts = TimingStore{ .inner = client_ms.store(), .io = io };
var rs: RoundSamples = .{};
defer rs.deinit(gpa);
var c = Counters{};
const rounds = try runClone(gpa, io, server_ms.store(), &ts, &rs, &c);
try testing.expect(rounds >= 1);
// Phase timings are meaningless if the clone is wrong; this is the license
// to print anything: the client store equals the server store, phantom-free.
const cd = try storeDigest(gpa, ts.inner);
const sd = try storeDigest(gpa, server_ms.store());
try testing.expectEqualSlices(u8, &sd, &cd);
try testing.expectEqual(@as(u64, 0), countPhantoms(ts.inner));
try testing.expectEqual(server_ms.store().count(), client_ms.store().count());
}
test "sync bench: the control cross-check agrees with sync_driver.runSync" {
const gpa = testing.allocator;
const io = testing.io;
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, 11, test_preset);
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var ts = TimingStore{ .inner = client_ms.store(), .io = io };
var rs: RoundSamples = .{};
defer rs.deinit(gpa);
var c = Counters{};
const rounds = try runClone(gpa, io, server_ms.store(), &ts, &rs, &c);
const digest = try storeDigest(gpa, ts.inner);
// Same seed, same preset: the real runSync must agree on rounds AND digest.
try crossCheck(gpa, 11, test_preset, rounds, digest);
// A perturbed round count must be caught (the guard actually fires).
try testing.expectError(error.ControlDivergence, crossCheck(gpa, 11, test_preset, rounds + 1, digest));
}
test "sync bench: counters are self-consistent and seed-deterministic across runs" {
const gpa = testing.allocator;
const io = testing.io;
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, 5, test_preset);
var first: Counters = undefined;
var run_i: usize = 0;
while (run_i < 2) : (run_i += 1) {
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var ts = TimingStore{ .inner = client_ms.store(), .io = io };
var rs: RoundSamples = .{};
defer rs.deinit(gpa);
var c = Counters{};
_ = try runClone(gpa, io, server_ms.store(), &ts, &rs, &c);
_ = try persistPhase(gpa, io, ts.inner, &c);
c.store_put_calls = ts.put_calls;
// Self-consistency: every received content card is a cfile or file, and
// the total equals the server's artifact count (a full clone).
try testing.expectEqual(c.reply_cfile_full + c.reply_cfile_delta + c.reply_file, c.artifacts_received);
try testing.expectEqual(@as(u64, 32), c.artifacts_received);
if (run_i == 0) first = c else try testing.expect(std.meta.eql(first, c));
}
}
test "sync bench: the persisted clone round-trips every artifact by hash" {
const gpa = testing.allocator;
const io = testing.io;
var server_ms = repostate.MemStore.init(gpa);
defer server_ms.deinit();
try generateCorpus(gpa, &server_ms, 3, test_preset);
var client_ms = repostate.MemStore.init(gpa);
defer client_ms.deinit();
var ts = TimingStore{ .inner = client_ms.store(), .io = io };
var rs: RoundSamples = .{};
defer rs.deinit(gpa);
var c = Counters{};
_ = try runClone(gpa, io, server_ms.store(), &ts, &rs, &c);
// Persist into a fresh log, then reopen and verify each artifact's content
// hashes back to its name (the logstore round-trip the persist phase times).
var temp = try Temp.create(io);
defer temp.cleanup(io);
var fs = storage.FileStorage.init(io, temp.file, 0);
var ls = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| return mapOpen(e);
const store = ts.inner;
const n = store.count();
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
const bytes = store.content(rid).?;
_ = ls.put(&rawHash(bytes), bytes) catch |e| return mapPut(e);
}
try ls.commit();
const total = ls.size();
ls.deinit(gpa);
var fs2 = storage.FileStorage.init(io, temp.file, total);
var ls2 = logstore.LogStore.open(gpa, fs2.storageFile(), 0) catch |e| return mapOpen(e);
defer ls2.deinit(gpa);
rid = 1;
while (rid <= n) : (rid += 1) {
const got = (ls2.content(gpa, rid) catch |e| return mapOpen(mapContentToOpen(e))) orelse return error.Corrupt;
defer gpa.free(got);
const h = hexHash(got);
try testing.expectEqualSlices(u8, store.hashOf(rid), &h);
}
}
/// ContentError -> OpenError shim for the round-trip test's single call site.
fn mapContentToOpen(e: logstore.ContentError) logstore.OpenError {
return switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.Storage => error.Storage,
error.Corrupt => error.Corrupt,
};
}
test "sync bench corpus: OutOfMemory is handled leak-free at every allocation point" {
const Ctx = struct {
fn gen(gpa: std.mem.Allocator) !void {
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
generateCorpus(gpa, &ms, 99, .{ .artifacts = 16, .blob = 64, .delta_frac = 50 }) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => unreachable, // only OOM is reachable on a valid preset
};
}
};
try testing.checkAllAllocationFailures(testing.allocator, Ctx.gen, .{});
}