40b14979

feat: link is the fd|pipe|quic union and the one await loop, as a row

a73x 2026-08-31 18:26

Commit message

feat: link is the fd|pipe|quic union and the one await loop, as a row

One change from the brief's code, and it is behavioural rather than a
spelling: awaitFrame polls before it reads on a BOUNDED fd or pipe
instead of reading first.

The brief read first on every arm, for the QUIC reason — a whole frame
may already sit in cl.in from an earlier pump, and poll never fires for
bytes already in userspace. But readFrame on the fd and pipe arms is a
read(2) on a stream, and on a quiet peer that blocks until the peer
speaks or goes away. The brief's own "silence consumes the deadline"
test therefore never returned: `make check` ran 420s to a timeout, and
a standalone repro confirmed read(2) on a socketpair whose peer is open
and silent does not come back at all. A test that hangs takes the whole
`zig build test` runner down with it and prints nothing, which is the
failure mode CLAUDE.md names first.

So the eager read is now the QUIC arm's and the deadline-free arm's,
where the block IS the wait; a bounded fd spends its wait in poll,
where the deadline can end it, exactly as dial.askOn has always done.
Once poll has announced bytes the fd arms still block until the frame
is whole — that price is unchanged.

One test beyond the brief's list. Its fixture held the arm constant —
every case was `.fd` — so the pipe arm's r/w assignment was unpinned
and a swap of the two would have shipped green. The added test runs a
real `cat` and reads its echo back, which fails if either direction
addresses the wrong end.

Verified by breaking one assertion in each test and seeing each fail by
name, then restoring: make check rc=0.

build.zig

src/link.zig

