a73x

3bb2f637

refactor: one owner for "send one frame, wait for one reply"

a73x   2026-08-29 10:01

Commit message
refactor: one owner for "send one frame, wait for one reply"

`oneShotQuery`, `askEndpointPort`, `confirmServing` and `upgradeCmd` each
wrote the same round trip: write the request, read frames until the one
type they want, give up on EOF. `askOnce` owns it now — including the
half-frame hazard note, which travelled with the read loop it describes.

The connect stays with each caller, and so does a failed send: the four
have four different things to say about a socket nobody serves, and
`upgradeCmd` alone can tell "closed before the request landed" from "this
daemon predates upgrade".

BEHAVIOUR UNCHANGED, deliberately. The ledger allowed `oneShotQuery` to
gain a deadline (`mux d dump` on a wedged daemon: hang -> rc 1); reading
the code found the decision that would reverse — the old comment on
`oneShotQuery` said it blocks so that a wedge SHOWS, rather than reading
as the same rc 1 an absent socket gets. So `deadline_ms` is optional and
`oneShotQuery` passes null. Pinned before the fold by "askOnce: a deadline
gives up on silence, and no deadline waits out a late reply", which was
RED (undeclared askOnce), then green, then mutation-checked: forcing a
10 ms bound on the unbounded arm kills it.

`upgradeCmd`'s empty-payload reply lands on the same "no reply" line it
reached before by being skipped until the deadline. e2e_01_boot,
e2e_04_handoff and e2e_14_upgrade all green.

