c651e9a0

refactor: one QUIC client, dialled by the product and by the tests

a73x 2026-08-29 10:01

Commit message

refactor: one QUIC client, dialled by the product and by the tests

`quic_server.TestClient` was a second hand-written QUIC client — psk
callback, conn_new, the writev drain, read_pkt — 200 product lines whose
only reason to exist was the folder rule "server names no client module".
The rule stands; the reason is gone. `Client` moves down into `quic.zig`
beside the handshake helpers both ends already share, `src/client/
quic_client.zig` is deleted with its build.zig row, and the listener's
own tests dial the shipping client through a 25-line `TestPeer` that adds
only what a test wants: a backlog it re-offers as the ring drains, and a
value a stack `defer` can deinit.

Two fields earn their place in the shipping client rather than a copy:
`extend_windows` (off, the blocked-stream branch is a documented ngtcp2
return with no other way in) and `readable` made pub (reading without
transmitting is what keeps "did THIS pump put the frame on the wire?"
answerable).

quic.zig's header claim "nothing here knows what a connection is" was
true and is now false, so it says what is true instead.

check rc 0; E2E_ONLY=04_handoff 6 scenarios rc 0, E2E_ONLY=10_agent 5
scenarios rc 0. Product lines 16517 -> 16326.

CLAUDE.md

build.zig

src/cli/muxa.zig