@@ @@ -0,0 +1,439 @@ @@
+//! The live connection to a daemon, however it was reached: a unix-socket fd,
+//! the stdio of a `--via`/handoff child, or a QUIC client. Mechanics only —
+//! send a frame, read a frame, wait, close. Policy stays with the owners:
+//! redial and backoff are `client.Transport`'s and muxa's, attach semantics
+//! are dial's and the callers'. This row exists because the fd|pipe|quic
+//! union used to live twice (client.Transport, muxa.AgentConnection), each
+//! with its own send, await and close.
+const std = @import("std");
+const proto = @import("term").protocol;
+const quic = @import("quic");
+/// Three outcomes, not two: QUIC's socket goes readable for acks and half
+/// frames, so `null` cannot keep the socket path's meaning of "peer gone"
+/// without making every partial frame a reconnect. (Moved from client.zig.)
+pub const Incoming = union(enum) {
+    frame: proto.Frame,
+    incomplete,
+    closed,
+};
+/// What a non-matching frame does to an `awaitFrame` wait. `on == null` is
+/// drop: the frame is freed and the wait continues — the observer-verb
+/// policy dial.ask always had. A non-null `on` BORROWS the frame for the
+/// duration of the call and must copy anything it keeps; awaitFrame frees
+/// the frame when `on` returns. An error out of `on` ends the wait with
+/// that error — some "other" frames are answers, not noise (muxa's
+/// exit_status), and only the caller knows which.
+pub const Sink = struct {
+    ctx: ?*anyopaque = null,
+    on: ?*const fn (ctx: ?*anyopaque, frame: proto.Frame) anyerror!void = null,
+};
+pub const Link = union(enum) {
+    /// A unix socket: one fd, read and write. -1 once closed — close() is
+    /// idempotent because a re-dial releases the dead link on entry and an
+    /// abort then closes the same value again through a defer; a second
+    /// close(2) on a stale fd is EBADF, which std.posix maps to unreachable.
+    fd: std.posix.fd_t,
+    /// `--via`, and the ssh half of a handoff: the child whose stdio IS the
+    /// transport. `r` is its stdout, `w` its stdin.
+    pipe: Pipe,
+    /// The connection that IS the transport; bytes go through the stream
+    /// layer, so there is nothing to write(2) to.
+    quic: Quic,
+    pub const Pipe = struct {
+        child: std.process.Child,
+        r: std.posix.fd_t,
+        w: std.posix.fd_t,
+    };
+    pub const Quic = struct {
+        cl: *quic.Client,
+        /// Bytes the QUIC ring would not take yet. Frames are appended whole
+        /// and handed over a prefix at a time, so a short accept can never
+        /// split one on the wire — the remainder is offered again next pass.
+        /// Lives here and not in the wrappers because the staging is a
+        /// property of the quic link, and it used to exist twice
+        /// (Transport.qout, muxa.sendFrameQuic's stack buffer).
+        qout: std.ArrayList(u8) = .empty,
+        alloc: std.mem.Allocator,
+    };
+    pub fn pollFd(self: *const Link) std.posix.fd_t {
+        return switch (self.*) {
+            .fd => |fd| fd,
+            .pipe => |p| p.r,
+            .quic => |q| q.cl.pollFd(),
+        };
+    }
+    /// Folds ngtcp2's next deadline in, so retransmits and idle timeouts
+    /// happen on time without a second timer.
+    pub fn timeoutMs(self: *Link, cap_ms: i32) i32 {
+        return switch (self.*) {
+            .quic => |*q| q.cl.timeoutMs(cap_ms),
+            .fd, .pipe => cap_ms,
+        };
+    }
+    /// Unconditional: a QUIC connection's timers are the only thing that
+    /// notices a peer which stopped answering.
+    pub fn service(self: *Link) void {
+        switch (self.*) {
+            .quic => |*q| {
+                q.cl.pump();
+                self.flushQuic();
+            },
+            .fd, .pipe => {},
+        }
+    }
+    /// Queue-and-offer, never blocking: fd and pipe write through; quic
+    /// appends whole and flushes what the ring takes. A caller that must
+    /// KNOW the bytes left (muxa's verbs) follows with flushWithin.
+    pub fn sendFrame(self: *Link, t: proto.MsgType, payload: []const u8) !void {
+        switch (self.*) {
+            .quic => |*q| {
+                try proto.appendFrame(&q.qout, q.alloc, t, payload);
+                self.flushQuic();
+            },
+            .fd => |fd| return proto.writeFrame(fd, t, payload),
+            .pipe => |p| return proto.writeFrame(p.w, t, payload),
+        }
+    }
+    /// Offer the outbound queue to the ring again. After every write and on
+    /// every service pass, because the room to accept comes from
+    /// acknowledgements, which arrive on their own schedule.
+    fn flushQuic(self: *Link) void {
+        const q = switch (self.*) {
+            .quic => |*q| q,
+            .fd, .pipe => return,
+        };
+        if (q.qout.items.len == 0) return;
+        const n = q.cl.send(q.qout.items);
+        if (n == 0) return;
+        q.qout.replaceRangeAssumeCapacity(0, n, &.{});
+    }
+    /// Drive the staged bytes out or say why not, within the deadline. A
+    /// no-op for fd/pipe (their sendFrame already either took the bytes or
+    /// failed). The QUIC arm is muxa's old sendFrameQuic loop: the ring is
+    /// full, only the peer's acks empty it, and they arrive through pump —
+    /// polling first keeps this from spinning.
+    pub fn flushWithin(self: *Link, deadline_ms: u32) !void {
+        const q = switch (self.*) {
+            .quic => |*q| q,
+            .fd, .pipe => return,
+        };
+        const end = std.time.milliTimestamp() + deadline_ms;
+        while (q.qout.items.len != 0) {
+            if (q.cl.dead) return error.ConnectionLost;
+            self.flushQuic();
+            if (q.qout.items.len == 0) return;
+            if (std.time.milliTimestamp() >= end) return error.SendStalled;
+            var fds = [_]std.posix.pollfd{
+                .{ .fd = q.cl.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
+            };
+            _ = std.posix.poll(&fds, q.cl.timeoutMs(50)) catch return error.ConnectionLost;
+            q.cl.pump();
+        }
+    }
+    /// The next whole frame, if there is one. (Moved from
+    /// client.Transport.readFrame; see Incoming for why a missing frame is
+    /// not automatically a dead transport.)
+    pub fn readFrame(self: *Link, alloc: std.mem.Allocator) !Incoming {
+        switch (self.*) {
+            .quic => |*q| {
+                // Death is checked after the pump, so bytes that arrived in
+                // the same pass as the close are still delivered before the
+                // tear.
+                const got = proto.takeFrame(alloc, &q.cl.in) catch |err| switch (err) {
+                    error.OutOfMemory => return err, // not a transport event
+                    else => return .closed,
+                };
+                if (got) |frame| return .{ .frame = frame };
+                return if (q.cl.dead) .closed else .incomplete;
+            },
+            .fd => |fd| {
+                const frame = (proto.readFrame(alloc, fd) catch |err| switch (err) {
+                    error.OutOfMemory => return err,
+                    else => return .closed,
+                }) orelse return .closed;
+                return .{ .frame = frame };
+            },
+            .pipe => |p| {
+                const frame = (proto.readFrame(alloc, p.r) catch |err| switch (err) {
+                    error.OutOfMemory => return err,
+                    else => return .closed,
+                }) orelse return .closed;
+                return .{ .frame = frame };
+            },
+        }
+    }
+    /// Wait for one frame of type `want`. THE primitive this row exists
+    /// for; every hand-rolled poll+readFrame loop in the tree is a copy of
+    /// this. Returns null when the deadline runs out (a null deadline waits
+    /// forever), error.Closed when the peer is gone — callers own the
+    /// wording for both (dial.ask maps Closed to its "no answer" null; muxa
+    /// maps it to DaemonGone/ConnectionLost). Non-matching frames go to the
+    /// sink (see Sink for ownership).
+    pub fn awaitFrame(
+        self: *Link,
+        alloc: std.mem.Allocator,
+        want: proto.MsgType,
+        deadline_ms: ?u32,
+        sink: Sink,
+    ) !?proto.Frame {
+        const end: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null;
+        // Whether this link may be asked for a frame without poll saying so
+        // first. QUIC may: a whole frame can already sit in cl.in from an
+        // earlier pump, and poll would never fire for bytes that are already
+        // in userspace. A BOUNDED fd or pipe may not: readFrame there is a
+        // read(2) on a stream, which on a quiet peer returns only when that
+        // peer speaks or goes away — the wait has to be spent in poll, where
+        // the deadline can end it, which is why dial.askOn polls first too.
+        // With no deadline the blocking read IS the wait, so they read
+        // straight away and never reach the poll below.
+        const eager = switch (self.*) {
+            .quic => true,
+            .fd, .pipe => end == null,
+        };
+        var ready = eager;
+        while (true) {
+            if (ready) {
+                // A non-match falls back to `eager`: on a bounded fd the next
+                // frame has to be announced by poll again, or draining noise
+                // would walk straight into a blocking read.
+                ready = eager;
+                switch (try self.readFrame(alloc)) {
+                    .frame => |frame| {
+                        if (frame.type == want) return frame;
+                        if (sink.on) |on| {
+                            defer frame.deinit(alloc);
+                            try on(sink.ctx, frame);
+                        } else frame.deinit(alloc);
+                        continue;
+                    },
+                    .closed => return error.Closed,
+                    .incomplete => {},
+                }
+            }
+            // Nothing whole in hand: wait. The wait is capped at 250ms even
+            // with no caller deadline, because a QUIC link's timers (loss
+            // detection, keepalive) need servicing on schedule rather than
+            // whenever the daemon happens to say something; a plain fd with
+            // no deadline may block indefinitely, which is what a caller
+            // wants when a stopped daemon should be a visible hang.
+            var cap: i32 = undefined;
+            if (end) |e| {
+                const left = e - std.time.milliTimestamp();
+                if (left <= 0) return null;
+                cap = @intCast(@min(left, 250));
+            } else cap = switch (self.*) {
+                .quic => 250,
+                .fd, .pipe => -1,
+            };
+            var fds = [_]std.posix.pollfd{
+                .{ .fd = self.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
+            };
+            const n = std.posix.poll(&fds, self.timeoutMs(cap)) catch return error.Closed;
+            self.service();
+            // A poll that timed out leaves a bounded fd with nothing to read,
+            // and going back to read(2) on it would block past the deadline;
+            // the next pass re-checks the clock instead. QUIC always looks
+            // again — readFrame never blocks there, and the pump just now may
+            // have completed a frame out of bytes that arrived earlier.
+            ready = eager or n != 0;
+            // Once poll HAS announced bytes, the fd arms still block in
+            // readFrame until that frame is whole, even past the deadline —
+            // the price of frames over a stream, same as dial.askOn always
+            // paid; a malformed frame is an error rather than an absent
+            // reply.
+        }
+    }
+    /// Idempotent, and it has to be: a re-dial releases the dead link on
+    /// entry, and an abort then closes the same value again through the
+    /// pump's defer. (The pipe-kill ordering is client.Transport.close's,
+    /// moved: stdin first so the command sees EOF and can wind down its
+    /// remote end, then TERM. kill() waitpid()s internally — no zombie.)
+    pub fn close(self: *Link) void {
+        switch (self.*) {
+            .fd => |fd| {
+                if (fd == -1) return; // already released
+                std.posix.close(fd);
+            },
+            .pipe => |*p| {
+                if (p.child.stdin) |*in| {
+                    in.close();
+                    p.child.stdin = null;
+                }
+                _ = p.child.kill() catch {};
+            },
+            .quic => |*q| {
+                q.qout.deinit(q.alloc);
+                // Closes the UDP socket with it, so pollFd's value must not
+                // be closed again.
+                q.cl.deinit();
+            },
+        }
+        self.* = .{ .fd = -1 };
+    }
+};
+const testing = std.testing;
+fn mkPair() ![2]std.posix.fd_t {
+    var pair: [2]i32 = undefined;
+    try testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
+    return .{ pair[0], pair[1] };
+}
+test "fd link: sendFrame puts one frame on the wire, readFrame takes it back" {
+    const alloc = testing.allocator;
+    const pair = try mkPair();
+    var a: Link = .{ .fd = pair[0] };
+    var b: Link = .{ .fd = pair[1] };
+    defer a.close();
+    defer b.close();
+    try a.sendFrame(.input, "hi");
+    const inc = try b.readFrame(alloc);
+    const frame = inc.frame;
+    defer frame.deinit(alloc);
+    try testing.expectEqual(proto.MsgType.input, frame.type);
+    try testing.expectEqualStrings("hi", frame.payload);
+}
+test "pipe link: the child's stdio IS the transport, and r and w are not interchangeable" {
+    const alloc = testing.allocator;
+    // A real child rather than a socketpair standing in for one, because the
+    // thing under test is which of the child's two fds each direction uses.
+    // `cat` echoes, so a frame that comes back proves sendFrame wrote the
+    // child's STDIN and readFrame read its STDOUT; swap the pair and both
+    // directions address the wrong end of a pipe and fail. A socketpair is
+    // symmetric and would pass either way.
+    var child = std.process.Child.init(&.{"cat"}, alloc);
+    child.stdin_behavior = .Pipe;
+    child.stdout_behavior = .Pipe;
+    try child.spawn();
+    var l: Link = .{ .pipe = .{
+        .child = child,
+        .r = child.stdout.?.handle,
+        .w = child.stdin.?.handle,
+    } };
+    defer l.close();
+    try l.sendFrame(.input, "echo me");
+    const got = (try l.awaitFrame(alloc, .input, 5000, .{})).?;
+    defer got.deinit(alloc);
+    try testing.expectEqualStrings("echo me", got.payload);
+}
+test "close is idempotent on every arm that can be closed twice" {
+    const pair = try mkPair();
+    std.posix.close(pair[1]);
+    var l: Link = .{ .fd = pair[0] };
+    l.close();
+    l.close(); // second close must be a no-op, not EBADF-unreachable
+}
+test "awaitFrame: null sink drops noise frames and returns the match" {
+    const alloc = testing.allocator;
+    const pair = try mkPair();
+    var l: Link = .{ .fd = pair[0] };
+    defer l.close();
+    // Two noise frames BEFORE the answer: the wait must survive a stream,
+    // not a single-frame fixture.
+    try proto.writeFrame(pair[1], .pty_mode, &.{0});
+    try proto.writeFrame(pair[1], .delta, "x");
+    try proto.writeFrame(pair[1], .stats_reply, "ok");
+    std.posix.close(pair[1]);
+    const got = (try l.awaitFrame(alloc, .stats_reply, 1000, .{})).?;
+    defer got.deinit(alloc);
+    try testing.expectEqualStrings("ok", got.payload);
+}
+test "awaitFrame: the sink sees every non-match, in order, and may end the wait" {
+    const alloc = testing.allocator;
+    const Seen = struct {
+        types: [8]proto.MsgType = undefined,
+        n: usize = 0,
+        fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
+            const self: *@This() = @ptrCast(@alignCast(ctx.?));
+            self.types[self.n] = frame.type;
+            self.n += 1;
+            if (frame.type == .exit_status) return error.SessionExited;
+        }
+    };
+    // Leg one: the sink observes and the wait completes.
+    {
+        const pair = try mkPair();
+        var l: Link = .{ .fd = pair[0] };
+        defer l.close();
+        try proto.writeFrame(pair[1], .snapshot, "");
+        try proto.writeFrame(pair[1], .stats_reply, "ok");
+        std.posix.close(pair[1]);
+        var seen: Seen = .{};
+        const got = (try l.awaitFrame(alloc, .stats_reply, 1000, .{ .ctx = &seen, .on = Seen.on })).?;
+        defer got.deinit(alloc);
+        try testing.expectEqual(@as(usize, 1), seen.n);
+        try testing.expectEqual(proto.MsgType.snapshot, seen.types[0]);
+    }
+    // Leg two: the sink's error IS the outcome — an "other" frame that is
+    // an answer, muxa's whole reason for a second copy.
+    {
+        const pair = try mkPair();
+        var l: Link = .{ .fd = pair[0] };
+        defer l.close();
+        try proto.writeFrame(pair[1], .exit_status, &.{7});
+        std.posix.close(pair[1]);
+        var seen: Seen = .{};
+        try testing.expectError(
+            error.SessionExited,
+            l.awaitFrame(alloc, .stats_reply, 1000, .{ .ctx = &seen, .on = Seen.on }),
+        );
+    }
+}
+test "awaitFrame: silence consumes the deadline and returns null; a closed peer is error.Closed" {
+    const alloc = testing.allocator;
+    {
+        const pair = try mkPair();
+        defer std.posix.close(pair[1]);
+        var l: Link = .{ .fd = pair[0] };
+        defer l.close();
+        const t0 = std.time.milliTimestamp();
+        try testing.expect((try l.awaitFrame(alloc, .stats_reply, 100, .{})) == null);
+        try testing.expect(std.time.milliTimestamp() - t0 >= 100);
+    }
+    {
+        const pair = try mkPair();
+        var l: Link = .{ .fd = pair[0] };
+        defer l.close();
+        std.posix.close(pair[1]);
+        try testing.expectError(error.Closed, l.awaitFrame(alloc, .stats_reply, 1000, .{}));
+    }
+}
+test "awaitFrame: no deadline waits out a late reply" {
+    const alloc = testing.allocator;
+    const pair = try mkPair();
+    var l: Link = .{ .fd = pair[0] };
+    defer l.close();
+    const Late = struct {
+        fn run(fd: std.posix.fd_t) void {
+            std.Thread.sleep(150 * std.time.ns_per_ms);
+            proto.writeFrame(fd, .stats_reply, "late") catch {};
+            std.posix.close(fd);
+        }
+    };
+    const t = try std.Thread.spawn(.{}, Late.run, .{pair[1]});
+    defer t.join();
+    const got = (try l.awaitFrame(alloc, .stats_reply, null, .{})).?;
+    defer got.deinit(alloc);
+    try testing.expectEqualStrings("late", got.payload);
+}

Old		New
@@ -167,6 +167,10 @@ const mod_table = [_]ModSpec{
167	// embedder reaches a daemon by linking this and the wire contract	167	// embedder reaches a daemon by linking this and the wire contract
168	// instead of the whole client module.	168	// instead of the whole client module.
169	.{ .name = "dial", .path = "src/dial.zig", .imports = &.{"term"} },	169	.{ .name = "dial", .path = "src/dial.zig", .imports = &.{"term"} },
		170	// The live connection itself — fd, pipe or QUIC — and the one wait-for-a-
		171	// frame loop. `term` for frames, `quic` for the third arm; policy stays
		172	// with the rows that import this one.
		173	.{ .name = "link", .path = "src/link.zig", .link_libc = true, .imports = &.{ "term", "quic" }, .quic_tests = true },
170	// Replays a captured client stdout stream and prints the final grid in	174	// Replays a captured client stdout stream and prints the final grid in
171	// `mux d dump`'s formats — the client half of the M11 render-vs-dump	175	// `mux d dump`'s formats — the client half of the M11 render-vs-dump
172	// convergence check. Imports term so both sides of the diff go through	176	// convergence check. Imports term so both sides of the diff go through
@@ -717,10 +721,10 @@ fn docGate(b: std.Build, target: std.Build.ResolvedTarget, check_step: std.Bui
717	/// of all: it carries every argument parser but muxa's, its mains being	721	/// of all: it carries every argument parser but muxa's, its mains being
718	/// child files — a test that is never built is not a test (decisions.md).	722	/// child files — a test that is never built is not a test (decisions.md).
719	const test_order = [_][]const u8{	723	const test_order = [_][]const u8{
720	"script", "cliflags", "testtmp", "spawn", "dial", "quic",	724	"script", "cliflags", "testtmp", "spawn", "dial", "link",
721	"webhub", "agent", "term", "rawmode", "delaypipe", "render",	725	"quic", "webhub", "agent", "term", "rawmode", "delaypipe",
722	"wsclient", "ptyclient", "pty", "sockpath", "xdg", "proxy",	726	"render", "wsclient", "ptyclient", "pty", "sockpath", "xdg",
723	"wall", "client", "daemon", "mux",	727	"proxy", "wall", "client", "daemon", "mux",
724	};	728	};
725		729
726	comptime {	730	comptime {