Net −4 product lines: the shared round trip is 33 lines and the four
bodies it replaced were 37. The win is one owner, not the count.
src/cli/main.zig
Old New
@@ -643,9 +643,55 @@ fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {
643 return try srv.run(); 643 return try srv.run();
644 } 644 }
645 645
646 /// `askEndpointPort` is not folded in here: it waits under a deadline 646 /// Whether a frame of the wanted type with NO payload is the answer.
647 /// for a daemon too old for `endpoint_req`. This one blocks, so a 647 /// `mux d upgrade` reads a status byte out of its reply, so an empty one is
648 /// wedge shows. 648 /// a peer that cannot answer, and the old loop kept waiting for a real
649 /// answer to the deadline. `dump` and `stats` print whatever arrived, empty
650 /// included: for them the frame IS the answer.
651 const EmptyPayload = enum { is_the_answer, keeps_waiting };
652
653 /// Send one frame, then read until `want` arrives; null is every way of
654 /// not getting it that names nothing; a null `deadline_ms` blocks.
655 fn askOnce(
656 alloc: std.mem.Allocator,
657 fd: std.posix.fd_t,
658 req: proto.MsgType,
659 payload: []const u8,
660 want: proto.MsgType,
661 deadline_ms: ?u32,
662 empty: EmptyPayload,
663 ) !?proto.Frame {
664 // Named apart from the read errors below, because `mux d upgrade` words
665 // the two differently: a request that never landed names the socket.
666 proto.writeFrame(fd, req, payload) catch return error.RequestNotSent;
667 const deadline: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null;
668 while (true) {
669 if (deadline) |end| {
670 const left = end - std.time.milliTimestamp();
671 if (left <= 0) return null;
672 var fds = [_]std.posix.pollfd{
673 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
674 };
675 if ((std.posix.poll(&fds, @intCast(left)) catch return null) == 0) return null;
676 if (fds[0].revents == 0) continue;
677 }
678 // Blocking, and bounded only by the poll that said bytes are here:
679 // a daemon that wrote half a seven-byte frame and then stopped
680 // would hold this past the deadline. That daemon is this same
681 // binary on the same box, writing that frame in one call — the
682 // hazard is worth naming and not worth a state machine. A frame it
683 // CAN read the length of but not the rest of is an error, not a
684 // null: `mux d dump` prints that name, where a null would spell a
685 // broken daemon exactly like an absent one.
686 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
687 if (frame.type == want and
688 !(empty == .keeps_waiting and frame.payload.len == 0)) return frame;
689 frame.deinit(alloc);
690 }
691 }
692
693 /// Unbounded on purpose: a daemon that has stopped answering must show as
694 /// the hang it is, not as the rc 1 an absent socket gets.
649 fn oneShotQuery( 695 fn oneShotQuery(
650 alloc: std.mem.Allocator, 696 alloc: std.mem.Allocator,
651 sock_path: []const u8, 697 sock_path: []const u8,
@@ -663,15 +709,11 @@ fn oneShotQuery(
663 }; 709 };
664 defer stream.close(); 710 defer stream.close();
665 711
666 try proto.writeFrame(stream.handle, req, req_payload); 712 const frame = (try askOnce(alloc, stream.handle, req, req_payload, want, null, .is_the_answer)) orelse return 1;
667 while (try proto.readFrame(alloc, stream.handle)) |frame| { 713 defer frame.deinit(alloc);
668 defer frame.deinit(alloc); 714 try proto.writeAllFd(std.posix.STDOUT_FILENO, frame.payload);
669 if (frame.type != want) continue; 715 try proto.writeAllFd(std.posix.STDOUT_FILENO, "\n");
670 try proto.writeAllFd(std.posix.STDOUT_FILENO, frame.payload); 716 return 0;
671 try proto.writeAllFd(std.posix.STDOUT_FILENO, "\n");
672 return 0;
673 }
674 return 1;
675 } 717 }
676 718
677 fn dump(alloc: std.mem.Allocator, sock_path: []const u8, vt_mode: bool, session: []const u8) !u8 { 719 fn dump(alloc: std.mem.Allocator, sock_path: []const u8, vt_mode: bool, session: []const u8) !u8 {
@@ -811,27 +853,26 @@ fn upgradeCmd(alloc: std.mem.Allocator, sock_path: []const u8, allow_same: bool)
811 std.debug.print("mux d upgrade: cannot name {s} in a request\n", .{exe}); 853 std.debug.print("mux d upgrade: cannot name {s} in a request\n", .{exe});
812 return 1; 854 return 1;
813 }; 855 };
814 proto.writeFrame(stream.handle, .upgrade_req, payload) catch {
815 std.debug.print("mux d upgrade: {s} closed before the request landed\n", .{sock_path});
816 return 1;
817 };
818
819 // Bounded, because a daemon older than this feature drops an unknown 856 // Bounded, because a daemon older than this feature drops an unknown
820 // frame without a word: the expiry is a diagnosis, not a timeout. 857 // frame without a word: the expiry is a diagnosis, not a timeout. So is
821 const deadline_ms: i64 = 5000; 858 // EOF — an older daemon that drops the connection over a frame it
822 const t0 = std.time.milliTimestamp(); 859 // cannot read reaches the same conclusion silence does.
823 while (true) { 860 const reply = askOnce(alloc, stream.handle, .upgrade_req, payload, .upgrade_reply, 5000, .keeps_waiting) catch |e| ask: {
824 const left = deadline_ms - (std.time.milliTimestamp() - t0); 861 if (e == error.RequestNotSent) {
825 if (left <= 0) break; 862 std.debug.print("mux d upgrade: {s} closed before the request landed\n", .{sock_path});
826 var pfd = [_]std.posix.pollfd{ 863 return 1;
827 .{ .fd = stream.handle, .events = std.posix.POLL.IN, .revents = 0 }, 864 }
828 }; 865 // A reply this side cannot read is no answer, which is where the
829 if ((std.posix.poll(&pfd, @intCast(left)) catch break) == 0) break; 866 // expiry and the EOF land too.
830 // EOF: an older daemon that drops the connection over a frame it 867 break :ask null;
831 // cannot read reaches the same conclusion as silence does. 868 };
832 const frame = (proto.readFrame(alloc, stream.handle) catch break) orelse break; 869 // `.keeps_waiting` above means an empty `upgrade_reply` never reaches
870 // here: the wait spends the rest of its deadline on a real answer and
871 // expires into the no-reply line below. This arm is the bounds check
872 // `payload[0]` needs, and it lands where silence lands.
873 if (reply) |frame| skip: {
833 defer frame.deinit(alloc); 874 defer frame.deinit(alloc);
834 if (frame.type != .upgrade_reply or frame.payload.len == 0) continue; 875 if (frame.payload.len == 0) break :skip;
835 if (frame.payload[0] != 0) { 876 if (frame.payload[0] != 0) {
836 // The daemon's words, verbatim: it is the side that knows which 877 // The daemon's words, verbatim: it is the side that knows which
837 // check failed, and paraphrasing here would lose the versions. 878 // check failed, and paraphrasing here would lose the versions.
@@ -876,19 +917,13 @@ fn confirmServing(alloc: std.mem.Allocator, sock_path: []const u8) u8 {
876 return 1; 917 return 1;
877 }; 918 };
878 defer stream.close(); 919 defer stream.close();
879 proto.writeFrame(stream.handle, .stats_req, "") catch return 1;
880 920
881 const deadline_ms: i64 = 5000; 921 const deadline_ms: u32 = 5000;
882 var pfd = [_]std.posix.pollfd{ 922 if (askOnce(alloc, stream.handle, .stats_req, "", .stats_reply, deadline_ms, .is_the_answer) catch return 1) |frame| {
883 .{ .fd = stream.handle, .events = std.posix.POLL.IN, .revents = 0 }, 923 frame.deinit(alloc);
884 }; 924 return 0;
885 if ((std.posix.poll(&pfd, deadline_ms) catch 0) > 0) {
886 if (proto.readFrame(alloc, stream.handle) catch null) |frame| {
887 defer frame.deinit(alloc);
888 if (frame.type == .stats_reply) return 0;
889 }
890 } 925 }
891 const secs = @divTrunc(deadline_ms, 1000); 926 const secs = deadline_ms / 1000;
892 var hint: [log_hint_len]u8 = undefined; 927 var hint: [log_hint_len]u8 = undefined;
893 std.debug.print( 928 std.debug.print(
894 "mux d upgrade: exec'd, but {s} has not answered in {d}s{s}\n", 929 "mux d upgrade: exec'd, but {s} has not answered in {d}s{s}\n",
@@ -1117,45 +1152,27 @@ fn reportKeyRefusal(path: []const u8, err: anyerror) void {
1117 ); 1152 );
1118 } 1153 }
1119 1154
1120 /// One observer round-trip: `endpoint_req`, then a bounded wait for the 1155 /// One observer round-trip: `endpoint_req`, then a bounded wait. 0 is
1121 /// reply. 0 is every failure — nothing listening, no reply, a reply that 1156 /// every failure — nothing listening, no reply, a reply we do not
1122 /// is not one — because the caller does the same thing with all of them. 1157 /// understand — because the caller treats them alike.
1123 ///
1124 /// The bound is what converts an old daemon's silence into the
1125 /// announce-none path instead of a hang: `MsgType` is non-exhaustive, so a
1126 /// binary from before the verb existed simply ignores it. It reuses
1127 /// `spawn.start_deadline_ms` rather than naming a number of its own — the
1128 /// same "how long can a local daemon reasonably take" the ensure above
1129 /// already waited on.
1130 /// 1158 ///
1131 /// The connect is outside the loop, so the ordinary failure — nothing on 1159 /// The bound converts an old daemon's silence (`MsgType` is
1132 /// the socket — is a refusal in microseconds and never touches the budget. 1160 /// non-exhaustive, so a binary from before the verb ignores it) into the
1161 /// announce-none path instead of a hang.
1133 fn askEndpointPort(alloc: std.mem.Allocator, sock_path: []const u8) u16 { 1162 fn askEndpointPort(alloc: std.mem.Allocator, sock_path: []const u8) u16 {
1134 const stream = std.net.connectUnixSocket(sock_path) catch return 0; 1163 const stream = std.net.connectUnixSocket(sock_path) catch return 0;
1135 defer stream.close(); 1164 defer stream.close();
1136 proto.writeFrame(stream.handle, .endpoint_req, "") catch return 0; 1165 const frame = (askOnce(
1137 1166 alloc,
1138 const deadline = std.time.milliTimestamp() + spawn.start_deadline_ms; 1167 stream.handle,
1139 while (std.time.milliTimestamp() < deadline) { 1168 .endpoint_req,
1140 var fds = [_]std.posix.pollfd{ 1169 "",
1141 .{ .fd = stream.handle, .events = std.posix.POLL.IN, .revents = 0 }, 1170 .endpoint_reply,
1142 }; 1171 spawn.start_deadline_ms,
1143 const remaining: i32 = @intCast(@max(1, deadline - std.time.milliTimestamp())); 1172 .is_the_answer,
1144 _ = std.posix.poll(&fds, remaining) catch return 0; 1173 ) catch null) orelse return 0;
1145 if (fds[0].revents == 0) continue; 1174 defer frame.deinit(alloc);
1146 // Blocking, and bounded only by the poll that said bytes are here: 1175 return proto.decodeEndpointReply(frame.payload) catch 0;
1147 // a daemon that wrote half a seven-byte frame and then stopped
1148 // would hold this past the deadline. That daemon is this same
1149 // binary on the same box, writing that frame in one call — the
1150 // hazard is worth naming and not worth a state machine.
1151 const f = (proto.readFrame(alloc, stream.handle) catch return 0) orelse return 0;
1152 defer f.deinit(alloc);
1153 if (f.type != .endpoint_reply) continue;
1154 // A reply of the wrong length is one we do not understand, which
1155 // lands in the same place as no reply at all.
1156 return proto.decodeEndpointReply(f.payload) catch 0;
1157 }
1158 return 0;
1159 } 1176 }
1160 1177
1161 /// `mux d start` = ensureDaemon under an explicit flag. Everything after 1178 /// `mux d start` = ensureDaemon under an explicit flag. Everything after
@@ -1658,6 +1675,96 @@ test "endpointCmd: a box with no daemon is refused, never started — the wall p
1658 try std.testing.expectEqual(@as(u64, 0), (try out.stat()).size); 1675 try std.testing.expectEqual(@as(u64, 0), (try out.stat()).size);
1659 } 1676 }
1660 1677
1678 test "askOnce: a deadline gives up on silence, and no deadline waits out a late reply" {
1679 const alloc = std.testing.allocator;
1680
1681 // A socket pair stands in for the daemon: this end asks, the test end
1682 // decides whether anything answers and when.
1683 //
1684 // A `zig build test` that prints nothing and never returns is this
1685 // block: the deadline is being ignored, and there is no reply coming.
1686 {
1687 var pair: [2]i32 = undefined;
1688 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1689 defer std.posix.close(pair[0]);
1690 defer std.posix.close(pair[1]);
1691 const t0 = std.time.milliTimestamp();
1692 try std.testing.expect((try askOnce(alloc, pair[0], .stats_req, "", .stats_reply, 100, .is_the_answer)) == null);
1693 // Waited the budget out rather than reading the silence as an
1694 // answer: the bounded callers turn "no reply" into a diagnosis
1695 // (an old daemon, a failed exec), which a fast null would fake.
1696 try std.testing.expect(std.time.milliTimestamp() - t0 >= 100);
1697 }
1698
1699 // The unbounded claim: a daemon that is slow to answer is WAITED for
1700 // (`oneShotQuery` owns the why).
1701 {
1702 var pair: [2]i32 = undefined;
1703 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1704 defer std.posix.close(pair[0]);
1705 const Late = struct {
1706 fn run(fd: std.posix.fd_t) void {
1707 std.Thread.sleep(150 * std.time.ns_per_ms);
1708 // Another type first: the wait is for the type asked for,
1709 // not for the next thing the daemon happens to say.
1710 proto.writeFrame(fd, .stats_req, "") catch {};
1711 proto.writeFrame(fd, .stats_reply, "late") catch {};
1712 std.posix.close(fd);
1713 }
1714 };
1715 const th = try std.Thread.spawn(.{}, Late.run, .{pair[1]});
1716 defer th.join();
1717 const t0 = std.time.milliTimestamp();
1718 const frame = (try askOnce(alloc, pair[0], .stats_req, "", .stats_reply, null, .is_the_answer)).?;
1719 defer frame.deinit(alloc);
1720 try std.testing.expectEqualStrings("late", frame.payload);
1721 try std.testing.expect(std.time.milliTimestamp() - t0 >= 150);
1722 }
1723 }
1724
1725 test "askOnce: an empty payload is the answer for stats and not for upgrade" {
1726 const alloc = std.testing.allocator;
1727
1728 // Both readings off ONE conversation: the same two frames, asked for
1729 // twice, so a helper that hard-coded either answer fails one arm.
1730 for ([_]EmptyPayload{ .is_the_answer, .keeps_waiting }) |empty| {
1731 var pair: [2]i32 = undefined;
1732 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1733 defer std.posix.close(pair[0]);
1734 defer std.posix.close(pair[1]);
1735 try proto.writeFrame(pair[1], .upgrade_reply, "");
1736 try proto.writeFrame(pair[1], .upgrade_reply, &.{0});
1737
1738 const frame = (try askOnce(alloc, pair[0], .upgrade_req, "", .upgrade_reply, 500, empty)).?;
1739 defer frame.deinit(alloc);
1740 // `mux d upgrade` indexes `payload[0]` for the status byte, so an
1741 // empty reply is a peer it cannot read, not an answer — it waits.
1742 switch (empty) {
1743 .is_the_answer => try std.testing.expectEqual(@as(usize, 0), frame.payload.len),
1744 .keeps_waiting => try std.testing.expectEqualSlices(u8, &.{0}, frame.payload),
1745 }
1746 }
1747 }
1748
1749 test "askOnce: a frame this side cannot read is an error, never silence" {
1750 // `mux d dump` and `mux d stats` hand this to std's main, which prints
1751 // `error: <name>` before the rc 1 — the reading of a corrupt frame the
1752 // verbs had before the round-trip was shared. Mapping it to null would
1753 // spell a broken daemon exactly like an absent one.
1754 var pair: [2]i32 = undefined;
1755 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1756 defer std.posix.close(pair[0]);
1757 defer std.posix.close(pair[1]);
1758 var hdr: [5]u8 = undefined;
1759 hdr[0] = @intFromEnum(proto.MsgType.stats_reply);
1760 std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
1761 try proto.writeAllFd(pair[1], &hdr);
1762 try std.testing.expectError(
1763 error.FrameTooLarge,
1764 askOnce(std.testing.allocator, pair[0], .stats_req, "", .stats_reply, null, .is_the_answer),
1765 );
1766 }
1767
1661 test "oneShotQuery: a socket nobody serves is exit 1" { 1768 test "oneShotQuery: a socket nobody serves is exit 1" {
1662 const testtmp = @import("testtmp"); 1769 const testtmp = @import("testtmp");
1663 var tmp = try testtmp.TmpDir.make(); 1770 var tmp = try testtmp.TmpDir.make();