src/server/server_test_deliver.zig
Ref: Size: 18.8 KiB History
const std = @import("std");
const delta_mod = @import("engine").delta;
const proto = @import("term").protocol;
const h = @import("server_test_harness.zig");
const dial = h.dial;
const connectedPair = h.connectedPair;
/// So a peer that never reads backs the socket up in a few KB rather than a
/// few hundred. Linux doubles and clamps the request: a floor, not a promise.
fn shrinkSendBuf(fd: std.posix.fd_t) !void {
const v: c_int = 1024;
try std.posix.setsockopt(
fd,
std.posix.SOL.SOCKET,
std.posix.SO.SNDBUF,
std.mem.asBytes(&v),
);
}
test "Server: a stalled client does not block delivery to others" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "stall", .{ .shell = "/bin/sh" });
defer td.deinit();
// A never reads a byte, so its send buffer fills and stays full. B
// drains normally. td.srv.deinit() closes both daemon-side fds.
const a = try connectedPair();
defer std.posix.close(a.peer);
try shrinkSendBuf(a.daemon);
const b = try connectedPair();
defer std.posix.close(b.peer);
td.srv.clients[0] = .{ .sink = .{ .socket = a.daemon }, .session = 0 };
td.srv.clients[1] = .{ .sink = .{ .socket = b.daemon }, .session = 0 };
// Snapshot everyone until A backs up. A blocking write into A's full
// buffer would hang here rather than fail, so reaching the assertions
// at all is itself the "never stalls" half of this test.
//
// The five counted loops in this file are all this shape and none is a
// `pumpUntil`: each round DOES work — it feeds the engine and
// resnapshots — so the loop applies load until the daemon reacts rather
// than waiting for something already in flight. The cap loop below goes
// further and asserts on the count itself.
var rounds: usize = 0;
while (rounds < 64) : (rounds += 1) {
if (td.srv.clients[0]) |slot| {
if (slot.pending.items.len > 0) break;
} else break;
td.srv.sessions.table[0].?.eng.feed("stalled-client-test output\r\n");
td.srv.resyncSnapshot(0);
}
// A is backed up but still attached: this is nowhere near the 8 MiB cap,
// so a slow peer is queued, not dropped and not waited on.
try std.testing.expect(td.srv.clients[0] != null);
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
// B is unaffected: its queue drained completely into the kernel...
try std.testing.expect(td.srv.clients[1] != null);
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[1].?.pending.items.len);
// ...and what it received is whole parseable snapshot/state pairs, one
// per round, not a stream truncated or interleaved by A's backpressure.
var frames: usize = 0;
while (frames < rounds) : (frames += 1) {
const f = (try proto.readFrame(alloc, b.peer)) orelse break;
defer f.deinit(alloc);
try std.testing.expectEqual(proto.MsgType.snapshot, f.type);
const p = try proto.readSnapshotPrefix(f.payload);
try std.testing.expectEqual(@as(u16, 80), p.cols);
const state = (try proto.readFrame(alloc, b.peer)) orelse return error.NoSnapshotSource;
defer state.deinit(alloc);
try std.testing.expectEqual(proto.MsgType.selection_reply, state.type);
const reply = try proto.decodeSelectionReply(state.payload);
try std.testing.expectEqual(@as(u32, 0), reply.id);
try std.testing.expectEqual(@as(u32, 0), reply.gesture);
}
try std.testing.expectEqual(rounds, frames);
}
test "Server: a client exceeding the pending cap is dropped" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "cap", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
try shrinkSendBuf(c.daemon);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
// Size the cap to a few snapshots' worth of backlog. Below one frame it
// would drop on the very first send, which would prove nothing about
// bounding a backlog; the production default is 8 MiB.
// Measured off the payload the daemon actually sends, not off any other
// serialization of the grid: a cap sized from a bigger encoding is a cap
// of far more than four frames, and the loop bound below would be what
// ended the test instead of the cap.
const snap = try delta_mod.buildSnapshot(alloc, td.srv.sessions.table[0].?.eng, .{
.seq = 0,
.history_rows = 0,
.cols = 80,
.rows = 24,
.epoch = 0,
});
defer alloc.free(snap);
const frame_len = 5 + snap.len;
td.srv.pending_cap = 4 * frame_len;
var rounds: usize = 0;
while (rounds < 256 and td.srv.clients[0] != null) : (rounds += 1) {
td.srv.sessions.table[0].?.eng.feed("cap-test output\r\n");
td.srv.resyncSnapshot(0);
}
// Past the cap the daemon drops the client rather than buffering for it
// without bound. `rounds` is read after the loop, so the count is part
// of the verdict here and not a budget.
try std.testing.expect(td.srv.clients[0] == null);
// The cap is what ended it, not the loop bound: a four-frame cap behind
// a socket buffer of about one frame cannot absorb many more than five.
try std.testing.expect(rounds <= 16);
// The fd was closed, not leaked: the peer drains whatever reached the
// socket and then sees EOF.
var drain: [8192]u8 = undefined;
var saw_eof = false;
var reads: usize = 0;
while (reads < 256 and !saw_eof) : (reads += 1) {
if (try std.posix.read(c.peer, &drain) == 0) saw_eof = true;
}
try std.testing.expect(saw_eof);
// And the session carries on without it: with nobody attached there is
// no one to snapshot, so the counters stand still.
const snapshots_at_drop = td.srv.stats.snapshots;
try std.testing.expect(snapshots_at_drop > 0);
td.srv.sessions.table[0].?.eng.feed("after the drop\r\n");
td.srv.resyncSnapshot(0);
try std.testing.expectEqual(snapshots_at_drop, td.srv.stats.snapshots);
}
test "Server: a writable backlog is flushed by poll, not mistaken for input" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "pollout", .{ .shell = "/bin/sh" });
defer td.deinit();
// Drain the shell's startup output first, so the pump below has exactly
// one thing to react to: the client's writability. Raw bytes off the pty
// master, so there is no Link here to await on.
while (true) {
var pfd = [_]std.posix.pollfd{
.{ .fd = td.srv.sessions.table[0].?.pty.master, .events = std.posix.POLL.IN, .revents = 0 },
};
if ((try std.posix.poll(&pfd, 300)) == 0) break;
var buf: [64 * 1024]u8 = undefined;
_ = std.posix.read(td.srv.sessions.table[0].?.pty.master, &buf) catch break;
}
const c = try connectedPair();
defer std.posix.close(c.peer);
try shrinkSendBuf(c.daemon);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
// Back the client up: the kernel stops taking bytes, so they queue.
var rounds: usize = 0;
while (rounds < 64) : (rounds += 1) {
if (td.srv.clients[0].?.pending.items.len > 0) break;
td.srv.sessions.table[0].?.eng.feed("pollout-test output\r\n");
td.srv.resyncSnapshot(0);
}
const backlog = td.srv.clients[0].?.pending.items.len;
try std.testing.expect(backlog > 0);
// The client reads, so its socket is writable again — but it sends
// nothing, so there is no input frame behind that POLLOUT. A pump that
// took writability for readability would block forever in
// serviceClient's read here, so returning at all is half the assertion.
var drain: [16 * 1024]u8 = undefined;
_ = try std.posix.read(c.peer, &drain);
try td.srv.pumpOnce(50);
// The other half: the pump used the POLLOUT for what it was.
try std.testing.expect(td.srv.clients[0] != null);
try std.testing.expect(td.srv.clients[0].?.pending.items.len < backlog);
}
test "Server: a partially flushed queue delivers every byte exactly once" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "pf", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
try shrinkSendBuf(c.daemon);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
td.srv.pending_cap = 64 * 1024 * 1024;
// Frames of varying length, queued while the peer is silent, so most of
// them pile up behind a full socket buffer and every subsequent flush
// resumes mid-queue.
const n_frames = 400;
var expect: std.ArrayList(u8) = .empty;
defer expect.deinit(alloc);
var k: usize = 0;
while (k < n_frames) : (k += 1) {
var pbuf: [512]u8 = undefined;
const full = try std.fmt.bufPrint(&pbuf, "frame-{d}-{s}", .{ k, "x" ** 320 });
const payload = full[0 .. 20 + (k % 280)];
try proto.appendFrame(&expect, alloc, .dump_reply, payload);
try std.testing.expect(td.srv.queueFrame(0, .dump_reply, payload));
}
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
// Drain and flush alternately: each flush starts where the last stopped.
// Byte-level on purpose, all the way down: the verdict below is that
// what the peer read equals `expect` BYTE for byte, so these two loops
// read a stream rather than frames and cannot be a frame await.
var got: std.ArrayList(u8) = .empty;
defer got.deinit(alloc);
var spins: usize = 0;
while (td.srv.clients[0] != null and
td.srv.clients[0].?.pending.items.len > 0 and spins < 100000) : (spins += 1)
{
var rbuf: [4096]u8 = undefined;
var pfd = [_]std.posix.pollfd{.{ .fd = c.peer, .events = std.posix.POLL.IN, .revents = 0 }};
if ((try std.posix.poll(&pfd, 200)) > 0) {
const n = try std.posix.read(c.peer, &rbuf);
if (n == 0) break;
try got.appendSlice(alloc, rbuf[0..n]);
}
td.srv.flushClient(0);
}
while (got.items.len < expect.items.len) {
var rbuf: [4096]u8 = undefined;
var pfd = [_]std.posix.pollfd{.{ .fd = c.peer, .events = std.posix.POLL.IN, .revents = 0 }};
if ((try std.posix.poll(&pfd, 200)) == 0) break;
const n = try std.posix.read(c.peer, &rbuf);
if (n == 0) break;
try got.appendSlice(alloc, rbuf[0..n]);
}
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.pending.items.len);
try std.testing.expectEqualSlices(u8, expect.items, got.items);
}
test "Server: the shell's exit status reaches an attached client" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "exit", .{ .shell = "/bin/sh" });
defer td.deinit();
try td.threaded();
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// Settle first, so the shell is ready for input and the exit below is
// the next thing that happens. Its own poll, like the quiet-drains in
// server_test_attach.zig: the condition is SILENCE, and inside
// `Link.awaitFrame` a poll that timed out is indistinguishable from the
// deadline running out.
var quiet_ms: u64 = 0;
var deadline_ms: u64 = 10_000;
while (deadline_ms > 0 and quiet_ms < 500) {
var pfd = [_]std.posix.pollfd{
.{ .fd = c.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, c.handle)) orelse break;
frame.deinit(alloc);
}
// The exit code must survive the daemon's own shutdown. Without the
// bounded drain this frame can be lost, and a client that misses it
// reports "connection to the daemon lost" and exits 1 instead of 7.
try proto.writeFrame(c.handle, .input, "exit 7\n");
var status: ?u8 = null;
if (try h.awaitFrameOn(alloc, c.handle, .exit_status, 10_000)) |frame| {
defer frame.deinit(alloc);
try std.testing.expectEqual(@as(usize, 1), frame.payload.len);
status = frame.payload[0];
}
try std.testing.expectEqual(@as(?u8, 7), status);
}
/// Test helper: read one fd until it has been quiet for `quiet_ms`, keeping
/// everything. Runs on its own thread so a drain under test has a peer that
/// is actually consuming. Raw bytes and a silence condition, so neither half
/// of it is a frame await.
const PeerDrainer = struct {
fd: std.posix.fd_t,
alloc: std.mem.Allocator,
out: std.ArrayList(u8) = .empty,
fn run(self: *PeerDrainer, quiet_ms: u64) void {
var quiet: u64 = 0;
while (quiet < quiet_ms) {
var pfd = [_]std.posix.pollfd{
.{ .fd = self.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&pfd, 100) catch return;
if (ready == 0) {
quiet += 100;
continue;
}
quiet = 0;
var buf: [8192]u8 = undefined;
const n = std.posix.read(self.fd, &buf) catch return;
if (n == 0) return;
self.out.appendSlice(self.alloc, buf[0..n]) catch return;
}
}
};
test "Server: the exit drain delivers a backlog once the peer resumes reading" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "exitdrain", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
try shrinkSendBuf(c.daemon);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
// Back the client up, so the exit_status queued next lands behind a
// backlog the kernel has already refused — the situation in which the
// frame would otherwise be lost.
var rounds: usize = 0;
while (rounds < 64) : (rounds += 1) {
if (td.srv.clients[0].?.pending.items.len > 0) break;
td.srv.sessions.table[0].?.eng.feed("exit-drain output\r\n");
td.srv.resyncSnapshot(0);
}
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
try std.testing.expect(td.srv.queueFrame(0, .exit_status, &.{7}));
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
var drainer = PeerDrainer{ .fd = c.peer, .alloc = alloc };
defer drainer.out.deinit(alloc);
const th = try std.Thread.spawn(.{}, PeerDrainer.run, .{ &drainer, 700 });
td.srv.drainPending(5000);
th.join();
// Everything owed went out...
try std.testing.expect(td.srv.clients[0] != null);
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.pending.items.len);
// ...and since the queue is FIFO and exit_status was queued last, the
// stream ends with exactly its wire bytes.
const tail = [_]u8{ 0x82, 1, 0, 0, 0, 7 };
try std.testing.expect(drainer.out.items.len >= tail.len);
try std.testing.expectEqualSlices(
u8,
&tail,
drainer.out.items[drainer.out.items.len - tail.len ..],
);
}
test "Server: the exit drain gives up on a peer that never reads" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "drain", .{ .shell = "/bin/sh" });
defer td.deinit();
const c = try connectedPair();
defer std.posix.close(c.peer);
try shrinkSendBuf(c.daemon);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon }, .session = 0 };
var rounds: usize = 0;
while (rounds < 64) : (rounds += 1) {
if (td.srv.clients[0].?.pending.items.len > 0) break;
td.srv.sessions.table[0].?.eng.feed("drain-test output\r\n");
td.srv.resyncSnapshot(0);
}
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
// The peer is silent and its buffer is full, so this can never finish.
// It must return on the deadline rather than hold the daemon open: a
// dead client must not be able to stop the session from exiting.
var timer = try std.time.Timer.start();
td.srv.drainPending(150);
const elapsed_ms = timer.read() / std.time.ns_per_ms;
try std.testing.expect(td.srv.clients[0] != null);
try std.testing.expect(td.srv.clients[0].?.pending.items.len > 0);
// Generous upper bound: the point is that it is bounded at all, not the
// precise number, which a loaded CI box would make flaky.
try std.testing.expect(elapsed_ms < 5000);
}
test "Server: broadcast stats count every send but the counterfactual once" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "stats", .{ .shell = "/bin/sh" });
defer td.deinit();
// Two stand-in clients, driving the server directly so the accounting is
// exact instead of hostage to shell timing. Real sockets, not pipes: the
// send path is send(2) now. td.srv.deinit() closes the daemon-side fds.
const pa = try connectedPair();
defer std.posix.close(pa.peer);
const pb = try connectedPair();
defer std.posix.close(pb.peer);
td.srv.clients[0] = .{ .sink = .{ .socket = pa.daemon }, .session = 0 };
td.srv.clients[1] = .{ .sink = .{ .socket = pb.daemon }, .session = 0 };
try td.srv.sessions.table[0].?.tracker.rebuild(alloc, td.srv.sessions.table[0].?.eng, 24, 80);
td.srv.sessions.table[0].?.eng.feed("one update, two recipients");
td.srv.sendUpdate(0);
const fa = (try proto.readFrame(alloc, pa.peer)).?;
defer fa.deinit(alloc);
const fb = (try proto.readFrame(alloc, pb.peer)).?;
defer fb.deinit(alloc);
try std.testing.expectEqual(proto.MsgType.delta, fa.type);
try std.testing.expectEqual(proto.MsgType.delta, fb.type);
try std.testing.expectEqualSlices(u8, fa.payload, fb.payload);
// Actuals are per send: two clients, two deltas on the wire.
try std.testing.expectEqual(@as(u64, 2), td.srv.stats.deltas);
try std.testing.expectEqual(@as(u64, @intCast(2 * fa.payload.len)), td.srv.stats.delta_bytes);
// The counterfactual is per event: what a snapshot-only daemon would
// have sent for this one update is one snapshot, however many clients
// received it.
// Rebuilt here the way the daemon builds it, so the stat is checked
// against the bytes a snapshot would really have cost rather than
// against a second serialization nobody sends.
const s = td.srv.sessions.table[0].?;
const snap = try delta_mod.buildSnapshot(alloc, s.eng, .{
.seq = s.tracker.seq,
.history_rows = s.eng.historyRows(),
.cols = 80,
.rows = 24,
.epoch = s.epoch,
});
defer alloc.free(snap);
try std.testing.expectEqual(
@as(u64, @intCast(snap.len)),
td.srv.stats.snapshot_equiv_bytes,
);
try std.testing.expectEqual(@as(u64, 0), td.srv.stats.snapshots);
}