28e91371
refactor: one owner answers the daemon's verbs, whichever table asked
a73x 2026-08-29 01:04
Commit message
src/server/server.zig
| Old | New | ||
|---|---|---|---|
| @@ -1926,39 +1926,132 @@ pub const Server = struct { | |||
| 1926 | /// One arm used to carry its own null check; it read like the others | 1926 | /// One arm used to carry its own null check; it read like the others |
| 1927 | /// were missing one. | 1927 | /// were missing one. |
| 1928 | fn handleFrame(self: *Server, i: usize, frame: proto.Frame) void { | 1928 | fn handleFrame(self: *Server, i: usize, frame: proto.Frame) void { |
| 1929 | if (self.handleDaemonVerb(.{ .client = i }, frame)) return; | ||
| 1929 | switch (frame.type) { | 1930 | switch (frame.type) { |
| 1930 | .attach => self.onAttach(i, frame), | 1931 | .attach => self.onAttach(i, frame), |
| 1931 | .resize => self.onResize(i, frame), | 1932 | .resize => self.onResize(i, frame), |
| 1932 | .input => self.onInput(i, frame), | 1933 | .input => self.onInput(i, frame), |
| 1933 | .stats_req => self.onStatsReq(i), | ||
| 1934 | .sessions_req => self.onSessionsReq(i), | ||
| 1935 | .fetch_scrollback => self.onFetchScrollback(i, frame), | 1934 | .fetch_scrollback => self.onFetchScrollback(i, frame), |
| 1936 | .selection_req => self.onSelectionReq(i, frame), | 1935 | .selection_req => self.onSelectionReq(i, frame), |
| 1937 | .detach => self.dropClient(i), | 1936 | .detach => self.dropClient(i), |
| 1938 | // Shut down via the signal path, not a second one: this is the | ||
| 1939 | // flag SIGTERM sets, so the poll loop, deinit's unlink and the | ||
| 1940 | // pty teardown are all already-tested code. No reply is sent — | ||
| 1941 | // the ack is the socket dying, which is what `mux d stop` polls | ||
| 1942 | // for. This arm has no product caller (`mux d stop` never | ||
| 1943 | // attaches, so it lands in serviceObserver); it is here so the | ||
| 1944 | // verb means the same thing on any connection, and it grants no | ||
| 1945 | // authority an attached client lacks — see the .input arm above, | ||
| 1946 | // which already writes to the pty master. | ||
| 1947 | .stop_req => shutdown_flag.store(true, .release), | ||
| 1948 | .endpoint_req => self.onEndpointReq(i), | ||
| 1949 | .debug_dump => self.onDebugDump(i, frame), | ||
| 1950 | .status_req => self.onStatusReq(i, frame), | 1937 | .status_req => self.onStatusReq(i, frame), |
| 1951 | .await_req => self.onAwaitReq(i, frame), | 1938 | .await_req => self.onAwaitReq(i, frame), |
| 1952 | .agent_offer => self.onAgentOffer(i), | 1939 | .agent_offer => self.onAgentOffer(i), |
| 1953 | .agent_data => self.onAgentData(i, frame), | 1940 | .agent_data => self.onAgentData(i, frame), |
| 1954 | .agent_close => self.onAgentClose(i, frame), | 1941 | .agent_close => self.onAgentClose(i, frame), |
| 1942 | else => {}, | ||
| 1943 | } | ||
| 1944 | } | ||
| 1945 | |||
| 1946 | /// Whichever table the asker sits in. | ||
| 1947 | const Peer = union(enum) { | ||
| 1948 | client: usize, | ||
| 1949 | observer: usize, | ||
| 1950 | }; | ||
| 1951 | |||
| 1952 | /// The one send for a daemon verb; see handleDaemonVerb for why the two | ||
| 1953 | /// arms differ and why a short write costs the connection. | ||
| 1954 | fn replyTo(self: *Server, p: Peer, t: proto.MsgType, payload: []const u8) void { | ||
| 1955 | switch (p) { | ||
| 1956 | .client => |i| _ = self.queueFrame(i, t, payload), | ||
| 1957 | .observer => |i| { | ||
| 1958 | const o = self.observers[i] orelse return; | ||
| 1959 | proto.writeFrameBounded(o.fd, t, payload, proto.reply_budget_ms) catch | ||
| 1960 | self.dropObserver(i); | ||
| 1961 | }, | ||
| 1962 | } | ||
| 1963 | } | ||
| 1964 | |||
| 1965 | fn dropPeer(self: *Server, p: Peer) void { | ||
| 1966 | switch (p) { | ||
| 1967 | .client => |i| self.dropClient(i), | ||
| 1968 | .observer => |i| self.dropObserver(i), | ||
| 1969 | } | ||
| 1970 | } | ||
| 1971 | |||
| 1972 | /// The verbs whose answer is the DAEMON rather than the connection: | ||
| 1973 | /// identical bytes for an attached client and for a one-shot `mux d` | ||
| 1974 | /// tool, so they are answered once here and neither dispatch table | ||
| 1975 | /// repeats them. Returns false for a verb this owner does not have, and | ||
| 1976 | /// the caller answers it its own way — an attach, an upgrade, or a | ||
| 1977 | /// status_req, each of which really does depend on which table asked. | ||
| 1978 | /// | ||
| 1979 | /// They were written twice before, and the two copies were not the same | ||
| 1980 | /// program: `sessions_req` had an answerer per table, and every reply | ||
| 1981 | /// spelled its own send. A verb that drifts here answers `mux d stats` | ||
| 1982 | /// differently depending on whether the asker had attached. | ||
| 1983 | /// | ||
| 1984 | /// The two tables differ only in how an answer LEAVES, which is all | ||
| 1985 | /// `replyTo` is: a client has a send queue and never blocks, while an | ||
| 1986 | /// observer has no ClientSlot to queue into — `mux d dump`, `mux d | ||
| 1987 | /// stats` and `mux d endpoint` read their one answer and exit. So the | ||
| 1988 | /// observer's write is bounded: a dump can be megabytes, it must | ||
| 1989 | /// tolerate a short write, and a peer that stops reading must cost this | ||
| 1990 | /// reply's budget rather than the daemon's single loop. Past the budget | ||
| 1991 | /// the frame is truncated, which is why a failed write drops the | ||
| 1992 | /// connection instead of leaving half a frame on it. | ||
| 1993 | fn handleDaemonVerb(self: *Server, p: Peer, frame: proto.Frame) bool { | ||
| 1994 | switch (frame.type) { | ||
| 1995 | .stats_req => { | ||
| 1996 | // Daemon-global, and the same text for whoever asks: an | ||
| 1997 | // attached client's own session no longer picks out a single | ||
| 1998 | // seq column, since every live session gets its own line now | ||
| 1999 | // (see statsText). | ||
| 2000 | var buf: [stats_text_len]u8 = undefined; | ||
| 2001 | const text = self.statsText(&buf) catch { | ||
| 2002 | self.dropPeer(p); | ||
| 2003 | return true; | ||
| 2004 | }; | ||
| 2005 | self.replyTo(p, .stats_reply, text); | ||
| 2006 | }, | ||
| 2007 | .sessions_req => { | ||
| 2008 | // Daemon-global for the same reason .stats_req is: the answer | ||
| 2009 | // is the session TABLE, so which session the asker sits in | ||
| 2010 | // (or whether it sits in one at all) cannot change it. | ||
| 2011 | // Answered on the asking connection only — it is an | ||
| 2012 | // observation, not activity, so it claims no grid. | ||
| 2013 | var buf: SessionsBuf = undefined; | ||
| 2014 | self.replyTo(p, .sessions_reply, self.sessions.text(&buf)); | ||
| 2015 | }, | ||
| 2016 | .endpoint_req => { | ||
| 2017 | const payload = proto.encodeEndpointReply(self.endpointPort()); | ||
| 2018 | self.replyTo(p, .endpoint_reply, &payload); | ||
| 2019 | }, | ||
| 2020 | .debug_dump => { | ||
| 2021 | // Answers whatever the payload's tail names — it is a read | ||
| 2022 | // against a name, not a question about this connection's own | ||
| 2023 | // session, and an attached client is free to peek at another | ||
| 2024 | // live session. See buildDump for the resolution and the | ||
| 2025 | // in-words unknown-name reply. | ||
| 2026 | const dump = self.buildDump(frame.payload) catch { | ||
| 2027 | self.dropPeer(p); | ||
| 2028 | return true; | ||
| 2029 | }; | ||
| 2030 | defer self.alloc.free(dump); | ||
| 2031 | self.replyTo(p, .dump_reply, dump); | ||
| 2032 | }, | ||
| 1955 | .end_req => { | 2033 | .end_req => { |
| 1956 | const v = self.endSession(frame.payload, i); | 2034 | // The asking client is excluded from the "others hold it" |
| 2035 | // count; an observer holds no session and excludes nobody. | ||
| 2036 | const v = self.endSession(frame.payload, switch (p) { | ||
| 2037 | .client => |i| i, | ||
| 2038 | .observer => null, | ||
| 2039 | }); | ||
| 1957 | var buf: [proto.end_reply_max_len]u8 = undefined; | 2040 | var buf: [proto.end_reply_max_len]u8 = undefined; |
| 1958 | _ = self.queueFrame(i, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason)); | 2041 | self.replyTo(p, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason)); |
| 1959 | }, | 2042 | }, |
| 1960 | else => {}, | 2043 | // Shut down via the signal path, not a second one: this is the |
| 2044 | // flag SIGTERM sets, so the poll loop, deinit's unlink and the | ||
| 2045 | // pty teardown are all already-tested code. No reply is sent — | ||
| 2046 | // the ack is the socket dying, which is what `mux d stop` polls | ||
| 2047 | // for. `mux d stop` never attaches, so in the product this is | ||
| 2048 | // always the observer arm; a client reaching it grants no | ||
| 2049 | // authority an attached client lacks — see the .input arm, which | ||
| 2050 | // already writes to the pty master. | ||
| 2051 | .stop_req => shutdown_flag.store(true, .release), | ||
| 2052 | else => return false, | ||
| 1961 | } | 2053 | } |
| 2054 | return true; | ||
| 1962 | } | 2055 | } |
| 1963 | 2056 | ||
| 1964 | fn onAttach(self: *Server, i: usize, frame: proto.Frame) void { | 2057 | fn onAttach(self: *Server, i: usize, frame: proto.Frame) void { |
| @@ -2054,29 +2147,6 @@ pub const Server = struct { | |||
| 2054 | proto.writeAllFd(fd, frame.payload) catch self.dropClient(i); | 2147 | proto.writeAllFd(fd, frame.payload) catch self.dropClient(i); |
| 2055 | } | 2148 | } |
| 2056 | 2149 | ||
| 2057 | fn onStatsReq(self: *Server, i: usize) void { | ||
| 2058 | // Daemon-global, and the same text for whoever asks: an | ||
| 2059 | // attached client's own session no longer picks out a | ||
| 2060 | // single seq column, since every live session gets its own | ||
| 2061 | // line now (see statsText). | ||
| 2062 | var buf: [stats_text_len]u8 = undefined; | ||
| 2063 | const text = self.statsText(&buf) catch { | ||
| 2064 | self.dropClient(i); | ||
| 2065 | return; | ||
| 2066 | }; | ||
| 2067 | _ = self.queueFrame(i, .stats_reply, text); | ||
| 2068 | } | ||
| 2069 | |||
| 2070 | fn onSessionsReq(self: *Server, i: usize) void { | ||
| 2071 | // Daemon-global for the same reason .stats_req is: the | ||
| 2072 | // answer is the session TABLE, so which session the asker | ||
| 2073 | // sits in (or whether it sits in one at all) cannot change | ||
| 2074 | // it. Answered on the asking connection only — it is an | ||
| 2075 | // observation, not activity, so it claims no grid. | ||
| 2076 | var buf: SessionsBuf = undefined; | ||
| 2077 | _ = self.queueFrame(i, .sessions_reply, self.sessions.text(&buf)); | ||
| 2078 | } | ||
| 2079 | |||
| 2080 | fn onFetchScrollback(self: *Server, i: usize, frame: proto.Frame) void { | 2150 | fn onFetchScrollback(self: *Server, i: usize, frame: proto.Frame) void { |
| 2081 | // Answered on this connection only: scroll position is | 2151 | // Answered on this connection only: scroll position is |
| 2082 | // client-local, so one client paging history never | 2152 | // client-local, so one client paging history never |
| @@ -2141,29 +2211,6 @@ pub const Server = struct { | |||
| 2141 | } | 2211 | } |
| 2142 | } | 2212 | } |
| 2143 | 2213 | ||
| 2144 | // The other half of the observer arm below, here for the same | ||
| 2145 | // reason stop_req's is: the verb means the same thing on any | ||
| 2146 | // connection. Replies through the queue like stats does. | ||
| 2147 | fn onEndpointReq(self: *Server, i: usize) void { | ||
| 2148 | const payload = proto.encodeEndpointReply(self.endpointPort()); | ||
| 2149 | _ = self.queueFrame(i, .endpoint_reply, &payload); | ||
| 2150 | } | ||
| 2151 | |||
| 2152 | fn onDebugDump(self: *Server, i: usize, frame: proto.Frame) void { | ||
| 2153 | // Unlike status_req/await_req below, dump answers whatever | ||
| 2154 | // the payload's tail names — it is a read against a name, | ||
| 2155 | // not a question about this connection's own session, and | ||
| 2156 | // an attached client is free to peek at another live | ||
| 2157 | // session. See buildDump for the resolution and the | ||
| 2158 | // in-words unknown-name reply. | ||
| 2159 | const dump = self.buildDump(frame.payload) catch { | ||
| 2160 | self.dropClient(i); | ||
| 2161 | return; | ||
| 2162 | }; | ||
| 2163 | defer self.alloc.free(dump); | ||
| 2164 | _ = self.queueFrame(i, .dump_reply, dump); | ||
| 2165 | } | ||
| 2166 | |||
| 2167 | fn onStatusReq(self: *Server, i: usize, frame: proto.Frame) void { | 2214 | fn onStatusReq(self: *Server, i: usize, frame: proto.Frame) void { |
| 2168 | const si = self.clients[i].?.session orelse { | 2215 | const si = self.clients[i].?.session orelse { |
| 2169 | // A session-less slot (promoted, never attached — the | 2216 | // A session-less slot (promoted, never attached — the |
| @@ -2477,6 +2524,7 @@ pub const Server = struct { | |||
| 2477 | /// One observer frame. Slot `i` is live on entry; the `.attach` arm is | 2524 | /// One observer frame. Slot `i` is live on entry; the `.attach` arm is |
| 2478 | /// the one that ends the slot without dropping it. | 2525 | /// the one that ends the slot without dropping it. |
| 2479 | fn handleObserverFrame(self: *Server, i: usize, frame: proto.Frame) void { | 2526 | fn handleObserverFrame(self: *Server, i: usize, frame: proto.Frame) void { |
| 2527 | if (self.handleDaemonVerb(.{ .observer = i }, frame)) return; | ||
| 2480 | const fd = self.observers[i].?.fd; | 2528 | const fd = self.observers[i].?.fd; |
| 2481 | switch (frame.type) { | 2529 | switch (frame.type) { |
| 2482 | .attach => { | 2530 | .attach => { |
| @@ -2536,12 +2584,7 @@ pub const Server = struct { | |||
| 2536 | // the client they were addressed to. | 2584 | // the client they were addressed to. |
| 2537 | self.pushInbound(slot, &.{}); | 2585 | self.pushInbound(slot, &.{}); |
| 2538 | }, | 2586 | }, |
| 2539 | .debug_dump => self.replyDumpObserver(fd, frame.payload) catch self.dropObserver(i), | ||
| 2540 | .stats_req => self.replyStatsObserver(fd) catch self.dropObserver(i), | ||
| 2541 | .detach => self.dropObserver(i), | 2587 | .detach => self.dropObserver(i), |
| 2542 | // Where `mux d stop` actually lands, since it never attaches. | ||
| 2543 | // Same signal path as the client arm. | ||
| 2544 | .stop_req => shutdown_flag.store(true, .release), | ||
| 2545 | // Where `mux d upgrade` lands: validate, reply, and let the run | 2588 | // Where `mux d upgrade` lands: validate, reply, and let the run |
| 2546 | // loop exec. The reply is blocking (observer has no send queue); | 2589 | // loop exec. The reply is blocking (observer has no send queue); |
| 2547 | // the exec is deferred to the run loop via pending_upgrade so | 2590 | // the exec is deferred to the run loop via pending_upgrade so |
| @@ -2589,11 +2632,6 @@ pub const Server = struct { | |||
| 2589 | self.dropObserver(i); | 2632 | self.dropObserver(i); |
| 2590 | self.pending_upgrade = .{ .path = path, .memfd = memfd }; | 2633 | self.pending_upgrade = .{ .path = path, .memfd = memfd }; |
| 2591 | }, | 2634 | }, |
| 2592 | // Where `mux d endpoint` actually lands, since it never attaches. | ||
| 2593 | .endpoint_req => { | ||
| 2594 | const payload = proto.encodeEndpointReply(self.endpointPort()); | ||
| 2595 | proto.writeFrameBounded(fd, .endpoint_reply, &payload, proto.reply_budget_ms) catch self.dropObserver(i); | ||
| 2596 | }, | ||
| 2597 | // Where `mux a status` actually lands: it asks and exits without | 2635 | // Where `mux a status` actually lands: it asks and exits without |
| 2598 | // ever attaching. Blocking reply for the same reason the stats | 2636 | // ever attaching. Blocking reply for the same reason the stats |
| 2599 | // and endpoint arms use one — an observer has no send queue. | 2637 | // and endpoint arms use one — an observer has no send queue. |
| @@ -2620,15 +2658,6 @@ pub const Server = struct { | |||
| 2620 | const payload = proto.encodeStatusReply(self.buildStatusReply(si)); | 2658 | const payload = proto.encodeStatusReply(self.buildStatusReply(si)); |
| 2621 | proto.writeFrameBounded(fd, .status_reply, &payload, proto.reply_budget_ms) catch self.dropObserver(i); | 2659 | proto.writeFrameBounded(fd, .status_reply, &payload, proto.reply_budget_ms) catch self.dropObserver(i); |
| 2622 | }, | 2660 | }, |
| 2623 | .end_req => { | ||
| 2624 | const v = self.endSession(frame.payload, null); | ||
| 2625 | var buf: [proto.end_reply_max_len]u8 = undefined; | ||
| 2626 | proto.writeFrameBounded(fd, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason), proto.reply_budget_ms) catch self.dropObserver(i); | ||
| 2627 | }, | ||
| 2628 | .sessions_req => { | ||
| 2629 | var buf: SessionsBuf = undefined; | ||
| 2630 | proto.writeFrameBounded(fd, .sessions_reply, self.sessions.text(&buf), proto.reply_budget_ms) catch self.dropObserver(i); | ||
| 2631 | }, | ||
| 2632 | else => {}, | 2661 | else => {}, |
| 2633 | } | 2662 | } |
| 2634 | } | 2663 | } |
| @@ -2663,21 +2692,6 @@ pub const Server = struct { | |||
| 2663 | try self.ses(si).eng.dumpPlain(self.alloc); | 2692 | try self.ses(si).eng.dumpPlain(self.alloc); |
| 2664 | } | 2693 | } |
| 2665 | 2694 | ||
| 2666 | // Observer replies go out unqueued, on `writeFrameBounded`. Observers | ||
| 2667 | // are local one-shot tools: `mux d dump`, `mux d stats` and `mux d | ||
| 2668 | // endpoint` read their one answer and exit, and `mux d stop` waits for | ||
| 2669 | // the socket to die rather than for a reply at all. So there is nothing | ||
| 2670 | // to queue into — an observer has no ClientSlot — but a dump can be | ||
| 2671 | // megabytes, so the write must tolerate a short one, and a peer that | ||
| 2672 | // stops reading must cost this reply's budget and not the daemon's | ||
| 2673 | // single loop. Past the budget the frame is truncated, which is why | ||
| 2674 | // every caller drops the connection on the error. | ||
| 2675 | fn replyDumpObserver(self: *Server, fd: std.posix.fd_t, payload: []const u8) !void { | ||
| 2676 | const dump = try self.buildDump(payload); | ||
| 2677 | defer self.alloc.free(dump); | ||
| 2678 | try proto.writeFrameBounded(fd, .dump_reply, dump, proto.reply_budget_ms); | ||
| 2679 | } | ||
| 2680 | |||
| 2681 | /// Take the grid to `cols` x `rows`. Returns whether the grid is now | 2695 | /// Take the grid to `cols` x `rows`. Returns whether the grid is now |
| 2682 | /// that size — false means the request was refused and nothing moved, | 2696 | /// that size — false means the request was refused and nothing moved, |
| 2683 | /// which is what keeps a refused size out of a client's slot. | 2697 | /// which is what keeps a refused size out of a client's slot. |
| @@ -3241,13 +3255,6 @@ pub const Server = struct { | |||
| 3241 | return @intCast(self.ses(si).eng.term.cols); | 3255 | return @intCast(self.ses(si).eng.term.cols); |
| 3242 | } | 3256 | } |
| 3243 | 3257 | ||
| 3244 | /// Unqueued and bounded — an observer has no send queue to accept into. | ||
| 3245 | /// See `statsText` for the format contract. | ||
| 3246 | fn replyStatsObserver(self: *Server, fd: std.posix.fd_t) !void { | ||
| 3247 | var buf: [stats_text_len]u8 = undefined; | ||
| 3248 | try proto.writeFrameBounded(fd, .stats_reply, try self.statsText(&buf), proto.reply_budget_ms); | ||
| 3249 | } | ||
| 3250 | |||
| 3251 | // The manifest an exec-ing daemon leaves for its replacement. The | 3258 | // The manifest an exec-ing daemon leaves for its replacement. The |
| 3252 | // writer's version and path are the rollback target's identity. | 3259 | // writer's version and path are the rollback target's identity. |
| 3253 | pub fn writeManifestTo( | 3260 | pub fn writeManifestTo( |