Artifact f0d99a1f8273a1675a2f482a4977bbb12fc07a432447905c991bd359db73acdf:
- File
src/sync_benchmark.zig
— part of check-in
[e17845e40d]
at
2026-07-25 13:37:43
on branch trunk
— serve: read the served store through a log-offset index, not a resident copy
Priming a serve snapshot copied every artifact's container out of the log into an in-memory store and held them there for the server's lifetime. The log had already been scanned and every frame CRC-checked to build the rid table, so the copy was a second full read whose only product was resident bytes: on a 254k-artifact repository it was two thirds of the prime time and a few hundred megabytes of RSS, all to re-hold what the log already holds.
The snapshot is now a log-backed store: an index of hash -> log location, and container bytes read from the log on demand as a round emits them, memoised only for the round and released after it. Persisted order is preserved so the clone pager still cursors correctly, the binary-digest to hex-wire-name conversion stays at its one site, and a read fault after the scan surfaces as a server storage error so a clone is never silently truncated. The old whole-store copy is retained only where it is actually needed: the pull client's holdings and the equivalence test's oracle.
Measured, 67499-artifact repository, ReleaseSafe: a clone off the indexed serve is byte-identical to a clone off the copy-based serve (both materialise all 983 files of trunk identically), while serve peak RSS falls from 157 MB to 32 MB. The one remaining O(store) cost, the startup scan itself, is a separate change. (user: dmestas, size: 67243) [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, 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 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; 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 /// `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, &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 { 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, 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: 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.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, .{}); }