src/server/server_test_attach.zig
Ref: Size: 72.8 KiB History
const std = @import("std");
const grid_mod = @import("term").grid;
const Grid = @import("term").grid.Grid;
const proto = @import("term").protocol;
const TmpDir = @import("testtmp").TmpDir;
const h = @import("server_test_harness.zig");
const dial = h.dial;
const srv_mod = @import("server.zig");
const Server = srv_mod.Server;
const installSignalHandlers = srv_mod.installSignalHandlers;
const max_clients = srv_mod.max_clients;
const applyFrame = h.applyFrame;
const awaitFrame = h.awaitFrame;
const connectedPair = h.connectedPair;
const firstStateFrame = h.firstStateFrame;
const awaitReplicaText = h.awaitReplicaText;
const ClientSlot = h.ClientSlot;
const Lead = h.Lead;
const pumpUntil = h.pumpUntil;
const serverThread = h.serverThread;
test "forward-role hello leaves terminal admission and the session untouched" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "forward-role", .{ .shell = "/bin/sh" });
defer td.deinit();
const session = td.srv.sessions.table[0].?.eng;
const epoch = td.srv.sessions.table[0].?.epoch;
const peer = try dial.dial(td.sock_path);
defer peer.close();
try proto.writeFrame(peer.handle, .forward_hello, &proto.encodeForwardHello());
const ready = (try awaitFrame(alloc, &td.srv, peer.handle, .forward_ready, 400)) orelse return error.NoForwardReady;
defer ready.deinit(alloc);
try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready.payload));
try std.testing.expectEqual(@as(?usize, 1), td.srv.forwards.freePeer());
try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
try std.testing.expectEqual(@as(usize, 1), srv_mod.countLive(td.srv.sessions.table));
try std.testing.expect(td.srv.sessions.table[0].?.eng == session);
try std.testing.expectEqual(epoch, td.srv.sessions.table[0].?.epoch);
}
/// A sink that replays every frame it is handed into a replica and lets the
/// wait run on. For the waits whose answer is one particular reply while the
/// session keeps broadcasting underneath it.
const ApplyEach = struct {
alloc: std.mem.Allocator,
replica: *Grid,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
try applyFrame(self.alloc, self.replica, frame);
}
fn sink(self: *ApplyEach) h.link.Sink {
return .{ .ctx = self, .on = ApplyEach.on };
}
};
/// Wait until a state frame reports at least `want` rows of history. Both
/// frame kinds carry the count, and only the attach is a snapshot, so a
/// wait for one kind alone would miss it on the other.
fn awaitHistoryRows(
alloc: std.mem.Allocator,
fd: std.posix.fd_t,
want: u32,
budget_ms: i64,
) !bool {
const Count = struct {
want: u32,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
const rows: u32 = switch (frame.type) {
.snapshot => (try proto.readSnapshotPrefix(frame.payload)).history_rows,
.delta => (try proto.readDeltaHeader(frame.payload)).history_rows,
else => return,
};
if (rows >= self.want) return error.HistoryReached;
}
};
var count: Count = .{ .want = want };
_ = h.awaitFrameOnSink(alloc, fd, h.never_from_daemon, budget_ms, .{
.ctx = &count,
.on = Count.on,
}) catch |err| switch (err) {
error.HistoryReached => return true,
else => return err,
};
return false;
}
test "Server: survives a client that dies without detaching; next attach works" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "kill", .{ .shell = "/bin/sh" });
defer td.deinit();
installSignalHandlers(); // includes SIGPIPE ignore
try td.threaded();
// Client 1 attaches, provokes output, then vanishes without detach.
const a = try dial.dialAttach(td.sock_path, 80, 24);
try proto.writeFrame(a.handle, .input, "echo pre-kill\n");
std.Thread.sleep(300 * std.time.ns_per_ms);
a.close(); // abrupt: no detach frame
// Force more output so the daemon writes into the dead socket.
std.Thread.sleep(300 * std.time.ns_per_ms);
// Daemon must still be serving: a fresh attach gets a snapshot.
const b = try dial.dialAttach(td.sock_path, 80, 24);
defer b.close();
var got_snapshot = false;
if (try h.awaitFrameOn(alloc, b.handle, .snapshot, 5000)) |frame| {
frame.deinit(alloc);
got_snapshot = true;
}
try std.testing.expect(got_snapshot);
}
test "Server: serves scrollback chunks on request" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "sb", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
try proto.writeFrame(c.handle, .input, "seq 1 100\n");
// Wait until an update reports enough history, then fetch the oldest
// page. Only the attach is a snapshot now; the rest are deltas, whose
// header carries the same history count.
try std.testing.expect(try awaitHistoryRows(alloc, c.handle, 50, 10_000));
try proto.writeFrame(c.handle, .fetch_scrollback, &proto.encodeScrollbackReq(0, 24));
var chunk: ?[]u8 = null;
defer if (chunk) |ch| alloc.free(ch);
if (try h.awaitFrameOn(alloc, c.handle, .scrollback_chunk, 5000)) |frame| {
chunk = frame.payload; // ownership taken
}
try std.testing.expect(chunk != null);
try std.testing.expect(chunk.?.len > 6);
const req_echo = try proto.decodeScrollbackReq(chunk.?[0..6]);
try std.testing.expectEqual(@as(u32, 0), req_echo.start);
// The oldest page contains the first line the shell printed.
try std.testing.expect(std.mem.indexOf(u8, chunk.?[6..], "seq 1 100") != null);
}
test "Server: a full client table refuses the next attach instead of displacing anyone" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "full", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
// Fill every client slot. Each attach is confirmed before the next
// connects, both to keep observer slots free (attach vacates one) and
// so the refusal below is unambiguously about client slots.
var streams: [max_clients]std.net.Stream = undefined;
var opened: usize = 0;
defer for (streams[0..opened]) |s| s.close();
while (opened < max_clients) : (opened += 1) {
streams[opened] = try dial.dialAttach(td.sock_path, 80, 24);
const fd = streams[opened].handle;
const first = try firstStateFrame(alloc, fd, 10_000);
try std.testing.expect(first != null);
}
// The attach past `max_clients` is refused; nobody already attached
// is evicted.
const extra = try dial.dialAttach(td.sock_path, 80, 24);
defer extra.close();
var refused = false;
if (try h.awaitFrameOn(alloc, extra.handle, .exit_status, 5000)) |frame| {
defer frame.deinit(alloc);
try std.testing.expect(frame.payload.len == 1 and frame.payload[0] == 1);
refused = true;
}
try std.testing.expect(refused);
// The client that attached first is still attached and still fed.
try proto.writeFrame(streams[0].handle, .input, "echo still-here\n");
var replica = try Grid.init(alloc, 80, 24);
defer replica.deinit();
try std.testing.expect(try awaitReplicaText(alloc, streams[0].handle, 10_000, .{
.replica = replica,
.needle = "still-here",
}));
}
test "Server: two clients converge on one session" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "two", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
const b = try dial.dialAttach(td.sock_path, 80, 24);
defer b.close();
// Input through A must reach both replicas.
try proto.writeFrame(a.handle, .input, "echo both-see-this\n");
var replica_a = try Grid.init(alloc, 80, 24);
defer replica_a.deinit();
var replica_b = try Grid.init(alloc, 80, 24);
defer replica_b.deinit();
for ([_]struct { fd: std.posix.fd_t, rep: *Grid }{
.{ .fd = a.handle, .rep = replica_a },
.{ .fd = b.handle, .rep = replica_b },
}) |side| {
try std.testing.expect(try awaitReplicaText(alloc, side.fd, 10_000, .{
.replica = side.rep,
.needle = "both-see-this",
}));
}
// Cell-level convergence: both replicas show what the daemon shows. The
// daemon's own dump is asked for as PLAIN and compared through
// `h.trimRowTails`, because trailing spaces are the one formatting
// difference between the two dumps and the VT formatter that fed the old
// wire trimmed them too.
try proto.writeFrame(a.handle, .debug_dump, &.{0});
var daemon_plain: ?[]u8 = null;
defer if (daemon_plain) |d| alloc.free(d);
var feed_a: ApplyEach = .{ .alloc = alloc, .replica = replica_a };
if (try h.awaitFrameOnSink(alloc, a.handle, .dump_reply, 5000, feed_a.sink())) |frame| {
daemon_plain = try h.trimRowTails(alloc, frame.payload);
frame.deinit(alloc);
}
try std.testing.expect(daemon_plain != null);
const va = try replica_a.dumpPlain(alloc);
defer alloc.free(va);
try std.testing.expectEqualStrings(daemon_plain.?, va);
// B saw the same broadcasts but may still have some in its socket
// buffer: drain until it agrees with the dump A already fetched.
const Converge = struct {
alloc: std.mem.Allocator,
replica: *Grid,
want: []const u8,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
try applyFrame(self.alloc, self.replica, frame);
const plain = try self.replica.dumpPlain(self.alloc);
defer self.alloc.free(plain);
if (std.mem.eql(u8, self.want, plain)) return error.Converged;
}
};
var conv: Converge = .{ .alloc = alloc, .replica = replica_b, .want = daemon_plain.? };
_ = h.awaitFrameOnSink(alloc, b.handle, h.never_from_daemon, 5000, .{
.ctx = &conv,
.on = Converge.on,
}) catch |err| switch (err) {
error.Converged => {},
else => return err,
};
// Dumped again rather than kept from the sink: the sink BORROWS its
// frame, and the comparison it made is the one this reproduces.
const vb = try replica_b.dumpPlain(alloc);
defer alloc.free(vb);
try std.testing.expectEqualStrings(daemon_plain.?, vb);
}
test "Server: a same-size join snapshots the joiner only" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "join", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
var replica_a = try Grid.init(alloc, 80, 24);
defer replica_a.deinit();
// Drain A's own join snapshot and everything the shell's startup
// produces, so that anything arriving after B joins is B's doing.
//
// Still its own poll loop, and has to be: the condition is that the
// socket went QUIET, and `Link.awaitFrame` does not report that — a
// poll that timed out inside it is indistinguishable from the deadline
// running out, which is the one fact this loop needs per turn.
var a_snapshots: usize = 0;
var quiet_ms: u64 = 0;
var deadline_ms: u64 = 10_000;
while (deadline_ms > 0 and (quiet_ms < 500 or a_snapshots == 0)) {
var pfd = [_]std.posix.pollfd{
.{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = try std.posix.poll(&pfd, 100);
deadline_ms -|= 100;
if (ready == 0) {
quiet_ms += 100;
continue;
}
quiet_ms = 0;
const frame = (try proto.readFrame(alloc, a.handle)) orelse break;
defer frame.deinit(alloc);
if (frame.type == .snapshot) a_snapshots += 1;
try applyFrame(alloc, replica_a, frame);
}
try std.testing.expectEqual(@as(usize, 1), a_snapshots);
// B joins at the same size: nothing about A's grid changed, so A must
// not be repainted. B, which has nothing, must be.
const b = try dial.dialAttach(td.sock_path, 80, 24);
defer b.close();
var b_snapshot = false;
if (try h.awaitFrameOn(alloc, b.handle, .snapshot, 5000)) |frame| {
frame.deinit(alloc);
b_snapshot = true;
}
try std.testing.expect(b_snapshot);
// A stays live across the join, by delta: the rebuild B triggered
// bumps the seq A will see next, which A neither notices nor needs.
try proto.writeFrame(a.handle, .input, "echo join-unicast\n");
try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
.replica = replica_a,
.needle = "join-unicast",
.reject = &.{.snapshot},
}));
// ...and B, the joiner, sees the same input.
var replica_b = try Grid.init(alloc, 80, 24);
defer replica_b.deinit();
try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{
.replica = replica_b,
.needle = "join-unicast",
}));
}
test "Server: latest attacher's size wins; earlier client is resnapshotted at the new size" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "latest", .{ .shell = "/bin/sh" });
defer td.deinit();
var stop = std.atomic.Value(bool).init(false);
const th = try std.Thread.spawn(.{}, serverThread, .{ &td.srv, &stop });
// Stopped and joined explicitly below so the final assertion can read
// server state without racing the daemon thread.
var joined = false;
defer if (!joined) {
stop.store(true, .release);
th.join();
};
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
// A is attached once it has been answered; only then can B's attach be
// the *later* event this test is about.
var a_attached = false;
if (try h.awaitFrameOn(alloc, a.handle, .snapshot, 10_000)) |frame| {
defer frame.deinit(alloc);
const p = try proto.readSnapshotPrefix(frame.payload);
try std.testing.expectEqual(@as(u16, 80), p.cols);
a_attached = true;
}
try std.testing.expect(a_attached);
const b = try dial.dialAttach(td.sock_path, 100, 30);
defer b.close();
// The grid follows the newest attacher, and A is told about it.
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 5000));
stop.store(true, .release);
th.join();
joined = true;
try std.testing.expectEqual(@as(u16, 100), @as(u16, @intCast(td.srv.sessions.table[0].?.eng.term.cols)));
try std.testing.expectEqual(@as(u16, 30), @as(u16, @intCast(td.srv.sessions.table[0].?.eng.term.rows)));
}
/// Lenient about what precedes it: the join snapshot and the shell's deltas
/// share the stream.
fn awaitSnapshotSize(
alloc: std.mem.Allocator,
fd: std.posix.fd_t,
cols: u16,
rows: u16,
timeout_ms: u64,
) !bool {
const Sized = struct {
cols: u16,
rows: u16,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
if (frame.type != .snapshot) return;
const self: *@This() = @ptrCast(@alignCast(ctx.?));
const p = try proto.readSnapshotPrefix(frame.payload);
if (p.cols == self.cols and p.rows == self.rows) return error.SizeArrived;
}
};
var sized: Sized = .{ .cols = cols, .rows = rows };
// Every frame has to reach the sink: a snapshot at some OTHER size is
// exactly what this is lenient about, and `want = .snapshot` would
// return the first one and call it the answer.
_ = h.awaitFrameOnSink(alloc, fd, h.never_from_daemon, @intCast(timeout_ms), .{
.ctx = &sized,
.on = Sized.on,
}) catch |err| switch (err) {
error.SizeArrived => return true,
else => return err,
};
return false;
}
test "Server: typing claims the grid for the typist (latest-wins on input)" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "typing", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
// A joins first and, being the only client, sets the grid: attach-wins
// is unchanged by any of this.
const a = try dial.dialAttach(td.sock_path, 100, 30);
defer a.close();
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
// B joins at a different size and takes the grid, which A is told about.
const b = try dial.dialAttach(td.sock_path, 80, 24);
defer b.close();
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 80, 24, 10_000));
// Quiet on both sides, so what follows is unambiguously the typing's
// doing and not a frame still in flight from the join.
_ = try drainHeld(alloc, a.handle, 10_000);
_ = try drainHeld(alloc, b.handle, 10_000);
// The point of the test: A types without resizing or re-attaching, and
// the grid moves back to A's size for everyone.
try proto.writeFrame(a.handle, .input, "echo from-a\n");
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 100, 30, 10_000));
_ = try drainHeld(alloc, a.handle, 10_000);
_ = try drainHeld(alloc, b.handle, 10_000);
// ...and it switches back when B types, so this is "most recent", not
// "first" or "largest".
try proto.writeFrame(b.handle, .input, "echo from-b\n");
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
try std.testing.expect(try awaitSnapshotSize(alloc, b.handle, 80, 24, 10_000));
_ = try drainHeld(alloc, a.handle, 10_000);
_ = try drainHeld(alloc, b.handle, 10_000);
// A second keystroke from the same client finds the grid already its
// size and must claim nothing: a snapshot here would mean every
// keystroke of an ordinary session pays for a full repaint.
try proto.writeFrame(b.handle, .input, "echo again-from-b\n");
const first = try firstStateFrame(alloc, b.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.delta, first.?.type);
}
test "Server: a size the grid refuses never becomes a claim" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "refused", .{ .shell = "/bin/sh" });
defer td.deinit();
var stop = std.atomic.Value(bool).init(false);
const th = try std.Thread.spawn(.{}, serverThread, .{ &td.srv, &stop });
// Stopped and joined explicitly below so the last assertion can read
// server state without racing the daemon thread.
var joined = false;
defer if (!joined) {
stop.store(true, .release);
th.join();
};
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 80, 24, 10_000));
_ = try drainHeld(alloc, a.handle, 10_000);
// D asks for a size applySize will not have (1x1 trips engine asserts).
// It still joins and is still served — it is only its *size* that is
// refused — but the grid must not move, and A must not be repainted for
// a resize that never happened.
const d = try dial.dialAttach(td.sock_path, 1, 1);
defer d.close();
try std.testing.expect(try awaitSnapshotSize(alloc, d.handle, 80, 24, 10_000));
// D types. Its slot holds no accepted size, so this claims nothing: A
// sees the output as deltas and no snapshot at all. That the marker
// arrives at all is the other half — a refused size must not cost the
// client its keystrokes.
try proto.writeFrame(d.handle, .input, "echo typed-by-refused\n");
{
// The replica starts blank on purpose — a delta carries every row it
// changed, so the row the marker lands on arrives whole, and a
// snapshot is the failure rather than the way the text gets here.
var seen = try Grid.init(alloc, 80, 24);
defer seen.deinit();
try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
.replica = seen,
.needle = "typed-by-refused",
.reject = &.{.snapshot},
}));
}
// The reviewer's repro: a real client moves the grid, and then D types
// again. If D's slot had been stamped with the grid it found at attach
// (80x24), this keystroke would drag everyone back to a size neither
// client asked for.
try proto.writeFrame(a.handle, .resize, &proto.encodeSize(100, 30));
try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 10_000));
_ = try drainHeld(alloc, a.handle, 10_000);
_ = try drainHeld(alloc, d.handle, 10_000);
try proto.writeFrame(d.handle, .input, "echo typed-again\n");
{
// Blank again, at the size the grid has moved to; a snapshot here
// would mean D's keystroke claimed a size it was refused.
var seen = try Grid.init(alloc, 100, 30);
defer seen.deinit();
try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
.replica = seen,
.needle = "typed-again",
.reject = &.{.snapshot},
}));
}
stop.store(true, .release);
th.join();
joined = true;
// The grid is where the only client with an accepted size put it.
try std.testing.expectEqual(@as(u16, 100), @as(u16, @intCast(td.srv.sessions.table[0].?.eng.term.cols)));
try std.testing.expectEqual(@as(u16, 30), @as(u16, @intCast(td.srv.sessions.table[0].?.eng.term.rows)));
}
test "Server: scrollback fetch is per-client and independent" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "percli", .{
.shell = "/bin/sh",
// Pin the prompt width so the command's screen-space columns do
// not depend on which implementation /bin/sh names on the host.
.extra_env = &.{.{ .key = "PS1", .value = ">>" }},
});
defer td.deinit();
try td.threaded();
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
const b = try dial.dialAttach(td.sock_path, 100, 30);
defer b.close();
// B attached LAST, and this waits for the daemon to have processed each
// attach in the order they were dialled — a state frame only exists once
// `seatClient` has run for that fd. The daemon is on its own thread here,
// so without the waits A's attach could be seated after B's and the grid
// would be at A's 80x24, which is the opposite of what the selections
// below are about. The frames are ASKED FOR, not discarded: a null here
// is a seat that never happened, and swallowing it would leave the wait
// in place while it proved nothing.
_ = (try h.firstStateFrame(alloc, a.handle, 10_000)) orelse return error.NoStateFrameForA;
_ = (try h.firstStateFrame(alloc, b.handle, 10_000)) orelse return error.NoStateFrameForB;
// Typed by B; both clients see the history it produces. Every output row
// carries the SAME marker, and the selections below read a row well down
// the list, so no assertion here depends on where the shell's own echo of
// the command landed or on what it looked like. That was not a spare
// precaution: macOS's /bin/sh is bash, whose readline redraws the input
// line whenever SIGWINCH arrives — carriage return, erase to end of line,
// prompt again — and the resize from the second attach arrives while that
// line is being echoed. Measured on macOS 26, the request that reads
// ">>seq 1 100" on Linux read "seq 1 100" there, and moving the target one
// row down landed on the echo instead of past it. Which column a
// REDRAWN input line starts in is the shell's business; a row the shell
// printed is the same on both.
try proto.writeFrame(b.handle, .input, "seq 1 100 | sed 's/.*/xxSELECTEDxx/'\n");
for ([_]std.posix.fd_t{ a.handle, b.handle }) |fd| {
try std.testing.expect(try awaitHistoryRows(alloc, fd, 50, 10_000));
}
// Selection is another client-local read, but unlike a scrollback page
// it is decoded against the authoritative grid. B attached last at
// 100x30, while A still advertises 80x24: asking from A must neither
// claim A's size nor disclose the reply to B.
const exact = proto.encodeSelectionReq(.{
.id = 77,
.anchor = .{ .row = 10, .col = 2 },
.active = .{ .row = 10, .col = 9 },
});
const invalid = proto.encodeSelectionReq(.{
.id = 78,
.anchor = .{ .row = std.math.maxInt(u32), .col = 0 },
.active = .{ .row = std.math.maxInt(u32), .col = 0 },
});
const one = proto.encodeSelectionReq(.{
.id = 79,
.anchor = .{ .row = 10, .col = 0 },
.active = .{ .row = 10, .col = 0 },
});
const sentinel = proto.encodeSelectionReq(.{
.id = 80,
.anchor = .{ .row = 10, .col = 2 },
.active = .{ .row = 10, .col = 9 },
});
try proto.writeFrame(a.handle, .selection_req, &exact);
try proto.writeFrame(a.handle, .selection_req, &invalid);
try proto.writeFrame(a.handle, .selection_req, &.{ 1, 2, 3 });
try proto.writeFrame(a.handle, .selection_req, &one);
try proto.writeFrame(a.handle, .selection_req, &sentinel);
const Replies = struct {
n: usize = 0,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
if (frame.type != .selection_reply) return;
const self: *@This() = @ptrCast(@alignCast(ctx.?));
const reply = try proto.decodeSelectionReply(frame.payload);
if (reply.id == 0) return;
switch (self.n) {
0 => {
try std.testing.expectEqual(@as(u32, 77), reply.id);
try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
try std.testing.expectEqualStrings("SELECTED", reply.text);
},
1 => {
try std.testing.expectEqual(@as(u32, 78), reply.id);
try std.testing.expectEqual(proto.SelectionStatus.invalid, reply.status);
try std.testing.expectEqual(@as(usize, 0), reply.text.len);
},
2 => {
try std.testing.expectEqual(@as(u32, 79), reply.id);
try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
try std.testing.expectEqualStrings("x", reply.text);
},
3 => {
// Reaching this request proves the malformed frame before
// id 79 was processed. Exactly four replies, in order,
// proves it generated none of its own without a timeout-only
// assertion.
try std.testing.expectEqual(@as(u32, 80), reply.id);
try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
try std.testing.expectEqualStrings("SELECTED", reply.text);
},
else => unreachable,
}
self.n += 1;
if (self.n == 4) return error.AllRepliesSeen;
}
};
var replies: Replies = .{};
_ = h.awaitFrameOnSink(alloc, a.handle, h.never_from_daemon, 5000, .{
.ctx = &replies,
.on = Replies.on,
}) catch |err| switch (err) {
error.AllRepliesSeen => {},
else => return err,
};
try std.testing.expectEqual(@as(usize, 4), replies.n);
try proto.writeFrame(a.handle, .status_req, "");
const NoSelection = struct {
fn on(_: ?*anyopaque, frame: proto.Frame) anyerror!void {
// A fifth reply to four requests would mean the malformed frame
// generated one of its own.
if (frame.type == .selection_reply and (try proto.decodeSelectionReply(frame.payload)).id != 0) return error.LateSelectionReply;
}
};
var status_reply: ?proto.StatusReply = null;
if (try h.awaitFrameOnSink(alloc, a.handle, .status_reply, 5000, .{ .on = NoSelection.on })) |frame| {
defer frame.deinit(alloc);
status_reply = try proto.decodeStatusReply(frame.payload);
}
try std.testing.expect(status_reply != null);
try std.testing.expectEqual(@as(u16, 100), status_reply.?.cols);
try std.testing.expectEqual(@as(u16, 30), status_reply.?.rows);
// A pages back through history...
try proto.writeFrame(a.handle, .fetch_scrollback, &proto.encodeScrollbackReq(0, 24));
var chunk: ?[]u8 = null;
defer if (chunk) |ch| alloc.free(ch);
if (try h.awaitFrameOn(alloc, a.handle, .scrollback_chunk, 5000)) |frame| {
chunk = frame.payload; // ownership taken
}
try std.testing.expect(chunk != null);
// Decoded rather than searched: a chunk is CellRows now, and a byte
// search would pass on a run header that happened to spell the text.
const chunk_count = std.mem.readInt(u16, chunk.?[4..6], .little);
const hist_rows = try grid_mod.decodeRows(alloc, chunk.?[6..], chunk_count, 100);
defer grid_mod.freeRows(alloc, hist_rows);
var seen_seq_line = false;
for (hist_rows) |*r| {
var line: std.ArrayList(u8) = .empty;
defer line.deinit(alloc);
for (r.cells) |c| {
if (c.text_len == 0) try line.append(alloc, ' ') else try line.appendSlice(alloc, r.textOf(c));
}
if (std.mem.indexOf(u8, line.items, "seq 1 100") != null) seen_seq_line = true;
}
try std.testing.expect(seen_seq_line);
// ...while B keeps streaming live updates, undisturbed: B's fetch was
// never asked for, so B must never see a scrollback_chunk.
try proto.writeFrame(b.handle, .input, "echo b-still-live\n");
var replica_b = try Grid.init(alloc, 100, 30);
defer replica_b.deinit();
// The replica starts empty, so only a snapshot or the deltas after it
// can produce the echoed text — and neither of A's answers may appear.
try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{
.replica = replica_b,
.needle = "b-still-live",
.reject = &.{ .scrollback_chunk, .selection_reply },
}));
}
test "Server: replica rebuilt from snapshots matches the authoritative grid" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "m2", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const stream = try dial.dialAttach(td.sock_path, 100, 30);
defer stream.close();
const fd = stream.handle;
var replica = try Grid.init(alloc, 100, 30);
defer replica.deinit();
try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n");
// Consume the attach snapshot and the deltas that follow it until the
// replica shows the command output.
try std.testing.expect(try awaitReplicaText(alloc, fd, 10_000, .{
.replica = replica,
.needle = "fidelity-ok",
}));
// Compare replica against the authoritative daemon grid. Plain, through
// `h.trimRowTails`: trailing spaces are the one formatting difference
// between a daemon's dump and a client's, and the old VT wire trimmed
// them as well.
try proto.writeFrame(fd, .debug_dump, &.{0});
var daemon_plain: ?[]u8 = null;
defer if (daemon_plain) |d| alloc.free(d);
// Late updates may arrive before the reply; the sink applies them so
// the replica stays current with what the dump will show.
var feed: ApplyEach = .{ .alloc = alloc, .replica = replica };
if (try h.awaitFrameOnSink(alloc, fd, .dump_reply, 10_000, feed.sink())) |frame| {
daemon_plain = try h.trimRowTails(alloc, frame.payload);
frame.deinit(alloc);
}
try std.testing.expect(daemon_plain != null);
const replica_plain = try replica.dumpPlain(alloc);
defer alloc.free(replica_plain);
try std.testing.expectEqualStrings(daemon_plain.?, replica_plain);
}
test "Server: typing produces deltas, not snapshots; stats track both" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "delta", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// The attach is answered with a full snapshot carrying the tracker seq.
var attach_seq: u64 = 0;
if (try h.awaitFrameOn(alloc, c.handle, .snapshot, 10_000)) |frame| {
defer frame.deinit(alloc);
attach_seq = (try proto.readSnapshotPrefix(frame.payload)).seq;
}
try std.testing.expect(attach_seq > 0);
// From here on every update must be a delta: a snapshot means the
// tracker isn't diffing, which is the regression this test guards.
try proto.writeFrame(c.handle, .input, "x");
const NoSnapshot = struct {
fn on(_: ?*anyopaque, frame: proto.Frame) anyerror!void {
if (frame.type != .snapshot) return;
std.debug.print(
"post-attach snapshot ({d} bytes) where a delta was due\n",
.{frame.payload.len},
);
return error.SnapshotInsteadOfDelta;
}
};
var hdr: ?proto.DeltaHeader = null;
if (try h.awaitFrameOnSink(alloc, c.handle, .delta, 10_000, .{ .on = NoSnapshot.on })) |frame| {
defer frame.deinit(alloc);
hdr = try proto.readDeltaHeader(frame.payload);
}
try std.testing.expect(hdr != null);
try std.testing.expect(hdr.?.seq > attach_seq);
// Echoing one character touches one row; a prompt redraw may add a
// couple more. Anything beyond that means the differ isn't working.
if (hdr.?.row_count > 3) {
std.debug.print("echo delta carried {d} rows\n", .{hdr.?.row_count});
}
try std.testing.expect(hdr.?.row_count <= 5);
try proto.writeFrame(c.handle, .stats_req, "");
var stats_text: ?[]u8 = null;
defer if (stats_text) |s| alloc.free(s);
if (try h.awaitFrameOn(alloc, c.handle, .stats_reply, 5000)) |frame| {
stats_text = frame.payload; // ownership taken
}
try std.testing.expect(stats_text != null);
try std.testing.expect(std.mem.indexOf(u8, stats_text.?, "deltas=") != null);
try std.testing.expect(std.mem.indexOf(u8, stats_text.?, "snapshots=") != null);
}
test "Server: reattach needs a recent have_seq AND this daemon's epoch to get a delta" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "resync", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
// Session 1: attach cold, produce output, note the newest seq we hold
// and the session epoch the daemon stamps into its snapshots.
var last_seq: u64 = 0;
var epoch: u64 = 0;
{
const c1 = try dial.dialAttach(td.sock_path, 80, 24);
defer c1.close();
// A resize is a discontinuity, so this pins reset_seq above 1 no
// matter how the shell's first output raced the attach — session 3
// below needs have_seq=1 to be reliably stale.
try proto.writeFrame(c1.handle, .resize, &proto.encodeSize(80, 24));
try proto.writeFrame(c1.handle, .input, "echo before-detach\n");
const held = try drainHeld(alloc, c1.handle, 10_000);
last_seq = held.seq;
epoch = held.epoch;
try std.testing.expect(last_seq > 0);
// 0 is reserved for "I hold no epoch", so a live daemon never has it.
try std.testing.expect(epoch != 0);
try dial.detach(c1.handle);
}
// Session 2: the seq is real but the epoch belongs to some other daemon
// instance. Those seqs describe a different history; honouring one would
// paint this session's screen with another's rows. Snapshot.
{
const c2 = try dial.dial(td.sock_path);
defer c2.close();
try proto.writeFrame(c2.handle, .attach, &proto.encodeAttach(80, 24, last_seq, epoch ^ 1));
const first = try firstStateFrame(alloc, c2.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
// That snapshot is itself a discontinuity (rebuild bumps seq and
// reset_seq), so the sessions below must resync off it, not off the
// seq session 1 learned.
last_seq = first.?.seq;
try std.testing.expectEqual(epoch, first.?.epoch);
try dial.detach(c2.handle);
}
// Session 3: a real seq with have_epoch = 0 — a pre-epoch client, or one
// hoping 0 means "any". It means "none", and none is never serviceable.
{
const c3 = try dial.dial(td.sock_path);
defer c3.close();
try proto.writeFrame(c3.handle, .attach, &proto.encodeAttach(80, 24, last_seq, 0));
const first = try firstStateFrame(alloc, c3.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
last_seq = first.?.seq;
try dial.detach(c3.handle);
}
// Session 4: right seq, right epoch — we really are up to date, so the
// daemon owes us a delta (possibly empty), never a full repaint.
{
const c4 = try dial.dial(td.sock_path);
defer c4.close();
try proto.writeFrame(c4.handle, .attach, &proto.encodeAttach(80, 24, last_seq, epoch));
const first = try firstStateFrame(alloc, c4.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.delta, first.?.type);
try dial.detach(c4.handle);
}
// Session 5: right epoch, but have_seq predates the last discontinuity,
// so no delta can reconstruct our state — full snapshot.
{
const c5 = try dial.dial(td.sock_path);
defer c5.close();
try proto.writeFrame(c5.handle, .attach, &proto.encodeAttach(80, 24, 1, epoch));
const first = try firstStateFrame(alloc, c5.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
}
}
test "Server: a daemon restart invalidates have_seq even with the old epoch presented" {
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}/restart.sock", .{dir_path});
defer alloc.free(sock_path);
// Daemon A: what a client is holding at the moment the daemon dies
// under it — a seq, and the epoch that seq is counted in.
var held_a: Held = undefined;
{
var srv_a = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
defer srv_a.deinit();
var stop_a = std.atomic.Value(bool).init(false);
const th_a = try std.Thread.spawn(.{}, serverThread, .{ &srv_a, &stop_a });
defer th_a.join();
defer stop_a.store(true, .release);
const c = try dial.dialAttach(sock_path, 80, 24);
defer c.close();
try proto.writeFrame(c.handle, .input, "echo before-restart\n");
held_a = try drainHeld(alloc, c.handle, 10_000);
try std.testing.expect(held_a.seq > 0);
try std.testing.expect(held_a.epoch != 0);
try dial.detach(c.handle);
}
// A is gone: thread joined, deinit ran, socket unlinked.
// Daemon B on the same path — the restart. Its seqs start over, so A's
// numbers now name rows B has never produced.
var srv_b = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
defer srv_b.deinit();
var stop_b = std.atomic.Value(bool).init(false);
const th_b = try std.Thread.spawn(.{}, serverThread, .{ &srv_b, &stop_b });
defer th_b.join();
defer stop_b.store(true, .release);
// Learn what B itself is up to, so the check below can vary the epoch
// alone. Two sequential inits are observed to differ here: that is the
// distinctness this whole mechanism rests on, and without pinning it a
// constant epoch would satisfy every other test in the file.
var held_b: Held = undefined;
{
const c = try dial.dialAttach(sock_path, 80, 24);
defer c.close();
held_b = try drainHeld(alloc, c.handle, 10_000);
try std.testing.expect(held_b.seq > 0);
try std.testing.expect(held_b.epoch != 0);
try std.testing.expect(held_b.epoch != held_a.epoch);
try dial.detach(c.handle);
}
// A seq B issued moments ago, quoted back with A's epoch. Everything
// about this attach is serviceable except the epoch, so a delta here
// would mean the epoch is not being checked at all.
{
const c = try dial.dial(sock_path);
defer c.close();
try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, held_b.seq, held_a.epoch));
const first = try firstStateFrame(alloc, c.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
// Read off the wire, not off srv_b: the server thread is live.
try std.testing.expectEqual(held_b.epoch, first.?.epoch);
try dial.detach(c.handle);
}
// And the literal restart case: the exact pair the pre-restart client
// held. B has no way to reconstruct that state, so: snapshot, stamped
// with B's own epoch, which the client adopts in place of A's.
{
const c = try dial.dial(sock_path);
defer c.close();
try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, held_a.seq, held_a.epoch));
const first = try firstStateFrame(alloc, c.handle, 10_000);
try std.testing.expect(first != null);
try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
try std.testing.expect(first.?.epoch != held_a.epoch);
}
}
test "Server: injected bytes reach frame handling, split anywhere" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "inject", .{ .shell = "/bin/sh" });
defer td.deinit();
// A real socket pair so the slot behaves like any other client: the
// point of this test is the INBOUND route, not the outbound one.
const c = try connectedPair();
defer std.posix.close(c.peer);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
// A resize frame, built exactly as a client would send it, then fed in
// as if it had arrived over some transport that is not this fd. The
// grid moving is proof the bytes reached the same handler a socket read
// would have reached.
var frame: std.ArrayList(u8) = .empty;
defer frame.deinit(alloc);
try proto.appendFrame(&frame, alloc, .resize, &proto.encodeSize(100, 30));
// One byte at a time: every possible split lands mid-header and
// mid-payload, which is the shape stream transports actually deliver
// and the reason the slot carries an inbound buffer at all.
for (frame.items) |b| {
try std.testing.expectEqual(@as(u16, 80), td.srv.colsNow(0)); // nothing yet
td.srv.pushInbound(0, &.{b});
}
try std.testing.expectEqual(@as(u16, 100), td.srv.colsNow(0));
try std.testing.expectEqual(@as(u16, 30), td.srv.rowsNow(0));
// Fully consumed: a frame that was handled must not sit in the buffer.
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.inbound.items.len);
// Two frames in one delivery, the other shape a stream produces.
var pair: std.ArrayList(u8) = .empty;
defer pair.deinit(alloc);
try proto.appendFrame(&pair, alloc, .resize, &proto.encodeSize(90, 20));
try proto.appendFrame(&pair, alloc, .resize, &proto.encodeSize(120, 40));
td.srv.pushInbound(0, pair.items);
try std.testing.expectEqual(@as(u16, 120), td.srv.colsNow(0));
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.inbound.items.len);
}
/// The rescue thread completes the frame after a deadline, so a pump that
/// blocked on the half frame is UNBLOCKED and the test fails with a message
/// instead of hanging the runner (which prints nothing for a wedged step).
const HalfFrameRescue = struct {
peer: std.posix.fd_t,
rest: []const u8,
fired: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
stop: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
fn run(self: *HalfFrameRescue) void {
var waited: usize = 0;
while (!self.stop.load(.acquire) and waited < 2000) : (waited += 10) {
std.Thread.sleep(10 * std.time.ns_per_ms);
}
if (self.stop.load(.acquire)) return;
self.fired.store(true, .release);
proto.writeAllFd(self.peer, self.rest) catch {};
}
};
test "Server: a client that sends half a frame does not stall the pump" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "half", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
var frame: std.ArrayList(u8) = .empty;
defer frame.deinit(alloc);
try proto.appendFrame(&frame, alloc, .resize, &proto.encodeSize(100, 30));
// Three of the five header bytes: the kind and half the length. This
// is the shape a peer that stalled mid-write leaves on the wire.
try proto.writeAllFd(c.peer, frame.items[0..3]);
var rescue: HalfFrameRescue = .{ .peer = c.peer, .rest = frame.items[3..] };
const th = try std.Thread.spawn(.{}, HalfFrameRescue.run, .{&rescue});
defer th.join();
// One pump with a 20ms poll must come back on its own — the readable
// socket is serviced, the partial frame is held, and the loop returns.
try td.srv.pumpOnce(20);
rescue.stop.store(true, .release);
if (rescue.fired.load(.acquire)) {
std.debug.print("pumpOnce blocked on a half frame; the rescue thread had to complete it\n", .{});
return error.PumpBlockedOnHalfFrame;
}
try std.testing.expectEqual(@as(u16, 80), td.srv.colsNow(0)); // nothing applied yet
try std.testing.expect(td.srv.clients[0] != null); // and nobody was dropped
// The rest arrives; the frame completes on the next pump.
try proto.writeAllFd(c.peer, frame.items[3..]);
try td.srv.pumpOnce(20);
try std.testing.expectEqual(@as(u16, 100), td.srv.colsNow(0));
try std.testing.expectEqual(@as(u16, 30), td.srv.rowsNow(0));
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.inbound.items.len);
}
/// What a client would be holding if it detached now: the newest seq it has
/// seen and the epoch stamped on the last snapshot (0 if only deltas came).
const Held = struct { seq: u64, epoch: u64 };
/// Test helper: read an attached connection until it goes quiet, then report
/// what it holds. "Quiet" rather than a frame count because a shell's
/// startup output arrives as an unpredictable number of frames.
///
/// One of the two poll loops left in this file, for the reason the other one
/// carries: `Link.awaitFrame` hides whether a turn ended in a timeout or in
/// a frame, and silence is exactly what this measures.
fn drainHeld(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !Held {
var held: Held = .{ .seq = 0, .epoch = 0 };
var quiet_ms: u64 = 0;
var deadline_ms = timeout_ms;
while (deadline_ms > 0 and quiet_ms < 500) {
var pfd = [_]std.posix.pollfd{
.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = try std.posix.poll(&pfd, 100);
deadline_ms -|= 100;
if (ready == 0) {
quiet_ms += 100;
continue;
}
quiet_ms = 0;
const frame = (try proto.readFrame(alloc, fd)) orelse break;
defer frame.deinit(alloc);
switch (frame.type) {
.snapshot => {
const p = try proto.readSnapshotPrefix(frame.payload);
held.seq = p.seq;
held.epoch = p.epoch;
},
.delta => held.seq = (try proto.readDeltaHeader(frame.payload)).seq,
else => {},
}
}
return held;
}
test "Server: a session is created at the attacher's size" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "bigsize", .{ .shell = "/bin/cat" });
defer td.deinit();
const c = try dial.dialAttachNamed(td.sock_path, 100, 30, "big");
defer c.close();
const f = (try awaitFrame(alloc, &td.srv, c.handle, .snapshot, 400)) orelse
return error.NoSnapshot;
defer f.deinit(alloc);
const p = try proto.readSnapshotPrefix(f.payload);
try std.testing.expectEqual(@as(u16, 100), p.cols);
try std.testing.expectEqual(@as(u16, 30), p.rows);
const si = td.srv.sessions.find("big") orelse return error.NotCreated;
try std.testing.expectEqual(@as(u16, 100), td.srv.colsNow(si));
try std.testing.expectEqual(@as(u16, 30), td.srv.rowsNow(si));
// The default session was not dragged to the new attacher's size.
try std.testing.expectEqual(@as(u16, 80), td.srv.colsNow(0));
}
test "Server: a 0x0 attach joins but never creates" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "zerosize", .{ .shell = "/bin/cat" });
defer td.deinit();
// mux a send attaches at 0x0 — it makes no size claim. Against a name
// that does not exist, that must be a refusal, never a shell spawned
// at a size nobody has.
const c1 = try dial.dialAttachNamed(td.sock_path, 0, 0, "b");
defer c1.close();
const f1 = (try awaitFrame(alloc, &td.srv, c1.handle, .exit_status, 400)) orelse
return error.NoRefusal;
defer f1.deinit(alloc);
try std.testing.expect(f1.payload.len == 1 and f1.payload[0] == 1);
try std.testing.expect(td.srv.sessions.table[1] == null);
// Create it properly, at a real size...
const c2 = try dial.dialAttachNamed(td.sock_path, 80, 24, "b");
defer c2.close();
const f2 = (try awaitFrame(alloc, &td.srv, c2.handle, .snapshot, 400)) orelse
return error.NoSnapshotOnCreate;
f2.deinit(alloc);
const si_b = td.srv.sessions.find("b") orelse return error.NotCreated;
// ...and the very same 0x0 attach now joins it.
const c3 = try dial.dialAttachNamed(td.sock_path, 0, 0, "b");
defer c3.close();
const f3 = (try awaitFrame(alloc, &td.srv, c3.handle, .snapshot, 400)) orelse
return error.ZeroSizeJoinRefused;
f3.deinit(alloc);
// Joined, not spawned: both clients point at the one "b".
var joined: usize = 0;
for (td.srv.clients) |slot| {
const cs = slot orelse continue;
if ((cs.session orelse continue) == si_b) joined += 1;
}
try std.testing.expectEqual(@as(usize, 2), joined);
}
test "Server: a 1x1 attach joins but never creates" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "tinysize", .{ .shell = "/bin/cat" });
defer td.deinit();
// Unlike 0x0, nothing spells passivity as 1x1 — which is exactly why
// this case needs its own pin. 0x0 is a contract and gets refused by
// one; 1x1 is a real terminal, one column wide, and the daemon has to
// stay unclaimable by it on the size threshold alone. e2e drives that
// client for real, on a genuine 1x1 pty.
//
// Refused for creation exactly as 0x0 is: applySize will not move a
// grid to 1x1, so a session created at 1x1 could never be resized by
// the client that caused it — a shell nobody can use, spawned by
// attaching with a name nobody had created yet.
const c1 = try dial.dialAttachNamed(td.sock_path, 1, 1, "tile");
defer c1.close();
const f1 = (try awaitFrame(alloc, &td.srv, c1.handle, .exit_status, 400)) orelse
return error.NoRefusal;
defer f1.deinit(alloc);
try std.testing.expect(f1.payload.len == 1 and f1.payload[0] == 1);
try std.testing.expect(td.srv.sessions.find("tile") == null);
// Created at a real size by someone who has one...
const c2 = try dial.dialAttachNamed(td.sock_path, 80, 24, "tile");
defer c2.close();
const f2 = (try awaitFrame(alloc, &td.srv, c2.handle, .snapshot, 400)) orelse
return error.NoSnapshotOnCreate;
f2.deinit(alloc);
const si = td.srv.sessions.find("tile") orelse return error.NotCreated;
// ...and the same 1x1 attach now JOINS it, leaving the grid alone.
// Refusing the creation must not cost the client its view: a terminal
// too small to spawn a session is still allowed to watch one, and may
// never be the reason one exists or the reason one resizes.
const c3 = try dial.dialAttachNamed(td.sock_path, 1, 1, "tile");
defer c3.close();
const f3 = (try awaitFrame(alloc, &td.srv, c3.handle, .snapshot, 400)) orelse
return error.TinyJoinRefused;
f3.deinit(alloc);
try std.testing.expectEqual(@as(u16, 80), td.srv.colsNow(si));
try std.testing.expectEqual(@as(u16, 24), td.srv.rowsNow(si));
}
// ---------------------------------------------------------------------------
// Activity order and the agent offer: what a slot must carry before an agent
// connection can be routed back to a client. Both are asserted here on the
// slot rather than through the routing that will read them, so an ordering
// bug and a routing bug cannot wear each other's clothes.
// ---------------------------------------------------------------------------
test "Server: attaching seats a client in activity order, and typing or resizing reorders it" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "activity");
defer td.deinit();
// /bin/cat never prints unprompted, so every bump below is one this test
// caused rather than a prompt redraw's.
try td.start(.{ .shell = "/bin/cat" });
const ca = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
defer ca.close();
try std.testing.expect(try pumpUntil(&td.srv, 2000, ClientSlot{ .srv = &td.srv, .n = 0 }, ClientSlot.seated));
// Seated one at a time so the slot indices below are the attach order:
// freeClientSlot hands out the lowest free slot.
const cb = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
defer cb.close();
try std.testing.expect(try pumpUntil(&td.srv, 2000, ClientSlot{ .srv = &td.srv, .n = 1 }, ClientSlot.seated));
const a_attached = (td.srv.clients[0] orelse return error.ClientANeverSeated).activity;
const b_attached = (td.srv.clients[1] orelse return error.ClientBNeverSeated).activity;
// Strict, not >=: the clock is a counter, so two attaches can never tie.
try std.testing.expect(b_attached > a_attached);
// Typing is activity too, and it takes the lead back.
try proto.writeFrame(ca.handle, .input, "x");
try std.testing.expect(try pumpUntil(&td.srv, 2000, Lead{ .srv = &td.srv, .ahead = 0, .behind = 1 }, Lead.taken));
// Resizing is the third activity verb, and B dragging its window back is
// enough to take the lead without B typing a character.
try proto.writeFrame(cb.handle, .resize, &proto.encodeSize(100, 30));
try std.testing.expect(try pumpUntil(&td.srv, 2000, Lead{ .srv = &td.srv, .ahead = 1, .behind = 0 }, Lead.taken));
}
test "Server: a slot promoted before it attached enters the activity order at its own attach" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "quicorder", .{ .shell = "/bin/cat" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
// A client slot with no session yet, which is how a QUIC client starts
// life: quicOnOpen seats it when the handshake completes, so its FIRST
// attach reaches handleFrame rather than serviceObserver. The promotion
// path's bump cannot cover this one, and the rest of the suite attaches
// over sockets — so without this test, deleting handleFrame's bump stays
// green (the hazard the reattach test above records hitting already).
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon } };
try std.testing.expectEqual(@as(u64, 0), td.srv.clients[0].?.activity);
var frame: std.ArrayList(u8) = .empty;
defer frame.deinit(alloc);
try proto.appendFrame(&frame, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));
// pushInbound dispatches the frame inline, so no pump is needed to see
// the arm's effect.
td.srv.pushInbound(0, frame.items);
// Off 0 is the claim: 0 means seated but never attached, which is
// precisely the slot an activity ranking must not pick.
const first = td.srv.clients[0].?.activity;
try std.testing.expect(first > 0);
// The same arm serves a reconnect over a live connection, and that is
// activity again rather than a no-op.
td.srv.pushInbound(0, frame.items);
try std.testing.expect(td.srv.clients[0].?.activity > first);
}
test "Server: an observer that sends one byte does not stall the pump, and finishes its frame later" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "obs", .{ .shell = "/bin/sh" });
defer td.deinit();
// Through the real listener, so the accept path seats it as the daemon
// would seat `mux d stats`.
const obs = try dial.dial(td.sock_path);
defer obs.close();
try td.srv.pumpOnce(20); // accept
try std.testing.expect(td.srv.observers[0] != null);
var frame: std.ArrayList(u8) = .empty;
defer frame.deinit(alloc);
try proto.appendFrame(&frame, alloc, .stats_req, "");
try proto.writeAllFd(obs.handle, frame.items[0..1]);
var rescue: HalfFrameRescue = .{ .peer = obs.handle, .rest = frame.items[1..] };
const th = try std.Thread.spawn(.{}, HalfFrameRescue.run, .{&rescue});
defer th.join();
try td.srv.pumpOnce(20);
rescue.stop.store(true, .release);
if (rescue.fired.load(.acquire)) {
std.debug.print("pumpOnce blocked on a one-byte observer; the rescue thread had to complete it\n", .{});
return error.PumpBlockedOnHalfFrame;
}
try std.testing.expect(td.srv.observers[0] != null); // held, not dropped
// A second observer is served while the first is still mid-frame: the
// daemon is answering everyone, which is the whole claim.
const obs2 = try dial.dial(td.sock_path);
defer obs2.close();
try proto.writeFrame(obs2.handle, .stats_req, "");
const r2 = (try awaitFrame(alloc, &td.srv, obs2.handle, .stats_reply, 100)) orelse
return error.NoStatsReplyWhileAnotherObserverStalls;
r2.deinit(alloc);
try proto.writeAllFd(obs.handle, frame.items[1..]);
const r1 = (try awaitFrame(alloc, &td.srv, obs.handle, .stats_reply, 100)) orelse
return error.NoStatsReplyAfterCompletion;
r1.deinit(alloc);
}
test "Server: stalled observers are dropped on the idle deadline, and the table is free again" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "idle", .{ .shell = "/bin/sh" });
defer td.deinit();
// Every slot, not one: the failure this pins is a table full of peers
// that each wrote one byte and went quiet, so the next `mux` is closed
// at accept with nothing said.
var peers: [srv_mod.max_observers]std.net.Stream = undefined;
for (&peers) |*p| {
p.* = try dial.dial(td.sock_path);
try proto.writeAllFd(p.handle, &[_]u8{@intFromEnum(proto.MsgType.stats_req)});
try td.srv.pumpOnce(20);
}
defer for (peers) |p| p.close();
for (td.srv.observers) |o| try std.testing.expect(o != null);
// Armed only once every peer is seated: peer 0's clock starts at ITS
// accept, three seating rounds before this assertion, and a Debug
// ghostty on a loaded box takes longer than the deadline to get here.
// A gate that fails for its own setup's timing pins nothing.
td.srv.observer_idle_ms = 50;
std.Thread.sleep(60 * std.time.ns_per_ms);
try td.srv.pumpOnce(20);
for (td.srv.observers) |o| try std.testing.expect(o == null);
// Dropped means closed: the peer reads EOF, not a hang.
var b: [1]u8 = undefined;
try std.testing.expectEqual(@as(usize, 0), try std.posix.read(peers[0].handle, &b));
const fresh = try dial.dial(td.sock_path);
defer fresh.close();
try proto.writeFrame(fresh.handle, .stats_req, "");
const r = (try awaitFrame(alloc, &td.srv, fresh.handle, .stats_reply, 100)) orelse
return error.NoStatsReplyAfterIdleReap;
r.deinit(alloc);
}
test "Server: bytes after an attach in the same write reach the promoted client" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "promote", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try dial.dial(td.sock_path);
defer c.close();
// attach + resize in ONE write: the resize lands in the observer's
// buffer behind the attach and must follow the fd into the client slot.
var both: std.ArrayList(u8) = .empty;
defer both.deinit(alloc);
try proto.appendFrame(&both, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));
try proto.appendFrame(&both, alloc, .resize, &proto.encodeSize(132, 50));
try proto.writeAllFd(c.handle, both.items);
// awaitFrame, not firstStateFrame: this test owns the pump itself (it
// reads td.srv state below, which a server thread would race), and
// firstStateFrame only polls the peer.
const first = (try awaitFrame(alloc, &td.srv, c.handle, .snapshot, 200)) orelse
return error.NoSnapshotAfterAttach;
first.deinit(alloc);
var i: usize = 0;
while (i < 100 and td.srv.colsNow(0) != 132) : (i += 1) try td.srv.pumpOnce(5);
try std.testing.expectEqual(@as(u16, 132), td.srv.colsNow(0));
try std.testing.expectEqual(@as(u16, 50), td.srv.rowsNow(0));
// Nothing left behind on either side of the promotion.
for (td.srv.observers) |o| try std.testing.expect(o == null);
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.inbound.items.len);
}
test "Server: an observer that never reads its replies is dropped, not allowed to stop the pump" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "nagger", .{ .shell = "/bin/sh" });
defer td.deinit();
// Long, so the drop under test is the write's answer and not the idle
// reaper's — a nagging peer is never idle anyway.
td.srv.observer_idle_ms = 60_000;
const peer = try dial.dial(td.sock_path);
defer peer.close();
try td.srv.pumpOnce(20);
// Every reply is up to sessions_text_len bytes and the peer never reads
// one, so a small buffer is what an unread socket looks like a few frames
// sooner. Set on BOTH ends, because the two kernels put the bytes in
// different places: for AF_UNIX Linux holds them against the SENDER's
// send buffer, while Darwin hands them to the peer and charges the
// RECEIVER's receive buffer, so capping only the daemon's end left the
// Mac with room for the whole burst and no drop to grade.
// 1 KB and not 4 KB: the replies to this burst are a few bytes each, so
// the cap has to sit below their total on BOTH systems, and what a given
// request buys differs — measured 2026-09-03, asking for 4096 leaves
// 1792 bytes of room on Linux and the full 4096 on Darwin, which the
// whole burst fitted inside. Asking for 1024 leaves about 1024 on each.
const fd = td.srv.observers[0].?.fd;
const small: c_int = 1024;
try std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.SNDBUF, std.mem.asBytes(&small));
try std.posix.setsockopt(peer.handle, std.posix.SOL.SOCKET, std.posix.SO.RCVBUF, std.mem.asBytes(&small));
// One write, hundreds of frames: they fit in a single 64 KB read, which
// is what makes the whole burst one uninterrupted drain.
var burst: std.ArrayList(u8) = .empty;
defer burst.deinit(alloc);
for (0..400) |_| try proto.appendFrame(&burst, alloc, .sessions_req, "");
try proto.writeAllFd(peer.handle, burst.items);
// Pumps, plural, and bounded: one pump answers at most
// `max_observer_frames_per_pump` of the burst, and how many of those
// replies it takes to fill a socket is the kernel's business — Darwin's
// smallest usable buffer holds more of them than Linux's. What is being
// graded is that the daemon DROPS this peer rather than waiting on it,
// and that no pump along the way blocked.
const t0 = std.time.milliTimestamp();
for (0..64) |_| {
try td.srv.pumpOnce(20);
if (td.srv.observers[0] == null) break;
}
const spent = std.time.milliTimestamp() - t0;
// The claim: every pump came back. A blocking write here never returns
// until the peer reads, and no session on the box is served meanwhile.
try std.testing.expect(spent < 5_000);
try std.testing.expect(td.srv.observers[0] == null);
}
test "Server: a burst of observer frames is answered across pumps, never all in one" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "burst", .{ .shell = "/bin/sh" });
defer td.deinit();
td.srv.observer_idle_ms = 60_000;
const peer = try dial.dial(td.sock_path);
defer peer.close();
try td.srv.pumpOnce(20);
const over = srv_mod.max_observer_frames_per_pump + 3;
var burst: std.ArrayList(u8) = .empty;
defer burst.deinit(alloc);
for (0..over) |_| try proto.appendFrame(&burst, alloc, .sessions_req, "");
try proto.writeAllFd(peer.handle, burst.items);
try td.srv.pumpOnce(20);
try std.testing.expect(td.srv.observers[0] != null);
// The cap is what bounds one pump's work; the leftovers are still owed.
try std.testing.expect(td.srv.observers[0].?.inbound.items.len > 0);
// And owed means answered without another byte from the peer: a burst
// bigger than the cap must not sit in the buffer until the peer
// happens to write again.
var pumps: usize = 0;
while (td.srv.observers[0] != null and td.srv.observers[0].?.inbound.items.len > 0 and pumps < 20) : (pumps += 1)
try td.srv.pumpOnce(20);
try std.testing.expect(td.srv.observers[0] != null);
try std.testing.expectEqual(@as(usize, 0), td.srv.observers[0].?.inbound.items.len);
// Every one of them was answered, in order, with the whole list.
for (0..over) |_| {
const r = (try awaitFrame(alloc, &td.srv, peer.handle, .sessions_reply, 200)) orelse
return error.MissingReply;
defer r.deinit(alloc);
try std.testing.expectEqualStrings("0", r.payload);
}
}
test "Server: the two bounded deadlines read a monotonic clock, not the calendar" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "mono");
defer td.deinit();
try td.startStubborn(alloc, .{});
// CLOCK_MONOTONIC counts from boot and CLOCK_REALTIME from 1970, so any
// machine that has been up less than a decade separates them by a
// margin no scheduling delay can close. That gap is the whole assertion:
// a deadline stamped from the calendar is postponed by an NTP step
// backwards — during which `pty.master` is -1 and `mux d upgrade` is
// refused — and an idle deadline stamped from it drops every healthy
// observer on a step forwards.
const wall_now = std.time.milliTimestamp();
const obs = try dial.dial(td.sock_path);
defer obs.close();
try td.srv.pumpOnce(20);
try std.testing.expect(td.srv.observers[0] != null);
try std.testing.expect(td.srv.observers[0].?.since_ms < wall_now - 1_000_000);
var rq: [proto.end_req_max_len]u8 = undefined;
try proto.writeFrame(obs.handle, .end_req, proto.encodeEndReq(&rq, true, ""));
const r = (try awaitFrame(alloc, &td.srv, obs.handle, .end_reply, 400)) orelse
return error.NoEndReply;
defer r.deinit(alloc);
try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted);
const by = td.srv.sessions.table[0].?.end_by_ms orelse return error.NoDeadline;
try std.testing.expect(by < wall_now - 1_000_000);
try std.testing.expect(by > srv_mod.monoMs());
}
test "Server: an observer that dribbles a huge frame is dropped at the cap, not held to the idle deadline" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "fat", .{ .shell = "/bin/sh" });
defer td.deinit();
const peer = try dial.dial(td.sock_path);
defer peer.close();
try td.srv.pumpOnce(20);
try std.testing.expect(td.srv.observers[0] != null);
// A header claiming a payload the protocol admits (16 MB is the CLIENT
// path's paste ceiling), then a dribble. Nothing an observer asks for
// is bigger than a path, so four slots' worth of that ceiling is 64 MB
// the daemon would hold for the whole idle deadline — and indefinitely,
// for a peer that completes one tiny frame every nine seconds.
var hdr: [5]u8 = undefined;
hdr[0] = @intFromEnum(proto.MsgType.upgrade_req);
std.mem.writeInt(u32, hdr[1..5], 15 * 1024 * 1024, .little);
try proto.writeAllFd(peer.handle, &hdr);
const dribble = try alloc.alloc(u8, srv_mod.observer_inbound_max + 1);
defer alloc.free(dribble);
@memset(dribble, 'x');
// Dribble it the way a real peer does — write, let the daemon pump,
// write again — rather than in one blocking writeAllFd. Nobody reads
// this socket except `pumpOnce`, so the whole dribble has to fit in the
// send buffer for a single write to return, and it does not everywhere:
// a unix stream socket holds a few hundred KB on Linux and about 8 KB on
// Darwin, which is the cap itself. The one write wedged the Mac suite in
// `write` with no output. Non-blocking so a full buffer is a short write
// to hand back to the pump instead of a stall, and the loop stops the
// moment the daemon drops the observer, which is the whole point.
const fl = try std.posix.fcntl(peer.handle, std.posix.F.GETFL, 0);
const nb: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
_ = try std.posix.fcntl(peer.handle, std.posix.F.SETFL, fl | nb);
var off: usize = 0;
// Bounded: every iteration either moves bytes or gives the daemon a pump
// to drain them, so a run that spends this many without the drop has
// stopped making progress and should fail rather than hang.
for (0..1000) |_| {
if (off == dribble.len or td.srv.observers[0] == null) break;
off += std.posix.write(peer.handle, dribble[off..]) catch |e| switch (e) {
// The drop closes the socket under us; that IS the outcome.
error.WouldBlock => @as(usize, 0),
error.BrokenPipe, error.ConnectionResetByPeer => break,
else => return e,
};
try td.srv.pumpOnce(20);
}
try td.srv.pumpOnce(20);
try std.testing.expect(td.srv.observers[0] == null);
// And the cap still clears a real upgrade_req: the longest thing an
// observer legitimately sends is a path.
var long_path: [4096]u8 = undefined;
@memset(&long_path, 'p');
long_path[0] = '/';
var ureq: [8192]u8 = undefined;
const req = try proto.encodeUpgradeReq(&ureq, .{
.allow_same_version = false,
.version = "9.9.9",
.path = &long_path,
});
try std.testing.expect(5 + req.len < srv_mod.observer_inbound_max);
}
fn seatedClients(s: *const Server) usize {
var n: usize = 0;
for (s.clients) |c| {
if (c != null) n += 1;
}
return n;
}
fn heldObservers(s: *const Server) usize {
var n: usize = 0;
for (s.observers) |o| {
if (o != null) n += 1;
}
return n;
}
fn eightSeated(s: *Server) bool {
return seatedClients(s) == 8;
}
test "Server: a burst of dials past the observer table waits in the backlog, and every one is seated" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "burst", .{ .shell = "/bin/sh" });
defer td.deinit();
// Eight connects with no attach frame behind any of them yet: twice the
// observer table, and the shape of an eight-tile wall whose pumps dial
// within a hundred microseconds of each other. The gap between the
// connect and the frame is the race: the daemon's accept loop used to
// run ahead of the frames, fill the four slots with silent peers and
// close the fifth outright, which lost one tile of eight at birth with
// nothing painted (test/e2e_13_birth.sh's refused-tile leg, flaky at
// one run in four on 2026-09-05).
var conns: [8]std.net.Stream = undefined;
for (&conns) |*c| c.* = try dial.dial(td.sock_path);
defer for (conns) |c| c.close();
// Let the daemon accept what it will before any frame arrives. Exactly
// the table's worth lands; the other four are the kernel's to hold.
for (0..20) |_| try td.srv.pumpOnce(5);
try std.testing.expectEqual(srv_mod.max_observers, heldObservers(&td.srv));
try std.testing.expectEqual(@as(usize, 0), seatedClients(&td.srv));
for (conns) |c| try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
try std.testing.expect(try h.pumpUntil(&td.srv, 5000, &td.srv, eightSeated));
try std.testing.expectEqual(@as(usize, 0), heldObservers(&td.srv));
// Seated is not served: every one of the eight reads its snapshot,
// the four that waited in the backlog included.
for (0..40) |_| try td.srv.pumpOnce(5);
for (conns) |c| {
try std.testing.expect((try h.firstStateFrame(alloc, c.handle, 2000)) != null);
}
}