src/client/client.zig
Ref: Size: 156.6 KiB History
//! The LINK, and nothing above it: how a client reaches a daemon and what it
//! says on arrival. The `Target` grammar, the dial (unix socket, `--via` pipe,
//! direct QUIC, the ssh→QUIC handoff), the words a failed dial exits with, the
//! attach frame, the reconnect backoff, and the pure answers a chord needs.
//!
//! Not a SESSION. There is ONE interaction loop and it is wallview.zig's tile
//! pump. What crosses the seam is `Transport` — opened by whoever holds the
//! tty, `adopt`ed by the thread that will own it — and nothing else.
const std = @import("std");
const proto = @import("term").protocol;
const TmpDir = @import("testtmp").TmpDir;
const quic = @import("quic");
const xdg = @import("xdg");
// The daemon-socket dial, under a name that is not `dial`: `Transport.open`
// already takes a `dial: ?*handoff.Dial` parameter, and a parameter that
// shadows a file-scope declaration does not compile.
const dialer = @import("dial");
// The fd|pipe|quic union and its await loop, under a name that is not
// `link`: `Transport.link` is the field this module reads everywhere, and
// several local variables are called `link` too.
const link_mod = @import("link");
// The client link's public seams: the wall, the hub and the mains reach
// these as client.X — the table stays one row, the files stay children.
pub const hosts = @import("hosts.zig");
pub const handoff = @import("handoff.zig");
pub const layout = @import("layout.zig");
pub const layoutfile = @import("layoutfile.zig");
// The dial and the prefix filter both watch for this local abort byte. A dial
// has no session to command and no surface input to encode.
pub const interrupt = @import("interrupt.zig");
pub const askpass = @import("askpass.zig");
pub const core = @import("client_core.zig");
pub const selection = @import("selection.zig");
pub const discovery = @import("discovery.zig");
pub const resolver = @import("resolver.zig");
const open_wait = @import("open_wait.zig");
pub const session_pump = @import("session_pump.zig");
pub const forward = @import("forward.zig");
/// A session name held by value. The names a switch travels on are decoded
/// out of a frame payload that is freed before the re-dial, so they cannot
/// be carried as slices.
pub const SessionName = struct {
buf: [proto.session_name_max]u8 = undefined,
len: u8 = 0,
pub fn of(name: []const u8) SessionName {
var n: SessionName = .{ .len = @intCast(name.len) };
@memcpy(n.buf[0..name.len], name);
return n;
}
pub fn slice(self: *const SessionName) []const u8 {
return self.buf[0..self.len];
}
};
/// `SessionName.of` memcpys with no bound of its own, and ring names arrive
/// in a peer's `sessions_reply` — so the guard sits where `of` is reached.
pub fn validPick(pick: ?[]const u8) ?SessionName {
const name = pick orelse return null;
if (!proto.validSessionName(name)) return null;
return .of(name);
}
/// The name `Ctrl-\ c` creates: the lowest non-negative integer not already a
/// session name, out of a `sessions_reply` payload. Numbering rather than
/// `new-1` because the default session is already called "0".
///
/// Both walks go through `proto.sessionsIter`, the one trust policy for that
/// payload. The answer it gives is unchanged by the filter — a candidate is
/// always a decimal string, which is a name `validSessionName` accepts, so no
/// line the filter drops could ever have matched one — but agreeing with the
/// wall's diff and the picker's count is what makes "the name the `c` chord
/// would have landed on" one sentence rather than three implementations.
pub fn nextFreeName(out: *[proto.session_name_max]u8, list: []const u8) []const u8 {
// Each existing name can rule out at most one candidate, so the first
// free number is somewhere in 0..count — no cap constant needed, and in
// particular no duplicate of the daemon's max_sessions.
var count: usize = 0;
var counter = proto.sessionsIter(list);
while (counter.next()) |_| count += 1;
var n: usize = 0;
while (n <= count) : (n += 1) {
// A usize's widest decimal form is 20 bytes, session_name_max is
// 32: this print cannot overflow `out` for any n reachable here.
const cand = std.fmt.bufPrint(out, "{d}", .{n}) catch unreachable;
var names = proto.sessionsIter(list);
const taken = while (names.next()) |name| {
if (std.mem.eql(u8, name, cand)) break true;
} else false;
if (!taken) return cand;
}
unreachable; // count+1 candidates, count names: one must be free.
}
/// Which chord is waiting for the daemon to answer. One field rather than a
/// flag per chord, so "one question outstanding" is a fact and not a rule.
/// They share the wait because they share the deadline: each is a verb an
/// older daemon simply does not hear.
pub const SwitchIntent = enum { none, new, end, end_force };
/// A chord that has asked the daemon for its session list, and when it stops
/// waiting. The DEADLINE is why this is a struct: a daemon older than
/// `sessions_req` drops frames it does not recognise, so the question is never
/// heard — and an intent with no expiry stays armed, making every later chord
/// a silent no-op the user blames on their keyboard.
pub const PendingSwitch = struct {
/// How long a chord waits. Far longer than any round trip a switch is
/// usable over, and short enough that the keystroke is still in the
/// user's head when the marker tells them it went nowhere.
const wait_ms: i64 = 2000;
intent: SwitchIntent = .none,
/// Meaningless while `intent` is `.none`; set by every `arm`.
until: i64 = 0,
pub fn arm(self: *PendingSwitch, intent: SwitchIntent, now: i64) void {
self.intent = intent;
self.until = now + wait_ms;
}
pub fn clear(self: *PendingSwitch) void {
self.intent = .none;
}
/// The intent, spent: one question is answered once, whatever the
/// answer turns out to mean.
pub fn take(self: *PendingSwitch) SwitchIntent {
defer self.intent = .none;
return self.intent;
}
/// True exactly once per armed intent — self-clearing, so a poll loop
/// cannot report the same silence twice.
pub fn expired(self: *PendingSwitch, now: i64) bool {
if (self.intent == .none or now < self.until) return false;
self.intent = .none;
return true;
}
};
/// Callers spell this `client.Incoming`; the union itself is the link row's.
pub const Incoming = link_mod.Incoming;
/// A `--via` that died before the first frame carried no connection, and the
/// cause is already on that command's stderr. `session_epoch` tells the two
/// apart: set from the first snapshot, never reset.
pub fn lostMsg(target: Target, session_epoch: u64) []const u8 {
if (target == .via and session_epoch == 0)
return "mux: transport command failed before a session started";
return "mux: connection to the daemon lost";
}
/// The client's name for the shared default; see `quic.default_idle_ms`
/// for what the number means and why it lives there.
pub const quic_idle_ms_default: u32 = quic.default_idle_ms;
pub const IdleMs = quic.IdleMs;
/// What a QUIC attach was asked for: where to dial and what key to prove
/// ourselves with.
pub const QuicTarget = struct {
host_port: []const u8,
key_path: []const u8,
idle_ms: u32 = quic_idle_ms_default,
/// The attach budget, split off idle_ms the way HandoffTarget
/// already does: the time we give a handshake is not the time we
/// give a quiet session. See handoff.deadline_ms for the number's
/// derivation.
deadline_ms: u32 = handoff.deadline_ms,
};
/// The bare-HOST recipe: everything a (re)connect needs to run the ssh→QUIC
/// handoff again. At the Transport layer so the reconnect loop re-runs the
/// WHOLE flow — a daemon restart invalidates the cached port.
pub const AskPass = struct { sock: []const u8, exe: []const u8 };
pub const HandoffTarget = struct {
/// The word the user typed. ssh's business entirely (aliases, `user@`,
/// ProxyJump); the QUIC dial uses `handoff.dialHost(host)`.
host: []const u8,
/// `ssh <host> mux d endpoint` as argv, prebuilt by mux_main — it has
/// the allocator, and it builds this once for the whole session.
ssh_argv: []const []const u8,
/// `ssh <host> mux d endpoint --start`, from the same
/// `handoff.recipeFor` call. Empty is "nothing to start": the dial
/// either finds a daemon or does not, and `openHandoff` reads the
/// bare word instead rather than exec'ing an argv with no argv[0].
asked_argv: []const []const u8 = &.{},
/// Where the last announce is remembered. Null means never cache (an
/// uncacheable host, or no resolvable cache directory): every attach is
/// then cold, which costs time and stays correct.
cache_path: ?[]const u8,
/// The per-attempt QUIC budget. A field rather than the constant so
/// tests can shrink it; production passes `handoff.deadline_ms`.
deadline_ms: u32 = handoff.deadline_ms,
idle_ms: u32 = quic_idle_ms_default,
/// Whether a USER asked for this dial: the entry attach and a picker
/// birth, never a poll, a reconnect or a restored tile. It decides two
/// things.
///
/// SAY SO when the session falls back to ssh — but only once, because a
/// reconnect re-runs this recipe forever and a line per retry would scroll
/// a live session's stderr onto the alternate screen.
///
/// PICK THE ARGV: `asked_argv` rather than `ssh_argv`, unless there is
/// none. The two are one ssh line a flag apart, so the far end decides
/// whether to start — and a poll's bare word cannot undo a `mux d stop`.
///
/// Defaults to the harmless half: a missing start is a `mux HOST` that
/// says so, a spurious one is a daemon on someone else's box.
asked: bool = false,
/// Whether the caller is a human at a bare prompt owed ssh's narration as
/// it happens: relay every stderr byte to mux's own fd 2. The remote's
/// progress rides that stderr, and the entry dial's user is sitting
/// through the wait it describes.
///
/// Nobody else relays. The bytes are READ either way — that is `Reason` —
/// but under a wall's alternate screen they would sit over tiles and
/// rails. Defaults to the harmless half, as `asked` does: a spurious line
/// corrupts a paint nobody can repair from.
narrate: bool = false,
/// Where this dial's ssh sends its prompts, and what carries them. Null is
/// "ssh keeps its own": the entry dial's user has a /dev/tty right there,
/// and a poll spells `BatchMode` and asks nothing. A wall dial that forgets
/// these reads /dev/tty under the alternate screen — a wait nobody can see.
/// Two fields because `spawn.selfExe` is under `src/cli/`, which a client
/// module may not name; the wall is tui and fills both in one place.
ask_sock: ?[]const u8 = null,
ask_exe: []const u8 = "",
/// Both or neither. `SSH_ASKPASS` pointing at nothing makes ssh FAIL
/// every prompt rather than ask one, which is worse than the tty read
/// this replaces.
pub fn askpassFor(self: HandoffTarget) ?AskPass {
const sock = self.ask_sock orelse return null;
if (self.ask_exe.len == 0) return null;
return .{ .sock = sock, .exe = self.ask_exe };
}
/// The recipe→target literal: a field added above is added here, not
/// at every dial. `asked` is a parameter with NO default though
/// the field has one — Zig cannot omit it, and that compile error is
/// the pin the field's default is not.
pub fn fromRecipe(host: []const u8, r: handoff.Recipe, idle_ms: u32, asked: bool) HandoffTarget {
return .{
.host = host,
.ssh_argv = r.ssh_argv,
.asked_argv = r.asked_argv,
.cache_path = r.cache_path,
.idle_ms = idle_ms,
.asked = asked,
};
}
};
/// A union, not four nullable fields — "exactly one is set" stops being a
/// rule a caller must remember.
pub const Target = union(enum) {
sock: []const u8,
via: []const u8,
quic: QuicTarget,
hand: HandoffTarget,
/// The one road from a spelling's `hosts.Spec` to the dial it names. Every
/// slice is OWNED by `alloc`: the spelling a caller parsed may be a scratch
/// buffer, and a target outlives the read that made it. `asked` is
/// required, never defaulted — see `HandoffTarget.fromRecipe`.
pub fn fromSpec(alloc: std.mem.Allocator, spec: hosts.Spec, key: ?[]const u8, idle_ms: u32, asked: bool) SpecError!Target {
return switch (spec) {
// No length guard: `Address.initUnix` answers `NameTooLong` and
// truncates nothing, so a doomed path fails as a dial like any
// other. The one BINDER is where a path is refused by name.
.sock => |path| .{ .sock = try alloc.dupe(u8, path) },
.host => |h| blk: {
const hd = try alloc.dupe(u8, h);
const r = try handoff.recipeFor(alloc, hd, false);
break :blk .{ .hand = HandoffTarget.fromRecipe(hd, r, idle_ms, asked) };
},
.quic => |hp| blk: {
const key_path = switch (xdg.resolveKeyPath(alloc, key) catch |err| switch (err) {
// No HOME is no default key path, which is the same
// outcome for this caller as a default that isn't
// there: nothing to authenticate the dial with.
error.NoHome => return error.MissingKey,
else => |e| return e,
}) {
// `.given` borrows from argv/env, which outlives
// nothing in particular from the target's point of view.
.given => |kp| try alloc.dupe(u8, kp),
.default => |kp| kp,
// `.missing` is an ALLOCATED path too — the refusal is
// the one arm that does not keep it, so it is the one
// arm that has to free it.
.missing => |kp| {
alloc.free(kp);
return error.MissingKey;
},
};
break :blk .{ .quic = .{
.host_port = try alloc.dupe(u8, hp),
.key_path = key_path,
.idle_ms = idle_ms,
} };
},
};
}
};
/// What resolving a spelling can fail at: a key that is not there to prove
/// the dial with, and a socket path the kernel cannot hold.
pub const SpecError = error{ MissingKey, OutOfMemory };
/// What the open produced — the live wire. Distinct from `Target` because a
/// `hand` recipe yields either a quic or a pipe link, and which one is
/// decided inside `openHandoff` at runtime. The union is the link row's;
/// this name is what the rest of the client spells, `std.meta.Tag(Link)`
/// included.
pub const Link = link_mod.Link;
/// One live connection to a daemon, however it was reached. The point of the
/// struct is that it can be closed and opened again from the same `Target`,
/// which is what lets a session outlive its transport instead of exiting
/// with it.
pub const Transport = struct {
/// The wire itself. Everything below the frame — send, read, wait,
/// close — is the link's. What stays on this struct is what the link has
/// no business knowing: the second fd a handoff's ssh talks on, the last
/// line it said, and whether that line is relayed onward.
link: Link,
/// The handoff ssh's stderr for a `.pipe` link born of a handoff; -1
/// for every other link and for `--via`, whose stderr is the user's.
err_fd: std.posix.fd_t = -1,
/// Where `drainErr` puts what it read, so a link that outlives its
/// dial still has somewhere to keep a line. Nothing reads it today —
/// a tile shows `connecting`, and the row quotes the POLL's copy — so
/// a reader that appears is free to define what it means.
reason: handoff.Reason = .{},
/// `HandoffTarget.narrate`, carried past the dial so a link that came up on
/// the pipe goes on relaying. The entry dial CLEARS it when the wall takes
/// the screen: past that, fd 2 is the alternate screen.
narrate: bool = false,
/// The handoff's coordination ssh, stderr included. `ask` non-null is
/// what turns its prompts into frames on a socket.
fn spawnPipe(alloc: std.mem.Allocator, argv: []const []const u8, ask: ?AskPass) !std.process.Child {
return spawnWithStderr(alloc, argv, .Pipe, ask);
}
/// `--via CMD` is the user's OWN program in the user's own terminal.
fn spawnVia(alloc: std.mem.Allocator, argv: []const []const u8) !std.process.Child {
return spawnWithStderr(alloc, argv, .Inherit, null);
}
/// One exec'd child, argv and never a shell line.
fn spawnWithStderr(
alloc: std.mem.Allocator,
argv: []const []const u8,
stderr_behavior: std.process.Child.StdIo,
ask: ?AskPass,
) !std.process.Child {
// The product runs `ssh` and the user's own `--via` program, and
// neither is worth a shell's expansions between us and it. `argv` need
// not outlive the call: `std.process.Child` copies it before the fork.
var child = std.process.Child.init(argv, alloc);
child.stdin_behavior = .Pipe;
child.stdout_behavior = .Pipe;
// The handoff's ssh is PIPED: its diagnostics are mux's to keep —
// the last line becomes the dial's `handoff.Reason`, which a picker
// row quotes — and a wall's alternate screen admits no foreign
// writer. `--via`'s stderr stays the user's, inherited.
child.stderr_behavior = stderr_behavior;
// `REQUIRE=force` is the whole trick: without it ssh uses its helper
// only when there is no tty, and a wall has one. The map lives to
// the `spawn` below and no longer — the child's envp is built there.
var env: ?std.process.EnvMap = if (ask == null) null else try std.process.getEnvMap(alloc);
defer if (env) |*e| e.deinit();
if (ask) |a| {
try env.?.put("SSH_ASKPASS", a.exe);
try env.?.put("SSH_ASKPASS_REQUIRE", "force");
try env.?.put(askpass.sock_env, a.sock);
child.env_map = &env.?;
}
try child.spawn();
return child;
}
/// `--via CMD` is argv WORDS: mux execs the command, so a quote, a
/// variable or a pipeline in CMD is bytes rather than syntax. The words
/// borrow from `cmd`; only the pointer array is allocated.
fn viaArgv(alloc: std.mem.Allocator, cmd: []const u8) ![]const []const u8 {
var words: std.ArrayList([]const u8) = .empty;
errdefer words.deinit(alloc);
var it = std.mem.tokenizeAny(u8, cmd, " \t\r\n");
while (it.next()) |w| try words.append(alloc, w);
// `std.process.Child` would spawn argv[0] out of an empty slice.
if (words.items.len == 0) return error.EmptyViaCommand;
return words.toOwnedSlice(alloc);
}
/// The transport a piped child IS: its stdio is the wire, and the child
/// itself is what has to be owned and reaped.
fn pipeTransport(child: std.process.Child) Transport {
return .{ .link = .{ .pipe = .{
.child = child,
.r = child.stdout.?.handle,
.w = child.stdin.?.handle,
} } };
}
/// `connect` only creates state, so the handshake wait belongs here:
/// "open succeeded" means one thing everywhere. `budget_ms` is the
/// attach budget, not `idle_ms`.
fn quicTransport(
alloc: std.mem.Allocator,
addr: std.net.Address,
key: quic.Key,
idle_ms: u32,
budget_ms: u32,
carry: ?*std.ArrayList(u8),
abort_fd: std.posix.fd_t,
inbound_cap: ?usize,
) !Transport {
const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
errdefer cl.deinit();
// The first handshake pump can already deliver stream data. Set a
// role's opaque bound before waitReady, not after its first frame.
cl.inbound_cap = inbound_cap;
try waitReady(cl, budget_ms, alloc, carry, abort_fd);
return .{ .link = .{ .quic = .{ .cl = cl, .alloc = alloc } } };
}
/// Errors are returned rather than reported here — the caller knows
/// which recipe it handed us and so which message fits.
/// `carry` collects any non-abort bytes typed while a QUIC handshake is
/// in flight; null means drop them. See waitReady.
pub fn open(
alloc: std.mem.Allocator,
target: Target,
carry: ?*std.ArrayList(u8),
/// Where the abort key (Ctrl-\) is watched for during the opening
/// waits: stdin in the CLI, -1 (no abort channel) in a hub that
/// has no terminal — its stray fd 0 must never be read.
abort_fd: std.posix.fd_t,
/// What the handoff leaves behind: ssh's last line, so a failed dial is
/// reported in ssh's own words, and the pid that said it, so a prompt
/// can be answered against the one dial it belongs to.
dial: ?*handoff.Dial,
) !Transport {
return openUntilBounded(alloc, target, carry, abort_fd, dial, null, null);
}
/// An absolute budget shared by DNS, connect, handshake, SSH fallback and
/// the caller's later request. Null preserves the legacy opening policy.
pub fn openUntil(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, deadline: ?i64) !Transport {
return openUntilBounded(alloc, target, carry, abort_fd, dial, deadline, null);
}
/// `inbound_cap` is an opaque transport receive budget for roles that
/// cannot retain terminal-sized traffic. Null preserves terminal policy.
pub fn openBounded(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, inbound_cap: ?usize) !Transport {
return openUntilBounded(alloc, target, carry, abort_fd, dial, null, inbound_cap);
}
fn openUntilBounded(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, deadline: ?i64, inbound_cap: ?usize) !Transport {
var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry, .deadline = deadline };
// Local first attach has always left stdin to the established client:
// even an already queued detach must attach before handling its chord.
// GUI callers have no carry and retain their dedicated cancel channel.
if (carry != null and (target == .sock or target == .via)) wait.abort_fd = -1;
if (target == .hand) try checkHandoffWait(&wait) else try wait.check();
switch (target) {
// Delegated whole, because the handoff can end up producing
// either of the two links below and owns the choice itself.
.hand => |h| return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait, inbound_cap),
.quic => |q| {
const key = try quic.Key.load(q.key_path);
const hp = try quic.splitHostPort(q.host_port);
const addr = try resolveOpening(alloc, hp.host, hp.port, q.deadline_ms, &wait);
return quicTransport(alloc, addr, key, q.idle_ms, try wait.remaining(q.deadline_ms), carry, abort_fd, inbound_cap);
},
.via => |cmd| {
const argv = try viaArgv(alloc, cmd);
defer alloc.free(argv);
return pipeTransport(try spawnVia(alloc, argv));
},
.sock => |path| {
const stream = try open_wait.connectUnix(path, &wait);
return .{ .link = .{ .fd = stream.handle } };
},
}
}
/// Process-wide, guarding the one handoff-cache write below. See there
/// for why one file can now have two writers.
var cache_write_mu: std.Thread.Mutex = .{};
/// The pipe that carried the announce IS the QUIC fallback, so a
/// UDP-blocked network costs one deadline, not two.
pub fn openHandoff(
alloc: std.mem.Allocator,
h: HandoffTarget,
carry: ?*std.ArrayList(u8),
abort_fd: std.posix.fd_t,
dial: ?*handoff.Dial,
) !Transport {
var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry };
return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait, null);
}
fn checkHandoffWait(wait: *open_wait.Wait) !void {
// First-attach SSH owns cooked stdin for passwords. Only an actual
// QUIC wait may collect carry bytes; phase/deadline checks must not.
if (wait.carry != null) {
var deadline_only = wait.*;
deadline_only.abort_fd = -1;
try deadline_only.check();
} else try wait.check();
}
fn openHandoffWait(alloc: std.mem.Allocator, h: HandoffTarget, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, wait: *open_wait.Wait, inbound_cap: ?usize) !Transport {
// The ORDER is `handoff.next`'s; this loop performs the step it is
// handed and reports what came of it.
var st: handoff.State = .{
.cached = if (h.cache_path) |cp| (handoff.readCache(cp) catch null) else null,
.asked = h.asked,
.has_cache = h.cache_path != null,
};
// Optional because the warm path can finish without ever spawning
// one. The errdefer owns it on EVERY error return, `fail` included,
// so no step spells a kill of its own.
var child: ?std.process.Child = null;
errdefer if (child) |*c| {
link_mod.terminateChild(c);
};
// A caller with nowhere to show a reason still needs one kept: the
// Transport carries it, and a failed dial's last line would
// otherwise have to be read twice.
var local_dial: handoff.Dial = .{};
const out = dial orelse &local_dial;
var errp: ErrPipe = .{
.fd = -1,
.reason = &out.reason,
.narrate = h.narrate,
};
errdefer errp.close();
// What the terminal steps need from the steps before them; `dialed`
// is the port the fallback line names.
var dialed: ?handoff.Endpoint = null;
var quic_t: ?Transport = null;
errdefer if (quic_t) |*transport| transport.close();
// `fail` returns the error the failing step actually got, so the
// caller's message names the real cause. Optional rather than a
// placeholder value: a `fail` that nothing recorded is a hole in the
// table, and a panic says so where an invented error would not.
var last_err: ?anyerror = null;
var step = handoff.next(&st, null);
while (true) {
try checkHandoffWait(wait);
const outcome: handoff.Outcome = switch (step) {
.dial_quic => |ep| blk: {
dialed = ep;
if (openQuicEndpointWait(alloc, h, ep, carry, abort_fd, wait, inbound_cap)) |t| {
quic_t = t;
break :blk .ok;
} else |err| {
last_err = err;
break :blk if (err == error.UserAbort) .user_abort else .failed;
}
},
.spawn_ssh => blk: {
// ONE run, picked by `asked` alone: the asking word ensures
// a daemon and announces on the same stdout, so there is no
// refusal to read and no second run. `and len > 0` keeps
// `asked_argv`'s doc true — an empty argv is not a no-op at
// the exec, the child null-unwraps `argv[0]` and dies.
const argv = if (h.asked and h.asked_argv.len > 0) h.asked_argv else h.ssh_argv;
child = spawnPipe(alloc, argv, h.askpassFor()) catch |err| {
last_err = err;
break :blk .failed;
};
// Taken OFF the child: `Child.kill` closes `stderr` with the
// other pipes, and this fd outlives the kill on `use_pipe`.
// `if` rather than an unwrap, so a spawn that stopped piping
// stderr FAILS the test named for it instead of panicking.
if (child.?.stderr) |f| {
errp.fd = f.handle;
child.?.stderr = null;
}
// Recorded HERE, not on the way out: the caller that needs it
// is the one this call is about to fail, and a refused
// password leaves no transport to read a pid off.
out.ssh_pid = child.?.id;
break :blk .ok;
},
.read_announce => blk: {
// ssh owns the terminal while it runs: its prompts read
// /dev/tty, and a competing stdin reader here steals whole
// cooked lines — auth fails on an empty password and the
// stolen line replays INTO the session via carry. On a first
// attach the tty is still cooked, so Ctrl-C aborts the group;
// the byte-read abort is only real on reconnect.
const announce_abort_fd: std.posix.fd_t = if (carry != null) -1 else abort_fd;
const got = readAnnounceUntil(child.?.stdout.?.handle, alloc, null, announce_abort_fd, &errp, wait.deadline) catch |err| {
// ssh says why on its way out, and its stdout's EOF can
// be the same poll pass as the last of it: read what is
// left BEFORE this error becomes the caller's answer.
errp.drainReady();
last_err = err;
break :blk .announce_failed;
};
break :blk if (got) |ep| .{ .announced = ep } else .none;
},
.write_cache => |ep| blk: {
// A failed cache write costs one cold attach and nothing
// else. Serialized because two tiles naming one host share a
// cache path and `writeCache` truncates in place: unserialized
// writers tear the line, and the likelier miss is the loser
// writing the STALER endpoint.
cache_write_mu.lock();
defer cache_write_mu.unlock();
handoff.writeCache(h.cache_path.?, ep) catch {};
break :blk .done;
},
.use_quic => {
// QUIC carries the session now, so the coordination ssh —
// if this handoff ran one at all — is done. kill()
// waits and closes the pipes with it;
// stderr is no longer among them, so it is closed here.
errp.close();
// ...and says nothing. Whatever the coordination ssh
// narrated belongs to a handoff that SUCCEEDED; left
// standing it would be quoted by the next thing that
// fails on this target, blaming a box that is up.
errp.reason.clear();
if (child) |*c| {
link_mod.terminateChild(c);
}
return quic_t.?;
},
.use_pipe => |say| {
// The line names what was tried and what is happening
// instead, and no cause: the client cannot tell a blocked
// port from a wrong key (both are silence), and guessing
// would be worse than the deadline it just spent.
if (say) std.debug.print(
"mux: quic://{s}:{d} unreachable, attaching over ssh\n",
.{ handoff.dialHost(h.host), dialed.?.port },
);
var t = pipeTransport(child.?);
// The pipe IS the session now, and ssh goes on talking
// for as long as it lives: the fd, the policy and the
// line so far all pass to whoever polls this transport.
t.err_fd = errp.fd;
t.narrate = errp.narrate;
t.reason = errp.reason.*;
return t;
},
.fail => return last_err.?,
};
step = handoff.next(&st, outcome);
}
}
/// The dial's budget is the handoff deadline while the CONNECTION keeps
/// the ordinary `idle_ms`; `quicTransport` spells that split.
pub fn openQuicEndpoint(
alloc: std.mem.Allocator,
h: HandoffTarget,
ep: handoff.Endpoint,
carry: ?*std.ArrayList(u8),
abort_fd: std.posix.fd_t,
) !Transport {
var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry };
return openQuicEndpointWait(alloc, h, ep, carry, abort_fd, &wait, null);
}
fn openQuicEndpointWait(alloc: std.mem.Allocator, h: HandoffTarget, ep: handoff.Endpoint, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, wait: *open_wait.Wait, inbound_cap: ?usize) !Transport {
const addr = try resolveOpening(alloc, handoff.dialHost(h.host), ep.port, h.deadline_ms, wait);
const key = quic.Key{ .bytes = ep.key };
return quicTransport(alloc, addr, key, h.idle_ms, try wait.remaining(h.deadline_ms), carry, abort_fd, inbound_cap);
}
fn resolveOpening(alloc: std.mem.Allocator, host: []const u8, port: u16, budget_ms: u32, wait: *open_wait.Wait) !std.net.Address {
if (wait.abort_fd < 0 and wait.deadline == null) return quic.resolveHost(alloc, host, port);
const previous = wait.deadline;
defer wait.deadline = previous;
wait.deadline = @min(previous orelse std.math.maxInt(i64), std.time.milliTimestamp() + budget_ms);
return resolver.resolve(alloc, host, port, wait);
}
/// Not where frames come from: QUIC polls UDP, reads frames above it.
pub fn pollFd(self: *const Transport) std.posix.fd_t {
return self.link.pollFd();
}
/// The second fd a handoff's owner polls, or null when there is none.
pub fn errFd(self: *const Transport) ?std.posix.fd_t {
// Skipped, ssh fills a 64k pipe and stops talking to the far end
// at all — a tile going silent for a reason no frame can explain.
return if (self.err_fd < 0) null else self.err_fd;
}
/// One read of that fd, kept as the reason and relayed if `narrate`.
/// Called when the poll says readable; EOF closes the fd for good.
pub fn drainErr(self: *Transport) void {
if (self.err_fd < 0) return;
var e: ErrPipe = .{ .fd = self.err_fd, .reason = &self.reason, .narrate = self.narrate };
_ = e.drain();
self.err_fd = e.fd;
}
/// One owning thread per `Transport`, but the ENTRY dial runs on main —
/// ssh may need the tty for a password. `qout` is what would otherwise
/// cross: two threads on one non-thread-safe arena.
pub fn adopt(self: *Transport, alloc: std.mem.Allocator) void {
const q = switch (self.link) {
.quic => |*q| q,
.fd, .pipe => return,
};
std.debug.assert(q.qout.items.len == 0);
q.qout.deinit(q.alloc);
q.qout = .empty;
q.alloc = alloc;
}
pub fn writeFrame(self: *Transport, t: proto.MsgType, payload: []const u8) !void {
return self.link.sendFrame(t, payload);
}
/// Unconditional: a QUIC connection's timers are the only thing that
/// notices a peer which stopped answering.
pub fn service(self: *Transport) void {
self.link.service();
}
/// Folds ngtcp2's next deadline in, so retransmits and idle timeouts
/// happen on time without a second timer.
pub fn timeoutMs(self: *Transport, cap_ms: i32) i32 {
return self.link.timeoutMs(cap_ms);
}
/// Offer the outbound queue to the ring again. Called after every write
/// and on every service pass, because the room to accept comes from
/// acknowledgements, which arrive on their own schedule.
pub fn flushQuic(self: *Transport) void {
self.link.flushQuic();
}
/// The next whole frame, if there is one. See `Incoming` for why a
/// missing frame is not automatically a dead transport, and
/// `link.Link.readFrame` for the two faults that stay errors.
pub fn readFrame(self: *Transport, alloc: std.mem.Allocator) !Incoming {
return self.link.readFrame(alloc);
}
/// Idempotent, and it has to be: a re-dial releases the dead transport on
/// entry, and an abort then closes the same value again through the pump's
/// `defer`. A second `close(2)` on a stale fd is EBADF, which `std.posix`
/// maps to `unreachable` — a panic that `--sock` does not hide.
///
/// Only the stderr fd is closed here: it is the one thing the link never
/// knew about, and `openHandoff` took it off the Child precisely so the
/// kill inside `Link.close` would leave it alone.
pub fn close(self: *Transport) void {
if (self.link == .fd and self.link.fd == -1) return; // already released
if (self.err_fd >= 0) {
std.posix.close(self.err_fd);
self.err_fd = -1;
}
self.link.close();
}
};
/// The path is a parameter, never the environment, so tests stay
/// environment-free. Null, never an error: "no agent" must be an answer the
/// pump can turn into an `agent_close`.
pub fn connectAgent(path: []const u8) ?std.posix.fd_t {
if (path.len == 0) return null;
const stream = dialer.dial(path) catch return null;
return stream.handle;
}
/// Bounded by the attach budget, not the idle timeout: the time given a
/// handshake is not the time given a quiet session.
fn waitReady(
cl: *quic.Client,
budget_ms: u32,
alloc: std.mem.Allocator,
carry: ?*std.ArrayList(u8),
abort_fd: std.posix.fd_t,
) !void {
const deadline = std.time.milliTimestamp() + budget_ms;
// A closed abort fd stays readable forever, so once it reports EOF it
// has to stop being polled or this loop spins hot for the rest of the
// bound instead of waiting on the socket. An abort_fd of -1 (a hub
// with no terminal) starts unwatched and stays that way.
var watch_stdin = abort_fd >= 0;
while (std.time.milliTimestamp() < deadline) {
cl.pump();
if (cl.isReady()) return;
if (cl.dead) return error.QuicHandshakeFailed;
// The abort fd is watched alongside the socket, and it has to be: this
// wait runs INSIDE `Transport.open`, where nothing else is looking for
// the abort key, and a re-dial's terminal is raw so `Ctrl-\` is the
// only way out. The uncapped retry loop is justified by that key.
var fds = abortPoll(cl.pollFd(), abort_fd, watch_stdin);
_ = std.posix.poll(&fds, cl.timeoutMs(50)) catch break;
// On a first attach the bytes drained here are the user's first
// keystrokes and are owed to the shell, so `carry` keeps them.
if (fds[1].revents != 0) watch_stdin = try drainAbortFd(abort_fd, alloc, carry);
}
cl.pump();
if (cl.isReady()) return;
return error.QuicHandshakeFailed;
}
/// The two-descriptor wait both handshake paths make: the thing being
/// waited on, and the abort key beside it. An unwatched abort fd polls -1,
/// which poll(2) ignores.
fn abortPoll(main_fd: std.posix.fd_t, abort_fd: std.posix.fd_t, watching: bool) [2]std.posix.pollfd {
return .{
.{ .fd = main_fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = if (watching) abort_fd else -1, .events = std.posix.POLL.IN, .revents = 0 },
};
}
/// Drain the abort fd; false means stop watching it, since a closed one stays
/// readable forever and the wait would spin hot. Non-abort bytes are the
/// caller's policy: `carry` keeps them, null drops them, and a reconnect
/// drops — replaying stale keystrokes on resume is worse than losing them.
fn drainAbortFd(abort_fd: std.posix.fd_t, alloc: std.mem.Allocator, carry: ?*std.ArrayList(u8)) error{UserAbort}!bool {
var buf: [1024]u8 = undefined;
const n = std.posix.read(abort_fd, &buf) catch 0;
if (n == 0) return false;
if (std.mem.indexOfScalar(u8, buf[0..n], interrupt.detach_key) != null) return error.UserAbort;
if (carry) |q| q.appendSlice(alloc, buf[0..n]) catch {};
return true;
}
/// The handoff ssh's stderr, while somebody is waiting on its stdout: the
/// fd, the line being kept off it, and whether the user is owed the bytes
/// live. One implementation, two owners — this wait and `Transport` — so
/// the rule cannot hold on one side of the announce and not the other.
const ErrPipe = struct {
fd: std.posix.fd_t,
reason: *handoff.Reason,
narrate: bool,
/// Linux's default pipe buffer, and `drainReady`'s bound.
const pipe_capacity = 64 * 1024;
/// One read, returning what it took. EOF (or any error) closes the fd
/// and forgets it: a closed pipe stays readable forever, and a caller
/// that went on polling it would spin hot — the hazard
/// `readAnnounceAbortable`'s `watch_stdin` documents, from the same side.
fn drain(self: *ErrPipe) usize {
if (self.fd < 0) return 0;
var buf: [512]u8 = undefined;
const n = std.posix.read(self.fd, &buf) catch 0;
if (n == 0) {
self.close();
return 0;
}
self.reason.feed(buf[0..n]);
// Relayed whole and unedited. `Reason`'s filtering is for the
// picker row; a user at a bare prompt is owed what ssh actually
// wrote, in the order it wrote it.
if (self.narrate) _ = std.posix.write(std.posix.STDERR_FILENO, buf[0..n]) catch {};
return n;
}
/// Everything the pipe holds NOW, without waiting for more.
fn drainReady(self: *ErrPipe) void {
// BYTES, not reads: the bound is "everything a dead writer can have
// left behind", which is the pipe's capacity and not a syscall count. A
// reads-shaped cap left the reason a line from the middle.
var total: usize = 0;
while (total < pipe_capacity and self.fd >= 0) {
var fds = [_]std.posix.pollfd{
.{ .fd = self.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&fds, 0) catch return;
if (ready == 0 or fds[0].revents == 0) return;
total += self.drain();
}
}
fn close(self: *ErrPipe) void {
if (self.fd < 0) return;
std.posix.close(self.fd);
self.fd = -1;
}
};
/// Deliberately NO deadline: a timer here races a cold `mux d endpoint`
/// spawn, and the abort key already covers a hung ssh.
fn readAnnounceAbortable(
fd: std.posix.fd_t,
alloc: std.mem.Allocator,
carry: ?*std.ArrayList(u8),
abort_fd: std.posix.fd_t,
/// ssh's stderr, joined to this wait so a diagnostic printed while we
/// block on the announce is kept rather than left to fill a pipe.
/// Null in the tests that drive this function off a bare fd.
errp: ?*ErrPipe,
) !?handoff.Endpoint {
return readAnnounceUntil(fd, alloc, carry, abort_fd, errp, null);
}
fn readAnnounceUntil(fd: std.posix.fd_t, alloc: std.mem.Allocator, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, errp: ?*ErrPipe, deadline: ?i64) !?handoff.Endpoint {
var buf: [handoff.announce_max_len]u8 = undefined;
var n: usize = 0;
// A closed abort fd stays readable forever, so once it reports EOF it
// has to stop being polled or this loop spins hot instead of waiting on
// the pipe. The same hazard waitReady documents, reached from the same
// side. -1 (no abort channel) starts unwatched and stays that way.
var watch_stdin = abort_fd >= 0;
while (true) {
// Before the wait, not after it: a line that has already filled the
// buffer is over-long whether or not another byte ever arrives, and
// waiting for one that may never come would turn an error into a hang.
if (n == buf.len) return error.LineTooLong;
const two = abortPoll(fd, abort_fd, watch_stdin);
// Three descriptors, not two: ssh's stderr is watched alongside,
// because the announce can be minutes away (a password prompt) and
// an unread stderr fills at 64k.
var fds = [_]std.posix.pollfd{
two[0],
two[1],
.{
.fd = if (errp) |e| e.fd else -1,
.events = std.posix.POLL.IN,
.revents = 0,
},
};
// No timeout, per the note above. std.posix.poll retries EINTR
// itself, so a SIGWINCH mid-wait is not an error to handle here.
const timeout: i32 = if (deadline) |end| @intCast(std.math.clamp(end - std.time.milliTimestamp(), 0, std.math.maxInt(i32))) else -1;
if (timeout == 0) return error.Timeout;
_ = try std.posix.poll(&fds, timeout);
if (deadline) |end| if (std.time.milliTimestamp() >= end) return error.Timeout;
// `openHandoff` always passes `carry` null: on a first attach this
// fd is -1 and never read — the keystrokes wait in the kernel's tty
// buffer for ssh's prompt, then the shell.
if (fds[1].revents != 0) watch_stdin = try drainAbortFd(abort_fd, alloc, carry);
if (fds[2].revents != 0) _ = errp.?.drain();
if (fds[0].revents == 0) continue;
// One byte, because the frame stream begins at the byte after the
// newline and a buffered read would swallow its first bytes. Polled
// readable, so it does not block; EOF and the over-long line are
// handoff's errors, told in handoff's words.
var one: [1]u8 = undefined;
if (try std.posix.read(fd, &one) == 0) return error.UnterminatedLine;
if (one[0] == '\n') return handoff.parseAnnounce(buf[0..n]);
buf[n] = one[0];
n += 1;
}
}
/// Did ssh work and the announce not, or did ssh not work? Exhaustive by
/// reflection over handoff.zig's error sets: a hand-kept list would let a
/// new member fall through to "cannot reach", the lie this prevents.
fn announceFailed(err: anyerror) bool {
const Announce = handoff.ParseError || handoff.ReadLineError;
inline for (@typeInfo(Announce).error_set.?) |e| {
if (err == @field(anyerror, e.name)) return true;
}
return false;
}
/// What `wallview.runAttach` says when the open fails, and what it exits with.
/// A pair rather than a message, because the abort paths print a line AND exit
/// 0 — two facts that would drift if they lived in two functions.
pub const OpenFailure = struct {
/// Points into the buffer passed to `openFailure` and lives only until the
/// next call on it — EXCEPT the abort paths, whose message is static. Print
/// before reusing; hold one across a second call and only some survive.
msg: []const u8,
/// 1 everywhere except the abort paths, where the user pressed Ctrl-\
/// and stopping when asked is not a failure to exit nonzero over.
exit: u8,
};
/// The buffer `wallview.runAttach` hands `openFailure`.
///
/// Not derived, and it cannot be: the longest operands are argv strings whose
/// only ceiling is ARG_MAX, and sizing for that puts a megabyte on the stack to
/// print one line. A chosen number, made safe by `failedMsg`'s truncation:
/// past this length the message clips instead of anything going wrong.
pub const open_err_len = 8192;
/// Truncating, not failing: this is the user's only account of why the
/// attach did not happen, so a clipped line beats none. The `\n` goes last,
/// so it is the first thing a clip loses.
fn failedMsg(buf: []u8, comptime fmt: []const u8, args: anytype) OpenFailure {
var w: std.Io.Writer = .fixed(buf);
w.print(fmt, args) catch {};
return .{ .msg = w.buffered(), .exit = 1 };
}
/// Ctrl-\ while we were still dialling. Nothing failed, so nothing is
/// reported as a failure — and the exit says so too.
const open_aborted: OpenFailure = .{ .msg = "mux: aborted before attaching\n", .exit = 0 };
/// What the entry dial prints and exits with when `Transport.open` fails. Pure,
/// so the whole error policy can be pinned; the caller owns only the printing.
/// `err` is `anyerror` by design: open unions error sets from five sources and
/// this classifies by VALUE. The cost is that a misspelled prong falls to
/// `else` rather than failing to compile — the literal pins below refuse that.
pub fn openFailure(buf: []u8, target: Target, err: anyerror, reason: []const u8) OpenFailure {
return switch (target) {
// A key the daemon would also have refused, said in the same
// words, because the user's mistake is the same one.
.quic => |q| switch (err) {
// Only the three key classes take the shared body. The `else` below
// is NOT its catch-all: down here an unclassified error is far more
// often a failed dial than an unreadable file, so it names the endpoint.
error.KeyFileMissing,
error.KeyFilePermissive,
error.KeyFileMalformed,
=> blk: {
var body: [quic.key_refusal_len]u8 = undefined;
break :blk failedMsg(
buf,
"mux: {s}\n",
.{quic.keyRefusalBody(&body, err, q.key_path)},
);
},
error.MalformedAddress, error.UnknownHostName => failedMsg(
buf,
"mux: cannot resolve quic://{s}\n",
.{q.host_port},
),
// The handshake is also where a wrong key lands: an external
// PSK that does not match produces no distinguishable
// rejection, just a handshake that never completes. Saying
// both is more honest than guessing which it was.
error.QuicHandshakeFailed => failedMsg(
buf,
"mux: quic://{s} did not answer (wrong key, or no mux d --quic there)\n",
.{q.host_port},
),
error.UserAbort => open_aborted,
else => failedMsg(
buf,
"mux: cannot reach quic://{s}: {s}\n",
.{ q.host_port, @errorName(err) },
),
},
// A QUIC failure is never fatal here — the pipe carries the
// session — so reaching this point means the ssh side failed,
// and there are two quite different ways for that to be true.
.hand => |h| switch (err) {
// Ctrl-\ while dialling or while waiting for the announce: the
// same answer the quic:// arm gives, for the same keystroke.
error.UserAbort => open_aborted,
// ssh, or the remote, said why in a whole sentence. It beats
// both of the lines below: `UnterminatedLine` names what mux
// observed, while `No route to host` names what happened.
else => if (handoff.classifyReason(reason) == .authentication_refused) failedMsg(
buf,
"mux: authentication refused by {s} over ssh: {s}\n",
.{ h.host, reason },
) else if (reason.len > 0) failedMsg(
buf,
"mux: {s} over ssh: {s}\n",
.{ h.host, reason },
) else if (announceFailed(err))
// All this observes is that no announce came. Whether ssh
// reached the host is NOT knowable here: a clean EOF is equally
// a refused connection, a rejected key, or a remote with no
// `mux` on PATH. So the line claims only the observation, which
// is weakly true where "cannot reach" was strongly false.
failedMsg(
buf,
"mux: no endpoint announce from {s} over ssh ({s})\n",
.{ h.host, @errorName(err) },
)
else
// The command never got far enough to say anything: it
// would not spawn, or the pipe itself failed. The error
// name is a fact rather than a guess at a cause — the
// quic:// arm's idiom, for the same reason.
failedMsg(
buf,
"mux: cannot reach {s} over ssh: {s}\n",
.{ h.host, @errorName(err) },
),
},
.via => |cmd| failedMsg(buf, "mux: cannot start --via command: {s}\n", .{cmd}),
// No "is the daemon running?": auto-start checked that moments ago.
// Reaching here means a daemon answered the probe (or was just
// spawned) and then vanished before this connect — the path is
// the whole of what we know, so the path is all we say.
.sock => |path| failedMsg(buf, "mux: cannot connect to {s}\n", .{path}),
};
}
/// What turning a `Target` back into a spelling can fail with — a tile's
/// label bar and the layout sidecar's leaf key. `NoSpelling` is not a
/// defect: the host grammar (hosts.zig) has no form for `--via`, so for
/// that transport there is nothing truthful to write down.
pub const SpellingError = error{ NoSpelling, NoSpace };
/// One session's tile as a user could have typed it: a single argv string
/// in the wall grammar, with `#NAME` last. The one place the client turns a
/// `Target` back into a spelling, so it is pure and pinned rather than
/// inlined into the spawn.
pub fn wallSpelling(out: []u8, target: Target, name: []const u8) SpellingError![]const u8 {
const written = switch (target) {
// The sock spelling carries its own flag INSIDE the string — one
// argv element per tile is the grammar, not two.
.sock => |path| std.fmt.bufPrint(out, hosts.sock_prefix ++ "{s}#{s}", .{ path, name }),
.via => return error.NoSpelling,
// `host_port` is what the user typed after the quic prefix, port and
// all, so it is written back out unexamined.
.quic => |q| std.fmt.bufPrint(out, hosts.quic_prefix ++ "{s}#{s}", .{ q.host_port, name }),
// The bare-HOST form: the wall re-runs the ssh→QUIC handoff from
// the host word, exactly as this client did.
.hand => |h| std.fmt.bufPrint(out, "{s}#{s}", .{ h.host, name }),
};
return written catch error.NoSpace;
}
/// The longest spelling this target can produce. Derived from the longer of
/// the two grammar prefixes rather than from a number: the bare-HOST form
/// has no prefix at all, so the widest case is always one of those two, and
/// a cap taken from them cannot fall short when a literal is re-spelled.
pub fn spellingCap(target: Target) usize {
const prefix_max = @max(hosts.sock_prefix.len, hosts.quic_prefix.len);
const operand: usize = switch (target) {
.sock => |path| path.len,
.via => 0,
.quic => |q| q.host_port.len,
.hand => |h| h.host.len,
};
return prefix_max + operand + 1 + proto.session_name_max;
}
// The grid a birth asks for: `main.DaemonArguments`'s own default, the size `mux d start`
// gives session 0. A session created for a client that claims no size has to be
// born at SOMETHING, and the daemon's own answer needs no explaining.
pub const birth_cols: u16 = 80;
pub const birth_rows: u16 = 24;
// How long a birth may take before the caller is told nothing happened.
// Generous for a fork+exec on a local socket and short enough that a
// browser tile is not left mute: past this the refusal is forwarded and
// the tile reads [refused], which is where it stood before.
const birth_budget_ms: i64 = 3000;
/// A caller that named more than one acceptable answer, waiting on a
/// primitive that takes exactly one. `Link.awaitFrame` waits for `want[0]`
/// and hands every other frame to this sink, which recognises the rest and
/// ends the wait by erroring — the escape the sink contract exists for.
/// The frame is COPIED because awaitFrame frees what it lends the sink.
const AltWant = struct {
alloc: std.mem.Allocator,
/// `want[1..]`: the types awaitFrame is not itself watching for.
rest: []const proto.MsgType,
got: ?proto.Frame = null,
/// Never leaves `roundTrip`, which turns it back into the frame.
const Answer = error.AltAnswer;
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *AltWant = @ptrCast(@alignCast(ctx.?));
for (self.rest) |w| {
if (frame.type != w) continue;
self.got = .{
.type = frame.type,
.payload = try self.alloc.dupe(u8, frame.payload),
};
return Answer;
}
// Narration the caller never asked for; awaitFrame frees it.
}
};
/// One question on a side connection: send `req`, then hand back the first
/// frame whose type is one the caller named, deinit-ing every other frame
/// the daemon says on the way. The frame returned is the caller's to deinit.
///
/// The errors are thin on purpose — `Timeout` for a budget spent, `Closed`
/// for a wait that ended without an answer, `FrameTooLarge` for a peer that
/// spoke and got the framing wrong, `OutOfMemory` for this machine's
/// allocator. What one MEANS is the caller's to say: the same close is a
/// refusal to a birth and a box that is down to a poll. `FrameTooLarge` is
/// deliberately NOT folded into `Closed` — a daemon whose reply cannot be
/// framed is a different fact from one that is not there, and a birth
/// reports the second as a refusal.
///
/// `Closed` is also what a failed `poll(2)` inside the wait becomes, since
/// `Link.awaitFrame` owns the poll and has one word for "the wait cannot go
/// on". That is a widening: this used to answer `Transport` there. The two
/// failures poll can report are out of kernel memory and a bug, so the
/// mislabel costs a "refused" instead of a "transport error" in a case
/// where nothing on the box is working anyway.
fn roundTrip(
tr: *Transport,
alloc: std.mem.Allocator,
req: proto.MsgType,
payload: []const u8,
want: []const proto.MsgType,
deadline: i64,
) !proto.Frame {
std.debug.assert(want.len != 0); // a question with no acceptable answer
try tr.writeFrame(req, payload);
const left = deadline - std.time.milliTimestamp();
if (left <= 0) return error.Timeout;
var alt: AltWant = .{ .alloc = alloc, .rest = want[1..] };
// The wait takes milliseconds as a u32. Every budget in the tree is
// seconds at most, and clamping is what a caller that passed more would
// have meant anyway — a cast would panic on it.
const ms: u32 = @intCast(@min(left, std.math.maxInt(u32)));
const f = tr.link.awaitFrame(alloc, want[0], ms, .{
.ctx = &alt,
.on = AltWant.on,
}) catch |e| switch (e) {
AltWant.Answer => return alt.got.?,
error.Closed => return error.Closed,
error.OutOfMemory => return error.OutOfMemory,
error.FrameTooLarge => return error.FrameTooLarge,
// Nothing reaches here today — awaitFrame raises only the three
// above and whatever the sink returns — but the arm keeps this
// function's error set the thin one its callers switch on rather
// than letting the sink's `anyerror` widen it.
else => return error.Transport,
};
return f orelse error.Timeout;
}
/// Creates `name` on `target` over a connection of its own, then leaves.
pub fn birthSession(
alloc: std.mem.Allocator,
target: Target,
name: []const u8,
cols: u16,
rows: u16,
) !void {
// `Refused` is the daemon's no, `Timeout` a daemon that never answered,
// `Transport` a wire that broke — they differ only in what a log can say.
// The budget below is the ANSWER budget and starts after the dial, since
// each target bounds its own. A SIDE connection, not the caller's: a tile
// that claimed a size to get its session made would keep claiming it.
var tr = try Transport.open(alloc, target, null, -1, null);
defer tr.close();
var buf: [proto.attach_max_len]u8 = undefined;
// Resume args 0/0: this connection holds nothing and wants the
// cheapest thing the daemon can answer with.
const attach = proto.encodeAttachNamed(&buf, cols, rows, 0, 0, proto.wireName(name));
const deadline = std.time.milliTimestamp() + birth_budget_ms;
// Both answers are asked for, because both end the wait:
// `server_sessions.resolve` refuses with an `exit_status` before any
// snapshot, and waiting past it for a snapshot that is not coming would
// spend the whole budget on an answer already given.
const f = roundTrip(&tr, alloc, .attach, attach, &.{ .snapshot, .exit_status }, deadline) catch |e| return switch (e) {
// A daemon that hung up before it answered refused this attach.
error.Closed => error.Refused,
error.Timeout => error.Timeout,
// Not `Refused`: a read that failed is a transport that broke, and
// telling the browser its attach was refused for something nothing
// refused is a lie the page then shows.
else => error.Transport,
};
defer f.deinit(alloc);
if (f.type != .snapshot) return error.Refused;
// The daemon made a grid, so the session exists — and this connection
// leaves, because the session outlives it and a lingering client would
// hold a slot the browser tile needs.
tr.writeFrame(.detach, "") catch {};
}
/// What the daemon said about an `end_req`. The reason is COPIED rather
/// than borrowed: the frame it arrived in is freed before this returns, and
/// the picker builds its notice from the reason after the call.
pub const EndOutcome = struct {
accepted: bool,
others: u8,
reason_buf: [proto.end_reply_max_len]u8 = undefined,
reason_len: usize = 0,
pub fn reason(self: *const EndOutcome) []const u8 {
return self.reason_buf[0..self.reason_len];
}
};
/// `end_req` on a side connection of its own: the picker ends a session
/// that may have no pane on this wall, so there is no pump to ask
/// through. The daemon owns the two-step; this only carries `force`.
pub fn endSession(alloc: std.mem.Allocator, target: Target, name: []const u8, force: bool) !EndOutcome {
var tr = try Transport.open(alloc, target, null, -1, null);
defer tr.close();
var buf: [proto.end_req_max_len]u8 = undefined;
const req = proto.encodeEndReq(&buf, force, proto.wireName(name));
const deadline = std.time.milliTimestamp() + birth_budget_ms;
// `exit_status` is asked for as well because it ends the wait: a daemon
// that kills the session before it answers has already said everything
// it is going to, and waiting out the budget for a reply that is not
// coming would park the popup for three seconds.
const f = roundTrip(&tr, alloc, .end_req, req, &.{ .end_reply, .exit_status }, deadline) catch |e| return switch (e) {
error.Timeout => error.Timeout,
else => error.Transport,
};
defer f.deinit(alloc);
// An `exit_status` (or a daemon too old to have an `end_req` arm at all)
// is not a verdict this can report a count from.
if (f.type != .end_reply) return error.Refused;
const r = proto.parseEndReply(f.payload) orelse return error.Refused;
var out: EndOutcome = .{ .accepted = r.accepted, .others = r.others };
// The tail is a PEER's bytes and the buffer is this frame's size, not
// the peer's: a reason longer than any word the daemon owns is cut.
const n = @min(r.reason.len, out.reason_buf.len);
@memcpy(out.reason_buf[0..n], r.reason[0..n]);
out.reason_len = n;
return out;
}
/// Every way of not reaching a daemon, except running out of memory.
fn oomOrTransport(e: anyerror) error{ OutOfMemory, Transport } {
// `mux hosts` prints `[unreachable]` for a Transport, so an allocation
// failure folded into it blames a box that is up for a fault here.
return if (e == error.OutOfMemory) error.OutOfMemory else error.Transport;
}
/// One `sessions_req` on a side connection: the wall's per-host poll.
pub fn listSessions(
alloc: std.mem.Allocator,
target: Target,
out: *[proto.sessions_reply_max]u8,
budget_ms: i64,
answered: ?*std.meta.Tag(Link),
/// Where the dial's leavings go when it fails: the picker row quotes
/// ssh's line, so a box that is down says why instead of `unreachable`.
dial: ?*handoff.Dial,
) ![]const u8 {
// A fresh connection per poll, so the observer idle deadline and the redial
// backoff stay the pump's problem. Recorded BEFORE the open, because a
// `hand` target's COST is the target's and not the reply's: an ssh login is
// spent either way, and only the caller's backoff can stop paying.
if (answered) |a| a.* = switch (target) {
.hand, .via => .pipe,
.quic => .quic,
.sock => .fd,
};
var tr = Transport.open(alloc, target, null, -1, dial) catch |e| return oomOrTransport(e);
defer tr.close();
// The handoff picks its own link, so only the success case knows it.
if (answered) |a| a.* = tr.link;
const deadline = std.time.milliTimestamp() + budget_ms;
// A daemon too old for the verb answers nothing at all, so the timeout is
// what names it; anything else on this connection is narration the poll
// never asked for, which `roundTrip` drops. A daemon that hung up is a
// box the wall cannot reach, the same as one that never came up.
const f = roundTrip(&tr, alloc, .sessions_req, "", &.{.sessions_reply}, deadline) catch |e|
return if (e == error.Timeout) e else oomOrTransport(e);
defer f.deinit(alloc);
if (f.payload.len > out.len) return error.Transport;
@memcpy(out[0..f.payload.len], f.payload);
return out[0..f.payload.len];
}
/// A daemon on a wall: what to dial, and the line that named it. The
/// spelling is the layout sidecar's key and what `mux hosts` prints back,
/// so it is kept verbatim rather than rebuilt.
pub const HostSpec = struct { spelling: []const u8, target: Target, poll_target: Target };
pub const HostResolveError = hosts.ParseError || SpecError;
/// A host line names a daemon, not a session; this is its target.
pub fn resolveHost(
alloc: std.mem.Allocator,
spelling: []const u8,
key: ?[]const u8,
idle_ms: u32,
) HostResolveError!HostSpec {
// A host line is a listing, not an attach anyone waited for. The
// POLLER runs off this spec once a second: an asked copy would print
// the fallback line onto the wall's alternate screen every cycle, and
// would start a daemon on a box whose owner just stopped one.
const target = try Target.fromSpec(alloc, try hosts.parse(spelling), key, idle_ms, false);
return .{ .spelling = spelling, .target = target, .poll_target = try pollTargetFor(alloc, target) };
}
/// What `HostSpec.poll_target` is: the same daemon, dialled by a recipe
/// nobody is sitting in front of.
pub fn pollTargetFor(alloc: std.mem.Allocator, target: Target) !Target {
// Only `hand` can ask a terminal for anything, so only `hand` needs a
// second recipe: a poll runs under a wall that owns the screen, where
// an ssh password prompt goes to /dev/tty under the panes and a
// fallback line goes onto the alternate screen.
const h = switch (target) {
.hand => |hd| hd,
else => return target,
};
const r = try handoff.recipeFor(alloc, h.host, true);
return .{ .hand = HandoffTarget.fromRecipe(h.host, r, h.idle_ms, false) };
}
/// Polling, not a push: a subscription is a new daemon concept, and one
/// small frame a second per host over a link that already carries deltas is
/// not a cost worth designing around.
const host_poll_ms: u64 = 1000;
/// How long before this host is asked again, given the link that answered.
pub fn pollDelayMs(link: std.meta.Tag(Link)) u64 {
// A pipe link cost a whole sshd login: the cached QUIC coordinates were
// dead or blocked, so this cycle spawned `ssh`, read the announce and
// killed it. A second of that, forever, is a remote auth log the wall
// wrote — the list is worth a tenth of the freshness.
return if (link == .pipe) host_poll_ms * 10 else host_poll_ms;
}
/// ONE host's session poll, for every front that shows a wall, so a tile born
/// in a terminal and one born in a browser come from the same question. Whether
/// to keep going and how to wake the reader are the CALLER's; everything else,
/// including riding out a blip, is the same on both fronts.
pub const SessionPoll = struct {
list_mu: std.Thread.Mutex = .{},
list: [proto.sessions_reply_max]u8 = undefined,
list_len: usize = 0,
/// News for the reader: a poll finished, well or badly.
list_ready: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
reachable: std.atomic.Value(bool) = std.atomic.Value(bool).init(true),
/// Why the last poll failed, in ssh's own words. Under `list_mu` with
/// the list, because a row paints both in one pass.
reason: handoff.Reason = .{},
/// A birth asks for the next poll NOW rather than in a second — no
/// wall may lag the session the user just made.
poke: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
/// The meta line off the last answer, parsed HERE once per cycle so no
/// row or bar re-walks the payload. Under `list_mu` with the list it
/// came from; `meta_present` false is an old daemon or no answer yet.
meta_version: [proto.sessions_meta_version_max]u8 = undefined,
meta_version_len: usize = 0,
meta_stale: bool = false,
meta_present: bool = false,
/// The last answer, copied out from under `list_mu` into the caller's
/// buffer: `list` is this poller's own and is overwritten by the next
/// poll, so a reader that kept a slice of it would read a half-written
/// list.
pub fn snapshot(self: *SessionPoll, buf: *[proto.sessions_reply_max]u8) []const u8 {
self.list_mu.lock();
defer self.list_mu.unlock();
@memcpy(buf[0..self.list_len], self.list[0..self.list_len]);
return buf[0..self.list_len];
}
/// `snapshot`'s shape for the other half of a row: copied out from
/// under the lock, because the poller overwrites its own on every
/// cycle.
pub fn reasonSnapshot(self: *SessionPoll, buf: *[handoff.reason_max]u8) []const u8 {
self.list_mu.lock();
defer self.list_mu.unlock();
const said = self.reason.slice();
@memcpy(buf[0..said.len], said);
return buf[0..said.len];
}
/// The daemon's word about itself off the last ANSWER, parsed once here
/// so every row and bar shares one verdict: absent until a reply carries
/// the line, absent again when a reply stops carrying it — last cycle's
/// drift must not dress a daemon that no longer states any.
pub fn metaSnapshot(
self: *SessionPoll,
buf: *[proto.sessions_meta_version_max]u8,
) ?proto.SessionsMeta {
self.list_mu.lock();
defer self.list_mu.unlock();
if (!self.meta_present) return null;
@memcpy(buf[0..self.meta_version_len], self.meta_version[0..self.meta_version_len]);
return .{ .version = buf[0..self.meta_version_len], .stale = self.meta_stale };
}
/// Runtime hooks, not a comptime context: one compiled loop, two
/// fronts, and a test may hand it a counter.
pub const Hooks = struct {
ctx: *anyopaque,
keep: *const fn (*anyopaque) bool,
wake: *const fn (*anyopaque) void,
};
/// Blocks until `keep` says stop. One thread per host.
pub fn run(self: *SessionPoll, target: Target, hooks: Hooks) void {
var out: [proto.sessions_reply_max]u8 = undefined;
while (hooks.keep(hooks.ctx)) {
// A connection of its own per poll: the observer idle deadline
// and the redial backoff stay the pump's problem, and this
// thread owns no transport between polls that a teardown would
// have to reach.
var link: std.meta.Tag(Link) = .fd;
// The poll's OWN reason, copied in under the lock below: this
// thread is the only writer, and a row must never read a
// sentence being written into it.
var said: handoff.Dial = .{};
const got = listSessions(std.heap.page_allocator, target, &out, 2000, &link, &said) catch null;
if (got) |list| {
self.list_mu.lock();
@memcpy(self.list[0..list.len], list);
self.list_len = list.len;
// A host that answered has no reason to give, and last
// cycle's would sit on a reachable row forever.
self.reason.clear();
// The meta verdict follows the same rule as the reason: it
// is the LAST answer's, so a reply without the line — an
// old daemon, or one downgraded under us — clears it.
if (proto.parseSessionsMeta(list)) |m| {
@memcpy(self.meta_version[0..m.version.len], m.version);
self.meta_version_len = m.version.len;
self.meta_stale = m.stale;
self.meta_present = true;
} else self.meta_present = false;
self.list_mu.unlock();
self.reachable.store(true, .release);
} else {
self.list_mu.lock();
self.reason = said.reason;
self.list_mu.unlock();
self.reachable.store(false, .release);
}
self.list_ready.store(true, .release);
hooks.wake(hooks.ctx);
var slept: u64 = 0;
const wait = pollDelayMs(link);
// Sliced so a poke or a stop is felt in 50 ms, not in a poll
// interval — a chord that births still pokes through the
// stretched `.pipe` wait, so a user's own action costs nothing.
while (slept < wait and
!self.poke.swap(false, .acq_rel) and
hooks.keep(hooks.ctx)) : (slept += 50)
std.Thread.sleep(50 * std.time.ns_per_ms);
}
}
};
/// Zero first — a link that just died usually reconnects now. No retry cap.
pub fn nextBackoffMs(prev: u64) u64 {
return if (prev == 0) 200 else @min(prev * 2, 2000);
}
test "pollDelayMs: a poll that cost an ssh login is asked ten times less often" {
// A pipe link is an sshd login per cycle — a remote auth log the wall
// writes. Every other link is a connect, and stays at the second.
try std.testing.expectEqual(@as(u64, 10_000), pollDelayMs(.pipe));
try std.testing.expectEqual(@as(u64, 1_000), pollDelayMs(.fd));
try std.testing.expectEqual(@as(u64, 1_000), pollDelayMs(.quic));
}
test "SessionPoll.run: a keep that says stop is felt within one sleep slice, not one poll interval" {
// The wall's teardown and the picker's `x` both end a poller by answering
// `keep` false. Re-reading it once per `pollDelayMs` would hold the wall's
// exit for a second per host, so the 50 ms slice is the claim.
var tmp = try TmpDir.make();
defer tmp.cleanup();
const path = try std.fmt.allocPrint(std.testing.allocator, "{s}/nobody.sock", .{tmp.path()});
defer std.testing.allocator.free(path);
const Ctx = struct {
left: u32 = 3,
woke: u32 = 0,
fn keep(p: *anyopaque) bool {
const self: *@This() = @ptrCast(@alignCast(p));
if (self.left == 0) return false;
self.left -= 1;
return true;
}
fn wake(p: *anyopaque) void {
const self: *@This() = @ptrCast(@alignCast(p));
self.woke += 1;
}
};
var ctx = Ctx{};
var poll: SessionPoll = .{};
const start = std.time.milliTimestamp();
poll.run(.{ .sock = path }, .{ .ctx = &ctx, .keep = Ctx.keep, .wake = Ctx.wake });
const elapsed = std.time.milliTimestamp() - start;
// One poll per pass, and a wake after each — the keyboard hears about a
// failed poll, or a host that went quiet would never be repainted.
try std.testing.expectEqual(@as(u32, 1), ctx.woke);
try std.testing.expect(!poll.reachable.load(.acquire));
try std.testing.expect(poll.list_ready.load(.acquire));
// Two of the three `keep`s are spent inside the sleep, so the run is
// slices long, nowhere near `pollDelayMs`.
try std.testing.expect(elapsed < 1_000);
}
test "SessionPoll: a failed poll keeps ssh's last line, and a good one clears it" {
// The picker row's other half. `unreachable` alone tells the user
// nothing they can act on; `No route to host` tells them the box is
// off and `Permission denied` tells them it is not.
var tmp = try TmpDir.make();
defer tmp.cleanup();
// One pass of the loop per run: `keep` says yes once, then no.
const Ctx = struct {
left: u32,
fn keep(p: *anyopaque) bool {
const self: *@This() = @ptrCast(@alignCast(p));
if (self.left == 0) return false;
self.left -= 1;
return true;
}
fn wake(_: *anyopaque) void {}
};
var poll: SessionPoll = .{};
// A host whose ssh dies with a sentence, which is what a box that is
// off the network looks like from here.
var ctx = Ctx{ .left = 1 };
poll.run(.{ .hand = .{
.host = "nowhere",
.ssh_argv = &.{ "/bin/sh", "-c", "printf 'ssh: no route\n' >&2; exit 255" },
.cache_path = null,
.deadline_ms = 200,
} }, .{ .ctx = &ctx, .keep = Ctx.keep, .wake = Ctx.wake });
try std.testing.expect(!poll.reachable.load(.acquire));
var said_buf: [handoff.reason_max]u8 = undefined;
try std.testing.expectEqualStrings("ssh: no route", poll.reasonSnapshot(&said_buf));
// The SAME poller then reaches a daemon. A reason that outlived its
// failure would sit on a row that is answering, blaming a box that is
// up — which is the state a poller spends most of its life in.
const sp = try std.fmt.allocPrint(std.testing.allocator, "{s}/heal.sock", .{tmp.path()});
defer std.testing.allocator.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = ListFake{ .listener = try addr.listen(.{}), .reply = "0\n" };
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, ListFake.serve, .{&fake});
ctx = Ctx{ .left = 1 };
poll.run(.{ .sock = sp }, .{ .ctx = &ctx, .keep = Ctx.keep, .wake = Ctx.wake });
th.join();
try std.testing.expect(poll.reachable.load(.acquire));
try std.testing.expectEqualStrings("", poll.reasonSnapshot(&said_buf));
}
test "SessionPoll: the daemon's meta rides the poll, and a wordless reply clears it" {
var tmp = try TmpDir.make();
defer tmp.cleanup();
const Ctx = struct {
left: u32,
fn keep(p: *anyopaque) bool {
const self: *@This() = @ptrCast(@alignCast(p));
if (self.left == 0) return false;
self.left -= 1;
return true;
}
fn wake(_: *anyopaque) void {}
};
var poll: SessionPoll = .{};
var vbuf: [proto.sessions_meta_version_max]u8 = undefined;
// A daemon that states a version and a replaced image, beside names the
// wall still has to see whole.
const sp = try std.fmt.allocPrint(std.testing.allocator, "{s}/meta.sock", .{tmp.path()});
defer std.testing.allocator.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = ListFake{
.listener = try addr.listen(.{}),
.reply = "0\nwork\n# mux 9.9.9-new stale",
};
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, ListFake.serve, .{&fake});
var ctx = Ctx{ .left = 1 };
poll.run(.{ .sock = sp }, .{ .ctx = &ctx, .keep = Ctx.keep, .wake = Ctx.wake });
th.join();
const meta = poll.metaSnapshot(&vbuf) orelse return error.MetaAbsent;
try std.testing.expectEqualStrings("9.9.9-new", meta.version);
try std.testing.expect(meta.stale);
var lbuf: [proto.sessions_reply_max]u8 = undefined;
try std.testing.expectEqualStrings("0\nwork\n# mux 9.9.9-new stale", poll.snapshot(&lbuf));
// The SAME poller then hears an old daemon — no line at all. The verdict
// leaves with the evidence, exactly as `reason` clears on an answer.
const sp2 = try std.fmt.allocPrint(std.testing.allocator, "{s}/old.sock", .{tmp.path()});
defer std.testing.allocator.free(sp2);
const addr2 = try std.net.Address.initUnix(sp2);
var fake2 = ListFake{ .listener = try addr2.listen(.{}), .reply = "0\n" };
defer fake2.listener.deinit();
const th2 = try std.Thread.spawn(.{}, ListFake.serve, .{&fake2});
ctx = Ctx{ .left = 1 };
poll.run(.{ .sock = sp2 }, .{ .ctx = &ctx, .keep = Ctx.keep, .wake = Ctx.wake });
th2.join();
try std.testing.expect(poll.metaSnapshot(&vbuf) == null);
}
test "reconnect backoff: 0 then 200 doubling to the 2s cap, never beyond" {
try std.testing.expectEqual(@as(u64, 200), nextBackoffMs(0));
try std.testing.expectEqual(@as(u64, 400), nextBackoffMs(200));
try std.testing.expectEqual(@as(u64, 800), nextBackoffMs(400));
try std.testing.expectEqual(@as(u64, 1600), nextBackoffMs(800));
try std.testing.expectEqual(@as(u64, 2000), nextBackoffMs(1600));
try std.testing.expectEqual(@as(u64, 2000), nextBackoffMs(2000));
}
test "Transport.close is idempotent: the abort path closes what a re-dial already closed" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const dir_path = tmp.path();
const sock_path = try std.fmt.allocPrint(alloc, "{s}/t.sock", .{dir_path});
defer alloc.free(sock_path);
const addr = try std.net.Address.initUnix(sock_path);
var listener = try addr.listen(.{});
defer listener.deinit();
var transport = try Transport.open(alloc, .{ .sock = sock_path }, null, -1, null);
// A re-dial closes the dead transport at entry, and an abort then closes it
// again through the pump's `defer`. Without a sentinel that is `close(2)` on
// a stale fd — EBADF, which `std.posix.close` maps to `unreachable`.
transport.close();
transport.close();
transport.close();
try std.testing.expect(transport.link == .fd);
try std.testing.expectEqual(@as(std.posix.fd_t, -1), transport.link.fd);
}
test "connectAgent: a live socket connects, a dead path returns null" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sock_path = try std.fmt.allocPrint(alloc, "{s}/agent.sock", .{tmp.path()});
defer alloc.free(sock_path);
const addr = try std.net.Address.initUnix(sock_path);
var listener = try addr.listen(.{});
defer listener.deinit();
const fd = connectAgent(sock_path) orelse {
std.debug.print("a bound agent socket must connect\n", .{});
return error.TestUnexpectedResult;
};
std.posix.close(fd);
// The whole point of the `?fd` signature: the daemon's fallback is a
// client that answers `agent_open` with `agent_close`, and it can only
// do that if a missing agent is a null rather than an error to bubble.
const gone = try std.fmt.allocPrint(alloc, "{s}.gone", .{sock_path});
defer alloc.free(gone);
try std.testing.expectEqual(@as(?std.posix.fd_t, null), connectAgent(gone));
// No SSH_AUTH_SOCK in the environment is an empty path, and that is a
// no rather than a connect attempt at "".
try std.testing.expectEqual(@as(?std.posix.fd_t, null), connectAgent(""));
}
test "Transport.open: a --via target yields a pipe, a --sock target an fd" {
// The dispatch decision itself, which nothing else pinned: every other open
// test asserts what the handoff CHOSE, so swapping open's two arms used to
// pass the whole suite. Ordered AFTER the idempotence test, because the
// double close below is safe only while the sentinel holds — and a panic
// prints no assertion, so running first would hide the pin that names it.
const alloc = std.testing.allocator;
// A command that stays alive on stdin, so the link is unambiguously a
// live child rather than one that raced us to exit.
var v = Transport.open(alloc, .{ .via = "cat" }, null, -1, null) catch |err| {
std.debug.print(
"a --via target must spawn a command, not connect a socket: open failed with {s}\n",
.{@errorName(err)},
);
return err;
};
// Paired with the explicit close below, which is safe precisely because
// close is idempotent — the property the preceding test pins.
defer v.close();
if (v.link != .pipe) std.debug.print(
"a --via target must yield a pipe link, got .{s}\n",
.{@tagName(v.link)},
);
try std.testing.expect(v.link == .pipe);
// Read before the close, not after: kill() reaps, and the child value
// goes with the link when close resets it.
const pid = v.link.pipe.child.id;
v.close();
// Observed, not assumed: close() owns the reaping (Child.kill waitpid()s
// internally), so the pid must be gone rather than a zombie — signal 0
// would still find a zombie. Pid reuse this fast would take wrapping the
// whole pid space between these two statements.
if (std.posix.kill(pid, 0)) |_| {
std.debug.print(
"close() must reap the via child: pid {d} still findable\n",
.{pid},
);
} else |err| if (err != error.ProcessNotFound) {
std.debug.print(
"close() must reap the via child: pid {d} answered {s}\n",
.{ pid, @errorName(err) },
);
}
try std.testing.expectError(error.ProcessNotFound, std.posix.kill(pid, 0));
// The other arm, against a listener that never accepts: connect(2)
// succeeds off the backlog, which is all this decision needs.
var tmp_s = try TmpDir.make();
defer tmp_s.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/dispatch.sock", .{tmp_s.path()});
defer alloc.free(sp);
const a = try std.net.Address.initUnix(sp);
var listener = try a.listen(.{});
defer listener.deinit();
var s = Transport.open(alloc, .{ .sock = sp }, null, -1, null) catch |err| {
std.debug.print(
"a --sock target must connect the socket, not spawn a command: open failed with {s}\n",
.{@errorName(err)},
);
return err;
};
defer s.close();
if (s.link != .fd) std.debug.print(
"a --sock target must yield an fd link, got .{s}\n",
.{@tagName(s.link)},
);
try std.testing.expect(s.link == .fd);
}
test "local first attach leaves preloaded detach input for the client while dedicated cancellation still aborts" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const path = try std.fmt.allocPrint(alloc, "{s}/queued-input.sock", .{tmp.path()});
defer alloc.free(path);
const addr = try std.net.Address.initUnix(path);
var listener = try addr.listen(.{});
defer listener.deinit();
const targets = [_]Target{ .{ .sock = path }, .{ .via = "cat" } };
for (targets) |target| {
for ([_][]const u8{ "\x1cd", "\x1c\x1c", "ordinary input\n" }) |input| {
var stdin = try FakeStdin.install(input);
defer stdin.deinit();
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var transport = try Transport.open(alloc, target, &carry, std.posix.STDIN_FILENO, null);
defer transport.close();
try std.testing.expectEqual(@as(usize, 0), carry.items.len);
// Bound the read: a regression must fail instead of waiting for
// bytes that opening mistakenly consumed.
var fds = [_]std.posix.pollfd{.{ .fd = std.posix.STDIN_FILENO, .events = std.posix.POLL.IN, .revents = 0 }};
try std.testing.expectEqual(@as(usize, 1), try std.posix.poll(&fds, 100));
var bytes: [32]u8 = undefined;
const n = try std.posix.read(std.posix.STDIN_FILENO, &bytes);
try std.testing.expectEqualStrings(input, bytes[0..n]);
}
// GUI pumps and discovery jobs have a dedicated channel, no carry.
// Their preloaded cancellation must still stop opening immediately.
const cancel = try std.posix.pipe();
defer std.posix.close(cancel[0]);
defer std.posix.close(cancel[1]);
_ = try std.posix.write(cancel[1], "\x1c");
try std.testing.expectError(error.UserAbort, Transport.open(alloc, target, null, cancel[0], null));
}
}
test "--via: the words reach the program verbatim — no shell splits, expands or quotes them" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
// The oracle is the CHILD's own view of its argv. A shell between us
// and it would eat the `;`, expand `$HOME` and drop the quotes, and
// every one of those would show up here as a different byte.
const prog = try std.fmt.allocPrint(alloc, "{s}/echoargs", .{tmp.path()});
defer alloc.free(prog);
try std.fs.cwd().writeFile(.{
.sub_path = prog,
.data = "#!/bin/sh\nfor a in \"$@\"; do printf '<%s>' \"$a\"; done\n",
.flags = .{ .mode = 0o755 },
});
const cmd = try std.fmt.allocPrint(alloc, "{s} a;b $HOME 'q'", .{prog});
defer alloc.free(cmd);
var v = try Transport.open(alloc, .{ .via = cmd }, null, -1, null);
defer v.close();
// To EOF: the child writes one arg per printf, so a single read sees
// only the first word and would pass on a shell that ate the rest.
var buf: [512]u8 = undefined;
var got: usize = 0;
while (true) {
const n = std.posix.read(v.link.pipe.r, buf[got..]) catch 0;
if (n == 0) break;
got += n;
}
try std.testing.expectEqualStrings("<a;b><$HOME><'q'>", buf[0..got]);
}
test "--via: a command of nothing but blanks names no program" {
try std.testing.expectError(
error.EmptyViaCommand,
Transport.viaArgv(std.testing.allocator, " \t "),
);
}
test "spawnPipe: the child is exec'd from a copy — an argv freed after spawn still ran" {
// The lifetime `Transport.open`'s `--via` arm depends on: it frees the
// argv the moment spawn returns. If std ever kept the slice instead of
// duplicating it, that free would be a use-after-free nothing else here
// would catch — the child would have exec'd correctly already.
const alloc = std.testing.allocator;
// The WORDS are duped as well as the array. String literals live for
// the whole program, so freeing an array of them would only have caught
// a std that kept `argv.ptr` — never one that kept `argv[i].ptr`, which
// is the same bug one level down and the one this claim also makes.
const argv = try alloc.alloc([]const u8, 2);
argv[0] = try alloc.dupe(u8, "/bin/echo");
argv[1] = try alloc.dupe(u8, "copied");
var child = try Transport.spawnPipe(alloc, argv, null);
for (argv) |w| alloc.free(w);
alloc.free(argv);
// Reuse the freed pages before reading, so a std that kept the pointer
// is reading somebody else's bytes rather than its own stale ones.
const churn = try alloc.alloc([]const u8, 2);
@memset(churn, "xxxxxxx");
alloc.free(churn);
const wchurn = try alloc.alloc(u8, 16);
@memset(wchurn, 'x');
alloc.free(wchurn);
var buf: [64]u8 = undefined;
const n = try std.posix.read(child.stdout.?.handle, &buf);
try std.testing.expectEqualStrings("copied\n", buf[0..n]);
_ = try child.kill();
}
test "spawnPipe: an askpass dial hands ssh the three variables, and a plain dial hands it none" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
// The oracle is the CHILD's own environment, read out of the process
// that was actually exec'd. Asserting on the EnvMap we built would pin
// the map and say nothing about whether the child ever saw it.
const prog = try std.fmt.allocPrint(alloc, "{s}/dumpenv", .{tmp.path()});
defer alloc.free(prog);
var body_buf: [512]u8 = undefined;
const body = try std.fmt.bufPrint(&body_buf, "#!/bin/sh\nenv > {s}/$1\n", .{tmp.path()});
try std.fs.cwd().writeFile(.{ .sub_path = prog, .data = body, .flags = .{ .mode = 0o755 } });
{
var c = try Transport.spawnPipe(alloc, &.{ prog, "with" }, .{
.sock = "/run/ask.sock",
.exe = "/opt/mux",
});
_ = try c.wait();
}
var with_buf: [8192]u8 = undefined;
const with = try shimSaid(tmp.path(), "with", &with_buf);
// `force` and not merely a helper: without it ssh uses SSH_ASKPASS only
// when there is no tty, and the wall this runs under has one.
try std.testing.expect(std.mem.indexOf(u8, with, "SSH_ASKPASS_REQUIRE=force") != null);
try std.testing.expect(std.mem.indexOf(u8, with, "SSH_ASKPASS=/opt/mux") != null);
try std.testing.expect(std.mem.indexOf(u8, with, "MUX_ASKPASS_SOCK=/run/ask.sock") != null);
// The rest of the environment is still the parent's: ssh reads
// SSH_AUTH_SOCK, HOME and TERM out of it, and a map built from nothing
// would break the agent forwarding this dial also carries.
try std.testing.expect(std.mem.indexOf(u8, with, "PATH=") != null);
{
var c = try Transport.spawnPipe(alloc, &.{ prog, "without" }, null);
_ = try c.wait();
}
var out_buf: [8192]u8 = undefined;
const without = try shimSaid(tmp.path(), "without", &out_buf);
// A poll and the entry dial come through here too, and either one
// carrying a socket would be a prompt raised where no popup can appear.
try std.testing.expect(std.mem.indexOf(u8, without, "MUX_ASKPASS_SOCK") == null);
try std.testing.expect(std.mem.indexOf(u8, without, "SSH_ASKPASS_REQUIRE") == null);
}
test "the announce reader consumes the newline and NOT the byte after it" {
// The property the announce-then-frames protocol stands on. A buffered read
// here would take 'X' — the frame stream's first byte — into a buffer that
// is thrown away, and the session would hang with nothing to point at.
const alloc = std.testing.allocator;
const fds = try std.posix.pipe();
defer std.posix.close(fds[0]);
var buf: [handoff.announce_max_len]u8 = undefined;
const ep: handoff.Endpoint = .{ .port = 4433, .key = [_]u8{0xAB} ** 32 };
_ = try std.posix.write(fds[1], try handoff.formatAnnounce(&buf, ep));
_ = try std.posix.write(fds[1], "X");
// Closed before the read: with the write end open, a reader that swallowed
// the 'X' would block the assertion below forever, and a hung suite names
// nothing. EOF turns the catch into a printed "expected 1, found 0".
std.posix.close(fds[1]);
const got = (try readAnnounceAbortable(fds[0], alloc, null, -1, null)).?;
try std.testing.expectEqual(ep.port, got.port);
try std.testing.expectEqualSlices(u8, &ep.key, &got.key);
var one: [1]u8 = undefined;
try std.testing.expectEqual(@as(usize, 1), try std.posix.read(fds[0], &one));
try std.testing.expectEqual(@as(u8, 'X'), one[0]);
}
test "the announce reader: `endpoint none` is null, EOF and an over-long line are handoff's errors" {
const alloc = std.testing.allocator;
{
const fds = try std.posix.pipe();
defer std.posix.close(fds[0]);
_ = try std.posix.write(fds[1], handoff.announce_none);
std.posix.close(fds[1]);
try std.testing.expectEqual(
@as(?handoff.Endpoint, null),
try readAnnounceAbortable(fds[0], alloc, null, -1, null),
);
}
{
const fds = try std.posix.pipe();
defer std.posix.close(fds[0]);
_ = try std.posix.write(fds[1], "endpoi");
std.posix.close(fds[1]);
try std.testing.expectError(
handoff.ReadLineError.UnterminatedLine,
readAnnounceAbortable(fds[0], alloc, null, -1, null),
);
}
{
const fds = try std.posix.pipe();
defer std.posix.close(fds[0]);
// Write end stays OPEN: the over-long line is an error the moment
// the buffer is full, not something the reader waits on a newline
// to discover. A reader that checked after the poll would hang here.
defer std.posix.close(fds[1]);
const long = [_]u8{'a'} ** handoff.announce_max_len;
_ = try std.posix.write(fds[1], &long);
try std.testing.expectError(
handoff.ReadLineError.LineTooLong,
readAnnounceAbortable(fds[0], alloc, null, -1, null),
);
}
}
test "handoff: endpoint-none rides the open pipe with no deadline paid" {
// A fake `ssh HOST mux d endpoint` that announces `none`. What this pins
// is Transport.open's DECISION — the announce-less remote gets a
// child-backed transport, immediately — not a whole session, which
// would need a daemon on the far end of the pipe.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
// Inert, but ours: this open is handed fd 0 as its abort fd, and a
// developer's terminal is not a fixture.
var stdin = try FakeStdin.install("");
defer stdin.deinit();
const t0 = std.time.milliTimestamp();
var t = try Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", "printf 'endpoint none\\n'; cat >/dev/null" },
.cache_path = null,
.deadline_ms = 200,
} }, &carry, std.posix.STDIN_FILENO, null);
defer t.close();
try std.testing.expect(t.link != .quic);
try std.testing.expect(t.link == .pipe);
// No coordinates were ever in play, so no budget may be spent looking
// for them. Generous against a loaded machine, and still an order of
// magnitude below the deadline this path must not touch.
try std.testing.expect(std.time.milliTimestamp() - t0 < 1000);
}
test "handoff: dead coordinates are a fast no, and the pipe is the fallback" {
// A well-formed announce naming 127.0.0.1:1, where nothing listens: the
// ICMP refusal is real, so this dial dies in one loopback round trip
// instead of running `deadline_ms` out. The upper bound proves it — a
// build that swallowed the refusal spends the whole 300 ms.
//
// With no cache here, this cannot witness that a dial happened at all: a
// build ignoring the coordinates lands on the child link just as fast.
// That half is e2e's. Pinned HERE: the fallback DECISION and the speed.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
// Inert, but ours — and this test really needs it: waitReady watches
// fd 0 for the whole QUIC dial.
var stdin = try FakeStdin.install("");
defer stdin.deinit();
const t0 = std.time.milliTimestamp();
var t = try Transport.open(alloc, .{ .hand = .{
.host = "127.0.0.1",
.ssh_argv = &.{ "/bin/sh", "-c", "printf 'endpoint 1 " ++ ("ab" ** 32) ++ "\\n'; cat >/dev/null" },
.cache_path = null,
.deadline_ms = 300,
} }, &carry, std.posix.STDIN_FILENO, null);
defer t.close();
const elapsed = std.time.milliTimestamp() - t0;
try std.testing.expect(t.link != .quic);
try std.testing.expect(t.link == .pipe);
// The refusal was SEEN, not waited out. 150 ms is derived from both ends:
// the failure spends the full 300 ms, and the 2 ms this measures leaves 75x
// of headroom — which it needs, since `elapsed` also covers a shell spawn.
if (elapsed >= 150) std.debug.print(
"refused dial took {d}ms of a 300ms budget: the ICMP refusal was swallowed, not acted on\n",
.{elapsed},
);
try std.testing.expect(elapsed < 150);
}
/// Put `bytes` on fd 0 for the duration of a test and give back a restorer.
/// The CLI passes STDIN_FILENO as the abort fd, so testing that spelling means
/// briefly owning fd 0. Zig runs a file's tests one at a time, so this is safe
/// as long as every caller restores — which `deinit` makes a defer.
const FakeStdin = struct {
saved: std.posix.fd_t,
w: std.posix.fd_t,
fn install(bytes: []const u8) !FakeStdin {
const p = try std.posix.pipe();
errdefer {
std.posix.close(p[0]);
std.posix.close(p[1]);
}
if (bytes.len > 0) _ = try std.posix.write(p[1], bytes);
const saved = try std.posix.dup(std.posix.STDIN_FILENO);
try std.posix.dup2(p[0], std.posix.STDIN_FILENO);
// fd 0 is the surviving copy of the read end.
std.posix.close(p[0]);
return .{ .saved = saved, .w = p[1] };
}
fn deinit(self: *FakeStdin) void {
std.posix.dup2(self.saved, std.posix.STDIN_FILENO) catch {};
std.posix.close(self.saved);
std.posix.close(self.w);
}
};
test "handoff: the announce wait still answers the abort key" {
// An announce that never comes must not cost the user their way out: a
// blocking read would sit in `read(2)` with nothing watching stdin, and on
// the reconnect path `Ctrl-\` is the only way out. RECONNECT spelling on
// purpose. The script exits on its own, so a blocking implementation FAILS
// with a printed expectation rather than hanging the suite.
const alloc = std.testing.allocator;
// The abort byte is already in the pipe when the wait starts, so the
// very first poll has it: no sleeping, no race with the child.
var stdin = try FakeStdin.install(&[_]u8{0x1c});
defer stdin.deinit();
const t0 = std.time.milliTimestamp();
try std.testing.expectError(error.UserAbort, Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", "sleep 2" },
.cache_path = null,
.deadline_ms = 200,
} }, null, std.posix.STDIN_FILENO, null));
// Well inside the script's own 2s, so this passing cannot mean "waited
// for the child to die and called that an abort".
try std.testing.expect(std.time.milliTimestamp() - t0 < 1000);
}
test "handoff: a first attach leaves stdin to ssh while the announce is pending — a typed password must not be swallowed" {
// ssh reads its password prompt from /dev/tty, and a client polling stdin
// during the announce wait steals whole cooked lines — auth fails, and the
// stolen line replays INTO the session via carry. A first attach must leave
// stdin untouched.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("hunter2\n");
defer stdin.deinit();
var t = try Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", "printf 'endpoint none\\n'; cat >/dev/null" },
.cache_path = null,
.deadline_ms = 200,
} }, &carry, std.posix.STDIN_FILENO, null);
t.close();
try std.testing.expectEqual(@as(usize, 0), carry.items.len);
var buf: [16]u8 = undefined;
const got = try std.posix.read(std.posix.STDIN_FILENO, &buf);
try std.testing.expectEqualStrings("hunter2\n", buf[0..got]);
}
test "handoff: abort_fd -1 means no abort channel — fd 0 is never read" {
// The hub's spelling. A Transport opened with abort_fd = -1 must not
// poll or read fd 0, even with the abort byte sitting right there on
// it: a hub dialling targets has no terminal, and a stray 0x1c on its
// stdin aborting a tile's dial would be this exact regression.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install(&[_]u8{0x1c});
defer stdin.deinit();
// Announces `endpoint none`, so the open completes as the ssh pipe —
// the abort byte, were anything watching, would have fired first (the
// sibling test above proves it fires within the same window).
var t = try Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", "printf 'endpoint none\\n'; cat >/dev/null" },
.cache_path = null,
.deadline_ms = 200,
} }, &carry, -1, null);
defer t.close();
try std.testing.expect(t.link == .pipe);
// And the byte is still in the pipe, unread: the carry took nothing.
try std.testing.expectEqualStrings("", carry.items);
}
/// How many times the ssh fake below ran. By NEWLINES, not by size: the
/// fake logs the WORD it was handed, so its lines have two lengths.
fn shimRuns(dir: []const u8, name: []const u8) !u64 {
var buf: [4096]u8 = undefined;
const said = shimSaid(dir, name, &buf) catch return 0;
if (said.len == 0) return 0;
return std.mem.count(u8, said, "\n") + 1;
}
/// One line the shim wrote, newline trimmed.
fn shimSaid(dir: []const u8, name: []const u8, buf: []u8) ![]const u8 {
var path_buf: [512]u8 = undefined;
const path = try std.fmt.bufPrint(&path_buf, "{s}/{s}", .{ dir, name });
const f = try std.fs.cwd().openFile(path, .{});
defer f.close();
const n = try f.readAll(buf);
return std.mem.trimRight(u8, buf[0..n], "\n");
}
fn shimMade(dir: []const u8, name: []const u8) !bool {
var buf: [512]u8 = undefined;
const path = try std.fmt.bufPrint(&buf, "{s}/{s}", .{ dir, name });
std.fs.cwd().access(path, .{}) catch return false;
return true;
}
/// The two remote words `handoff.recipeFor` builds, as the fakes below are
/// handed them. Spelled out rather than imported so that a test which
/// believes the client ran the asking word is reading a literal, not the
/// same expression the product built it from.
const read_word = "mux d endpoint";
const asked_word = "mux d endpoint --start";
/// The fake box: ONE script behind both argvs, so only the WORD differs,
/// as on the wire. `--start` leaves a daemon; without one there is
/// nothing to announce and it exits 1 in silence.
fn boxScript(buf: []u8, dir: []const u8) ![]const u8 {
return std.fmt.bufPrint(buf,
\\echo "$*" >> {[d]s}/runs
\\case "$*" in *--start*) : > {[d]s}/started ;; esac
\\test -e {[d]s}/started || exit 1
\\printf 'endpoint none\n'
\\cat >/dev/null
, .{ .d = dir });
}
/// The dial must not have reached a daemon. NOT `expectError`: it renders the
/// success value with `{any}`, and a live `Transport` holds an allocator vtable
/// that FAULTS on formatting — the runner dies inside the message and names no
/// test. The session it should not have is closed here; it owns a child.
fn expectNoSession(claim: []const u8, r: anytype) !void {
if (r) |t| {
var live = t;
live.close();
std.debug.print("{s}\n", .{claim});
return error.TestUnexpectedResult;
} else |err| {
if (err == error.UnterminatedLine) return;
std.debug.print("{s} — got {s}, want UnterminatedLine\n", .{ claim, @errorName(err) });
return err;
}
}
/// A box that cannot hold a daemon — a full disk, a broken shell, a `mux`
/// too old to know the flag. It records the word and refuses either way.
fn refusingBoxScript(buf: []u8, dir: []const u8) ![]const u8 {
return std.fmt.bufPrint(buf,
\\echo "$*" >> {[d]s}/runs
\\exit 1
, .{ .d = dir });
}
test "openHandoff: a HandoffTarget nobody configured runs the reading word, never the asking one" {
// The default is what a NEW dial path inherits by forgetting the line, and
// nothing fails loudly: the symptom is a daemon on someone else's box. So
// the default is the harmless half. `asked` picks the ARGV, so the oracle
// is the word the fake was handed — what RAN, not what was meant.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("");
defer stdin.deinit();
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try boxScript(&script_buf, tmp.path());
try expectNoSession("a dial nobody configured reached a daemon: it ran the asking word", Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.asked_argv = &.{ "/bin/sh", "-c", script, "sh", asked_word },
.cache_path = null,
.deadline_ms = 200,
} }, &carry, std.posix.STDIN_FILENO, null));
try std.testing.expect(!try shimMade(tmp.path(), "started"));
// The whole log, not a count: one line, and it is the reading word.
var buf: [512]u8 = undefined;
try std.testing.expectEqualStrings(read_word, try shimSaid(tmp.path(), "runs", &buf));
}
test "openHandoff: the dial a user ASKED for runs the asking word ONCE and rides the announce that comes back on it" {
// Three ssh runs became one: the remote ensures the daemon and announces on
// the same stdout, so there is no refusal to read, no exit code to tell from
// ssh's own 255, and no second dial to pay a password prompt for.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("");
defer stdin.deinit();
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try boxScript(&script_buf, tmp.path());
var t = try Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.asked_argv = &.{ "/bin/sh", "-c", script, "sh", asked_word },
.cache_path = null,
.deadline_ms = 200,
.asked = true,
} }, &carry, std.posix.STDIN_FILENO, null);
defer t.close();
// `endpoint none` is a real announce: the remote says ssh IS the
// session, and the run that started the daemon is the run carrying it.
try std.testing.expect(t.link == .pipe);
// Asked of the FILESYSTEM. "Was a daemon started" is a question about
// the world, and the client is not a witness to it.
try std.testing.expect(try shimMade(tmp.path(), "started"));
var buf: [512]u8 = undefined;
try std.testing.expectEqualStrings(asked_word, try shimSaid(tmp.path(), "runs", &buf));
}
test "openHandoff: an asked dial whose box still announces nothing fails after that ONE run" {
// A box that refuses to hold a daemon used to cost the attach an extra
// round trip — the ask bought one start and one retry. It buys neither
// now: there is nothing left for a second run to learn, because the
// run that could have started a daemon already tried.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("");
defer stdin.deinit();
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try refusingBoxScript(&script_buf, tmp.path());
try expectNoSession("a box that announces nothing gave the asked dial a session", Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.asked_argv = &.{ "/bin/sh", "-c", script, "sh", asked_word },
.cache_path = null,
.deadline_ms = 200,
.asked = true,
} }, &carry, std.posix.STDIN_FILENO, null));
// ONE line, and the asking word: a client that kept a retry would show
// two, and a client that fell back to the reading word would show the
// wrong one.
try std.testing.expectEqual(@as(u64, 1), try shimRuns(tmp.path(), "runs"));
var buf: [512]u8 = undefined;
try std.testing.expectEqualStrings(asked_word, try shimSaid(tmp.path(), "runs", &buf));
}
test "openHandoff: a dial nobody asked for, against a box with nothing, reports the failure and ran the bare word only" {
// The wall polls every listed host once a second, and a poll that started a
// daemon gave a listed box one from a READ — undoing a `mux d stop` a second
// after it was typed. The rule is argv's, so the log of what ran IS the proof.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("");
defer stdin.deinit();
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try boxScript(&script_buf, tmp.path());
try expectNoSession("a poll reached a daemon on an empty box: it ran the asking word", Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.asked_argv = &.{ "/bin/sh", "-c", script, "sh", asked_word },
.cache_path = null,
.deadline_ms = 200,
.asked = false,
} }, &carry, std.posix.STDIN_FILENO, null));
try std.testing.expect(!try shimMade(tmp.path(), "started"));
var buf: [512]u8 = undefined;
try std.testing.expectEqualStrings(read_word, try shimSaid(tmp.path(), "runs", &buf));
}
test "openHandoff: an asked target with no asking argv runs the reading word, and starts nothing" {
// `asked_argv` defaults to empty, and nothing in the type stops a caller
// setting `asked` on such a target — an empty argv is not a no-op at the
// exec: the child null-unwraps `argv[0]` and dies, which looks exactly like
// a box with no daemon. This pins the DOC, since `fromRecipe` always fills it.
const alloc = std.testing.allocator;
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var stdin = try FakeStdin.install("");
defer stdin.deinit();
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try boxScript(&script_buf, tmp.path());
try expectNoSession("an asked target with no asking argv did not fall back to the reading word", Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.cache_path = null,
.deadline_ms = 200,
.asked = true,
} }, &carry, std.posix.STDIN_FILENO, null));
// The reading word ran, once, and no daemon came of it: an exec that
// died on a null argv[0] would have left this log EMPTY.
try std.testing.expect(!try shimMade(tmp.path(), "started"));
var buf: [512]u8 = undefined;
try std.testing.expectEqualStrings(read_word, try shimSaid(tmp.path(), "runs", &buf));
}
/// fd 2, captured into a pipe this test owns and restored on `take` —
/// `FakeStdin`'s shape for the other direction. fd 2 and never fd 1: a byte on
/// the runner's stdout wedges `zig build test` silently, at 0 CPU.
const CapturedStderr = struct {
saved: std.posix.fd_t,
r: std.posix.fd_t,
fn install() !CapturedStderr {
const p = try std.posix.pipe();
errdefer {
std.posix.close(p[0]);
std.posix.close(p[1]);
}
const saved = try std.posix.dup(std.posix.STDERR_FILENO);
try std.posix.dup2(p[1], std.posix.STDERR_FILENO);
// fd 2 is the surviving copy of the write end.
std.posix.close(p[1]);
return .{ .saved = saved, .r = p[0] };
}
/// Restore FIRST, then read: while fd 2 still holds a write end the
/// read below would block instead of seeing the end of the capture.
fn take(self: *CapturedStderr, buf: []u8) ![]const u8 {
std.posix.dup2(self.saved, std.posix.STDERR_FILENO) catch {};
std.posix.close(self.saved);
defer std.posix.close(self.r);
return buf[0..try std.posix.read(self.r, buf)];
}
};
test "openHandoff: the handoff ssh's stderr is a pipe, and only `narrate` relays it" {
// A hosts line naming a box that is down used to put ssh's `No route to
// host` onto the wall's alternate screen every poll, because the child's
// stderr was INHERITED. It is a pipe mux reads now, whoever dialled. The
// fake records what KIND of file its stderr is, off `/dev/fd/2`, so
// "piped" is exact — `/dev/fd` because every OS this builds for has it
// and the Linux-only spelling would have to be ported alongside the test.
// Asked with `test -p` rather than by reading a link target: on Linux
// `/dev/fd/2` is a symlink to `pipe:[N]` and on Darwin it is an entry of
// the fdesc filesystem that is not a symlink at all, so `readlink` there
// answers nothing and the check passed on an empty file. `test -p` stats
// the path on both and reports the underlying object's type, which is the
// claim being made.
// BOTH values of `asked`, because the rule is the spawn's.
const alloc = std.testing.allocator;
var stdin = try FakeStdin.install("");
defer stdin.deinit();
for ([_]bool{ false, true }) |narrate| for ([_]bool{ false, true }) |asked| {
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var tmp = try TmpDir.make();
defer tmp.cleanup();
var script_buf: [1024]u8 = undefined;
const script = try std.fmt.bufPrint(&script_buf,
\\if [ -p /dev/fd/2 ]; then echo pipe > {[d]s}/e; else echo "not a pipe" > {[d]s}/e; fi
\\printf 'boom: no route\n' >&2
\\exit 1
, .{ .d = tmp.path() });
var dial: handoff.Dial = .{};
var cap = try CapturedStderr.install();
const opened = Transport.open(alloc, .{ .hand = .{
.host = "fake",
.ssh_argv = &.{ "/bin/sh", "-c", script, "sh", read_word },
.asked_argv = &.{ "/bin/sh", "-c", script, "sh", asked_word },
.cache_path = null,
.deadline_ms = 200,
.asked = asked,
.narrate = narrate,
} }, &carry, std.posix.STDIN_FILENO, &dial);
var relayed_buf: [256]u8 = undefined;
const relayed = try cap.take(&relayed_buf);
try expectNoSession("the stderr fixture's box announced a session it has no daemon for", opened);
var err_buf: [std.fs.max_path_bytes]u8 = undefined;
const on_err = try shimSaid(tmp.path(), "e", &err_buf);
try std.testing.expectEqualStrings("pipe", on_err);
// Kept in every case: the picker row is painted from a dial
// nobody narrated, which is the whole point of keeping it here
// rather than letting the bytes fall out onto a screen.
try std.testing.expectEqualStrings("boom: no route", dial.reason.slice());
// A dial that spawned an ssh knows which one, whatever became of
// it: the refusal a prompt earns is keyed on this pid, and the
// dial it belongs to is the one that just failed.
try std.testing.expect(dial.ssh_pid > 0);
if (narrate) {
try std.testing.expectEqualStrings("boom: no route\n", relayed);
} else {
try std.testing.expectEqualStrings("", relayed);
}
};
}
test "lostMsg: only a --via transport that never connected gets the new wording" {
// The case the message exists for: a command that failed to start. It
// names what happened and guesses no cause — ssh's own stderr passes
// through and has already named the real one (host key, DNS, refused,
// no such binary). The lower layer spoke; this line must not talk over it.
try std.testing.expectEqualStrings(
"mux: transport command failed before a session started",
lostMsg(.{ .via = "ssh box mux d proxy" }, 0),
);
// Same transport, but a session existed — there WAS a connection, and
// saying otherwise would be the new lie in place of the old one.
try std.testing.expectEqualStrings("mux: connection to the daemon lost", lostMsg(.{ .via = "ssh box mux d proxy" }, 7));
// No command to have failed: a socket or quic:// target keeps the
// original wording however early it dies.
try std.testing.expectEqualStrings("mux: connection to the daemon lost", lostMsg(.{ .sock = "/run/muxd.sock" }, 0));
}
// The three tests below pin every line the entry dial can print when the
// open fails — a policy that until now was reachable only by making a real dial
// fail in a real terminal, which is why none of it was pinned at all.
// `openFailure` is pure, so each class is one call with a literal answer.
test "Transport.drainErr: a session whose ssh floods stderr keeps serving" {
// A pipe holds 64k; past that the writer BLOCKS, and ssh blocked on stderr
// is ssh not moving the session's bytes either — nothing on screen says so,
// the tile simply stops. The fixture floods in the FOREGROUND, before it
// serves, so an owner that never drains cannot get past it.
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
var carry: std.ArrayList(u8) = .empty;
defer carry.deinit(alloc);
var t = try Transport.open(alloc, .{ .hand = .{
.host = "flood",
.ssh_argv = &.{
"/bin/sh", "-c",
\\printf 'endpoint none\n'
\\head -c 1048576 /dev/zero >&2
\\cat
,
},
.cache_path = null,
.deadline_ms = 200,
} }, &carry, -1, null);
defer t.close();
try std.testing.expect(t.link == .pipe);
// The whole point of the fd being on the Transport: whoever owns the
// link can find it without knowing how the handoff went.
try std.testing.expect(t.errFd() != null);
// `cat` echoes, so a frame written comes back as itself once the
// flood is out of the way.
const blob = "z" ** 4096;
try t.writeFrame(.input, blob);
const deadline = std.time.milliTimestamp() + 2000;
var echoed = false;
while (!echoed and std.time.milliTimestamp() < deadline) {
var fds: [2]std.posix.pollfd = undefined;
fds[0] = .{ .fd = t.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
var n: usize = 1;
if (t.errFd()) |efd| {
fds[1] = .{ .fd = efd, .events = std.posix.POLL.IN, .revents = 0 };
n = 2;
}
_ = std.posix.poll(fds[0..n], 50) catch break;
if (n == 2 and fds[1].revents != 0) t.drainErr();
if (fds[0].revents == 0) continue;
switch (t.readFrame(alloc) catch break) {
.incomplete => {},
.closed => break,
.frame => |f| {
defer f.deinit(alloc);
if (f.type == .input and std.mem.eql(u8, f.payload, blob)) echoed = true;
},
}
}
// The rule before the failure, since a bare `expect` names nothing:
// this test's whole subject is WHY no frame came back.
if (!echoed) std.debug.print(
"the tile stopped serving: with stderr undrained ssh blocks at the pipe's " ++
"64k and never runs the `cat` that carries the session\n",
.{},
);
try std.testing.expect(echoed);
}
test "openFailure: a quic:// target names the key or the address, and only an abort exits 0" {
var buf: [open_err_len]u8 = undefined;
const q: Target = .{ .quic = .{ .host_port = "box:4433", .key_path = "/etc/mux/key" } };
// The three key-file classes say what is wrong with the file, in the
// daemon's own words — the user's mistake is the same one either end
// catches it at.
try std.testing.expectEqualStrings(
"mux: no such key file: /etc/mux/key\n",
openFailure(&buf, q, error.KeyFileMissing, "").msg,
);
try std.testing.expectEqualStrings(
"mux: /etc/mux/key is readable by group or other; chmod 600 it\n",
openFailure(&buf, q, error.KeyFilePermissive, "").msg,
);
try std.testing.expectEqualStrings(
"mux: /etc/mux/key is not a key: want 32 raw bytes or 64 hex characters\n",
openFailure(&buf, q, error.KeyFileMalformed, "").msg,
);
// Both address failures are one class: a name that will not resolve and
// a string that will not parse are the same fact to the user.
try std.testing.expectEqualStrings(
"mux: cannot resolve quic://box:4433\n",
openFailure(&buf, q, error.MalformedAddress, "").msg,
);
try std.testing.expectEqualStrings(
"mux: cannot resolve quic://box:4433\n",
openFailure(&buf, q, error.UnknownHostName, "").msg,
);
// The handshake line must keep naming both causes: a wrong key and an
// absent daemon are indistinguishable from here, and dropping either
// half would turn an honest ambiguity into a wrong guess.
try std.testing.expectEqualStrings(
"mux: quic://box:4433 did not answer (wrong key, or no mux d --quic there)\n",
openFailure(&buf, q, error.QuicHandshakeFailed, "").msg,
);
// Anything unclassified still reports the error's name rather than
// inventing a cause for it.
try std.testing.expectEqualStrings(
"mux: cannot reach quic://box:4433: ConnectionRefused\n",
openFailure(&buf, q, error.ConnectionRefused, "").msg,
);
// Every one of those is a failure and exits 1.
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, q, error.KeyFileMissing, "").exit);
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, q, error.QuicHandshakeFailed, "").exit);
// The abort is not one of them: the user asked to stop, so the line
// says the attach ended and the status says nothing went wrong.
const ab = openFailure(&buf, q, error.UserAbort, "");
try std.testing.expectEqualStrings("mux: aborted before attaching\n", ab.msg);
try std.testing.expectEqual(@as(u8, 0), ab.exit);
}
test "openFailure: a handoff separates a missing announce from an ssh that never spoke" {
var buf: [open_err_len]u8 = undefined;
const h: Target = .{ .hand = .{
.host = "box",
.ssh_argv = &.{ "ssh", "box", "mux d endpoint" },
.cache_path = null,
} };
// We waited for an announce and did not get one. Both halves of
// announceFailed's reflection are exercised — a parse error and a
// read error — because the classification is what decides the wording.
try std.testing.expectEqualStrings(
"mux: no endpoint announce from box over ssh (AnnounceMissingPrefix)\n",
openFailure(&buf, h, error.AnnounceMissingPrefix, "").msg,
);
try std.testing.expectEqualStrings(
"mux: no endpoint announce from box over ssh (UnterminatedLine)\n",
openFailure(&buf, h, error.UnterminatedLine, "").msg,
);
// Never got that far: the command would not spawn, or the pipe failed.
// The other wording, and the split between them is the point — the
// announce line must not claim we reached the host, and this one must
// not claim we waited for a line we never got to wait for.
try std.testing.expectEqualStrings(
"mux: cannot reach box over ssh: AccessDenied\n",
openFailure(&buf, h, error.AccessDenied, "").msg,
);
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, h, error.AccessDenied, "").exit);
// The abort answers the same as the quic:// arm, for the same keystroke.
const ab = openFailure(&buf, h, error.UserAbort, "");
try std.testing.expectEqualStrings("mux: aborted before attaching\n", ab.msg);
try std.testing.expectEqual(@as(u8, 0), ab.exit);
}
test "openFailure: a reason replaces the error name for a hand target" {
// What the user actually needs is the sentence ssh printed. Before the
// stderr pipe it was not mux's to have — it went straight to the
// terminal, and this line could only say what mux had OBSERVED, which
// is that no announce arrived.
var buf: [open_err_len]u8 = undefined;
const h: Target = .{ .hand = .{ .host = "box", .ssh_argv = &.{"ssh"}, .cache_path = null } };
const said = "ssh: connect to host box port 22: No route to host";
// Both arms of the hand target — the one that read a broken announce
// and the one that could not spawn ssh at all — because a reason is a
// better answer than either error name, and one arm keeping its name
// would be the half nobody notices.
try std.testing.expectEqualStrings(
"mux: box over ssh: ssh: connect to host box port 22: No route to host\n",
openFailure(&buf, h, error.UnterminatedLine, said).msg,
);
try std.testing.expectEqualStrings(
"mux: box over ssh: ssh: connect to host box port 22: No route to host\n",
openFailure(&buf, h, error.AccessDenied, said).msg,
);
// ...and with nothing said, the two old lines stand byte for byte: a
// silent ssh is exactly the case they were written for.
try std.testing.expectEqualStrings(
"mux: no endpoint announce from box over ssh (UnterminatedLine)\n",
openFailure(&buf, h, error.UnterminatedLine, "").msg,
);
try std.testing.expectEqualStrings(
"mux: cannot reach box over ssh: AccessDenied\n",
openFailure(&buf, h, error.AccessDenied, "").msg,
);
// An abort is still not a failure, whatever ssh had been saying.
try std.testing.expectEqual(@as(u8, 0), openFailure(&buf, h, error.UserAbort, said).exit);
// Other transports have no ssh to quote, so the reason is not theirs
// to print even when a caller passes one.
try std.testing.expectEqualStrings(
"mux: cannot connect to /run/muxd.sock\n",
openFailure(&buf, .{ .sock = "/run/muxd.sock" }, error.ConnectionRefused, said).msg,
);
}
test "openFailure: an OpenSSH authentication refusal is visible as terminal" {
var buf: [open_err_len]u8 = undefined;
const h: Target = .{ .hand = .{ .host = "box", .ssh_argv = &.{"ssh"}, .cache_path = null } };
try std.testing.expectEqualStrings(
"mux: authentication refused by box over ssh: Permission denied (publickey).\n",
openFailure(&buf, h, error.UnterminatedLine, "Permission denied (publickey).").msg,
);
}
test "openFailure: --via and --sock say what they know and nothing more" {
var buf: [open_err_len]u8 = undefined;
// One line whatever the error was: the command's own stderr is
// inherited and has already named the cause.
const v: Target = .{ .via = "ssh box mux d proxy" };
try std.testing.expectEqualStrings(
"mux: cannot start --via command: ssh box mux d proxy\n",
openFailure(&buf, v, error.FileNotFound, "").msg,
);
try std.testing.expectEqualStrings(
"mux: cannot start --via command: ssh box mux d proxy\n",
openFailure(&buf, v, error.AccessDenied, "").msg,
);
// No "is the daemon running?" — auto-start checked that moments ago, so the
// path is the whole of what is known and the whole of what is said.
const s: Target = .{ .sock = "/run/muxd.sock" };
try std.testing.expectEqualStrings(
"mux: cannot connect to /run/muxd.sock\n",
openFailure(&buf, s, error.ConnectionRefused, "").msg,
);
// Neither transport can abort — there is no dial to interrupt — so
// both exit 1 for every error, the abort key's included.
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, v, error.FileNotFound, "").exit);
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, s, error.ConnectionRefused, "").exit);
try std.testing.expectEqual(@as(u8, 1), openFailure(&buf, v, error.UserAbort, "").exit);
}
test "openFailure: a message too long for the buffer clips, and still fails" {
// `open_err_len` is chosen rather than derived, so what makes choosing safe
// is this: past the buffer the line CLIPS and everything else holds. A
// future `bufPrint` here would answer an empty message or a crash instead.
const cmd = "x" ** (open_err_len + 808);
var buf: [open_err_len]u8 = undefined;
const f = openFailure(&buf, .{ .via = cmd }, error.FileNotFound, "");
// Full buffer, and really the message's own prefix. The LENGTH goes first
// and the slices below are cut to what came back: an expected slice sized
// from the constant is a comptime bounds error the day this input stops
// overflowing, and that fails the whole module instead of one property.
try std.testing.expectEqual(@as(usize, open_err_len), f.msg.len);
const head = "mux: cannot start --via command: ";
try std.testing.expectEqualStrings(head, f.msg[0..head.len]);
const tail = f.msg[head.len..];
try std.testing.expectEqualStrings(cmd[0..tail.len], tail);
// The newline is written last, so a clip is exactly where it is lost.
// Stated as a test because it is what the user sees: a line that stops
// mid-command with no terminator, not a line that silently vanished.
try std.testing.expectEqual(@as(u8, 'x'), f.msg[f.msg.len - 1]);
// Clipped is still a failure. The exit must not ride on formatting.
try std.testing.expectEqual(@as(u8, 1), f.exit);
}
test "client: the new session's name is the lowest free integer" {
var buf: [proto.session_name_max]u8 = undefined;
try std.testing.expectEqualStrings("0", nextFreeName(&buf, ""));
try std.testing.expectEqualStrings("1", nextFreeName(&buf, "0"));
try std.testing.expectEqualStrings("2", nextFreeName(&buf, "0\n1"));
// A hole is filled before the series is extended, so names stay short
// and a closed session's number comes back.
try std.testing.expectEqualStrings("1", nextFreeName(&buf, "0\n2"));
// Named sessions (`mux --session work`) rule out nothing numeric.
try std.testing.expectEqualStrings("1", nextFreeName(&buf, "0\ndev"));
try std.testing.expectEqualStrings("0", nextFreeName(&buf, "dev\nwork"));
// Slot order is the daemon's, not sorted: the answer must not depend
// on it.
try std.testing.expectEqualStrings("2", nextFreeName(&buf, "1\n0"));
}
test "client: a hostile session list cannot move the free name" {
var buf: [proto.session_name_max]u8 = undefined;
// A `sessions_reply` is bounded only in total, so a buggy or hostile
// daemon can put an over-long line, a line with a space, or a blank line
// in the list. `proto.sessionsIter` drops each; the free name is the same
// one the honest prefix alone would have produced, because a candidate is
// always a decimal string and no line the filter drops can equal one.
const long = "x" ** 1056;
try std.testing.expectEqualStrings("1", nextFreeName(&buf, "0\n" ++ long));
try std.testing.expectEqualStrings("2", nextFreeName(&buf, "0\nhas space\n1"));
try std.testing.expectEqualStrings("1", nextFreeName(&buf, "\n\n0\n\n"));
// And the counting walk shrinking does not cut the search short: the
// honest names still rule out their own numbers.
try std.testing.expectEqualStrings("3", nextFreeName(&buf, "0\n" ++ long ++ "\n1\n2"));
}
test "client: a picked name off the wire is validated before it is copied" {
// The reason the guard exists: `SessionName.of` memcpys into a
// `session_name_max` buffer, so this name is an out-of-bounds write
// rather than a rejected frame.
const too_long = "a" ** (proto.session_name_max + 1);
try std.testing.expectEqual(@as(?SessionName, null), validPick(too_long));
// Spellings the daemon would never accept as a name either, so
// switching to one could only ever fail.
try std.testing.expectEqual(@as(?SessionName, null), validPick("has space"));
try std.testing.expectEqual(@as(?SessionName, null), validPick("has/slash"));
try std.testing.expectEqual(@as(?SessionName, null), validPick(""));
// Nothing to go to is the same answer, and the one the ring already
// gave: stay put.
try std.testing.expectEqual(@as(?SessionName, null), validPick(null));
// A name the daemon could have meant still moves the focus.
const ok = validPick("work").?;
try std.testing.expectEqualStrings("work", ok.slice());
const at_cap = validPick("a" ** proto.session_name_max).?;
try std.testing.expectEqual(@as(usize, proto.session_name_max), at_cap.slice().len);
}
test "client: an unanswered chord expires once, and says so once" {
var p: PendingSwitch = .{};
// Nothing armed is nothing to report, however long we wait.
try std.testing.expect(!p.expired(1_000_000));
p.arm(.end, 1000);
// Inside the window the chord is still waiting for a daemon that may
// yet answer.
try std.testing.expect(!p.expired(1000 + PendingSwitch.wait_ms - 1));
try std.testing.expectEqual(SwitchIntent.end, p.intent);
try std.testing.expect(p.expired(1000 + PendingSwitch.wait_ms));
// Once: the poll loop runs this every 100ms, so a marker per pass would
// be a chord that keeps shouting.
try std.testing.expect(!p.expired(1_000_000));
// An answered chord never expires — `take` spends it.
p.arm(.new, 1000);
try std.testing.expectEqual(SwitchIntent.new, p.take());
try std.testing.expectEqual(SwitchIntent.none, p.take());
try std.testing.expect(!p.expired(1_000_000));
// Nor does one a re-dial dropped: the daemon it asked no longer has
// the question, and the silence is already explained by the
// `[reconnecting]` the user was shown.
p.arm(.new, 1000);
p.clear();
try std.testing.expect(!p.expired(1_000_000));
}
test "client: a target spells itself back as one wall argument per session" {
var buf: [256]u8 = undefined;
// One argv string with the flag inside it — the grammar, not two args.
try std.testing.expectEqualStrings(
"--sock /tmp/m.sock#b",
try wallSpelling(&buf, .{ .sock = "/tmp/m.sock" }, "b"),
);
// The port the user typed rides along untouched, and so does its
// absence.
try std.testing.expectEqualStrings(
"quic://box:8787#0",
try wallSpelling(&buf, .{ .quic = .{ .host_port = "box:8787", .key_path = "/k" } }, "0"),
);
try std.testing.expectEqualStrings(
"quic://box#0",
try wallSpelling(&buf, .{ .quic = .{ .host_port = "box", .key_path = "/k" } }, "0"),
);
try std.testing.expectEqualStrings(
"vm1#work",
try wallSpelling(&buf, .{ .hand = .{
.host = "vm1",
.ssh_argv = &.{ "ssh", "vm1", "mux d endpoint" },
.cache_path = null,
} }, "work"),
);
}
test "client: every spelling this writes, the host grammar reads back the same" {
// Writer/reader identity across the module boundary: a tile's label and the
// sidecar leaf keyed by it are this string, and a drift in either half heals
// a saved layout onto the wrong tile. `refAllDecls` compiles both.
var buf: [256]u8 = undefined;
const cases = .{
.{ Target{ .sock = "/run/user/1000/muxd.sock" }, "0" },
// `user@host`: nothing inside it parses, which is what keeps ssh's
// own config working — and the wall must not start parsing it now.
.{ Target{ .hand = .{ .host = "ubuntu@sandbox-a609d8", .ssh_argv = &.{"x"}, .cache_path = null } }, "build" },
.{ Target{ .hand = .{ .host = "vm1", .ssh_argv = &.{"x"}, .cache_path = null } }, "0" },
// The port rides through untouched, and so does its absence.
.{ Target{ .quic = .{ .host_port = "box:8787", .key_path = "/k" } }, "work" },
.{ Target{ .quic = .{ .host_port = "box", .key_path = "/k" } }, "0" },
};
inline for (cases) |c| {
const spelling = try wallSpelling(&buf, c[0], c[1]);
// `#NAME` is last and a session name may not hold a '#', so the
// split is the last one — the same rule the label bar reads by.
const hash = std.mem.lastIndexOfScalar(u8, spelling, '#').?;
try std.testing.expectEqualStrings(c[1], spelling[hash + 1 ..]);
const spec = try hosts.parse(spelling[0..hash]);
switch (c[0]) {
.sock => |path| try std.testing.expectEqualStrings(path, spec.sock),
.hand => |h| try std.testing.expectEqualStrings(h.host, spec.host),
.quic => |q| try std.testing.expectEqualStrings(q.host_port, spec.quic),
.via => unreachable,
}
}
}
test "client: the default session spells as #0, not as an empty name" {
var buf: [256]u8 = undefined;
// Bare `mux` attaches under the empty WIRE name (older-daemon compat),
// and a spelling has to be one the user could type back — so the tile
// everyone gains on first use is `--sock <default>#0`, per the spec.
try std.testing.expectEqualStrings(
"--sock /run/muxd.sock#0",
try wallSpelling(&buf, .{ .sock = "/run/muxd.sock" }, proto.resolveName("")),
);
}
test "client: a --via target has no wall spelling at all" {
var buf: [256]u8 = undefined;
// Not a formatting failure to be worked around: the wall grammar has
// no form for "an arbitrary command's stdio", so the honest answer is
// that this transport cannot be walled.
try std.testing.expectError(
error.NoSpelling,
wallSpelling(&buf, .{ .via = "ssh h mux d proxy" }, "0"),
);
}
test "client: a wall spelling that does not fit fails rather than truncates" {
// The cap is what every caller sizes its buffer from, so a spelling
// that fits `spellingCap` must never be the one that clips.
const target: Target = .{ .sock = "/tmp/m.sock" };
const cap = spellingCap(target);
const buf = try std.testing.allocator.alloc(u8, cap);
defer std.testing.allocator.free(buf);
const longest = "n" ** proto.session_name_max;
try std.testing.expectEqualStrings(
"--sock /tmp/m.sock#" ++ longest,
try wallSpelling(buf, target, longest),
);
try std.testing.expectError(
error.NoSpace,
wallSpelling(buf[0 .. cap - 1], target, longest),
);
}
// 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 {
_ = discovery;
_ = resolver;
_ = open_wait;
std.testing.refAllDeclsRecursive(@This());
_ = @import("client_core.zig");
_ = @import("hosts.zig");
_ = @import("handoff.zig");
_ = @import("layout.zig");
_ = @import("interrupt.zig");
_ = @import("askpass.zig");
}
/// A daemon stand-in for the birth tests: accepts once, records every frame
/// type it is sent, and answers the first one with `reply`.
const BirthFake = struct {
listener: std.net.Server,
got: [4]proto.MsgType = undefined,
n: usize = 0,
attach_cols: u16 = 0,
attach_rows: u16 = 0,
attach_name: [proto.session_name_max]u8 = undefined,
attach_name_len: usize = 0,
reply: proto.MsgType,
fn serve(self: *BirthFake) void {
const alloc = std.testing.allocator;
const conn = self.listener.accept() catch return;
defer conn.stream.close();
while (self.n < self.got.len) {
const f = (proto.readFrame(alloc, conn.stream.handle) catch return) orelse return;
defer f.deinit(alloc);
self.got[self.n] = f.type;
self.n += 1;
if (f.type == .attach) {
const req = proto.decodeAttach(f.payload) catch return;
self.attach_cols = req.cols;
self.attach_rows = req.rows;
self.attach_name_len = req.name.len;
@memcpy(self.attach_name[0..req.name.len], req.name);
// The daemon's real order on the admitted path:
// `Server.sendPtyModeTo` goes out BEFORE the snapshot, so
// a birth that treated the first frame as the answer
// would read a mode byte as a refusal.
if (self.reply == .snapshot)
proto.writeFrame(conn.stream.handle, .pty_mode, &.{0}) catch return;
proto.writeFrame(conn.stream.handle, self.reply, &.{0}) catch return;
}
}
}
};
test "birthSession: the birth is the snapshot, not the first frame the daemon sends" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/birth.sock", .{tmp.path()});
defer alloc.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = BirthFake{ .listener = try addr.listen(.{}), .reply = .snapshot };
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, BirthFake.serve, .{&fake});
try birthSession(alloc, .{ .sock = sp }, "wghost", 80, 24);
th.join();
// The whole point of the side connection: it claims a size, so the
// daemon has a grid to build the session on. A 0x0 attach — what every
// browser tile sends — is join-only and would create nothing.
try std.testing.expectEqual(@as(u16, 80), fake.attach_cols);
try std.testing.expectEqual(@as(u16, 24), fake.attach_rows);
try std.testing.expectEqualStrings("wghost", fake.attach_name[0..fake.attach_name_len]);
// The `pty_mode` the fake sent first was skipped rather than taken
// for an answer; only e2e exercises that ordering otherwise.
// ...and it leaves: the session outlives this connection, and a
// lingering client would hold a slot the browser tile needs.
try std.testing.expectEqual(@as(usize, 2), fake.n);
try std.testing.expectEqual(proto.MsgType.attach, fake.got[0]);
try std.testing.expectEqual(proto.MsgType.detach, fake.got[1]);
}
test "birthSession: an exit_status before any snapshot is Refused, and nothing is detached" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/refuse.sock", .{tmp.path()});
defer alloc.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = BirthFake{ .listener = try addr.listen(.{}), .reply = .exit_status };
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, BirthFake.serve, .{&fake});
// A full session table answers exactly this, and the caller must be
// able to tell it from a birth so it stops trying rather than looping.
try std.testing.expectError(error.Refused, birthSession(alloc, .{ .sock = sp }, "wghost", 80, 24));
th.join();
try std.testing.expectEqual(@as(usize, 1), fake.n);
}
/// A daemon stand-in for the end test. One connection per `endSession`, the
/// same shape `BirthFake` stands in for a birth with — and for the same
/// reason: the `client` module has no import edge to `daemon`, so a REAL
/// daemon cannot be stood up in this test binary. The verdicts are scripted
/// here because they are the DAEMON's to make; that half is pinned against a
/// real daemon in `server_test_session.zig` ("end_req with another client
/// attached is refused with the count"). What this file owns, and what this
/// fake therefore records, is the request `endSession` puts on the wire and
/// the outcome it makes of the reply.
const EndFake = struct {
listener: std.net.Server,
/// One scripted answer per connection, in order.
replies: []const proto.EndReply,
/// What each request carried, so the client's half is assertable.
force: [4]bool = @splat(false),
names: [4][proto.session_name_max]u8 = undefined,
name_lens: [4]usize = @splat(0),
n: usize = 0,
fn serve(self: *EndFake) void {
const alloc = std.testing.allocator;
while (self.n < self.replies.len) {
const conn = self.listener.accept() catch return;
defer conn.stream.close();
const f = (proto.readFrame(alloc, conn.stream.handle) catch return) orelse return;
defer f.deinit(alloc);
if (f.type != .end_req or f.payload.len < proto.end_req_len) return;
const at = self.n;
self.force[at] = f.payload[0] != 0;
const name = f.payload[proto.end_req_len..];
@memcpy(self.names[at][0..name.len], name);
self.name_lens[at] = name.len;
self.n += 1;
const r = self.replies[at];
// The daemon's own encoder, so the bytes this test decodes are
// the bytes a daemon writes rather than a second spelling.
var buf: [proto.end_reply_max_len]u8 = undefined;
proto.writeFrame(
conn.stream.handle,
.end_reply,
proto.encodeEndReply(&buf, r.accepted, r.others, r.reason),
) catch return;
}
}
fn nameOf(self: *const EndFake, i: usize) []const u8 {
return proto.resolveName(self.names[i][0..self.name_lens[i]]);
}
};
test "endSession: a held session refuses with the count, force ends it, and an unknown name is refused with the daemon's reason" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/end.sock", .{tmp.path()});
defer alloc.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = EndFake{
.listener = try addr.listen(.{}),
.replies = &.{
.{ .accepted = false, .others = 1, .reason = proto.end_reason.others_attached },
.{ .accepted = true, .others = 0, .reason = proto.end_reason.accepted },
.{ .accepted = false, .others = 0, .reason = proto.end_reason.no_session },
},
};
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, EndFake.serve, .{&fake});
const first = try endSession(alloc, .{ .sock = sp }, "0", false);
try std.testing.expect(!first.accepted);
try std.testing.expectEqual(@as(u8, 1), first.others);
// The picker's notice is built from this, so a refusal that lost its
// reason would read as an end that happened.
try std.testing.expectEqualStrings(proto.end_reason.others_attached, first.reason());
const forced = try endSession(alloc, .{ .sock = sp }, "0", true);
try std.testing.expect(forced.accepted);
try std.testing.expectEqualStrings("", forced.reason());
const missing = try endSession(alloc, .{ .sock = sp }, "nope", false);
try std.testing.expect(!missing.accepted);
try std.testing.expectEqualStrings(proto.end_reason.no_session, missing.reason());
th.join();
// Three connections, one question each: the picker ends a session with
// no pane on this wall, so there is no pump's link to ride.
try std.testing.expectEqual(@as(usize, 3), fake.n);
// `force` is the SECOND press and nothing else — the client never
// decides to force, it only remembers being told to ask twice.
try std.testing.expectEqual([3]bool{ false, true, false }, fake.force[0..3].*);
// The default session rides the wire as an empty tail; `endSession` must
// spell it that way or a daemon reads a session called "0" it has not got.
try std.testing.expectEqual(@as(usize, 0), fake.name_lens[0]);
try std.testing.expectEqualStrings("0", fake.nameOf(0));
try std.testing.expectEqualStrings("nope", fake.nameOf(2));
}
/// A daemon stand-in for the list test: accepts once and answers the first
/// `sessions_req` with a fixed reply.
const ListFake = struct {
listener: std.net.Server,
reply: []const u8,
fn serve(self: *ListFake) void {
const alloc = std.testing.allocator;
const conn = self.listener.accept() catch return;
defer conn.stream.close();
const f = (proto.readFrame(alloc, conn.stream.handle) catch return) orelse return;
defer f.deinit(alloc);
if (f.type != .sessions_req) return;
proto.writeFrame(conn.stream.handle, .sessions_reply, self.reply) catch return;
}
};
test "listSessions: answers with the daemon's whole list, and a socket nobody listens on is a transport error" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/list.sock", .{tmp.path()});
defer alloc.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = ListFake{ .listener = try addr.listen(.{}), .reply = "p\nq\n" };
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, ListFake.serve, .{&fake});
var out: [proto.sessions_reply_max]u8 = undefined;
var link: std.meta.Tag(Link) = .quic;
const list = try listSessions(alloc, .{ .sock = sp }, &out, 2000, &link, null);
th.join();
// Which link answered is what the wall's poll interval is chosen from,
// so a socket must say `fd` and not merely "not an error".
try std.testing.expectEqual(std.meta.Tag(Link).fd, link);
// Every name, not the first: the poll diffs the whole list against the
// wall, so a reply read short would vanish tiles the daemon still has.
try std.testing.expectEqualStrings("p\nq\n", list);
const dead = try std.fmt.allocPrint(alloc, "{s}/nobody.sock", .{tmp.path()});
defer alloc.free(dead);
try std.testing.expectError(error.Transport, listSessions(alloc, .{ .sock = dead }, &out, 200, null, null));
}
test "listSessions: an allocation failure is not a host that is down" {
// `mux hosts` prints `[unreachable]` for anything this returns
// `error.Transport` for, so collapsing OOM into it blames a box that is
// up and answering for a fault on this machine.
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const sp = try std.fmt.allocPrint(alloc, "{s}/oom.sock", .{tmp.path()});
defer alloc.free(sp);
const addr = try std.net.Address.initUnix(sp);
var fake = ListFake{ .listener = try addr.listen(.{}), .reply = "p\nq\n" };
defer fake.listener.deinit();
const th = try std.Thread.spawn(.{}, ListFake.serve, .{&fake});
var out: [proto.sessions_reply_max]u8 = undefined;
try std.testing.expectError(
error.OutOfMemory,
listSessions(std.testing.failing_allocator, .{ .sock = sp }, &out, 2000, null, null),
);
th.join();
}
test "listSessions: a poll that failed still reports the login it paid for" {
const alloc = std.testing.allocator;
var out: [proto.sessions_reply_max]u8 = undefined;
// An ssh that dies without an announce: the login is spent, the poll has
// nothing. Left at `.fd` the wall would ask again in a second, forever —
// one sshd auth line per second per dead host, which is the whole reason
// `pollDelayMs` stretches a `.pipe` answer tenfold.
var link: std.meta.Tag(Link) = .quic;
try std.testing.expectError(error.Transport, listSessions(alloc, .{ .hand = .{
.host = "nowhere",
.ssh_argv = &.{ "/bin/sh", "-c", "exit 255" },
.cache_path = null,
.asked = false,
} }, &out, 200, &link, null));
try std.testing.expectEqual(std.meta.Tag(Link).pipe, link);
// A `--sock` open that fails cost a connect(2) and nothing else, so the
// backoff must not follow it: the local daemon a user just stopped is
// back a second later, not ten.
link = .quic;
try std.testing.expectError(error.Transport, listSessions(alloc, .{ .sock = "/nonexistent/mux.sock" }, &out, 200, &link, null));
try std.testing.expectEqual(std.meta.Tag(Link).fd, link);
}
test "Target.fromSpec: asked is the caller's word, never a default" {
// The COMPILE-TIME half cannot be a runtime assertion: `fromSpec` and
// `fromRecipe` take `asked` with no default, so a call that omits it does
// not build. What IS runtime-checkable is that the word is carried rather
// than dropped and re-defaulted between here and the recipe.
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena.deinit();
const alloc = arena.allocator();
// Both values, off one spelling: a fromSpec that hard-coded either one
// would pass a test that only ever asked for the other.
for ([_]bool{ true, false }) |asked| {
const t = try Target.fromSpec(alloc, .{ .host = "box" }, null, 30_000, asked);
try std.testing.expectEqual(asked, t.hand.asked);
try std.testing.expectEqual(@as(u32, 30_000), t.hand.idle_ms);
// The asking line is the recipe's, and it is what `asked` selects.
try std.testing.expect(t.hand.asked_argv.len > 0);
}
}
test "Target.fromSpec: the target owns every slice, so a scratch spelling may be reused" {
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena.deinit();
const alloc = arena.allocator();
// The hub parses a POST body and the wall parses a filter's buffer;
// both are overwritten before the dial the target describes.
var scratch: [15]u8 = "box ".*;
const t = try Target.fromSpec(alloc, .{ .host = scratch[0..3] }, null, 30_000, false);
@memset(&scratch, 'z');
try std.testing.expectEqualStrings("box", t.hand.host);
var kbuf: [7]u8 = "/k ".*;
const q = try Target.fromSpec(alloc, .{ .quic = "h:1" }, kbuf[0..2], 30_000, false);
@memset(&kbuf, 'z');
try std.testing.expectEqualStrings("/k", q.quic.key_path);
try std.testing.expectEqualStrings("h:1", q.quic.host_port);
}
test "Target.fromSpec: a quic spelling with no key frees the path it refused" {
// `std.testing.allocator` IS the assertion: `resolveKeyPath`'s `.missing`
// arm hands back an allocated path, and the refusal that does not keep it
// must free it. The live callers pass arenas, so nothing else would say so.
var tmp = try TmpDir.make();
defer tmp.cleanup();
// Pointed at an empty directory rather than the developer's own config:
// whether ~/.config/mux/key exists on the machine running the suite
// decides which arm this takes, and a test that grades a different arm
// per box grades nothing.
const libc = @cImport({
@cInclude("stdlib.h");
});
const alloc = std.testing.allocator;
const prior = std.posix.getenv("XDG_CONFIG_HOME");
const prior_z = if (prior) |p| try alloc.dupeZ(u8, p) else null;
defer if (prior_z) |p| alloc.free(p);
const cfg = try alloc.dupeZ(u8, tmp.path());
defer alloc.free(cfg);
_ = libc.setenv("XDG_CONFIG_HOME", cfg.ptr, 1);
defer if (prior_z) |p| {
_ = libc.setenv("XDG_CONFIG_HOME", p.ptr, 1);
} else {
_ = libc.unsetenv("XDG_CONFIG_HOME");
};
try std.testing.expectError(
error.MissingKey,
Target.fromSpec(alloc, .{ .quic = "h:1" }, null, 30_000, false),
);
}