0a285e41

feat: lazy QUIC bind — endpoint_req stands up an ephemeral-port listener on demand

a73x 2026-08-11 10:48

Commit message

feat: lazy QUIC bind — endpoint_req stands up an ephemeral-port listener on demand

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

build.zig

Old		New
@@ -146,6 +146,9 @@ pub fn build(b: *std.Build) void {
146	server_mod.addImport("protocol", protocol_mod);	146	server_mod.addImport("protocol", protocol_mod);
147	server_mod.addImport("quic", quic_mod);	147	server_mod.addImport("quic", quic_mod);
148	server_mod.addImport("testtmp", testtmp_mod);	148	server_mod.addImport("testtmp", testtmp_mod);
		149	// For endpoint_req's lazy bind: the default key path, resolved by the
		150	// daemon itself when nobody handed it a --key.
		151	server_mod.addImport("xdg", xdg_mod);
149		152
150	// The client's QUIC transport. Imports the listener's module for the	153	// The client's QUIC transport. Imports the listener's module for the
151	// pieces both ends must agree on (the key, the egress ring's lifetime	154	// pieces both ends must agree on (the key, the egress ring's lifetime

src/server.zig

@@ @@ -10,6 +10,7 @@ const Engine = @import("engine").Engine; @@
+const Pty = @import("pty").Pty;
+const proto = @import("protocol");
+const quic = @import("quic");
+const xdg = @import("xdg");
+const TmpDir = @import("testtmp").TmpDir;
+const max_clients = 8;
@@ @@ -57,6 +58,17 @@ fn drainWaitMs(remaining: i64, quic_hint: ?i32) i32 { @@
+    return @max(1, @min(hint, rem));
+/// The port the kernel actually gave a listener, asked of the socket rather
+/// than remembered: a lazy bind names port 0 and only getsockname knows what
+/// came back. 0 on failure, which is the same "no port to announce" the
+/// endpoint_reply already spells that way.
+fn boundUdpPort(l: *quic.Listener) u16 {
+    var actual: std.posix.sockaddr.storage = undefined;
+    var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
+    std.posix.getsockname(l.pollFd(), @ptrCast(&actual), &len) catch return 0;
+    return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort();
+}
+const Wyhash = std.hash.Wyhash;
+const Stats = struct {
@@ @@ -385,6 +397,11 @@ pub const Server = struct { @@
+    /// QUIC is opt-in per invocation and the unix socket is unaffected by
+    /// its presence or absence.
+    quic_listener: ?*quic.Listener = null,
+    /// True when the listener was bound lazily by endpoint_req, and is
+    /// therefore ours to deinit. A listener handed in by main.zig (the
+    /// explicit --quic path) has its own deferred deinit out there, and
+    /// freeing it twice would be a use-after-free at shutdown.
+    quic_owned: bool = false,
+    /// The pty's line-discipline bits as last put on the wire, or null
+    /// before the first poll. Deliberately "what clients have been told"
+    /// rather than "what the pty says": the two differ for exactly the span
@@ @@ -492,6 +509,16 @@ pub const Server = struct { @@
+        for (self.observers) |slot| {
+            if (slot) |fd| std.posix.close(fd);
+        // After the client slots, never before: a QUIC sink closes its
+        // connection THROUGH the listener, so the listener has to outlive
+        // the slots that hold it. That is the ordering main.zig's defers
+        // give the explicit --quic path, kept here for the lazy one.
+        // Only a listener we bound ourselves — see quic_owned.
+        if (self.quic_owned) {
+            if (self.quic_listener) |l| l.deinit();
+            self.quic_listener = null;
+            self.quic_owned = false;
+        }
+        // Only unlink the socket path if it still refers to *our* socket — a
+        // newer daemon may have replaced the file since we bound it, and
+        // deleting that one would hand its clients the same field incident
@@ @@ -914,6 +941,71 @@ pub const Server = struct { @@
+        self.quic_listener = listener;
+    /// The endpoint_req answer: the bound QUIC port, standing a listener up
+    /// on demand if none exists. 0 means "could not", and the reason goes to
+    /// the daemon log (stderr) rather than into the frame — the asker can
+    /// do nothing with it but relay, and `muxd endpoint`'s announce-none
+    /// already tells the client everything it can act on.
+    fn endpointPort(self: *Server) u16 {
+        return self.endpointPortFrom(
+            std.posix.getenv("MUX_KEY_FILE"),
+            std.posix.getenv("XDG_CONFIG_HOME"),
+            std.posix.getenv("HOME"),
+        );
+    }
+    /// Env handed in, nothing read: the xdg *From pattern, for the same
+    /// reason — tests cannot setenv. Key resolution is MUX_KEY_FILE then
+    /// the default path; there is no --key half because a daemon being
+    /// asked lazily is one that was never handed a flag.
+    fn endpointPortFrom(
+        self: *Server,
+        env_key: ?[]const u8,
+        xdg_config_home: ?[]const u8,
+        home: ?[]const u8,
+    ) u16 {
+        if (self.quic_listener) |q| return boundUdpPort(q);
+        // "Set but empty" is unset, the same reading main.zig's envKey gives
+        // it: an empty path could only ever be a mistake.
+        const env: ?[]const u8 = if (env_key) |v|
+            (if (v.len == 0) null else v)
+        else
+            null;
+        var owned: ?[]const u8 = null;
+        defer if (owned) |p| self.alloc.free(p);
+        const key_path = env orelse blk: {
+            const dflt = xdg.keyPathFrom(self.alloc, xdg_config_home, home) catch break :blk null;
+            owned = dflt;
+            break :blk if (std.fs.cwd().access(dflt, .{})) |_| dflt else |_| @as(?[]const u8, null);
+        } orelse {
+            std.debug.print("muxd: endpoint_req: no key to listen with (run `muxd keygen`)\n", .{});
+            return 0;
+        };
+        const key = quic.Key.load(key_path) catch |err| {
+            std.debug.print(
+                "muxd: endpoint_req: cannot load key {s}: {s}\n",
+                .{ key_path, @errorName(err) },
+            );
+            return 0;
+        };
+        return self.lazyBindQuic(key) catch |err| {
+            std.debug.print("muxd: endpoint_req: cannot bind udp: {s}\n", .{@errorName(err)});
+            return 0;
+        };
+    }
+    /// Bind 0.0.0.0 on a kernel-assigned port and wire it in. The poll loop
+    /// re-reads `quic_listener` every iteration, so there is no loop surgery
+    /// here — setting the field IS the integration.
+    fn lazyBindQuic(self: *Server, key: quic.Key) !u16 {
+        const addr = try std.net.Address.parseIp("0.0.0.0", 0);
+        const l = try quic.Listener.bind(self.alloc, addr, key, quic.default_idle_ms);
+        l.setHandler(self.quicHandler());
+        self.quic_listener = l;
+        self.quic_owned = true;
+        return boundUdpPort(l);
+    }
+    /// One frame, on the stack, for the two places the QUIC path has to
+    /// speak before a client slot exists.
+    fn frameBytes(t: proto.MsgType, payload: []const u8) [6]u8 {
@@ @@ -1172,6 +1264,13 @@ pub const Server = struct { @@
+            // authority an attached client lacks — see the .input arm above,
+            // which already writes to the pty master.
+            .stop_req => shutdown_flag.store(true, .release),
+            // The other half of the observer arm below, here for the same
+            // reason stop_req's is: the verb means the same thing on any
+            // connection. Replies through the queue like stats does.
+            .endpoint_req => {
+                const payload = proto.encodeEndpointReply(self.endpointPort());
+                _ = self.queueFrame(i, .endpoint_reply, &payload);
+            },
+            .debug_dump => {
+                const dump = self.buildDump(frame.payload) catch {
+                    self.dropClient(i);
@@ @@ -1234,6 +1333,11 @@ pub const Server = struct { @@
+            // Where `muxd stop` actually lands, since it never attaches.
+            // Same signal path as the client arm.
+            .stop_req => shutdown_flag.store(true, .release),
+            // Where `muxd endpoint` actually lands, since it never attaches.
+            .endpoint_req => {
+                const payload = proto.encodeEndpointReply(self.endpointPort());
+                proto.writeFrame(fd, .endpoint_reply, &payload) catch self.dropObserver(i);
+            },
+            else => {},
+    try std.testing.expect(shutdown_flag.load(.acquire));
+// ---------------------------------------------------------------------------
+// endpoint_req: the lazy QUIC bind.
+//
+// Consolidated into two tests on purpose. `quic.Listener` keeps a
+// process-global "one listener at a time" latch, so every binding test has to
+// give it back before the next one asks — fewer tests is fewer places that
+// can fail to.
+// ---------------------------------------------------------------------------
+/// Pump the daemon until a frame of `want` arrives on `fd`, or the budget of
+/// iterations runs out. Single-threaded on purpose: the loop that services
+/// the socket is the same one that has to answer, so the pump and the read
+/// have to interleave. Caller owns the returned frame.
+fn awaitFrame(
+    alloc: std.mem.Allocator,
+    srv: *Server,
+    fd: std.posix.fd_t,
+    want: proto.MsgType,
+    iters: usize,
+) !?proto.Frame {
+    var i: usize = 0;
+    while (i < iters) : (i += 1) {
+        _ = try srv.pumpOnce(5);
+        var pfd = [_]std.posix.pollfd{
+            .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
+        };
+        if ((std.posix.poll(&pfd, 1) catch 0) == 0) continue;
+        const frame = (try proto.readFrame(alloc, fd)) orelse return null;
+        if (frame.type == want) return frame;
+        // Anything else on the way (a snapshot for the attached half of this
+        // test) is not what was asked for; drop it and keep pumping.
+        frame.deinit(alloc);
+    }
+    return null;
+}
+test "Server: endpoint_req binds a listener lazily, answers the same port on both dispatches, and deinit hands it back" {
+    const alloc = std.testing.allocator;
+    var tmp = try TmpDir.make();
+    defer tmp.cleanup();
+    var kbuf: [128]u8 = undefined;
+    const key_path = try std.fmt.bufPrint(&kbuf, "{s}/key", .{tmp.path()});
+    try xdg.writeNewKey(key_path);
+    const key = try quic.Key.load(key_path);
+    var sbuf: [128]u8 = undefined;
+    const sock_path = try std.fmt.bufPrint(&sbuf, "{s}/ep.sock", .{tmp.path()});
+    // Scoped so srv.deinit() runs before the release check below, and still
+    // runs if an assertion inside fails.
+    var first_port: u16 = 0;
+    {
+        var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
+        defer srv.deinit();
+        // Key resolution first, through the env seam: MUX_KEY_FILE's value
+        // handed in rather than set, since tests cannot setenv.
+        first_port = srv.endpointPortFrom(key_path, null, null);
+        try std.testing.expect(first_port != 0);
+        try std.testing.expect(srv.quic_listener != null);
+        // Now the wire, on a listener that already exists — which is what
+        // keeps this half from depending on the test machine having a real
+        // ~/.config/mux/key. The observer arm is the load-bearing one:
+        // `muxd endpoint` never attaches.
+        const obs = try std.net.connectUnixSocket(sock_path);
+        defer obs.close();
+        try proto.writeFrame(obs.handle, .endpoint_req, "");
+        const reply = (try awaitFrame(alloc, &srv, obs.handle, .endpoint_reply, 200)) orelse
+            return error.NoEndpointReply;
+        defer reply.deinit(alloc);
+        try std.testing.expectEqual(first_port, try proto.decodeEndpointReply(reply.payload));
+        // Asked twice, answered the same: a second bind attempt would be
+        // refused by the process-global latch and surface as port 0, so
+        // equality is the stronger claim than "nonzero again".
+        try proto.writeFrame(obs.handle, .endpoint_req, "");
+        const reply2 = (try awaitFrame(alloc, &srv, obs.handle, .endpoint_reply, 200)) orelse
+            return error.NoSecondEndpointReply;
+        defer reply2.deinit(alloc);
+        try std.testing.expectEqual(first_port, try proto.decodeEndpointReply(reply2.payload));
+        // The attached-client arm answers the same verb the same way, on a
+        // separate connection that attaches first.
+        const cl = try std.net.connectUnixSocket(sock_path);
+        defer cl.close();
+        try proto.writeFrame(cl.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
+        try proto.writeFrame(cl.handle, .endpoint_req, "");
+        const reply3 = (try awaitFrame(alloc, &srv, cl.handle, .endpoint_reply, 400)) orelse
+            return error.NoAttachedEndpointReply;
+        defer reply3.deinit(alloc);
+        try std.testing.expectEqual(first_port, try proto.decodeEndpointReply(reply3.payload));
+    }
+    // A lazily bound listener is the server's to free, and this is the
+    // observable for that: the latch is only released by a deinit that
+    // happened, so a fresh bind succeeding proves the old one is gone.
+    const addr = try std.net.Address.parseIp("0.0.0.0", 0);
+    const l2 = try quic.Listener.bind(alloc, addr, key, quic.default_idle_ms);
+    l2.deinit();
+}
+test "Server: endpointPortFrom refuses without a key, survives it, and prefers a listener already bound" {
+    const alloc = std.testing.allocator;
+    var tmp = try TmpDir.make();
+    defer tmp.cleanup();
+    var kbuf: [128]u8 = undefined;
+    const key_path = try std.fmt.bufPrint(&kbuf, "{s}/key", .{tmp.path()});
+    try xdg.writeNewKey(key_path);
+    const key = try quic.Key.load(key_path);
+    var sbuf: [128]u8 = undefined;
+    const sock_path = try std.fmt.bufPrint(&sbuf, "{s}/ep2.sock", .{tmp.path()});
+    var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
+    defer srv.deinit();
+    // No env at all: keyPathFrom has nothing to build a default from, so
+    // there is no key and the answer is 0 rather than a crash.
+    try std.testing.expectEqual(@as(u16, 0), srv.endpointPortFrom(null, null, null));
+    // Named but absent: Key.load fails, same answer.
+    var mbuf: [128]u8 = undefined;
+    const missing = try std.fmt.bufPrint(&mbuf, "{s}/absent-key", .{tmp.path()});
+    try std.testing.expectEqual(@as(u16, 0), srv.endpointPortFrom(missing, null, null));
+    // A default path pointing at a directory with no key in it: resolvable,
+    // not present, so the access check refuses before Key.load is asked.
+    try std.testing.expectEqual(@as(u16, 0), srv.endpointPortFrom(null, tmp.path(), null));
+    // Refusing left no half-state behind: nothing was bound, and the daemon
+    // is still able to bind when a real key does turn up.
+    try std.testing.expect(srv.quic_listener == null);
+    const port = try srv.lazyBindQuic(key);
+    try std.testing.expect(port != 0);
+    // With a listener in hand the answer is the bound port, and key
+    // resolution is never reached — all-null env would otherwise return 0.
+    try std.testing.expectEqual(port, srv.endpointPortFrom(null, null, null));
+}