a73x

9273c18c

fix(hub): never call readSmallMessage speculatively; dead-leg ping; one owner per recipe

a73x   2026-08-13 12:20

Commit message
fix(hub): never call readSmallMessage speculatively; dead-leg ping; one owner per recipe

The hub's browser leg had two ways to park a tile thread with the daemon
leg unattended, and both came from the same mistake: readSmallMessage
BLOCKS — it fills from the socket whenever the frame it is decoding runs
past what the reader holds — and the pump called it on nothing more than
a poll wakeup.

  - webhub.headFrame decides, from the reader's buffered bytes alone,
    what the next readSmallMessage will do: incomplete (wait, and poll
    WILL fire again, because a short frame is one whose rest is in
    flight), a complete pong (tossed here — readSmallMessage swallows
    pongs and then blocks on whatever is behind them), ready, or a frame
    too big to ever be buffered whole. Every read in the pump and in
    dialLoop is gated on it. Unit-tested at every split boundary: 1-byte
    header, mid-extended-length, mid-mask, mid-payload, and the
    whole-frame-vs-payload capacity edge.
  - poll reports what the KERNEL holds and headFrame reads what the
    READER holds, so a readable event is turned into buffered bytes with
    one fillMore first. Without it the gate deadlocks on the very first
    attach — caught by the e2e hub scenarios, which is what they are for.
  - A re-dial now `continue`s the outer loop instead of falling through
    to fds[1].revents, which described a socket state from before the
    dial and a message dialLoop may already have eaten.

The tile thread also had no way to notice a browser that vanished with a
half-open socket — a slept laptop, a killed browser, a dropped ssh -L —
and it held a daemon client slot the whole time. Silence is now measured
on the wall clock (the poll stays capped at 100ms because the QUIC timer
needs the tick): a ping at 30s and 60s, and the tile ends at 90s.
Incoming pings are answered with a pong, as RFC 6455 asks and as a
browser heartbeat needs.