Old		New
@@ -17,7 +17,6 @@
17	const std = @import("std");	17	const std = @import("std");
18	const proto = @import("protocol");	18	const proto = @import("protocol");
19	const sockpath = @import("sockpath");	19	const sockpath = @import("sockpath");
20	const quic_client = @import("quic_client");
21	const quic = @import("quic");	20	const quic = @import("quic");
22	const xdg = @import("xdg");	21	const xdg = @import("xdg");
23	const cliflags = @import("cliflags");	22	const cliflags = @import("cliflags");
@@ -307,9 +306,9 @@ test "parseArgs: --quic and --key, and the pairs that make no sense" {
307	/// rotated under us) and would then fail the handshake for a reason that	306	/// rotated under us) and would then fail the handshake for a reason that
308	/// has nothing to do with why the first connection died.	307	/// has nothing to do with why the first connection died.
309	const Quic = struct {	308	const Quic = struct {
310	cl: *quic_client.Client,	309	cl: *quic.Client,
311	addr: std.net.Address,	310	addr: std.net.Address,
312	key: quic_client.Key,	311	key: quic.Key,
313	idle_ms: u32,	312	idle_ms: u32,
314	/// Wall-clock milliseconds the FIRST handshake took, which is this	313	/// Wall-clock milliseconds the FIRST handshake took, which is this
315	/// client's only measurement of how far away the daemon is. `graceMs`	314	/// client's only measurement of how far away the daemon is. `graceMs`
@@ -372,12 +371,12 @@ const Conn = struct {
372	fn openQuic(	371	fn openQuic(
373	alloc: std.mem.Allocator,	372	alloc: std.mem.Allocator,
374	addr: std.net.Address,	373	addr: std.net.Address,
375	key: quic_client.Key,	374	key: quic.Key,
376	idle_ms: u32,	375	idle_ms: u32,
377	deadline_ms: i64,	376	deadline_ms: i64,
378	) !Conn {	377	) !Conn {
379	const started = std.time.milliTimestamp();	378	const started = std.time.milliTimestamp();
380	const cl = try quic_client.Client.connect(alloc, addr, key, idle_ms);	379	const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
381	errdefer cl.deinit();	380	errdefer cl.deinit();
382	try waitReady(cl, deadline_ms);	381	try waitReady(cl, deadline_ms);
383	return .{	382	return .{
@@ -564,7 +563,7 @@ const Conn = struct {
564	/// it after this point anyway.	563	/// it after this point anyway.
565	fn reconnect(self: *Conn, deadline_ms: i64) !void {	564	fn reconnect(self: *Conn, deadline_ms: i64) !void {
566	const q = &self.link.quic;	565	const q = &self.link.quic;
567	const cl = try quic_client.Client.connect(self.alloc, q.addr, q.key, q.idle_ms);	566	const cl = try quic.Client.connect(self.alloc, q.addr, q.key, q.idle_ms);
568	errdefer cl.deinit();	567	errdefer cl.deinit();
569	try waitReady(cl, deadline_ms);	568	try waitReady(cl, deadline_ms);
570	q.cl.deinit();	569	q.cl.deinit();
@@ -611,13 +610,13 @@ test "graceMs: flat over a socket, RTT-derived over QUIC, and capped" {
611		610
612	/// Drive a fresh connection until it can carry bytes, or give up.	611	/// Drive a fresh connection until it can carry bytes, or give up.
613	///	612	///
614	/// A refused port ends this early — quic_client turns the ICMP unreachable	613	/// A refused port ends this early — quic.Client turns the ICMP unreachable
615	/// into `dead` — so the common mistake (no daemon on that port) costs	614	/// into `dead` — so the common mistake (no daemon on that port) costs
616	/// milliseconds. A blackholed one produces no error at all, and there the	615	/// milliseconds. A blackholed one produces no error at all, and there the
617	/// deadline is the only thing that ends the wait; even an unbounded one	616	/// deadline is the only thing that ends the wait; even an unbounded one
618	/// (`--timeout 0`) terminates, because the connection's own idle timeout	617	/// (`--timeout 0`) terminates, because the connection's own idle timeout
619	/// kills it after `idle_ms`.	618	/// kills it after `idle_ms`.
620	fn waitReady(cl: *quic_client.Client, deadline_ms: i64) !void {	619	fn waitReady(cl: *quic.Client, deadline_ms: i64) !void {
621	while (true) {	620	while (true) {
622	cl.pump();	621	cl.pump();
623	if (cl.isReady()) return;	622	if (cl.isReady()) return;
@@ -635,11 +634,11 @@ fn waitReady(cl: *quic_client.Client, deadline_ms: i64) !void {
635	test "reconnect: redials the same coordinates, and a dead port is a fast no" {	634	test "reconnect: redials the same coordinates, and a dead port is a fast no" {
636	const alloc = std.testing.allocator;	635	const alloc = std.testing.allocator;
637	// 127.0.0.1:1, where nothing listens: the refusal is REAL — an ICMP	636	// 127.0.0.1:1, where nothing listens: the refusal is REAL — an ICMP
638	// unreachable comes back and quic_client acts on it — which is what	637	// unreachable comes back and the quic client acts on it — which is what
639	// lets this exercise the whole redial path (dial, handshake wait,	638	// lets this exercise the whole redial path (dial, handshake wait,
640	// verdict) in a couple of loopback round trips instead of a timeout.	639	// verdict) in a couple of loopback round trips instead of a timeout.
641	const addr = try std.net.Address.parseIp("127.0.0.1", 1);	640	const addr = try std.net.Address.parseIp("127.0.0.1", 1);
642	const key: quic_client.Key = .{ .bytes = [_]u8{7} ** quic_client.key_len };	641	const key: quic.Key = .{ .bytes = [_]u8{7} ** quic.key_len };
643		642
644	// The dial that stands in for the connection this client had before	643	// The dial that stands in for the connection this client had before
645	// the network went away. It dies for the same reason the redial will,	644	// the network went away. It dies for the same reason the redial will,
@@ -648,7 +647,7 @@ test "reconnect: redials the same coordinates, and a dead port is a fast no" {
648	const deadline = std.time.milliTimestamp() + 2_000;	647	const deadline = std.time.milliTimestamp() + 2_000;
649	var conn = Conn{	648	var conn = Conn{
650	.link = .{ .quic = .{	649	.link = .{ .quic = .{
651	.cl = try quic_client.Client.connect(alloc, addr, key, 1_000),	650	.cl = try quic.Client.connect(alloc, addr, key, 1_000),
652	.addr = addr,	651	.addr = addr,
653	.key = key,	652	.key = key,
654	.idle_ms = 1_000,	653	.idle_ms = 1_000,
@@ -1211,12 +1210,12 @@ fn openQuicConn(
1211	p,	1210	p,
1212	) },	1211	) },
1213	};	1212	};
1214	const key = quic_client.Key.load(key_path) catch \|e\| {	1213	const key = quic.Key.load(key_path) catch \|e\| {
1215	// The daemon's words for a key the daemon would also refuse —	1214	// The daemon's words for a key the daemon would also refuse —
1216	// including the group/other-readable refusal, which this binary	1215	// including the group/other-readable refusal, which this binary
1217	// gets for free by loading the key the same way.	1216	// gets for free by loading the key the same way.
1218	var buf: [quic_client.key_refusal_len]u8 = undefined;	1217	var buf: [quic.key_refusal_len]u8 = undefined;
1219	return .{ .exit = fail("quic: unusable key", quic_client.keyRefusalBody(&buf, e, key_path)) };	1218	return .{ .exit = fail("quic: unusable key", quic.keyRefusalBody(&buf, e, key_path)) };
1220	};	1219	};
1221	const addr = quic.parseAddr(alloc, host_port) catch \|e\| {	1220	const addr = quic.parseAddr(alloc, host_port) catch \|e\| {
1222	var buf: [512]u8 = undefined;	1221	var buf: [512]u8 = undefined;
@@ -1224,7 +1223,7 @@ fn openQuicConn(
1224	@errorName(e);	1223	@errorName(e);
1225	return .{ .exit = fail("quic: cannot read HOST:PORT", detail) };	1224	return .{ .exit = fail("quic: cannot read HOST:PORT", detail) };
1226	};	1225	};
1227	const conn = Conn.openQuic(alloc, addr, key, quic_client.default_idle_ms, deadline) catch \|e\| {	1226	const conn = Conn.openQuic(alloc, addr, key, quic.default_idle_ms, deadline) catch \|e\| {
1228	var buf: [512]u8 = undefined;	1227	var buf: [512]u8 = undefined;
1229	const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ host_port, @errorName(e) }) catch	1228	const detail = std.fmt.bufPrint(&buf, "{s}: {s}", .{ host_port, @errorName(e) }) catch
1230	@errorName(e);	1229	@errorName(e);

src/client/client.zig

Old		New
@@ -20,7 +20,6 @@ const Replica = @import("replica").Replica;
20	const keymap = @import("keymap");	20	const keymap = @import("keymap");
21	const proto = @import("protocol");	21	const proto = @import("protocol");
22	const TmpDir = @import("testtmp").TmpDir;	22	const TmpDir = @import("testtmp").TmpDir;
23	const quic_client = @import("quic_client");
24	const quic = @import("quic");	23	const quic = @import("quic");
25	const handoff = @import("handoff");	24	const handoff = @import("handoff");
26		25
@@ -161,8 +160,8 @@ pub fn lostMsg(target: Target, session_epoch: u64) []const u8 {
161		160
162	/// The client's name for the shared default; see `quic.default_idle_ms`	161	/// The client's name for the shared default; see `quic.default_idle_ms`
163	/// for what the number means and why it lives there.	162	/// for what the number means and why it lives there.
164	pub const quic_idle_ms_default: u32 = quic_client.default_idle_ms;	163	pub const quic_idle_ms_default: u32 = quic.default_idle_ms;
165	pub const IdleMs = quic_client.IdleMs;	164	pub const IdleMs = quic.IdleMs;
166		165
167	/// What a QUIC attach was asked for: where to dial and what key to prove	166	/// What a QUIC attach was asked for: where to dial and what key to prove
168	/// ourselves with.	167	/// ourselves with.
@@ -257,7 +256,7 @@ pub const Link = union(enum) {
257	/// The connection that IS the transport. `conn` holds the UDP socket in	256	/// The connection that IS the transport. `conn` holds the UDP socket in
258	/// `.r` so the poll path needs no special case, and `.w` is -1 because	257	/// `.r` so the poll path needs no special case, and `.w` is -1 because
259	/// there is nothing to write(2) to — bytes go through the stream layer.	258	/// there is nothing to write(2) to — bytes go through the stream layer.
260	quic: *quic_client.Client,	259	quic: *quic.Client,
261	};	260	};
262		261
263	/// One live connection to a daemon, however it was reached. The point of the	262	/// One live connection to a daemon, however it was reached. The point of the
@@ -318,18 +317,18 @@ pub const Transport = struct {
318	}	317	}
319		318
320	/// `connect` only creates state, so the handshake wait belongs here:	319	/// `connect` only creates state, so the handshake wait belongs here:
321	/// "open succeeded" means the same on every transport. `budget_ms` is	320	/// "open succeeded" means one thing everywhere. `budget_ms` is the
322	/// the attach budget, not `idle_ms`.	321	/// attach budget, not `idle_ms`.
323	fn quicTransport(	322	fn quicTransport(
324	alloc: std.mem.Allocator,	323	alloc: std.mem.Allocator,
325	addr: std.net.Address,	324	addr: std.net.Address,
326	key: quic_client.Key,	325	key: quic.Key,
327	idle_ms: u32,	326	idle_ms: u32,
328	budget_ms: u32,	327	budget_ms: u32,
329	carry: ?*std.ArrayList(u8),	328	carry: ?*std.ArrayList(u8),
330	abort_fd: std.posix.fd_t,	329	abort_fd: std.posix.fd_t,
331	) !Transport {	330	) !Transport {
332	const cl = try quic_client.Client.connect(alloc, addr, key, idle_ms);	331	const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
333	errdefer cl.deinit();	332	errdefer cl.deinit();
334	try waitReady(cl, budget_ms, alloc, carry, abort_fd);	333	try waitReady(cl, budget_ms, alloc, carry, abort_fd);
335	return .{	334	return .{
@@ -357,7 +356,7 @@ pub const Transport = struct {
357	// either of the two links below and owns the choice itself.	356	// either of the two links below and owns the choice itself.
358	.hand => \|h\| return openHandoff(alloc, h, carry, abort_fd),	357	.hand => \|h\| return openHandoff(alloc, h, carry, abort_fd),
359	.quic => \|q\| {	358	.quic => \|q\| {
360	const key = try quic_client.Key.load(q.key_path);	359	const key = try quic.Key.load(q.key_path);
361	const addr = try quic.parseAddr(alloc, q.host_port);	360	const addr = try quic.parseAddr(alloc, q.host_port);
362	return quicTransport(alloc, addr, key, q.idle_ms, q.deadline_ms, carry, abort_fd);	361	return quicTransport(alloc, addr, key, q.idle_ms, q.deadline_ms, carry, abort_fd);
363	},	362	},
@@ -400,7 +399,7 @@ pub const Transport = struct {
400	// which is authoritative. What that costs depends on	399	// which is authoritative. What that costs depends on
401	// how the coordinates are dead: a resolve failure and	400	// how the coordinates are dead: a resolve failure and
402	// a REFUSED port are both instant — the ICMP comes	401	// a REFUSED port are both instant — the ICMP comes
403	// back and quic_client acts on it — while anything	402	// back and the quic client acts on it — while anything
404	// SILENT (blackholed UDP, a listener holding another	403	// SILENT (blackholed UDP, a listener holding another
405	// key) spends the whole deadline first. See	404	// key) spends the whole deadline first. See
406	// handoff.deadline_ms for both numbers.	405	// handoff.deadline_ms for both numbers.
@@ -582,7 +581,7 @@ pub const Transport = struct {
582	abort_fd: std.posix.fd_t,	581	abort_fd: std.posix.fd_t,
583	) !Transport {	582	) !Transport {
584	const addr = try quic.resolveHost(alloc, handoff.dialHost(h.host), ep.port);	583	const addr = try quic.resolveHost(alloc, handoff.dialHost(h.host), ep.port);
585	const key = quic_client.Key{ .bytes = ep.key };	584	const key = quic.Key{ .bytes = ep.key };
586	return quicTransport(alloc, addr, key, h.idle_ms, h.deadline_ms, carry, abort_fd);	585	return quicTransport(alloc, addr, key, h.idle_ms, h.deadline_ms, carry, abort_fd);
587	}	586	}
588		587
@@ -725,7 +724,7 @@ pub fn connectAgent(path: []const u8) ?std.posix.fd_t {
725	/// Bounded by the attach budget, not the idle timeout: the time given a	724	/// Bounded by the attach budget, not the idle timeout: the time given a
726	/// handshake is not the time given a quiet session.	725	/// handshake is not the time given a quiet session.
727	fn waitReady(	726	fn waitReady(
728	cl: *quic_client.Client,	727	cl: *quic.Client,
729	budget_ms: u32,	728	budget_ms: u32,
730	alloc: std.mem.Allocator,	729	alloc: std.mem.Allocator,
731	carry: ?*std.ArrayList(u8),	730	carry: ?*std.ArrayList(u8),
@@ -924,11 +923,11 @@ pub fn openFailure(buf: []u8, target: Target, err: anyerror) OpenFailure {
924	error.KeyFilePermissive,	923	error.KeyFilePermissive,
925	error.KeyFileMalformed,	924	error.KeyFileMalformed,
926	=> blk: {	925	=> blk: {
927	var body: [quic_client.key_refusal_len]u8 = undefined;	926	var body: [quic.key_refusal_len]u8 = undefined;
928	break :blk failedMsg(	927	break :blk failedMsg(
929	buf,	928	buf,
930	"mux: {s}\n",	929	"mux: {s}\n",
931	.{quic_client.keyRefusalBody(&body, err, q.key_path)},	930	.{quic.keyRefusalBody(&body, err, q.key_path)},
932	);	931	);
933	},	932	},
934	error.MalformedAddress, error.UnknownHostName => failedMsg(	933	error.MalformedAddress, error.UnknownHostName => failedMsg(
@@ -1533,7 +1532,7 @@ test "handoff: endpoint-none rides the open pipe with no deadline paid" {
1533	test "handoff: dead coordinates are a fast no, and the pipe is the fallback" {	1532	test "handoff: dead coordinates are a fast no, and the pipe is the fallback" {
1534	// A well-formed announce naming 127.0.0.1:1, where nothing listens.	1533	// A well-formed announce naming 127.0.0.1:1, where nothing listens.
1535	// The refusal is REAL — an ICMP unreachable comes back — and both	1534	// The refusal is REAL — an ICMP unreachable comes back — and both
1536	// socket paths in quic_client act on it, so this dial dies	1535	// socket paths in the quic client act on it, so this dial dies
1537	// in about one loopback round trip instead of running deadline_ms out.	1536	// in about one loopback round trip instead of running deadline_ms out.
1538	// The upper bound below is what proves that: it sits far under the	1537	// The upper bound below is what proves that: it sits far under the
1539	// budget, so an implementation that swallowed the refusal again (the	1538	// budget, so an implementation that swallowed the refusal again (the

src/client/quic_client.zig

Old		New
@@ -1,419 +0,0 @@
1	//! mux's QUIC client transport: one UDP socket, one connection, one
2	//! bidirectional stream of opaque bytes.
3	//!
4	//! The mirror of `quic_server.zig`: it knows NOTHING about the frame
5	//! protocol it carries, which is why there is no `proto` import here.
6	//! Everything both ends must agree on is imported from quic.zig, so there
7	//! is one copy to drift.
8	//!
9	//! Three ngtcp2 constraints:
10	//!
11	//! 1. ngtcp2 does not copy stream payload. It keeps the vector it is
12	//! handed and re-reads those bytes to retransmit, so outbound bytes
13	//! must not move or be reused until the peer acknowledges them.
14	//! 2. A blocked stream is not a dead one. NGTCP2_ERR_STREAM_DATA_BLOCKED
15	//! is a documented return; abandoning the egress loop on it also
16	//! abandons the ACKs that would reopen the window.
17	//! 3. Both flow-control windows get extended on consume. Extending
18	//! only the stream's leaves the connection window to close, and the
19	//! DAEMON stalls a few hundred kilobytes in — a freeze with no error
20	//! on either side.
21	const std = @import("std");
22	const quic = @import("quic");
23
24	/// quic.zig's cImport, and deliberately not one of our own: two @cImport
25	/// blocks over the same headers are two distinct type universes, and the
26	/// egress ring hands `ngtcp2_vec`s across this boundary.
27	const c = quic.c;
28
29	pub const Key = quic.Key;
30	pub const key_len = quic.key_len;
31	pub const default_port = quic.default_port;
32	pub const default_idle_ms = quic.default_idle_ms;
33	pub const IdleMs = quic.IdleMs;
34	/// The refusal words, re-exported for the same reason `Key` is: client.zig
35	/// prints the daemon's sentences for a key the daemon would also refuse,
36	/// and it reaches the vocabulary module through this one.
37	pub const key_refusal_len = quic.key_refusal_len;
38	pub const keyRefusalBody = quic.keyRefusalBody;
39
40	/// wolfSSL's PSK callback carries no user pointer, so the key has to be
41	/// reachable without one. A client process runs one connection at a time,
42	/// which makes this correct rather than merely convenient.
43	var g_key: ?Key = null;
44
45	fn pskClientCb(
46	_: ?*c.WOLFSSL,
47	_: [*c]const u8,
48	identity: [*c]u8,
49	id_max: c_uint,
50	key_out: [*c]u8,
51	key_max: c_uint,
52	ciphersuite: [c][c]const u8,
53	) callconv(.c) c_uint {
54	const k = g_key orelse return 0;
55	return quic.answerPsk(k, identity, id_max, key_out, key_max, ciphersuite);
56	}
57
58	fn getConnCb(ref: [c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?c.ngtcp2_conn {
59	const self: Client = @ptrCast(@alignCast(ref..user_data));
60	return self.conn;
61	}
62
63	fn handshakeCompletedCb(_: ?c.ngtcp2_conn, ud: ?anyopaque) callconv(.c) c_int {
64	const self: *Client = @ptrCast(@alignCast(ud.?));
65	self.handshake_done = true;
66	return 0;
67	}
68
69	/// The peer granted stream credit: open the one stream this transport uses.
70	/// It cannot be opened before the handshake, which is why this is a callback
71	/// rather than a line in `connect`.
72	fn extendStreamsCb(conn: ?c.ngtcp2_conn, _: u64, ud: ?anyopaque) callconv(.c) c_int {
73	const self: *Client = @ptrCast(@alignCast(ud.?));
74	if (self.stream_id == -1) {
75	var sid: i64 = -1;
76	if (c.ngtcp2_conn_open_bidi_stream(conn, &sid, null) == 0) self.stream_id = sid;
77	}
78	return 0;
79	}
80
81	fn recvStreamDataCb(
82	conn: ?*c.ngtcp2_conn,
83	_: u32,
84	stream_id: i64,
85	_: u64,
86	data: [*c]const u8,
87	datalen: usize,
88	ud: ?*anyopaque,
89	_: ?*anyopaque,
90	) callconv(.c) c_int {
91	const self: *Client = @ptrCast(@alignCast(ud.?));
92
93	// BOTH windows, and the connection-level one is the half that is easy to
94	// forget: extend only the stream and the daemon stops sending a few
95	// hundred kilobytes into a session — a scrollback fetch, a big
96	// snapshot — with no error anywhere. It presents as a freeze.
97	_ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen);
98	_ = c.ngtcp2_conn_extend_max_offset(conn, datalen);
99
100	if (datalen > 0) {
101	self.in.appendSlice(self.alloc, data[0..datalen]) catch {
102	self.dead = true;
103	return 0;
104	};
105	}
106	return 0;
107	}
108
109	fn ackedStreamDataCb(
110	_: ?*c.ngtcp2_conn,
111	_: i64,
112	_: u64,
113	datalen: u64,
114	ud: ?*anyopaque,
115	_: ?*anyopaque,
116	) callconv(.c) c_int {
117	const self: *Client = @ptrCast(@alignCast(ud.?));
118	self.out.ack(@intCast(datalen));
119	return 0;
120	}
121
122	/// One connection to a daemon's QUIC listener.
123	pub const Client = struct {
124	alloc: std.mem.Allocator,
125	fd: std.posix.fd_t,
126	ssl_ctx: ?*c.WOLFSSL_CTX = null,
127	ssl: ?*c.WOLFSSL = null,
128	conn: ?*c.ngtcp2_conn = null,
129	conn_ref: c.ngtcp2_crypto_conn_ref = undefined,
130	remote: std.posix.sockaddr.storage = undefined,
131	remote_len: std.posix.socklen_t = 0,
132	local: std.posix.sockaddr.storage = undefined,
133	local_len: std.posix.socklen_t = 0,
134	stream_id: i64 = -1,
135	handshake_done: bool = false,
136	/// Set once this connection can never carry another byte: an idle
137	/// timeout, a protocol error, a peer that stopped answering. The caller
138	/// reads it as "the transport is gone" and reconnects.
139	dead: bool = false,
140	/// Depth of nesting inside ngtcp2, mirroring the listener's. This side
141	/// has no live re-entrancy today — its callbacks only set flags and
142	/// append to `in`, none of them reaches `send` — so `send` may still
143	/// drain, which is what keeps a keystroke from waiting for the next poll
144	/// cycle. The counter and the assert in `drain` are what make that a
145	/// checked invariant rather than a property someone has to re-verify
146	/// every time a callback grows a line.
147	ngtcp2_depth: u8 = 0,
148	/// Outbound bytes, in the ring that does not move them until they are
149	/// acknowledged. See the module comment for why that matters.
150	out: quic.Egress,
151	/// Stream bytes the caller has not consumed yet.
152	in: std.ArrayList(u8) = .empty,
153
154	pub fn connect(
155	alloc: std.mem.Allocator,
156	addr: std.net.Address,
157	key: Key,
158	idle_ms: u32,
159	) !*Client {
160	const fd = try std.posix.socket(
161	addr.any.family,
162	std.posix.SOCK.DGRAM \| std.posix.SOCK.NONBLOCK \| std.posix.SOCK.CLOEXEC,
163	0,
164	);
165	errdefer std.posix.close(fd);
166
167	const self = try alloc.create(Client);
168	errdefer alloc.destroy(self);
169	const ring = try alloc.alloc(u8, quic.egress_cap);
170	errdefer alloc.free(ring);
171
172	self.* = .{
173	.alloc = alloc,
174	.fd = fd,
175	.out = .{ .buf = ring },
176	.remote_len = addr.getOsSockLen(),
177	.local_len = @sizeOf(std.posix.sockaddr.storage),
178	};
179	@memcpy(
180	std.mem.asBytes(&self.remote)[0..self.remote_len],
181	std.mem.asBytes(&addr.any)[0..self.remote_len],
182	);
183	// Connected UDP: the kernel filters out anything from another
184	// address, which is one fewer thing this code has to check.
185	try std.posix.connect(fd, &addr.any, self.remote_len);
186	try std.posix.getsockname(fd, @ptrCast(&self.local), &self.local_len);
187
188	g_key = key;
189	try self.startTls();
190	try self.startConn(idle_ms);
191	self.drain();
192	return self;
193	}
194
195	fn startTls(self: *Client) !void {
196	// Set before clientTls, which hands wolfSSL a pointer to it.
197	self.conn_ref = .{ .get_conn = getConnCb, .user_data = self };
198	const tls = try quic.clientTls(pskClientCb, &self.conn_ref);
199	self.ssl_ctx = tls.ctx;
200	self.ssl = tls.ssl;
201	}
202
203	fn startConn(self: *Client, idle_ms: u32) !void {
204	var dcid: c.ngtcp2_cid = undefined;
205	dcid.datalen = 16;
206	std.crypto.random.bytes(dcid.data[0..16]);
207	var scid: c.ngtcp2_cid = undefined;
208	scid.datalen = 8;
209	std.crypto.random.bytes(scid.data[0..8]);
210
211	var cbs = quic.clientCallbacks();
212	cbs.handshake_completed = handshakeCompletedCb;
213	cbs.extend_max_local_streams_bidi = extendStreamsCb;
214	cbs.recv_stream_data = recvStreamDataCb;
215	cbs.acked_stream_data_offset = ackedStreamDataCb;
216
217	var settings: c.ngtcp2_settings = undefined;
218	c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings);
219	settings.initial_ts = quic.timestampNs();
220
221	var params = quic.clientParams();
222	// Tunable for the same reason as the daemon's: the reconnect loop
223	// has to see a dead transport on a schedule a test can wait for.
224	params.max_idle_timeout = @as(u64, idle_ms) * 1_000_000;
225
226	var path = quic.pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
227	var conn: ?*c.ngtcp2_conn = null;
228	if (c.ngtcp2_conn_client_new_versioned(
229	&conn,
230	&dcid,
231	&scid,
232	&path,
233	c.NGTCP2_PROTO_VER_V1,
234	c.NGTCP2_CALLBACKS_VERSION,
235	&cbs,
236	c.NGTCP2_SETTINGS_VERSION,
237	&settings,
238	c.NGTCP2_TRANSPORT_PARAMS_VERSION,
239	&params,
240	null,
241	self,
242	) != 0) return error.ConnInit;
243	self.conn = conn;
244	c.ngtcp2_conn_set_tls_native_handle(conn, self.ssl);
245	// Silence is not death: a terminal nobody is typing into must not be
246	// dropped at the idle timeout. A third of it, matching the daemon.
247	c.ngtcp2_conn_set_keep_alive_timeout(conn, quic.keepAliveNs(idle_ms));
248	}
249
250	pub fn deinit(self: *Client) void {
251	self.in.deinit(self.alloc);
252	// ngtcp2 before the egress ring: it holds vectors into that ring for
253	// anything unacknowledged, so freeing the ring first leaves it
254	// reading memory that is already gone for the length of its own
255	// teardown. Same ordering as the listener's Conn.deinit.
256	if (self.conn) \|cn\| c.ngtcp2_conn_del(cn);
257	self.out.deinit(self.alloc);
258	if (self.ssl) \|s\| c.wolfSSL_free(s);
259	if (self.ssl_ctx) \|x\| c.wolfSSL_CTX_free(x);
260	std.posix.close(self.fd);
261	self.alloc.destroy(self);
262	}
263
264	/// Readable means a datagram arrived, not that a frame is waiting.
265	pub fn pollFd(self: *const Client) std.posix.fd_t {
266	return self.fd;
267	}
268
269	/// Handshake finished AND the stream open: a handshake without a stream
270	/// silently holds everything in the ring.
271	pub fn isReady(self: *const Client) bool {
272	return self.handshake_done and self.stream_id != -1 and !self.dead;
273	}
274
275	/// Milliseconds until ngtcp2 next wants servicing, capped. Feeds the
276	/// client's existing poll timeout — no timer fd, same as the daemon.
277	pub fn timeoutMs(self: *Client, cap_ms: i32) i32 {
278	// A negative cap means poll forever; the @intCast below would panic
279	// on it instead of honouring it.
280	if (cap_ms < 0) return cap_ms;
281	const conn = self.conn orelse return cap_ms;
282	const expiry = c.ngtcp2_conn_get_expiry(conn);
283	if (expiry == std.math.maxInt(u64)) return cap_ms;
284	const now = quic.timestampNs();
285	if (expiry <= now) return 0;
286	return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000));
287	}
288
289	/// One service pass: read, run due timers, push ready egress. Safe to call
290	/// at any time.
291	pub fn pump(self: *Client) void {
292	if (self.dead) return;
293	self.readable();
294	self.tick();
295	self.drain();
296	}
297
298	/// ECONNREFUSED is fatal and reaches whichever syscall runs first, so
299	/// both paths must act.
300	fn sendRecvFailed(
301	self: *Client,
302	err: (std.posix.RecvFromError \|\| std.posix.SendError),
303	) void {
304	switch (err) {
305	error.ConnectionRefused => self.dead = true,
306	else => {},
307	}
308	}
309
310	fn readable(self: *Client) void {
311	var buf: [65536]u8 = undefined;
312	while (true) {
313	const n = std.posix.recv(self.fd, &buf, 0) catch \|err\| switch (err) {
314	error.WouldBlock => return,
315	else => {
316	self.sendRecvFailed(err);
317	return;
318	},
319	};
320	if (n == 0) continue;
321	const conn = self.conn orelse return;
322	var path = quic.pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
323	var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
324	const rv = blk: {
325	self.ngtcp2_depth += 1;
326	defer self.ngtcp2_depth -= 1;
327	break :blk c.ngtcp2_conn_read_pkt(conn, &path, &pi, &buf, n, quic.timestampNs());
328	};
329	if (rv != 0) {
330	self.dead = true;
331	return;
332	}
333	}
334	}
335
336	fn tick(self: *Client) void {
337	const conn = self.conn orelse return;
338	const now = quic.timestampNs();
339	if (c.ngtcp2_conn_get_expiry(conn) > now) return;
340	// An idle timeout arrives here, which is how the reconnect loop
341	// learns the daemon stopped answering.
342	if (c.ngtcp2_conn_handle_expiry(conn, now) != 0) self.dead = true;
343	}
344
345	fn drain(self: *Client) void {
346	// The listener learned this the hard way: writing to a connection
347	// while ngtcp2 is reading a packet on it corrupts loss detection and
348	// can abort outright. Enforced here so it stays true of this side.
349	std.debug.assert(self.ngtcp2_depth == 0);
350	const conn = self.conn orelse return;
351	var buf: [quic.max_udp]u8 = undefined;
352	var stream_blocked = false;
353	while (true) {
354	var ps: c.ngtcp2_path_storage = undefined;
355	c.ngtcp2_path_storage_zero(&ps);
356	var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
357	var wrote: c.ngtcp2_ssize = 0;
358
359	var vecs: [2]c.ngtcp2_vec = undefined;
360	var vcnt: usize = 0;
361	var sid: i64 = -1;
362	if (!stream_blocked and self.stream_id != -1) {
363	vcnt = self.out.vecs(&vecs);
364	if (vcnt > 0) sid = self.stream_id;
365	}
366
367	const n = c.ngtcp2_conn_writev_stream_versioned(
368	conn,
369	&ps.path,
370	c.NGTCP2_PKT_INFO_VERSION,
371	&pi,
372	&buf,
373	buf.len,
374	&wrote,
375	0,
376	sid,
377	if (vcnt > 0) &vecs else null,
378	vcnt,
379	quic.timestampNs(),
380	);
381	if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) {
382	// Documented, not fatal. Retry the same iteration carrying
383	// no stream data so ACKs and keepalives still leave: they
384	// are how the peer's window reopens, and a client that
385	// stopped here would stall exactly as the daemon did.
386	stream_blocked = true;
387	continue;
388	}
389	switch (quic.accountWrite(&self.out, wrote, n)) {
390	.stop => {
391	self.dead = true;
392	return;
393	},
394	.brk => return,
395	.cont => {},
396	}
397	_ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch \|err\| {
398	self.sendRecvFailed(err);
399	return;
400	};
401	}
402	}
403
404	/// A short return is the caller's signal to keep the rest and offer it
405	/// again: the ring is bounded, so the backlog belongs with somebody who
406	/// can see how big it is.
407	pub fn send(self: *Client, bytes: []const u8) usize {
408	if (self.dead or self.stream_id == -1) return 0;
409	const n = self.out.push(bytes);
410	if (n > 0) self.drain();
411	return n;
412	}
413	};
414
415	// Plain, not recursive: this module reaches the QUIC stack's @cImport, and
416	// a recursive walk would force-analyze the entire wolfSSL/ngtcp2 namespace.
417	test {
418	std.testing.refAllDecls(@This());
419	}

src/quic.zig

@@ @@ -1,10 +1,12 @@ @@
+//! The QUIC vocabulary both ends share: the C import, the pre-shared key,
+//! the wire constants, and the egress ring's lifetime discipline.
+//!
+//! Nothing here knows what a LISTENER is: quic_server.zig owns that, and
+//! imports this — which is what makes "the two ends agree" a fact about one
+//! file rather than a convention two files are trusted to keep. The dialling
+//! end lives here, below the shared vocabulary, because both the client and
+//! the listener's own tests need a real peer and the folder rule forbids the
+//! server naming a client module.
+//!
+//! THE OWNERSHIP RULE, because it is load-bearing rather than tidy: this
+//! file holds the one and only @cImport of the QUIC stack. Two blocks over
@@ @@ -700,11 +702,11 @@ pub fn getNewCidCb( @@
+// ---------------------------------------------------------------------------
+// The client side of the handshake, kept separate from `Client` below
+// because the listener's tests once dialled with a second copy of it: a
+// drift between the two failed the handshake with nothing to read but a
+// TLS alert. `Client` sets its own key, stream handlers and idle timeout
+// on what these hand back.
+// ---------------------------------------------------------------------------
+/// Identity "mux", the key's bytes, our one ciphersuite: none is a choice.
@@ @@ -801,6 +803,413 @@ pub fn pathFrom( @@
+    };
+// ---------------------------------------------------------------------------
+// The client transport: one UDP socket, one connection, one bidirectional
+// stream of opaque bytes. It knows NOTHING about the frame protocol it
+// carries, which is why `protocol` is not imported here.
+//
+// Three ngtcp2 constraints shape it, and the listener across the seam obeys
+// the same three:
+//
+//   1. **ngtcp2 does not copy stream payload.** It keeps the vector it is
+//      handed and re-reads those bytes to retransmit, so outbound bytes
+//      must not move or be reused until the peer acknowledges them.
+//   2. **A blocked stream is not a dead one.** NGTCP2_ERR_STREAM_DATA_BLOCKED
+//      is a documented return; abandoning the egress loop on it also
+//      abandons the ACKs that would reopen the window.
+//   3. **Both flow-control windows get extended on consume.** Extending
+//      only the stream's leaves the connection window to close, and the
+//      DAEMON stalls a few hundred kilobytes in — a freeze with no error
+//      on either side.
+// ---------------------------------------------------------------------------
+/// wolfSSL's PSK callback carries no user pointer, so the key has to be
+/// reachable without one. A client process runs one connection at a time,
+/// which makes this correct rather than merely convenient.
+var g_key: ?Key = null;
+fn pskClientCb(
+    _: ?*c.WOLFSSL,
+    _: [*c]const u8,
+    identity: [*c]u8,
+    id_max: c_uint,
+    key_out: [*c]u8,
+    key_max: c_uint,
+    ciphersuite: [*c][*c]const u8,
+) callconv(.c) c_uint {
+    const k = g_key orelse return 0;
+    return answerPsk(k, identity, id_max, key_out, key_max, ciphersuite);
+}
+fn getConnCb(ref: [*c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?*c.ngtcp2_conn {
+    const self: *Client = @ptrCast(@alignCast(ref.*.user_data));
+    return self.conn;
+}
+fn handshakeCompletedCb(_: ?*c.ngtcp2_conn, ud: ?*anyopaque) callconv(.c) c_int {
+    const self: *Client = @ptrCast(@alignCast(ud.?));
+    self.handshake_done = true;
+    return 0;
+}
+/// The peer granted stream credit: open the one stream this transport uses.
+/// It cannot be opened before the handshake, which is why this is a callback
+/// rather than a line in `connect`.
+fn extendStreamsCb(conn: ?*c.ngtcp2_conn, _: u64, ud: ?*anyopaque) callconv(.c) c_int {
+    const self: *Client = @ptrCast(@alignCast(ud.?));
+    if (self.stream_id == -1) {
+        var sid: i64 = -1;
+        if (c.ngtcp2_conn_open_bidi_stream(conn, &sid, null) == 0) self.stream_id = sid;
+    }
+    return 0;
+}
+fn recvStreamDataCb(
+    conn: ?*c.ngtcp2_conn,
+    _: u32,
+    stream_id: i64,
+    _: u64,
+    data: [*c]const u8,
+    datalen: usize,
+    ud: ?*anyopaque,
+    _: ?*anyopaque,
+) callconv(.c) c_int {
+    const self: *Client = @ptrCast(@alignCast(ud.?));
+    // BOTH windows, and the connection-level one is the half that is easy to
+    // forget: extend only the stream and the daemon stops sending a few
+    // hundred kilobytes into a session — a scrollback fetch, a big
+    // snapshot — with no error anywhere. It presents as a freeze.
+    if (self.extend_windows) {
+        _ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen);
+        _ = c.ngtcp2_conn_extend_max_offset(conn, datalen);
+    }
+    if (datalen > 0) {
+        self.in.appendSlice(self.alloc, data[0..datalen]) catch {
+            self.dead = true;
+            return 0;
+        };
+    }
+    return 0;
+}
+fn ackedStreamDataCb(
+    _: ?*c.ngtcp2_conn,
+    _: i64,
+    _: u64,
+    datalen: u64,
+    ud: ?*anyopaque,
+    _: ?*anyopaque,
+) callconv(.c) c_int {
+    const self: *Client = @ptrCast(@alignCast(ud.?));
+    self.out.ack(@intCast(datalen));
+    return 0;
+}
+/// One connection to a daemon's QUIC listener.
+pub const Client = struct {
+    alloc: std.mem.Allocator,
+    fd: std.posix.fd_t,
+    ssl_ctx: ?*c.WOLFSSL_CTX = null,
+    ssl: ?*c.WOLFSSL = null,
+    conn: ?*c.ngtcp2_conn = null,
+    conn_ref: c.ngtcp2_crypto_conn_ref = undefined,
+    remote: std.posix.sockaddr.storage = undefined,
+    remote_len: std.posix.socklen_t = 0,
+    local: std.posix.sockaddr.storage = undefined,
+    local_len: std.posix.socklen_t = 0,
+    stream_id: i64 = -1,
+    handshake_done: bool = false,
+    /// Set once this connection can never carry another byte: an idle
+    /// timeout, a protocol error, a peer that stopped answering. The caller
+    /// reads it as "the transport is gone" and reconnects.
+    dead: bool = false,
+    /// Depth of nesting inside ngtcp2, mirroring the listener's. This side
+    /// has no live re-entrancy today — its callbacks only set flags and
+    /// append to `in`, none of them reaches `send` — so `send` may still
+    /// drain, which is what keeps a keystroke from waiting for the next poll
+    /// cycle. The counter and the assert in `drain` are what make that a
+    /// checked invariant rather than a property someone has to re-verify
+    /// every time a callback grows a line.
+    ngtcp2_depth: u8 = 0,
+    /// Outbound bytes, in the ring that does not move them until they are
+    /// acknowledged. See the module comment for why that matters.
+    out: Egress,
+    /// Stream bytes the caller has not consumed yet.
+    in: std.ArrayList(u8) = .empty,
+    /// Whether arriving bytes buy the peer more window. A shipping client
+    /// always extends (constraint 3 above); the listener's own tests turn
+    /// it off to reach the blocked-stream branch, which is a documented
+    /// ngtcp2 return rather than a failure and has no other way in.
+    extend_windows: bool = true,
+    pub fn connect(
+        alloc: std.mem.Allocator,
+        addr: std.net.Address,
+        key: Key,
+        idle_ms: u32,
+    ) !*Client {
+        const fd = try std.posix.socket(
+            addr.any.family,
+            std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
+,
+        );
+        errdefer std.posix.close(fd);
+        const self = try alloc.create(Client);
+        errdefer alloc.destroy(self);
+        const ring = try alloc.alloc(u8, egress_cap);
+        errdefer alloc.free(ring);
+        self.* = .{
+            .alloc = alloc,
+            .fd = fd,
+            .out = .{ .buf = ring },
+            .remote_len = addr.getOsSockLen(),
+            .local_len = @sizeOf(std.posix.sockaddr.storage),
+        };
+        @memcpy(
+            std.mem.asBytes(&self.remote)[0..self.remote_len],
+            std.mem.asBytes(&addr.any)[0..self.remote_len],
+        );
+        // Connected UDP: the kernel filters out anything from another
+        // address, which is one fewer thing this code has to check.
+        try std.posix.connect(fd, &addr.any, self.remote_len);
+        try std.posix.getsockname(fd, @ptrCast(&self.local), &self.local_len);
+        g_key = key;
+        try self.startTls();
+        try self.startConn(idle_ms);
+        self.drain();
+        return self;
+    }
+    fn startTls(self: *Client) !void {
+        // Set before clientTls, which hands wolfSSL a pointer to it.
+        self.conn_ref = .{ .get_conn = getConnCb, .user_data = self };
+        const tls = try clientTls(pskClientCb, &self.conn_ref);
+        self.ssl_ctx = tls.ctx;
+        self.ssl = tls.ssl;
+    }
+    fn startConn(self: *Client, idle_ms: u32) !void {
+        var dcid: c.ngtcp2_cid = undefined;
+        dcid.datalen = 16;
+        std.crypto.random.bytes(dcid.data[0..16]);
+        var scid: c.ngtcp2_cid = undefined;
+        scid.datalen = 8;
+        std.crypto.random.bytes(scid.data[0..8]);
+        var cbs = clientCallbacks();
+        cbs.handshake_completed = handshakeCompletedCb;
+        cbs.extend_max_local_streams_bidi = extendStreamsCb;
+        cbs.recv_stream_data = recvStreamDataCb;
+        cbs.acked_stream_data_offset = ackedStreamDataCb;
+        var settings: c.ngtcp2_settings = undefined;
+        c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings);
+        settings.initial_ts = timestampNs();
+        var params = clientParams();
+        // Tunable for the same reason as the daemon's: the reconnect loop
+        // has to see a dead transport on a schedule a test can wait for.
+        params.max_idle_timeout = @as(u64, idle_ms) * 1_000_000;
+        var path = pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
+        var conn: ?*c.ngtcp2_conn = null;
+        if (c.ngtcp2_conn_client_new_versioned(
+            &conn,
+            &dcid,
+            &scid,
+            &path,
+            c.NGTCP2_PROTO_VER_V1,
+            c.NGTCP2_CALLBACKS_VERSION,
+            &cbs,
+            c.NGTCP2_SETTINGS_VERSION,
+            &settings,
+            c.NGTCP2_TRANSPORT_PARAMS_VERSION,
+            &params,
+            null,
+            self,
+        ) != 0) return error.ConnInit;
+        self.conn = conn;
+        c.ngtcp2_conn_set_tls_native_handle(conn, self.ssl);
+        // Silence is not death: a terminal nobody is typing into must not be
+        // dropped at the idle timeout. A third of it, matching the daemon.
+        c.ngtcp2_conn_set_keep_alive_timeout(conn, keepAliveNs(idle_ms));
+    }
+    pub fn deinit(self: *Client) void {
+        self.in.deinit(self.alloc);
+        // ngtcp2 before the egress ring: it holds vectors into that ring for
+        // anything unacknowledged, so freeing the ring first leaves it
+        // reading memory that is already gone for the length of its own
+        // teardown. Same ordering as the listener's Conn.deinit.
+        if (self.conn) |cn| c.ngtcp2_conn_del(cn);
+        self.out.deinit(self.alloc);
+        if (self.ssl) |s| c.wolfSSL_free(s);
+        if (self.ssl_ctx) |x| c.wolfSSL_CTX_free(x);
+        std.posix.close(self.fd);
+        self.alloc.destroy(self);
+    }
+    /// Readable means a datagram arrived, not that a frame is waiting.
+    pub fn pollFd(self: *const Client) std.posix.fd_t {
+        return self.fd;
+    }
+    /// Handshake finished AND the stream open: a handshake without a stream
+    /// silently holds everything in the ring.
+    pub fn isReady(self: *const Client) bool {
+        return self.handshake_done and self.stream_id != -1 and !self.dead;
+    }
+    /// Milliseconds until ngtcp2 next wants servicing, capped. Feeds the
+    /// client's existing poll timeout — no timer fd, same as the daemon.
+    pub fn timeoutMs(self: *Client, cap_ms: i32) i32 {
+        // A negative cap means poll forever; the @intCast below would panic
+        // on it instead of honouring it.
+        if (cap_ms < 0) return cap_ms;
+        const conn = self.conn orelse return cap_ms;
+        const expiry = c.ngtcp2_conn_get_expiry(conn);
+        if (expiry == std.math.maxInt(u64)) return cap_ms;
+        const now = timestampNs();
+        if (expiry <= now) return 0;
+        return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000));
+    }
+    /// One service pass: read, run due timers, push ready egress. Safe to call
+    /// at any time.
+    pub fn pump(self: *Client) void {
+        if (self.dead) return;
+        self.readable();
+        self.tick();
+        self.drain();
+    }
+    /// ECONNREFUSED is fatal and reaches whichever syscall runs first, so
+    /// both paths must act.
+    fn sendRecvFailed(
+        self: *Client,
+        err: (std.posix.RecvFromError || std.posix.SendError),
+    ) void {
+        switch (err) {
+            error.ConnectionRefused => self.dead = true,
+            else => {},
+        }
+    }
+    /// Read whatever has arrived, WITHOUT transmitting — `pump` is the pass
+    /// that also acknowledges. The daemon's tests need the two apart: a
+    /// client that acks is a client the listener gets an inbound packet
+    /// from, and an inbound packet carries a drain the test was trying to
+    /// rule out.
+    pub fn readable(self: *Client) void {
+        var buf: [65536]u8 = undefined;
+        while (true) {
+            const n = std.posix.recv(self.fd, &buf, 0) catch |err| switch (err) {
+                error.WouldBlock => return,
+                else => {
+                    self.sendRecvFailed(err);
+                    return;
+                },
+            };
+            if (n == 0) continue;
+            const conn = self.conn orelse return;
+            var path = pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
+            var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
+            const rv = blk: {
+                self.ngtcp2_depth += 1;
+                defer self.ngtcp2_depth -= 1;
+                break :blk c.ngtcp2_conn_read_pkt(conn, &path, &pi, &buf, n, timestampNs());
+            };
+            if (rv != 0) {
+                self.dead = true;
+                return;
+            }
+        }
+    }
+    fn tick(self: *Client) void {
+        const conn = self.conn orelse return;
+        const now = timestampNs();
+        if (c.ngtcp2_conn_get_expiry(conn) > now) return;
+        // An idle timeout arrives here, which is how the reconnect loop
+        // learns the daemon stopped answering.
+        if (c.ngtcp2_conn_handle_expiry(conn, now) != 0) self.dead = true;
+    }
+    fn drain(self: *Client) void {
+        // The listener learned this the hard way: writing to a connection
+        // while ngtcp2 is reading a packet on it corrupts loss detection and
+        // can abort outright. Enforced here so it stays true of this side.
+        std.debug.assert(self.ngtcp2_depth == 0);
+        const conn = self.conn orelse return;
+        var buf: [max_udp]u8 = undefined;
+        var stream_blocked = false;
+        while (true) {
+            var ps: c.ngtcp2_path_storage = undefined;
+            c.ngtcp2_path_storage_zero(&ps);
+            var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
+            var wrote: c.ngtcp2_ssize = 0;
+            var vecs: [2]c.ngtcp2_vec = undefined;
+            var vcnt: usize = 0;
+            var sid: i64 = -1;
+            if (!stream_blocked and self.stream_id != -1) {
+                vcnt = self.out.vecs(&vecs);
+                if (vcnt > 0) sid = self.stream_id;
+            }
+            const n = c.ngtcp2_conn_writev_stream_versioned(
+                conn,
+                &ps.path,
+                c.NGTCP2_PKT_INFO_VERSION,
+                &pi,
+                &buf,
+                buf.len,
+                &wrote,
+,
+                sid,
+                if (vcnt > 0) &vecs else null,
+                vcnt,
+                timestampNs(),
+            );
+            if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) {
+                // Documented, not fatal. Retry the same iteration carrying
+                // no stream data so ACKs and keepalives still leave: they
+                // are how the peer's window reopens, and a client that
+                // stopped here would stall exactly as the daemon did.
+                stream_blocked = true;
+                continue;
+            }
+            switch (accountWrite(&self.out, wrote, n)) {
+                .stop => {
+                    self.dead = true;
+                    return;
+                },
+                .brk => return,
+                .cont => {},
+            }
+            _ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch |err| {
+                self.sendRecvFailed(err);
+                return;
+            };
+        }
+    }
+    /// A short return is the caller's signal to keep the rest and offer it
+    /// again: the ring is bounded, so the backlog belongs with somebody who
+    /// can see how big it is.
+    pub fn send(self: *Client, bytes: []const u8) usize {
+        if (self.dead or self.stream_id == -1) return 0;
+        const n = self.out.push(bytes);
+        if (n > 0) self.drain();
+        return n;
+    }
+};
+// Plain, not recursive: this module reaches the QUIC stack's @cImport, and
+// a recursive walk would force-analyze the entire wolfSSL/ngtcp2 namespace.
+test {

src/server/quic_server.zig

Old		New
@@ -3,10 +3,8 @@
3	//! both ends share — the C import, the key, the egress ring — is quic.zig's,	3	//! both ends share — the C import, the key, the egress ring — is quic.zig's,
4	//! and this file imports it.	4	//! and this file imports it.
5	//!	5	//!
6	//! Two things live here, not one. Below the test banner further down sits	6	//! The peer the tests below dial with is quic.Client — the shipping one, not
7	//! `TestClient`, a real QUIC peer the daemon's own tests drive; it is	7	//! a second copy of it living here.
8	//! exported for server.zig's benefit and is scaffolding, not a transport.
9	//! The shipping client is quic_client.zig.
10	//!	8	//!
11	//! This file follows proxy.zig's discipline: it knows NOTHING about the	9	//! This file follows proxy.zig's discipline: it knows NOTHING about the
12	//! frame protocol it carries. If a `proto.` import ever appears here,	10	//! frame protocol it carries. If a `proto.` import ever appears here,
@@ -947,7 +945,7 @@ pub const Listener = struct {
947	.cont => {},	945	.cont => {},
948	}	946	}
949		947
950	// No refusal check here, unlike quic_client's send: this socket	948	// No refusal check here, unlike quic.Client's send: this socket
951	// is UNCONNECTED, so the kernel has no peer to attribute an ICMP	949	// is UNCONNECTED, so the kernel has no peer to attribute an ICMP
952	// unreachable to and never delivers one. The asymmetry with the	950	// unreachable to and never delivers one. The asymmetry with the
953	// client is the sockets', not drift.	951	// client is the sockets', not drift.
@@ -983,246 +981,50 @@ pub const Listener = struct {
983	// ---------------------------------------------------------------------------	981	// ---------------------------------------------------------------------------
984	// Tests: a real handshake against a real client, in one process.	982	// Tests: a real handshake against a real client, in one process.
985	//	983	//
986	// The client below is test scaffolding, and not a stand-in for the real	984	// The listener cannot be tested by inspection — a QUIC handshake either
987	// one: quic_client.zig is the shipping transport. It exists because the	985	// completes against a real peer or it does not — so the tests below dial it
988	// listener cannot be tested by inspection — a QUIC handshake either	986	// with the shipping client, wrapped for the two conveniences a test wants.
989	// completes against a real peer or it does not.
990	// ---------------------------------------------------------------------------	987	// ---------------------------------------------------------------------------
991		988
992	var g_client_key: quic.Key = undefined;	989	/// The shipping `quic.Client`, plus the two things only a test wants: a
993		990	/// backlog it re-offers as the ring drains — `send` takes what fits, and a
994	fn pskClientCb(	991	/// payload larger than `egress_cap` needs somebody to hold the rest — and a
995	ssl: ?*c.WOLFSSL,	992	/// value a stack `defer` can deinit.
996	hint: [*c]const u8,	993	///
997	identity: [*c]u8,	994	/// It replaced a second hand-written QUIC client that lived here. That copy
998	id_max: c_uint,	995	/// existed because the folder rule forbids the server naming a client
999	key_out: [*c]u8,	996	/// module; `quic.Client` moving down to quic.zig, which both ends already
1000	key_max: c_uint,	997	/// import, dissolved the reason rather than the rule.
1001	ciphersuite: [c][c]const u8,	998	pub const TestPeer = struct {
1002	) callconv(.c) c_uint {	999	cl: *quic.Client,
1003	_ = ssl;
1004	_ = hint;
1005	return quic.answerPsk(g_client_key, identity, id_max, key_out, key_max, ciphersuite);
1006	}
1007
1008	/// Test scaffolding, exported so the daemon's own tests can drive a real
1009	/// QUIC peer. Not the client transport — that is quic_client.zig — and
1010	/// deliberately less: no reconnect, no resumption, a fixed receive buffer.
1011	pub const TestClient = struct {
1012	fd: std.posix.fd_t,
1013	ssl_ctx: ?*c.WOLFSSL_CTX,
1014	ssl: ?*c.WOLFSSL,
1015	conn: ?*c.ngtcp2_conn,
1016	conn_ref: c.ngtcp2_crypto_conn_ref,
1017	remote: std.posix.sockaddr.storage,
1018	remote_len: std.posix.socklen_t,
1019	local: std.posix.sockaddr.storage,
1020	local_len: std.posix.socklen_t,
1021	stream_id: i64 = -1,
1022	handshake_done: bool = false,
1023	echoed: usize = 0,
1024	/// When set, every arriving byte is compared against this at the offset
1025	/// it arrives at. A transport that echoes the right COUNT of the wrong
1026	/// bytes satisfies a length assertion and fails this one.
1027	verify: ?[]const u8 = null,
1028	mismatch: bool = false,
1029	/// Whether to give the peer more stream window as data is consumed.
1030	/// Turning it off is how a test reaches the blocked-stream path, which
1031	/// is a documented return value rather than a failure.
1032	extend: bool = true,
1033	out: []const u8 = &.{},	1000	out: []const u8 = &.{},
1034	out_sent: usize = 0,	1001	out_sent: usize = 0,
1035	/// Everything the peer sent, so a test can walk frames out of it.
1036	recv_buf: [128 * 1024]u8 = undefined,
1037	recv_len: usize = 0,
1038	/// Set when the peer's CONNECTION_CLOSE reaches this side. The oracle
1039	/// for closeAll is the client observing the close, not bytes the server
1040	/// queued — a packet not received is an unobserved close.
1041	draining: bool = false,
1042
1043	fn getConn(ref: [c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?c.ngtcp2_conn {
1044	const s: TestClient = @ptrCast(@alignCast(ref..user_data));
1045	return s.conn;
1046	}
1047		1002
1048	fn onHandshake(_: ?c.ngtcp2_conn, ud: ?anyopaque) callconv(.c) c_int {	1003	/// Idle 30s, not the listener's: a peer that PINGed on its own
1049	const s: *TestClient = @ptrCast(@alignCast(ud.?));	1004	/// schedule would answer the keepalive test's question for it.
1050	s.handshake_done = true;	1005	pub fn init(addr: std.net.Address, key: quic.Key) !TestPeer {
1051	return 0;	1006	return .{ .cl = try quic.Client.connect(std.testing.allocator, addr, key, 30_000) };
1052	}	1007	}
1053		1008
1054	fn onStreams(cn: ?c.ngtcp2_conn, _: u64, ud: ?anyopaque) callconv(.c) c_int {	1009	pub fn deinit(self: *TestPeer) void {
1055	const s: *TestClient = @ptrCast(@alignCast(ud.?));	1010	self.cl.deinit();
1056	if (s.stream_id == -1) {
1057	var sid: i64 = -1;
1058	if (c.ngtcp2_conn_open_bidi_stream(cn, &sid, null) == 0) s.stream_id = sid;
1059	}
1060	return 0;
1061	}	1011	}
1062		1012
1063	fn onData(	1013	/// Bytes to hand the ring, from the top, as room appears.
1064	cn: ?*c.ngtcp2_conn,	1014	pub fn offer(self: *TestPeer, bytes: []const u8) void {
1065	_: u32,	1015	self.out = bytes;
1066	stream_id: i64,	1016	self.out_sent = 0;
1067	_: u64,
1068	data: [*c]const u8,
1069	datalen: usize,
1070	ud: ?*anyopaque,
1071	_: ?*anyopaque,
1072	) callconv(.c) c_int {
1073	const s: *TestClient = @ptrCast(@alignCast(ud.?));
1074	// The client must extend too: without this the SERVER stalls once
1075	// it has echoed a window's worth back, and the large-payload test
1076	// below would hang rather than fail.
1077	if (s.extend) {
1078	_ = c.ngtcp2_conn_extend_max_stream_offset(cn, stream_id, datalen);
1079	_ = c.ngtcp2_conn_extend_max_offset(cn, datalen);
1080	}
1081	if (s.verify) \|exp\| {
1082	const off = s.echoed;
1083	if (off + datalen > exp.len or
1084	!std.mem.eql(u8, exp[off..][0..datalen], data[0..datalen]))
1085	{
1086	s.mismatch = true;
1087	}
1088	}
1089	s.echoed += datalen;
1090	if (s.recv_len + datalen <= s.recv_buf.len) {
1091	@memcpy(s.recv_buf[s.recv_len..][0..datalen], data[0..datalen]);
1092	s.recv_len += datalen;
1093	}
1094	return 0;
1095	}	1017	}
1096		1018
1097	pub fn init(server_addr: std.net.Address, key: quic.Key) !TestClient {	1019	/// One service pass: top the ring up, then read, tick and write.
1098	g_client_key = key;	1020	pub fn drain(self: *TestPeer) void {
1099	const fd = try std.posix.socket(	1021	if (self.out_sent < self.out.len) self.out_sent += self.cl.send(self.out[self.out_sent..]);
1100	server_addr.any.family,	1022	self.cl.pump();
1101	std.posix.SOCK.DGRAM \| std.posix.SOCK.NONBLOCK \| std.posix.SOCK.CLOEXEC,
1102	0,
1103	);
1104	var self: TestClient = .{
1105	.fd = fd,
1106	.ssl_ctx = null,
1107	.ssl = null,
1108	.conn = null,
1109	.conn_ref = undefined,
1110	.remote = undefined,
1111	.remote_len = server_addr.getOsSockLen(),
1112	.local = undefined,
1113	.local_len = @sizeOf(std.posix.sockaddr.storage),
1114	};
1115	@memcpy(std.mem.asBytes(&self.remote)[0..self.remote_len], std.mem.asBytes(&server_addr.any)[0..self.remote_len]);
1116	try std.posix.connect(fd, &server_addr.any, self.remote_len);
1117	try std.posix.getsockname(fd, @ptrCast(&self.local), &self.local_len);
1118	return self;
1119	}	1023	}
1120		1024
1121	pub fn start(self: *TestClient) !void {	1025	/// Nothing consumes, so this counts and `cl.in.items` is the bytes.
1122	// Set before clientTls, which hands wolfSSL a pointer to it.	1026	pub fn echoed(self: *const TestPeer) usize {
1123	self.conn_ref = .{ .get_conn = getConn, .user_data = self };	1027	return self.cl.in.items.len;
1124	const tls = try quic.clientTls(pskClientCb, &self.conn_ref);
1125	self.ssl_ctx = tls.ctx;
1126	const ssl = tls.ssl;
1127	self.ssl = ssl;
1128
1129	var dcid: c.ngtcp2_cid = undefined;
1130	dcid.datalen = 16;
1131	std.crypto.random.bytes(dcid.data[0..16]);
1132	var scid: c.ngtcp2_cid = undefined;
1133	scid.datalen = 8;
1134	std.crypto.random.bytes(scid.data[0..8]);
1135
1136	var cbs = quic.clientCallbacks();
1137	cbs.handshake_completed = onHandshake;
1138	cbs.extend_max_local_streams_bidi = onStreams;
1139	cbs.recv_stream_data = onData;
1140
1141	var settings: c.ngtcp2_settings = undefined;
1142	c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings);
1143	settings.initial_ts = quic.timestampNs();
1144	var params = quic.clientParams();
1145
1146	var path = quic.pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
1147	var conn: ?*c.ngtcp2_conn = null;
1148	if (c.ngtcp2_conn_client_new_versioned(
1149	&conn,
1150	&dcid,
1151	&scid,
1152	&path,
1153	c.NGTCP2_PROTO_VER_V1,
1154	c.NGTCP2_CALLBACKS_VERSION,
1155	&cbs,
1156	c.NGTCP2_SETTINGS_VERSION,
1157	&settings,
1158	c.NGTCP2_TRANSPORT_PARAMS_VERSION,
1159	&params,
1160	null,
1161	self,
1162	) != 0) return error.ConnInit;
1163	self.conn = conn;
1164	c.ngtcp2_conn_set_tls_native_handle(conn, ssl);
1165	}
1166
1167	pub fn drain(self: *TestClient) void {
1168	const conn = self.conn orelse return;
1169	var buf: [quic.max_udp]u8 = undefined;
1170	while (true) {
1171	var ps: c.ngtcp2_path_storage = undefined;
1172	c.ngtcp2_path_storage_zero(&ps);
1173	var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
1174	var wrote: c.ngtcp2_ssize = 0;
1175	var vec: c.ngtcp2_vec = undefined;
1176	var vcnt: usize = 0;
1177	var sid: i64 = -1;
1178	if (self.stream_id != -1 and self.out_sent < self.out.len) {
1179	vec = .{ .base = @constCast(self.out.ptr) + self.out_sent, .len = self.out.len - self.out_sent };
1180	vcnt = 1;
1181	sid = self.stream_id;
1182	}
1183	const n = c.ngtcp2_conn_writev_stream_versioned(
1184	conn,
1185	&ps.path,
1186	c.NGTCP2_PKT_INFO_VERSION,
1187	&pi,
1188	&buf,
1189	buf.len,
1190	&wrote,
1191	0,
1192	sid,
1193	if (vcnt > 0) &vec else null,
1194	vcnt,
1195	quic.timestampNs(),
1196	);
1197	// Same ordering as the listener's drain, and for the same
1198	// reason: ngtcp2 can commit the stream offset and still return
1199	// an error, so the bytes are accounted before the check or they
1200	// get offered again at an offset the peer has moved past.
1201	if (wrote > 0) self.out_sent += @intCast(wrote);
1202	if (n < 0) return;
1203	if (n == 0) return;
1204	_ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch return;
1205	}
1206	}
1207
1208	pub fn readable(self: *TestClient) void {
1209	var buf: [65536]u8 = undefined;
1210	while (true) {
1211	const n = std.posix.recv(self.fd, &buf, 0) catch return;
1212	if (n == 0) return;
1213	const conn = self.conn orelse return;
1214	var path = quic.pathFrom(&self.local, self.local_len, &self.remote, self.remote_len);
1215	var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
1216	_ = c.ngtcp2_conn_read_pkt(conn, &path, &pi, &buf, n, quic.timestampNs());
1217	if (c.ngtcp2_conn_in_draining_period(conn) != 0) self.draining = true;
1218	}
1219	}
1220
1221	pub fn deinit(self: *TestClient) void {
1222	if (self.conn) \|cn\| c.ngtcp2_conn_del(cn);
1223	if (self.ssl) \|s\| c.wolfSSL_free(s);
1224	if (self.ssl_ctx) \|x\| c.wolfSSL_CTX_free(x);
1225	std.posix.close(self.fd);
1226	}	1028	}
1227	};	1029	};
1228		1030
@@ -1277,18 +1079,17 @@ const EchoOwner = struct {
1277	/// Drive both ends until `done` or the deadline. Single-threaded on purpose:	1079	/// Drive both ends until `done` or the deadline. Single-threaded on purpose:
1278	/// a test that needs threads to make a handshake happen is a test that will	1080	/// a test that needs threads to make a handshake happen is a test that will
1279	/// wedge one day.	1081	/// wedge one day.
1280	fn pump(l: Listener, cl: TestClient, ms: u64, done: const fn (EchoOwner, TestClient) bool, owner: EchoOwner) bool {	1082	fn pump(l: Listener, cl: TestPeer, ms: u64, done: const fn (EchoOwner, TestPeer) bool, owner: EchoOwner) bool {
1281	var waited: u64 = 0;	1083	var waited: u64 = 0;
1282	while (waited < ms) {	1084	while (waited < ms) {
1283	if (done(owner, cl)) return true;	1085	if (done(owner, cl)) return true;
1284	var fds = [_]std.posix.pollfd{	1086	var fds = [_]std.posix.pollfd{
1285	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },	1087	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
1286	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	1088	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
1287	};	1089	};
1288	const ready = std.posix.poll(&fds, 10) catch return false;	1090	const ready = std.posix.poll(&fds, 10) catch return false;
1289	if (ready > 0) {	1091	if (ready > 0) {
1290	if (fds[0].revents != 0) l.readable();	1092	if (fds[0].revents != 0) l.readable();
1291	if (fds[1].revents != 0) cl.readable();
1292	} else {	1093	} else {
1293	waited += 10;	1094	waited += 10;
1294	}	1095	}
@@ -1328,18 +1129,16 @@ test "Listener: PSK handshake, Retry, and a payload larger than the initial wind
1328	const setup = try loopbackListener(alloc, key, &owner, 5000);	1129	const setup = try loopbackListener(alloc, key, &owner, 5000);
1329	defer setup.l.deinit();	1130	defer setup.l.deinit();
1330		1131
1331	var cl = try TestClient.init(setup.addr, key);	1132	var cl = try TestPeer.init(setup.addr, key);
1332	defer cl.deinit();	1133	defer cl.deinit();
1333	try cl.start();
1334	cl.drain();
1335		1134
1336	// Both sides, not just the client: the client declares the handshake	1135	// Both sides, not just the client: the client declares the handshake
1337	// complete one flight before the server does, so waiting only on the	1136	// complete one flight before the server does, so waiting only on the
1338	// client stops the pump while the server's final flight is still in the	1137	// client stops the pump while the server's final flight is still in the
1339	// air.	1138	// air.
1340	try std.testing.expect(pump(setup.l, &cl, 5000, struct {	1139	try std.testing.expect(pump(setup.l, &cl, 5000, struct {
1341	fn f(o: EchoOwner, t: TestClient) bool {	1140	fn f(o: EchoOwner, t: TestPeer) bool {
1342	return t.handshake_done and o.opened >= 1;	1141	return t.cl.handshake_done and o.opened >= 1;
1343	}	1142	}
1344	}.f, &owner));	1143	}.f, &owner));
1345		1144
@@ -1365,31 +1164,29 @@ test "Listener: PSK handshake, Retry, and a payload larger than the initial wind
1365	// production too, which is why it is worth a test rather than a comment.	1164	// production too, which is why it is worth a test rather than a comment.
1366	const payload = try alloc.alloc(u8, 3 * 1024 * 1024);	1165	const payload = try alloc.alloc(u8, 3 * 1024 * 1024);
1367	defer alloc.free(payload);	1166	defer alloc.free(payload);
1368	// A repeating byte pattern, checked as it arrives rather than counted	1167	// A repeating byte pattern, compared byte for byte, not merely counted.
1369	// at the end. Counting alone cannot tell a working transport from one	1168	// Counting alone cannot tell a working transport from one that delivers
1370	// that delivers the right NUMBER of the wrong bytes — reordered,	1169	// the right NUMBER of the wrong bytes — reordered, duplicated, or read
1371	// duplicated, or read back out of a buffer that had been recycled	1170	// back out of a buffer that had been recycled underneath it, which is
1372	// underneath it, which is precisely the failure this file has already	1171	// precisely the failure this file has already had once.
1373	// had once.
1374	for (payload, 0..) \|b, i\| b. = @truncate(i);	1172	for (payload, 0..) \|b, i\| b. = @truncate(i);
1375	cl.verify = payload;	1173	cl.offer(payload);
1376	cl.out = payload;
1377	cl.drain();	1174	cl.drain();
1378		1175
1379	const ok = pump(setup.l, &cl, 30_000, struct {	1176	const ok = pump(setup.l, &cl, 30_000, struct {
1380	fn f(_: EchoOwner, t: TestClient) bool {	1177	fn f(_: EchoOwner, t: TestPeer) bool {
1381	return t.echoed >= 3 * 1024 * 1024;	1178	return t.echoed() >= 3 * 1024 * 1024;
1382	}	1179	}
1383	}.f, &owner);	1180	}.f, &owner);
1384	if (!ok) {	1181	if (!ok) {
1385	std.debug.print(	1182	std.debug.print(
1386	"stalled: server saw {d} bytes, client got {d} of {d} back\n",	1183	"stalled: server saw {d} bytes, client got {d} of {d} back\n",
1387	.{ owner.received, cl.echoed, payload.len },	1184	.{ owner.received, cl.echoed(), payload.len },
1388	);	1185	);
1389	}	1186	}
1390	try std.testing.expect(ok);	1187	try std.testing.expect(ok);
1391	try std.testing.expectEqual(@as(usize, 3 * 1024 * 1024), owner.received);	1188	try std.testing.expectEqual(@as(usize, 3 * 1024 * 1024), owner.received);
1392	try std.testing.expect(!cl.mismatch);	1189	try std.testing.expectEqualSlices(u8, payload, cl.cl.in.items);
1393		1190
1394	// Piggybacking on the one place a LIVE connection with a finite expiry	1191	// Piggybacking on the one place a LIVE connection with a finite expiry
1395	// exists: negative is poll(2)'s "wait forever", and the timeout fold	1192	// exists: negative is poll(2)'s "wait forever", and the timeout fold
@@ -1411,20 +1208,18 @@ test "Listener: a client holding the wrong key never completes a handshake" {
1411	const setup = try loopbackListener(alloc, server_key, &owner, 5000);	1208	const setup = try loopbackListener(alloc, server_key, &owner, 5000);
1412	defer setup.l.deinit();	1209	defer setup.l.deinit();
1413		1210
1414	var cl = try TestClient.init(setup.addr, wrong_key);	1211	var cl = try TestPeer.init(setup.addr, wrong_key);
1415	defer cl.deinit();	1212	defer cl.deinit();
1416	try cl.start();
1417	cl.drain();
1418		1213
1419	// Deliberately generous: the point is that it never succeeds, not that	1214	// Deliberately generous: the point is that it never succeeds, not that
1420	// it fails fast. A shorter window could pass for the wrong reason.	1215	// it fails fast. A shorter window could pass for the wrong reason.
1421	_ = pump(setup.l, &cl, 3000, struct {	1216	_ = pump(setup.l, &cl, 3000, struct {
1422	fn f(_: EchoOwner, t: TestClient) bool {	1217	fn f(_: EchoOwner, t: TestPeer) bool {
1423	return t.handshake_done;	1218	return t.cl.handshake_done;
1424	}	1219	}
1425	}.f, &owner);	1220	}.f, &owner);
1426		1221
1427	try std.testing.expect(!cl.handshake_done);	1222	try std.testing.expect(!cl.cl.handshake_done);
1428	try std.testing.expectEqual(@as(usize, 0), owner.opened);	1223	try std.testing.expectEqual(@as(usize, 0), owner.opened);
1429	try std.testing.expectEqual(@as(usize, 0), owner.received);	1224	try std.testing.expectEqual(@as(usize, 0), owner.received);
1430		1225
@@ -1451,13 +1246,11 @@ test "Listener: keepalive carries an idle connection past its idle timeout" {
1451	const setup = try loopbackListener(alloc, key, &owner, idle_ms);	1246	const setup = try loopbackListener(alloc, key, &owner, idle_ms);
1452	defer setup.l.deinit();	1247	defer setup.l.deinit();
1453		1248
1454	var cl = try TestClient.init(setup.addr, key);	1249	var cl = try TestPeer.init(setup.addr, key);
1455	defer cl.deinit();	1250	defer cl.deinit();
1456	try cl.start();
1457	cl.drain();
1458	try std.testing.expect(pump(setup.l, &cl, 5000, struct {	1251	try std.testing.expect(pump(setup.l, &cl, 5000, struct {
1459	fn f(o: EchoOwner, t: TestClient) bool {	1252	fn f(o: EchoOwner, t: TestPeer) bool {
1460	return t.handshake_done and o.opened > 0;	1253	return t.cl.handshake_done and o.opened > 0;
1461	}	1254	}
1462	}.f, &owner));	1255	}.f, &owner));
1463		1256
@@ -1466,7 +1259,7 @@ test "Listener: keepalive carries an idle connection past its idle timeout" {
1466	// least this long — the keepalive PINGs it does carry are the whole	1259	// least this long — the keepalive PINGs it does carry are the whole
1467	// point, and they are not application traffic.	1260	// point, and they are not application traffic.
1468	_ = pump(setup.l, &cl, idle_ms * 3, struct {	1261	_ = pump(setup.l, &cl, idle_ms * 3, struct {
1469	fn f(_: EchoOwner, _: TestClient) bool {	1262	fn f(_: EchoOwner, _: TestPeer) bool {
1470	return false;	1263	return false;
1471	}	1264	}
1472	}.f, &owner);	1265	}.f, &owner);
@@ -1475,11 +1268,10 @@ test "Listener: keepalive carries an idle connection past its idle timeout" {
1475	try std.testing.expectEqual(@as(usize, 0), owner.closed);	1268	try std.testing.expectEqual(@as(usize, 0), owner.closed);
1476	// ...and the connection is not merely un-reaped but still usable, which	1269	// ...and the connection is not merely un-reaped but still usable, which
1477	// a live conn struct on its own would not prove.	1270	// a live conn struct on its own would not prove.
1478	cl.out = "still-here";	1271	cl.offer("still-here");
1479	cl.out_sent = 0;
1480	try std.testing.expect(pump(setup.l, &cl, 5000, struct {	1272	try std.testing.expect(pump(setup.l, &cl, 5000, struct {
1481	fn f(_: EchoOwner, t: TestClient) bool {	1273	fn f(_: EchoOwner, t: TestPeer) bool {
1482	return t.echoed >= "still-here".len;	1274	return t.echoed() >= "still-here".len;
1483	}	1275	}
1484	}.f, &owner));	1276	}.f, &owner));
1485	try std.testing.expectEqual(@as(usize, 0), owner.closed);	1277	try std.testing.expectEqual(@as(usize, 0), owner.closed);
@@ -1491,44 +1283,41 @@ test "Listener: keepalive carries an idle connection past its idle timeout" {
1491	/// client say something first. Waits until the round trip is acknowledged,	1283	/// client say something first. Waits until the round trip is acknowledged,
1492	/// so the egress ring is empty again and the next assertion is about the	1284	/// so the egress ring is empty again and the next assertion is about the
1493	/// test's own bytes.	1285	/// test's own bytes.
1494	fn openStream(l: Listener, cl: TestClient, owner: *EchoOwner) !void {	1286	fn openStream(l: Listener, cl: TestPeer, owner: *EchoOwner) !void {
1495	cl.out = "hi";	1287	cl.offer("hi");
1496	cl.out_sent = 0;
1497	cl.drain();	1288	cl.drain();
1498	const ok = pumpUntil(l, cl, owner, 10_000, struct {	1289	const ok = pumpUntil(l, cl, owner, 10_000, struct {
1499	fn f(o: EchoOwner, t: TestClient) bool {	1290	fn f(o: EchoOwner, t: TestPeer) bool {
1500	return o.received >= 2 and t.echoed >= 2 and l_pending(o) == 0;	1291	return o.received >= 2 and t.echoed() >= 2 and l_pending(o) == 0;
1501	}	1292	}
1502	fn l_pending(o: *EchoOwner) usize {	1293	fn l_pending(o: *EchoOwner) usize {
1503	return o.listener.pendingBytes(o.id);	1294	return o.listener.pendingBytes(o.id);
1504	}	1295	}
1505	}.f);	1296	}.f);
1506	if (!ok) return error.StreamNeverOpened;	1297	if (!ok) return error.StreamNeverOpened;
1507	cl.out = &.{};	1298	cl.offer(&.{});
1508	cl.out_sent = 0;	1299	cl.cl.in.clearRetainingCapacity();
1509	cl.echoed = 0;
1510	}	1300	}
1511		1301
1512	/// `pump` counts only quiet iterations toward its deadline, which is wrong	1302	/// `pump` counts only quiet iterations toward its deadline, which is wrong
1513	/// for a bulk transfer that is busy throughout.	1303	/// for a bulk transfer that is busy throughout.
1514	fn pumpUntil(	1304	fn pumpUntil(
1515	l: *Listener,	1305	l: *Listener,
1516	cl: *TestClient,	1306	cl: *TestPeer,
1517	owner: *EchoOwner,	1307	owner: *EchoOwner,
1518	wall_ms: i64,	1308	wall_ms: i64,
1519	done: const fn (EchoOwner, *TestClient) bool,	1309	done: const fn (EchoOwner, *TestPeer) bool,
1520	) bool {	1310	) bool {
1521	const deadline = std.time.milliTimestamp() + wall_ms;	1311	const deadline = std.time.milliTimestamp() + wall_ms;
1522	while (std.time.milliTimestamp() < deadline) {	1312	while (std.time.milliTimestamp() < deadline) {
1523	if (done(owner, cl)) return true;	1313	if (done(owner, cl)) return true;
1524	var fds = [_]std.posix.pollfd{	1314	var fds = [_]std.posix.pollfd{
1525	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },	1315	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
1526	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	1316	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
1527	};	1317	};
1528	const ready = std.posix.poll(&fds, 5) catch return false;	1318	const ready = std.posix.poll(&fds, 5) catch return false;
1529	if (ready > 0) {	1319	if (ready > 0) {
1530	if (fds[0].revents != 0) l.readable();	1320	if (fds[0].revents != 0) l.readable();
1531	if (fds[1].revents != 0) cl.readable();
1532	}	1321	}
1533	l.tick();	1322	l.tick();
1534	owner.flush();	1323	owner.flush();
@@ -1549,7 +1338,7 @@ test "Listener: bytes survive retransmission, which is what the buffer is for" {
1549	const setup = try loopbackListener(alloc, key, &owner, 10_000);	1338	const setup = try loopbackListener(alloc, key, &owner, 10_000);
1550	defer setup.l.deinit();	1339	defer setup.l.deinit();
1551		1340
1552	var cl = try TestClient.init(setup.addr, key);	1341	var cl = try TestPeer.init(setup.addr, key);
1553	defer cl.deinit();	1342	defer cl.deinit();
1554		1343
1555	// A deliberately tiny receive buffer, which is how loopback is made to	1344	// A deliberately tiny receive buffer, which is how loopback is made to
@@ -1561,17 +1350,15 @@ test "Listener: bytes survive retransmission, which is what the buffer is for" {
1561	// reached the same path by accident, about one run in twenty, and	1350	// reached the same path by accident, about one run in twenty, and
1562	// segfaulted when it did.	1351	// segfaulted when it did.
1563	try std.posix.setsockopt(	1352	try std.posix.setsockopt(
1564	cl.fd,	1353	cl.cl.fd,
1565	std.posix.SOL.SOCKET,	1354	std.posix.SOL.SOCKET,
1566	std.posix.SO.RCVBUF,	1355	std.posix.SO.RCVBUF,
1567	&std.mem.toBytes(@as(c_int, 64 * 1024)),	1356	&std.mem.toBytes(@as(c_int, 64 * 1024)),
1568	);	1357	);
1569		1358
1570	try cl.start();
1571	cl.drain();
1572	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {	1359	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1573	fn f(o: EchoOwner, t: TestClient) bool {	1360	fn f(o: EchoOwner, t: TestPeer) bool {
1574	return t.handshake_done and o.opened > 0;	1361	return t.cl.handshake_done and o.opened > 0;
1575	}	1362	}
1576	}.f));	1363	}.f));
1577		1364
@@ -1582,14 +1369,12 @@ test "Listener: bytes survive retransmission, which is what the buffer is for" {
1582	defer alloc.free(payload);	1369	defer alloc.free(payload);
1583	for (payload, 0..) \|b, i\| b. = @truncate(i *% 31 +% 7);	1370	for (payload, 0..) \|b, i\| b. = @truncate(i *% 31 +% 7);
1584		1371
1585	cl.verify = payload;	1372	cl.offer(payload);
1586	cl.out = payload;
1587	cl.out_sent = 0;
1588		1373
1589	const ok = pumpUntil(setup.l, &cl, &owner, 60_000, struct {	1374	const ok = pumpUntil(setup.l, &cl, &owner, 60_000, struct {
1590	fn f(o: EchoOwner, t: TestClient) bool {	1375	fn f(o: EchoOwner, t: TestPeer) bool {
1591	_ = o;	1376	_ = o;
1592	return t.echoed >= t.verify.?.len;	1377	return t.echoed() >= t.out.len;
1593	}	1378	}
1594	}.f);	1379	}.f);
1595	try std.testing.expect(ok);	1380	try std.testing.expect(ok);
@@ -1597,8 +1382,8 @@ test "Listener: bytes survive retransmission, which is what the buffer is for" {
1597	// Count AND content. A transport that lost a retransmission and carried	1382	// Count AND content. A transport that lost a retransmission and carried
1598	// on would land the right number of bytes in the wrong order, which no	1383	// on would land the right number of bytes in the wrong order, which no
1599	// length assertion can see.	1384	// length assertion can see.
1600	try std.testing.expectEqual(size, cl.echoed);	1385	try std.testing.expectEqual(size, cl.echoed());
1601	try std.testing.expect(!cl.mismatch);	1386	try std.testing.expectEqualSlices(u8, payload, cl.cl.in.items);
1602	try std.testing.expectEqual(size, owner.received);	1387	try std.testing.expectEqual(size, owner.received);
1603	try std.testing.expectEqual(@as(usize, 0), owner.closed);	1388	try std.testing.expectEqual(@as(usize, 0), owner.closed);
1604	}	1389	}
@@ -1612,13 +1397,11 @@ test "Listener.send: takes what fits, refuses when full, and recovers on acks" {
1612	const setup = try loopbackListener(alloc, key, &owner, 10_000);	1397	const setup = try loopbackListener(alloc, key, &owner, 10_000);
1613	defer setup.l.deinit();	1398	defer setup.l.deinit();
1614		1399
1615	var cl = try TestClient.init(setup.addr, key);	1400	var cl = try TestPeer.init(setup.addr, key);
1616	defer cl.deinit();	1401	defer cl.deinit();
1617	try cl.start();
1618	cl.drain();
1619	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {	1402	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1620	fn f(o: EchoOwner, t: TestClient) bool {	1403	fn f(o: EchoOwner, t: TestPeer) bool {
1621	return t.handshake_done and o.opened > 0;	1404	return t.cl.handshake_done and o.opened > 0;
1622	}	1405	}
1623	}.f));	1406	}.f));
1624		1407
@@ -1644,11 +1427,11 @@ test "Listener.send: takes what fits, refuses when full, and recovers on acks" {
1644	// client is not the same event as the ring being free to reuse it, and	1427	// client is not the same event as the ring being free to reuse it, and
1645	// conflating them is what this buffer exists to stop anyone doing.	1428	// conflating them is what this buffer exists to stop anyone doing.
1646	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 20_000, struct {	1429	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 20_000, struct {
1647	fn f(o: EchoOwner, _: TestClient) bool {	1430	fn f(o: EchoOwner, _: TestPeer) bool {
1648	return o.listener.pendingBytes(o.id) == 0;	1431	return o.listener.pendingBytes(o.id) == 0;
1649	}	1432	}
1650	}.f));	1433	}.f));
1651	try std.testing.expectEqual(quic.egress_cap, cl.echoed);	1434	try std.testing.expectEqual(quic.egress_cap, cl.echoed());
1652	try std.testing.expect(try setup.l.send(owner.id, "room again") > 0);	1435	try std.testing.expect(try setup.l.send(owner.id, "room again") > 0);
1653		1436
1654	// An id nobody owns is an error, not a silent success.	1437	// An id nobody owns is an error, not a silent success.
@@ -1669,13 +1452,11 @@ test "Listener: a full peer window blocks the stream without stopping the ACKs"
1669	const setup = try loopbackListener(alloc, key, &owner, idle_ms);	1452	const setup = try loopbackListener(alloc, key, &owner, idle_ms);
1670	defer setup.l.deinit();	1453	defer setup.l.deinit();
1671		1454
1672	var cl = try TestClient.init(setup.addr, key);	1455	var cl = try TestPeer.init(setup.addr, key);
1673	defer cl.deinit();	1456	defer cl.deinit();
1674	try cl.start();
1675	cl.drain();
1676	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {	1457	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1677	fn f(o: EchoOwner, t: TestClient) bool {	1458	fn f(o: EchoOwner, t: TestPeer) bool {
1678	return t.handshake_done and o.opened > 0;	1459	return t.cl.handshake_done and o.opened > 0;
1679	}	1460	}
1680	}.f));	1461	}.f));
1681		1462
@@ -1683,7 +1464,7 @@ test "Listener: a full peer window blocks the stream without stopping the ACKs"
1683		1464
1684	// The client consumes but never grants more window, so the peer's	1465	// The client consumes but never grants more window, so the peer's
1685	// stream credit runs out and stays out.	1466	// stream credit runs out and stays out.
1686	cl.extend = false;	1467	cl.cl.extend_windows = false;
1687		1468
1688	const big = try alloc.alloc(u8, 4 * quic.egress_cap);	1469	const big = try alloc.alloc(u8, 4 * quic.egress_cap);
1689	defer alloc.free(big);	1470	defer alloc.free(big);
@@ -1693,13 +1474,13 @@ test "Listener: a full peer window blocks the stream without stopping the ACKs"
1693		1474
1694	// Well past the idle timeout, with the stream blocked the entire time.	1475	// Well past the idle timeout, with the stream blocked the entire time.
1695	_ = pumpUntil(setup.l, &cl, &owner, idle_ms * 3, struct {	1476	_ = pumpUntil(setup.l, &cl, &owner, idle_ms * 3, struct {
1696	fn f(_: EchoOwner, _: TestClient) bool {	1477	fn f(_: EchoOwner, _: TestPeer) bool {
1697	return false;	1478	return false;
1698	}	1479	}
1699	}.f);	1480	}.f);
1700		1481
1701	// The stream really did block — the client cannot have taken it all.	1482	// The stream really did block — the client cannot have taken it all.
1702	try std.testing.expect(cl.echoed < big.len);	1483	try std.testing.expect(cl.echoed() < big.len);
1703	// ...and the connection is alive, which it can only be if packets that	1484	// ...and the connection is alive, which it can only be if packets that
1704	// are not stream data kept flowing while it was blocked.	1485	// are not stream data kept flowing while it was blocked.
1705	try std.testing.expectEqual(@as(usize, 0), owner.closed);	1486	try std.testing.expectEqual(@as(usize, 0), owner.closed);
@@ -1715,13 +1496,11 @@ test "Listener: a packet addressed to any advertised CID reaches its connection"
1715	const setup = try loopbackListener(alloc, key, &owner, 10_000);	1496	const setup = try loopbackListener(alloc, key, &owner, 10_000);
1716	defer setup.l.deinit();	1497	defer setup.l.deinit();
1717		1498
1718	var cl = try TestClient.init(setup.addr, key);	1499	var cl = try TestPeer.init(setup.addr, key);
1719	defer cl.deinit();	1500	defer cl.deinit();
1720	try cl.start();
1721	cl.drain();
1722	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {	1501	try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1723	fn f(o: EchoOwner, t: TestClient) bool {	1502	fn f(o: EchoOwner, t: TestPeer) bool {
1724	return t.handshake_done and o.opened > 0;	1503	return t.cl.handshake_done and o.opened > 0;
1725	}	1504	}
1726	}.f));	1505	}.f));
1727		1506
@@ -1864,34 +1643,31 @@ test "Listener: a reply queued just before a close still reaches the peer" {
1864	try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);	1643	try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);
1865	const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));	1644	const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));
1866		1645
1867	var cl = try TestClient.init(addr, key);	1646	var cl = try TestPeer.init(addr, key);
1868	defer cl.deinit();	1647	defer cl.deinit();
1869	try cl.start();
1870	cl.drain();
1871		1648
1872	cl.out = "attach";	1649	cl.offer("attach");
1873	var waited: u64 = 0;	1650	var waited: u64 = 0;
1874	while (waited < 10_000) : (waited += 5) {	1651	while (waited < 10_000) : (waited += 5) {
1875	var fds = [_]std.posix.pollfd{	1652	var fds = [_]std.posix.pollfd{
1876	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },	1653	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
1877	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	1654	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
1878	};	1655	};
1879	_ = std.posix.poll(&fds, 5) catch break;	1656	_ = std.posix.poll(&fds, 5) catch break;
1880	if (fds[0].revents != 0) l.readable();	1657	if (fds[0].revents != 0) l.readable();
1881	if (fds[1].revents != 0) cl.readable();
1882	l.tick();	1658	l.tick();
1883	cl.drain();	1659	cl.drain();
1884	for (l.conns) \|slot\| {	1660	for (l.conns) \|slot\| {
1885	if (slot) \|cn\| l.drain(cn);	1661	if (slot) \|cn\| l.drain(cn);
1886	}	1662	}
1887	if (cl.recv_len >= "REFUSED-FULL".len) break;	1663	if (cl.echoed() >= "REFUSED-FULL".len) break;
1888	}	1664	}
1889		1665
1890	try std.testing.expectEqual(@as(usize, "REFUSED-FULL".len), owner.queued);	1666	try std.testing.expectEqual(@as(usize, "REFUSED-FULL".len), owner.queued);
1891	// The bytes are on the peer, not merely in a ring that was then freed.	1667	// The bytes are on the peer, not merely in a ring that was then freed.
1892	try std.testing.expect(std.mem.indexOf(	1668	try std.testing.expect(std.mem.indexOf(
1893	u8,	1669	u8,
1894	cl.recv_buf[0..cl.recv_len],	1670	cl.cl.in.items,
1895	"REFUSED-FULL",	1671	"REFUSED-FULL",
1896	) != null);	1672	) != null);
1897	// ...and the connection really did go.	1673	// ...and the connection really did go.
@@ -1950,21 +1726,18 @@ test "Listener: closing a connection from inside a receive callback is deferred"
1950	try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);	1726	try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);
1951	const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));	1727	const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));
1952		1728
1953	var cl = try TestClient.init(addr, key);	1729	var cl = try TestPeer.init(addr, key);
1954	defer cl.deinit();	1730	defer cl.deinit();
1955	try cl.start();
1956	cl.drain();
1957		1731
1958	var waited: u64 = 0;	1732	var waited: u64 = 0;
1959	cl.out = "goodbye";	1733	cl.offer("goodbye");
1960	while (waited < 10_000 and owner.received == 0) : (waited += 5) {	1734	while (waited < 10_000 and owner.received == 0) : (waited += 5) {
1961	var fds = [_]std.posix.pollfd{	1735	var fds = [_]std.posix.pollfd{
1962	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },	1736	.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
1963	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	1737	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
1964	};	1738	};
1965	_ = std.posix.poll(&fds, 5) catch break;	1739	_ = std.posix.poll(&fds, 5) catch break;
1966	if (fds[0].revents != 0) l.readable();	1740	if (fds[0].revents != 0) l.readable();
1967	if (fds[1].revents != 0) cl.readable();
1968	l.tick();	1741	l.tick();
1969	cl.drain();	1742	cl.drain();
1970	for (l.conns) \|slot\| {	1743	for (l.conns) \|slot\| {
@@ -2032,14 +1805,12 @@ test "initFromFd: an adopted bound fd serves a handshake" {
2032		1805
2033	try std.testing.expectEqual(expected_port, l.boundAddr().getPort());	1806	try std.testing.expectEqual(expected_port, l.boundAddr().getPort());
2034		1807
2035	var cl = try TestClient.init(l.boundAddr(), key);	1808	var cl = try TestPeer.init(l.boundAddr(), key);
2036	defer cl.deinit();	1809	defer cl.deinit();
2037	try cl.start();
2038	cl.drain();
2039		1810
2040	try std.testing.expect(pump(l, &cl, 5000, struct {	1811	try std.testing.expect(pump(l, &cl, 5000, struct {
2041	fn f(o: EchoOwner, t: TestClient) bool {	1812	fn f(o: EchoOwner, t: TestPeer) bool {
2042	return t.handshake_done and o.opened >= 1;	1813	return t.cl.handshake_done and o.opened >= 1;
2043	}	1814	}
2044	}.f, &owner));	1815	}.f, &owner));
2045	try std.testing.expectEqual(@as(usize, 1), owner.opened);	1816	try std.testing.expectEqual(@as(usize, 1), owner.opened);
@@ -2057,14 +1828,12 @@ test "closeAll: a connected client hears the goodbye" {
2057	const setup = try loopbackListener(alloc, key, &owner, 5000);	1828	const setup = try loopbackListener(alloc, key, &owner, 5000);
2058	defer setup.l.deinit();	1829	defer setup.l.deinit();
2059		1830
2060	var cl = try TestClient.init(setup.addr, key);	1831	var cl = try TestPeer.init(setup.addr, key);
2061	defer cl.deinit();	1832	defer cl.deinit();
2062	try cl.start();
2063	cl.drain();
2064		1833
2065	try std.testing.expect(pump(setup.l, &cl, 5000, struct {	1834	try std.testing.expect(pump(setup.l, &cl, 5000, struct {
2066	fn f(o: EchoOwner, t: TestClient) bool {	1835	fn f(o: EchoOwner, t: TestPeer) bool {
2067	return t.handshake_done and o.opened >= 1;	1836	return t.cl.handshake_done and o.opened >= 1;
2068	}	1837	}
2069	}.f, &owner));	1838	}.f, &owner));
2070		1839
@@ -2072,13 +1841,16 @@ test "closeAll: a connected client hears the goodbye" {
2072	// packet leaves before the caller proceeds to exec.	1841	// packet leaves before the caller proceeds to exec.
2073	setup.l.closeAll();	1842	setup.l.closeAll();
2074		1843
2075	// The oracle is the client's observed close, not bytes the server queued.	1844	// The oracle is the client's observed close, not bytes the server queued
		1845	// — and specifically the draining period, which only a CONNECTION_CLOSE
		1846	// it actually read can enter. `dead` would also be set by any other read
		1847	// failure, and would pass for the wrong reason.
2076	try std.testing.expect(pump(setup.l, &cl, 5000, struct {	1848	try std.testing.expect(pump(setup.l, &cl, 5000, struct {
2077	fn f(_: EchoOwner, t: TestClient) bool {	1849	fn f(_: EchoOwner, t: TestPeer) bool {
2078	return t.draining;	1850	return c.ngtcp2_conn_in_draining_period(t.cl.conn.?) != 0;
2079	}	1851	}
2080	}.f, &owner));	1852	}.f, &owner));
2081	try std.testing.expect(cl.draining);	1853	try std.testing.expect(c.ngtcp2_conn_in_draining_period(cl.cl.conn.?) != 0);
2082	}	1854	}
2083		1855
2084	// Plain, not recursive: this module reaches the QUIC stack's @cImport, and	1856	// Plain, not recursive: this module reaches the QUIC stack's @cImport, and

src/server/server_test_agent.zig

Old		New
@@ -711,10 +711,8 @@ test "Server: a QUIC client's agent channels die with the client" {
711	const q = try quicTestServer(&srv, key);	711	const q = try quicTestServer(&srv, key);
712	defer q.l.deinit();	712	defer q.l.deinit();
713		713
714	var cl = try quic_server.TestClient.init(q.addr, key);	714	var cl = try quic_server.TestPeer.init(q.addr, key);
715	defer cl.deinit();	715	defer cl.deinit();
716	try cl.start();
717	cl.drain();
718		716
719	// attachOver cannot say "offering", and the offer rides its own frame	717	// attachOver cannot say "offering", and the offer rides its own frame
720	// re-sent after every attach, so both go out in one flight here.	718	// re-sent after every attach, so both go out in one flight here.
@@ -725,7 +723,7 @@ test "Server: a QUIC client's agent channels die with the client" {
725	cl.out = cbuf.items;	723	cl.out = cbuf.items;
726	cl.drain();	724	cl.drain();
727		725
728	var only = [_]*quic_server.TestClient{&cl};	726	var only = [_]*quic_server.TestPeer{&cl};
729	try quicPump(&srv, &only, 10000, &srv, struct {	727	try quicPump(&srv, &only, 10000, &srv, struct {
730	fn f(s: *Server) bool {	728	fn f(s: *Server) bool {
731	return s.clients[0] != null and s.clients[0].?.agent_offer;	729	return s.clients[0] != null and s.clients[0].?.agent_offer;
@@ -737,8 +735,8 @@ test "Server: a QUIC client's agent channels die with the client" {
737	const agent = try std.net.connectUnixSocket(path);	735	const agent = try std.net.connectUnixSocket(path);
738	defer agent.close();	736	defer agent.close();
739	try quicPump(&srv, &only, 10000, &cl, struct {	737	try quicPump(&srv, &only, 10000, &cl, struct {
740	fn f(t: *quic_server.TestClient) bool {	738	fn f(t: *quic_server.TestPeer) bool {
741	return findFrame(t.recv_buf[0..t.recv_len], .agent_open) != null;	739	return findFrame(t.cl.in.items, .agent_open) != null;
742	}	740	}
743	}.f);	741	}.f);
744	if (srv.agents.chans[0] == null) return error.NoAgentOpen;	742	if (srv.agents.chans[0] == null) return error.NoAgentOpen;

src/server/server_test_harness.zig

Old		New
@@ -108,7 +108,7 @@ pub fn quicTestServer(srv: Server, key: quic.Key) !struct { l: quic_server.Lis
108	/// is the integration under test.	108	/// is the integration under test.
109	pub fn quicPump(	109	pub fn quicPump(
110	srv: *Server,	110	srv: *Server,
111	clients: []*quic_server.TestClient,	111	clients: []*quic_server.TestPeer,
112	budget_ms: u64,	112	budget_ms: u64,
113	ctx: anytype,	113	ctx: anytype,
114	done: *const fn (@TypeOf(ctx)) bool,	114	done: *const fn (@TypeOf(ctx)) bool,
@@ -118,9 +118,9 @@ pub fn quicPump(
118	if (done(ctx)) return;	118	if (done(ctx)) return;
119	try srv.pumpOnce(5);	119	try srv.pumpOnce(5);
120	for (clients) \|cl\| {	120	for (clients) \|cl\| {
		121	var pfd = [_]std.posix.pollfd{.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
		122	_ = std.posix.poll(&pfd, 1) catch 0;
121	cl.drain();	123	cl.drain();
122	var pfd = [_]std.posix.pollfd{.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
123	if ((std.posix.poll(&pfd, 1) catch 0) > 0) cl.readable();
124	}	124	}
125	waited += 6;	125	waited += 6;
126	}	126	}

src/server/server_test_quic.zig

Old		New
@@ -13,7 +13,7 @@ const findFrame = h.findFrame;
13	const quicPump = h.quicPump;	13	const quicPump = h.quicPump;
14	const quicTestServer = h.quicTestServer;	14	const quicTestServer = h.quicTestServer;
15		15
16	fn attachOver(cl: quic_server.TestClient, buf: std.ArrayList(u8), alloc: std.mem.Allocator) !void {	16	fn attachOver(cl: quic_server.TestPeer, buf: std.ArrayList(u8), alloc: std.mem.Allocator) !void {
17	try proto.appendFrame(buf, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));	17	try proto.appendFrame(buf, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));
18	cl.out = buf.items;	18	cl.out = buf.items;
19	cl.drain();	19	cl.drain();
@@ -105,22 +105,20 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
105	const q = try quicTestServer(&srv, key);	105	const q = try quicTestServer(&srv, key);
106	defer q.l.deinit();	106	defer q.l.deinit();
107		107
108	var cl = try quic_server.TestClient.init(q.addr, key);	108	var cl = try quic_server.TestPeer.init(q.addr, key);
109	defer cl.deinit();	109	defer cl.deinit();
110	try cl.start();
111	cl.drain();
112		110
113	var out: std.ArrayList(u8) = .empty;	111	var out: std.ArrayList(u8) = .empty;
114	defer out.deinit(alloc);	112	defer out.deinit(alloc);
115	try attachOver(&cl, &out, alloc);	113	try attachOver(&cl, &out, alloc);
116		114
117	var only = [_]*quic_server.TestClient{&cl};	115	var only = [_]*quic_server.TestPeer{&cl};
118	try quicPump(&srv, &only, 8000, &cl, struct {	116	try quicPump(&srv, &only, 8000, &cl, struct {
119	fn f(t: *quic_server.TestClient) bool {	117	fn f(t: *quic_server.TestPeer) bool {
120	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	118	return findFrame(t.cl.in.items, .snapshot) != null;
121	}	119	}
122	}.f);	120	}.f);
123	try std.testing.expect(findFrame(cl.recv_buf[0..cl.recv_len], .snapshot) != null);	121	try std.testing.expect(findFrame(cl.cl.in.items, .snapshot) != null);
124		122
125	// From here the client says NOTHING. That is the whole point: the	123	// From here the client says NOTHING. That is the whole point: the
126	// listener's send only queues, and the two drains that fire on their own	124	// listener's send only queues, and the two drains that fire on their own
@@ -139,13 +137,13 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
139	try srv.pumpOnce(5);	137	try srv.pumpOnce(5);
140	cl.drain();	138	cl.drain();
141	var pfd = [_]std.posix.pollfd{	139	var pfd = [_]std.posix.pollfd{
142	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	140	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
143	};	141	};
144	if ((std.posix.poll(&pfd, 1) catch 0) > 0) cl.readable();	142	if ((std.posix.poll(&pfd, 1) catch 0) > 0) cl.cl.readable();
145	}	143	}
146	try std.testing.expect(q.l.timeoutMs(1000) > 100);	144	try std.testing.expect(q.l.timeoutMs(1000) > 100);
147		145
148	const before = cl.recv_len;	146	const before = cl.echoed();
149	cl.out = &.{};	147	cl.out = &.{};
150	srv.sessions.table[0].?.eng.feed("quiet-client-marker\r\n");	148	srv.sessions.table[0].?.eng.feed("quiet-client-marker\r\n");
151	srv.sendUpdate(0);	149	srv.sendUpdate(0);
@@ -168,15 +166,15 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
168	// Reading is not transmitting: the client takes whatever already	166	// Reading is not transmitting: the client takes whatever already
169	// arrived, and never gives the server an inbound packet to react to.	167	// arrived, and never gives the server an inbound packet to react to.
170	var tries: usize = 0;	168	var tries: usize = 0;
171	while (tries < 60 and findFrame(cl.recv_buf[0..cl.recv_len], .delta) == null) : (tries += 1) {	169	while (tries < 60 and findFrame(cl.cl.in.items, .delta) == null) : (tries += 1) {
172	var pfd = [_]std.posix.pollfd{	170	var pfd = [_]std.posix.pollfd{
173	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	171	.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
174	};	172	};
175	if ((std.posix.poll(&pfd, 2) catch 0) > 0) cl.readable();	173	if ((std.posix.poll(&pfd, 2) catch 0) > 0) cl.cl.readable();
176	}	174	}
177		175
178	try std.testing.expect(cl.recv_len > before);	176	try std.testing.expect(cl.echoed() > before);
179	try std.testing.expect(findFrame(cl.recv_buf[0..cl.recv_len], .delta) != null);	177	try std.testing.expect(findFrame(cl.cl.in.items, .delta) != null);
180	}	178	}
181		179
182	test "Server: a QUIC client still receives the shell's exit status" {	180	test "Server: a QUIC client still receives the shell's exit status" {
@@ -193,10 +191,8 @@ test "Server: a QUIC client still receives the shell's exit status" {
193	const q = try quicTestServer(&srv, key);	191	const q = try quicTestServer(&srv, key);
194	defer q.l.deinit();	192	defer q.l.deinit();
195		193
196	var cl = try quic_server.TestClient.init(q.addr, key);	194	var cl = try quic_server.TestPeer.init(q.addr, key);
197	defer cl.deinit();	195	defer cl.deinit();
198	try cl.start();
199	cl.drain();
200		196
201	var out: std.ArrayList(u8) = .empty;	197	var out: std.ArrayList(u8) = .empty;
202	defer out.deinit(alloc);	198	defer out.deinit(alloc);
@@ -204,13 +200,13 @@ test "Server: a QUIC client still receives the shell's exit status" {
204		200
205	// Attached and carrying state: the frame path works before we start	201	// Attached and carrying state: the frame path works before we start
206	// asking about the harder one.	202	// asking about the harder one.
207	var only = [_]*quic_server.TestClient{&cl};	203	var only = [_]*quic_server.TestPeer{&cl};
208	try quicPump(&srv, &only, 8000, &cl, struct {	204	try quicPump(&srv, &only, 8000, &cl, struct {
209	fn f(t: *quic_server.TestClient) bool {	205	fn f(t: *quic_server.TestPeer) bool {
210	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	206	return findFrame(t.cl.in.items, .snapshot) != null;
211	}	207	}
212	}.f);	208	}.f);
213	try std.testing.expect(findFrame(cl.recv_buf[0..cl.recv_len], .snapshot) != null);	209	try std.testing.expect(findFrame(cl.cl.in.items, .snapshot) != null);
214		210
215	// Now the case drainPending gets wrong when a QUIC slot contributes a	211	// Now the case drainPending gets wrong when a QUIC slot contributes a
216	// -1 descriptor: the shell exits, the exit_status is queued, and the	212	// -1 descriptor: the shell exits, the exit_status is queued, and the
@@ -220,12 +216,12 @@ test "Server: a QUIC client still receives the shell's exit status" {
220	cl.drain();	216	cl.drain();
221		217
222	try quicPump(&srv, &only, 15000, &cl, struct {	218	try quicPump(&srv, &only, 15000, &cl, struct {
223	fn f(t: *quic_server.TestClient) bool {	219	fn f(t: *quic_server.TestPeer) bool {
224	return findFrame(t.recv_buf[0..t.recv_len], .exit_status) != null;	220	return findFrame(t.cl.in.items, .exit_status) != null;
225	}	221	}
226	}.f);	222	}.f);
227		223
228	const status = findFrame(cl.recv_buf[0..cl.recv_len], .exit_status);	224	const status = findFrame(cl.cl.in.items, .exit_status);
229	try std.testing.expect(status != null);	225	try std.testing.expect(status != null);
230	try std.testing.expectEqual(@as(usize, 1), status.?.len);	226	try std.testing.expectEqual(@as(usize, 1), status.?.len);
231	try std.testing.expectEqual(@as(u8, 7), status.?[0]);	227	try std.testing.expectEqual(@as(u8, 7), status.?[0]);
@@ -245,14 +241,10 @@ test "Server: one QUIC client leaving does not disturb the other" {
245	const q = try quicTestServer(&srv, key);	241	const q = try quicTestServer(&srv, key);
246	defer q.l.deinit();	242	defer q.l.deinit();
247		243
248	var a = try quic_server.TestClient.init(q.addr, key);	244	var a = try quic_server.TestPeer.init(q.addr, key);
249	defer a.deinit();	245	defer a.deinit();
250	try a.start();	246	var b = try quic_server.TestPeer.init(q.addr, key);
251	a.drain();
252	var b = try quic_server.TestClient.init(q.addr, key);
253	defer b.deinit();	247	defer b.deinit();
254	try b.start();
255	b.drain();
256		248
257	var abuf: std.ArrayList(u8) = .empty;	249	var abuf: std.ArrayList(u8) = .empty;
258	defer abuf.deinit(alloc);	250	defer abuf.deinit(alloc);
@@ -261,14 +253,14 @@ test "Server: one QUIC client leaving does not disturb the other" {
261	try attachOver(&a, &abuf, alloc);	253	try attachOver(&a, &abuf, alloc);
262	try attachOver(&b, &bbuf, alloc);	254	try attachOver(&b, &bbuf, alloc);
263		255
264	var both = [_]*quic_server.TestClient{ &a, &b };	256	var both = [_]*quic_server.TestPeer{ &a, &b };
265	try quicPump(&srv, &both, 10000, &b, struct {	257	try quicPump(&srv, &both, 10000, &b, struct {
266	fn f(t: *quic_server.TestClient) bool {	258	fn f(t: *quic_server.TestPeer) bool {
267	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	259	return findFrame(t.cl.in.items, .snapshot) != null;
268	}	260	}
269	}.f);	261	}.f);
270	try std.testing.expect(findFrame(a.recv_buf[0..a.recv_len], .snapshot) != null);	262	try std.testing.expect(findFrame(a.cl.in.items, .snapshot) != null);
271	try std.testing.expect(findFrame(b.recv_buf[0..b.recv_len], .snapshot) != null);	263	try std.testing.expect(findFrame(b.cl.in.items, .snapshot) != null);
272		264
273	// A leaves. Sink.close() must tear down A's connection and NOT the UDP	265	// A leaves. Sink.close() must tear down A's connection and NOT the UDP
274	// socket every other client is reached through — for a socket client	266	// socket every other client is reached through — for a socket client
@@ -287,18 +279,18 @@ test "Server: one QUIC client leaving does not disturb the other" {
287	}.f);	279	}.f);
288		280
289	// B is still served: a marker typed now must come back to it.	281	// B is still served: a marker typed now must come back to it.
290	const before = b.recv_len;	282	const before = b.echoed();
291	try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");	283	try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");
292	b.out = bbuf.items;	284	b.out = bbuf.items;
293	b.drain();	285	b.drain();
294	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	286	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
295	defer replica.deinit();	287	defer replica.deinit();
296	try quicPump(&srv, &both, 15000, &b, struct {	288	try quicPump(&srv, &both, 15000, &b, struct {
297	fn f(t: *quic_server.TestClient) bool {	289	fn f(t: *quic_server.TestPeer) bool {
298	return t.recv_len > 0 and findFrame(t.recv_buf[0..t.recv_len], .delta) != null;	290	return t.echoed() > 0 and findFrame(t.cl.in.items, .delta) != null;
299	}	291	}
300	}.f);	292	}.f);
301	try std.testing.expect(b.recv_len > before);	293	try std.testing.expect(b.echoed() > before);
302	}	294	}
303		295
304	test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" {	296	test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" {
@@ -315,18 +307,17 @@ test "Server: a QUIC client that stops reading is dropped by the cap, not tolera
315	defer q.l.deinit();	307	defer q.l.deinit();
316	defer srv.deinit();	308	defer srv.deinit();
317		309
318	var cl = try quic_server.TestClient.init(q.addr, key);	310	var cl = try quic_server.TestPeer.init(q.addr, key);
319	defer cl.deinit();	311	defer cl.deinit();
320	try cl.start();
321		312
322	var bbuf: std.ArrayList(u8) = .empty;	313	var bbuf: std.ArrayList(u8) = .empty;
323	defer bbuf.deinit(alloc);	314	defer bbuf.deinit(alloc);
324	try attachOver(&cl, &bbuf, alloc);	315	try attachOver(&cl, &bbuf, alloc);
325		316
326	var only = [_]*quic_server.TestClient{&cl};	317	var only = [_]*quic_server.TestPeer{&cl};
327	try quicPump(&srv, &only, 8000, &cl, struct {	318	try quicPump(&srv, &only, 8000, &cl, struct {
328	fn f(t: *quic_server.TestClient) bool {	319	fn f(t: *quic_server.TestPeer) bool {
329	return t.handshake_done and t.echoed > 0;	320	return t.cl.handshake_done and t.echoed() > 0;
330	}	321	}
331	}.f);	322	}.f);
332	try std.testing.expect(srv.clients[0] != null);	323	try std.testing.expect(srv.clients[0] != null);
@@ -371,22 +362,21 @@ test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
371	defer q.l.deinit();	362	defer q.l.deinit();
372	defer srv.deinit();	363	defer srv.deinit();
373		364
374	var cl = try quic_server.TestClient.init(q.addr, key);	365	var cl = try quic_server.TestPeer.init(q.addr, key);
375	defer cl.deinit();	366	defer cl.deinit();
376	try cl.start();
377		367
378	var bbuf: std.ArrayList(u8) = .empty;	368	var bbuf: std.ArrayList(u8) = .empty;
379	defer bbuf.deinit(alloc);	369	defer bbuf.deinit(alloc);
380	try attachOver(&cl, &bbuf, alloc);	370	try attachOver(&cl, &bbuf, alloc);
381		371
382	var only = [_]*quic_server.TestClient{&cl};	372	var only = [_]*quic_server.TestPeer{&cl};
383	try quicPump(&srv, &only, 8000, &cl, struct {	373	try quicPump(&srv, &only, 8000, &cl, struct {
384	fn f(t: *quic_server.TestClient) bool {	374	fn f(t: *quic_server.TestPeer) bool {
385	return t.handshake_done and t.echoed > 0;	375	return t.cl.handshake_done and t.echoed() > 0;
386	}	376	}
387	}.f);	377	}.f);
388	try std.testing.expect(srv.clients[0] != null);	378	try std.testing.expect(srv.clients[0] != null);
389	const before = cl.echoed;	379	const before = cl.echoed();
390		380
391	// Queue more than one ring's worth in one go, so the drain cannot	381	// Queue more than one ring's worth in one go, so the drain cannot
392	// possibly finish by handing everything over once: some of it is still	382	// possibly finish by handing everything over once: some of it is still
@@ -406,14 +396,14 @@ test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
406	var done = false;	396	var done = false;
407	const t0 = std.time.milliTimestamp();	397	const t0 = std.time.milliTimestamp();
408	const th = try std.Thread.spawn(.{}, struct {	398	const th = try std.Thread.spawn(.{}, struct {
409	fn f(client: quic_server.TestClient, flag: bool) void {	399	fn f(client: quic_server.TestPeer, flag: bool) void {
410	// The peer: reads and acknowledges until the daemon says stop.	400	// The peer: reads and acknowledges until the daemon says stop.
411	while (!flag.*) {	401	while (!flag.*) {
412	client.drain();	402	client.drain();
413	var pfd = [_]std.posix.pollfd{	403	var pfd = [_]std.posix.pollfd{
414	.{ .fd = client.fd, .events = std.posix.POLL.IN, .revents = 0 },	404	.{ .fd = client.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
415	};	405	};
416	if ((std.posix.poll(&pfd, 5) catch 0) > 0) client.readable();	406	if ((std.posix.poll(&pfd, 5) catch 0) > 0) client.cl.readable();
417	}	407	}
418	}	408	}
419	}.f, .{ &cl, &done });	409	}.f, .{ &cl, &done });
@@ -425,6 +415,6 @@ test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
425	// Everything owed actually left, and it left inside the budget.	415	// Everything owed actually left, and it left inside the budget.
426	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.pending.items.len);	416	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.pending.items.len);
427	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.sink.inFlight());	417	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.sink.inFlight());
428	try std.testing.expect(cl.echoed >= before + payload.len);	418	try std.testing.expect(cl.echoed() >= before + payload.len);
429	try std.testing.expect(elapsed < 15_000);	419	try std.testing.expect(elapsed < 15_000);
430	}	420	}

Old		New
@@ -51,7 +51,7 @@ engine and a client can link those folders and paint its own way:
51	\|---\|---\|	51	\|---\|---\|
52	\| `src/engine/` \| `protocol` `engine` `delta` `replica` `predict` \|	52	\| `src/engine/` \| `protocol` `engine` `delta` `replica` `predict` \|
53	\| `src/server/` \| `server`(+`_agent` `_sessions` `_test_*`) `pty` `quic_server` `cmd` `shellint` `upgrade` \|	53	\| `src/server/` \| `server`(+`_agent` `_sessions` `_test_*`) `pty` `quic_server` `cmd` `shellint` `upgrade` \|
54	\| `src/client/` \| `client` `client_core` `hosts` `handoff` `quic_client` `layout` `wall` `webhub` `keymap` `wasm_core` \|	54	\| `src/client/` \| `client` `client_core` `hosts` `handoff` `layout` `wall` `webhub` `keymap` `wasm_core` \|
55	\| `src/tui/` \| `wallview`(+`wall_host` `wall_picker` `wall_pump` `wall_layout` `wall_test_*`) `interact` `paint` `select` \|	55	\| `src/tui/` \| `wallview`(+`wall_host` `wall_picker` `wall_pump` `wall_layout` `wall_test_*`) `interact` `paint` `select` \|
56	\| `src/cli/` \| `mux`(dispatch) `main`(daemon) `mux_main`(client) `muxa`(agent) `webhub_main`(hub) `flags` `spawn` \|	56	\| `src/cli/` \| `mux`(dispatch) `main`(daemon) `mux_main`(client) `muxa`(agent) `webhub_main`(hub) `flags` `spawn` \|
57	\| `src/` \| `xdg` `sockpath` `proxy` `quic` `testtmp` — what both sides link \|	57	\| `src/` \| `xdg` `sockpath` `proxy` `quic` `testtmp` — what both sides link \|
@@ -77,7 +77,7 @@ Layers are enforced in the same module table (grep `.layer =` for the graph).
77	\| Layer \| Modules \|	77	\| Layer \| Modules \|
78	\|---\|---\|	78	\|---\|---\|
79	\| 0 \| `protocol` `engine` `pty` `quic` `keymap` `xdg` `sockpath` `proxy` `cliflags` `testtmp` \|	79	\| 0 \| `protocol` `engine` `pty` `quic` `keymap` `xdg` `sockpath` `proxy` `cliflags` `testtmp` \|
80	\| 1 \| `client_core` `quic_server` `quic_client` `predict` `spawn`(cli) `handoff` `delta` `cmd` `wall` `shellint` `replica` `paint` `layout` `select` `upgrade` \|	80	\| 1 \| `client_core` `quic_server` `predict` `spawn`(cli) `handoff` `delta` `cmd` `wall` `shellint` `replica` `paint` `layout` `select` `upgrade` \|
81	\| 2 \| `server` `agent_main` `interact` `hosts` \|	81	\| 2 \| `server` `agent_main` `interact` `hosts` \|
82	\| 3 \| `client` `daemon_main` \|	82	\| 3 \| `client` `daemon_main` \|
83	\| 4 \| `webhub` `wallview` \|	83	\| 4 \| `webhub` `wallview` \|

Old		New
@@ -163,12 +163,6 @@ const mod_table = [_]ModSpec{
163	// table, and deliberately no protocol import — it carries opaque bytes,	163	// table, and deliberately no protocol import — it carries opaque bytes,
164	// exactly as proxy.zig does.	164	// exactly as proxy.zig does.
165	.{ .name = "quic_server", .path = "src/server/quic_server.zig", .layer = 1, .link_libc = true, .imports = &.{"quic"}, .quic_tests = true },	165	.{ .name = "quic_server", .path = "src/server/quic_server.zig", .layer = 1, .link_libc = true, .imports = &.{"quic"}, .quic_tests = true },
166	// The client's QUIC transport. Imports the vocabulary module — NOT the
167	// listener — for the pieces both ends must agree on (the key, the egress
168	// ring's lifetime discipline, the PSK identity and ALPN); duplicating
169	// those would make a handshake failure the first sign they had drifted.
170	// It has no business knowing a listener exists, and now it cannot.
171	.{ .name = "quic_client", .path = "src/client/quic_client.zig", .layer = 1, .link_libc = true, .imports = &.{"quic"}, .quic_tests = true },
172	// Speculative local echo: the overlay and its policy, and deliberately	166	// Speculative local echo: the overlay and its policy, and deliberately
173	// nothing else. No engine import, which is what lets the whole state	167	// nothing else. No engine import, which is what lets the whole state
174	// machine be exercised without a terminal — or a daemon — anywhere in	168	// machine be exercised without a terminal — or a daemon — anywhere in
@@ -259,10 +253,10 @@ const mod_table = [_]ModSpec{
259	// The agent-facing client. It speaks frames and owns no terminal, which	253	// The agent-facing client. It speaks frames and owns no terminal, which
260	// is the whole point — it attaches at 0x0 and never claims the grid.	254	// is the whole point — it attaches at 0x0 and never claims the grid.
261	// The transport modules are the CLI client's, minus everything that	255	// The transport modules are the CLI client's, minus everything that
262	// renders: `quic_client` for the remote arm and `xdg` for the one	256	// renders: `quic` for the remote arm and `xdg` for the one
263	// key-resolution rule all three binaries obey. Deliberately still no	257	// key-resolution rule all three binaries obey. Deliberately still no
264	// engine and no replica — muxa has nothing to draw.	258	// engine and no replica — muxa has nothing to draw.
265	.{ .name = "agent_main", .path = "src/cli/muxa.zig", .layer = 2, .link_libc = true, .imports = &.{ "protocol", "sockpath", "quic_client", "quic", "xdg", "cliflags" }, .quic_tests = true },	259	.{ .name = "agent_main", .path = "src/cli/muxa.zig", .layer = 2, .link_libc = true, .imports = &.{ "protocol", "sockpath", "quic", "xdg", "cliflags" }, .quic_tests = true },
266	.{ .name = "wsclient", .path = "test/wsclient.zig", .layer = 2, .link_libc = true, .imports = &.{ "engine", "replica", "protocol", "script" } },	260	.{ .name = "wsclient", .path = "test/wsclient.zig", .layer = 2, .link_libc = true, .imports = &.{ "engine", "replica", "protocol", "script" } },
267	// ---- layer 3 ----	261	// ---- layer 3 ----
268	// Dialling, and what a chord means. The client is the only thing that	262	// Dialling, and what a chord means. The client is the only thing that
@@ -274,7 +268,7 @@ const mod_table = [_]ModSpec{
274	// attach records its own tile (the wall is attach history), and the	268	// attach records its own tile (the wall is attach history), and the
275	// chord switches that re-dial from inside client.attach have to record	269	// chord switches that re-dial from inside client.attach have to record
276	// theirs too, so the writer cannot live up in mux_main.	270	// theirs too, so the writer cannot live up in mux_main.
277	.{ .name = "client", .path = "src/client/client.zig", .layer = 3, .link_libc = true, .imports = &.{ "protocol", "replica", "keymap", "quic_client", "quic", "handoff", "wall" }, .test_imports = &.{"testtmp"}, .quic_tests = true },	271	.{ .name = "client", .path = "src/client/client.zig", .layer = 3, .link_libc = true, .imports = &.{ "protocol", "replica", "keymap", "quic", "handoff", "wall" }, .test_imports = &.{"testtmp"}, .quic_tests = true },
278	// The daemon entrypoint loads the key and constructs the listener, so	272	// The daemon entrypoint loads the key and constructs the listener, so
279	// it needs quic/quic_server directly rather than through the server.	273	// it needs quic/quic_server directly rather than through the server.
280	// `muxd endpoint` prints the announce line handoff spells; sockpath is	274	// `muxd endpoint` prints the announce line handoff spells; sockpath is
@@ -806,12 +800,12 @@ fn docGate(b: std.Build, target: std.Build.ResolvedTarget, check_step: std.Bui
806	/// they carry the argument parsers — a test that is never built is not a	800	/// they carry the argument parsers — a test that is never built is not a
807	/// test (decisions.md).	801	/// test (decisions.md).
808	const test_order = [_][]const u8{	802	const test_order = [_][]const u8{
809	"script", "select", "protocol", "client_core", "interact", "engine", "pty",	803	"script", "select", "protocol", "client_core", "interact", "engine", "pty",
810	"delta", "cmd", "wall", "hosts", "upgrade", "shellint", "replica",	804	"delta", "cmd", "wall", "hosts", "upgrade", "shellint", "replica",
811	"keymap", "webhub", "wallview", "sockpath", "agent_main", "server", "client",	805	"keymap", "webhub", "wallview", "sockpath", "agent_main", "server", "client",
812	"proxy", "client_main", "quic", "quic_server", "daemon_main", "testtmp", "quic_client",	806	"proxy", "client_main", "quic", "quic_server", "daemon_main", "testtmp", "predict",
813	"predict", "rawmode", "delaypipe", "xdg", "spawn", "handoff", "paint",	807	"rawmode", "delaypipe", "xdg", "spawn", "handoff", "paint", "layout",
814	"layout", "render", "ptyclient", "hub_main", "wsclient", "cliflags", "mux",	808	"render", "ptyclient", "hub_main", "wsclient", "cliflags", "mux",
815	};	809	};
816		810
817	comptime {	811	comptime {