a73x

ac90c478

refactor: one header spelling, one discontinuity guard

a73x   2026-09-01 13:39

Commit message
refactor: one header spelling, one discontinuity guard

protocol.encodeHeader is the single place the 5-byte frame header is
built, and the drift test becomes a layout pin against golden bytes
on both the encode and decode side. DeltaTracker.continuous states
the resync conditions once for update and noteBlind. proxy's writeAll
clone and the hub's one-frame-per-event drain get comments pinning
them as deliberate.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
src/client/webhub.zig
Old New
@@ -680,14 +680,20 @@ pub fn pumpTile(
680 }; 680 };
681 defer frame.deinit(alloc); 681 defer frame.deinit(alloc);
682 dial.onFrame(frame.type); 682 dial.onFrame(frame.type);
683 var hdr: [proto.frame_header_len]u8 = undefined; 683 const hdr = proto.encodeHeader(frame.type, frame.payload.len);
684 hdr[0] = @intFromEnum(frame.type);
685 std.mem.writeInt(u32, hdr[1..5], @intCast(frame.payload.len), .little);
686 var vecs = [_][]const u8{ &.{env_frame}, &hdr, frame.payload }; 684 var vecs = [_][]const u8{ &.{env_frame}, &hdr, frame.payload };
687 ws.writeMessageVec(&vecs, .binary) catch return; 685 ws.writeMessageVec(&vecs, .binary) catch return;
688 // Only the socket link carries the guarantee that one 686 // Only the socket link carries the guarantee that one
689 // readable event is one frame; QUIC may have buffered 687 // readable event is one frame; QUIC may have buffered
690 // more, so drain until .incomplete. 688 // more, so drain until .incomplete.
689 //
690 // The socket arm breaks after one frame where wall_pump.zig
691 // re-polls at a zero timeout to drain the whole burst. That
692 // divergence is correct: the CLI judges KEYS between frames
693 // and misreads a wheel notch it sees before the `term_modes`
694 // trailing it, so it must not stop mid-burst. The hub judges
695 // nothing — it re-frames onto the WebSocket in order — and a
696 // burst still pending makes the next poll return immediately.
691 if (transport.link != .quic) break :frames; 697 if (transport.link != .quic) break :frames;
692 } 698 }
693 } 699 }
src/engine/delta.zig
Old New
@@ -94,17 +94,29 @@ pub const DeltaTracker = struct {
94 advanced, 94 advanced,
95 }; 95 };
96 96
97 /// Whether the tracked state can still be described as a delta against
98 /// `eng`, or whether the client has to be resynced from scratch.
99 ///
100 /// Geometry is load-bearing, not decorative: row_hashes is indexed by the
101 /// tracker's own row count, and dumpVtRow asserts against the engine's. A
102 /// tracker left stale by a failed rebuild resyncs here instead of running
103 /// off the end of the grid. A tracker that was never built has no rows to
104 /// diff at all, and an alt-screen flip replaces the whole grid, so no row
105 /// seq from before it means anything.
106 ///
107 /// Both `update` and `noteBlind` advance the tracker, so both must ask
108 /// this before they touch a row seq.
109 fn continuous(self: *const DeltaTracker, eng: *Engine) bool {
110 if (self.rows == 0) return false;
111 if (self.rows != eng.term.rows or self.cols != eng.term.cols) return false;
112 return eng.onAltScreen() == self.on_alt;
113 }
114
97 /// Diff current engine state against the tracked state. Advances seq 115 /// Diff current engine state against the tracked state. Advances seq
98 /// and tracked rows when anything changed. Allocates only the per-row 116 /// and tracked rows when anything changed. Allocates only the per-row
99 /// dumps it hashes. 117 /// dumps it hashes.
100 pub fn update(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine) !Update { 118 pub fn update(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine) !Update {
101 if (self.rows == 0) return .discontinuity; // never built 119 if (!self.continuous(eng)) return .discontinuity;
102 // Geometry is load-bearing, not decorative: row_hashes is indexed
103 // by the tracker's own row count, and dumpVtRow asserts against the
104 // engine's. A tracker left stale by a failed rebuild resyncs here
105 // instead of running off the end of the grid.
106 if (self.rows != eng.term.rows or self.cols != eng.term.cols) return .discontinuity;
107 if (eng.onAltScreen() != self.on_alt) return .discontinuity;
108 120
109 // Safe to stamp row_seqs with the seq we may not end up taking: it 121 // Safe to stamp row_seqs with the seq we may not end up taking: it
110 // is only written for rows whose hash changed, and any changed row 122 // is only written for rows whose hash changed, and any changed row
@@ -144,9 +156,7 @@ pub const DeltaTracker = struct {
144 /// those bytes go nowhere. Takes no allocator — a signature that can 156 /// those bytes go nowhere. Takes no allocator — a signature that can
145 /// allocate is one that can render. 157 /// allocate is one that can render.
146 pub fn noteBlind(self: *DeltaTracker, eng: *Engine) Update { 158 pub fn noteBlind(self: *DeltaTracker, eng: *Engine) Update {
147 if (self.rows == 0) return .discontinuity; 159 if (!self.continuous(eng)) return .discontinuity;
148 if (self.rows != eng.term.rows or self.cols != eng.term.cols) return .discontinuity;
149 if (eng.onAltScreen() != self.on_alt) return .discontinuity;
150 160
151 self.seq += 1; 161 self.seq += 1;
152 for (self.row_seqs) |*row_seq| row_seq.* = self.seq; 162 for (self.row_seqs) |*row_seq| row_seq.* = self.seq;
src/engine/protocol.zig
Old New
@@ -54,12 +54,22 @@ pub const MsgType = enum(u8) {
54 54
55 pub const max_payload = 16 * 1024 * 1024; 55 pub const max_payload = 16 * 1024 * 1024;
56 56
57 /// One type byte + u32 LE payload length. The writers below spell the 5 57 /// One type byte + u32 LE payload length.
58 /// inline in their fixed-size buffers; this name exists for READERS that
59 /// delimit frames out of a buffer they did not fill (the daemon's
60 /// pushInbound, the hub's WebSocket messages).
61 pub const frame_header_len = 5; 58 pub const frame_header_len = 5;
62 59
60 /// The one place the header layout is spelled. Every writer — blocking,
61 /// bounded, queued, and the hub's re-frame onto a WebSocket — builds its
62 /// header here, so the layout cannot drift between one path and another;
63 /// `delimitFrame` and `readFrame` are the decode side of these same bytes.
64 /// Returns by value because a 5-byte array is cheaper to copy than to
65 /// borrow, and the caller wants it beside a payload slice anyway.
66 pub fn encodeHeader(t: MsgType, len: usize) [frame_header_len]u8 {
67 var hdr: [frame_header_len]u8 = undefined;
68 hdr[0] = @intFromEnum(t);
69 std.mem.writeInt(u32, hdr[1..5], @intCast(len), .little);
70 return hdr;
71 }
72
63 pub const Frame = struct { 73 pub const Frame = struct {
64 type: MsgType, 74 type: MsgType,
65 payload: []u8, 75 payload: []u8,
@@ -112,9 +122,7 @@ pub fn takeFrame(alloc: std.mem.Allocator, buf: *std.ArrayList(u8)) !?Frame {
112 } 122 }
113 123
114 pub fn writeFrame(fd: std.posix.fd_t, t: MsgType, payload: []const u8) !void { 124 pub fn writeFrame(fd: std.posix.fd_t, t: MsgType, payload: []const u8) !void {
115 var hdr: [5]u8 = undefined; 125 const hdr = encodeHeader(t, payload.len);
116 hdr[0] = @intFromEnum(t);
117 std.mem.writeInt(u32, hdr[1..5], @intCast(payload.len), .little);
118 try writeAllFd(fd, &hdr); 126 try writeAllFd(fd, &hdr);
119 try writeAllFd(fd, payload); 127 try writeAllFd(fd, payload);
120 } 128 }
@@ -129,9 +137,7 @@ pub const reply_budget_ms: i32 = 250;
129 /// caller, while a merely slow one still gets every byte. WouldBlock 137 /// caller, while a merely slow one still gets every byte. WouldBlock
130 /// leaves a TRUNCATED frame — hence every caller drops the connection. 138 /// leaves a TRUNCATED frame — hence every caller drops the connection.
131 pub fn writeFrameBounded(fd: std.posix.fd_t, t: MsgType, payload: []const u8, budget_ms: i32) !void { 139 pub fn writeFrameBounded(fd: std.posix.fd_t, t: MsgType, payload: []const u8, budget_ms: i32) !void {
132 var hdr: [5]u8 = undefined; 140 const hdr = encodeHeader(t, payload.len);
133 hdr[0] = @intFromEnum(t);
134 std.mem.writeInt(u32, hdr[1..5], @intCast(payload.len), .little);
135 var left = budget_ms; 141 var left = budget_ms;
136 try writeAllFdBounded(fd, &hdr, &left); 142 try writeAllFdBounded(fd, &hdr, &left);
137 try writeAllFdBounded(fd, payload, &left); 143 try writeAllFdBounded(fd, payload, &left);
@@ -164,9 +170,7 @@ pub fn appendFrame(
164 t: MsgType, 170 t: MsgType,
165 payload: []const u8, 171 payload: []const u8,
166 ) !void { 172 ) !void {
167 var hdr: [5]u8 = undefined; 173 const hdr = encodeHeader(t, payload.len);
168 hdr[0] = @intFromEnum(t);
169 std.mem.writeInt(u32, hdr[1..5], @intCast(payload.len), .little);
170 try list.appendSlice(alloc, &hdr); 174 try list.appendSlice(alloc, &hdr);
171 try list.appendSlice(alloc, payload); 175 try list.appendSlice(alloc, payload);
172 } 176 }
@@ -1164,19 +1168,28 @@ pub fn composeDelta(alloc: std.mem.Allocator, payload: []const u8) !ComposedDelt
1164 return .{ .header = hdr, .bytes = try out.toOwnedSlice(alloc) }; 1168 return .{ .header = hdr, .bytes = try out.toOwnedSlice(alloc) };
1165 } 1169 }
1166 1170
1167 test "appendFrame encodes the same bytes writeFrame sends" { 1171 test "the frame header layout is these bytes, and the decoders read them back" {
1168 const alloc = std.testing.allocator; 1172 const alloc = std.testing.allocator;
1169 const golden = [_]u8{ 0x02, 3, 0, 0, 0, 'a', 'b', 'c' }; 1173 const golden = [_]u8{ 0x02, 3, 0, 0, 0, 'a', 'b', 'c' };
1170 1174
1175 // Every writer builds its header in `encodeHeader`, so they cannot
1176 // disagree with each other. What is still worth pinning is the LAYOUT:
1177 // the golden bytes are the wire contract, and a peer running an older
1178 // build is decoding them with its own copy of this rule. So the encode
1179 // side is checked against literal bytes, and the decode side is checked
1180 // against the same literal bytes rather than against a round trip — a
1181 // round trip through readFrame passes even if encode and decode drift
1182 // together, and says nothing about what actually goes on the wire.
1171 var list: std.ArrayList(u8) = .empty; 1183 var list: std.ArrayList(u8) = .empty;
1172 defer list.deinit(alloc); 1184 defer list.deinit(alloc);
1173 try appendFrame(&list, alloc, .input, "abc"); 1185 try appendFrame(&list, alloc, .input, "abc");
1174 try std.testing.expectEqualSlices(u8, &golden, list.items); 1186 try std.testing.expectEqualSlices(u8, &golden, list.items);
1175 1187
1176 // writeFrame is driven for real rather than round-tripped: a round trip 1188 const d = (try delimitFrame(&golden)).?;
1177 // through readFrame passes without pinning a single byte, so it can say 1189 try std.testing.expectEqual(MsgType.input, d.type);
1178 // nothing about whether writeFrame and appendFrame agree — which is the 1190 try std.testing.expectEqualSlices(u8, "abc", d.payload);
1179 // drift this test names. 1191 try std.testing.expectEqual(golden.len, d.consumed);
1192
1180 const p = try std.posix.pipe(); 1193 const p = try std.posix.pipe();
1181 defer std.posix.close(p[0]); 1194 defer std.posix.close(p[0]);
1182 try writeFrame(p[1], .input, "abc"); 1195 try writeFrame(p[1], .input, "abc");
src/proxy.zig
Old New
@@ -61,6 +61,9 @@ pub fn pump(in_fd: std.posix.fd_t, out_fd: std.posix.fd_t, sock_path: []const u8
61 } 61 }
62 } 62 }
63 63
64 /// Byte-identical to `protocol.writeAllFd`, and deliberately not shared: this
65 /// file's contract is that it imports nothing of the wire, and `protocol` is
66 /// the wire. The duplicate is the price of that ban, not an oversight.
64 fn writeAll(fd: std.posix.fd_t, data: []const u8) !void { 67 fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
65 var i: usize = 0; 68 var i: usize = 0;
66 while (i < data.len) i += try std.posix.write(fd, data[i..]); 69 while (i < data.len) i += try std.posix.write(fd, data[i..]);