Also here, all from the same review round:

  - handoff.recipeFor and xdg.resolveKeyPath are now the single owners of
    the bare-HOST ssh line and the quic:// key rule; `mux` and `muxweb`
    both call them rather than each spelling ~30 lines. A drift between
    the two spellings would have pointed the binaries at different remote
    commands and different keys.
  - The handoff cache write is behind a process-wide mutex: openHandoff
    stopped being single-threaded when the hub started running one per
    tile, and two tiles naming the same host share a cache path.
  - web/verify.js runs as part of `make test`, gated on node's presence.
    Both its path and the wasm are FILE args, not strings — as a string
    the script is not an input the build graph hashes, so a doctored
    verify.js stayed cached and the check was decorative. Confirmed by
    doctoring one: it now fails the test step.
  - build.zig.zon .paths gains "web" (webhub_main @embedFiles it).
  - The dead @constCast around controlMessage: writeMessage takes a
    const slice.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -501,6 +501,8 @@ pub fn build(b: *std.Build) void {
501 webhub_main_mod.addImport("client", client_mod); 501 webhub_main_mod.addImport("client", client_mod);
502 webhub_main_mod.addImport("webhub", webhub_mod); 502 webhub_main_mod.addImport("webhub", webhub_mod);
503 webhub_main_mod.addImport("xdg", xdg_mod); 503 webhub_main_mod.addImport("xdg", xdg_mod);
504 // The HOST tile's recipe comes from the same owner mux_main uses.
505 webhub_main_mod.addImport("handoff", handoff_mod);
504 webhub_main_mod.addImport("build_options", version_opts.createModule()); 506 webhub_main_mod.addImport("build_options", version_opts.createModule());
505 webhub_main_mod.addAnonymousImport("index.html", .{ .root_source_file = b.path("web/index.html") }); 507 webhub_main_mod.addAnonymousImport("index.html", .{ .root_source_file = b.path("web/index.html") });
506 webhub_main_mod.addAnonymousImport("mux.js", .{ .root_source_file = b.path("web/mux.js") }); 508 webhub_main_mod.addAnonymousImport("mux.js", .{ .root_source_file = b.path("web/mux.js") });
@@ -540,6 +542,32 @@ pub fn build(b: *std.Build) void {
540 test_step.dependOn(&b.addRunArtifact(t).step); 542 test_step.dependOn(&b.addRunArtifact(t).step);
541 } 543 }
542 544
545 // web/verify.js drives mux_core.wasm through the page's real call
546 // sequence, and it was the only check of the wasm ABI that `make test`
547 // did not run — so an export the JS shell depends on could be renamed,
548 // or its return contract changed, with every Zig test still green.
549 // Gated on node rather than required: the wasm core builds and the Zig
550 // suite passes without it, and a missing node must not turn `make test`
551 // red on a machine that never opens a browser.
552 if (b.findProgram(&.{"node"}, &.{})) |node| {
553 const verify = b.addSystemCommand(&.{node});
554 // Both as FILE args, not strings: a string argument is not an input
555 // the build graph hashes, so a doctored verify.js would have stayed
556 // cached and the check would have been decorative. (Caught by
557 // doctoring one, exactly as the fix round asked.)
558 verify.addFileArg(b.path("web/verify.js"));
559 verify.addFileArg(wasm_exe.getEmittedBin());
560 verify.setName("verify wasm ABI (web/verify.js)");
561 // stdio is the assertion: a failing check exits non-zero.
562 verify.expectExitCode(0);
563 test_step.dependOn(&verify.step);
564 } else |_| {
565 std.debug.print(
566 "build: node not found — skipping web/verify.js (the wasm ABI check)\n",
567 .{},
568 );
569 }
570
543 const e2e = b.addSystemCommand(&.{"test/e2e.sh"}); 571 const e2e = b.addSystemCommand(&.{"test/e2e.sh"});
544 e2e.addArtifactArg(exe); 572 e2e.addArtifactArg(exe);
545 e2e.addArtifactArg(mux_exe); 573 e2e.addArtifactArg(mux_exe);
build.zig.zon
Old New
@@ -7,6 +7,9 @@
7 "build.zig", 7 "build.zig",
8 "build.zig.zon", 8 "build.zig.zon",
9 "src", 9 "src",
10 // The page muxweb serves: index.html and mux.js are @embedFile'd
11 // by webhub_main.zig, so a package without them does not build.
12 "web",
10 }, 13 },
11 .dependencies = .{ 14 .dependencies = .{
12 .ghostty = .{ 15 .ghostty = .{
src/client.zig
Old New
@@ -271,6 +271,10 @@ pub const Transport = struct {
271 } 271 }
272 } 272 }
273 273
274 /// Process-wide, guarding the one handoff-cache write below. See there
275 /// for why one file can now have two writers.
276 var cache_write_mu: std.Thread.Mutex = .{};
277
274 /// The handoff, in transport terms. Warm: cached coordinates, dial 278 /// The handoff, in transport terms. Warm: cached coordinates, dial
275 /// direct, no ssh process at all. Cold: one ssh child whose first stdout 279 /// direct, no ssh process at all. Cold: one ssh child whose first stdout
276 /// line is the mandatory announce; QUIC success kills it, QUIC failure 280 /// line is the mandatory announce; QUIC success kills it, QUIC failure
@@ -319,7 +323,18 @@ pub const Transport = struct {
319 }; 323 };
320 // A cache write that fails costs a cold attach next time and nothing 324 // A cache write that fails costs a cold attach next time and nothing
321 // else, so it is not worth a line of the user's attention. 325 // else, so it is not worth a line of the user's attention.
322 if (h.cache_path) |cp| handoff.writeCache(cp, ep) catch {}; 326 //
327 // Serialized because openHandoff is no longer single-threaded: the
328 // web hub runs one of these per tile, and two tiles naming the same
329 // host share a cache path. writeCache is write-then-rename, so the
330 // worst interleaving loses a write rather than tearing a file — but
331 // "the loser wrote the STALER endpoint" is a cold attach that looks
332 // like a bug, and a mutex is cheaper than the afternoon.
333 if (h.cache_path) |cp| {
334 cache_write_mu.lock();
335 defer cache_write_mu.unlock();
336 handoff.writeCache(cp, ep) catch {};
337 }
323 338
324 if (openQuicEndpoint(alloc, h, ep, carry, abort_fd)) |t| { 339 if (openQuicEndpoint(alloc, h, ep, carry, abort_fd)) |t| {
325 // QUIC carries the session now; the coordination ssh is done. 340 // QUIC carries the session now; the coordination ssh is done.
src/handoff.zig
Old New
@@ -166,6 +166,32 @@ pub fn dialHost(host: []const u8) []const u8 {
166 return host[at + 1 ..]; 166 return host[at + 1 ..];
167 } 167 }
168 168
169 /// The two allocations a bare-HOST target needs before it can be dialed:
170 /// the coordination command and the per-host cache path.
171 pub const Recipe = struct {
172 ssh_cmd: []const u8,
173 /// null means attach UNCACHED — an uncacheable host (a separator in
174 /// the name) or no resolvable cache directory. Always cold, never
175 /// wrong; the rule lives here rather than at each call site.
176 cache_path: ?[]const u8,
177
178 pub fn deinit(self: Recipe, alloc: std.mem.Allocator) void {
179 alloc.free(self.ssh_cmd);
180 if (self.cache_path) |c| alloc.free(c);
181 }
182 };
183
184 /// ONE owner for the handoff recipe: `mux HOST` and a `muxweb` HOST tile
185 /// build the identical thing, and a drift between two spellings of the
186 /// ssh line would quietly point the two binaries at different remote
187 /// commands. Building it here (rather than in client.zig) is what keeps
188 /// the client free of XDG and of allocating a command line.
189 pub fn recipeFor(alloc: std.mem.Allocator, host: []const u8) !Recipe {
190 const cmd = try std.fmt.allocPrint(alloc, "ssh {s} muxd endpoint", .{host});
191 errdefer alloc.free(cmd);
192 return .{ .ssh_cmd = cmd, .cache_path = xdg.hostCachePath(alloc, host) catch null };
193 }
194
169 /// One newline-terminated line from `fd`, returned WITHOUT the newline. 195 /// One newline-terminated line from `fd`, returned WITHOUT the newline.
170 /// 196 ///
171 /// Read a byte at a time, deliberately. The frame stream begins at the 197 /// Read a byte at a time, deliberately. The frame stream begins at the
src/mux_main.zig
Old New
@@ -8,6 +8,7 @@ const client = @import("client");
8 const build_options = @import("build_options"); 8 const build_options = @import("build_options");
9 const xdg = @import("xdg"); 9 const xdg = @import("xdg");
10 const spawn = @import("spawn"); 10 const spawn = @import("spawn");
11 const handoff = @import("handoff");
11 12
12 const usage = 13 const usage =
13 \\usage: mux [HOST | --sock PATH | --via CMD | quic://HOST[:PORT]] 14 \\usage: mux [HOST | --sock PATH | --via CMD | quic://HOST[:PORT]]
@@ -154,19 +155,20 @@ pub fn main() !u8 {
154 return 2; 155 return 2;
155 }, 156 },
156 .quic => |q| { 157 .quic => |q| {
157 var key_owned: ?[]const u8 = null; 158 const res = try xdg.resolveKeyPath(alloc, q.key);
158 defer if (key_owned) |p| alloc.free(p); 159 defer switch (res) {
159 const key_path = q.key orelse blk: { 160 .given => {},
160 const p = try xdg.keyPath(alloc); 161 .default, .missing => |p| alloc.free(p),
161 key_owned = p; 162 };
162 std.fs.cwd().access(p, .{}) catch { 163 const key_path = switch (res) {
164 .given, .default => |p| p,
165 .missing => |p| {
163 std.debug.print( 166 std.debug.print(
164 "mux: no key: pass --key, set MUX_KEY_FILE, or run `muxd keygen` (default {s})\n", 167 "mux: no key: pass --key, set MUX_KEY_FILE, or run `muxd keygen` (default {s})\n",
165 .{p}, 168 .{p},
166 ); 169 );
167 return 2; 170 return 2;
168 }; 171 },
169 break :blk p;
170 }; 172 };
171 return client.attach(alloc, .{ .quic = .{ 173 return client.attach(alloc, .{ .quic = .{
172 .host_port = q.host_port, 174 .host_port = q.host_port,
@@ -178,19 +180,14 @@ pub fn main() !u8 {
178 // The handoff recipe: ssh fetches the coordinates (and, on a 180 // The handoff recipe: ssh fetches the coordinates (and, on a
179 // cold attach, carries the session if QUIC cannot), while a 181 // cold attach, carries the session if QUIC cannot), while a
180 // warm attach dials from the cache and never spawns ssh at all. 182 // warm attach dials from the cache and never spawns ssh at all.
181 // Building the pieces here keeps client.zig free of XDG and of 183 // handoff.recipeFor owns both pieces; muxweb builds its HOST
182 // allocating a command line — the same split as the ssh proxy 184 // tiles from the same call.
183 // sugar this replaces. 185 const r = try handoff.recipeFor(alloc, h);
184 const cmd = try std.fmt.allocPrint(alloc, "ssh {s} muxd endpoint", .{h}); 186 defer r.deinit(alloc);
185 defer alloc.free(cmd);
186 // An uncacheable host (a separator in the name) or no resolvable
187 // cache directory: attach uncached — always cold, never wrong.
188 const cache: ?[]const u8 = xdg.hostCachePath(alloc, h) catch null;
189 defer if (cache) |c| alloc.free(c);
190 return client.attach(alloc, .{ .hand = .{ 187 return client.attach(alloc, .{ .hand = .{
191 .host = h, 188 .host = h,
192 .ssh_cmd = cmd, 189 .ssh_cmd = r.ssh_cmd,
193 .cache_path = cache, 190 .cache_path = r.cache_path,
194 } }); 191 } });
195 }, 192 },
196 .attach => |t| { 193 .attach => |t| {
src/webhub.zig
Old New
@@ -124,6 +124,94 @@ pub fn parseFrameMessage(msg: []const u8) FrameMsgError!ParsedFrame {
124 return .{ .t = @enumFromInt(f[0]), .payload = f[proto.frame_header_len..] }; 124 return .{ .t = @enumFromInt(f[0]), .payload = f[proto.frame_header_len..] };
125 } 125 }
126 126
127 /// What the WS reader's ALREADY-BUFFERED bytes will do to the next
128 /// `readSmallMessage`, decided without touching the socket.
129 ///
130 /// This exists because readSmallMessage blocks: it fills from the socket
131 /// whenever the frame it is decoding runs past what the reader holds. A
132 /// pump that calls it on a partial frame parks the tile thread with the
133 /// daemon leg unattended — and poll will NOT rescue it, because the
134 /// missing bytes are not in the kernel; the bytes that are already in
135 /// userspace were what made poll fire in the first place.
136 pub const HeadFrame = union(enum) {
137 /// No complete frame at the head. Leave POLLIN armed and re-check
138 /// after the next readable event: a frame that is short is a frame
139 /// whose rest is still in flight, so more bytes ARE coming and poll
140 /// will fire again. (A peer that sends half a frame and then goes
141 /// quiet parks this tile in poll — which is exactly where the
142 /// dead-leg ping below finds it and ends the tile.)
143 incomplete,
144 /// A complete pong at the head, `bytes` long. readSmallMessage
145 /// SWALLOWS pongs and loops to the frame behind them, so handing it
146 /// one is the blocking hazard all over again; the pump tosses the
147 /// pong itself and counts it as the liveness proof it is.
148 pong: usize,
149 /// A whole frame is buffered: readSmallMessage returns it (or names
150 /// it a close) without touching the socket.
151 ready,
152 /// The frame does not FIT the reader's buffer, so no amount of
153 /// waiting makes it readable without blocking — and a pump that kept
154 /// waiting would eventually fill the buffer with a frame it can never
155 /// finish. readSmallMessage answers most of these with MessageTooBig
156 /// and the pump ends the tile either way, so it is said here instead:
157 /// never hand that function a frame it would have to block on. Our
158 /// own page's largest message is a 32 KiB paste chunk against a
159 /// 64 KiB buffer, so reaching this means a peer that is not our page.
160 too_big,
161 };
162
163 const ws_opcode_pong: u4 = 10;
164
165 /// `buffered` is the reader's unread bytes; `capacity` its whole buffer
166 /// (readSmallMessage's own MessageTooBig bound).
167 pub fn headFrame(buffered: []const u8, capacity: usize) HeadFrame {
168 if (buffered.len < 2) return .incomplete;
169 const opcode: u4 = @truncate(buffered[0]);
170 const masked = buffered[1] & 0x80 != 0;
171 const len7: u7 = @truncate(buffered[1]);
172
173 var off: usize = 2;
174 var payload_len: u64 = len7;
175 switch (len7) {
176 126 => {
177 if (buffered.len < off + 2) return .incomplete;
178 payload_len = std.mem.readInt(u16, buffered[off..][0..2], .big);
179 off += 2;
180 },
181 127 => {
182 if (buffered.len < off + 8) return .incomplete;
183 payload_len = std.mem.readInt(u64, buffered[off..][0..8], .big);
184 off += 8;
185 },
186 else => {},
187 }
188 // Browser→server frames are always masked; the bit is checked by
189 // readSmallMessage, but the four key bytes count toward the length
190 // either way.
191 if (masked) {
192 if (buffered.len < off + 4) return .incomplete;
193 off += 4;
194 }
195 // HEADER AND PAYLOAD against the capacity, not the payload alone: the
196 // reader holds both, so a payload that only just fits still leaves a
197 // frame that never completes — and the pump would keep waiting on it
198 // until the buffer was full of bytes it could not use.
199 if (off + payload_len > capacity) return .too_big;
200 if (buffered.len - off < payload_len) return .incomplete;
201 if (opcode == ws_opcode_pong) return .{ .pong = off + @as(usize, @intCast(payload_len)) };
202 return .ready;
203 }
204
205 /// Dead browser leg. The pump's poll is capped at 100 ms because the QUIC
206 /// timer needs the tick, so silence is measured on the wall clock rather
207 /// than by a poll timeout: after `ping_idle_ms` with nothing inbound the
208 /// hub pings, and after `dead_intervals` of them it gives up. What this
209 /// releases is not just a thread — it is the daemon client slot behind
210 /// it, which a browser that vanished with a half-open socket would
211 /// otherwise hold until the daemon exits.
212 pub const ping_idle_ms: i64 = 30_000;
213 pub const dead_intervals: i64 = 3;
214
127 /// One thread per tile, and blocking is the design: the tile's Transport 215 /// 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 216 /// is private to this thread, so Transport.readFrame's blocking read
129 /// (fatal to a multiplexing hub) is simply correct here. A slow browser 217 /// (fatal to a multiplexing hub) is simply correct here. A slow browser
@@ -147,12 +235,20 @@ pub fn pumpTile(
147 // same way reconnect() quiets retries. 235 // same way reconnect() quiets retries.
148 if (target == .hand) target.hand.report_fallback = false; 236 if (target == .hand) target.hand.report_fallback = false;
149 237
150 ws.writeMessage(@constCast(controlMessage(.connecting)), .binary) catch return; 238 ws.writeMessage(controlMessage(.connecting), .binary) catch return;
151 var transport = dialLoop(alloc, target, ws, ws_fd) orelse return; 239 var transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
152 defer transport.close(); 240 defer transport.close();
153 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return; 241 // `up` is not decoration: the browser re-attaches when it reads this,
242 // and the hub never re-attaches on its behalf. mux.js's ENV_CONTROL
243 // handler is the other half of that contract — the residual coupling
244 // this module has to the page it serves, named here where the message
245 // is written rather than left to be discovered.
246 ws.writeMessage(controlMessage(.up), .binary) catch return;
154 247
155 while (true) { 248 var last_inbound_ms = std.time.milliTimestamp();
249 var pings_sent: i64 = 0;
250
251 outer: while (true) {
156 var fds = [_]std.posix.pollfd{ 252 var fds = [_]std.posix.pollfd{
157 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 }, 253 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
158 .{ .fd = ws_fd, .events = std.posix.POLL.IN, .revents = 0 }, 254 .{ .fd = ws_fd, .events = std.posix.POLL.IN, .revents = 0 },
@@ -170,11 +266,15 @@ pub fn pumpTile(
170 .frame => |f| f, 266 .frame => |f| f,
171 .incomplete => break :frames, 267 .incomplete => break :frames,
172 .closed => { 268 .closed => {
173 ws.writeMessage(@constCast(controlMessage(.reconnecting)), .binary) catch return; 269 ws.writeMessage(controlMessage(.reconnecting), .binary) catch return;
174 transport.close(); 270 transport.close();
175 transport = dialLoop(alloc, target, ws, ws_fd) orelse return; 271 transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
176 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return; 272 ws.writeMessage(controlMessage(.up), .binary) catch return;
177 break :frames; 273 // fds[1].revents describes a socket state from
274 // BEFORE the re-dial, and dialLoop may have eaten
275 // the very message it described. Re-poll instead
276 // of reading on a readiness that is now a rumour.
277 continue :outer;
178 }, 278 },
179 }; 279 };
180 defer frame.deinit(alloc); 280 defer frame.deinit(alloc);
@@ -192,28 +292,66 @@ pub fn pumpTile(
192 292
193 // Browser → daemon. One WS message is one frame; drain what the 293 // Browser → daemon. One WS message is one frame; drain what the
194 // reader has already buffered too, or it waits for a poll the 294 // reader has already buffered too, or it waits for a poll the
195 // socket will never signal. 295 // socket will never signal. Every read is gated on headFrame:
296 // readSmallMessage may not be called speculatively, here or
297 // anywhere, because it blocks.
196 if (fds[1].revents != 0) { 298 if (fds[1].revents != 0) {
299 // poll reports what the KERNEL holds; headFrame reads what the
300 // reader holds, and on the first pass that is nothing at all.
301 // One fill turns the readable event into buffered bytes, and
302 // it cannot block — poll just said there are bytes (or an EOF,
303 // which ends the tile right here).
304 ws.input.fillMore() catch return;
197 while (true) { 305 while (true) {
306 switch (headFrame(ws.input.buffered(), ws.input.buffer.len)) {
307 .incomplete => break,
308 .too_big => return,
309 .pong => |n| {
310 ws.input.toss(n);
311 last_inbound_ms = std.time.milliTimestamp();
312 pings_sent = 0;
313 continue;
314 },
315 .ready => {},
316 }
198 const msg = ws.readSmallMessage() catch return; 317 const msg = ws.readSmallMessage() catch return;
199 if (msg.opcode == .binary or msg.opcode == .text) { 318 last_inbound_ms = std.time.milliTimestamp();
200 if (parseFrameMessage(msg.data)) |parsed| { 319 pings_sent = 0;
201 transport.writeFrame(parsed.t, parsed.payload) catch { 320 switch (msg.opcode) {
202 ws.writeMessage(@constCast(controlMessage(.reconnecting)), .binary) catch return; 321 // RFC 6455: a pong carrying the ping's payload back.
203 transport.close(); 322 // Ignoring pings meant a browser heartbeat could not
204 transport = dialLoop(alloc, target, ws, ws_fd) orelse return; 323 // tell a wedged hub from a busy one.
205 ws.writeMessage(@constCast(controlMessage(.up)), .binary) catch return; 324 .ping => ws.writeMessage(msg.data, .pong) catch return,
206 }; 325 .binary, .text => {
207 } else |_| { 326 if (parseFrameMessage(msg.data)) |parsed| {
208 // Not a frame message: dropped, deliberately — 327 transport.writeFrame(parsed.t, parsed.payload) catch {
209 // the browser side is ours, so this is a bug's 328 ws.writeMessage(controlMessage(.reconnecting), .binary) catch return;
210 // signature, and killing every tile for it would 329 transport.close();
211 // make the page unusable exactly when debugging. 330 transport = dialLoop(alloc, target, ws, ws_fd) orelse return;
212 } 331 ws.writeMessage(controlMessage(.up), .binary) catch return;
332 continue :outer; // same stale-revents reason as above
333 };
334 } else |_| {
335 // Not a frame message: dropped, deliberately —
336 // the browser side is ours, so this is a bug's
337 // signature, and killing every tile for it would
338 // make the page unusable exactly when debugging.
339 }
340 },
341 else => {},
213 } 342 }
214 if (ws.input.bufferedLen() == 0) break;
215 } 343 }
216 } 344 }
345
346 // Is anyone still there? A closed tab usually arrives as a close
347 // frame or EOF, but a laptop that slept, a killed browser, or a
348 // dropped ssh -L leaves the socket half-open and silent forever.
349 const silence = std.time.milliTimestamp() - last_inbound_ms;
350 if (silence >= ping_idle_ms * dead_intervals) return;
351 if (silence >= ping_idle_ms * (pings_sent + 1)) {
352 ws.writeMessage("", .ping) catch return;
353 pings_sent += 1;
354 }
217 } 355 }
218 } 356 }
219 357
@@ -240,8 +378,27 @@ fn dialLoop(
240 }; 378 };
241 const n = std.posix.poll(&fds, @intCast(backoff_ms)) catch return null; 379 const n = std.posix.poll(&fds, @intCast(backoff_ms)) catch return null;
242 if (n > 0 and fds[0].revents != 0) { 380 if (n > 0 and fds[0].revents != 0) {
243 const msg = ws.readSmallMessage() catch return null; 381 // Gated exactly as the pump's reads are: a partial frame here
244 _ = msg; // dropped: nothing to carry it yet 382 // would block the dial loop too, and this one has no second
383 // leg to notice. A close frame still ends the tile, which is
384 // the whole point of watching the socket during a backoff.
385 ws.input.fillMore() catch return null;
386 drain: while (true) {
387 switch (headFrame(ws.input.buffered(), ws.input.buffer.len)) {
388 .incomplete => break :drain,
389 .too_big => return null,
390 .pong => |p| ws.input.toss(p),
391 .ready => {
392 const msg = ws.readSmallMessage() catch return null;
393 // Answer a ping even with no transport: the
394 // browser is asking whether the hub is alive, and
395 // "dialing" is a live answer.
396 if (msg.opcode == .ping) ws.writeMessage(msg.data, .pong) catch return null;
397 // Anything else is dropped: nothing to carry it
398 // yet, and the browser re-attaches on `up`.
399 },
400 }
401 }
245 } 402 }
246 } 403 }
247 } 404 }
@@ -413,6 +570,114 @@ test "control messages: the closed vocabulary, envelope included" {
413 try std.testing.expectEqualStrings("\x01{\"state\":\"gone\"}", controlMessage(.gone)); 570 try std.testing.expectEqualStrings("\x01{\"state\":\"gone\"}", controlMessage(.gone));
414 } 571 }
415 572
573 test "head frame: every split boundary is INCOMPLETE, the whole frame is READY" {
574 const cap = ws_buffer_len;
575 // A masked binary frame, 3-byte payload: the shape the browser sends.
576 const small = [_]u8{ 0x82, 0x83, 1, 2, 3, 4, 'a' ^ 1, 'b' ^ 2, 'c' ^ 3 };
577 // Every proper prefix is incomplete — 1-byte header included.
578 for (0..small.len) |n| {
579 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(small[0..n], cap));
580 }
581 try std.testing.expectEqual(HeadFrame.ready, headFrame(&small, cap));
582 // Trailing bytes of the NEXT frame do not make this one incomplete.
583 try std.testing.expectEqual(HeadFrame.ready, headFrame(&(small ++ [_]u8{0x82}), cap));
584
585 // 16-bit extended length: the split lands mid-length and mid-mask.
586 var ext16: [4 + 4 + 200]u8 = undefined;
587 ext16[0] = 0x82;
588 ext16[1] = 0x80 | 126;
589 std.mem.writeInt(u16, ext16[2..4], 200, .big);
590 @memset(ext16[4..8], 0); // mask
591 @memset(ext16[8..], 'x');
592 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext16[0..3], cap)); // mid-length
593 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext16[0..6], cap)); // mid-mask
594 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext16[0..8], cap)); // header only
595 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext16[0 .. ext16.len - 1], cap)); // mid-payload
596 try std.testing.expectEqual(HeadFrame.ready, headFrame(&ext16, cap));
597
598 // 64-bit extended length: same, one byte at a time through the length.
599 var ext64: [2 + 8 + 4 + 70000]u8 = undefined;
600 ext64[0] = 0x82;
601 ext64[1] = 0x80 | 127;
602 std.mem.writeInt(u64, ext64[2..10], 70000, .big);
603 @memset(ext64[10..14], 0);
604 @memset(ext64[14..], 'y');
605 for (2..10) |n| {
606 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext64[0..n], cap));
607 }
608 // 70000 > the reader's buffer: named from the header alone, because
609 // waiting for a frame that can never be buffered whole would fill the
610 // buffer with bytes the pump can never use.
611 try std.testing.expectEqual(HeadFrame.too_big, headFrame(ext64[0..14], cap));
612 // ...and that verdict does NOT come before the header is buffered.
613 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(ext64[0..11], cap));
614
615 // The boundary the payload-only check got wrong: a payload that fits
616 // the buffer EXACTLY still needs 14 bytes of header alongside it, so
617 // the frame as a whole never fits and the pump must not wait for it.
618 var edge: [14]u8 = undefined;
619 edge[0] = 0x82;
620 edge[1] = 0x80 | 127;
621 std.mem.writeInt(u64, edge[2..10], cap, .big);
622 @memset(edge[10..14], 0);
623 try std.testing.expectEqual(HeadFrame.too_big, headFrame(&edge, cap));
624 // One byte under the whole-frame budget is an ordinary wait.
625 std.mem.writeInt(u64, edge[2..10], cap - 14, .big);
626 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(&edge, cap));
627
628 // A pong is named separately: readSmallMessage swallows it and blocks
629 // on whatever is behind it, so the pump must toss it itself.
630 const pong = [_]u8{ 0x8a, 0x84, 0, 0, 0, 0, 'p', 'i', 'n', 'g' };
631 try std.testing.expectEqual(@as(usize, 10), headFrame(&pong, cap).pong);
632 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(pong[0..9], cap));
633 // An UNMASKED pong (what a server sends) is 4 bytes shorter.
634 const server_pong = [_]u8{ 0x8a, 0x00 };
635 try std.testing.expectEqual(@as(usize, 2), headFrame(&server_pong, cap).pong);
636
637 // Ping and close are ordinary reads: readSmallMessage returns the
638 // ping and errors ConnectionClose on the close, both without a socket
639 // read, and both end up handled rather than waited on.
640 try std.testing.expectEqual(HeadFrame.ready, headFrame(&[_]u8{ 0x89, 0x80, 0, 0, 0, 0 }, cap));
641 try std.testing.expectEqual(HeadFrame.ready, headFrame(&[_]u8{ 0x88, 0x80, 0, 0, 0, 0 }, cap));
642 // Empty buffer: nothing to do, and above all no speculative read.
643 try std.testing.expectEqual(HeadFrame.incomplete, headFrame(&.{}, cap));
644 }
645
646 test "tiles json: order preserved, the unrepresentable bytes escaped" {
647 const alloc = std.testing.allocator;
648 {
649 const j = try tilesJson(alloc, &.{});
650 defer alloc.free(j);
651 try std.testing.expectEqualStrings("[]", j);
652 }
653 {
654 // Index order IS the tile order — /ws/<idx> indexes the same list.
655 const j = try tilesJson(alloc, &.{ "box1", "box2", "box3" });
656 defer alloc.free(j);
657 try std.testing.expectEqualStrings("[\"box1\",\"box2\",\"box3\"]", j);
658 }
659 {
660 // A path with a quote in it must not break the page.
661 const j = try tilesJson(alloc, &.{"/tmp/we\"ird\\path"});
662 defer alloc.free(j);
663 try std.testing.expectEqualStrings("[\"/tmp/we\\\"ird\\\\path\"]", j);
664 }
665 {
666 // Control bytes go to \u00XX, including the ones JSON has short
667 // spellings for — one rule, no table to get wrong.
668 const j = try tilesJson(alloc, &.{"a\nb\tc\x00d\x1fe"});
669 defer alloc.free(j);
670 try std.testing.expectEqualStrings("[\"a\\u000ab\\u0009c\\u0000d\\u001fe\"]", j);
671 }
672 {
673 // Bytes above 0x7f pass through: labels are argv, and a UTF-8
674 // hostname stays itself.
675 const j = try tilesJson(alloc, &.{ "", "héllo" });
676 defer alloc.free(j);
677 try std.testing.expectEqualStrings("[\"\",\"héllo\"]", j);
678 }
679 }
680
416 test "frame messages: exact framing in, everything else named" { 681 test "frame messages: exact framing in, everything else named" {
417 // Types spelled via the enum so the test cannot drift from the wire. 682 // Types spelled via the enum so the test cannot drift from the wire.
418 const input_byte: u8 = @intFromEnum(proto.MsgType.input); 683 const input_byte: u8 = @intFromEnum(proto.MsgType.input);
src/webhub_main.zig
Old New
@@ -11,6 +11,7 @@ const client = @import("client");
11 const webhub = @import("webhub"); 11 const webhub = @import("webhub");
12 const build_options = @import("build_options"); 12 const build_options = @import("build_options");
13 const xdg = @import("xdg"); 13 const xdg = @import("xdg");
14 const handoff = @import("handoff");
14 15
15 const usage = 16 const usage =
16 \\usage: muxweb TARGET [TARGET ...] [--port N] 17 \\usage: muxweb TARGET [TARGET ...] [--port N]
@@ -140,9 +141,10 @@ pub fn main() !u8 {
140 }; 141 };
141 defer parsed.deinit(alloc); 142 defer parsed.deinit(alloc);
142 143
143 // Resolve spellings into dialable Targets. The recipes mirror 144 // Resolve spellings into dialable Targets through the SAME owners
144 // mux_main's .host/.quic arms (a cross-file duplication, noted as a 145 // mux_main uses — handoff.recipeFor and xdg.resolveKeyPath — so the
145 // dedup candidate); labels are the argv spellings verbatim. 146 // two binaries cannot drift on what a bare HOST or a `quic://` means.
147 // Labels are the argv spellings verbatim.
146 var targets: std.ArrayList(client.Target) = .empty; 148 var targets: std.ArrayList(client.Target) = .empty;
147 defer targets.deinit(alloc); 149 defer targets.deinit(alloc);
148 var labels: std.ArrayList([]const u8) = .empty; 150 var labels: std.ArrayList([]const u8) = .empty;
@@ -163,30 +165,32 @@ pub fn main() !u8 {
163 try labels.append(alloc, path); 165 try labels.append(alloc, path);
164 }, 166 },
165 .host => |h| { 167 .host => |h| {
166 const cmd = try std.fmt.allocPrint(alloc, "ssh {s} muxd endpoint", .{h}); 168 const r = try handoff.recipeFor(alloc, h);
167 try owned.append(alloc, cmd); 169 try owned.append(alloc, r.ssh_cmd);
168 const cache: ?[]const u8 = xdg.hostCachePath(alloc, h) catch null; 170 if (r.cache_path) |c| try owned.append(alloc, c);
169 if (cache) |c| try owned.append(alloc, c);
170 try targets.append(alloc, .{ .hand = .{ 171 try targets.append(alloc, .{ .hand = .{
171 .host = h, 172 .host = h,
172 .ssh_cmd = cmd, 173 .ssh_cmd = r.ssh_cmd,
173 .cache_path = cache, 174 .cache_path = r.cache_path,
174 .idle_ms = parsed.idle_ms, 175 .idle_ms = parsed.idle_ms,
175 } }); 176 } });
176 try labels.append(alloc, h); 177 try labels.append(alloc, h);
177 }, 178 },
178 .quic => |hp| { 179 .quic => |hp| {
179 const key_path = parsed.key orelse blk: { 180 const key_path = switch (try xdg.resolveKeyPath(alloc, parsed.key)) {
180 const p = try xdg.keyPath(alloc); 181 .given => |p| p,
181 try owned.append(alloc, p); 182 .default => |p| blk: {
182 std.fs.cwd().access(p, .{}) catch { 183 try owned.append(alloc, p);
184 break :blk p;
185 },
186 .missing => |p| {
187 defer alloc.free(p);
183 std.debug.print( 188 std.debug.print(
184 "muxweb: no key for quic://{s}: pass --key, set MUX_KEY_FILE, or run `muxd keygen` (default {s})\n", 189 "muxweb: no key for quic://{s}: pass --key, set MUX_KEY_FILE, or run `muxd keygen` (default {s})\n",
185 .{ hp, p }, 190 .{ hp, p },
186 ); 191 );
187 return 2; 192 return 2;
188 }; 193 },
189 break :blk p;
190 }; 194 };
191 try targets.append(alloc, .{ .quic = .{ 195 try targets.append(alloc, .{ .quic = .{
192 .host_port = hp, 196 .host_port = hp,
src/xdg.zig
Old New
@@ -14,6 +14,33 @@ pub fn keyPath(alloc: std.mem.Allocator) ![]const u8 {
14 return keyPathFrom(alloc, std.posix.getenv("XDG_CONFIG_HOME"), std.posix.getenv("HOME")); 14 return keyPathFrom(alloc, std.posix.getenv("XDG_CONFIG_HOME"), std.posix.getenv("HOME"));
15 } 15 }
16 16
17 /// Where a `quic://` dial's key came from, so each binary can spell its
18 /// own refusal around the path it actually looked at.
19 pub const KeyResolution = union(enum) {
20 /// `--key` or `$MUX_KEY_FILE`. BORROWED from argv/env — do not free —
21 /// and deliberately unchecked: naming a key explicitly is the user
22 /// asserting it, and a wrong path fails at the dial with its own
23 /// error rather than being second-guessed here.
24 given: []const u8,
25 /// The XDG default, which exists. Owned by the caller.
26 default: []const u8,
27 /// The XDG default, which does not. Owned by the caller, and carried
28 /// out rather than printed: `mux` and `muxweb` word this differently
29 /// (the hub names the tile that wanted it) and both need the path.
30 missing: []const u8,
31 };
32
33 /// The key-resolution rule both binaries follow: an explicit spelling
34 /// wins, otherwise the XDG default must already exist. ONE owner, because
35 /// a drift here would mean two binaries disagreeing about which key a
36 /// `quic://` target authenticates with.
37 pub fn resolveKeyPath(alloc: std.mem.Allocator, given: ?[]const u8) !KeyResolution {
38 if (given) |g| return .{ .given = g };
39 const p = try keyPath(alloc);
40 std.fs.cwd().access(p, .{}) catch return .{ .missing = p };
41 return .{ .default = p };
42 }
43
17 pub fn keyPathFrom( 44 pub fn keyPathFrom(
18 alloc: std.mem.Allocator, 45 alloc: std.mem.Allocator,
19 xdg_config_home: ?[]const u8, 46 xdg_config_home: ?[]const u8,