a73x

8fe125a8

refactor: Transport wraps the Link, policy stays and mechanics move

a73x   2026-08-31 19:15

Commit message
refactor: Transport wraps the Link, policy stays and mechanics move

Fields become one `link_mod.Link`; `Conn`, `qout` and the per-arm
`alloc` are gone with the union that used to hold them. Send, read,
wait and close delegate; `close` keeps only what the link never knew
about — the handoff ssh's stderr fd.

`roundTrip` retires its hand-rolled poll+read loop for
`Link.awaitFrame`. A caller naming two acceptable answers gets them
through the sink, which is also where `FrameTooLarge` stops being a
closed peer: a daemon whose reply cannot be framed now reads as a
transport fault rather than a refusal.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TWxBL1HBULH1ZwTNzzKTja
build.zig
Old New
@@ -215,7 +215,7 @@ const mod_table = [_]ModSpec{
215 // sits here rather than in either front so the CLI wall and the browser 215 // sits here rather than in either front so the CLI wall and the browser
216 // hub resolve a host line the same way. Nothing here WRITES that file — 216 // hub resolve a host line the same way. Nothing here WRITES that file —
217 // `wall_host.recordHost` and `webhub_main` do. 217 // `wall_host.recordHost` and `webhub_main` do.
218 .{ .name = "client", .path = "src/client/client.zig", .link_libc = true, .imports = &.{ "term", "quic", "xdg", "sockpath", "dial" }, .test_imports = &.{"testtmp"}, .quic_tests = true }, 218 .{ .name = "client", .path = "src/client/client.zig", .link_libc = true, .imports = &.{ "term", "quic", "xdg", "sockpath", "dial", "link" }, .test_imports = &.{"testtmp"}, .quic_tests = true },
219 // ---- the two fronts ---- 219 // ---- the two fronts ----
220 // The browser hub's HTTP/WebSocket decisions: Origin gate, route table, 220 // The browser hub's HTTP/WebSocket decisions: Origin gate, route table,
221 // WS endpoint naming. Assets are injected (the exe root @embedFiles 221 // WS endpoint naming. Assets are injected (the exe root @embedFiles
src/client/client.zig
Old New
@@ -17,6 +17,10 @@ const sockpath = @import("sockpath");
17 // already takes a `dial: ?*handoff.Dial` parameter, and a parameter that 17 // already takes a `dial: ?*handoff.Dial` parameter, and a parameter that
18 // shadows a file-scope declaration does not compile. 18 // shadows a file-scope declaration does not compile.
19 const dialer = @import("dial"); 19 const dialer = @import("dial");
20 // The fd|pipe|quic union and its await loop, under a name that is not
21 // `link`: `Transport.link` is the field this module reads everywhere, and
22 // several local variables are called `link` too.
23 const link_mod = @import("link");
20 24
21 // The client link's public seams: the wall, the hub and the mains reach 25 // The client link's public seams: the wall, the hub and the mains reach
22 // these as client.X — the table stays one row, the files stay children. 26 // these as client.X — the table stays one row, the files stay children.
@@ -135,22 +139,8 @@ pub const PendingSwitch = struct {
135 } 139 }
136 }; 140 };
137 141
138 /// The client's transport: a read fd and a write fd. For a unix socket they 142 /// Callers spell this `client.Incoming`; the union itself is the link row's.
139 /// are one and the same; under `--via` they are the child command's stdout 143 pub const Incoming = link_mod.Incoming;
140 /// and stdin. Nothing below the transport setup knows which it is — that
141 /// blindness is the point: transport is a swap, not a redesign.
142 const Conn = struct { r: std.posix.fd_t, w: std.posix.fd_t };
143
144 /// Three outcomes, not two: QUIC's socket goes readable for acks and half
145 /// frames, so `null` cannot keep the socket path's meaning of "peer gone"
146 /// without making every partial frame reconnect.
147 pub const Incoming = union(enum) {
148 frame: proto.Frame,
149 /// The transport is healthy; there is simply no whole frame yet.
150 incomplete,
151 /// Finished: EOF, a frame torn in half, a connection that died.
152 closed,
153 };
154 144
155 /// A `--via` that died before the first frame carried no connection, and the 145 /// A `--via` that died before the first frame carried no connection, and the
156 /// cause is already on that command's stderr. `session_epoch` tells the two 146 /// cause is already on that command's stderr. `session_epoch` tells the two
@@ -322,32 +312,21 @@ pub const SpecError = error{ MissingKey, OutOfMemory };
322 312
323 /// What the open produced — the live wire. Distinct from `Target` because a 313 /// What the open produced — the live wire. Distinct from `Target` because a
324 /// `hand` recipe yields either a quic or a pipe link, and which one is 314 /// `hand` recipe yields either a quic or a pipe link, and which one is
325 /// decided inside `openHandoff` at runtime. 315 /// decided inside `openHandoff` at runtime. The union is the link row's;
326 pub const Link = union(enum) { 316 /// this name is what the rest of the client spells, `std.meta.Tag(Link)`
327 /// A unix socket: `conn.r == conn.w`, and there is nothing else to own. 317 /// included.
328 fd, 318 pub const Link = link_mod.Link;
329 /// `--via`, and the ssh half of a handoff: the command whose stdio *is*
330 /// the transport.
331 pipe: std.process.Child,
332 /// The connection that IS the transport. `conn` holds the UDP socket in
333 /// `.r` so the poll path needs no special case, and `.w` is -1 because
334 /// there is nothing to write(2) to — bytes go through the stream layer.
335 quic: *quic.Client,
336 };
337 319
338 /// One live connection to a daemon, however it was reached. The point of the 320 /// One live connection to a daemon, however it was reached. The point of the
339 /// struct is that it can be closed and opened again from the same `Target`, 321 /// struct is that it can be closed and opened again from the same `Target`,
340 /// which is what lets a session outlive its transport instead of exiting 322 /// which is what lets a session outlive its transport instead of exiting
341 /// with it. 323 /// with it.
342 pub const Transport = struct { 324 pub const Transport = struct {
343 conn: Conn, 325 /// The wire itself. Everything below the frame — send, read, wait,
344 link: Link = .fd, 326 /// close — is the link's. What stays on this struct is what the link has
345 /// Bytes the QUIC ring would not take yet. Frames are appended whole and 327 /// no business knowing: the second fd a handoff's ssh talks on, the last
346 /// handed over a prefix at a time, so a short accept can never split one 328 /// line it said, and whether that line is relayed onward.
347 /// on the wire — it just means the rest waits here. Keystrokes and 329 link: Link,
348 /// attach frames are all this ever holds.
349 qout: std.ArrayList(u8) = .empty,
350 alloc: std.mem.Allocator = undefined,
351 /// The handoff ssh's stderr for a `.pipe` link born of a handoff; -1 330 /// The handoff ssh's stderr for a `.pipe` link born of a handoff; -1
352 /// for every other link and for `--via`, whose stderr is the user's. 331 /// for every other link and for `--via`, whose stderr is the user's.
353 err_fd: std.posix.fd_t = -1, 332 err_fd: std.posix.fd_t = -1,
@@ -421,10 +400,11 @@ pub const Transport = struct {
421 /// The transport a piped child IS: its stdio is the wire, and the child 400 /// The transport a piped child IS: its stdio is the wire, and the child
422 /// itself is what has to be owned and reaped. 401 /// itself is what has to be owned and reaped.
423 fn pipeTransport(child: std.process.Child) Transport { 402 fn pipeTransport(child: std.process.Child) Transport {
424 return .{ 403 return .{ .link = .{ .pipe = .{
425 .conn = .{ .r = child.stdout.?.handle, .w = child.stdin.?.handle }, 404 .child = child,
426 .link = .{ .pipe = child }, 405 .r = child.stdout.?.handle,
427 }; 406 .w = child.stdin.?.handle,
407 } } };
428 } 408 }
429 409
430 /// `connect` only creates state, so the handshake wait belongs here: 410 /// `connect` only creates state, so the handshake wait belongs here:
@@ -442,11 +422,7 @@ pub const Transport = struct {
442 const cl = try quic.Client.connect(alloc, addr, key, idle_ms); 422 const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
443 errdefer cl.deinit(); 423 errdefer cl.deinit();
444 try waitReady(cl, budget_ms, alloc, carry, abort_fd); 424 try waitReady(cl, budget_ms, alloc, carry, abort_fd);
445 return .{ 425 return .{ .link = .{ .quic = .{ .cl = cl, .alloc = alloc } } };
446 .conn = .{ .r = cl.pollFd(), .w = -1 },
447 .link = .{ .quic = cl },
448 .alloc = alloc,
449 };
450 } 426 }
451 427
452 /// Errors are returned rather than reported here — the caller knows 428 /// Errors are returned rather than reported here — the caller knows
@@ -482,10 +458,7 @@ pub const Transport = struct {
482 }, 458 },
483 .sock => |path| { 459 .sock => |path| {
484 const stream = try dialer.dial(path); 460 const stream = try dialer.dial(path);
485 return .{ 461 return .{ .link = .{ .fd = stream.handle } };
486 .conn = .{ .r = stream.handle, .w = stream.handle },
487 .link = .fd,
488 };
489 }, 462 },
490 } 463 }
491 } 464 }
@@ -661,7 +634,7 @@ pub const Transport = struct {
661 634
662 /// Not where frames come from: QUIC polls UDP, reads frames above it. 635 /// Not where frames come from: QUIC polls UDP, reads frames above it.
663 pub fn pollFd(self: *const Transport) std.posix.fd_t { 636 pub fn pollFd(self: *const Transport) std.posix.fd_t {
664 return self.conn.r; 637 return self.link.pollFd();
665 } 638 }
666 639
667 /// The second fd a handoff's owner polls, or null when there is none. 640 /// The second fd a handoff's owner polls, or null when there is none.
@@ -684,125 +657,61 @@ pub const Transport = struct {
684 /// ssh may need the tty for a password. `qout` is what would otherwise 657 /// ssh may need the tty for a password. `qout` is what would otherwise
685 /// cross: two threads on one non-thread-safe arena. 658 /// cross: two threads on one non-thread-safe arena.
686 pub fn adopt(self: *Transport, alloc: std.mem.Allocator) void { 659 pub fn adopt(self: *Transport, alloc: std.mem.Allocator) void {
687 if (self.link != .quic) return; 660 const q = switch (self.link) {
688 std.debug.assert(self.qout.items.len == 0); 661 .quic => |*q| q,
689 self.qout.deinit(self.alloc); 662 .fd, .pipe => return,
690 self.qout = .empty; 663 };
691 self.alloc = alloc; 664 std.debug.assert(q.qout.items.len == 0);
665 q.qout.deinit(q.alloc);
666 q.qout = .empty;
667 q.alloc = alloc;
692 } 668 }
693 669
694 pub fn writeFrame(self: *Transport, t: proto.MsgType, payload: []const u8) !void { 670 pub fn writeFrame(self: *Transport, t: proto.MsgType, payload: []const u8) !void {
695 switch (self.link) { 671 return self.link.sendFrame(t, payload);
696 .quic => {
697 // Appended whole, handed over a prefix at a time. A frame can
698 // therefore never be split across a refusal — the ring takes
699 // what it takes and the remainder is offered again next pass.
700 try proto.appendFrame(&self.qout, self.alloc, t, payload);
701 self.flushQuic();
702 },
703 .fd, .pipe => return proto.writeFrame(self.conn.w, t, payload),
704 }
705 } 672 }
706 673
707 /// Unconditional: a QUIC connection's timers are the only thing that 674 /// Unconditional: a QUIC connection's timers are the only thing that
708 /// notices a peer which stopped answering. 675 /// notices a peer which stopped answering.
709 pub fn service(self: *Transport) void { 676 pub fn service(self: *Transport) void {
710 switch (self.link) { 677 self.link.service();
711 .quic => |cl| {
712 cl.pump();
713 self.flushQuic();
714 },
715 .fd, .pipe => {},
716 }
717 } 678 }
718 679
719 /// Folds ngtcp2's next deadline in, so retransmits and idle timeouts 680 /// Folds ngtcp2's next deadline in, so retransmits and idle timeouts
720 /// happen on time without a second timer. 681 /// happen on time without a second timer.
721 pub fn timeoutMs(self: *Transport, cap_ms: i32) i32 { 682 pub fn timeoutMs(self: *Transport, cap_ms: i32) i32 {
722 return switch (self.link) { 683 return self.link.timeoutMs(cap_ms);
723 .quic => |cl| cl.timeoutMs(cap_ms),
724 .fd, .pipe => cap_ms,
725 };
726 } 684 }
727 685
728 /// Offer the outbound queue to the ring again. Called after every write 686 /// Offer the outbound queue to the ring again. Called after every write
729 /// and on every service pass, because the room to accept comes from 687 /// and on every service pass, because the room to accept comes from
730 /// acknowledgements, which arrive on their own schedule. 688 /// acknowledgements, which arrive on their own schedule.
731 pub fn flushQuic(self: *Transport) void { 689 pub fn flushQuic(self: *Transport) void {
732 const cl = switch (self.link) { 690 self.link.flushQuic();
733 .quic => |cl| cl,
734 .fd, .pipe => return,
735 };
736 if (self.qout.items.len == 0) return;
737 const n = cl.send(self.qout.items);
738 if (n == 0) return;
739 self.qout.replaceRangeAssumeCapacity(0, n, &.{});
740 } 691 }
741 692
742 /// The next whole frame, if there is one. See `Incoming` for why a 693 /// The next whole frame, if there is one. See `Incoming` for why a
743 /// missing frame is not automatically a dead transport. 694 /// missing frame is not automatically a dead transport, and
695 /// `link.Link.readFrame` for the two faults that stay errors.
744 pub fn readFrame(self: *Transport, alloc: std.mem.Allocator) !Incoming { 696 pub fn readFrame(self: *Transport, alloc: std.mem.Allocator) !Incoming {
745 switch (self.link) { 697 return self.link.readFrame(alloc);
746 .quic => |cl| {
747 // Death is checked after the pump, so bytes that arrived in the
748 // same pass as the close are still delivered before the tear.
749 const got = proto.takeFrame(alloc, &cl.in) catch |err| switch (err) {
750 // Not a transport event, and it stays loud.
751 error.OutOfMemory => return err,
752 else => return .closed,
753 };
754 if (got) |frame| return .{ .frame = frame };
755 return if (cl.dead) .closed else .incomplete;
756 },
757 .fd, .pipe => {
758 const frame = (proto.readFrame(alloc, self.conn.r) catch |err| switch (err) {
759 // Not a transport event, and it stays loud.
760 error.OutOfMemory => return err,
761 else => return .closed,
762 }) orelse return .closed;
763 return .{ .frame = frame };
764 },
765 }
766 } 698 }
767 699
768 /// Idempotent, and it has to be: a re-dial releases the dead transport on 700 /// Idempotent, and it has to be: a re-dial releases the dead transport on
769 /// entry, and an abort then closes the same value again through the pump's 701 /// entry, and an abort then closes the same value again through the pump's
770 /// `defer`. A second `close(2)` on a stale fd is EBADF, which `std.posix` 702 /// `defer`. A second `close(2)` on a stale fd is EBADF, which `std.posix`
771 /// maps to `unreachable` — a panic that `--sock` does not hide. 703 /// maps to `unreachable` — a panic that `--sock` does not hide.
704 ///
705 /// Only the stderr fd is closed here: it is the one thing the link never
706 /// knew about, and `openHandoff` took it off the Child precisely so the
707 /// kill inside `Link.close` would leave it alone.
772 pub fn close(self: *Transport) void { 708 pub fn close(self: *Transport) void {
773 if (self.conn.r == -1) return; // already released 709 if (self.link == .fd and self.link.fd == -1) return; // already released
774 defer self.conn = .{ .r = -1, .w = -1 }; 710 if (self.err_fd >= 0) {
775 // The link is reset to `.fd` with the fds, so nothing that outlives 711 std.posix.close(self.err_fd);
776 // this call holds a freed client or a reaped child behind a tag that 712 self.err_fd = -1;
777 // says otherwise.
778 defer self.link = .fd;
779 switch (self.link) {
780 .quic => |cl| {
781 self.qout.deinit(self.alloc);
782 self.qout = .empty;
783 // Closes the UDP socket with it, so the fd this returns
784 // through `conn.r` must not be closed again.
785 cl.deinit();
786 },
787 .pipe => |*c| {
788 // Close stdin first so the command sees EOF and can wind down
789 // its remote end cleanly; then TERM it. kill() waitpid()s
790 // internally, so this also reaps — no zombie is left behind.
791 if (c.stdin) |*in| {
792 in.close();
793 c.stdin = null;
794 }
795 // Ours, not the child's: `openHandoff` took stderr off the
796 // Child so `kill` would leave it alone, so nothing else
797 // will close it.
798 if (self.err_fd >= 0) {
799 std.posix.close(self.err_fd);
800 self.err_fd = -1;
801 }
802 _ = c.kill() catch {};
803 },
804 .fd => std.posix.close(self.conn.r),
805 } 713 }
714 self.link.close();
806 } 715 }
807 }; 716 };
808 717
@@ -1173,32 +1082,53 @@ pub const birth_rows: u16 = 24;
1173 // the tile reads [refused], which is where it stood before. 1082 // the tile reads [refused], which is where it stood before.
1174 const birth_budget_ms: i64 = 3000; 1083 const birth_budget_ms: i64 = 3000;
1175 1084
1176 /// One bounded wait on a side connection: false means loop again without 1085 /// A caller that named more than one acceptable answer, waiting on a
1177 /// reading, `error.Timeout` means the budget is spent. No EINTR arm — 1086 /// primitive that takes exactly one. `Link.awaitFrame` waits for `want[0]`
1178 /// `std.posix.poll` retries INTR itself, and its error set holds only failures 1087 /// and hands every other frame to this sink, which recognises the rest and
1179 /// the fd never recovers from. 1088 /// ends the wait by erroring — the escape the sink contract exists for.
1180 fn awaitFrames(tr: *Transport, deadline: i64) !bool { 1089 /// The frame is COPIED because awaitFrame frees what it lends the sink.
1181 const left = deadline - std.time.milliTimestamp(); 1090 const AltWant = struct {
1182 if (left <= 0) return error.Timeout; 1091 alloc: std.mem.Allocator,
1183 var fds = [_]std.posix.pollfd{ 1092 /// `want[1..]`: the types awaitFrame is not itself watching for.
1184 .{ .fd = tr.pollFd(), .events = std.posix.POLL.IN, .revents = 0 }, 1093 rest: []const proto.MsgType,
1185 }; 1094 got: ?proto.Frame = null,
1186 _ = std.posix.poll(&fds, tr.timeoutMs(@intCast(@min(left, 100)))) catch 1095
1187 return error.Transport; 1096 /// Never leaves `roundTrip`, which turns it back into the frame.
1188 tr.service(); 1097 const Answer = error.AltAnswer;
1189 // Over QUIC a frame can arrive with the socket never going readable. 1098
1190 return fds[0].revents != 0 or tr.link == .quic; 1099 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
1191 } 1100 const self: *AltWant = @ptrCast(@alignCast(ctx.?));
1101 for (self.rest) |w| {
1102 if (frame.type != w) continue;
1103 self.got = .{
1104 .type = frame.type,
1105 .payload = try self.alloc.dupe(u8, frame.payload),
1106 };
1107 return Answer;
1108 }
1109 // Narration the caller never asked for; awaitFrame frees it.
1110 }
1111 };
1192 1112
1193 /// One question on a side connection: send `req`, then hand back the first 1113 /// One question on a side connection: send `req`, then hand back the first
1194 /// frame whose type is one the caller named, deinit-ing every other frame 1114 /// frame whose type is one the caller named, deinit-ing every other frame
1195 /// the daemon says on the way. The frame returned is the caller's to deinit. 1115 /// the daemon says on the way. The frame returned is the caller's to deinit.
1196 /// 1116 ///
1197 /// The errors are thin on purpose — `Timeout` for a budget spent, `Closed` 1117 /// The errors are thin on purpose — `Timeout` for a budget spent, `Closed`
1198 /// for a peer that hung up before answering, `Transport` for a poll that 1118 /// for a wait that ended without an answer, `FrameTooLarge` for a peer that
1199 /// failed, `OutOfMemory` for this machine's allocator. What one MEANS is 1119 /// spoke and got the framing wrong, `OutOfMemory` for this machine's
1200 /// the caller's to say: the same close is a refusal to a birth and a box 1120 /// allocator. What one MEANS is the caller's to say: the same close is a
1201 /// that is down to a poll. 1121 /// refusal to a birth and a box that is down to a poll. `FrameTooLarge` is
1122 /// deliberately NOT folded into `Closed` — a daemon whose reply cannot be
1123 /// framed is a different fact from one that is not there, and a birth
1124 /// reports the second as a refusal.
1125 ///
1126 /// `Closed` is also what a failed `poll(2)` inside the wait becomes, since
1127 /// `Link.awaitFrame` owns the poll and has one word for "the wait cannot go
1128 /// on". That is a widening: this used to answer `Transport` there. The two
1129 /// failures poll can report are out of kernel memory and a bug, so the
1130 /// mislabel costs a "refused" instead of a "transport error" in a case
1131 /// where nothing on the box is working anyway.
1202 fn roundTrip( 1132 fn roundTrip(
1203 tr: *Transport, 1133 tr: *Transport,
1204 alloc: std.mem.Allocator, 1134 alloc: std.mem.Allocator,
@@ -1207,25 +1137,30 @@ fn roundTrip(
1207 want: []const proto.MsgType, 1137 want: []const proto.MsgType,
1208 deadline: i64, 1138 deadline: i64,
1209 ) !proto.Frame { 1139 ) !proto.Frame {
1140 std.debug.assert(want.len != 0); // a question with no acceptable answer
1210 try tr.writeFrame(req, payload); 1141 try tr.writeFrame(req, payload);
1211 while (true) { 1142 const left = deadline - std.time.milliTimestamp();
1212 if (!try awaitFrames(tr, deadline)) continue; 1143 if (left <= 0) return error.Timeout;
1213 while (true) { 1144 var alt: AltWant = .{ .alloc = alloc, .rest = want[1..] };
1214 // `readFrame` turns every wire fault into `.closed`, so the one 1145 // The wait takes milliseconds as a u32. Every budget in the tree is
1215 // error it can still raise is this machine's own allocator. 1146 // seconds at most, and clamping is what a caller that passed more would
1216 switch (try tr.readFrame(alloc)) { 1147 // have meant anyway — a cast would panic on it.
1217 .incomplete => break, 1148 const ms: u32 = @intCast(@min(left, std.math.maxInt(u32)));
1218 .closed => return error.Closed, 1149 const f = tr.link.awaitFrame(alloc, want[0], ms, .{
1219 .frame => |f| { 1150 .ctx = &alt,
1220 for (want) |w| if (f.type == w) return f; 1151 .on = AltWant.on,
1221 f.deinit(alloc); 1152 }) catch |e| switch (e) {
1222 }, 1153 AltWant.Answer => return alt.got.?,
1223 } 1154 error.Closed => return error.Closed,
1224 // One readable event is one frame on a socket link, and a 1155 error.OutOfMemory => return error.OutOfMemory,
1225 // second read would block until the daemon spoke again. 1156 error.FrameTooLarge => return error.FrameTooLarge,
1226 if (tr.link != .quic) break; 1157 // Nothing reaches here today — awaitFrame raises only the three
1227 } 1158 // above and whatever the sink returns — but the arm keeps this
1228 } 1159 // function's error set the thin one its callers switch on rather
1160 // than letting the sink's `anyerror` widen it.
1161 else => return error.Transport,
1162 };
1163 return f orelse error.Timeout;
1229 } 1164 }
1230 1165
1231 /// Creates `name` on `target` over a connection of its own, then leaves. 1166 /// Creates `name` on `target` over a connection of its own, then leaves.
@@ -1587,7 +1522,8 @@ test "Transport.close is idempotent: the abort path closes what a re-dial alread
1587 transport.close(); 1522 transport.close();
1588 transport.close(); 1523 transport.close();
1589 transport.close(); 1524 transport.close();
1590 try std.testing.expectEqual(@as(std.posix.fd_t, -1), transport.conn.r); 1525 try std.testing.expect(transport.link == .fd);
1526 try std.testing.expectEqual(@as(std.posix.fd_t, -1), transport.link.fd);
1591 } 1527 }
1592 1528
1593 test "connectAgent: a live socket connects, a dead path returns null" { 1529 test "connectAgent: a live socket connects, a dead path returns null" {
@@ -1647,7 +1583,7 @@ test "Transport.open: a --via target yields a pipe, a --sock target an fd" {
1647 1583
1648 // Read before the close, not after: kill() reaps, and the child value 1584 // Read before the close, not after: kill() reaps, and the child value
1649 // goes with the link when close resets it. 1585 // goes with the link when close resets it.
1650 const pid = v.link.pipe.id; 1586 const pid = v.link.pipe.child.id;
1651 v.close(); 1587 v.close();
1652 // Observed, not assumed: close() owns the reaping (Child.kill waitpid()s 1588 // Observed, not assumed: close() owns the reaping (Child.kill waitpid()s
1653 // internally), so the pid must be gone rather than a zombie — signal 0 1589 // internally), so the pid must be gone rather than a zombie — signal 0
@@ -1717,7 +1653,7 @@ test "--via: the words reach the program verbatim — no shell splits, expands o
1717 var buf: [512]u8 = undefined; 1653 var buf: [512]u8 = undefined;
1718 var got: usize = 0; 1654 var got: usize = 0;
1719 while (true) { 1655 while (true) {
1720 const n = std.posix.read(v.conn.r, buf[got..]) catch 0; 1656 const n = std.posix.read(v.link.pipe.r, buf[got..]) catch 0;
1721 if (n == 0) break; 1657 if (n == 0) break;
1722 got += n; 1658 got += n;
1723 } 1659 }
src/link.zig
Old New
@@ -107,7 +107,7 @@ pub const Link = union(enum) {
107 /// Offer the outbound queue to the ring again. After every write and on 107 /// Offer the outbound queue to the ring again. After every write and on
108 /// every service pass, because the room to accept comes from 108 /// every service pass, because the room to accept comes from
109 /// acknowledgements, which arrive on their own schedule. 109 /// acknowledgements, which arrive on their own schedule.
110 fn flushQuic(self: *Link) void { 110 pub fn flushQuic(self: *Link) void {
111 const q = switch (self.*) { 111 const q = switch (self.*) {
112 .quic => |*q| q, 112 .quic => |*q| q,
113 .fd, .pipe => return, 113 .fd, .pipe => return,
src/tui/wall_test_pump.zig
Old New
@@ -304,13 +304,14 @@ test "agent channels: slots fill in order, a full table refuses, and a close is
304 try std.testing.expectEqual(@as(?usize, 1), wall_pump.findLocal(&locals, 11)); 304 try std.testing.expectEqual(@as(?usize, 1), wall_pump.findLocal(&locals, 11));
305 try std.testing.expectEqual(@as(?usize, null), wall_pump.findLocal(&locals, 99)); 305 try std.testing.expectEqual(@as(?usize, null), wall_pump.findLocal(&locals, 99));
306 306
307 // A pipe for the link, which is what a `.fd` transport writes to. The 307 // A pipe for the link: the transport holds the WRITE end, since these
308 // frame that comes back out of it is the daemon's only notice that this 308 // tests only ever send, and reads the pipe themselves. The frame that
309 // end hung up on the channel. 309 // comes back out of it is the daemon's only notice that this end hung
310 // up on the channel.
310 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 311 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
311 defer std.posix.close(link[0]); 312 defer std.posix.close(link[0]);
312 defer std.posix.close(link[1]); 313 defer std.posix.close(link[1]);
313 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 314 var transport: client.Transport = .{ .link = .{ .fd = link[1] } };
314 315
315 wall_pump.closeLocal(&locals, 1, &transport); 316 wall_pump.closeLocal(&locals, 1, &transport);
316 try std.testing.expectEqual(@as(?AgentLocal, null), locals[1]); 317 try std.testing.expectEqual(@as(?AgentLocal, null), locals[1]);
@@ -339,7 +340,7 @@ test "agent channels: an oversize frame hangs the channel up, a full one lands"
339 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 340 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
340 defer std.posix.close(link[0]); 341 defer std.posix.close(link[0]);
341 defer std.posix.close(link[1]); 342 defer std.posix.close(link[1]);
342 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 343 var transport: client.Transport = .{ .link = .{ .fd = link[1] } };
343 344
344 // The ordinary case first, so the refusal below is a refusal and not a 345 // The ordinary case first, so the refusal below is a refusal and not a
345 // delivery path that never worked. 346 // delivery path that never worked.
@@ -412,7 +413,7 @@ test "agent forwarding is per tile: no -A offers nothing and opens nothing" {
412 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 413 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
413 defer std.posix.close(link[0]); 414 defer std.posix.close(link[0]);
414 defer std.posix.close(link[1]); 415 defer std.posix.close(link[1]);
415 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 416 var transport: client.Transport = .{ .link = .{ .fd = link[1] } };
416 417
417 try wall_pump.sendAttach(&t, &transport, 0, 0); 418 try wall_pump.sendAttach(&t, &transport, 0, 0);
418 const first = (try proto.readFrame(std.testing.allocator, link[0])) orelse 419 const first = (try proto.readFrame(std.testing.allocator, link[0])) orelse
@@ -457,7 +458,7 @@ test "a view tile's attach makes no size claim; a tile the user asked for does"
457 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 458 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
458 defer std.posix.close(link[0]); 459 defer std.posix.close(link[0]);
459 defer std.posix.close(link[1]); 460 defer std.posix.close(link[1]);
460 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 461 var transport: client.Transport = .{ .link = .{ .fd = link[1] } };
461 462
462 // A view tile: 0x0 on the wire, and the rect owed to the doorbell. 463 // A view tile: 0x0 on the wire, and the rect owed to the doorbell.
463 t.creates = false; 464 t.creates = false;
@@ -509,7 +510,7 @@ test "a newborn tile's first claim is its real stripe, not the placeholder" {
509 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 510 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
510 defer std.posix.close(link[0]); 511 defer std.posix.close(link[0]);
511 defer std.posix.close(link[1]); 512 defer std.posix.close(link[1]);
512 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 513 var transport: client.Transport = .{ .link = .{ .fd = link[1] } };
513 try wall_pump.sendAttach(&t, &transport, 0, 0); 514 try wall_pump.sendAttach(&t, &transport, 0, 0);
514 const fr = (try proto.readFrame(std.testing.allocator, link[0])) orelse 515 const fr = (try proto.readFrame(std.testing.allocator, link[0])) orelse
515 return error.NoAttach; 516 return error.NoAttach;