Artifact f4b18c21053a0cece333d08d5230854f6348a56364de29c086d27732d018f675:
- File
src/sync_benchmark.zig
— part of check-in
[1fd84b4327]
at
2026-08-16 02:37:51
on branch trunk
— sync: per-round credentials over the cookie channel (#74)
An authenticated exchange now carries a fresh credential on every round: body login until the peer's gate is known, then an x-f-l-c cookie whose nonce covers that round's own body. The transport attaches it; serve now verifies it (and refuses tampered, replayed, and duplicate logins) instead of ignoring the header its own announced version solicits. (user: dmestas, size: 68122) [annotate] [blame] [check-ins using]
//! 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 `container.decode` of a per-round igot flood the clone re-sends //! and then ignores, the `respondWithMedia` parse+store-walk, and the per-cfile //! `container.encode` 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), so parse and the store walk are NOT separated and //! `respond` is reported as one measured phase. The compression share alone is //! attributed, by a standalone `container.encode` micro-measurement over the //! corpus — faithful because a clone emits each artifact as exactly one //! full-content cfile via the same `writeCfile`→`container.encode(.default)` — //! and reported as a share of the `respond` phase, not as a partition of it. The //! parse+walk remainder is left unpartitioned rather than mislabeled. //! //! 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, 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; /// The runaway backstop, mirroring `runSync`: consecutive rounds that moved no /// bytes, NOT total rounds. A clone pages the repository in /// `clone_batch_max`-sized cfile batches, so its round count scales with the /// corpus — which is exactly why a total-round cap cannot express this bound /// without becoming a limit on corpus size instead. /// /// It was `1 << 20`, sized as a total-round budget. Reinterpreted as a stall /// budget that number is meaningless rather than merely generous: a bench whose /// peer stalls is broken, and a million stalled rounds is not a backstop. Sized /// here the way `max_stalled_sync_rounds` is, with headroom over §3.7's longest /// by-design stall. const max_bench_rounds: usize = 32; /// §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 prev_hash: [64]u8 = undefined; 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); // Model the production ingest policy (logstore.putArtifact): a chained // artifact is delta-encoded against its predecessor when the delta is // materially smaller (< 90% of the full artifact), and the store then // serves that delta cfile verbatim on clone (via putWireDelta). Every // other artifact is held as its full §4.1 container (via putWire). All // compression/delta cost is paid HERE, at ingest, off the timed serve // path — exactly the design. var stored_delta = false; if (chained) { if (zit.delta.create(gpa, prev, cur)) |raw| { defer gpa.free(raw); // Same profitability ratio as the store: raw.len * 10 < blob * 9. if (raw.len * 10 < blob * 9) { const dc = zit.container.encode(gpa, raw, .default) catch return error.OutOfMemory; defer gpa.free(dc); // The reconstructed target IS `cur` (the delta was created // from it), so the §7.2 USIZE is exactly `cur.len` — no need // to re-parse the header we just wrote. _ = ms.putWireDelta(&hash, dc, &prev_hash, cur.len) catch return error.OutOfMemory; stored_delta = true; } } else |e| switch (e) { error.OutOfMemory => return error.OutOfMemory, error.TargetTooLarge => {}, // never at bench sizes; falls through to full } } if (!stored_delta) { const container = zit.container.encode(gpa, cur, .default) catch return error.OutOfMemory; defer gpa.free(container); _ = ms.putWire(&hash, container) catch return error.OutOfMemory; } @memcpy(prev, cur); @memcpy(&prev_hash, &hash); } 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, retain_ns: u64 = 0, put_calls: u64 = 0, content_calls: u64 = 0, retain_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 + self.retain_ns; } const vtable = repostate.ArtifactStore.VTable{ .intern = internImpl, .put = putImpl, .content = contentImpl, .hashOf = hashOfImpl, .lookup = lookupImpl, .count = countImpl, .wire = wireImpl, .retain = retainImpl, .held = heldImpl, }; fn wireImpl(ctx: *anyopaque, rid: repostate.Rid) ?repostate.WireFrame { const self: *TimingStore = @ptrCast(@alignCast(ctx)); return self.inner.wire(rid); } fn heldImpl(ctx: *anyopaque, rid: repostate.Rid) bool { const self: *TimingStore = @ptrCast(@alignCast(ctx)); return self.inner.held(rid); } fn retainImpl(ctx: *anyopaque, rid: repostate.Rid, ingest: repostate.WireIngest) error{OutOfMemory}!void { const self: *TimingStore = @ptrCast(@alignCast(ctx)); const t0 = now(self.io); try self.inner.retain(rid, ingest); self.retain_ns += elapsed(t0, now(self.io)); self.retain_calls += 1; } 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 container.decode 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; // Mirrors `runSync`'s guard, which bounds CONSECUTIVE byte-free rounds // rather than total rounds — a total cap would bound the transfer itself. var stalled: usize = 0; while (client.state == .exchange or client.state == .clone_round0) { if (stalled == 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, null) 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; if (sync.roundDeliveredContent(obs)) stalled = 0 else stalled += 1; 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 /// `container.decode` (the growing igot-flood the clone re-announces every /// round), `respond` is `respondWithMedia` (parse + store walk + per-cfile /// compression), `reframe` is the outer framing (`container.encode` 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 = zit.container.decode(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, null, &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) (zit.container.encode(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 cfile-compression cost, measured standalone: time one /// `container.encode(.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 is a faithful measure /// of the respond phase's compression work. It is the one server sub-cost /// attributable at the pub boundary: `scanRequest`/`emitCloneReply` are private, /// so the parse and store walk that `respondWithMedia` also does stay bundled in /// respond and are not separated. 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 = zit.container.encode(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 content_bytes += bytes.len; // §7.7 — the LogStore is addressed by the RAW digest (<=32 bytes), not // the 64-hex wire NAME the sync store keys on; `rawHash(bytes)` IS the // §7.7 recompute over the reconstructed bytes and the log key at once. const digest = rawHash(bytes); // D4 §7.2/§4.1 — persist the wire form the client retained during // consume verbatim, so a re-serve (a clone-of-clone) retransmits the // compact delta byte-for-byte. Fall back to compressing the plaintext // when nothing was retained or a delta's basis is not in the log (§3.3). // Mirrors main.zig's persistStore, adapted to the rawHash keying. if (client.wire(rid)) |wf| { if (wf.delta) |d| { var stored = false; if (client.lookup(d.src_hash)) |base_sync_rid| { if (client.content(base_sync_rid)) |base_bytes| { const base_key = rawHash(base_bytes); if (ls.lookup(&base_key)) |base_log_rid| { if (!ls.isPhantom(base_log_rid) and ls.depthOf(base_log_rid) < logstore.max_delta_chain) { _ = ls.putDeltaContainer(&digest, base_log_rid, wf.container) catch |e| return mapPut(e); stored = true; } } } } if (!stored) _ = ls.putZlib(&digest, bytes) catch |e| return mapPut(e); } else { _ = ls.putContainer(&digest, wf.container) catch |e| return mapPut(e); } } else { _ = ls.putZlib(&digest, bytes) catch |e| return mapPut(e); } } 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 { /// §6.5 (#74) — the driver hands every round's cookie to the transport; /// this double runs sans-HTTP, so the credential has no header to ride. pub fn setLoginCookie(self: *@This(), value: ?[]const u8) void { _ = self; _ = value; } 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) — a faithful standalone measure of the respond phase's // compression work (parse and walk stay bundled in respond, not attributed). 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 phase = decode + respond + reframe, all measured. `respond` bundles // parse + store walk + per-cfile compression; `scanRequest`/`emitCloneReply` // are file-private, so parse and walk are NOT separable at the pub boundary // and no parse-vs-walk (or igot-vs-compression) split is claimed. The one // honestly attributable server sub-cost is compression: a standalone // per-artifact `container.encode` micro over the corpus, reported below as // a share of its enclosing `respond` phase, with the parse+walk remainder // left unpartitioned rather than mislabeled. 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); // Compression reported honestly: faithful (a clone emits each artifact as one // full-content cfile through the same writeCfile->container.encode) and as // a share of respond, not summed into wall. try out.print( " compression {d:>10.3} {d:>5.1}% of respond (standalone per-artifact micro, faithful)\n", .{ nsToMs(r.compress), pct(r.compress, r.respond) }, ); try out.writeAll(" note: a reliable measure of compression cost; the remaining respond time (parse+walk) is not separable without core instrumentation.\n"); 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, }; } /// 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: zit.container.DecodeError) 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: zit.container.EncodeError) 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.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, .{}); }