src/server/server.zig
Ref: Size: 146.8 KiB History
//! The daemon core (`mux d start`): up to `max_sessions` sessions, each an
//! engine, a pty and a command tracker, named at attach. One listener on the
//! unix socket and a second on UDP when QUIC is configured. Every state
//! update broadcasts to that session's clients, and each grid follows its
//! most recently active one (latest wins). Single-threaded: `pumpOnce` is one
//! poll iteration, so tests can drive the loop.
const std = @import("std");
const Engine = @import("engine").Engine;
const Pty = @import("pty").Pty;
const proto = @import("term").protocol;
const delta_mod = @import("engine").delta;
const DeltaTracker = delta_mod.DeltaTracker;
const cmdmod = @import("cmd.zig");
const shellint = @import("shellint.zig");
const sockpath = @import("sockpath");
const serve = @import("serve");
const quic = @import("quic");
const xdg = @import("xdg");
const server_os = @import("server_os");
// Re-exported for the daemon's own main (src/cli/main.zig) — the only
// consumer outside this folder; nobody else may know these exist.
pub const quic_server = @import("quic_server.zig");
pub const upgrade = @import("upgrade.zig");
const proxy = @import("proxy");
// The agent relay is a sub-file of this module, not a row of its own: it is
// one cluster of Server's state, and a module row would make it a seam
// anything in the build could name.
const agent_mod = @import("server_agent.zig");
const upgrade_ops = @import("server_upgrade.zig");
const SessionTable = @import("server_sessions.zig").SessionTable;
const AgentRelay = agent_mod.AgentRelay;
const AgentSock = agent_mod.AgentSock;
const forward_mod = @import("server_forward.zig");
pub const max_agent_chans = agent_mod.max_agent_chans;
/// One slot per ATTACH, not per session: every tile on a wall is its own
/// attach, and two clients watching one session spend two slots. At 8 an
/// ordinary wall refused its own next tile, because seven tiles plus a
/// QUIC host's once-a-second poll already held every slot. Matching
/// `max_sessions` and `wallview.max_tiles` means one full wall of 32 tiles
/// fits exactly — and leaves nothing over, so a SECOND wall on the same
/// daemon is still refused, which is the intended ceiling and not a bug.
pub const max_clients = 32;
pub const max_observers = 4;
pub const max_forward_peers = forward_mod.max_peers;
/// Every transport-level QUIC connection can briefly be unclassified. Making
/// this smaller than the listener silently narrows the existing concurrent
/// attach capacity before the first stream frame has a chance to claim a role.
pub const max_quic_pending = quic_server.max_conns;
/// One legal daemon reply frame: `debug_dump` may use protocol's entire
/// 16 MiB payload allowance. This is per provisional peer; together with the
/// listener's 256 KiB egress ring it bounds a peer at this plus `egress_cap`.
pub const quic_one_shot_pending_cap = proto.max_payload + proto.frame_header_len;
pub const quic_one_shot_deadline_ms: i64 = 10_000;
const QuicPending = struct {
id: u64,
inbound: std.ArrayList(u8) = .empty,
/// A one-shot reply is retained outside the terminal table until the
/// QUIC egress ring has accepted and acknowledged its complete frame.
/// This keeps large dumps truthful without lending observer verbs a
/// terminal slot.
pending: std.ArrayList(u8) = .empty,
one_shot: bool = false,
since_ms: i64,
};
/// A connection that has not attached yet: see `Server.observers`.
pub const Observer = struct {
fd: std.posix.fd_t,
inbound: std.ArrayList(u8) = .empty,
/// `monoMs` at the accept, or at the last whole frame; the idle
/// deadline measures from here.
since_ms: i64,
};
/// How long an observer may hold a slot without completing a frame. Without
/// it, four silent peers close every later attach at accept with no
/// diagnostic. 10 s is for a `--via` relay's first frame over a slow link.
pub const observer_idle_ms_default: i64 = 10_000;
/// How often `Server.watchSockPath` stats the socket path. A second is
/// well under the wall's own dial-then-auto-start, which is what a lost
/// path has to beat. The Server field of the same name is what the watch
/// reads, so a test can tick it in milliseconds instead of waiting out
/// three real seconds.
pub const sock_watch_interval_ms_default: i64 = 1000;
/// `watchSockPath`'s state: when it last looked, whether the path is
/// currently lost (so the loss logs once, not once a second), and the last
/// re-bind refusal (so that logs once per reason).
pub const SockWatch = struct {
checked_ms: i64 = 0,
lost: bool = false,
refused: ?anyerror = null,
};
/// Every line about the daemon's own socket, in one shape, on the
/// daemon's stderr — which `forkDetached` pointed at the xdg log. The
/// 2026-09-04 incident (a live daemon's socket file deleted, a second
/// daemon started on the path) left a log with nothing in it about
/// either; these lines are the trail that would have dated it
/// (issue 04b3019d).
pub fn logSocket(path: []const u8, comptime fmt: []const u8, args: anytype) void {
std.debug.print("mux d: socket {s}: " ++ fmt ++ "\n", .{path} ++ args);
}
/// The clock the daemon's bounded deadlines measure against: a calendar
/// step must not move one, and a boot's clock survives an upgrade's exec.
pub fn monoMs() i64 {
const t = std.posix.clock_gettime(.MONOTONIC) catch return std.time.milliTimestamp();
return @as(i64, t.sec) * 1000 + @divFloor(t.nsec, 1_000_000);
}
/// Observer frames one pump may answer. A 64 KB read can hold thousands of
/// five-byte `sessions_req`s, so without a cap a batching peer decides how
/// long every session on the box goes dark. The rest drain next pump.
pub const max_observer_frames_per_pump: usize = 16;
/// What one observer may hold part-received. Sized to the traffic this
/// endpoint carries — `stats_req` and `sessions_req` are empty, `end_req`
/// and `status_req` a name, and `upgrade_req` a path — not to
/// `max_payload`, the 16 MB paste ceiling of the CLIENT path.
pub const observer_inbound_max: usize = 8 * 1024;
/// Consecutive drain wakeups that may retire nothing before `drainPending`
/// concludes the peer is stuck. One proves nothing — a coalesced or
/// duplicate ack, or a bare flow-control update, wakes the loop without
/// retiring a byte. Bounded, so a stuck-writable socket cannot spin here.
const max_drain_stalls = 64;
/// The bound is on a RUN of unproductive wakeups, not their total: a slow peer
/// that keeps taking bytes is not stuck.
pub fn stallExhausted(stalls: *usize, progressed: bool) bool {
if (progressed) {
stalls.* = 0;
return false;
}
stalls.* += 1;
return stalls.* >= max_drain_stalls;
}
/// Convert the engine's authoritative extraction verdict to the wire. A
/// null status is the fourth wire case: the extraction itself could not be
/// performed, so there is no Engine result to inspect.
pub fn selectionReplyStatus(status: ?Engine.SelectionExtract.Status) proto.SelectionStatus {
const extracted = status orelse return .unavailable;
return switch (extracted) {
.ok => .ok,
.invalid => .invalid,
.too_large => .too_large,
};
}
/// Both bounds are load-bearing: sleeping past ngtcp2's expiry starves the PTO
/// timer, so a lost packet is never resent; an unfloored timeout spins hot on
/// an expiry ngtcp2 reports as already past.
pub fn drainWaitMs(remaining: i64, quic_hint: ?i32) i32 {
const rem: i32 = @intCast(@min(remaining, @as(i64, std.math.maxInt(i32))));
const hint = quic_hint orelse return rem;
return @max(1, @min(hint, rem));
}
/// Asked of the socket rather than remembered: a lazy bind names port 0 and
/// only getsockname knows what came back. 0 on failure, which is the same "no
/// port to announce" `endpoint_reply` already spells that way.
pub fn boundUdpPort(l: *quic_server.Listener) u16 {
var actual: std.posix.sockaddr.storage = undefined;
var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
std.posix.getsockname(l.pollFd(), @ptrCast(&actual), &len) catch return 0;
return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort();
}
/// The one occupied-slot count behind every `live` gauge: a loop per
/// table is a place per table to forget one.
pub fn countLive(slots: anytype) usize {
var n: usize = 0;
for (slots) |slot| {
if (slot != null) n += 1;
}
return n;
}
/// One read-only poll slot. A dead one polls -1, which poll(2) ignores.
fn pollIn(fd: std.posix.fd_t) std.posix.pollfd {
return .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 };
}
pub var shutdown_flag = std.atomic.Value(bool).init(false);
fn onShutdownSignal(_: c_int) callconv(.c) void {
shutdown_flag.store(true, .release);
}
/// Install SIGINT/SIGTERM handlers so a foreground `mux d start` shuts down
/// cleanly (socket file removed, shell reaped). Called by main; one test
/// borrows it for the SIGPIPE ignore.
pub fn installSignalHandlers() void {
var sa: std.posix.Sigaction = .{
.handler = .{ .handler = onShutdownSignal },
.mask = std.posix.sigemptyset(),
.flags = 0,
};
std.posix.sigaction(std.posix.SIG.INT, &sa, null);
std.posix.sigaction(std.posix.SIG.TERM, &sa, null);
// A client killed mid-run leaves a dead socket; the next snapshot
// write must fail with EPIPE, not deliver a fatal SIGPIPE.
proxy.ignoreSigpipe();
}
/// Where one client's bytes go, and where they come from. A QUIC client
/// shares ONE UDP socket with every other peer and is named by connection
/// ID, so "the client's fd" is not something that can be polled or sent to.
const Sink = union(enum) {
socket: std.posix.fd_t,
/// A QUIC peer: the listener that owns the shared UDP socket, plus the
/// connection id within it. Note what is NOT here — a descriptor.
quic: struct { listener: *quic_server.Listener, id: u64 },
/// -1 for QUIC: poll(2) ignores negative fds; its readability is the
/// listener's shared socket.
fn pollFd(self: Sink) std.posix.fd_t {
return switch (self) {
.socket => |fd| fd,
.quic => -1,
};
}
/// A short return leaves the rest in `pending` for the cap to judge; while
/// the QUIC arm accepted everything, `pending_cap` could never trip and
/// unbounded growth just moved into the listener.
fn send(self: Sink, bytes: []const u8) !usize {
return switch (self) {
.socket => |fd| server_os.sendNoSigNoWait(fd, bytes),
.quic => |q| q.listener.send(q.id, bytes),
};
}
/// Zero for a socket: once the kernel has it, it is the kernel's problem.
/// QUIC is not done until the peer acks.
pub fn inFlight(self: Sink) usize {
return switch (self) {
.socket => 0,
.quic => |q| q.listener.pendingBytes(q.id),
};
}
/// THIS CLIENT'S channel, never the transport it shares: one UDP socket
/// carries every QUIC client on this daemon.
pub fn close(self: Sink) void {
switch (self) {
.socket => |fd| std.posix.close(fd),
.quic => |q| q.listener.closeConn(q.id),
}
}
};
/// One attached interactive client: its connection plus whatever the daemon
/// still owes it.
const ClientSlot = struct {
sink: Sink,
/// Bytes that arrived for this client but do not yet form a whole
/// frame. Only the injection path fills this (see `pushInbound`): a
/// socket client reads whole frames straight off its fd and never
/// buffers here; a QUIC client's stream chunks land and wait here.
inbound: std.ArrayList(u8) = .empty,
/// The grid size this client last asked for *and got*: written only after
/// an `applySize` that succeeded, so it is never another client's size.
/// 0x0 until the first accepted attach, which `claimGrid` reads as no claim.
cols: u16 = 0,
rows: u16 = 0,
/// Frames queued but not yet accepted by the kernel. Bounded by
/// Server.pending_cap; a peer that stops reading gets dropped, never
/// waited on — one slow WAN client must not stall the session.
pending: std.ArrayList(u8) = .empty,
/// An await_req held open. At most one per client: a second one
/// replaces the first (the client is a serial CLI; queueing two would
/// be inventing a use case).
await_state: ?AwaitState = null,
/// Which session this client is attached to; null between a QUIC
/// handshake-promotion and its first attach. Only ever the INDEX — the
/// wire name dies with the frame it rode in.
session: ?usize = null,
/// Where this client sits in the daemon's activity order — see
/// `Server.activity_clock`. Still 0 means never attached, which is also
/// what a promoted-but-unattached QUIC slot's null `session` says.
activity: u64 = 0,
/// This client volunteered an SSH agent (`.agent_offer`). Opt-in and
/// per-connection: a redial is a new slot and must offer again.
agent_offer: bool = false,
selection: ?struct { id: u32, tracked: *Engine.TrackedSelection } = null,
fn clearSelection(self: *ClientSlot) void {
if (self.selection) |*selection| selection.tracked.deinit();
self.selection = null;
}
};
/// One client's outstanding await: the request as asked, plus the two pieces
/// of state resolving it needs to carry across pumps.
const AwaitState = struct {
since_seq: u64,
settle_ms: u32,
timeout_ms: u32,
/// milliTimestamp at acceptance; timeout measures from here.
started_ms: i64,
/// pgid fallback edge detector: set once the fg pgid has been seen off
/// the shell, so "back on the shell" means returned, not never-left. It
/// cannot say WHICH command returned — that needs marks.
saw_busy: bool = false,
};
// Matches `wallview.max_tiles`, and now `max_clients` too: 32 nameable
// sessions, and enough client slots that one wall can watch every one of
// them at once. The two are still different questions — a slot is spent
// per attach, so a session two clients hold costs one name and two slots,
// and the client table is what refuses first whenever anything beyond a
// single full wall dials in.
pub const max_sessions = 32;
/// The smallest grid a session may exist at, read by `resolveSession` and
/// `applySize` alike so a 1x1 attach cannot spawn a session no resize will
/// move. It lives in protocol.zig because the client's stripe floor is the
/// same contract on the other end of the wire.
pub const min_session_cols = proto.min_session_cols;
pub const min_session_rows = proto.min_session_rows;
/// One shell and everything the daemon knows about it. The command tracker
/// lives here and not on `Server` for the same reason the engine does: marks
/// are one shell's lifecycle, and two shells would interleave into nonsense.
pub const Session = struct {
eng: *Engine,
pty: Pty,
/// Row-level change tracking behind the delta stream.
tracker: DeltaTracker = .{},
/// Identifies this session instance in every snapshot it sends. Seqs mean
/// nothing across instances: a client quoting a dead epoch must be
/// snapshotted, never delta-served content it has never seen. Never 0 —
/// that is a client saying "I hold nothing".
epoch: u64,
/// This session's agent socket, or null when the daemon has no agent
/// directory. One listener per SESSION: the socket's path IS that
/// session's `SSH_AUTH_SOCK`, and an agent connection says nothing else
/// about who is calling. The listener and the path travel as one because
/// the unlink at teardown has to be guarded by what the listener was
/// bound to — see AgentSock.release. The path is owned and freed with
/// it in both teardown paths (Session.closeAgent).
agent_sock: ?AgentSock = null,
/// The pty's line-discipline bits as last put on the wire, or null
/// before the first poll. Deliberately "what clients have been told"
/// rather than "what the pty says": the two differ for exactly the span
/// of one pump, and closing that gap is the send.
mode_sent: ?proto.PtyModeFlags = null,
/// The terminal modes as last put on the wire, or null before the first
/// sample. Same discipline as mode_sent: what clients have been TOLD.
term_modes_sent: ?proto.TermModes = null,
/// The window title as last put on the wire. OWNED: the engine rewrites
/// its one title buffer in place, so a borrowed slice would compare the
/// new title against itself.
title_sent: ?[]const u8 = null,
/// The last side-channel event of each kind a reconnecting client might
/// still be owed, with the seq it happened at. One slot per EVENT kind,
/// not per clipboard target: two copies during a gap means the last wins,
/// and forty bells means one ding. The clipboard slot holds the USER'S
/// TEXT, so it is dropped as soon as its seq stops being servable.
pending_clipboard: ?PendingEvent = null,
pending_bell: ?PendingEvent = null,
/// The session's live command (OSC 133 marks). Seq-stamped copies of its
/// transitions are what cmd_state/await_reply/status_reply carry.
cmd: cmdmod.LiveCommand = .{},
/// The last completed command, frozen as it returned, and the one owner
/// of the return watermark: "a return happened at or before this seq" and
/// "here is what it was" are one fact, not two fields to keep in step.
/// An await asks about a PAST event, which `LiveCommand` by definition
/// cannot describe: a shell's `command_end` and the next `prompt_start`
/// arrive in a single write, so the live view loses the verdict a fraction
/// of a pump after it.
last_return: ?proto.CmdState = null,
/// `milliTimestamp` of the last byte the pty produced; the settle floor.
/// 0 is a session that has never spoken, which no elapsed silence should
/// be read as a finished command. Per-session: this shell's silence.
last_pty_ms: i64 = 0,
/// The `monoMs` at which the SIGTERM an accepted `end_req` sent stops being
/// waited on and `reap` sends SIGKILL. Null except between that accept and
/// the kill, so an accepted end is bounded and no shell can refuse to die.
end_by_ms: ?i64 = null,
/// The session's name, valid only up to name_len — the rest of the
/// buffer is undefined, so read it through name() and nowhere else.
/// From the attach-or-create task on, never empty on a live session:
/// the wire's "" maps to the default name before creation.
name_buf: [proto.session_name_max]u8 = undefined,
name_len: u8 = 0,
const PendingEvent = struct {
/// tracker.seq as of the drain that recorded it. A reattach is owed
/// the event only if this is ABOVE the seq it quotes.
seq: u64,
/// Owned; the wire payload exactly as drainSideEvents built it.
payload: []const u8,
};
pub fn name(self: *const Session) []const u8 {
return self.name_buf[0..self.name_len];
}
fn pendingSlot(self: *Session, kind: Engine.SideEvent.Kind) *?PendingEvent {
return switch (kind) {
.clipboard => &self.pending_clipboard,
.bell => &self.pending_bell,
};
}
pub const kinds = std.enums.values(Engine.SideEvent.Kind);
/// Derived from the enum: a new `SideEvent.Kind` fails to compile at
/// `pendingSlot` instead of leaking unreplayed. Enum order is
/// `replayPending`'s wire order.
pub fn pendingSlots(self: *Session) [kinds.len]*?PendingEvent {
var out: [kinds.len]*?PendingEvent = undefined;
inline for (kinds, 0..) |k, i| out[i] = self.pendingSlot(k);
return out;
}
/// Guarded by the expiry's predicate: an event recorded while `canServe`
/// is false could never be replayed.
pub fn recordPending(
self: *Session,
alloc: std.mem.Allocator,
kind: Engine.SideEvent.Kind,
payload: []const u8,
) void {
if (!self.tracker.canServe(self.tracker.seq)) return;
const owned = alloc.dupe(u8, payload) catch return;
const slot = self.pendingSlot(kind);
// The old one freed only once the new one exists, so an OOM leaves
// the slot holding something real rather than emptying it.
if (slot.*) |old| alloc.free(old.payload);
slot.* = .{ .seq = self.tracker.seq, .payload = owned };
}
/// Called wherever the tracker is rebuilt: a rebuild is the only thing
/// that moves reset_seq, and so the only thing that can put a recorded
/// event permanently out of reach.
pub fn dropUnservablePending(self: *Session, alloc: std.mem.Allocator) void {
for (self.pendingSlots()) |slot| {
const p = slot.* orelse continue;
if (self.tracker.canServe(p.seq)) continue;
alloc.free(p.payload);
slot.* = null;
}
}
pub fn freePending(self: *Session, alloc: std.mem.Allocator) void {
for (self.pendingSlots()) |slot| {
if (slot.*) |p| alloc.free(p.payload);
slot.* = null;
}
}
/// Free every allocation this Session owns EXCEPT two: the pty and the
/// agent socket. Those two are not memory, and each caller ends them
/// differently — `Server.deinit` reaps its child against a shared
/// deadline, `SessionTable.reap` deinits one pty on its own, and the
/// upgrade tests deliberately leave both descriptors open for the
/// adopting Server. So a caller pairs this with whatever pty and agent
/// teardown its path calls for, and anything else the session allocated
/// is handled here for all of them.
pub fn freeOwned(self: *Session, alloc: std.mem.Allocator) void {
self.tracker.deinit(alloc);
if (self.title_sent) |t| alloc.free(t);
self.freePending(alloc);
self.eng.deinit();
}
/// Give up this session's agent socket. Called from BOTH teardown
/// paths — a session reaped mid-life and the whole daemon exiting —
/// and idempotent, because a session that never got one is the
/// ordinary case, not an error.
pub fn closeAgent(self: *Session, alloc: std.mem.Allocator) void {
if (self.agent_sock) |*a| a.release(alloc);
self.agent_sock = null;
}
/// The listening descriptor, or -1 for a session that has none: the
/// pollfd table and the upgrade's cloexec sweep want a number, and -1
/// is what `pollIn` and those sweeps already read as "no fd here".
pub fn agentFd(self: *const Session) std.posix.fd_t {
const a = self.agent_sock orelse return -1;
return a.bound.fd;
}
/// The value the shell was handed as `SSH_AUTH_SOCK`, or null.
pub fn agentPath(self: *const Session) ?[:0]const u8 {
const a = self.agent_sock orelse return null;
return a.path;
}
};
pub const Server = struct {
alloc: std.mem.Allocator,
sessions: SessionTable = .{},
/// The one plan every session's shell is spawned from — this, not
/// `opts.shell`. Computed once and shared: `shellint.install` mints a
/// fresh directory per call, so a plan per session would orphan all but
/// the last from teardown. Every slice points into `shellint_arena`.
spawn_plan: SpawnPlan,
// Spawn inputs retained for the manifest: SpawnPlan is the computed
// result (argv/env after injection), not the inputs that produced it.
spawn_shell: [:0]const u8,
spawn_shell_integration: bool,
spawn_extra_env: []const Pty.EnvPair,
/// The listening socket and what its file was when we bound it, so
/// teardown can tell our socket from one that replaced it. One field
/// rather than a descriptor beside a std.net.Server holding the same
/// number: the fd has one owner, and `serve.Bound.close` is the only
/// thing that closes it. Not optional — init cannot return without one,
/// and an absent-means-false arm is the silent no-unlink 6090604 fixed.
bound: serve.Bound,
sock_path: []const u8,
/// The once-a-second check that `sock_path` still names `bound`, and
/// the re-bind when it does not. See `watchSockPath`.
sock_watch: SockWatch = .{},
/// The attached interactive clients, across every session; each sees
/// every update of the one session it is attached to.
clients: [max_clients]?ClientSlot = @splat(null),
/// Ticks once per activity verb, stamping ClientSlot.activity. Monotonic
/// and not a timestamp: two clients acting inside the same millisecond
/// must still order, and under test they routinely do.
activity_clock: u64 = 0,
/// How much unsent output one client may accumulate before the daemon
/// gives up on it. Overridden small in tests; 8 MiB is far more than a
/// live session ever queues, so tripping it means the peer is gone.
pending_cap: usize = 8 * 1024 * 1024,
/// Connections that have not attached (`mux d dump`, or a client waiting
/// to attach); an attach promotes one into a client slot. An fd and a
/// buffer, never a Sink: QUIC serves clients, and an observer is a local
/// one-shot tool that reads one answer over the unix socket and exits.
observers: [max_observers]?Observer = @splat(null),
observer_idle_ms: i64 = observer_idle_ms_default,
sock_watch_interval_ms: i64 = sock_watch_interval_ms_default,
/// `.none` is a first-class answer, not a failure: QUIC is opt-in per
/// invocation.
quic: union(enum) {
none,
borrowed: *quic_server.Listener,
owned: *quic_server.Listener,
} = .none,
/// Holds every string the shell-integration injection handed the spawn:
/// the shim directory's path, the argv, the env pairs. An arena because
/// they are allocated once, in init, and freed once, together.
shellint_arena: std.heap.ArenaAllocator,
/// The shim directory to remove at teardown, or null when nothing was
/// written — integration off, or a shell this daemon has no scripts for.
/// Distinct from "the arena is empty": only a directory that exists is
/// one we are responsible for deleting.
shellint_dir: ?[]const u8 = null,
/// The agent-forwarding relay: the channel table, its counters and the
/// directory the per-session sockets are bound in. See server_agent.zig
/// for why it is handed a `*Server` rather than holding one.
agents: AgentRelay = .{},
/// Forwarding-role connections have their own admission and never enter
/// the terminal client/session tables.
forwards: forward_mod.Relay,
/// Authenticated QUIC connections wait here only until their first frame
/// declares terminal/observer traffic or the dedicated forward role.
quic_pending: [max_quic_pending]?QuicPending = @splat(null),
/// Descriptors the manifest handed over that no session could adopt — an
/// agent listener whose socket file vanished mid-upgrade. Closing one
/// where the adopt fails would be wrong: until the last rollback point
/// is behind us, a rollback exec hands EVERY manifest fd back to the old
/// binary, which expects this one among them. So the close is deferred
/// to `sealAdoptedFds`. It cannot be skipped either — `execUpgrade`
/// cleared CLOEXEC on these fds, and no session names them any more, so
/// one left open is a listening socket inherited by every shell this
/// daemon forks for the rest of its life. At most one per session.
orphaned_fds: [max_sessions]std.posix.fd_t = @splat(-1),
orphaned_n: usize = 0,
stats: upgrade.Counters = .{},
// Set by the `upgrade_req` handler, checked by the run loop: the reply
// must drain before the exec, and a mid-handler exec would strand the
// observer fd and skip the close-all.
pending_upgrade: ?PendingUpgrade = null,
// The daemon's own version string, set at init from build_options.
// Stored on the struct so the upgrade handler can reach it without
// importing build_options (which would conflict with exe's own import).
version: []const u8 = "",
// What the run loop needs to exec: the candidate's path and the carrier
// holding the manifest (`server_os.anonFd`). Set by validateUpgrade +
// writeManifestTo.
const PendingUpgrade = struct {
path: []const u8,
carrier: std.posix.fd_t,
};
pub const Options = struct {
sock_path: []const u8,
shell: [:0]const u8,
cols: u16 = 80,
rows: u16 = 24,
/// Inject the OSC 133 mark scripts into the session shell. OFF by
/// default: the shim costs the user their `~/.zshenv` under zsh and
/// their DEBUG trap under bash, and only `mux a` reads what it buys.
shell_integration: bool = false,
/// Extra variables for the session shell, set after the injection's
/// own so a caller can override one. The shell-integration tests point
/// HOME at a temp dir with it, so no verdict depends on whose box ran.
extra_env: []const Pty.EnvPair = &.{},
// The daemon's own version, for the upgrade skew check.
version: []const u8 = "",
};
pub fn init(alloc: std.mem.Allocator, opts: Options) !Server {
// Before any request can ask: the comparison is against the image
// that BOOTED, not the first one asked about.
server_os.noteBootImage();
// Before the shell is spawned, so refusing costs nobody a fork and
// leaves no process to reap. `serve.bind` below runs the same refusal
// again, and the repeat is not redundant: it is the one that decides,
// covering the window this early check opens by refusing before the
// fork rather than at the bind.
const claimed = try sockpath.claim(opts.sock_path);
logSocket(opts.sock_path, "claimed ({s})", .{switch (claimed) {
.free => "nothing there",
.cleared_leftover => "cleared a dead daemon's leftover",
}});
// Shell integration, decided and written before the fork: whatever
// the child is going to be told has to exist on disk by the time it
// execs, and a failure here is still cheap — no process yet.
var shellint_arena = std.heap.ArenaAllocator.init(alloc);
errdefer shellint_arena.deinit();
const plan = try prepareSpawn(shellint_arena.allocator(), opts);
// Before the first session, because a session is born with its
// agent socket or without one for good — the env pair reaches the
// shell at the exec and never again.
const agent_dir = AgentRelay.makeDir(alloc, opts.sock_path);
errdefer if (agent_dir) |d| {
std.fs.cwd().deleteTree(d) catch {};
alloc.free(d);
};
var s0 = try SessionTable.create(
alloc,
plan,
proto.default_session,
opts.cols,
opts.rows,
AgentRelay.bindSock(alloc, agent_dir, proto.default_session),
);
errdefer {
s0.closeAgent(alloc);
s0.pty.deinit();
s0.eng.deinit();
}
const bound = try serve.bind(opts.sock_path, .{ .policy = .refuse_live });
logSocket(opts.sock_path, "bound dev={d} ino={d}", .{ bound.path_id.dev, bound.path_id.ino });
var srv: Server = .{
.alloc = alloc,
.spawn_plan = plan,
.spawn_shell = opts.shell,
.spawn_shell_integration = opts.shell_integration,
.spawn_extra_env = opts.extra_env,
.version = opts.version,
.bound = bound,
.sock_path = opts.sock_path,
.shellint_arena = shellint_arena,
.shellint_dir = plan.shellint_dir,
.agents = .{ .dir = agent_dir },
.forwards = forward_mod.Relay.init(alloc),
};
srv.sessions.table[0] = s0;
return srv;
}
pub const initFromManifest = upgrade_ops.initFromManifest;
/// What `Pty.spawnArgv` has to be handed, once shell integration has
/// had its say. Every slice points into the arena `prepareSpawn` was
/// given, which must outlive the spawn — spawnArgv reads all of it in
/// the child, after the fork.
pub const SpawnPlan = struct {
argv: [*:null]const ?[*:0]const u8,
env: []const Pty.EnvPair,
/// Straight from the injection: the shim directory to delete at
/// teardown, or null when nothing was written.
shellint_dir: ?[]const u8,
};
/// Turn the session options into that plan. Split out of `init` because
/// it is the one part of starting a daemon that is neither the engine,
/// the pty nor the listener, and inlining it buried those three.
fn prepareSpawn(a: std.mem.Allocator, opts: Options) !SpawnPlan {
// Beside the socket: that directory is already private, runtime-
// appropriate and per-user, which is what the shims need. What it is
// called and what to say on failure are `shellint.install`'s to report.
const injection: shellint.Injection = if (opts.shell_integration)
shellint.install(a, std.fs.path.dirname(opts.sock_path) orelse ".", opts.shell)
else
shellint.no_injection;
return planFrom(a, opts, injection);
}
/// The plan for an injection somebody else decided on.
pub fn planFrom(
a: std.mem.Allocator,
opts: Options,
injection: shellint.Injection,
) !SpawnPlan {
// Split from `prepareSpawn` for adoption: a resumed daemon must not
// mint a second shim directory. shellint and pty each speak their own
// `EnvPair` to stay leaves, so the mapping lives here.
const env = try a.alloc(Pty.EnvPair, injection.env.len + opts.extra_env.len + 1);
for (injection.env, env[0..injection.env.len]) |src, *dst| {
dst.* = .{ .key = src.key, .value = src.value };
}
// Last, so setenv's overwrite makes the caller's spelling the one
// the child sees (see the loop in Pty.spawnArgv).
@memcpy(env[injection.env.len..][0..opts.extra_env.len], opts.extra_env);
// After `extra_env`, not before: this is the daemon's identity, and a
// caller that could overwrite it would hand a shell a lie about where
// it is (`wallview.showsSelf` reads it back). One plan serves every
// session, so only the socket belongs here; `createSession` adds the name.
env[env.len - 1] = .{ .key = proto.sock_env, .value = try a.dupeZ(u8, opts.sock_path) };
// argv is the shell plus whatever the injection adds, null-terminated
// for execve. With no extra argv and no env this is the bare
// one-word argv, which is what keeps a /bin/sh session exactly the
// session it was before shell integration existed.
const argv = try a.allocSentinel(?[*:0]const u8, 1 + injection.extra_argv.len, null);
argv[0] = opts.shell.ptr;
for (injection.extra_argv, argv[1..]) |src, *dst| dst.* = src.ptr;
return .{ .argv = argv.ptr, .env = env, .shellint_dir = injection.dir };
}
pub fn deinit(self: *Server) void {
// A listener owns every QUIC connection, including provisional
// one-shot peers and forward-role peers. Announce the close before
// any of those owners quietly reap their entries, while their sink
// references and the borrowed listener are still live.
if (self.quicListener()) |listener| listener.closeAll();
self.forwards.deinit();
for (0..max_quic_pending) |i| self.dropQuicPending(i, true);
// Before the client slots, and silent: this loop frees the queues a
// notification would append to, and there is nobody left to tell —
// the daemon is going. Every channel, not a per-session sweep: what
// has to happen here is that no descriptor outlives the table.
for (0..max_agent_chans) |s| self.agents.closeChan(self, s, .silent);
for (0..self.clients.len) |i| self.teardownClient(i, true);
for (0..self.observers.len) |i| self.dropObserver(i);
// After the client slots, never before: a QUIC sink closes its
// connection THROUGH the listener, so the listener has to outlive
// the slots that hold it. That is the ordering main.zig's defers
// give the explicit --quic path, kept here for the lazy one.
switch (self.quic) {
// We bound it, so we return it.
.owned => |l| {
l.deinit();
self.quic = .none;
},
// Somebody else's, with its own deferred deinit still to run out
// there: freeing it here would be the double free.
.borrowed => {},
.none => {},
}
// Close, then unlink only if the path still names *our* socket — the
// guard and its rationale live in serve.Bound.close now. Which way
// it went is logged: "already gone" at a stop is the only trace a
// path deleted under a running daemon leaves once it exits.
logSocket(self.sock_path, "{s}", .{switch (self.bound.close(self.sock_path)) {
.unlinked => "unlinked",
.spared_successor => "left in place: it names another daemon's socket now",
.already_gone => "left in place: already gone",
.unlink_failed => "left in place: the unlink was refused; the next start will clear it as a dead leftover",
.closed_before => "already closed",
}});
// Two passes over one deadline: every child is asked to go before any
// is waited on, so a table of shells that ignore TERM costs one
// `term_grace_ms` and not one each.
for (&self.sessions.table) |*slot| {
if (slot.* != null) slot.*.?.pty.requestExit();
}
const reap_by = std.time.milliTimestamp() + Pty.term_grace_ms;
for (&self.sessions.table) |*slot| {
if (slot.* == null) continue;
const s = &slot.*.?;
// `reap` only signals and waits on the child, so it is free to
// run either side of the memory frees.
s.pty.reap(reap_by);
s.freeOwned(self.alloc);
s.closeAgent(self.alloc);
}
// After the ptys, so every shell is gone before the files it was
// reading are: a shim removed out from under a live shell would be
// a session that half-sourced its own integration. Best-effort —
// a daemon that cannot tidy /tmp must still exit.
if (self.shellint_dir) |dir| std.fs.cwd().deleteTree(dir) catch {};
// The sockets inside are already unlinked by the loop above; this
// is the directory itself, plus anything a failed unlink left.
if (self.agents.dir) |dir| {
std.fs.cwd().deleteTree(dir) catch {};
self.alloc.free(dir);
self.agents.dir = null;
}
self.shellint_arena.deinit();
}
/// Callers hold a live `si` or they do not call; nothing invents an index.
pub fn ses(self: *Server, si: usize) *Session {
return &self.sessions.table[si].?;
}
/// One poll iteration. A session whose shell exited is torn down first,
/// its clients told and dropped, and the pump carries on: no session's
/// death ends the daemon, so an emptied one keeps serving its socket.
pub fn pumpOnce(self: *Server, timeout_ms: i32) !void {
self.sessions.reap(self);
const listener_idx = max_sessions;
const client_base = max_sessions + 1;
const obs_base = client_base + max_clients;
const agent_listener_base = obs_base + max_observers;
const agent_chan_base = agent_listener_base + max_sessions;
const forward_base = agent_chan_base + max_agent_chans;
var fds: [
max_sessions + 1 + max_clients + max_observers +
max_sessions + max_agent_chans + forward_mod.poll_len + 1
]std.posix.pollfd = undefined;
for (&self.sessions.table, 0..) |*slot, si| {
fds[si] = pollIn(if (slot.*) |*s| s.pty.master else -1);
}
// The listener is polled only while an observer slot is free to
// land the connection in. With every slot taken, a peer waits in
// the kernel's listen backlog until a promotion or an idle drop
// frees one; polling the listener anyway would wake this loop
// every pass for a connection `acceptConn` could not take.
fds[listener_idx] = pollIn(if (self.freeObserverSlot() != null) self.bound.fd else -1);
for (&self.clients, 0..) |*slot, i| {
if (slot.*) |*c| {
// POLLOUT only while something is owed: asking for it on an
// idle client would spin the loop, since an empty socket
// buffer is always writable.
var events: i16 = std.posix.POLL.IN;
if (c.pending.items.len > 0) events |= std.posix.POLL.OUT;
fds[client_base + i] = .{ .fd = c.sink.pollFd(), .events = events, .revents = 0 };
} else {
fds[client_base + i] = pollIn(-1);
}
}
for (&self.observers, 0..) |*slot, i| {
fds[obs_base + i] = pollIn(if (slot.*) |*o| o.fd else -1);
}
// Each session's agent socket and every live channel on it. In the
// main poll rather than a loop of their own: an agent exchange is
// several round trips deep inside an ssh handshake, and a second
// loop would pay it a poll cycle per leg.
for (&self.sessions.table, 0..) |*slot, si| {
fds[agent_listener_base + si] = pollIn(if (slot.*) |*s| s.agentFd() else -1);
}
for (self.agents.chans, 0..) |slot, s| {
fds[agent_chan_base + s] = pollIn(if (slot) |ch| ch.fd else -1);
}
self.forwards.fillPoll(fds[forward_base .. forward_base + forward_mod.poll_len]);
// One extra descriptor for every QUIC client there will ever be:
// they share it, which is the whole reason a client slot cannot be
// a descriptor.
const quic_idx = fds.len - 1;
fds[quic_idx] = pollIn(if (self.quicListener()) |q| q.pollFd() else -1);
// ngtcp2's timers, folded in: the listener's earliest deadline
// shortens this poll, so retransmits and idle timeouts happen on
// time without a timerfd or a second loop.
var wait_ms = timeout_ms;
if (self.observerBacklog()) wait_ms = 0;
if (self.forwards.backlog()) wait_ms = 0;
if (self.quicListener()) |q| wait_ms = q.timeoutMs(wait_ms);
_ = try std.posix.poll(&fds, wait_ms);
if (self.quicListener()) |q| {
if (fds[quic_idx].revents != 0) q.readable();
// Unconditional: expiry work is due whether or not a packet
// arrived, and this is the only place it can happen.
q.tick();
// Acks are what free room in a connection's egress ring, and a
// QUIC client has no descriptor of its own to go writable: without
// this a filled ring waits for the next frame by coincidence.
self.flushQuicClients();
self.forwards.flushQuic();
}
for (&self.sessions.table, 0..) |*slot, si| {
if (slot.* == null) continue;
if (fds[si].revents == 0) continue;
const s = &slot.*.?;
var buf: [64 * 1024]u8 = undefined;
const n = std.posix.read(s.pty.master, &buf) catch 0;
if (n > 0) {
// Stamped on arrival, before anything is made of the bytes:
// the settle floor measures silence on the wire, not how
// long the engine took to digest what broke it.
s.last_pty_ms = std.time.milliTimestamp();
s.eng.feed(buf[0..n]);
self.flushPtyOutput(si);
self.sendUpdate(si);
self.drainMarkEvents(si);
self.drainSideEvents(si);
self.sampleTermModes(si);
self.sampleTermTitle(si);
}
}
// Outside the arm above: a program can call tcsetattr and print
// nothing, which is what every password prompt looks like from here.
// Checking only when the pty spoke would miss the quietest changes.
for (&self.sessions.table, 0..) |*slot, si| {
if (slot.* != null) self.pollPtyMode(si);
}
if (fds[listener_idx].revents & std.posix.POLL.IN != 0) self.acceptConn();
self.watchSockPath();
// Re-check each slot: an earlier arm may have dropped a client whose
// fd is still in `fds`. The null check suffices only because no slot
// can be REFILLED between poll and here — move the observer loop above
// this one and it stops being true.
for (0..max_clients) |i| {
if (self.clients[i] == null) continue;
const revents = fds[client_base + i].revents;
if (revents & std.posix.POLL.OUT != 0) self.flushClient(i);
// POLLOUT alone must NOT reach serviceClient: a read on a
// writable-but-silent socket has nothing to return and would
// block. Everything else (POLLIN, and the error/hangup bits the
// kernel reports unasked) is a read event as before.
const readable = revents & ~@as(i16, std.posix.POLL.OUT);
if (self.clients[i] != null and readable != 0) self.serviceClient(i);
}
for (0..max_observers) |i| {
const o = if (self.observers[i]) |*p| p else continue;
if (fds[obs_base + i].revents != 0) {
self.serviceObserver(i);
} else if (o.inbound.items.len > 0) {
// A capped drain's leftovers, answered without waiting for
// the peer to write again — it may never write again.
self.drainObserver(i);
}
}
self.reapIdleObservers(monoMs());
self.reapIdleQuicPending(monoMs());
self.flushQuicOneShots();
self.forwards.servicePoll(fds[forward_base .. forward_base + forward_mod.poll_len]);
// After the client arms, so a channel opened this pump is routed by
// the activity order those frames left. Servicing BEFORE accepting
// keeps `revents` and the table in step: accepting first could free
// and refill a slot in one pass, and the loop below would then read a
// descriptor on the strength of the previous occupant's poll.
for (0..max_agent_chans) |s| {
if (self.agents.chans[s] == null) continue;
if (fds[agent_chan_base + s].revents != 0) self.agents.service(self, s);
}
self.agents.sweepMute(self);
for (0..max_sessions) |si| {
if (self.sessions.table[si] == null) continue;
if (fds[agent_listener_base + si].revents & std.posix.POLL.IN != 0) {
self.agents.accept(self, si);
}
}
// After every arm that can move the session on, so an await sees THIS
// pump's marks and silence — and before the QUIC drain, which is what
// puts the frame it queues on the wire.
self.checkAwaits();
// Last in the pump: the listener's send only QUEUES — draining inside
// an ngtcp2 callback is the defect that avoids — so this is where a
// QUIC client's bytes actually leave, after all the frame handling.
if (self.quicListener()) |q| q.drainAll();
}
/// Pumps until something asks the daemon to stop. Always 0: getting here
/// means the daemon served, and a supervisor reads a nonzero exit on a
/// clean shutdown as a crash. Only a boot failure in main exits nonzero.
pub fn run(self: *Server) !u8 {
// The pump cannot start on a listener whose handler points anywhere
// but here. An adopted listener (initFromManifest) is built before
// the Server has its final address, and re-binding a listener the
// caller already wired is the same assignment twice.
if (self.quicListener()) |l| l.setHandler(self.quicHandler());
while (true) {
if (shutdown_flag.load(.acquire)) return 0;
// An upgrade was accepted: the reply has drained (pumpOnce
// ran the observer handler), so exec now. Shaped like
// shutdown_flag but per-instance because the exec carries the
// candidate's path and carrier.
if (self.pending_upgrade) |up| {
self.execUpgrade(up.path, up.carrier);
// execUpgrade only returns on failure; the daemon carries on.
continue;
}
try self.pumpOnce(100);
}
}
/// The observer slot the next accepted connection lands in, or null
/// when all four are taken. `pumpOnce` asks it before polling the
/// listener, so a burst of dials is never accepted faster than the
/// slots drain: the fifth peer of a burst waits in the kernel backlog
/// (128 deep) until a promotion or an idle drop frees a slot. Accepting
/// it and closing it, which is what this loop did until 2026-09-05,
/// lost the fifth tile of an eight-tile wall at birth with no
/// diagnostic, because eight pumps dial within a hundred microseconds
/// and the accept loop ran ahead of their attach frames — measured at
/// 118 of 240 attaches lost in eight-way bursts on a Debug daemon.
fn freeObserverSlot(self: *const Server) ?usize {
for (self.observers, 0..) |slot, i| {
if (slot == null) return i;
}
return null;
}
fn acceptConn(self: *Server) void {
const slot = self.freeObserverSlot() orelse return;
// CLOEXEC on the accepted fd, which is what std.net.Server.accept
// did here before: the daemon forks a shell per session and a live
// client connection leaked into one outlives every close of ours.
// Client connections are also the fds `mux d upgrade` DROPS, so
// unlike the listener this one has no reason to cross an exec.
const fd = std.posix.accept(self.bound.fd, null, null, std.posix.SOCK.CLOEXEC) catch return;
const conn: std.net.Stream = .{ .handle = fd };
// Nonblocking from the accept, and through the promotion into a
// client slot: a blocking fd lets one peer that stops reading stop the
// daemon's only loop. Every write here tolerates a short write.
setNonblocking(conn.handle) catch {
conn.close();
return;
};
self.observers[slot] = .{ .fd = conn.handle, .since_ms = monoMs() };
}
/// Once a second: does `sock_path` still name the socket we bound? A
/// unix listener survives the deletion of its path — the daemon keeps
/// every session and keeps listening on an inode nothing can reach by
/// name — and until this check existed that was permanent: nothing
/// re-bound, no admin verb takes anything but `--sock`, and the next
/// `mux` to dial the path found nothing and auto-started a second daemon
/// on it, stranding the first (issue 145807a2, 2026-09-04).
///
/// The re-bind goes through the same `serve.bind(.refuse_live)` a start
/// does, so "no socket stealing" holds by construction: a successor that
/// already holds the path is refused by `sockpath.claim` and this daemon
/// stays path-less, logs it once, and keeps looking — a successor's own
/// `mux d stop` unlinks its file, and the next tick takes the path back.
/// Only a path with NOTHING at it, or a dead socket file, is re-bound.
/// One `fstatat` per second is the whole cost.
fn watchSockPath(self: *Server) void {
const now = monoMs();
if (now - self.sock_watch.checked_ms < self.sock_watch_interval_ms) return;
self.sock_watch.checked_ms = now;
if (self.bound.path_id.stillAt(self.sock_path)) return;
if (!self.sock_watch.lost) {
self.sock_watch.lost = true;
if (sockpath.PathId.of(self.sock_path)) |other| {
logSocket(self.sock_path, "no longer names our listener (was dev={d} ino={d}, now dev={d} ino={d})", .{
self.bound.path_id.dev, self.bound.path_id.ino, other.dev, other.ino,
});
} else |_| {
logSocket(self.sock_path, "no longer names our listener (was dev={d} ino={d}, now missing)", .{
self.bound.path_id.dev, self.bound.path_id.ino,
});
}
}
const fresh = serve.bind(self.sock_path, .{ .policy = .refuse_live }) catch |err| {
// Once per reason, not once per second: a successor holding the
// path for a week would otherwise write 600k identical lines.
const e: anyerror = err;
if (self.sock_watch.refused == null or self.sock_watch.refused.? != e) {
self.sock_watch.refused = e;
logSocket(self.sock_path, "cannot re-bind: {s}", .{@errorName(e)});
}
return;
};
// Accepted connections are descriptors of their own and are not
// touched; the old listener's backlog held nothing reachable by
// name, which is the whole reason we are here.
std.posix.close(self.bound.fd);
self.bound = fresh;
self.sock_watch = .{ .checked_ms = now };
logSocket(self.sock_path, "re-bound dev={d} ino={d}", .{ fresh.path_id.dev, fresh.path_id.ino });
}
fn setNonblocking(fd: std.posix.fd_t) !void {
const fl = try std.posix.fcntl(fd, std.posix.F.GETFL, 0);
const nb: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
_ = try std.posix.fcntl(fd, std.posix.F.SETFL, fl | nb);
}
/// Exists so the relay never reaches into `ClientSlot`.
pub fn revokeAgentOffer(self: *Server, i: usize) void {
if (self.clients[i]) |*c| c.agent_offer = false;
}
/// Latest wins, as the grid does: the person typing is the person whose
/// agent signs. Decided once per connection — swapping identities under a
/// mid-exchange ssh fails the signature rather than moving it.
pub fn agentAnswerer(self: *const Server, si: usize) ?usize {
var best: ?usize = null;
for (self.clients, 0..) |c, i| {
const slot = c orelse continue;
if (slot.session != si or !slot.agent_offer) continue;
if (best == null or slot.activity > self.clients[best.?].?.activity) best = i;
}
return best;
}
/// Every client death comes through here, so no exit path can skip the
/// channel sweep and leave a channel routing into the next slot to take
/// it. `close_sink` is false where quic_server already freed the conn.
fn teardownClient(self: *Server, i: usize, close_sink: bool) void {
self.agents.closeOfClient(self, i);
if (self.clients[i]) |*slot| {
slot.clearSelection();
slot.pending.deinit(self.alloc);
slot.inbound.deinit(self.alloc);
if (close_sink) slot.sink.close();
}
self.clients[i] = null;
}
pub fn dropClient(self: *Server, i: usize) void {
self.teardownClient(i, true);
}
/// Queue one frame for the client in slot `i` and offer it to the kernel.
/// False if the client is gone, so callers can skip the accounting. The
/// only way anything reaches a client fd, and it never blocks: the daemon
/// is single-threaded and one stalled peer would freeze everyone.
pub fn queueFrame(self: *Server, i: usize, t: proto.MsgType, payload: []const u8) bool {
if (self.clients[i] == null) return false;
const slot = &self.clients[i].?;
proto.appendFrame(&slot.pending, self.alloc, t, payload) catch {
// A half-appended frame would corrupt every byte after it, so
// the connection cannot be salvaged.
self.dropClient(i);
return false;
};
// Checked after queueing, not before: the frame just added is what
// tips a hopeless client over, and the queue is the honest measure
// of how far behind it is.
if (slot.pending.items.len > self.pending_cap) {
self.dropClient(i);
return false;
}
self.flushClient(i);
return self.clients[i] != null;
}
pub fn queueSelectionReply(self: *Server, i: usize, reply: proto.SelectionReply) void {
var bytes: std.ArrayList(u8) = .empty;
defer bytes.deinit(self.alloc);
// Reserve the refusal before formatting a potentially large reply.
bytes.ensureTotalCapacity(self.alloc, proto.selection_reply_prefix_len) catch {
self.dropClient(i);
return;
};
proto.encodeSelectionReply(&bytes, self.alloc, reply) catch {
bytes.clearRetainingCapacity();
var refusal = reply;
refusal.status = .unavailable;
refusal.text = &.{};
proto.encodeSelectionReply(&bytes, self.alloc, refusal) catch unreachable;
};
_ = self.queueFrame(i, .selection_reply, bytes.items);
}
/// Wait up to `budget_ms` for clients to accept what they are owed. Only
/// the path where a client gets no next pump uses it — a session's death —
/// because there the alternative to a bounded wait is losing the frame.
pub fn drainPending(self: *Server, budget_ms: i64) void {
const deadline = std.time.milliTimestamp() + budget_ms;
var stalls: usize = 0;
while (true) {
// One extra slot for the QUIC listener: its readability is how
// acknowledgements arrive, and acknowledgements are the only
// thing that frees room in a connection's egress ring.
var fds: [max_clients + 1]std.posix.pollfd = undefined;
var slots: [max_clients]usize = undefined;
var n_sock: usize = 0;
var owed: usize = 0;
var quic_owed: usize = 0;
for (&self.clients, 0..) |*slot, i| {
if (slot.* == null) continue;
if (slot.*.?.sink == .quic) {
// A QUIC client has no descriptor to wait on, so its
// egress is driven directly. It owes in TWO places: bytes
// queued in the slot, and bytes the ring took but the peer
// has not acked. A send finishes when it is acked.
self.flushClient(i);
if (self.clients[i]) |*c| {
const still = c.pending.items.len + c.sink.inFlight();
owed += still;
quic_owed += still;
}
continue;
}
const c = &self.clients[i].?;
if (c.pending.items.len == 0) continue;
owed += c.pending.items.len;
slots[n_sock] = i;
fds[n_sock] = .{ .fd = c.sink.pollFd(), .events = std.posix.POLL.OUT, .revents = 0 };
n_sock += 1;
}
// Give QUIC egress a chance to leave before deciding there is
// nothing to wait for. `drainAll` as well as `tick`: tick only
// services connections whose timer is due.
if (self.quicListener()) |q| {
q.tick();
q.drainAll();
}
if (owed == 0) return;
const remaining = deadline - std.time.milliTimestamp();
if (remaining <= 0) return;
var n = n_sock;
var quic_idx: ?usize = null;
if (quic_owed > 0) {
if (self.quicListener()) |q| {
quic_idx = n;
fds[n] = .{ .fd = q.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
n += 1;
}
}
if (n == 0) return; // owed, but nothing left that could pay
const hint: ?i32 = if (self.quicListener()) |q|
q.timeoutMs(@intCast(@min(remaining, @as(i64, std.math.maxInt(i32)))))
else
null;
const ready = std.posix.poll(fds[0..n], drainWaitMs(remaining, hint)) catch return;
// A timeout is a verdict on this slice, not on the peer: the
// deadline at the top of the loop ends this. Returning here gives
// up exactly when a lossy path was about to make progress.
if (ready == 0) continue;
if (quic_idx) |qi| {
if (fds[qi].revents != 0) {
if (self.quicListener()) |q| q.readable();
}
}
for (fds[0..n_sock], slots[0..n_sock]) |pfd, i| {
// Non-POLLOUT revents (the peer hung up) reach flushClient
// too: its send fails and the client is dropped, which is
// also what stops this loop from retrying a dead fd.
if (pfd.revents != 0) self.flushClient(i);
}
// Guard against a poll that keeps reporting ready while nothing
// actually moves: without this the loop could spin hot until the
// deadline. Bounded by a RUN of unproductive wakeups rather than
// by one, because one proves nothing — see max_drain_stalls.
var still: usize = 0;
for (&self.clients) |*slot| {
if (slot.*) |*c| still += c.pending.items.len + c.sink.inFlight();
}
if (stallExhausted(&stalls, still < owed)) return;
}
}
/// Hand as much of a client's queue to the kernel as it will take right
/// now. WouldBlock leaves the remainder queued for the next POLLOUT; any
/// other error means the peer is unusable and the client is dropped.
pub fn flushClient(self: *Server, i: usize) void {
if (self.clients[i] == null) return;
const slot = &self.clients[i].?;
// A local offset, so a partial send never shifts the queue mid-loop.
var off: usize = 0;
while (off < slot.pending.items.len) {
const n = slot.sink.send(slot.pending.items[off..]) catch |err| switch (err) {
error.WouldBlock => break,
else => {
self.dropClient(i);
return;
},
};
if (n == 0) break; // no progress; wait for POLLOUT
off += n;
}
if (off == slot.pending.items.len) {
// Retaining the allocation is the right default, but a one-off
// burst would keep its buffer for the session's life, up to
// `pending_cap` x `max_clients`. Oversized buffers go back.
if (slot.pending.capacity > 64 * 1024) {
slot.pending.clearAndFree(self.alloc);
} else {
slot.pending.clearRetainingCapacity();
}
} else if (off > 0) {
// Compact once per flush rather than once per send, so
// pending.items.len always means "bytes still owed" — which is
// what the cap is checked against.
slot.pending.replaceRangeAssumeCapacity(0, off, &.{});
}
}
/// Offer every QUIC client's backlog to its ring again. Cheap when there
/// is nothing owed, which is the common case.
fn flushQuicClients(self: *Server) void {
for (&self.clients, 0..) |*slot, i| {
const cl = &(slot.* orelse continue);
if (cl.sink != .quic) continue;
if (cl.pending.items.len > 0) self.flushClient(i);
}
}
// ---- QUIC glue -------------------------------------------------------
// Three callbacks hold the entire difference between a QUIC client and a
// socket client. Everything past `pushInbound` has heard of neither.
fn quicOnOpen(ctx: *anyopaque, id: u64) void {
const self: *Server = @ptrCast(@alignCast(ctx));
const listener = self.quicListener() orelse return;
for (&self.quic_pending) |*slot| if (slot.* == null) {
slot.* = .{ .id = id, .since_ms = monoMs() };
return;
};
listener.closeConn(id);
}
fn quicOnData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
const self: *Server = @ptrCast(@alignCast(ctx));
if (self.forwards.pushQuic(id, bytes)) return;
if (self.slotForQuic(id)) |i| return self.pushInbound(i, bytes);
const qi = self.pendingForQuic(id) orelse return;
const q = &self.quic_pending[qi].?;
if (q.one_shot) return;
if (q.inbound.items.len + bytes.len > observer_inbound_max) return self.dropQuicPending(qi, true);
q.inbound.appendSlice(self.alloc, bytes) catch return self.dropQuicPending(qi, true);
const delimited = proto.delimitFrame(q.inbound.items) catch return self.dropQuicPending(qi, true);
const first = delimited orelse return;
if (first.type == .forward_hello) {
if ((proto.decodeForwardHello(first.payload) catch return self.dropQuicPending(qi, true)) != proto.forward_version)
return self.dropQuicPending(qi, true);
q.inbound.replaceRangeAssumeCapacity(0, first.consumed, &.{});
var moved = q.inbound;
q.inbound = .empty;
self.quic_pending[qi] = null;
if (self.forwards.adoptQuic(self.quicListener().?, id, moved) == .no_room) {
moved.deinit(self.alloc);
self.quicListener().?.closeConn(id);
}
return;
}
// Only ATTACH owns an interactive client slot. One-shot replies stay
// in this bounded provisional table until their queued frame drains.
if (first.type != .attach) {
const owned = (proto.takeFrame(self.alloc, &q.inbound) catch {
self.dropQuicPending(qi, true);
return;
}) orelse unreachable;
defer owned.deinit(self.alloc);
// Set this before any reply allocation. replyTo may OOM-drop this
// entry, so no code below may dereference `q` after it replies.
q.one_shot = true;
if (!self.handleDaemonVerb(.{ .quic_one = id }, owned))
self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
return;
}
// Validate before promotion. A malformed or unresolvable attach must
// not consume a terminal slot merely because its header was complete.
const req = proto.decodeAttach(first.payload) catch return self.dropQuicPending(qi, true);
const slot = self.freeClientSlot() orelse {
// replyTo can free this slot on OOM; mark before calling it.
q.one_shot = true;
self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
return;
};
const si = self.sessions.resolve(self, req.name, req.cols, req.rows) orelse {
q.one_shot = true;
self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
return;
};
const owned_attach = (proto.takeFrame(self.alloc, &q.inbound) catch return self.dropQuicPending(qi, true)) orelse unreachable;
defer owned_attach.deinit(self.alloc);
const moved = q.inbound;
q.inbound = .empty;
self.quic_pending[qi] = null;
self.clients[slot] = .{ .sink = .{ .quic = .{ .listener = self.quicListener().?, .id = id } }, .session = si, .inbound = moved };
self.seatClient(slot, si, req);
self.pushInbound(slot, &.{});
}
pub fn quicOnClose(ctx: *anyopaque, id: u64) void {
const self: *Server = @ptrCast(@alignCast(ctx));
if (self.forwards.closeQuic(id)) return;
if (self.pendingForQuic(id)) |i| return self.dropQuicPending(i, false);
const i = self.slotForQuic(id) orelse return;
// The conn is already freed, so there is no sink left to close.
self.teardownClient(i, false);
}
fn slotForQuic(self: *Server, id: u64) ?usize {
for (self.clients, 0..) |slot, i| {
const cs = slot orelse continue;
switch (cs.sink) {
.quic => |q| if (q.id == id) return i,
else => {},
}
}
return null;
}
fn pendingForQuic(self: *const Server, id: u64) ?usize {
for (self.quic_pending, 0..) |slot, i| if (slot != null and slot.?.id == id) return i;
return null;
}
fn dropQuicPending(self: *Server, i: usize, close_conn: bool) void {
if (self.quic_pending[i]) |*q| {
const id = q.id;
q.inbound.deinit(self.alloc);
q.pending.deinit(self.alloc);
self.quic_pending[i] = null;
if (close_conn) if (self.quicListener()) |listener| listener.closeConn(id);
}
}
pub fn quicOneShotExpired(since_ms: i64, now_ms: i64) bool {
return now_ms - since_ms >= quic_one_shot_deadline_ms;
}
/// The pre-allocation admission check for the one retained response.
/// Keeping it pure makes the 16 MiB boundary and overflow refusal pinable.
pub fn quicOneShotCanQueue(owed: usize, payload_len: usize) bool {
if (payload_len > proto.max_payload) return false;
const frame_len = std.math.add(usize, payload_len, proto.frame_header_len) catch return false;
return owed <= quic_one_shot_pending_cap and frame_len <= quic_one_shot_pending_cap - owed;
}
fn reapIdleQuicPending(self: *Server, now_ms: i64) void {
for (0..max_quic_pending) |i| {
const q = self.quic_pending[i] orelse continue;
// One-shots cannot be exempt forever: an ACK-withholding peer
// otherwise occupies both a listener connection and this role
// slot indefinitely. The explicit helper keeps this deterministic
// under test without a sleep.
if (if (q.one_shot) quicOneShotExpired(q.since_ms, now_ms) else now_ms - q.since_ms > self.observer_idle_ms)
self.dropQuicPending(i, true);
}
}
/// Re-offer queued one-shot frames as ACKs release the listener ring. A
/// close happens only after the whole frame is acknowledged, never after
/// a short enqueue that would leave a valid header with a truncated body.
fn flushQuicOneShots(self: *Server) void {
const listener = self.quicListener() orelse return;
for (0..max_quic_pending) |i| {
const q = if (self.quic_pending[i]) |*p| p else continue;
if (!q.one_shot) continue;
if (q.pending.items.len != 0) {
const n = listener.send(q.id, q.pending.items) catch {
self.dropQuicPending(i, false);
continue;
};
if (n != 0) q.pending.replaceRangeAssumeCapacity(0, n, &.{});
}
// The stop requester is its own shutdown acknowledgement: retain
// its peer until Server.deinit's closeAll sends CONNECTION_CLOSE.
// Quietly reaping it here turns a successful stop into an idle
// timeout at the client.
if (!shutdown_flag.load(.acquire) and q.pending.items.len == 0 and listener.pendingBytes(q.id) == 0)
self.dropQuicPending(i, true);
}
}
pub fn quicHandler(self: *Server) quic_server.Handler {
return .{
.ctx = self,
.onOpen = quicOnOpen,
.onData = quicOnData,
.onClose = quicOnClose,
};
}
/// Access does not transfer ownership. Deinit notifies peers through this
/// accessor, then frees only an owned listener after closing its clients.
pub fn quicListener(self: *const Server) ?*quic_server.Listener {
return switch (self.quic) {
.none => null,
.borrowed, .owned => |l| l,
};
}
/// Adopt a listener built by the caller (main.zig, or a test). The
/// server does not own the socket, only the reference — deinit leaves
/// the listener to its creator, which keeps the ownership story the
/// same as the unix listener's.
pub fn attachQuic(self: *Server, listener: *quic_server.Listener) void {
// Only before the loop runs: a lazily-bound listener in this field is
// ours to free, and overwriting it would drop that one into deinit's
// `.borrowed` arm, which frees nothing: the listener, its fd, and the
// bind latch it holds all leak.
std.debug.assert(self.quic == .none);
self.quic = .{ .borrowed = listener };
}
/// 0 means "could not", and the reason goes to the daemon log rather than
/// into the frame: the asker can do nothing with it but relay.
fn endpointPort(self: *Server) u16 {
return self.endpointPortFrom(
std.posix.getenv("MUX_KEY_FILE"),
std.posix.getenv("XDG_CONFIG_HOME"),
std.posix.getenv("HOME"),
);
}
/// Env handed in: tests cannot setenv. Refusals log as `mux d: <wire-verb>:
/// <what>`, the arriving verb.
pub fn endpointPortFrom(
self: *Server,
env_key: ?[]const u8,
xdg_config_home: ?[]const u8,
home: ?[]const u8,
) u16 {
if (self.quicListener()) |q| return boundUdpPort(q);
// The same three-way answer `mux d --quic` and every client get, so
// this cannot come to disagree with them about which file "the key"
// names. `xdg.pickKey` reads "set but empty" as unset.
var owned: ?[]const u8 = null;
defer if (owned) |p| self.alloc.free(p);
const key_path = switch (xdg.resolveKeyPathFrom(
self.alloc,
xdg.pickKey(null, env_key),
xdg_config_home,
home,
) catch {
// keyPathFrom refused, which means there was no HOME to build a
// default under in the first place — a different failure from an
// absent key, and it says so.
std.debug.print(
"mux d: endpoint_req: no key to listen with and no HOME to find one under (run `mux d keygen`)\n",
.{},
);
return 0;
}) {
.given => |g| g,
.default => |p| blk: {
owned = p;
break :blk p;
},
.missing => |p| {
owned = p;
std.debug.print(
"mux d: endpoint_req: no key at {s} (run `mux d keygen`)\n",
.{p},
);
return 0;
},
};
// The same refusals main.zig gives `--quic`, in the same words:
// `quic.keyRefusalBody` owns them and this site owns only the prefix.
// Every error routes there, catch-all included — nothing else in this
// expression can fail, so an unclassified one is still an unread key.
const key = quic.Key.load(key_path) catch |err| {
var buf: [quic.key_refusal_len]u8 = undefined;
std.debug.print(
"mux d: endpoint_req: {s}\n",
.{quic.keyRefusalBody(&buf, err, key_path)},
);
return 0;
};
return self.lazyBindQuic(key) catch |err| {
std.debug.print("mux d: endpoint_req: cannot bind udp: {s}\n", .{@errorName(err)});
return 0;
};
}
/// Bind 0.0.0.0 on a kernel-assigned port and wire it in. The poll loop
/// re-reads `quic` every iteration, so there is no loop surgery here —
/// setting the field IS the integration.
pub fn lazyBindQuic(self: *Server, key: quic.Key) !u16 {
const addr = try std.net.Address.parseIp("0.0.0.0", 0);
const l = try quic_server.Listener.bind(self.alloc, addr, key, quic.default_idle_ms);
l.setHandler(self.quicHandler());
self.quic = .{ .owned = l };
return boundUdpPort(l);
}
/// The one frame the QUIC path speaks before a client slot exists: the
/// full-session refusal. Fixed bytes, because there is no second pre-slot
/// frame to generalize for.
pub fn refusalFrame() [6]u8 {
var buf: [6]u8 = undefined;
buf[0] = @intFromEnum(proto.MsgType.exit_status);
std.mem.writeInt(u32, buf[1..5], 1, .little);
buf[5] = 1;
return buf;
}
// ---- pty mode -----------------------------------------------------
// The daemon holds the only fd that knows whether a keystroke will be
// echoed. Shipping that turns a client's local echo from a guess into a
// deduction: it is told what the terminal IS and decides for itself.
fn readPtyMode(self: *Server, si: usize) ?proto.PtyModeFlags {
// Not merely defensive: tcgetattr on a closed master is `unreachable`
// in std, so this is the difference between no answer and no daemon.
if (self.ptyFd(si) == null) return null;
const m = self.ses(si).pty.mode() catch return null;
return .{ .icanon = m.icanon, .echo = m.echo };
}
/// Read the pty's line discipline and, if it moved since clients were
/// last told, tell them. One tcgetattr per pump: polling is the only
/// mechanism there is, since nothing notifies us of another process.
fn pollPtyMode(self: *Server, si: usize) void {
const flags = self.readPtyMode(si) orelse return;
if (self.ses(si).mode_sent) |prev| {
// Compared as the byte that goes on the wire, so a bit that gets
// defined later is covered by this the day it exists rather than
// the day someone remembers to extend the comparison.
if (@as(u8, @bitCast(prev)) == @as(u8, @bitCast(flags))) return;
}
self.ses(si).mode_sent = flags;
const payload = proto.encodePtyMode(flags);
for (0..max_clients) |i| {
if (self.inSession(i, si)) _ = self.queueFrame(i, .pty_mode, &payload);
}
}
/// Tell one client what the pty is doing right now. The broadcast above
/// only fires on a CHANGE, so without this a client joining a session
/// sitting quietly at a prompt would wait forever to learn the mode.
fn sendPtyModeTo(self: *Server, si: usize, i: usize) void {
const flags = self.ses(si).mode_sent orelse blk: {
// Reached whenever an attach lands before the mode was ever
// polled: a QUIC handshake and attach completing in one pump get
// here with `mode_sent` still null. Recorded as sent, so no
// broadcast repeats the value to the client just handed it.
const f = self.readPtyMode(si) orelse return;
self.ses(si).mode_sent = f;
break :blk f;
};
_ = self.queueFrame(i, .pty_mode, &proto.encodePtyMode(flags));
}
/// Gated on marks_seen: a push means a mark was read, never a heuristic
/// guess. Sent after the resync — start_row and end_row point into a grid
/// the client must already hold.
fn sendCmdStateTo(self: *Server, si: usize, i: usize) void {
if (!self.ses(si).cmd.marks_seen) return;
_ = self.queueFrame(i, .cmd_state, &proto.encodeCmdState(self.cmdState(si, .marks)));
}
/// The one filter every per-session broadcast applies. A promoted-but-
/// unattached slot holds no session and receives nothing until it says
/// which shell it wants.
fn inSession(self: *const Server, i: usize, si: usize) bool {
const slot = self.clients[i] orelse return false;
return (slot.session orelse return false) == si;
}
pub fn hasClientsIn(self: *const Server, si: usize) bool {
for (0..max_clients) |i| {
if (self.inSession(i, si)) return true;
}
return false;
}
/// Null from `endSession`'s hang-up until `reap` clears the slot.
fn ptyFd(self: *Server, si: usize) ?std.posix.fd_t {
const fd = self.ses(si).pty.master;
return if (fd >= 0) fd else null;
}
fn clientsIn(self: *const Server, si: usize, exclude: ?usize) u8 {
comptime {
// `+=` below, not `+|=`: a saturating add implies a bound that
// can be reached. It cannot — but only while this holds, and
// `others` is a u8 on the wire either way.
std.debug.assert(max_clients <= std.math.maxInt(u8));
}
var n: u8 = 0;
for (0..max_clients) |i| {
if (exclude == i) continue;
if (self.inSession(i, si)) n += 1;
}
return n;
}
/// Accepting hangs up and starts a clock. `reap` still owns the teardown,
/// exactly as for a shell that ended itself, so every attached client gets
/// the same exit_status — but a shell free to ignore TERM and HUP would
/// otherwise outlive the session it was asked to end.
fn endSession(self: *Server, payload: []const u8, exclude: ?usize) proto.EndReply {
if (payload.len < proto.end_req_len) return .{ .accepted = false, .others = 0, .reason = proto.end_reason.bad_frame };
const force = payload[0] & 1 != 0;
const si = self.sessions.find(payload[proto.end_req_len..]) orelse
return .{ .accepted = false, .others = 0, .reason = proto.end_reason.no_session };
const others = self.clientsIn(si, exclude);
if (others > 0 and !force) return .{ .accepted = false, .others = others, .reason = proto.end_reason.others_attached };
self.ses(si).pty.requestExit();
// Only while no deadline is already pending: re-arming would let a
// client asking again every 400ms keep a stubborn shell past every
// deadline it was ever given.
if (self.ses(si).end_by_ms == null)
self.ses(si).end_by_ms = monoMs() + Pty.term_grace_ms;
// An accepted end cancels a pending upgrade: `validateUpgrade`
// decides a pump before `run` execs, and an end accepted in between
// leaves the exec calling `clearCloexec(-1)` — a panic, not a refusal.
self.cancelUpgrade();
return .{ .accepted = true, .others = others, .reason = proto.end_reason.accepted };
}
fn cancelUpgrade(self: *Server) void {
const up = self.pending_upgrade orelse return;
self.alloc.free(up.path);
std.posix.close(up.carrier);
self.pending_upgrade = null;
}
fn freeClientSlot(self: *const Server) ?usize {
for (self.clients, 0..) |slot, i| {
if (slot == null) return i;
}
return null;
}
fn dropObserver(self: *Server, i: usize) void {
if (self.observers[i]) |*o| {
o.inbound.deinit(self.alloc);
std.posix.close(o.fd);
}
self.observers[i] = null;
}
/// One read per readiness, for both connection tables: `.bytes` arrived,
/// `.again` is a readiness with nothing behind it, `.gone` is EOF or a
/// dead socket. Stated once so client and observer cannot drift apart.
const ReadOutcome = union(enum) { bytes: []u8, again, gone };
fn readConn(fd: std.posix.fd_t, buf: []u8) ReadOutcome {
const n = std.posix.read(fd, buf) catch |e| switch (e) {
error.WouldBlock => return .again,
else => return .gone,
};
return if (n == 0) .gone else .{ .bytes = buf[0..n] };
}
/// A socket client is readable: one read, then whatever whole frames
/// that made.
fn serviceClient(self: *Server, i: usize) void {
const fd = switch (self.clients[i].?.sink) {
.socket => |fd| fd,
// A QUIC client has no descriptor of its own to read: its bytes
// arrive from the shared UDP socket and enter through
// pushInbound. Poll never reports it readable (pollFd is -1), so
// reaching here would mean the poll bookkeeping had drifted.
.quic => return,
};
// One read per readiness: a partial frame waits in `inbound`.
var buf: [64 * 1024]u8 = undefined;
switch (readConn(fd, &buf)) {
.bytes => |b| self.pushInbound(i, b),
.again => {},
.gone => self.dropClient(i),
}
}
/// Feed bytes that arrived for client `i`, extracting whole frames as
/// they complete. Every transport lands here, so from the frame onward
/// they take byte-identical paths. Only the byte-split test below asks
/// for the adversarial chunk boundaries a live link rarely produces.
pub fn pushInbound(self: *Server, i: usize, bytes: []const u8) void {
if (self.clients[i] == null) return;
self.clients[i].?.inbound.appendSlice(self.alloc, bytes) catch {
// The peer is sending faster than we can hold; dropping it is
// the same answer a corrupt queue gets on the way out.
self.dropClient(i);
return;
};
while (true) {
if (self.clients[i] == null) return; // a handler dropped it
const frame = proto.takeFrame(self.alloc, &self.clients[i].?.inbound) catch {
self.dropClient(i);
return;
} orelse return; // header or payload still coming
defer frame.deinit(self.alloc);
self.handleFrame(i, frame);
}
}
/// Everything a client can ask for, once its bytes are a frame. Slot `i`
/// is live on entry — `pushInbound` re-reads it before every dispatch,
/// because a handler can drop the client mid-loop — so the arms below
/// index with `.?` rather than each re-testing it.
fn handleFrame(self: *Server, i: usize, frame: proto.Frame) void {
if (self.handleDaemonVerb(.{ .client = i }, frame)) return;
switch (frame.type) {
.attach => self.onAttach(i, frame),
.resize => self.onResize(i, frame),
.input => self.onInput(i, frame),
.fetch_scrollback => self.onFetchScrollback(i, frame),
.selection_req => self.onSelectionReq(i, frame.payload),
.detach => self.dropClient(i),
.status_req => self.onStatusReq(i, frame),
.await_req => self.onAwaitReq(i, frame),
.agent_offer => self.onAgentOffer(i),
.agent_data => self.onAgentData(i, frame),
.agent_close => self.onAgentClose(i, frame),
else => {},
}
}
/// Whichever table the asker sits in.
const Peer = union(enum) {
client: usize,
observer: usize,
/// A first-frame QUIC daemon verb. It is never a terminal client.
quic_one: u64,
};
/// The one send for a daemon verb; see handleDaemonVerb for why the two
/// arms differ and why a short write costs the connection.
fn replyTo(self: *Server, p: Peer, t: proto.MsgType, payload: []const u8) void {
switch (p) {
.client => |i| _ = self.queueFrame(i, t, payload),
.observer => |i| {
const o = self.observers[i] orelse return;
proto.writeFrameBounded(o.fd, t, payload, proto.reply_budget_ms) catch
self.dropObserver(i);
},
.quic_one => |id| {
// A one-shot owns exactly one reply. Check before append so
// an oversized/failed allocation never grows its ArrayList;
// the legal 16 MiB debug dump plus its header still fits.
const qi = self.pendingForQuic(id) orelse return;
const q = &self.quic_pending[qi].?;
if (!quicOneShotCanQueue(q.pending.items.len, payload.len)) {
self.dropQuicPending(qi, true);
return;
}
proto.appendFrame(&q.pending, self.alloc, t, payload) catch self.dropQuicPending(qi, true);
},
}
}
/// exit_status 1 is the only "no" this wire has; the close follows it.
fn refuseObserver(self: *Server, i: usize) void {
self.replyTo(.{ .observer = i }, .exit_status, &.{1});
self.dropObserver(i);
}
/// An upgrade refusal: byte 1, then the reason in words, then the close
/// — a `mux d upgrade` that will not happen has nothing further to say
/// on this socket. Five sites spelled this out, each remembering the
/// drop for itself.
fn refuseUpgrade(self: *Server, i: usize, reason: []const u8) void {
var reply: [proto.upgrade_reply_max_len]u8 = undefined;
self.replyTo(.{ .observer = i }, .upgrade_reply, proto.encodeUpgradeReply(&reply, .{
.ok = false,
.reason = reason,
}));
self.dropObserver(i);
}
fn dropPeer(self: *Server, p: Peer) void {
switch (p) {
.client => |i| self.dropClient(i),
.observer => |i| self.dropObserver(i),
.quic_one => |id| if (self.quicListener()) |listener| listener.closeConn(id),
}
}
/// The verbs whose answer is the DAEMON rather than the connection:
/// identical bytes for an attached client and a one-shot `mux d` tool.
/// False for a verb that does depend on who asked. The two owners differ
/// only in how an answer LEAVES, which is all `replyTo` is: a client has a
/// send queue, an observer gets a bounded write and a truncation drops it.
fn handleDaemonVerb(self: *Server, p: Peer, frame: proto.Frame) bool {
switch (frame.type) {
.stats_req => {
// Daemon-global, and the same text for whoever asks: an
// attached client's own session no longer picks out a single
// seq column, since every live session gets its own line now
// (see statsText).
var buf: [stats_text_len]u8 = undefined;
const text = self.statsText(&buf) catch {
self.dropPeer(p);
return true;
};
self.replyTo(p, .stats_reply, text);
},
.sessions_req => {
// Daemon-global like `.stats_req`: the answer is the session
// TABLE, so who asked cannot change it. An observation, not
// activity, so it claims no grid.
var buf: [proto.sessions_reply_max]u8 = undefined;
const names = self.sessions.text(buf[0..proto.sessions_text_max]);
// The holds lines and the meta line ride the same gate: a
// daemon with no version to state appends nothing, so a bare
// fixture's payload stays byte-identical to the old wire and
// the sibling exact-equality test keeps pinning it.
var len = names.len;
if (self.version.len != 0) {
for (self.sessions.table, 0..) |slot, si| {
const s = slot orelse continue;
const holds: u8 = @intCast(@min(self.clientsInSession(si), std.math.maxInt(u8)));
len = proto.appendSessionsHolds(&buf, len, s.name(), holds);
}
len = proto.appendSessionsMeta(&buf, len, self.version, server_os.selfImageStale());
}
self.replyTo(p, .sessions_reply, buf[0..len]);
},
.endpoint_req => {
const payload = proto.encodeEndpointReply(self.endpointPort());
self.replyTo(p, .endpoint_reply, &payload);
},
.status_req => switch (p) {
.quic_one => {
const si = self.sessions.find(frame.payload) orelse {
self.replyTo(p, .exit_status, &.{1});
return true;
};
const payload = proto.encodeStatusReply(self.buildStatusReply(si));
self.replyTo(p, .status_reply, &payload);
},
else => return false,
},
.debug_dump => {
// A read against a NAME, not a question about this
// connection's session: an attached client may peek at any
// live one. `buildDump` owns the resolution and the refusal.
const dump = self.buildDump(frame.payload) catch {
self.dropPeer(p);
return true;
};
defer self.alloc.free(dump);
self.replyTo(p, .dump_reply, dump);
},
.create_req => {
// Resolution and creation share the daemon thread. A competing
// creator is refused without attaching to or resizing its shell.
const outcome: proto.CreateReply = blk: {
const req = proto.parseCreateReq(frame.payload) catch
break :blk .{ .status = .refused, .reason = "invalid request" };
if (req.cols < min_session_cols or req.rows < min_session_rows or req.cols > proto.max_cols)
break :blk .{ .status = .refused, .reason = "invalid size" };
if (self.sessions.find(req.name) != null)
break :blk .{ .status = .exists, .reason = "session name already exists" };
_ = self.sessions.resolve(self, req.name, req.cols, req.rows) orelse
break :blk .{ .status = .refused, .reason = "session could not be created" };
break :blk .{ .status = .created, .reason = "" };
};
var buf: [proto.create_reply_max_len]u8 = undefined;
self.replyTo(p, .create_reply, proto.encodeCreateReply(&buf, outcome.status, outcome.reason));
},
.end_req => {
// The asking client is excluded from the "others hold it"
// count; an observer holds no session and excludes nobody.
const v = self.endSession(frame.payload, switch (p) {
.client => |i| i,
.observer, .quic_one => null,
});
var buf: [proto.end_reply_max_len]u8 = undefined;
self.replyTo(p, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason));
},
// The flag SIGTERM sets, not a second shutdown path, so the poll
// loop and both teardowns stay already-tested code. The ack is the
// socket dying, which is what `mux d stop` polls for.
.stop_req => shutdown_flag.store(true, .release),
else => return false,
}
return true;
}
fn onAttach(self: *Server, i: usize, frame: proto.Frame) void {
// Not an unconditional snapshot: a QUIC connection is promoted to a
// client slot at handshake, so its FIRST attach arrives here. Snapshot
// -serving them all would deny every QUIC client a delta resume.
const req = proto.decodeAttach(frame.payload) catch return;
// Re-resolved on every attach, because an attach on an
// established connection IS the reconnect path — the name
// is looked up fresh each time. Only the INDEX is stored:
// req.name borrows the frame's payload and dies with it.
const si = self.sessions.resolve(self, req.name, req.cols, req.rows) orelse {
// Same honest no a full client table gets: exit_status
// 1, and the peer decides what to do with the line.
_ = self.queueFrame(i, .exit_status, &.{1});
return;
};
if (self.clients[i]) |*c| {
c.clearSelection();
// Seq series are per-session, so an await's `since_seq` from the
// old one would answer instantly or never, arbitrarily. Only on a
// real change: a reconnect re-resolving the same name keeps its await.
if (c.session != si) c.await_state = null;
c.session = si;
}
self.seatClient(i, si, req);
}
/// The seating both attach arms end on, past every refusal so `attaches`
/// never grows for an attach that seated nobody. Size is recorded only
/// when the grid really went there, and modes precede the state they describe.
fn seatClient(self: *Server, i: usize, si: usize, req: proto.AttachReq) void {
self.stats.attaches += 1;
self.bumpActivity(i);
const size_changed = (req.cols != self.colsNow(si) or req.rows != self.rowsNow(si));
const applied = self.applySize(si, req.cols, req.rows);
if (applied) self.recordSize(si, i);
self.sendPtyModeTo(si, i);
self.sendResync(si, i, req.have_seq, req.have_epoch, size_changed and applied);
self.sendCmdStateTo(si, i);
}
fn onResize(self: *Server, i: usize, frame: proto.Frame) void {
// Latest wins: whoever resized last sets the grid, and the
// broadcast snapshot tells every other client about it.
const si = self.clients[i].?.session orelse return;
const sz = proto.decodeSize(frame.payload) catch return;
self.bumpActivity(i);
if (self.applySize(si, sz.cols, sz.rows)) self.recordSize(si, i);
self.resyncSnapshot(si);
}
fn onInput(self: *Server, i: usize, frame: proto.Frame) void {
// Latest wins follows ACTIVITY, and typing is activity: the console
// you type at claims the grid. Scrollback, stats and detach are
// deliberately not — paging history from a small terminal would yank
// the grid from whoever is working. Claimed before the bytes go out,
// so the shell reacts at the size the typist is watching.
const si = self.clients[i].?.session orelse return;
self.bumpActivity(i);
self.claimGrid(si, i);
const fd = self.ptyFd(si) orelse return;
proto.writeAllFd(fd, frame.payload) catch self.dropClient(i);
}
fn onFetchScrollback(self: *Server, i: usize, frame: proto.Frame) void {
// Answered on this connection only: scroll position is
// client-local, so one client paging history never
// disturbs another's live stream.
const si = self.clients[i].?.session orelse return;
const req = proto.decodeScrollbackReq(frame.payload) catch return;
const got = self.ses(si).eng.encodeScrollback(self.alloc, req.start, req.count) catch return;
defer self.alloc.free(got.bytes);
const payload = self.alloc.alloc(u8, 6 + got.bytes.len) catch return;
defer self.alloc.free(payload);
// The echoed header is what the encoder CLAMPED to, not what was
// asked for: the client decodes exactly `count` rows out of the body,
// and a request past the end of history answers with fewer.
@memcpy(payload[0..6], &proto.encodeScrollbackReq(got.first, got.count));
@memcpy(payload[6..], got.bytes);
_ = self.queueFrame(i, .scrollback_chunk, payload);
self.sendSelectionState(si, i);
}
fn onSelectionReq(self: *Server, i: usize, payload: []const u8) void {
const req = proto.decodeSelectionReq(payload) catch return;
const si = self.clients[i].?.session orelse {
self.queueSelectionReply(i, .{ .id = req.id, .gesture = req.gesture, .history_rows = 0 });
return;
};
const s = self.ses(si);
if (req.action == .clear) {
if (self.clients[i].?.selection) |selection| {
if (selection.id == req.gesture) self.clients[i].?.clearSelection();
}
return;
}
var state = self.selectionState(si, i);
state.id = req.id;
state.gesture = req.gesture;
if (req.action == .start) {
const tracked = if (req.gesture != 0 and req.epoch == s.epoch and req.source == s.eng.selectionSource())
s.eng.trackSelection(req.anchor, req.active) catch null
else
null;
if (tracked) |selection| {
self.clients[i].?.clearSelection();
self.clients[i].?.selection = .{ .id = req.gesture, .tracked = selection };
state = self.selectionState(si, i);
state.id = req.id;
} else {
state.status = .unavailable;
self.queueSelectionReply(i, state);
return;
}
}
const result: ?Engine.SelectionExtract = if (req.action == .extract)
s.eng.extractSelection(self.alloc, req.anchor.row, req.anchor.col, req.active.row, req.active.col, proto.selection_text_max) catch null
else if (self.clients[i].?.selection) |*selection|
if (selection.id == req.gesture and req.epoch == s.epoch)
selection.tracked.extract(self.alloc, proto.selection_text_max) catch null
else
null
else
null;
defer if (result) |value| value.deinit(self.alloc);
if (result) |value| {
state.status = selectionReplyStatus(value.status);
state.text = value.text orelse &.{};
} else state.status = .unavailable;
self.queueSelectionReply(i, state);
}
fn selectionState(self: *Server, si: usize, i: usize) proto.SelectionReply {
const s = self.ses(si);
var state: proto.SelectionReply = .{
.seq = s.tracker.seq,
.source = s.eng.selectionSource(),
.history_rows = s.eng.historyRows(),
};
if (self.clients[i].?.selection) |*selection| {
state.gesture = selection.id;
if (selection.tracked.points()) |points| {
state.status = .ok;
state.anchor = points.anchor;
state.active = points.active;
} else self.clients[i].?.clearSelection();
}
return state;
}
fn sendSelectionState(self: *Server, si: usize, i: usize) void {
if (!self.inSession(i, si)) return;
self.queueSelectionReply(i, self.selectionState(si, i));
}
fn onStatusReq(self: *Server, i: usize, frame: proto.Frame) void {
const si = self.clients[i].?.session orelse {
// A session-less slot has nothing to compare a tail against, so
// the tail IS the question — an observer's `status_req`, reached
// over QUIC. A one-shot ask against a name, so the resolved index
// is used and never stored on the slot.
const found = self.sessions.find(frame.payload) orelse {
const name = SessionTable.safeName(frame.payload);
// Same wording as the observer arm: one stderr line
// an operator can grep for, since status_reply is a
// fixed binary layout with no room for words.
std.debug.print("mux d: status_req for unknown session: {s}\n", .{name});
// The one word the wire has for no. Before the drop, not
// instead of it: silence made `mux a status` report a dead
// daemon. It outlives the drop only inside an ngtcp2 callback,
// where the close defers to `.closing_quiet` and `reapClosing`
// puts the byte on the wire.
_ = self.queueFrame(i, .exit_status, &.{1});
self.dropClient(i);
return;
};
const payload = proto.encodeStatusReply(self.buildStatusReply(found));
_ = self.queueFrame(i, .status_reply, &payload);
return;
};
// One rule, no aliasing: a non-empty tail must name the slot's OWN
// session or the frame is ignored — never answered against the tail.
// The client asked two questions at once, and neither reading wins.
if (frame.payload.len != 0 and !std.mem.eql(u8, frame.payload, self.ses(si).name())) return;
const payload = proto.encodeStatusReply(self.buildStatusReply(si));
_ = self.queueFrame(i, .status_reply, &payload);
}
fn onAwaitReq(self: *Server, i: usize, frame: proto.Frame) void {
const req = proto.decodeAwaitReq(frame.payload) catch return;
const si = self.clients[i].?.session orelse return;
// Same one rule as status_req's, on the tail decodeAwaitReq
// already borrowed out as `.name`.
if (req.name.len != 0 and !std.mem.eql(u8, req.name, self.ses(si).name())) return;
self.clients[i].?.await_state = .{
.since_seq = req.since_seq,
.settle_ms = req.settle_ms,
.timeout_ms = req.timeout_ms,
.started_ms = std.time.milliTimestamp(),
};
// A return that already happened answers immediately, which is what
// makes a reconnect re-issue safe — and keeps that true independent
// of the pump-end pass's ordering.
self.checkAwaits();
}
fn onAgentOffer(self: *Server, i: usize) void {
// Idempotent by design: a redial re-offers on a slot that may already
// be flagged, including one `sweepMuteAgentChans` took away — a
// reattach may bring a working agent.
self.clients[i].?.agent_offer = true;
}
fn onAgentData(self: *Server, i: usize, frame: proto.Frame) void {
const id = proto.decodeAgentId(frame.payload) catch return;
// An id naming no channel of THIS client's is dropped in silence: a
// refusal that told "not yours" from "no such channel" apart would
// tell a client what other clients hold.
const s = self.agents.find(id, i) orelse return;
// The cap, enforced on the side that did not choose it. The
// blocking write below is argued from agent traffic being
// small — a property of the peer, which is exactly the kind
// of thing that has to be checked rather than trusted.
if (proto.agentDataOversize(frame.payload)) {
self.agents.closeChan(self, s, .notify);
return;
}
// A reply proves; a volley before any question does not, or
// a mute peer clears the clock by talking first.
if (self.agents.chans[s].?.answer == .asked) self.agents.chans[s].?.answer = .proven;
// The daemon's one deliberate blocking write. "Never block on a
// client" holds because a client is a stranger across a link; this
// peer dialled a socket in `makeAgentDir`'s 0700 directory, so it is
// same-uid and could already signal this daemon. Agent traffic is
// small request-response, so the wedge to guard against does not exist.
proto.writeAllFd(self.agents.chans[s].?.fd, frame.payload[proto.agent_id_len..]) catch
self.agents.closeChan(self, s, .notify);
}
fn onAgentClose(self: *Server, i: usize, frame: proto.Frame) void {
const id = proto.decodeAgentId(frame.payload) catch return;
// Silent: the client asked for this, so an agent_close back
// would be an echo it has to learn to ignore.
if (self.agents.find(id, i)) |s| self.agents.closeChan(self, s, .silent);
}
/// Resolve any awaits that can be answered this pump, at the run loop's
/// 100ms granularity. Each client resolves against its OWN slot's
/// session: another session's return must never answer its await.
fn checkAwaits(self: *Server) void {
const now = std.time.milliTimestamp();
// One ioctl per SESSION being awaited on, not per waiting client: the
// foreground process group belongs to a session's pty, so every client
// of it reads the same number. Null covers both "nobody asked" and
// "the ioctl failed". `saw_busy` stays per-client: the transition is its own.
var wanted: [max_sessions]bool = @splat(false);
for (&self.clients) |*cslot| {
if (cslot.*) |*c| {
if (c.await_state != null) {
if (c.session) |si| wanted[si] = true;
}
}
}
var fg_pgids: [max_sessions]?std.posix.pid_t = @splat(null);
for (&self.sessions.table, 0..) |*sslot, si| {
if (!wanted[si]) continue;
if (sslot.*) |*s| {
if (!s.cmd.marksOpen()) fg_pgids[si] = s.pty.fgPgid() catch null;
}
}
for (0..max_clients) |i| {
if (self.clients[i] == null) continue;
const slot = &self.clients[i].?;
if (slot.await_state == null) continue;
// Unreachable: `.await_req` only accepts from a slot with a
// session, and no client survives its session's exit. Kept because
// a slot that lost one has nothing an answer could be about.
const si = slot.session orelse continue;
const fg_pgid = fg_pgids[si];
const a = &slot.await_state.?;
// 1. Marks: a return newer than `since_seq` answers with the full
// story. Strictly greater — `since_seq` is "what I have".
// The WATERMARK alone decides, never live phase: a shell's
// `D;code`+`A` burst lands in one pty read, so phase is back to
// `at_prompt` before this line runs.
if (self.ses(si).last_return) |st| {
if (st.seq > a.since_seq) {
self.answerAwait(i, st, .returned);
continue;
}
}
// 2. pgid: only when marks do not hold the floor (folded into
// fg_pgid above). The shell is the session leader, so its pid
// is the resting pgid.
if (fg_pgid) |pg| {
if (pg != self.ses(si).pty.child) {
a.saw_busy = true;
} else if (a.saw_busy) {
// The pgid went out and came back: something ran and is
// over. WHAT its code was, this mechanism cannot say.
const st = self.fallbackState(si, .pgid, .{
.phase = .returned,
.clear_exit_code = true,
});
self.answerAwait(i, st, .returned);
continue;
}
}
// 3. Settle: output silence, if the caller asked for a floor.
if (a.settle_ms > 0 and self.ses(si).last_pty_ms > 0 and
now - self.ses(si).last_pty_ms >= a.settle_ms and
now - a.started_ms >= a.settle_ms)
{
// Phase is left as the session's own: silence says the
// output stopped, never that a command returned.
const st = self.fallbackState(si, .settle, .{ .clear_exit_code = true });
self.answerAwait(i, st, .settled);
continue;
}
// 4. Timeout: the bound the client set on the whole wait. 0 is
// not a zero-length deadline but the absence of one — such an
// await ends only when marks, the pgid or settle end it.
if (a.timeout_ms > 0 and now - a.started_ms >= a.timeout_ms) {
// The phase stays the session's own — a timeout reports where
// things stand — but the code goes: nothing returned during
// THIS wait, so any code still standing is the previous
// command's verdict, and `mechanism: "marks"` would sell it.
const st = self.fallbackState(
si,
if (self.ses(si).cmd.marks_seen) .marks else .pgid,
.{ .clear_exit_code = true },
);
self.answerAwait(i, st, .timeout);
continue;
}
}
}
fn answerAwait(self: *Server, i: usize, st: proto.CmdState, reason: proto.AwaitReason) void {
if (self.clients[i] == null) return;
self.clients[i].?.await_state = null;
const payload = proto.encodeAwaitReply(st, reason);
_ = self.queueFrame(i, .await_reply, &payload);
}
/// An observer is readable: one read, then every whole frame that made.
fn serviceObserver(self: *Server, i: usize) void {
const o = &self.observers[i].?;
var buf: [64 * 1024]u8 = undefined;
const bytes = switch (readConn(o.fd, &buf)) {
.bytes => |b| b,
.again => return,
.gone => {
self.dropObserver(i);
return;
},
};
o.inbound.appendSlice(self.alloc, bytes) catch {
self.dropObserver(i);
return;
};
if (o.inbound.items.len > observer_inbound_max) {
self.dropObserver(i);
return;
}
self.drainObserver(i);
}
/// Frames off an observer's buffer until it holds only a partial one,
/// the slot was dropped, or an attach promoted it — after which any
/// bytes behind the attach are the CLIENT's and go through pushInbound.
fn drainObserver(self: *Server, i: usize) void {
for (0..max_observer_frames_per_pump) |_| {
const o = if (self.observers[i]) |*p| p else return;
const frame = proto.takeFrame(self.alloc, &o.inbound) catch {
self.dropObserver(i);
return;
} orelse return;
defer frame.deinit(self.alloc);
o.since_ms = monoMs();
self.handleObserverFrame(i, frame);
}
}
/// Frames already read but unanswered, the cap having stopped a drain.
/// Owed NOW: a peer that has said all it will say never wakes the poll.
fn observerBacklog(self: *const Server) bool {
for (self.observers) |slot| {
const o = slot orelse continue;
if (o.inbound.items.len > 0) return true;
}
return false;
}
/// Drop every observer past `observer_idle_ms` without a whole frame.
/// Runs once per pump, which is why the pump's poll is bounded (100 ms)
/// rather than indefinite.
fn reapIdleObservers(self: *Server, now_ms: i64) void {
for (0..max_observers) |i| {
const o = self.observers[i] orelse continue;
if (now_ms - o.since_ms > self.observer_idle_ms) self.dropObserver(i);
}
}
/// One observer frame. Slot `i` is live on entry; the `.attach` arm is
/// the one that ends the slot without dropping it.
fn handleObserverFrame(self: *Server, i: usize, frame: proto.Frame) void {
if (frame.type == .forward_hello) {
const version = proto.decodeForwardHello(frame.payload) catch return self.dropObserver(i);
if (version != proto.forward_version) return self.dropObserver(i);
const o = &self.observers[i].?;
const fd = o.fd;
var moved = o.inbound;
o.inbound = .empty;
self.observers[i] = null;
if (self.forwards.adoptSocket(fd, moved) == .no_room) {
moved.deinit(self.alloc);
std.posix.close(fd);
}
return;
}
if (self.handleDaemonVerb(.{ .observer = i }, frame)) return;
const fd = self.observers[i].?.fd;
switch (frame.type) {
.attach => {
const sz = proto.decodeAttach(frame.payload) catch {
self.dropObserver(i);
return;
};
// Attach joins the session; it no longer displaces whoever
// was already there (takeover is retired; attach joins).
const slot = self.freeClientSlot() orelse {
// Client table full: the client exits nonzero rather than
// hanging on a silent socket. Checked BEFORE the name
// resolves, so an unseatable client spawns no shell.
return self.refuseObserver(i);
};
// Attach-or-create; null is the refusal (bad name, session
// table full, or an unknown name with no size to create
// at), answered exactly like the full-table no above.
const si = self.sessions.resolve(self, sz.name, sz.cols, sz.rows) orelse {
return self.refuseObserver(i);
};
// Promote: the buffer moves with the fd.
const moved = self.observers[i].?.inbound;
self.observers[i] = null;
self.clients[slot] = .{ .sink = .{ .socket = fd }, .session = si, .inbound = moved };
self.seatClient(slot, si, sz);
// Frames the same write carried behind the attach (a wall
// tile sends agent_offer on its heels) are handled now, as
// the client they were addressed to.
self.pushInbound(slot, &.{});
},
.detach => self.dropObserver(i),
// Where `mux d upgrade` lands: validate, reply, and let the run
// loop exec. Deferred via `pending_upgrade` so the close-all and
// execve happen outside the observer's read cycle.
.upgrade_req => {
const req = proto.parseUpgradeReq(frame.payload) catch return self.refuseUpgrade(i, "bad frame");
if (self.validateUpgrade(req, self.version)) |reason| {
defer self.alloc.free(reason);
return self.refuseUpgrade(i, reason);
}
// Accepted: write the manifest to its carrier (not CLOEXEC —
// the new binary must inherit it), reply, and arm the exec.
const carrier = server_os.anonFd("mux-upgrade") catch return self.refuseUpgrade(i, "carrier");
self.writeManifestTo(carrier, self.version) catch {
std.posix.close(carrier);
return self.refuseUpgrade(i, "manifest");
};
// Owned, because `req.path` points into the frame payload
// this handler's caller frees on return, and the exec runs a
// pump later. execUpgrade frees it if the exec fails.
const path = self.alloc.dupe(u8, req.path) catch {
std.posix.close(carrier);
return self.refuseUpgrade(i, "oom");
};
var accepted: [1]u8 = undefined;
self.replyTo(.{ .observer = i }, .upgrade_reply, proto.encodeUpgradeReply(&accepted, .{
.ok = true,
.reason = "",
}));
self.dropObserver(i);
self.pending_upgrade = .{ .path = path, .carrier = carrier };
},
// Where `mux a status` actually lands: it asks and exits without
// ever attaching. Blocking reply for the same reason the stats
// and endpoint arms use one — an observer has no send queue.
.status_req => {
// The whole payload is the name; empty means the default
// session. Unlike an attached client's status_req, there is
// no slot to fall back to and nothing here to compare the
// tail against — the tail IS the question.
const si = self.sessions.find(frame.payload) orelse {
const name = SessionTable.safeName(frame.payload);
// `status_reply` is a fixed binary layout with no room for
// words, so the answer is the `exit_status` 1 a refused
// attach gets and the name goes on stderr. Bounded write:
// the byte must be gone before the fd, not at the pump's price.
std.debug.print("mux d: status_req for unknown session: {s}\n", .{name});
return self.refuseObserver(i);
};
const payload = proto.encodeStatusReply(self.buildStatusReply(si));
proto.writeFrameBounded(fd, .status_reply, &payload, proto.reply_budget_ms) catch self.dropObserver(i);
},
else => {},
}
}
/// `payload` is 1 byte (0 = plain, 1 = vt) ++ an optional session-name
/// tail; empty names the default. Resolved via `findSession`, never
/// `resolveSession`: a read that could spawn a shell would make
/// `mux d dump --session typo` a way to stand one up. An unknown name
/// answers IN WORDS, because `dump_reply` is the only frame this verb
/// gets. The name quoted is the RESOLVED one. Caller owns the result.
fn buildDump(self: *Server, payload: []const u8) ![]const u8 {
const want_vt = payload.len >= 1 and payload[0] == 1;
const wire_name = if (payload.len >= 2) payload[1..] else "";
const si = self.sessions.find(wire_name) orelse {
// Through `safeName` like the two `status_req` arms: `mux d dump`
// prints this straight to a console, and one rule for "a name in a
// message" beats remembering which site was the safe one.
const name = SessionTable.safeName(wire_name);
return std.fmt.allocPrint(self.alloc, "mux d: no such session: {s}\n", .{name});
};
return if (want_vt)
try self.ses(si).eng.dumpVt(self.alloc)
else
try self.ses(si).eng.dumpPlain(self.alloc);
}
/// Take the grid to `cols` x `rows`. Returns whether the grid is now
/// that size — false means the request was refused and nothing moved,
/// which is what keeps a refused size out of a client's slot.
pub fn applySize(self: *Server, si: usize, cols: u16, rows: u16) bool {
// A degenerate size (0x0 pty, buggy client) would trip engine
// asserts; keep the current grid instead. Same constant
// resolveSession creates by — see min_session_cols.
if (cols < min_session_cols or rows < min_session_rows) return false;
self.ses(si).eng.resize(cols, rows) catch return false;
self.ses(si).pty.resize(cols, rows) catch {};
// A resize is the one engine event that answers the app without a
// feed (the mode-2048 size report), so it flushes on its own.
self.flushPtyOutput(si);
return true;
}
/// Whatever the engine owes the app — query answers, size reports —
/// goes to the pty now. Best-effort: a pty that will not take it is a
/// session on its way out.
fn flushPtyOutput(self: *Server, si: usize) void {
const s = self.ses(si);
const resp = s.eng.ptyOutput();
if (resp.len == 0) return;
proto.writeAllFd(s.pty.master, resp) catch {};
s.eng.clearPtyOutput();
}
/// ONLY after an applySize that returned true. After a refusal you record
/// some other client's size, which this client then claims the moment it
/// typed.
fn recordSize(self: *Server, si: usize, i: usize) void {
if (self.clients[i] == null) return;
self.clients[i].?.cols = self.colsNow(si);
self.clients[i].?.rows = self.rowsNow(si);
}
/// Latest wins on activity: bring the grid to client `i`'s size if it
/// isn't there already, and tell everyone. Same discontinuity path a
/// resize takes, because that is exactly what this is — the difference
/// is only what triggered it.
fn claimGrid(self: *Server, si: usize, i: usize) void {
if (self.clients[i] == null) return;
const slot = &self.clients[i].?;
if (slot.cols == self.colsNow(si) and slot.rows == self.rowsNow(si)) return;
// A client that has never had a size accepted is still 0x0 and
// claims nothing — this is what stops a degenerate attacher from
// broadcasting a repaint on every keystroke for a resize that
// cannot happen. Two compares and a refusal, once per keystroke.
if (!self.applySize(si, slot.cols, slot.rows)) return;
self.resyncSnapshot(si);
}
/// Called from the three activity verbs — attach, input, resize — and
/// nowhere else. Four call sites, not three: attach has two arms.
fn bumpActivity(self: *Server, i: usize) void {
if (self.clients[i] == null) return;
self.activity_clock += 1;
self.clients[i].?.activity = self.activity_clock;
}
/// Accrued once per update, not once per recipient, so the delta-vs-
/// snapshot ratio means the same however many clients attach. Pays a full
/// serialization per update purely to measure the saving; fine for a
/// prototype.
fn accrueSnapshotEquiv(self: *Server, si: usize) void {
// Built and thrown away: the stat means "what a full snapshot would
// have cost", so it has to measure the payload this daemon would
// actually have sent rather than any other serialization of the grid.
if (self.buildSnapshotPayload(si)) |payload| {
self.stats.snapshot_equiv_bytes += payload.len;
self.alloc.free(payload);
} else |_| {}
}
/// The only place a send is counted, so the four paths cannot drift on
/// what a byte means. `to` is one client, null every client of THIS
/// session. Counted bytes exclude the 5-byte header here and in the
/// counterfactual alike (~8% understated), as the bench measures it.
fn countedSend(self: *Server, si: usize, t: proto.MsgType, payload: []const u8, to: ?usize) void {
var sent = false;
for (0..max_clients) |i| {
if (if (to) |only| i != only else !self.inSession(i, si)) continue;
if (!self.queueFrame(i, t, payload)) continue;
self.sendSelectionState(si, i);
sent = true;
if (t == .delta) {
self.stats.deltas += 1;
self.stats.delta_bytes += payload.len;
} else {
self.stats.snapshots += 1;
self.stats.snapshot_bytes += payload.len;
}
}
if (!sent) return;
if (t == .delta) self.accrueSnapshotEquiv(si) else self.stats.snapshot_equiv_bytes += payload.len;
}
fn sendDeltaTo(self: *Server, si: usize, i: usize, payload: []const u8) void {
self.countedSend(si, .delta, payload, i);
}
fn broadcastDelta(self: *Server, si: usize, payload: []const u8) void {
self.countedSend(si, .delta, payload, null);
}
/// Per-update path: diff and broadcast a delta; discontinuities resync.
/// The tracker advances with nobody attached, so a quiet detacher can
/// still be caught up — but WITHOUT diffing, since rendering every row
/// to hash it would be thrown away. The returner gets a whole-grid delta.
pub fn sendUpdate(self: *Server, si: usize) void {
const s = self.ses(si);
// Before the diff, not after: `update()` renders every row to hash
// it, and with nobody attached that payload has nowhere to go — ~60%
// of the daemon's cycles on a full-width repaint.
if (!self.hasClientsIn(si)) {
// Still a discontinuity's caller: resyncSnapshot's rebuild is
// documented to happen with nobody attached, so the tracker is
// usable for the next one to arrive.
if (s.tracker.noteBlind(s.eng) == .discontinuity) self.resyncSnapshot(si);
return;
}
const upd = s.tracker.update(self.alloc, s.eng) catch return;
switch (upd) {
.none => for (0..max_clients) |i| self.sendSelectionState(si, i),
.discontinuity => self.resyncSnapshot(si),
.advanced => {
// Losing the payload would strand the clients a seq behind
// with no later delta covering these rows; resync instead.
const payload = s.tracker.buildDeltaSince(
self.alloc,
s.eng,
s.tracker.seq - 1,
) catch {
self.resyncSnapshot(si);
return;
};
defer self.alloc.free(payload);
self.broadcastDelta(si, payload);
},
}
}
/// Fold the engine's OSC 133 events into the command tracker and tell
/// clients about transitions. After `sendUpdate`, so `tracker.seq`
/// already covers the same pty chunk. The pump's pty-read arm is the only
/// caller: marks from a test that feeds an engine directly sit pending.
fn drainMarkEvents(self: *Server, si: usize) void {
const s = self.ses(si);
for (s.eng.markEvents()) |ev| {
const tr = s.cmd.apply(ev) orelse continue;
if (tr == .returned) {
// Copied out of the tracker while it still describes this
// command — the `A` that ends the burst is usually the very
// next event in this same loop, and the next command's `C`
// clears the exit code outright.
s.last_return = .{
.phase = .returned,
.mechanism = .marks,
.exit_code = s.cmd.exit_code,
.start_row = s.cmd.start_row,
.end_row = s.cmd.end_row,
.seq = s.tracker.seq,
};
}
if (tr == .reset) continue; // believed nothing, tell no one
// The session's own clients alone: a mark is one shell's
// lifecycle, and a push crossing sessions would tell a client
// about a command it cannot see.
const payload = proto.encodeCmdState(self.cmdState(si, .marks));
for (0..max_clients) |i| {
if (self.inSession(i, si)) _ = self.queueFrame(i, .cmd_state, &payload);
}
}
s.eng.clearMarkEvents();
}
/// Ship the engine's side-channel events to THIS session's clients, in
/// the pty-read arm beside `drainMarkEvents` because the events describe
/// the chunk just fed. A clipboard push crossing sessions would set the
/// user's clipboard from a session they are not looking at.
fn drainSideEvents(self: *Server, si: usize) void {
const s = self.ses(si);
// No bail when the session has no clients: with nobody attached is
// when a recorded event matters MOST, since that is the gap a
// reconnecting client asks to be caught up across. Recording needs the
// encode — the slot holds the wire payload. Bounded: one payload per
// kind survives, and one buffer serves the whole drain.
var payload: std.ArrayList(u8) = .empty;
defer payload.deinit(self.alloc);
// At most one bell per DRAIN: N rings inside one 64 KiB chunk are one
// ring to a human, and `cat` on a binary produces ~256 of them per
// chunk. Per drain and not per session — a bell a second later is a
// separate ring. The pending slot is unaffected: it is one slot per
// kind at a `tracker.seq` that does not move inside this loop.
var rang = false;
for (s.eng.sideEvents()) |ev| {
if (ev.kind == .bell) {
if (rang) continue;
rang = true;
}
payload.clearRetainingCapacity();
switch (ev.kind) {
// A failed encode gives up this one event, silently: there is
// nowhere in this file to say it. `continue` and not `return`,
// because a 64 KiB clipboard can exhaust memory while the
// 1-byte bell behind it would have gone out fine.
.clipboard => proto.encodeClipboardEvent(
&payload,
self.alloc,
ev.target,
ev.payload,
) catch continue,
.bell => proto.encodeBellEvent(&payload, self.alloc) catch continue,
}
for (0..max_clients) |i| {
if (self.inSession(i, si)) _ = self.queueFrame(i, .term_event, payload.items);
}
// After the live clients and unconditionally: one client watching
// is no reason to leave a reconnecting one's gap empty. Stamped at
// the `tracker.seq` `sendUpdate` already advanced over this chunk.
s.recordPending(self.alloc, ev.kind, payload.items);
}
// Unconditional, and outside the loop: an encode that failed on OOM
// still consumed its event, and leaving it queued would replay it
// against the next chunk.
s.eng.clearSideEvents();
}
/// The engine's sampled terminal modes on the wire. One place, because
/// a mode the change-detecting sampler carries and the attach-time send
/// does not would be a mode that only exists until you reattach.
fn sampledModes(eng: *const Engine) proto.TermModes {
const m = eng.mouseModes();
return .{
.bracketed_paste = eng.bracketedPaste(),
.mouse_x10 = m.x10,
.mouse_normal = m.normal,
.mouse_button = m.button,
.mouse_any = m.any,
.mouse_utf8 = m.utf8,
.mouse_sgr = m.sgr,
.mouse_urxvt = m.urxvt,
.mouse_sgr_pixels = m.sgr_pixels,
.alt_screen = eng.onAltScreen(),
.cursor_keys = eng.cursorKeys(),
};
}
/// No history is kept: a reattaching client needs the current value, which
/// is why `sendResync` sends it unconditionally. The early return is the
/// point — modes change twice in a session's life, chunks arrive by the
/// thousand, and a frame per chunk is not proportional to what changed.
fn sampleTermModes(self: *Server, si: usize) void {
const s = self.ses(si);
const now = sampledModes(s.eng);
if (s.term_modes_sent) |sent| {
if (std.meta.eql(sent, now)) return;
}
s.term_modes_sent = now;
const payload = proto.encodeTermModes(now);
for (0..max_clients) |i| {
if (self.inSession(i, si)) _ = self.queueFrame(i, .term_modes, &payload);
}
}
/// An empty title is NOT sent: clearing would wipe whatever the user's own
/// terminal had in its title bar, and silence is not "set it to empty".
/// Accepted consequence — a session that genuinely clears its title leaves
/// the last one standing. `sendResync` must apply the same two rules.
fn sampleTermTitle(self: *Server, si: usize) void {
const s = self.ses(si);
const now = s.eng.title();
if (now.len == 0 or now.len > proto.term_title_max) return;
if (s.title_sent) |sent| {
if (std.mem.eql(u8, sent, now)) return;
}
// Duped before anything is sent, and the old one freed only once the
// new one exists: an OOM here leaves the session claiming to have
// sent what it did send, so the next sample retries rather than
// recording a title no client ever saw.
const owned = self.alloc.dupe(u8, now) catch return;
if (s.title_sent) |old| self.alloc.free(old);
s.title_sent = owned;
for (0..max_clients) |i| {
if (self.inSession(i, si)) _ = self.queueFrame(i, .term_title, owned);
}
}
/// The current command state as a wire struct. `mechanism` is the
/// caller's claim about how the verdict was reached: marks pushes say
/// .marks; await resolutions say what actually resolved them.
fn cmdState(self: *Server, si: usize, mechanism: proto.Mechanism) proto.CmdState {
const s = self.ses(si);
return .{
.phase = s.cmd.phase,
.mechanism = mechanism,
.exit_code = s.cmd.exit_code,
.start_row = s.cmd.start_row,
.end_row = s.cmd.end_row,
// The return watermark, read off its one owner. Absent means no
// command has returned this session, which is exactly what 0
// has always meant on the wire.
.seq = if (s.last_return) |lr| lr.seq else 0,
};
}
/// The seq override swaps cmdState's RETURN watermark for the tracker's
/// seq; `proto.CmdState.seq` documents both meanings.
fn fallbackState(
self: *Server,
si: usize,
mechanism: proto.Mechanism,
overrides: struct {
phase: ?proto.CmdPhase = null,
clear_exit_code: bool = false,
},
) proto.CmdState {
var st = self.cmdState(si, mechanism);
if (overrides.phase) |p| st.phase = p;
// Only marks can know a code (proto.Mechanism says so); a fallback
// that passed one through would be attributing the PREVIOUS
// command's verdict to this one.
if (overrides.clear_exit_code) st.exit_code = null;
st.seq = self.ses(si).tracker.seq;
return st;
}
fn buildStatusReply(self: *Server, si: usize) proto.StatusReply {
const s = self.ses(si);
const cur = s.eng.cursorPos();
return .{
.cols = self.colsNow(si),
.rows = self.rowsNow(si),
.cursor_x = cur.x,
.cursor_y = cur.y,
.history_rows = s.eng.historyRows(),
.alt_screen = s.eng.onAltScreen(),
// Live read, then the last one that worked, then a default. The
// middle term is the point: `mode_sent` is a real tcgetattr from
// last pump, so a failed read reports a remembered truth rather
// than a fabricated "canonical and echoing".
.mode = self.readPtyMode(si) orelse
(s.mode_sent orelse .{ .icanon = true, .echo = true }),
.cmd = self.cmdState(si, if (s.cmd.marks_seen) .marks else .pgid),
};
}
/// The snapshot payload for the grid as it stands: the fixed prefix, the
/// cursor, then every viewport row as cells. Caller owns the result.
/// Reads tracker.seq, so it must be called after the rebuild that stamps
/// it.
fn buildSnapshotPayload(self: *Server, si: usize) ![]u8 {
const s = self.ses(si);
return delta_mod.buildSnapshot(self.alloc, s.eng, .{
.seq = s.tracker.seq,
.history_rows = s.eng.historyRows(),
.cols = self.colsNow(si),
.rows = self.rowsNow(si),
.epoch = s.epoch,
});
}
/// One call because a rebuild is the ONLY thing that narrows the servable
/// span. The `defer` registers before the call because a FAILED rebuild
/// narrows it too — payloads permanently undeliverable and still resident.
fn rebuildTracker(self: *Server, si: usize) bool {
const s = self.ses(si);
defer s.dropUnservablePending(self.alloc);
s.tracker.rebuild(self.alloc, s.eng, self.rowsNow(si), self.colsNow(si)) catch return false;
return true;
}
/// Every client of THIS session gets it, and the rebuild happens with
/// nobody attached too, so the tracker stays usable. No term_modes,
/// unlike sendResync: every attached client was told the mode on attach
/// and at each change; a resync that skips it breaks that, silently.
pub fn resyncSnapshot(self: *Server, si: usize) void {
for (0..max_clients) |i| {
if (self.inSession(i, si)) self.clients[i].?.clearSelection();
}
if (!self.rebuildTracker(si)) return;
if (!self.hasClientsIn(si)) return;
const payload = self.buildSnapshotPayload(si) catch return;
defer self.alloc.free(payload);
self.countedSend(si, .snapshot, payload, null);
}
/// A join at the current size is a discontinuity for the joiner alone,
/// so repainting anyone else would be waste. The rebuild still bumps
/// seq for them all; a client never checks seq for contiguity.
fn snapshotTo(self: *Server, si: usize, i: usize) void {
if (!self.rebuildTracker(si)) return;
if (self.clients[i] == null) return;
const payload = self.buildSnapshotPayload(si) catch return;
defer self.alloc.free(payload);
self.countedSend(si, .snapshot, payload, i);
}
/// Attach/reattach for the client in slot `i`. Three cases: the size
/// changed, so latest wins and everyone gets a snapshot; the size held
/// and `have_seq` is serviceable, so this client alone gets a delta; or
/// it is unserviceable and this client alone gets a snapshot — the case a
/// FIRST attach takes, which is why it must not broadcast. Serviceable
/// means `have_epoch` names THIS daemon instance, or a client holding a
/// restarted daemon's seq would be told it is current.
fn sendResync(self: *Server, si: usize, i: usize, have_seq: u64, have_epoch: u64, size_changed: bool) void {
// After whichever content this call sends. Modes and the title are
// state, not history, so a snapshot needs them as much as a delta. One
// registration ahead of all three returns, so a fourth cannot skip it.
defer self.sendSampledStateTo(si, i);
if (size_changed) {
self.resyncSnapshot(si);
return;
}
const s = self.ses(si);
if (have_epoch == s.epoch and s.tracker.canServe(have_seq)) {
const payload = s.tracker.buildDeltaSince(self.alloc, s.eng, have_seq) catch {
self.snapshotTo(si, i);
return;
};
defer self.alloc.free(payload);
self.sendDeltaTo(si, i, payload);
self.replayPending(si, i, have_seq);
return;
}
self.snapshotTo(si, i);
}
/// The session's sampled state — modes and title — for one client. What
/// `sendResync` owes every attach whichever branch it takes; kept out of
/// line so that function is one defer plus a three-way branch.
fn sendSampledStateTo(self: *Server, si: usize, i: usize) void {
_ = self.queueFrame(i, .term_modes, &proto.encodeTermModes(
sampledModes(self.ses(si).eng),
));
// Not unconditional, unlike the modes: empty and over-cap are both
// refused here exactly as `sampleTermTitle` refuses them, and for its
// reasons. Borrowing the engine's buffer is safe because `queueFrame`
// copies into the client's pending bytes before returning.
const t = self.ses(si).eng.title();
if (t.len > 0 and t.len <= proto.term_title_max) {
_ = self.queueFrame(i, .term_title, t);
}
}
/// The delta branch of `sendResync` is the only caller: a client that
/// watched continuously is owed its gap, while one repainted from scratch
/// is a stranger whose clipboard a replay would hijack. Wire order is
/// delta → events, since a terminal ACTS on a bell. `> have_seq`, not
/// `>=`: an event at the client's own seq is one it saw.
fn replayPending(self: *Server, si: usize, i: usize, have_seq: u64) void {
const s = self.ses(si);
// Enum declaration order, so clipboard precedes bell. Fixed rather
// than meaningful — the two are independent — but an order that
// varies is one no test can pin, and this one cannot drift because
// nothing here restates it.
for (s.pendingSlots()) |slot| {
const p = slot.* orelse continue;
if (p.seq <= have_seq) continue;
_ = self.queueFrame(i, .term_event, p.payload);
}
}
pub fn rowsNow(self: *Server, si: usize) u16 {
return @intCast(self.ses(si).eng.term.rows);
}
pub fn colsNow(self: *Server, si: usize) u16 {
return @intCast(self.ses(si).eng.term.cols);
}
pub const writeManifestTo = upgrade_ops.writeManifestTo;
pub const validateUpgrade = upgrade_ops.validateUpgrade;
pub const clearCloexec = upgrade_ops.clearCloexec;
pub const setCloexec = upgrade_ops.setCloexec;
pub const sealAdoptedFds = upgrade_ops.sealAdoptedFds;
pub const execUpgrade = upgrade_ops.execUpgrade;
pub const stats_main_fmt =
"snapshots={d} snapshot_bytes={d} deltas={d} delta_bytes={d}" ++
" snapshot_equiv_bytes={d} clients={d} attaches={d} sessions={d}" ++
" agent_chans={d} agent_refused_no_offer={d} agent_refused_full={d}";
pub const stats_session_fmt = " session {s} clients={d} seq={d}";
/// Widest text `fmt` can print: every `{d}` a 20-digit u64, every `{s}` a
/// full-length name. Read off the format string, because a hand count does
/// not move when a counter is added and too small drops the client.
fn fmtWorstCase(comptime fmt: []const u8) usize {
comptime {
var n: usize = 0;
var i: usize = 0;
while (i < fmt.len) {
if (fmt[i] != '{') {
n += 1;
i += 1;
} else if (std.mem.startsWith(u8, fmt[i..], "{d}")) {
n += 20;
i += 3;
} else if (std.mem.startsWith(u8, fmt[i..], "{s}")) {
n += proto.session_name_max;
i += 3;
} else @compileError("stats format grew a specifier fmtWorstCase cannot size: " ++ fmt);
}
return n;
}
}
pub const stats_text_len =
fmtWorstCase(stats_main_fmt) + max_sessions * fmtWorstCase(stats_session_fmt);
/// Every live name plus one separator each — one more separator than a
/// join uses, which is the slack that lets sessionsText be infallible.
pub const sessions_text_len = max_sessions * (proto.session_name_max + 1);
/// What a `sessions_reply` is rendered into, wherever it is rendered. The
/// client and observer arms cannot share a function, but they must not each
/// spell the size: a caller that copied an array literal slips the assert.
pub const SessionsBuf = [sessions_text_len]u8;
comptime {
// The wall reads a reply into a buffer of its own — sized
// `proto.sessions_reply_max`, names plus the meta line — and a
// daemon that could say more than that would be truncated into a
// false list. Names alone must agree with the protocol's half.
std.debug.assert(@sizeOf(SessionsBuf) == proto.sessions_text_max);
}
/// A gauge: an unattached QUIC handshake holds a slot, unobservably.
fn liveClients(self: *const Server) usize {
return countLive(&self.clients);
}
/// Who is watching THIS shell, not how many sockets are open at all.
fn clientsInSession(self: *const Server, si: usize) usize {
return self.clientsIn(si, null);
}
/// Text, but machine-parsed: the bench harness and the e2e tests split on
/// these key=value pairs, so renaming or reordering fields breaks them.
/// Byte counters accrue when a frame is ACCEPTED INTO A CLIENT'S QUEUE,
/// which `pending_cap` bounds. Daemon-global, with per-session segments in
/// slot order — and still one line, so no caller trips on a newline.
pub fn statsText(self: *const Server, buf: []u8) ![]const u8 {
var w: std.Io.Writer = .fixed(buf);
try w.print(
stats_main_fmt,
.{
self.stats.snapshots, self.stats.snapshot_bytes,
self.stats.deltas, self.stats.delta_bytes,
self.stats.snapshot_equiv_bytes, self.liveClients(),
self.stats.attaches, self.sessions.live(),
// `agents.*`, not `stats.*`: the relay counts the refusals
// live and `stats` only carries the adoption-time pair.
self.agents.live(), self.agents.refused_no_offer,
self.agents.refused_full,
},
);
for (self.sessions.table, 0..) |slot, si| {
const s = slot orelse continue;
try w.print(stats_session_fmt, .{
s.name(), self.clientsInSession(si), s.tracker.seq,
});
}
return w.buffered();
}
};
// Forces semantic analysis of every pub decl under `zig build test`, so an
// unreferenced decl must at least compile (the silent-module-loss hazard,
// decisions.md). Pub decls only: std.meta.declarations sees nothing private.
test {
std.testing.refAllDeclsRecursive(@This());
_ = @import("cmd.zig");
_ = @import("shellint.zig");
_ = @import("upgrade.zig");
// Reaching a file is what registers its tests; build.zig gates the list.
// quic_server.zig chains here rather than above: its suite binds real UDP
// sockets, so a wedge in it costs the same silence a daemon test's does.
_ = @import("server_test_agent.zig");
_ = @import("server_test_attach.zig");
_ = @import("server_test_await.zig");
_ = @import("server_test_clipboard.zig");
_ = @import("server_test_deliver.zig");
_ = @import("server_test_harness.zig");
_ = @import("server_test_modes.zig");
_ = @import("server_test_quic.zig");
_ = @import("server_test_session.zig");
_ = @import("server_test_upgrade.zig");
_ = @import("server_forward.zig");
_ = @import("quic_server.zig");
}