a73x

54fd740a

dial.ask and a decoder for upgrade_reply

a73x   2026-08-31 17:06

Commit message
dial.ask and a decoder for upgrade_reply

`mux d`'s observer verbs each spelled the same round trip: connect, write
one frame, poll with a deadline, read until the wanted type arrives, close.
main.zig held it as a private `askOnce` that four callers reached for, which
put a primitive the client and the agent could also use behind the daemon
entrypoint's door. It moves to `dial.ask`, beside `dial`/`dialAttach`, where
the rest of "reach a daemon over its socket" already lives.

The two failures a caller must tell apart are named — `error.NoDaemon` for
the dial and `error.RequestNotSent` for a peer that hung up before the write
— because every verb has its own words for a daemon that never heard the
question, and none of them wants to enumerate a connect errno set to find
out. `confirmServing` keeps its meaning exactly: only a real `stats_reply`
proves the post-exec image is serving.

`askOnce` also carried an `EmptyPayloadPolicy`, whose whole reason to exist
was that `upgrade_reply` had no decoder: the transport skipped empty frames
on upgrade's behalf so the hand-rolled `payload[0]` read downstream could not
index past the end. `protocol.parseUpgradeReply` answers that where the rest
of the upgrade wire is read — a payload with no status byte is no answer, and
joins timeout and EOF at the client's "no reply" line — so the transport is
back to one rule for every verb and the enum is gone. Its encode twin owns
the daemon's truncation bound, so the two sites in server.zig that spelled
the status byte by hand now name the codec.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
build.zig
Old New
@@ -251,9 +251,13 @@ const mod_table = [_]ModSpec{
251 // no-arg client opens, `term` for `term.protocol`'s SessionName.parseCLI 251 // no-arg client opens, `term` for `term.protocol`'s SessionName.parseCLI
252 // (a bad --session is a usage error at parse, not bytes a daemon 252 // (a bad --session is a usage error at parse, not bytes a daemon
253 // downstream must refuse) and `sockpath` for the sun_path bound every verb 253 // downstream must refuse) and `sockpath` for the sun_path bound every verb
254 // checks before acting on a path. `testtmp` is the keygen round-trip's: 254 // checks before acting on a path. `dial` is how `mux d`'s observer verbs
255 // it needs a directory to generate into, which the daemon never touches. 255 // — stats, dump, endpoint, upgrade — ask their one question: the same
256 .{ .name = "mux", .path = "src/cli/mux.zig", .link_libc = true, .imports = &.{ "daemon", "client", "wall", "agent", "webhub", "term", "proxy", "quic", "xdg", "spawn", "sockpath", "cliflags" }, .test_imports = &.{"testtmp"}, .quic_tests = true }, 256 // round trip the client and the agent already reach for, rather than a
257 // fourth copy of connect-write-poll-read here. `testtmp` is the keygen
258 // round-trip's: it needs a directory to generate into, which the daemon
259 // never touches.
260 .{ .name = "mux", .path = "src/cli/mux.zig", .link_libc = true, .imports = &.{ "daemon", "client", "wall", "agent", "webhub", "term", "proxy", "quic", "xdg", "spawn", "sockpath", "cliflags", "dial" }, .test_imports = &.{"testtmp"}, .quic_tests = true },
257 }; 261 };
258 262
259 /// Comptime row lookup. Every hand-written module name in this file goes 263 /// Comptime row lookup. Every hand-written module name in this file goes
src/cli/main.zig
Old New
@@ -17,6 +17,7 @@ const handoff = @import("client").handoff;
17 const sockpath = @import("sockpath"); 17 const sockpath = @import("sockpath");
18 const upgrade = @import("daemon").upgrade; 18 const upgrade = @import("daemon").upgrade;
19 const cliflags = @import("cliflags"); 19 const cliflags = @import("cliflags");
20 const dial = @import("dial");
20 21
21 const usage = 22 const usage =
22 \\usage: 23 \\usage:
@@ -583,45 +584,6 @@ fn run(alloc: std.mem.Allocator, o: DaemonArguments, sock_path: []const u8) !u8
583 return try srv.run(); 584 return try srv.run();
584 } 585 }
585 586
586 /// Policy for an expected frame with an empty payload. Dump and stats accept an
587 /// empty result, while upgrade requires a status byte and continues waiting.
588 const EmptyPayloadPolicy = enum { is_the_answer, keeps_waiting };
589
590 /// Send one request and read until a frame of type `want` arrives. Return null
591 /// for timeout, EOF, or other no-reply conditions; a null deadline blocks.
592 fn askOnce(
593 alloc: std.mem.Allocator,
594 fd: std.posix.fd_t,
595 req: proto.MsgType,
596 payload: []const u8,
597 want: proto.MsgType,
598 deadline_ms: ?u32,
599 empty: EmptyPayloadPolicy,
600 ) !?proto.Frame {
601 // Preserve a distinct send error because upgrade reports a request that
602 // could not be delivered differently from a missing reply.
603 proto.writeFrame(fd, req, payload) catch return error.RequestNotSent;
604 const deadline: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null;
605 while (true) {
606 if (deadline) |end| {
607 const left = end - std.time.milliTimestamp();
608 if (left <= 0) return null;
609 var fds = [_]std.posix.pollfd{
610 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
611 };
612 if ((std.posix.poll(&fds, @intCast(left)) catch return null) == 0) return null;
613 if (fds[0].revents == 0) continue;
614 }
615 // The frame read may extend past the poll deadline if a peer writes only
616 // part of a frame. Treat malformed or partial frames as errors rather
617 // than as an absent reply.
618 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
619 if (frame.type == want and
620 !(empty == .keeps_waiting and frame.payload.len == 0)) return frame;
621 frame.deinit(alloc);
622 }
623 }
624
625 /// Perform an unbounded one-shot query. A connected daemon that stops replying 587 /// Perform an unbounded one-shot query. A connected daemon that stops replying
626 /// remains a visible hang rather than being reported like an absent socket. 588 /// remains a visible hang rather than being reported like an absent socket.
627 fn oneShotQuery( 589 fn oneShotQuery(
@@ -632,16 +594,19 @@ fn oneShotQuery(
632 req_payload: []const u8, 594 req_payload: []const u8,
633 want: proto.MsgType, 595 want: proto.MsgType,
634 ) !u8 { 596 ) !u8 {
635 const stream = std.net.connectUnixSocket(sock_path) catch { 597 const frame = (dial.ask(alloc, sock_path, req, req_payload, want, null) catch |e| switch (e) {
636 std.debug.print( 598 error.NoDaemon => {
637 "mux d {s}: nothing listening on {s} (`mux d start -d` starts one)\n", 599 std.debug.print(
638 .{ verb, sock_path }, 600 "mux d {s}: nothing listening on {s} (`mux d start -d` starts one)\n",
639 ); 601 .{ verb, sock_path },
640 return 1; 602 );
641 }; 603 return 1;
642 defer stream.close(); 604 },
643 605 // A reply this side could not read, or a daemon that hung up over the
644 const frame = (try askOnce(alloc, stream.handle, req, req_payload, want, null, .is_the_answer)) orelse return 1; 606 // request: both stay errors, so a broken daemon never reads as an
607 // absent one.
608 else => return e,
609 }) orelse return 1;
645 defer frame.deinit(alloc); 610 defer frame.deinit(alloc);
646 try proto.writeAllFd(std.posix.STDOUT_FILENO, frame.payload); 611 try proto.writeAllFd(std.posix.STDOUT_FILENO, frame.payload);
647 try proto.writeAllFd(std.posix.STDOUT_FILENO, "\n"); 612 try proto.writeAllFd(std.posix.STDOUT_FILENO, "\n");
@@ -748,12 +713,6 @@ fn upgradeCmd(alloc: std.mem.Allocator, sock_path: []const u8, allow_same: bool)
748 return 1; 713 return 1;
749 }; 714 };
750 715
751 const stream = std.net.connectUnixSocket(sock_path) catch {
752 std.debug.print("mux d upgrade: nothing listening on {s}\n", .{sock_path});
753 return 1;
754 };
755 defer stream.close();
756
757 var buf: [std.fs.max_path_bytes + 64]u8 = undefined; 716 var buf: [std.fs.max_path_bytes + 64]u8 = undefined;
758 const payload = proto.encodeUpgradeReq(&buf, .{ 717 const payload = proto.encodeUpgradeReq(&buf, .{
759 .allow_same_version = allow_same, 718 .allow_same_version = allow_same,
@@ -767,28 +726,34 @@ fn upgradeCmd(alloc: std.mem.Allocator, sock_path: []const u8, allow_same: bool)
767 // frame without a word: the expiry is a diagnosis, not a timeout. So is 726 // frame without a word: the expiry is a diagnosis, not a timeout. So is
768 // EOF — an older daemon that drops the connection over a frame it 727 // EOF — an older daemon that drops the connection over a frame it
769 // cannot read reaches the same conclusion silence does. 728 // cannot read reaches the same conclusion silence does.
770 const reply = askOnce(alloc, stream.handle, .upgrade_req, payload, .upgrade_reply, 5000, .keeps_waiting) catch |e| ask: { 729 const reply = dial.ask(alloc, sock_path, .upgrade_req, payload, .upgrade_reply, 5000) catch |e| ask: {
771 if (e == error.RequestNotSent) { 730 switch (e) {
772 std.debug.print("mux d upgrade: {s} closed before the request landed\n", .{sock_path}); 731 error.NoDaemon => {
773 return 1; 732 std.debug.print("mux d upgrade: nothing listening on {s}\n", .{sock_path});
733 return 1;
734 },
735 error.RequestNotSent => {
736 std.debug.print("mux d upgrade: {s} closed before the request landed\n", .{sock_path});
737 return 1;
738 },
739 // Treat an unreadable reply like timeout or EOF: no usable upgrade
740 // response was received.
741 else => break :ask null,
774 } 742 }
775 // Treat an unreadable reply like timeout or EOF: no usable upgrade
776 // response was received.
777 break :ask null;
778 }; 743 };
779 // `.keeps_waiting` filters empty upgrade replies. Keep this payload bounds 744 // A frame with no status byte is not an answer — `parseUpgradeReply` says
780 // check and classify any remaining empty frame as no reply. 745 // so — and joins timeout and EOF at the "no reply" line below.
781 if (reply) |frame| skip: { 746 if (reply) |frame| skip: {
782 defer frame.deinit(alloc); 747 defer frame.deinit(alloc);
783 if (frame.payload.len == 0) break :skip; 748 const answer = proto.parseUpgradeReply(frame.payload) orelse break :skip;
784 if (frame.payload[0] != 0) { 749 if (!answer.ok) {
785 // The daemon's words, verbatim: it is the side that knows which 750 // The daemon's words, verbatim: it is the side that knows which
786 // check failed, and paraphrasing here would lose the versions. 751 // check failed, and paraphrasing here would lose the versions.
787 std.debug.print("mux d upgrade: refused: {s}\n", .{frame.payload[1..]}); 752 std.debug.print("mux d upgrade: refused: {s}\n", .{answer.reason});
788 // Add compatibility context when version probing fails: v0.0.1-15 753 // Add compatibility context when version probing fails: v0.0.1-15
789 // and older expect `muxd <version>`, while this binary reports 754 // and older expect `muxd <version>`, while this binary reports
790 // `mux <version>`. 755 // `mux <version>`.
791 if (std.mem.eql(u8, frame.payload[1..], "version: output mismatch")) 756 if (std.mem.eql(u8, answer.reason, "version: output mismatch"))
792 std.debug.print( 757 std.debug.print(
793 "mux d upgrade: if that daemon is v0.0.1-15 or older, it wants a " ++ 758 "mux d upgrade: if that daemon is v0.0.1-15 or older, it wants a " ++
794 "candidate that prints `muxd <version>` and this one is `mux`. " ++ 759 "candidate that prints `muxd <version>` and this one is `mux`. " ++
@@ -814,14 +779,14 @@ fn confirmServing(alloc: std.mem.Allocator, sock_path: []const u8) u8 {
814 // A connect is insufficient because the listener descriptor survives exec 779 // A connect is insufficient because the listener descriptor survives exec
815 // and can queue connections throughout handover. A valid reply proves that 780 // and can queue connections throughout handover. A valid reply proves that
816 // the new image is accepting and processing frames. 781 // the new image is accepting and processing frames.
817 const stream = std.net.connectUnixSocket(sock_path) catch {
818 std.debug.print("mux d upgrade: {s} stopped answering after the exec\n", .{sock_path});
819 return 1;
820 };
821 defer stream.close();
822
823 const deadline_ms: u32 = 5000; 782 const deadline_ms: u32 = 5000;
824 if (askOnce(alloc, stream.handle, .stats_req, "", .stats_reply, deadline_ms, .is_the_answer) catch return 1) |frame| { 783 if (dial.ask(alloc, sock_path, .stats_req, "", .stats_reply, deadline_ms) catch |e| switch (e) {
784 error.NoDaemon => {
785 std.debug.print("mux d upgrade: {s} stopped answering after the exec\n", .{sock_path});
786 return 1;
787 },
788 else => return 1,
789 }) |frame| {
825 frame.deinit(alloc); 790 frame.deinit(alloc);
826 return 0; 791 return 0;
827 } 792 }
@@ -1011,16 +976,13 @@ fn reportKeyRefusal(path: []const u8, err: anyerror) void {
1011 /// failure so old daemons that ignore the request fall back to SSH instead of 976 /// failure so old daemons that ignore the request fall back to SSH instead of
1012 /// hanging. 977 /// hanging.
1013 fn askEndpointPort(alloc: std.mem.Allocator, sock_path: []const u8) u16 { 978 fn askEndpointPort(alloc: std.mem.Allocator, sock_path: []const u8) u16 {
1014 const stream = std.net.connectUnixSocket(sock_path) catch return 0; 979 const frame = (dial.ask(
1015 defer stream.close();
1016 const frame = (askOnce(
1017 alloc, 980 alloc,
1018 stream.handle, 981 sock_path,
1019 .endpoint_req, 982 .endpoint_req,
1020 "", 983 "",
1021 .endpoint_reply, 984 .endpoint_reply,
1022 start_deadline_ms, 985 start_deadline_ms,
1023 .is_the_answer,
1024 ) catch null) orelse return 0; 986 ) catch null) orelse return 0;
1025 defer frame.deinit(alloc); 987 defer frame.deinit(alloc);
1026 return proto.decodeEndpointReply(frame.payload) catch 0; 988 return proto.decodeEndpointReply(frame.payload) catch 0;
@@ -1721,86 +1683,6 @@ test "endpointCmd: polling an absent daemon does not start one" {
1721 try std.testing.expectEqual(@as(u64, 0), (try out.stat()).size); 1683 try std.testing.expectEqual(@as(u64, 0), (try out.stat()).size);
1722 } 1684 }
1723 1685
1724 test "askOnce: a deadline gives up on silence, and no deadline waits out a late reply" {
1725 const alloc = std.testing.allocator;
1726
1727 // A socket pair models the daemon and controls whether and when a reply
1728 // arrives.
1729 {
1730 var pair: [2]i32 = undefined;
1731 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1732 defer std.posix.close(pair[0]);
1733 defer std.posix.close(pair[1]);
1734 const t0 = std.time.milliTimestamp();
1735 try std.testing.expect((try askOnce(alloc, pair[0], .stats_req, "", .stats_reply, 100, .is_the_answer)) == null);
1736 // Verify that silence consumes the deadline before returning null.
1737 try std.testing.expect(std.time.milliTimestamp() - t0 >= 100);
1738 }
1739
1740 // Without a deadline, wait for a delayed reply.
1741 {
1742 var pair: [2]i32 = undefined;
1743 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1744 defer std.posix.close(pair[0]);
1745 const Late = struct {
1746 fn run(fd: std.posix.fd_t) void {
1747 std.Thread.sleep(150 * std.time.ns_per_ms);
1748 // Ignore an unrelated frame before returning the requested type.
1749 proto.writeFrame(fd, .stats_req, "") catch {};
1750 proto.writeFrame(fd, .stats_reply, "late") catch {};
1751 std.posix.close(fd);
1752 }
1753 };
1754 const th = try std.Thread.spawn(.{}, Late.run, .{pair[1]});
1755 defer th.join();
1756 const t0 = std.time.milliTimestamp();
1757 const frame = (try askOnce(alloc, pair[0], .stats_req, "", .stats_reply, null, .is_the_answer)).?;
1758 defer frame.deinit(alloc);
1759 try std.testing.expectEqualStrings("late", frame.payload);
1760 try std.testing.expect(std.time.milliTimestamp() - t0 >= 150);
1761 }
1762 }
1763
1764 test "askOnce: an empty payload is the answer for stats and not for upgrade" {
1765 const alloc = std.testing.allocator;
1766
1767 // Exercise both policies against the same pair of reply frames.
1768 for ([_]EmptyPayloadPolicy{ .is_the_answer, .keeps_waiting }) |empty| {
1769 var pair: [2]i32 = undefined;
1770 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1771 defer std.posix.close(pair[0]);
1772 defer std.posix.close(pair[1]);
1773 try proto.writeFrame(pair[1], .upgrade_reply, "");
1774 try proto.writeFrame(pair[1], .upgrade_reply, &.{0});
1775
1776 const frame = (try askOnce(alloc, pair[0], .upgrade_req, "", .upgrade_reply, 500, empty)).?;
1777 defer frame.deinit(alloc);
1778 // Upgrade requires payload[0] for its status and therefore skips the
1779 // empty reply; stats accepts it.
1780 switch (empty) {
1781 .is_the_answer => try std.testing.expectEqual(@as(usize, 0), frame.payload.len),
1782 .keeps_waiting => try std.testing.expectEqualSlices(u8, &.{0}, frame.payload),
1783 }
1784 }
1785 }
1786
1787 test "askOnce: a frame this side cannot read is an error, never silence" {
1788 // Preserve corrupt-frame errors so dump and stats distinguish a broken
1789 // daemon response from an absent daemon.
1790 var pair: [2]i32 = undefined;
1791 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
1792 defer std.posix.close(pair[0]);
1793 defer std.posix.close(pair[1]);
1794 var hdr: [5]u8 = undefined;
1795 hdr[0] = @intFromEnum(proto.MsgType.stats_reply);
1796 std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
1797 try proto.writeAllFd(pair[1], &hdr);
1798 try std.testing.expectError(
1799 error.FrameTooLarge,
1800 askOnce(std.testing.allocator, pair[0], .stats_req, "", .stats_reply, null, .is_the_answer),
1801 );
1802 }
1803
1804 test "oneShotQuery: a socket nobody serves is exit 1" { 1686 test "oneShotQuery: a socket nobody serves is exit 1" {
1805 const testtmp = @import("testtmp"); 1687 const testtmp = @import("testtmp");
1806 var tmp = try testtmp.TmpDir.make(); 1688 var tmp = try testtmp.TmpDir.make();
src/dial.zig
Old New
@@ -46,6 +46,78 @@ pub fn dialAttachNamed(sock_path: []const u8, cols: u16, rows: u16, name: []cons
46 return s; 46 return s;
47 } 47 }
48 48
49 /// Dial, ask one question, read the one answer, hang up. The observer verbs'
50 /// round trip: `stats_req`, `sessions_req`, `debug_dump`, `endpoint_req`,
51 /// `upgrade_req` — every one of them a question asked by a caller that holds
52 /// no other connection to that daemon and wants none after the answer.
53 ///
54 /// Null is "no answer": the deadline ran out, or the peer closed without
55 /// sending one. A null `deadline_ms` waits forever, which is what a caller
56 /// wants when a daemon that has stopped replying should be a visible hang
57 /// rather than a report of an absent socket.
58 ///
59 /// Two failures are named so the caller does not have to tell them apart from
60 /// a connect errno set: `error.NoDaemon` is the dial, and
61 /// `error.RequestNotSent` is a peer that closed between the connect and the
62 /// write. Both mean the daemon never heard the question, and every caller has
63 /// its own words for that — which is why this returns the distinction instead
64 /// of printing it. Any other error is a reply this side could not read, and
65 /// stays an error precisely so a corrupt frame is never reported as silence.
66 pub fn ask(
67 alloc: std.mem.Allocator,
68 sock_path: []const u8,
69 req: proto.MsgType,
70 payload: []const u8,
71 want: proto.MsgType,
72 deadline_ms: ?u32,
73 ) !?proto.Frame {
74 const s = dial(sock_path) catch return error.NoDaemon;
75 defer s.close();
76 return askOn(alloc, s.handle, req, payload, want, deadline_ms);
77 }
78
79 /// `ask` once the connection exists. Private on purpose: this loop DROPS every
80 /// frame that is not the one it was told to wait for, which is right for a
81 /// socket opened to ask one question and wrong for a client connection
82 /// carrying a snapshot and its deltas. The tests below reach it through a
83 /// socketpair — the one fd of this shape that no dial produced.
84 fn askOn(
85 alloc: std.mem.Allocator,
86 fd: std.posix.fd_t,
87 req: proto.MsgType,
88 payload: []const u8,
89 want: proto.MsgType,
90 deadline_ms: ?u32,
91 ) !?proto.Frame {
92 // A request that could not be delivered is its own answer: the caller
93 // reports a daemon that never heard the question differently from one
94 // that heard it and said nothing.
95 proto.writeFrame(fd, req, payload) catch return error.RequestNotSent;
96 const deadline: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null;
97 while (true) {
98 if (deadline) |end| {
99 const left = end - std.time.milliTimestamp();
100 if (left <= 0) return null;
101 var fds = [_]std.posix.pollfd{
102 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
103 };
104 if ((std.posix.poll(&fds, @intCast(left)) catch return null) == 0) return null;
105 if (fds[0].revents == 0) continue;
106 }
107 // The frame read may extend past the poll deadline if a peer writes
108 // only part of a frame. A malformed or partial frame is an error
109 // rather than an absent reply.
110 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
111 // An empty payload of the wanted type IS the answer here. What an
112 // empty reply means belongs to that verb's decoder — `stats_reply`
113 // says "nothing to report" with one, `parseUpgradeReply` calls one no
114 // answer at all — and a transport that guessed for them would have to
115 // be told the verb's policy by every caller.
116 if (frame.type == want) return frame;
117 frame.deinit(alloc);
118 }
119 }
120
49 // A daemon of our own is the server suite's business, not this module's: 121 // A daemon of our own is the server suite's business, not this module's:
50 // what is worth pinning without one is that a path nobody bound fails as a 122 // what is worth pinning without one is that a path nobody bound fails as a
51 // dial — an error the caller can report — rather than blocking or reaching 123 // dial — an error the caller can report — rather than blocking or reaching
@@ -58,3 +130,86 @@ test "the attach helpers fail at the dial, before any frame" {
58 try std.testing.expectError(error.FileNotFound, dialAttach("/nonexistent-dir/mux-dial-test.sock", 80, 24)); 130 try std.testing.expectError(error.FileNotFound, dialAttach("/nonexistent-dir/mux-dial-test.sock", 80, 24));
59 try std.testing.expectError(error.FileNotFound, dialAttachNamed("/nonexistent-dir/mux-dial-test.sock", 80, 24, "work")); 131 try std.testing.expectError(error.FileNotFound, dialAttachNamed("/nonexistent-dir/mux-dial-test.sock", 80, 24, "work"));
60 } 132 }
133
134 test "ask: a path nothing is bound at is error.NoDaemon, not a connect errno" {
135 try std.testing.expectError(
136 error.NoDaemon,
137 ask(std.testing.allocator, "/nonexistent-dir/mux-dial-test.sock", .stats_req, "", .stats_reply, 100),
138 );
139 }
140
141 test "askOn: a deadline gives up on silence, and no deadline waits out a late reply" {
142 const alloc = std.testing.allocator;
143
144 // A socket pair models the daemon and controls whether and when a reply
145 // arrives.
146 {
147 var pair: [2]i32 = undefined;
148 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
149 defer std.posix.close(pair[0]);
150 defer std.posix.close(pair[1]);
151 const t0 = std.time.milliTimestamp();
152 try std.testing.expect((try askOn(alloc, pair[0], .stats_req, "", .stats_reply, 100)) == null);
153 // Verify that silence consumes the deadline before returning null.
154 try std.testing.expect(std.time.milliTimestamp() - t0 >= 100);
155 }
156
157 // Without a deadline, wait for a delayed reply.
158 {
159 var pair: [2]i32 = undefined;
160 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
161 defer std.posix.close(pair[0]);
162 const Late = struct {
163 fn run(fd: std.posix.fd_t) void {
164 std.Thread.sleep(150 * std.time.ns_per_ms);
165 // Ignore an unrelated frame before returning the requested type.
166 proto.writeFrame(fd, .stats_req, "") catch {};
167 proto.writeFrame(fd, .stats_reply, "late") catch {};
168 std.posix.close(fd);
169 }
170 };
171 const th = try std.Thread.spawn(.{}, Late.run, .{pair[1]});
172 defer th.join();
173 const t0 = std.time.milliTimestamp();
174 const frame = (try askOn(alloc, pair[0], .stats_req, "", .stats_reply, null)).?;
175 defer frame.deinit(alloc);
176 try std.testing.expectEqualStrings("late", frame.payload);
177 try std.testing.expect(std.time.milliTimestamp() - t0 >= 150);
178 }
179 }
180
181 test "askOn: an empty payload of the wanted type is the answer, not a frame to skip" {
182 // The verb's decoder decides what an empty reply means. This loop used to
183 // carry a per-caller policy so that `upgrade_reply` could keep waiting
184 // through one; `protocol.parseUpgradeReply` now answers that question
185 // where the rest of the upgrade wire is read.
186 const alloc = std.testing.allocator;
187 var pair: [2]i32 = undefined;
188 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
189 defer std.posix.close(pair[0]);
190 defer std.posix.close(pair[1]);
191 try proto.writeFrame(pair[1], .upgrade_reply, "");
192 try proto.writeFrame(pair[1], .upgrade_reply, &.{0});
193
194 const frame = (try askOn(alloc, pair[0], .upgrade_req, "", .upgrade_reply, 500)).?;
195 defer frame.deinit(alloc);
196 try std.testing.expectEqual(@as(usize, 0), frame.payload.len);
197 try std.testing.expect(proto.parseUpgradeReply(frame.payload) == null);
198 }
199
200 test "askOn: a frame this side cannot read is an error, never silence" {
201 // Preserve corrupt-frame errors so dump and stats distinguish a broken
202 // daemon response from an absent daemon.
203 var pair: [2]i32 = undefined;
204 try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
205 defer std.posix.close(pair[0]);
206 defer std.posix.close(pair[1]);
207 var hdr: [5]u8 = undefined;
208 hdr[0] = @intFromEnum(proto.MsgType.stats_reply);
209 std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
210 try proto.writeAllFd(pair[1], &hdr);
211 try std.testing.expectError(
212 error.FrameTooLarge,
213 askOn(std.testing.allocator, pair[0], .stats_req, "", .stats_reply, null),
214 );
215 }
src/engine/protocol.zig
Old New
@@ -328,6 +328,36 @@ pub fn parseUpgradeReq(payload: []const u8) error{BadPayload}!UpgradeReq {
328 }; 328 };
329 } 329 }
330 330
331 // `upgrade_reply`: the daemon's yes or no, and — when it is a no — the words
332 // for it. The reason is the daemon's own prose because it is the side that
333 // knows which check failed; the client quotes it rather than paraphrasing.
334 pub const UpgradeReply = struct {
335 ok: bool,
336 /// Empty when the daemon accepted, or when it refused without saying why.
337 reason: []const u8,
338 };
339
340 /// The daemon's reply buffer. A reason longer than this is truncated rather
341 /// than refused: the bytes are prose for a person reading a terminal, and a
342 /// truncated explanation beats a refusal that never reaches them.
343 pub const upgrade_reply_max_len = 256;
344
345 pub fn encodeUpgradeReply(buf: []u8, reply: UpgradeReply) []const u8 {
346 buf[0] = if (reply.ok) 0 else 1;
347 const n = @min(reply.reason.len, buf.len - 1);
348 @memcpy(buf[1..][0..n], reply.reason[0..n]);
349 return buf[0 .. 1 + n];
350 }
351
352 /// Null is a payload with no status byte at all — nothing this side can call
353 /// an answer, which the caller reports as "no reply" rather than as a refusal
354 /// it could quote. A daemon that predates this verb sends no frame at all and
355 /// reaches the same conclusion by silence.
356 pub fn parseUpgradeReply(payload: []const u8) ?UpgradeReply {
357 if (payload.len == 0) return null;
358 return .{ .ok = payload[0] == 0, .reason = payload[1..] };
359 }
360
331 /// Screen-space rows; the grid's owner normalizes. 361 /// Screen-space rows; the grid's owner normalizes.
332 pub const SelectionPoint = struct { 362 pub const SelectionPoint = struct {
333 row: u32, 363 row: u32,
@@ -2248,6 +2278,34 @@ test "parseUpgradeReq: a relative path is BadPayload" {
2248 try std.testing.expectError(error.BadPayload, parseUpgradeReq(bad)); 2278 try std.testing.expectError(error.BadPayload, parseUpgradeReq(bad));
2249 } 2279 }
2250 2280
2281 test "encodeUpgradeReply/parseUpgradeReply: accepted, refused with words, and no status byte" {
2282 var buf: [upgrade_reply_max_len]u8 = undefined;
2283
2284 const ok = parseUpgradeReply(encodeUpgradeReply(&buf, .{ .ok = true, .reason = "" })) orelse
2285 return error.ReplyDidNotParse;
2286 try std.testing.expect(ok.ok);
2287 try std.testing.expectEqualStrings("", ok.reason);
2288
2289 const wire = encodeUpgradeReply(&buf, .{ .ok = false, .reason = "version: output mismatch" });
2290 try std.testing.expectEqual(@as(u8, 1), wire[0]);
2291 const no = parseUpgradeReply(wire) orelse return error.ReplyDidNotParse;
2292 try std.testing.expect(!no.ok);
2293 // The client matches this string to add its rename hint, so the reason
2294 // must survive the round trip byte for byte.
2295 try std.testing.expectEqualStrings("version: output mismatch", no.reason);
2296
2297 // An empty payload is not an accepted upgrade: without a status byte
2298 // there is no answer to read, and reading one as "accepted" would tell a
2299 // user their daemon exec'd when nothing did.
2300 try std.testing.expect(parseUpgradeReply("") == null);
2301
2302 // A reason past the buffer is cut, never refused and never overrun.
2303 var small: [8]u8 = undefined;
2304 const cut = parseUpgradeReply(encodeUpgradeReply(&small, .{ .ok = false, .reason = "0123456789" })) orelse
2305 return error.ReplyDidNotParse;
2306 try std.testing.expectEqualStrings("0123456", cut.reason);
2307 }
2308
2251 test "parseUpgradeReq: an empty version is BadPayload" { 2309 test "parseUpgradeReq: an empty version is BadPayload" {
2252 // Manually crafted: flag ++ NUL ++ path (version is zero-length) 2310 // Manually crafted: flag ++ NUL ++ path (version is zero-length)
2253 const bad = &[_]u8{ 0, 0, '/', 'a' }; 2311 const bad = &[_]u8{ 0, 0, '/', 'a' };
src/server/server.zig
Old New
@@ -1689,11 +1689,11 @@ pub const Server = struct {
1689 /// on this socket. Five sites spelled this out, each remembering the 1689 /// on this socket. Five sites spelled this out, each remembering the
1690 /// drop for itself. 1690 /// drop for itself.
1691 fn refuseUpgrade(self: *Server, i: usize, reason: []const u8) void { 1691 fn refuseUpgrade(self: *Server, i: usize, reason: []const u8) void {
1692 var reply: [256]u8 = undefined; 1692 var reply: [proto.upgrade_reply_max_len]u8 = undefined;
1693 reply[0] = 1; 1693 self.replyTo(.{ .observer = i }, .upgrade_reply, proto.encodeUpgradeReply(&reply, .{
1694 const n = @min(reason.len, reply.len - 1); 1694 .ok = false,
1695 @memcpy(reply[1..][0..n], reason[0..n]); 1695 .reason = reason,
1696 self.replyTo(.{ .observer = i }, .upgrade_reply, reply[0 .. 1 + n]); 1696 }));
1697 self.dropObserver(i); 1697 self.dropObserver(i);
1698 } 1698 }
1699 1699
@@ -2195,7 +2195,11 @@ pub const Server = struct {
2195 std.posix.close(memfd); 2195 std.posix.close(memfd);
2196 return self.refuseUpgrade(i, "oom"); 2196 return self.refuseUpgrade(i, "oom");
2197 }; 2197 };
2198 self.replyTo(.{ .observer = i }, .upgrade_reply, &.{0}); 2198 var accepted: [1]u8 = undefined;
2199 self.replyTo(.{ .observer = i }, .upgrade_reply, proto.encodeUpgradeReply(&accepted, .{
2200 .ok = true,
2201 .reason = "",
2202 }));
2199 self.dropObserver(i); 2203 self.dropObserver(i);
2200 self.pending_upgrade = .{ .path = path, .memfd = memfd }; 2204 self.pending_upgrade = .{ .path = path, .memfd = memfd };
2201 }, 2205 },