a73x

1aba332b

feat: webhub tile pump — envelope, frame messages, backoff shared with the CLI

a73x   2026-08-13 10:33

Commit message
feat: webhub tile pump — envelope, frame messages, backoff shared with the CLI

The hub↔browser wire (M-web Task 6): 0x00 + frame verbatim both ways,
0x01 + the four-word JSON control vocabulary hub→browser. Browser
messages parse the 5-byte header and never the payload; the length must
match the message exactly (WebSocket already delimits, so a trailing
byte is upstream's bug). MsgType being non-exhaustive BY DESIGN means a
future type byte parses and transits untouched — the test pins 0x40
passing through rather than dying, which is the proxy thesis holding
for vocabulary the protocol has not learned yet.

pumpTile: one thread per tile, blocking by design (the Transport is
thread-private, so readFrame's blocking read is correct here); dials
with abort_fd=-1; narrates reconnecting/up around re-dials on the CLI's
own schedule — nextBackoffMs extracted from reconnect() as a pure fn
the CLI now also calls, no-retry-cap contract included, pinned by test.
protocol.zig gains frame_header_len for readers that delimit frames out
of buffers they did not fill.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -213,6 +213,7 @@ pub fn build(b: *std.Build) void {
213 .target = target, 213 .target = target,
214 .optimize = optimize, 214 .optimize = optimize,
215 }); 215 });
216 webhub_mod.addImport("protocol", protocol_mod);
216 217
217 // The socket path's identity and the right to bind it: the stale-socket 218 // The socket path's identity and the right to bind it: the stale-socket
218 // claim and the dev+ino record teardown compares against. A leaf — it 219 // claim and the dev+ino record teardown compares against. A leaf — it
@@ -301,6 +302,10 @@ pub fn build(b: *std.Build) void {
301 client_mod.addImport("proxy", proxy_mod); 302 client_mod.addImport("proxy", proxy_mod);
302 client_mod.addImport("paint", paint_mod); 303 client_mod.addImport("paint", paint_mod);
303 304
305 // Late-bound: webhub is defined before client in this file, but the
306 // import graph only needs both to exist by the time the exes compile.
307 webhub_mod.addImport("client", client_mod);
308
304 const mux_mod = b.createModule(.{ 309 const mux_mod = b.createModule(.{
305 .root_source_file = b.path("src/mux_main.zig"), 310 .root_source_file = b.path("src/mux_main.zig"),
306 .target = target, 311 .target = target,
@@ -493,7 +498,7 @@ pub fn build(b: *std.Build) void {
493 // found no libraries and no explanation. 498 // found no libraries and no explanation.
494 if (mod == server_mod or mod == quic_mod or mod == quic_server_mod or 499 if (mod == server_mod or mod == quic_mod or mod == quic_server_mod or
495 mod == exe_mod or mod == client_mod or mod == mux_mod or 500 mod == exe_mod or mod == client_mod or mod == mux_mod or
496 mod == quic_client_mod) linkQuic(b, t, quic); 501 mod == quic_client_mod or mod == webhub_mod) linkQuic(b, t, quic);
497 test_step.dependOn(&b.addRunArtifact(t).step); 502 test_step.dependOn(&b.addRunArtifact(t).step);
498 } 503 }
499 504
src/client.zig
Old New
@@ -1600,7 +1600,7 @@ fn reconnect(
1600 var backoff_ms: u64 = 0; 1600 var backoff_ms: u64 = 0;
1601 while (true) { 1601 while (true) {
1602 if (drainStdinForQuit(stdin_fd, backoff_ms)) return false; 1602 if (drainStdinForQuit(stdin_fd, backoff_ms)) return false;
1603 backoff_ms = if (backoff_ms == 0) 200 else @min(backoff_ms * 2, 2000); 1603 backoff_ms = nextBackoffMs(backoff_ms);
1604 1604
1605 var fresh = Transport.open(alloc, quiet, null, std.posix.STDIN_FILENO) catch |err| { 1605 var fresh = Transport.open(alloc, quiet, null, std.posix.STDIN_FILENO) catch |err| {
1606 // Ctrl-\ during the handshake is the same answer as Ctrl-\ 1606 // Ctrl-\ during the handshake is the same answer as Ctrl-\
@@ -1628,6 +1628,24 @@ fn reconnect(
1628 } 1628 }
1629 } 1629 }
1630 1630
1631 /// The reconnect pacing, M7's numbers: iteration zero waits not at all
1632 /// (see reconnect's comment for the measurement), then 200ms doubling to
1633 /// a 2s cap — pacing a flapping link without ever giving up. Pure so the
1634 /// muxweb hub runs the SAME schedule against its tiles; the deliberate
1635 /// no-retry-cap is part of the contract.
1636 pub fn nextBackoffMs(prev: u64) u64 {
1637 return if (prev == 0) 200 else @min(prev * 2, 2000);
1638 }
1639
1640 test "reconnect backoff: 0 then 200 doubling to the 2s cap, never beyond" {
1641 try std.testing.expectEqual(@as(u64, 200), nextBackoffMs(0));
1642 try std.testing.expectEqual(@as(u64, 400), nextBackoffMs(200));
1643 try std.testing.expectEqual(@as(u64, 800), nextBackoffMs(400));
1644 try std.testing.expectEqual(@as(u64, 1600), nextBackoffMs(800));
1645 try std.testing.expectEqual(@as(u64, 2000), nextBackoffMs(1600));
1646 try std.testing.expectEqual(@as(u64, 2000), nextBackoffMs(2000));
1647 }
1648
1631 fn devNull() !std.posix.fd_t { 1649 fn devNull() !std.posix.fd_t {
1632 return std.posix.open("/dev/null", .{ .ACCMODE = .WRONLY }, 0); 1650 return std.posix.open("/dev/null", .{ .ACCMODE = .WRONLY }, 0);
1633 } 1651 }
src/protocol.zig
Old New
@@ -36,6 +36,12 @@ pub const MsgType = enum(u8) {
36 36
37 pub const max_payload = 16 * 1024 * 1024; 37 pub const max_payload = 16 * 1024 * 1024;
38 38
39 /// One type byte + u32 LE payload length. The writers below spell the 5
40 /// inline in their fixed-size buffers; this name exists for READERS that
41 /// delimit frames out of a buffer they did not fill (the daemon's
42 /// pushInbound, the hub's WebSocket messages).
43 pub const frame_header_len = 5;
44
39 pub const Frame = struct { 45 pub const Frame = struct {
40 type: MsgType, 46 type: MsgType,
41 payload: []u8, 47 payload: []u8,
src/webhub.zig
Old New
@@ -10,6 +10,8 @@
10 //! authenticated by ssh like everything else in this project. 10 //! authenticated by ssh like everything else in this project.
11 11
12 const std = @import("std"); 12 const std = @import("std");
13 const proto = @import("protocol");
14 const client = @import("client");
13 15
14 pub const default_port: u16 = 7681; 16 pub const default_port: u16 = 7681;
15 17
@@ -77,6 +79,172 @@ pub fn route(assets: Assets, path: []const u8) ?Asset {
77 } 79 }
78 80
79 // --------------------------------------------------------------------------- 81 // ---------------------------------------------------------------------------
82 // The wire between hub and browser (M-web Task 6): WebSocket binary
83 // messages, one envelope byte. 0x00 + mux protocol frame verbatim, both
84 // directions; 0x01 + UTF-8 JSON control message, hub→browser only. That
85 // is the whole vocabulary — anything the mux protocol learns to say
86 // later transits untouched.
87
88 pub const env_frame: u8 = 0x00;
89 pub const env_control: u8 = 0x01;
90
91 pub const TileState = enum { connecting, up, reconnecting, gone };
92
93 /// The full control MESSAGE — envelope byte included — ready for one
94 /// writeMessage. Comptime because the vocabulary is closed.
95 pub fn controlMessage(comptime s: TileState) []const u8 {
96 return &[1]u8{env_control} ++ "{\"state\":\"" ++ @tagName(s) ++ "\"}";
97 }
98
99 pub const ParsedFrame = struct { t: proto.MsgType, payload: []const u8 };
100
101 pub const FrameMsgError = error{
102 BadEnvelope,
103 ShortFrame,
104 LengthMismatch,
105 Oversize,
106 };
107
108 /// One browser→hub WebSocket message → one mux frame. The hub parses the
109 /// 5-byte header (it must — the WS leg is message-delimited while the
110 /// daemon leg is a byte stream) and NEVER the payload; the spec's
111 /// amendment 2 is this function's contract. The length must match the
112 /// message exactly: WebSocket already delimits, so a trailing byte is a
113 /// bug upstream, not framing to resynchronize.
114 pub fn parseFrameMessage(msg: []const u8) FrameMsgError!ParsedFrame {
115 if (msg.len < 1 or msg[0] != env_frame) return error.BadEnvelope;
116 const f = msg[1..];
117 if (f.len < proto.frame_header_len) return error.ShortFrame;
118 const len = std.mem.readInt(u32, f[1..5], .little);
119 if (len > proto.max_payload) return error.Oversize;
120 if (f.len - proto.frame_header_len != len) return error.LengthMismatch;
121 // MsgType is non-exhaustive BY DESIGN, so an unknown type byte is a
122 // valid value that transits untouched — the proxy thesis holding for
123 // vocabulary the protocol has not learned yet.
124 return .{ .t = @enumFromInt(f[0]), .payload = f[proto.frame_header_len..] };
125 }
126
127 /// One thread per tile, and blocking is the design: the tile's Transport
128 /// is private to this thread, so Transport.readFrame's blocking read
129 /// (fatal to a multiplexing hub) is simply correct here. A slow browser
130 /// stalls only its own tile (v1 stance: the daemon side is protected by
131 /// its own 8 MiB pending cap; the hub's upstream reads just stall).
132 ///
133 /// The hub owns reconnection; the browser owns re-attach. On transport
134 /// death this narrates `reconnecting`, re-dials on the CLI's own backoff
135 /// schedule (client.nextBackoffMs — no retry cap, deliberately), then
136 /// narrates `up`; the browser's replica quotes have_seq/have_epoch in a
137 /// fresh attach and M7's snapshot-vs-delta resolution does the rest.
138 pub fn pumpTile(
139 alloc: std.mem.Allocator,
140 ws: *std.http.Server.WebSocket,
141 ws_fd: std.posix.fd_t,
142 target_in: client.Target,
143 ) void {
144 var target = target_in;
145 // No terminal to spam and a control channel that already narrates:
146 // the one fallback line the CLI allows itself is quieted here, the
147 // same way reconnect() quiets retries.
148 if (target == .hand) target.hand.report_fallback = false;
149
150 ws.writeMessage(@constCast(controlMessage(.connecting)), .binary) catch return;
151 var transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
152 defer transport.close();
153 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return;
154
155 while (true) {
156 var fds = [_]std.posix.pollfd{
157 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
158 .{ .fd = ws_fd, .events = std.posix.POLL.IN, .revents = 0 },
159 };
160 _ = std.posix.poll(&fds, transport.timeoutMs(100)) catch return;
161 transport.service();
162
163 // Daemon → browser. The `.quic` disjunct is the CLI's lesson
164 // verbatim: there frames can arrive from the stream layer with
165 // the socket never going readable.
166 if (fds[0].revents != 0 or transport.link == .quic) frames: {
167 while (true) {
168 const incoming = transport.readFrame(alloc) catch return;
169 const frame = switch (incoming) {
170 .frame => |f| f,
171 .incomplete => break :frames,
172 .closed => {
173 ws.writeMessage(@constCast(controlMessage(.reconnecting)), .binary) catch return;
174 transport.close();
175 transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
176 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return;
177 break :frames;
178 },
179 };
180 defer frame.deinit(alloc);
181 var hdr: [proto.frame_header_len]u8 = undefined;
182 hdr[0] = @intFromEnum(frame.type);
183 std.mem.writeInt(u32, hdr[1..5], @intCast(frame.payload.len), .little);
184 var vecs = [_][]const u8{ &.{env_frame}, &hdr, frame.payload };
185 ws.writeMessageVec(&vecs, .binary) catch return;
186 // Only the socket link carries the guarantee that one
187 // readable event is one frame; QUIC may have buffered
188 // more, so drain until .incomplete.
189 if (transport.link != .quic) break :frames;
190 }
191 }
192
193 // Browser → daemon. One WS message is one frame; drain what the
194 // reader has already buffered too, or it waits for a poll the
195 // socket will never signal.
196 if (fds[1].revents != 0) {
197 while (true) {
198 const msg = ws.readSmallMessage() catch return;
199 if (msg.opcode == .binary or msg.opcode == .text) {
200 if (parseFrameMessage(msg.data)) |parsed| {
201 transport.writeFrame(parsed.t, parsed.payload) catch {
202 ws.writeMessage(@constCast(controlMessage(.reconnecting)), .binary) catch return;
203 transport.close();
204 transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
205 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return;
206 };
207 } else |_| {
208 // Not a frame message: dropped, deliberately —
209 // the browser side is ours, so this is a bug's
210 // signature, and killing every tile for it would
211 // make the page unusable exactly when debugging.
212 }
213 }
214 if (ws.input.bufferedLen() == 0) break;
215 }
216 }
217 }
218 }
219
220 /// Dial with the CLI's backoff schedule until the transport opens or the
221 /// browser hangs up (null). While waiting out a backoff the WS is
222 /// watched: a message that arrives with no transport to carry it is
223 /// dropped (the browser re-attaches on `up` anyway), a dead WS ends the
224 /// tile.
225 fn dialLoop(
226 alloc: std.mem.Allocator,
227 target: client.Target,
228 ws: *std.http.Server.WebSocket,
229 ws_fd: std.posix.fd_t,
230 ) ?client.Transport {
231 var backoff_ms: u64 = 0;
232 while (true) {
233 if (client.Transport.open(alloc, target, null, -1)) |t| {
234 return t;
235 } else |_| {}
236 backoff_ms = client.nextBackoffMs(backoff_ms);
237 // The backoff doubles as the WS liveness window.
238 var fds = [_]std.posix.pollfd{
239 .{ .fd = ws_fd, .events = std.posix.POLL.IN, .revents = 0 },
240 };
241 const n = std.posix.poll(&fds, @intCast(backoff_ms)) catch return null;
242 if (n > 0 and fds[0].revents != 0) {
243 const msg = ws.readSmallMessage() catch return null;
244 _ = msg; // dropped: nothing to carry it yet
245 }
246 }
247 }
80 248
81 test "origin: exactly our two spellings pass, everything else refuses" { 249 test "origin: exactly our two spellings pass, everything else refuses" {
82 const cases = [_]struct { origin: ?[]const u8, port: u16, want: bool }{ 250 const cases = [_]struct { origin: ?[]const u8, port: u16, want: bool }{
@@ -134,3 +302,41 @@ test "routes: the three assets with their content types, 404 for the rest" {
134 try std.testing.expectEqual(@as(?Asset, null), route(assets, "/ws/0")); 302 try std.testing.expectEqual(@as(?Asset, null), route(assets, "/ws/0"));
135 try std.testing.expectEqual(@as(?Asset, null), route(assets, "/../src/main.zig")); 303 try std.testing.expectEqual(@as(?Asset, null), route(assets, "/../src/main.zig"));
136 } 304 }
305
306 test "control messages: the closed vocabulary, envelope included" {
307 try std.testing.expectEqualStrings("\x01{\"state\":\"connecting\"}", controlMessage(.connecting));
308 try std.testing.expectEqualStrings("\x01{\"state\":\"up\"}", controlMessage(.up));
309 try std.testing.expectEqualStrings("\x01{\"state\":\"reconnecting\"}", controlMessage(.reconnecting));
310 try std.testing.expectEqualStrings("\x01{\"state\":\"gone\"}", controlMessage(.gone));
311 }
312
313 test "frame messages: exact framing in, everything else named" {
314 // Types spelled via the enum so the test cannot drift from the wire.
315 const input_byte: u8 = @intFromEnum(proto.MsgType.input);
316 const good = [_]u8{ 0x00, input_byte, 2, 0, 0, 0, 'h', 'i' };
317 const parsed = try parseFrameMessage(&good);
318 try std.testing.expectEqual(proto.MsgType.input, parsed.t);
319 try std.testing.expectEqualStrings("hi", parsed.payload);
320
321 // Empty payload is legal (detach sends one).
322 const detach_byte: u8 = @intFromEnum(proto.MsgType.detach);
323 const empty = [_]u8{ 0x00, detach_byte, 0, 0, 0, 0 };
324 try std.testing.expectEqual(proto.MsgType.detach, (try parseFrameMessage(&empty)).t);
325
326 // Each refusal by name.
327 try std.testing.expectError(error.BadEnvelope, parseFrameMessage(&[_]u8{}));
328 try std.testing.expectError(error.BadEnvelope, parseFrameMessage(&[_]u8{ 0x01, 'x' }));
329 try std.testing.expectError(error.ShortFrame, parseFrameMessage(&[_]u8{ 0x00, input_byte, 1, 0 }));
330 // Length says 3, message carries 2.
331 try std.testing.expectError(error.LengthMismatch, parseFrameMessage(&[_]u8{ 0x00, input_byte, 3, 0, 0, 0, 'h', 'i' }));
332 // A trailing byte is upstream's bug, not framing to resync.
333 try std.testing.expectError(error.LengthMismatch, parseFrameMessage(&[_]u8{ 0x00, input_byte, 1, 0, 0, 0, 'h', 'i' }));
334 // Length field claims more than max_payload.
335 var oversize = [_]u8{ 0x00, input_byte, 0, 0, 0, 0 };
336 std.mem.writeInt(u32, oversize[2..6], proto.max_payload + 1, .little);
337 try std.testing.expectError(error.Oversize, parseFrameMessage(&oversize));
338 // MsgType is non-exhaustive by design: a type byte from a future
339 // protocol PARSES and transits untouched rather than dying here.
340 const future = try parseFrameMessage(&[_]u8{ 0x00, 0x40, 0, 0, 0, 0 });
341 try std.testing.expectEqual(@as(u8, 0x40), @intFromEnum(future.t));
342 }