src/server/server_test_quic.zig
Ref: Size: 28.0 KiB History
const std = @import("std");
const Grid = @import("term").grid.Grid;
const proto = @import("term").protocol;
const quic = @import("quic");
const quic_server = @import("quic_server.zig");
const TmpDir = @import("testtmp").TmpDir;
const h = @import("server_test_harness.zig");
const srv_mod = @import("server.zig");
const Server = srv_mod.Server;
const drainWaitMs = srv_mod.drainWaitMs;
const stallExhausted = srv_mod.stallExhausted;
const shutdown_flag = &srv_mod.shutdown_flag;
const findFrame = h.findFrame;
const quicPump = h.quicPump;
const quicTestServer = h.quicTestServer;
fn attachOver(cl: *quic_server.TestPeer, buf: *std.ArrayList(u8), alloc: std.mem.Allocator) !void {
try proto.appendFrame(buf, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));
cl.out = buf.items;
cl.drain();
}
test "the QUIC listener holds the admitted peers and unclassified role table" {
// A QUIC connection exists from the moment its handshake completes and
// only THEN asks for a client slot, so the listener needs room the client
// table does not: peers mid-handshake, and peers whose slot request is
// about to be refused, all occupy a connection first.
//
// A listener no deeper than the client table drops peers the daemon still
// has seats for, and that drop is SILENT — `acceptConn` answers a full
// table with "full: drop, the peer will retry", so those peers hang on
// retries instead of being told anything, and a host polled that way
// reads `unreachable` on a daemon that is running fine.
//
// Pinned rather than derived: `quic_server.zig` is a transport and does
// not read the daemon's tables, so its number is restated there. This is
// what catches the restatement going stale.
try std.testing.expect(quic_server.max_conns > srv_mod.max_clients + srv_mod.max_forward_peers);
try std.testing.expectEqual(quic_server.max_conns, srv_mod.max_quic_pending);
}
test "Server: QUIC forward-role hello uses no terminal client slot" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qforward", .{ .shell = "/bin/sh" });
const key: quic.Key = .{ .bytes = [_]u8{0x46} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
defer q.l.deinit();
defer td.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try proto.appendFrame(&out, alloc, .forward_hello, &proto.encodeForwardHello());
cl.out = out.items;
cl.drain();
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 8000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .forward_ready) != null;
}
}.f);
const ready = findFrame(cl.cl.in.items, .forward_ready) orelse return error.NoForwardReady;
try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready));
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));
}
test "Server: QUIC one-shot stats and status reply without consuming terminal slots" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qoneshot", .{ .shell = "/bin/cat" });
const key: quic.Key = .{ .bytes = [_]u8{0x52} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
defer q.l.deinit();
defer td.deinit();
// These are real QUIC peers, deliberately not socket observers. Each
// request is its own connection because one-shot requests close only
// after their full reply has drained and must never be promoted to a
// terminal slot.
var stats = try quic_server.TestPeer.init(q.addr, key);
defer stats.deinit();
var status = try quic_server.TestPeer.init(q.addr, key);
defer status.deinit();
var stats_out: std.ArrayList(u8) = .empty;
defer stats_out.deinit(alloc);
var status_out: std.ArrayList(u8) = .empty;
defer status_out.deinit(alloc);
try proto.appendFrame(&stats_out, alloc, .stats_req, "");
try proto.appendFrame(&status_out, alloc, .status_req, "");
stats.out = stats_out.items;
status.out = status_out.items;
stats.drain();
status.drain();
var both = [_]*quic_server.TestPeer{ &stats, &status };
try quicPump(&td.srv, &both, 8000, &status, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .status_reply) != null;
}
}.f);
const stats_reply = findFrame(stats.cl.in.items, .stats_reply) orelse return error.NoQuicStatsReply;
try std.testing.expect(std.mem.indexOf(u8, stats_reply, "clients=0") != null);
const status_reply = findFrame(status.cl.in.items, .status_reply) orelse return error.NoQuicStatusReply;
const decoded = try proto.decodeStatusReply(status_reply);
try std.testing.expectEqual(@as(u16, 80), decoded.cols);
try std.testing.expectEqual(@as(u16, 24), decoded.rows);
try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
// A completed one-shot has no retained provisional entry either. Pumping
// until both replies arrived rules out observing a merely queued response.
try quicPump(&td.srv, &both, 8000, &td.srv, struct {
fn f(s: *Server) bool {
for (s.quic_pending) |slot| if (slot != null) return false;
return true;
}
}.f);
}
test "Server: a malformed first QUIC frame releases its provisional slot" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qbadfirst", .{ .shell = "/bin/cat" });
const key: quic.Key = .{ .bytes = [_]u8{0x53} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
defer q.l.deinit();
defer td.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
// input is neither an attach nor an observer verb. It is a complete,
// legal frame in the wrong first-frame position, so it exercises role
// classification rather than a parser rejection.
try proto.appendFrame(&out, alloc, .input, "x");
cl.out = out.items;
cl.drain();
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 8000, &td.srv, struct {
fn f(s: *Server) bool {
for (s.quic_pending) |slot| if (slot != null) return false;
return true;
}
}.f);
try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
try std.testing.expectEqual(@as(?usize, 0), td.srv.forwards.freePeer());
}
test "refusalFrame: the exact refusal wire bytes" {
// The literal pins what this daemon's QUIC path puts on the wire, and
// nothing about the client that reads it. Spelled out rather than
// computed from proto constants so that changing the verb or the
// length has to be done here too, deliberately:
// type=exit_status (0x82), len=1 LE, payload=exit code 1.
const f = Server.refusalFrame();
try std.testing.expectEqualSlices(u8, &.{ 0x82, 1, 0, 0, 0, 1 }, &f);
// serviceObserver frames the same refusal independently for the socket
// path, and quicOnOpen's comment calls the two "the same answer sent
// the same way" — this is what stops that from being wishful. The
// socket path calls writeFrame, which takes an fd; appendFrame stands
// in for it because protocol.zig pins the two to one layout.
var socket_bytes: std.ArrayList(u8) = .empty;
defer socket_bytes.deinit(std.testing.allocator);
try proto.appendFrame(&socket_bytes, std.testing.allocator, .exit_status, &.{1});
try std.testing.expectEqualSlices(u8, socket_bytes.items, &f);
}
test "drainPending: one quiet wakeup is not a verdict; a run of them is" {
var stalls: usize = 0;
// Written with literal counts rather than a loop over max_drain_stalls:
// a loop parameterised by the constant moves its own goalposts and would
// pass at any value of it, which is the whole failure this bound exists
// to fix restated as a test bug.
var i: usize = 0;
while (i < 63) : (i += 1) {
try std.testing.expect(!stallExhausted(&stalls, false));
}
// The 64th consecutive nothing is where patience runs out.
try std.testing.expect(stallExhausted(&stalls, false));
// And progress resets the run. A slow peer that keeps taking bytes is
// not a stuck one, however many quiet wakeups it accumulates between
// them — the bound is on consecutive silence, not on a lifetime total.
stalls = 0;
i = 0;
while (i < 63) : (i += 1) {
try std.testing.expect(!stallExhausted(&stalls, false));
}
try std.testing.expect(!stallExhausted(&stalls, true));
try std.testing.expectEqual(@as(usize, 0), stalls);
i = 0;
while (i < 63) : (i += 1) {
try std.testing.expect(!stallExhausted(&stalls, false));
}
}
test "drainPending: the poll slice is floored at 1ms and capped by ngtcp2" {
// No QUIC listener: the whole remaining budget, as before.
try std.testing.expectEqual(@as(i32, 250), drainWaitMs(250, null));
// ngtcp2 says something is already due. Unfloored this returns 0, and a
// zero-timeout poll spins the loop hot for the entire budget instead of
// waiting for the acknowledgement that would end it — ngtcp2 keeps
// reporting that expiry as past until the event clearing it arrives.
try std.testing.expectEqual(@as(i32, 1), drainWaitMs(250, 0));
// A real deadline is honoured: sleeping past it starves the PTO timer,
// and since retransmission lives only in tick() a lost packet would
// never be resent.
try std.testing.expectEqual(@as(i32, 10), drainWaitMs(250, 10));
// ...but never past the budget that is actually left.
try std.testing.expectEqual(@as(i32, 5), drainWaitMs(5, 100));
try std.testing.expectEqual(@as(i32, 1), drainWaitMs(1, 100));
}
test "Server: output reaches a silent QUIC client without waiting for it to speak" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "qquiet");
// /bin/cat: a session that emits nothing on its own, so the only output
// in this test is the output the test causes.
const key: quic.Key = .{ .bytes = [_]u8{0x4D} ** quic.key_len };
const q = setup: {
// Only until the pair below is registered: an error before that
// leaves nobody to free the daemon. Scoped to this block so it
// cannot fire alongside the `defer td.deinit()` that follows.
errdefer td.deinit();
try td.start(.{ .shell = "/bin/cat" });
break :setup try quicTestServer(&td.srv, key);
};
// ORDER IS LOAD-BEARING, and it is the order main.zig gives the real
// daemon: defers run in reverse, so the server tears down FIRST and the
// borrowed listener outlives it. A QUIC sink closes its connection
// THROUGH the listener (`Sink.close` -> `Listener.closeConn`), so
// `Server.deinit` walking its client table reads the listener after the
// other order would have freed it. `server.zig`'s deinit says the same
// rule for the lazily-bound arm.
defer q.l.deinit();
defer td.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try attachOver(&cl, &out, alloc);
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 8000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .snapshot) != null;
}
}.f);
try std.testing.expect(findFrame(cl.cl.in.items, .snapshot) != null);
// From here the client says NOTHING. That is the whole point: the
// listener's send only queues, and the two drains that fire on their own
// are the one after read_pkt — which needs an inbound packet — and tick,
// which only services connections whose timer is due. A broadcast to a
// quiet client is covered by neither, and the end-of-pump drainAll is
// the only thing that moves it.
// Quiesce first, or the measurement is meaningless. Straight after the
// attach exchange ngtcp2 has a loss-detection timer already past due, so
// tick() drains on expiry and delivers the frame below whether or not
// anything else does. Settle until the next timer is comfortably in the
// future; only then is tick ruled out as the deliverer.
// Every poll in this file is on the QUIC socket, driving a TestPeer's
// own ngtcp2 state — none of them is a frame await, so none becomes a
// `Link.awaitFrame`. A QUIC Link owns its client and would close it.
var settle: usize = 0;
while (settle < 400) : (settle += 1) {
if (q.l.timeoutMs(1000) > 100) break;
try td.srv.pumpOnce(5);
cl.drain();
var pfd = [_]std.posix.pollfd{
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
if ((std.posix.poll(&pfd, 1) catch 0) > 0) cl.cl.readable();
}
try std.testing.expect(q.l.timeoutMs(1000) > 100);
const before = cl.echoed();
cl.out = &.{};
td.srv.sessions.table[0].?.eng.feed("quiet-client-marker\r\n");
td.srv.sendUpdate(0);
// EXACTLY ONE pump, and the client only ever listens. Both halves are
// the test, and both were learned by watching weaker versions pass:
//
// quicPump calls cl.drain() every iteration, which makes the client
// TRANSMIT — its acknowledgements alone suffice — and every inbound
// packet hands feed() a drain to carry the reply out on. And looping
// over several pumps is no better: on loopback the PTO expiry comes due
// within a few hundred milliseconds, and tick drains on expiry, so any
// budget generous enough not to be flaky is generous enough to pass
// without drainAll existing at all.
//
// One pump answers the only question that isolates it: the frame was
// queued before this call, so did THIS call put it on the wire?
try td.srv.pumpOnce(5);
// Reading is not transmitting: the client takes whatever already
// arrived, and never gives the server an inbound packet to react to.
// And this one must NOT pump the daemon at all, which is the whole
// isolation the paragraph above sets up.
var tries: usize = 0;
while (tries < 60 and findFrame(cl.cl.in.items, .delta) == null) : (tries += 1) {
var pfd = [_]std.posix.pollfd{
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
if ((std.posix.poll(&pfd, 2) catch 0) > 0) cl.cl.readable();
}
try std.testing.expect(cl.echoed() > before);
try std.testing.expect(findFrame(cl.cl.in.items, .delta) != null);
}
test "Server: a QUIC client still receives the shell's exit status" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qexit", .{ .shell = "/bin/sh" });
const key: quic.Key = .{ .bytes = [_]u8{0x31} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
// The listener outlives the server: see the ordering note above.
defer q.l.deinit();
defer td.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try attachOver(&cl, &out, alloc);
// Attached and carrying state: the frame path works before we start
// asking about the harder one.
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 8000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .snapshot) != null;
}
}.f);
try std.testing.expect(findFrame(cl.cl.in.items, .snapshot) != null);
// Now the case drainPending gets wrong when a QUIC slot contributes a
// -1 descriptor: the shell exits, the exit_status is queued, and the
// 250ms budget must actually deliver it rather than sleeping through it.
try proto.appendFrame(&out, alloc, .input, "exit 7\n");
cl.out = out.items;
cl.drain();
try quicPump(&td.srv, &only, 15000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .exit_status) != null;
}
}.f);
const status = findFrame(cl.cl.in.items, .exit_status);
try std.testing.expect(status != null);
try std.testing.expectEqual(@as(usize, 1), status.?.len);
try std.testing.expectEqual(@as(u8, 7), status.?[0]);
}
test "Server: one QUIC client leaving does not disturb the other" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qtwo", .{ .shell = "/bin/sh" });
const key: quic.Key = .{ .bytes = [_]u8{0x77} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
// The listener outlives the server: see the ordering note above.
defer q.l.deinit();
defer td.deinit();
var a = try quic_server.TestPeer.init(q.addr, key);
defer a.deinit();
var b = try quic_server.TestPeer.init(q.addr, key);
defer b.deinit();
var abuf: std.ArrayList(u8) = .empty;
defer abuf.deinit(alloc);
var bbuf: std.ArrayList(u8) = .empty;
defer bbuf.deinit(alloc);
try attachOver(&a, &abuf, alloc);
try attachOver(&b, &bbuf, alloc);
var both = [_]*quic_server.TestPeer{ &a, &b };
try quicPump(&td.srv, &both, 10000, &b, struct {
fn f(t: *quic_server.TestPeer) bool {
return findFrame(t.cl.in.items, .snapshot) != null;
}
}.f);
try std.testing.expect(findFrame(a.cl.in.items, .snapshot) != null);
try std.testing.expect(findFrame(b.cl.in.items, .snapshot) != null);
// A leaves. Sink.close() must tear down A's connection and NOT the UDP
// socket every other client is reached through — for a socket client
// those are the same object, which is exactly why this needs a test.
try proto.appendFrame(&abuf, alloc, .detach, "");
a.out = abuf.items;
a.drain();
try quicPump(&td.srv, &both, 4000, &td.srv, struct {
fn f(s: *Server) bool {
var n: usize = 0;
for (s.clients) |slot| {
if (slot != null) n += 1;
}
return n == 1;
}
}.f);
// B is still served: a marker typed now must come back to it.
const before = b.echoed();
try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");
b.out = bbuf.items;
b.drain();
var replica = try Grid.init(alloc, 80, 24);
defer replica.deinit();
// The predicate has to be the ASSERTION, not a weaker relative of it:
// `echoed()` counts every byte B has ever taken, and B took a snapshot
// long before this marker, so "more than zero, and some delta has
// arrived" was already true the moment the pump started. It returned at
// once and the assertion below then graded bytes from earlier in the
// test. On Linux the delta happened to be late enough that the pump
// waited anyway; on the Mac it was not, and the test failed.
const Grew = struct {
peer: *quic_server.TestPeer,
was: usize,
fn f(self: *@This()) bool {
return self.peer.echoed() > self.was;
}
};
var grew: Grew = .{ .peer = &b, .was = before };
try quicPump(&td.srv, &both, 15000, &grew, Grew.f);
try std.testing.expect(b.echoed() > before);
}
test "Server: a QUIC stop requester receives CONNECTION_CLOSE during deinit" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "qstopclose", .{ .shell = "/bin/cat" });
const key: quic.Key = .{ .bytes = [_]u8{0x5C} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
// The borrowed listener remains live through Server.deinit.
defer q.l.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
shutdown_flag.store(false, .release);
defer shutdown_flag.store(false, .release);
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try proto.appendFrame(&out, alloc, .stop_req, "");
cl.out = out.items;
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 8000, shutdown_flag, struct {
fn f(flag: *std.atomic.Value(bool)) bool {
return flag.load(.acquire);
}
}.f);
try std.testing.expect(shutdown_flag.load(.acquire));
td.deinit();
// No server pump remains: receiving this packet is solely closeAll's
// synchronous CONNECTION_CLOSE, not the listener's idle timeout.
var i: usize = 0;
while (i < 100 and !cl.cl.dead) : (i += 1) {
var pfd = [_]std.posix.pollfd{.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
_ = std.posix.poll(&pfd, 2) catch 0;
cl.drain();
}
try std.testing.expect(cl.cl.dead);
}
test "Server: a large QUIC dump drains over ACK refill and one-shot bounds expire" {
const alloc = std.testing.allocator;
// This viewport yields a real dump larger than the 256 KiB transport ring.
var td = try h.TestDaemon.init(alloc, "qlargedump", .{ .shell = "/bin/cat", .cols = 1024, .rows = 512 });
defer td.deinit();
const key: quic.Key = .{ .bytes = [_]u8{0x5B} ** quic.key_len };
const q = try quicTestServer(&td.srv, key);
defer q.l.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
td.srv.ses(0).eng.feed("x" ** (1024 * 512));
const expected = try td.srv.ses(0).eng.dumpPlain(alloc);
defer alloc.free(expected);
try std.testing.expect(expected.len > quic.egress_cap);
var out: std.ArrayList(u8) = .empty;
defer out.deinit(alloc);
try proto.appendFrame(&out, alloc, .debug_dump, &.{0});
cl.out = out.items;
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&td.srv, &only, 20_000, &cl, struct {
fn f(peer: *quic_server.TestPeer) bool {
return findFrame(peer.cl.in.items, .dump_reply) != null;
}
}.f);
const dump = findFrame(cl.cl.in.items, .dump_reply) orelse return error.NoDumpReply;
try std.testing.expectEqualSlices(u8, expected, dump);
try quicPump(&td.srv, &only, 8000, &td.srv, struct {
fn f(s: *Server) bool {
return srv_mod.countLive(s.quic_pending) == 0;
}
}.f);
// This is checked before append/allocation; an impossible reply cannot
// expand an ArrayList past the stated per-peer response budget.
try std.testing.expect(Server.quicOneShotCanQueue(0, proto.max_payload));
try std.testing.expect(!Server.quicOneShotCanQueue(0, proto.max_payload + 1));
try std.testing.expect(!Server.quicOneShotCanQueue(srv_mod.quic_one_shot_pending_cap, 0));
try std.testing.expect(!Server.quicOneShotExpired(100, 100 + srv_mod.quic_one_shot_deadline_ms - 1));
try std.testing.expect(Server.quicOneShotExpired(100, 100 + srv_mod.quic_one_shot_deadline_ms));
}
test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" {
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}/qcap.sock", .{dir_path});
defer alloc.free(sock_path);
var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
const key: quic.Key = .{ .bytes = [_]u8{0x5D} ** quic.key_len };
const q = try quicTestServer(&srv, key);
defer q.l.deinit();
defer srv.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var bbuf: std.ArrayList(u8) = .empty;
defer bbuf.deinit(alloc);
try attachOver(&cl, &bbuf, alloc);
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&srv, &only, 8000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return t.cl.handshake_done and t.echoed() > 0;
}
}.f);
try std.testing.expect(srv.clients[0] != null);
// From here the client acknowledges nothing and reads nothing, so the
// connection's egress ring fills and stays full. Everything the daemon
// queues after that has nowhere to go but the slot's own queue, which is
// what pending_cap is for. While the QUIC sink accepted every byte
// unconditionally, this backlog lived inside the listener where no cap
// could see it, and the daemon would have grown without bound rather
// than dropping one hopeless client.
srv.pending_cap = 512 * 1024;
const chunk = try alloc.alloc(u8, 32 * 1024);
defer alloc.free(chunk);
@memset(chunk, 'q');
// Not a `pumpUntil`: each round QUEUES another frame, so the loop is
// applying load rather than waiting for a condition.
var i: usize = 0;
while (i < 200 and srv.clients[0] != null) : (i += 1) {
_ = srv.queueFrame(0, .snapshot, chunk);
// The daemon's own machinery, not a hand-rolled loop: expiry,
// egress and the flush that follows an ack all live in pumpOnce.
srv.pumpOnce(1) catch {};
}
try std.testing.expect(srv.clients[0] == null);
// The listener let go of the connection with the slot: a dropped client
// must not leave its conn behind holding a ring.
try std.testing.expect(q.l.pendingBytes(1) == 0);
}
test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
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}/qdrain.sock", .{dir_path});
defer alloc.free(sock_path);
var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
const key: quic.Key = .{ .bytes = [_]u8{0x6D} ** quic.key_len };
const q = try quicTestServer(&srv, key);
defer q.l.deinit();
defer srv.deinit();
var cl = try quic_server.TestPeer.init(q.addr, key);
defer cl.deinit();
var bbuf: std.ArrayList(u8) = .empty;
defer bbuf.deinit(alloc);
try attachOver(&cl, &bbuf, alloc);
var only = [_]*quic_server.TestPeer{&cl};
try quicPump(&srv, &only, 8000, &cl, struct {
fn f(t: *quic_server.TestPeer) bool {
return t.cl.handshake_done and t.echoed() > 0;
}
}.f);
try std.testing.expect(srv.clients[0] != null);
const before = cl.echoed();
// Queue more than one ring's worth in one go, so the drain cannot
// possibly finish by handing everything over once: some of it is still
// in the slot's queue, and the rest is in ngtcp2's hands unacknowledged.
// Both have to be waited out, which is the arm that used to return
// immediately and call the job done.
const payload = try alloc.alloc(u8, 700 * 1024);
defer alloc.free(payload);
@memset(payload, 'D');
_ = srv.queueFrame(0, .snapshot, payload);
try std.testing.expect(srv.clients[0] != null);
try std.testing.expect(srv.clients[0].?.pending.items.len > 0);
// drainPending has to do the waiting itself. Nothing else is driving the
// daemon here — the only thing servicing the listener for the rest of
// this test is the loop inside drainPending.
var done = false;
const t0 = std.time.milliTimestamp();
const th = try std.Thread.spawn(.{}, struct {
fn f(client: *quic_server.TestPeer, flag: *bool) void {
// The peer: reads and acknowledges until the daemon says stop.
while (!flag.*) {
client.drain();
var pfd = [_]std.posix.pollfd{
.{ .fd = client.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
if ((std.posix.poll(&pfd, 5) catch 0) > 0) client.cl.readable();
}
}
}.f, .{ &cl, &done });
srv.drainPending(15_000);
done = true;
th.join();
const elapsed = std.time.milliTimestamp() - t0;
// Everything owed actually left, and it left inside the budget.
try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.pending.items.len);
try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.sink.inFlight());
try std.testing.expect(cl.echoed() >= before + payload.len);
try std.testing.expect(elapsed < 15_000);
}