a73x

118384a1

refactor: one counted send for both frame kinds and both fan-outs

a73x   2026-08-29 10:01

Commit message
refactor: one counted send for both frame kinds and both fan-outs

sendDeltaTo, broadcastDelta, snapshotTo and resyncSnapshot each spelled
their own queueFrame + counter pair + counterfactual accrual. countedSend
is the one owner: `to` names a single client or null every client of the
session, and the frame kind picks the counter pair, so a fifth send path
cannot invent a fifth accounting. Three doc blocks shrank with the bodies
they document, server.zig stays at budget 0.

Pinned by server_test_deliver's counter assertions and statsText's field
order (the bench reads it).
src/server/server.zig
Old New
@@ -2687,34 +2687,36 @@ pub const Server = struct {
2687 } else |_| {} 2687 } else |_| {}
2688 } 2688 }
2689 2689
2690 /// Send one delta to one client (the attach path, which serves a 2690 /// The only place a send is counted, so the four paths cannot drift on
2691 /// have_seq only that client holds). 2691 /// what a byte means. `to` is one client, null every client of THIS
2692 fn sendDeltaTo(self: *Server, si: usize, i: usize, payload: []const u8) void { 2692 /// session — where a failed write drops only that client. Counted bytes
2693 if (!self.queueFrame(i, .delta, payload)) return; 2693 /// are payload actually written, the 5-byte header excluded here and in
2694 self.stats.deltas += 1; 2694 /// the counterfactual (~8% understated). The counterfactual accrues once
2695 // Payload bytes only: the 5-byte frame header is excluded here and 2695 /// per EVENT and only if someone got it, as the bench measures it.
2696 // in the counterfactual, so the ratio stays honest even though it 2696 fn countedSend(self: *Server, si: usize, t: proto.MsgType, payload: []const u8, to: ?usize) void {
2697 // understates delta wire cost by ~8% at a typical delta size.
2698 self.stats.delta_bytes += payload.len;
2699 self.accrueSnapshotEquiv(si);
2700 }
2701
2702 /// The live stream's exit: one delta to every client of THIS session.
2703 /// A write failure drops only that client — the rest of the broadcast
2704 /// still goes out.
2705 fn broadcastDelta(self: *Server, si: usize, payload: []const u8) void {
2706 var sent = false; 2697 var sent = false;
2707 for (0..max_clients) |i| { 2698 for (0..max_clients) |i| {
2708 if (!self.inSession(i, si)) continue; 2699 if (if (to) |only| i != only else !self.inSession(i, si)) continue;
2709 if (!self.queueFrame(i, .delta, payload)) continue; 2700 if (!self.queueFrame(i, t, payload)) continue;
2710 sent = true; 2701 sent = true;
2711 // Per-send actuals: delta_bytes is bytes we really wrote. 2702 if (t == .delta) {
2712 self.stats.deltas += 1; 2703 self.stats.deltas += 1;
2713 self.stats.delta_bytes += payload.len; 2704 self.stats.delta_bytes += payload.len;
2705 } else {
2706 self.stats.snapshots += 1;
2707 self.stats.snapshot_bytes += payload.len;
2708 }
2714 } 2709 }
2715 // Per-event counterfactual, and only if the update reached someone 2710 if (!sent) return;
2716 // — matching the single-client behaviour the bench measures. 2711 if (t == .delta) self.accrueSnapshotEquiv(si) else self.stats.snapshot_equiv_bytes += payload.len;
2717 if (sent) self.accrueSnapshotEquiv(si); 2712 }
2713
2714 fn sendDeltaTo(self: *Server, si: usize, i: usize, payload: []const u8) void {
2715 self.countedSend(si, .delta, payload, i);
2716 }
2717
2718 fn broadcastDelta(self: *Server, si: usize, payload: []const u8) void {
2719 self.countedSend(si, .delta, payload, null);
2718 } 2720 }
2719 2721
2720 /// Per-update path: diff and broadcast a delta; discontinuities resync. 2722 /// Per-update path: diff and broadcast a delta; discontinuities resync.
@@ -3046,49 +3048,28 @@ pub const Server = struct {
3046 return true; 3048 return true;
3047 } 3049 }
3048 3050
3049 /// Every client of THIS session gets it: these events change the grid 3051 /// Every client of THIS session gets it, and the rebuild happens with
3050 /// under everyone watching it. The rebuild happens even with nobody 3052 /// nobody attached too, so the tracker stays usable. No term_modes,
3051 /// attached, so the tracker stays usable for the next attach. 3053 /// unlike sendResync: reaching here for a client that has never been
3052 /// 3054 /// through an attach breaks that, silently.
3053 /// Carries no term_modes, unlike sendResync: no event reaching here
3054 /// changes a mode, and every attached client has already been told the
3055 /// current value — sampleTermModes broadcasts on change, and sendResync
3056 /// runs on BOTH attach paths. A caller that reaches here for a client
3057 /// which has never been through an attach breaks that, silently.
3058 pub fn resyncSnapshot(self: *Server, si: usize) void { 3055 pub fn resyncSnapshot(self: *Server, si: usize) void {
3059 if (!self.rebuildTracker(si)) return; 3056 if (!self.rebuildTracker(si)) return;
3060 if (!self.hasClientsIn(si)) return; 3057 if (!self.hasClientsIn(si)) return;
3061 const payload = self.buildSnapshotPayload(si) catch return; 3058 const payload = self.buildSnapshotPayload(si) catch return;
3062 defer self.alloc.free(payload); 3059 defer self.alloc.free(payload);
3063 3060
3064 var sent = false; 3061 self.countedSend(si, .snapshot, payload, null);
3065 for (0..max_clients) |i| {
3066 if (!self.inSession(i, si)) continue;
3067 if (!self.queueFrame(i, .snapshot, payload)) continue;
3068 sent = true;
3069 self.stats.snapshots += 1;
3070 self.stats.snapshot_bytes += payload.len;
3071 }
3072 // Per-event counterfactual: this snapshot IS what a snapshot-only
3073 // daemon would have sent, so it counts once however many clients
3074 // received it.
3075 if (sent) self.stats.snapshot_equiv_bytes += payload.len;
3076 } 3062 }
3077 3063
3078 /// A join at the current size is a discontinuity for the joiner alone: 3064 /// A join at the current size is a discontinuity for the joiner alone,
3079 /// everyone else is already current, so sending them a full repaint would 3065 /// so repainting anyone else would be waste. The rebuild still bumps
3080 /// be pure waste. The rebuild still bumps seq for all of them, which they 3066 /// seq for them all; a client never checks seq for contiguity.
3081 /// absorb silently — a client reads the seq it is given and never checks
3082 /// it for contiguity, so the next delta simply carries a higher number.
3083 fn snapshotTo(self: *Server, si: usize, i: usize) void { 3067 fn snapshotTo(self: *Server, si: usize, i: usize) void {
3084 if (!self.rebuildTracker(si)) return; 3068 if (!self.rebuildTracker(si)) return;
3085 if (self.clients[i] == null) return; 3069 if (self.clients[i] == null) return;
3086 const payload = self.buildSnapshotPayload(si) catch return; 3070 const payload = self.buildSnapshotPayload(si) catch return;
3087 defer self.alloc.free(payload); 3071 defer self.alloc.free(payload);
3088 if (!self.queueFrame(i, .snapshot, payload)) return; 3072 self.countedSend(si, .snapshot, payload, i);
3089 self.stats.snapshots += 1;
3090 self.stats.snapshot_bytes += payload.len;
3091 self.stats.snapshot_equiv_bytes += payload.len;
3092 } 3073 }
3093 3074
3094 /// Attach/reattach for the client in slot `i`. Three cases: 3075 /// Attach/reattach for the client in slot `i`. Three cases: