src/server/quic_server.zig
Ref: Size: 69.7 KiB History
//! The daemon's QUIC listener: one UDP socket, N authenticated connections,
//! each carrying one bidirectional stream of opaque bytes. The vocabulary both
//! ends share is quic.zig's, and the tests below dial with the shipping
//! `quic.Client`.
//!
//! It follows proxy.zig's discipline and knows NOTHING about the frame
//! protocol it carries: a `proto.` import here is a mistake. Authentication is
//! TLS 1.3 external PSK — both ends hold the same 32-byte key, no certificate.
const std = @import("std");
const quic = @import("quic");
const server_os = @import("server_os");
/// The C view of the QUIC stack, imported once in quic.zig and shared:
/// two @cImport blocks over the same headers produce two *distinct* Zig
/// types, so a second block here would find that `ngtcp2_vec` is not
/// `ngtcp2_vec`. One import, one type universe.
const c = quic.c;
// ---------------------------------------------------------------------------
// The listener
// ---------------------------------------------------------------------------
/// What the owner of a listener gets told. Deliberately three opaque events
/// and a byte slice: nothing here names a frame, a client slot, or a
/// protocol, which is what keeps this file swappable.
pub const Handler = struct {
ctx: *anyopaque,
/// A connection completed its handshake and is authenticated. The id is
/// this listener's handle for it, stable until onClose.
onOpen: *const fn (ctx: *anyopaque, id: u64) void,
/// Stream bytes arrived. Arbitrary chunking: the owner reassembles.
onData: *const fn (ctx: *anyopaque, id: u64, bytes: []const u8) void,
/// A connection the LISTENER gave up on: an idle timeout, a protocol error,
/// a peer that went away. It does NOT pair with `onOpen`: a handshake that
/// never completes produces neither, and a close the owner asked for via
/// `closeConn` produces no callback. This means "ended without you asking".
onClose: *const fn (ctx: *anyopaque, id: u64) void,
};
/// Connections the listener will hold at once, and deliberately MORE than
/// the daemon's `max_clients`. A connection exists from the moment its
/// handshake completes and only then asks for a client slot, so the surplus
/// is room for handshakes in flight and for peers about to be refused a slot.
/// One that finds no slot is answered and closed, which is a peer that knows.
///
/// Running level with the client table, or under it, is the failure to avoid:
/// a peer that finds no CONNECTION is dropped SILENTLY — `acceptConn` spells
/// it "full: drop, the peer will retry" — so it hangs on retries while the
/// daemon still has seats, and a host polled over QUIC reads `unreachable`
/// though it is running fine.
///
/// Restated here rather than derived: this file is a transport, and a
/// transport does not read the daemon's tables. The restatement is pinned
/// instead — `server_test_quic` asserts room for terminal and forwarding
/// admissions and pins the provisional role table to this cap, so changing one
/// without the others fails the build rather than going quiet.
// Terminal clients, forwarding-role peers and the small unclassified role
// table must all fit before the first authenticated stream frame arrives.
pub const max_conns = 48;
/// Source Connection IDs cached per connection. ngtcp2 caps its own pool at
/// NGTCP2_MAX_SCID_POOL_SIZE (8), so this is headroom; and because the
/// primary CID is matched separately, a cache that ever did fill would
/// degrade to the old primary-only behaviour rather than misroute.
const max_cids = 16;
/// wolfSSL's PSK callbacks carry no user pointer, so the key has to be
/// reachable without one. A daemon runs a single listener, which makes a
/// module-level key correct rather than merely convenient — but it is the
/// reason `Listener.init` refuses a second concurrent listener.
var g_key: ?quic.Key = null;
var g_listener_live: bool = false;
fn pskServerCb(
ssl: ?*c.WOLFSSL,
identity: [*c]const u8,
key_out: [*c]u8,
key_max: c_uint,
ciphersuite: [*c][*c]const u8,
) callconv(.c) c_uint {
_ = ssl;
// The out-param is `const char **` — the trap the spike recorded, and
// the reason this is written out rather than copied from the client
// callback, which takes a different shape.
if (ciphersuite) |cs| cs.* = quic.psk_ciphersuite;
const k = g_key orelse return 0;
if (identity == null) return 0;
if (std.mem.orderZ(u8, identity, quic.psk_identity) != .eq) return 0;
if (key_max < quic.key_len) return 0;
@memcpy(key_out[0..quic.key_len], &k.bytes);
return quic.key_len;
}
/// One authenticated peer: an ngtcp2 connection, its TLS object, and the
/// single bidirectional stream that carries everything.
const Conn = struct {
id: u64,
conn: ?*c.ngtcp2_conn = null,
ssl: ?*c.WOLFSSL = null,
conn_ref: c.ngtcp2_crypto_conn_ref = undefined,
listener: *Listener,
remote: std.posix.sockaddr.storage = undefined,
remote_len: std.posix.socklen_t = 0,
local_storage: std.posix.sockaddr.storage = undefined,
local_len: std.posix.socklen_t = 0,
scid: c.ngtcp2_cid = undefined,
/// Every Connection ID this endpoint has advertised and not retired. A peer
/// may address us by any of them — that is what makes migration work — and
/// matching only `scid` drops a migrated client's packets as noise.
cids: [max_cids]c.ngtcp2_cid = undefined,
ncids: usize = 0,
cids_dirty: bool = true,
/// Set when this connection must go but cannot be freed yet, because
/// ngtcp2 is inside a call on it. See `inNgtcp2` and `ngtcp2_depth`.
close_state: enum { open, closing_quiet, closing_notify } = .open,
stream_id: i64 = -1,
/// Bytes owed to this peer. See Egress: they do not move until acked.
out: quic.Egress,
opened: bool = false,
/// Re-read the advertised CIDs from ngtcp2. Lazy, because the answer
/// only changes when one is issued or retired and both of those tell us.
fn refreshCids(self: *Conn) void {
self.cids_dirty = false;
self.ncids = 0;
const conn = self.conn orelse return;
const n = c.ngtcp2_conn_get_scid2(conn, null);
if (n == 0 or n > max_cids) return;
self.ncids = c.ngtcp2_conn_get_scid2(conn, &self.cids);
}
/// Is this connection addressed by `dcid`? The primary is checked first
/// and without the cache, so this can only ever be more permissive than
/// the CID the connection was created with — never less.
fn matches(self: *Conn, dcid: []const u8) bool {
if (cidEql(&self.scid, dcid)) return true;
if (self.cids_dirty) self.refreshCids();
for (self.cids[0..self.ncids]) |*cid| {
if (cidEql(cid, dcid)) return true;
}
return false;
}
pub fn deinit(self: *Conn, alloc: std.mem.Allocator) void {
// ngtcp2 first, egress second: it holds vectors into the ring until the
// peer acknowledges them, so freeing the ring first leaves ngtcp2
// reading freed memory for the length of its own teardown.
if (self.conn) |cn| c.ngtcp2_conn_del(cn);
if (self.ssl) |s| c.wolfSSL_free(s);
self.conn = null;
self.ssl = null;
self.out.deinit(alloc);
}
};
fn getConnCb(ref: [*c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?*c.ngtcp2_conn {
const cn: *Conn = @ptrCast(@alignCast(ref.*.user_data));
return cn.conn;
}
/// The handler a listener carries between `bind` and `setHandler`: it
/// exists so that window has no null to check on every packet.
fn ignoreOpen(_: *anyopaque, _: u64) void {}
fn ignoreData(_: *anyopaque, _: u64, _: []const u8) void {}
fn ignoreClose(_: *anyopaque, _: u64) void {}
fn cidEql(cid: *const c.ngtcp2_cid, bytes: []const u8) bool {
if (cid.datalen != bytes.len) return false;
return std.mem.eql(u8, cid.data[0..cid.datalen], bytes);
}
/// Separate from `getNewCidCb` because the test client shares that one and
/// its user_data is not a Conn. Marks the CID cache stale.
fn serverGetNewCidCb(
conn: ?*c.ngtcp2_conn,
cid: [*c]c.ngtcp2_cid,
token: [*c]c.ngtcp2_stateless_reset_token,
cidlen: usize,
user_data: ?*anyopaque,
) callconv(.c) c_int {
const rv = quic.getNewCidCb(conn, cid, token, cidlen, user_data);
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
cn.cids_dirty = true;
return rv;
}
/// A retired CID must stop matching, or this listener answers to a name
/// the peer forgot.
fn serverRemoveCidCb(
_: ?*c.ngtcp2_conn,
_: [*c]const c.ngtcp2_cid,
user_data: ?*anyopaque,
) callconv(.c) c_int {
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
cn.cids_dirty = true;
return 0;
}
fn handshakeCompletedCb(_: ?*c.ngtcp2_conn, user_data: ?*anyopaque) callconv(.c) c_int {
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
cn.opened = true;
cn.listener.handler.onOpen(cn.listener.handler.ctx, cn.id);
return 0;
}
/// The ONLY thing that frees egress space; registering it is what makes
/// the no-move invariant affordable.
fn ackedStreamDataCb(
_: ?*c.ngtcp2_conn,
_: i64,
_: u64,
datalen: u64,
user_data: ?*anyopaque,
_: ?*anyopaque,
) callconv(.c) c_int {
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
cn.out.ack(@intCast(datalen));
return 0;
}
fn streamOpenCb(_: ?*c.ngtcp2_conn, stream_id: i64, user_data: ?*anyopaque) callconv(.c) c_int {
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
// One stream per connection: the first the client opens is the session.
if (cn.stream_id == -1) cn.stream_id = stream_id;
return 0;
}
fn recvStreamDataCb(
conn: ?*c.ngtcp2_conn,
_: u32,
stream_id: i64,
_: u64,
data: [*c]const u8,
datalen: usize,
user_data: ?*anyopaque,
_: ?*anyopaque,
) callconv(.c) c_int {
const cn: *Conn = @ptrCast(@alignCast(user_data.?));
// MANDATORY: ngtcp2 never extends its own windows. Consume without these
// two and the peer stops the moment it has sent the initial window, with no
// error on either side — it presents as a freeze.
_ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen);
_ = c.ngtcp2_conn_extend_max_offset(conn, datalen);
// The window is extended BEFORE the owner has touched these bytes, so the
// credit is granted against a consume that has not happened. Safe while
// `onData` is synchronous and bounded by the daemon's own cap; it stops
// being safe the moment an owner can hold bytes indefinitely.
if (datalen > 0 and cn.close_state == .open) {
cn.listener.handler.onData(cn.listener.handler.ctx, cn.id, data[0..datalen]);
}
return 0;
}
/// The UDP socket, the connection table, and the TLS context they share.
pub const Listener = struct {
alloc: std.mem.Allocator,
fd: std.posix.fd_t,
ssl_ctx: ?*c.WOLFSSL_CTX,
handler: Handler,
conns: [max_conns]?*Conn = @splat(null),
next_id: u64 = 1,
/// Secret behind Retry tokens. Random per daemon: a token from a
/// previous run must not validate against this one.
retry_secret: [32]u8,
idle_ms: u64,
/// Nonzero while ngtcp2 owns the stack — inside `ngtcp2_conn_read_pkt`.
///
/// A connection MUST NOT be freed in that window: after our receive
/// callback ngtcp2 calls `conn_emit_pending_stream_data`, whose FIRST
/// statement dereferences the connection. Our callback reaches the daemon,
/// which may decide the client is finished — a `.detach` frame is exactly
/// that — so freeing there is a use-after-free on every receive.
///
/// A DEPTH rather than a flag: as a bool it is balanced only by there being
/// no early return between the set and the clear, and one added `return`
/// would leave the listener permanently believing ngtcp2 owns the stack.
ngtcp2_depth: u8 = 0,
/// Bound BEFORE the daemon's socket: a refused UDP port must not cost a
/// live shell.
pub fn bind(
alloc: std.mem.Allocator,
bind_addr: std.net.Address,
key: quic.Key,
idle_ms: u64,
) !*Listener {
return init(alloc, bind_addr, key, .{
.ctx = undefined,
.onOpen = ignoreOpen,
.onData = ignoreData,
.onClose = ignoreClose,
}, idle_ms);
}
pub fn setHandler(self: *Listener, h: Handler) void {
self.handler = h;
}
/// Anything that frees or writes a connection has to ask.
fn inNgtcp2(self: *const Listener) bool {
return self.ngtcp2_depth > 0;
}
pub fn init(
alloc: std.mem.Allocator,
bind_addr: std.net.Address,
key: quic.Key,
handler: Handler,
idle_ms: u64,
) !*Listener {
if (g_listener_live) return error.ListenerAlreadyRunning;
const fd = try std.posix.socket(
bind_addr.any.family,
std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
0,
);
errdefer std.posix.close(fd);
// Deliberately NO SO_REUSEADDR: on UDP it lets a second daemon bind the
// same address, and the kernel then splits datagrams between them —
// two sessions silently sharing one port. Fail the bind, loudly.
try std.posix.bind(fd, &bind_addr.any, bind_addr.getOsSockLen());
return finishInit(alloc, fd, key, handler, idle_ms);
}
// Adopt a bound UDP fd from another process (the upgrade exec path).
// Refuses unless the fd is a datagram socket: a stream fd would
// accept a handshake and then lose every packet to recvfrom.
pub fn initFromFd(
alloc: std.mem.Allocator,
fd: std.posix.fd_t,
key: quic.Key,
handler: Handler,
idle_ms: u64,
) !*Listener {
if (g_listener_live) return error.ListenerAlreadyRunning;
if ((server_os.sockType(fd) catch return error.NotAUdpSocket) != std.posix.SOCK.DGRAM) return error.NotAUdpSocket;
return finishInit(alloc, fd, key, handler, idle_ms);
}
// The runtime bound address. lazyBindQuic can own a kernel-assigned
// port no flag names; the manifest records this.
pub fn boundAddr(self: *const Listener) std.net.Address {
var storage: std.posix.sockaddr.storage = undefined;
var len: std.posix.socklen_t = @sizeOf(@TypeOf(storage));
std.posix.getsockname(self.fd, @ptrCast(&storage), &len) catch
return std.net.Address.initIp4(.{ 0, 0, 0, 0 }, 0);
return std.net.Address.initPosix(@ptrCast(@alignCast(&storage)));
}
// The key the manifest carries as bytes. The PSK callback reads it
// from this module global, not from Server.
pub fn currentKey() ?quic.Key {
return g_key;
}
/// wolfSSL/TLS setup and Listener allocation shared by init and
/// initFromFd. The fd is already bound.
fn finishInit(
alloc: std.mem.Allocator,
fd: std.posix.fd_t,
key: quic.Key,
handler: Handler,
idle_ms: u64,
) !*Listener {
if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit;
const ctx = c.wolfSSL_CTX_new(c.wolfTLSv1_3_server_method()) orelse return error.TlsInit;
errdefer c.wolfSSL_CTX_free(ctx);
if (c.ngtcp2_crypto_wolfssl_configure_server_context(ctx) != 0) return error.TlsInit;
c.wolfSSL_CTX_set_psk_server_tls13_callback(ctx, pskServerCb);
_ = c.wolfSSL_CTX_use_psk_identity_hint(ctx, "");
_ = c.wolfSSL_CTX_set_cipher_list(ctx, quic.psk_ciphersuite);
const self = try alloc.create(Listener);
errdefer alloc.destroy(self);
self.* = .{
.alloc = alloc,
.fd = fd,
.ssl_ctx = ctx,
.handler = handler,
.retry_secret = undefined,
.idle_ms = idle_ms,
};
std.crypto.random.bytes(&self.retry_secret);
g_key = key;
g_listener_live = true;
return self;
}
pub fn deinit(self: *Listener) void {
for (&self.conns) |*slot| {
if (slot.*) |cn| {
cn.deinit(self.alloc);
self.alloc.destroy(cn);
slot.* = null;
}
}
if (self.ssl_ctx) |ctx| c.wolfSSL_CTX_free(ctx);
std.posix.close(self.fd);
g_listener_live = false;
g_key = null;
self.alloc.destroy(self);
}
/// One socket for every peer — a client slot cannot be an fd.
pub fn pollFd(self: *const Listener) std.posix.fd_t {
return self.fd;
}
fn find(self: *Listener, id: u64) ?*Conn {
for (self.conns) |slot| {
if (slot) |cn| {
if (cn.id == id) return cn;
}
}
return null;
}
/// QUEUE ONLY; a short return is backpressure. ngtcp2 is not re-entrant,
/// and draining from inside its stream callback aborts. `drainAll` drains.
pub fn send(self: *Listener, id: u64, bytes: []const u8) !usize {
const cn = self.find(id) orelse return error.NoSuchConn;
return cn.out.push(bytes);
}
/// Push every connection's egress. Called once per pump after frame
/// handling, so queue-only costs no latency.
pub fn drainAll(self: *Listener) void {
for (&self.conns) |*slot| {
const cn = slot.* orelse continue;
self.drain(cn);
}
}
/// Bytes accepted from the owner but not acknowledged. A QUIC send is
/// done at the ack, not at the syscall.
pub fn pendingBytes(self: *Listener, id: u64) usize {
const cn = self.find(id) orelse return 0;
return cn.out.held;
}
/// Close ONE connection and never the socket it shares: every peer is
/// multiplexed over one UDP socket. It does NOT invoke `onClose` — calling
/// back mid-teardown is how re-entrancy bugs start — and does not send
/// CONNECTION_CLOSE, so the peer finds out when its idle timer expires.
pub fn closeConn(self: *Listener, id: u64) void {
for (&self.conns) |*slot| {
if (slot.*) |cn| {
if (cn.id != id) continue;
// Deferred if ngtcp2 is mid-call: it dereferences `conn` again
// after our callback returns. `feed` reaps on the way out,
// quietly — a close the owner asked for gets no callback.
if (self.inNgtcp2()) {
cn.close_state = .closing_quiet;
return;
}
cn.deinit(self.alloc);
self.alloc.destroy(cn);
slot.* = null;
return;
}
}
}
// Loud goodbye: CONNECTION_CLOSE on every live conn, drained once.
// closeConn is quiet — the peer learns from the idle timer. An upgrade
// exec drops the socket before that timer fires; a WAN client would
// wait out its backoff. This makes the close immediate.
pub fn closeAll(self: *Listener) void {
for (&self.conns) |*slot| {
const cn = slot.* orelse continue;
const conn = cn.conn orelse continue;
var ccerr: c.ngtcp2_ccerr = undefined;
c.ngtcp2_ccerr_default(&ccerr);
var buf: [quic.max_udp]u8 = undefined;
var ps: c.ngtcp2_path_storage = undefined;
c.ngtcp2_path_storage_zero(&ps);
var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
const n = c.ngtcp2_conn_write_connection_close_versioned(
conn,
&ps.path,
c.NGTCP2_PKT_INFO_VERSION,
&pi,
&buf,
buf.len,
&ccerr,
quic.timestampNs(),
);
if (n > 0) {
_ = std.posix.sendto(
self.fd,
buf[0..@intCast(n)],
0,
@ptrCast(&cn.remote),
cn.remote_len,
) catch {};
}
}
// One drain pass: clears the egress ring before exec.
for (&self.conns) |*slot| {
if (slot.*) |cn| self.drain(cn);
}
}
/// Free every connection that asked to go while ngtcp2 held the stack.
fn reapClosing(self: *Listener) void {
for (&self.conns) |*slot| {
const cn = slot.* orelse continue;
if (cn.close_state == .open) continue;
const notify = cn.close_state == .closing_notify and cn.opened;
const id = cn.id;
// One last push before it goes: anything the owner queued while
// deciding to close — the session-full refusal — was only queued,
// and this is its one chance to reach the wire.
self.drain(cn);
cn.deinit(self.alloc);
self.alloc.destroy(cn);
slot.* = null;
if (notify) self.handler.onClose(self.handler.ctx, id);
}
}
/// An unknown connection ID. The address is unvalidated, so amplification
/// protection lives here.
fn accept(
self: *Listener,
pkt: []const u8,
from: *std.posix.sockaddr.storage,
from_len: std.posix.socklen_t,
) void {
var hd: c.ngtcp2_pkt_hd = undefined;
if (c.ngtcp2_accept(&hd, pkt.ptr, pkt.len) != 0) return;
// No token: answer with a Retry and nothing else. Until the peer echoes
// a token we minted for its address there is no evidence the address is
// real, and handing out handshake bytes anyway is an amplification
// reflector. Sent before any conn state, so a flood costs no memory.
if (hd.tokenlen == 0) {
self.sendRetry(&hd, from, from_len);
return;
}
// The original DCID the client first used, recovered from the token
// we minted — it goes into transport params so the client can verify
// the Retry was ours and not an injection.
var odcid: c.ngtcp2_cid = undefined;
if (c.ngtcp2_crypto_verify_retry_token(
&odcid,
hd.token,
hd.tokenlen,
&self.retry_secret,
self.retry_secret.len,
hd.version,
@ptrCast(from),
from_len,
&hd.dcid,
60 * 1_000_000_000, // a token is good for 60s
quic.timestampNs(),
) != 0) return;
const slot = for (&self.conns) |*sl| {
if (sl.* == null) break sl;
} else return; // full: drop, the peer will retry
const cn = self.alloc.create(Conn) catch return;
const ring = self.alloc.alloc(u8, quic.egress_cap) catch {
self.alloc.destroy(cn);
return;
};
cn.* = .{ .id = self.next_id, .listener = self, .out = .{ .buf = ring } };
self.next_id += 1;
@memcpy(std.mem.asBytes(&cn.remote)[0..from_len], std.mem.asBytes(from)[0..from_len]);
cn.remote_len = from_len;
cn.local_len = @sizeOf(std.posix.sockaddr.storage);
std.posix.getsockname(self.fd, @ptrCast(&cn.local_storage), &cn.local_len) catch {
cn.out.deinit(self.alloc);
self.alloc.destroy(cn);
return;
};
// The source CID must be the one the client is already addressing — the
// SCID we put in the Retry, i.e. this Initial's DCID. A fresh one leaves
// ngtcp2 owning a CID nobody sends to, and the handshake stalls.
cn.scid = hd.dcid;
const ssl = c.wolfSSL_new(self.ssl_ctx) orelse {
cn.out.deinit(self.alloc);
self.alloc.destroy(cn);
return;
};
cn.ssl = ssl;
cn.conn_ref = .{ .get_conn = getConnCb, .user_data = cn };
_ = c.wolfSSL_set_app_data(ssl, &cn.conn_ref);
var cbs: c.ngtcp2_callbacks = std.mem.zeroes(c.ngtcp2_callbacks);
cbs.recv_client_initial = c.ngtcp2_crypto_recv_client_initial_cb;
cbs.recv_crypto_data = c.ngtcp2_crypto_recv_crypto_data_cb;
cbs.encrypt = c.ngtcp2_crypto_encrypt_cb;
cbs.decrypt = c.ngtcp2_crypto_decrypt_cb;
cbs.hp_mask = c.ngtcp2_crypto_hp_mask_cb;
cbs.update_key = c.ngtcp2_crypto_update_key_cb;
cbs.delete_crypto_aead_ctx = c.ngtcp2_crypto_delete_crypto_aead_ctx_cb;
cbs.delete_crypto_cipher_ctx = c.ngtcp2_crypto_delete_crypto_cipher_ctx_cb;
cbs.get_path_challenge_data = c.ngtcp2_crypto_get_path_challenge_data_cb;
cbs.version_negotiation = c.ngtcp2_crypto_version_negotiation_cb;
cbs.rand = quic.randCb;
cbs.get_new_connection_id2 = serverGetNewCidCb;
cbs.remove_connection_id = serverRemoveCidCb;
cbs.handshake_completed = handshakeCompletedCb;
cbs.stream_open = streamOpenCb;
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 = quic.timestampNs();
settings.token = hd.token;
settings.tokenlen = hd.tokenlen;
var params: c.ngtcp2_transport_params = undefined;
c.ngtcp2_transport_params_default_versioned(c.NGTCP2_TRANSPORT_PARAMS_VERSION, ¶ms);
params.initial_max_streams_bidi = 4;
params.initial_max_stream_data_bidi_local = 256 * 1024;
params.initial_max_stream_data_bidi_remote = 256 * 1024;
params.initial_max_data = 1024 * 1024;
params.original_dcid = odcid;
params.original_dcid_present = 1;
params.retry_scid = hd.dcid;
params.retry_scid_present = 1;
// Idle timeout is tunable because the reconnect tests need a dead
// transport to be declared dead quickly and deterministically.
params.max_idle_timeout = self.idle_ms * 1_000_000;
var path = quic.pathFrom(&cn.local_storage, cn.local_len, from, from_len);
var conn: ?*c.ngtcp2_conn = null;
if (c.ngtcp2_conn_server_new_versioned(
&conn,
&hd.scid,
&cn.scid,
&path,
hd.version,
c.NGTCP2_CALLBACKS_VERSION,
&cbs,
c.NGTCP2_SETTINGS_VERSION,
&settings,
c.NGTCP2_TRANSPORT_PARAMS_VERSION,
¶ms,
null,
cn,
) != 0) {
c.wolfSSL_free(ssl);
cn.out.deinit(self.alloc);
self.alloc.destroy(cn);
return;
}
cn.conn = conn;
c.ngtcp2_conn_set_tls_native_handle(conn, ssl);
// Without this, a terminal nobody is typing into is indistinguishable
// from a peer that has gone away, and a session dies while its owner
// reads the screen. The keep-alive PING is ack-eliciting, so a peer that
// has genuinely vanished still times out on schedule.
c.ngtcp2_conn_set_keep_alive_timeout(conn, quic.keepAliveNs(self.idle_ms));
slot.* = cn;
self.feed(slot, pkt, from, from_len);
}
fn sendRetry(
self: *Listener,
hd: *const c.ngtcp2_pkt_hd,
from: *std.posix.sockaddr.storage,
from_len: std.posix.socklen_t,
) void {
var scid: c.ngtcp2_cid = undefined;
scid.datalen = 8;
std.crypto.random.bytes(scid.data[0..8]);
var token: [c.NGTCP2_CRYPTO_MAX_RETRY_TOKENLEN]u8 = undefined;
const tlen = c.ngtcp2_crypto_generate_retry_token(
&token,
&self.retry_secret,
self.retry_secret.len,
hd.version,
@ptrCast(from),
from_len,
&scid,
&hd.dcid,
quic.timestampNs(),
);
if (tlen < 0) return;
var buf: [quic.max_udp]u8 = undefined;
const n = c.ngtcp2_crypto_write_retry(
&buf,
buf.len,
hd.version,
&hd.scid,
&scid,
&hd.dcid,
&token,
@intCast(tlen),
);
if (n < 0) return;
_ = std.posix.sendto(self.fd, buf[0..@intCast(n)], 0, @ptrCast(from), from_len) catch {};
}
/// Milliseconds until the earliest thing ngtcp2 wants doing, folded
/// across every connection — the MINIMUM, because servicing one
/// connection late is servicing it wrong. Feeds the daemon's existing
/// poll timeout; no timerfd is needed and that is deliberate.
pub fn timeoutMs(self: *Listener, cap_ms: i32) i32 {
// A negative cap is poll(2)'s "wait forever", and the @intCast below
// would panic on it rather than honour it. No caller passes one
// today; this costs a compare and stops that being load-bearing.
if (cap_ms < 0) return cap_ms;
var best: i32 = cap_ms;
const now = quic.timestampNs();
for (self.conns) |slot| {
const cn = slot orelse continue;
best = @min(best, quic.expiryMs(cn.conn, cap_ms, now));
}
return best;
}
/// Service every connection whose deadline has passed. Called by the
/// daemon after each poll, whether or not the socket was readable.
pub fn tick(self: *Listener) void {
const now = quic.timestampNs();
for (&self.conns) |*slot| {
const cn = slot.* orelse continue;
const conn = cn.conn orelse continue;
if (c.ngtcp2_conn_get_expiry(conn) > now) continue;
if (c.ngtcp2_conn_handle_expiry(conn, now) != 0) {
self.kill(slot);
continue;
}
self.drain(cn);
}
}
/// The socket is readable: take everything queued and route it.
pub fn readable(self: *Listener) void {
var buf: [65536]u8 = undefined;
while (true) {
var from: std.posix.sockaddr.storage = undefined;
var from_len: std.posix.socklen_t = @sizeOf(@TypeOf(from));
const n = std.posix.recvfrom(
self.fd,
&buf,
0,
@ptrCast(&from),
&from_len,
) catch |err| switch (err) {
error.WouldBlock => return,
else => return,
};
if (n == 0) continue;
self.route(buf[0..n], &from, from_len);
}
}
fn route(
self: *Listener,
pkt: []const u8,
from: *std.posix.sockaddr.storage,
from_len: std.posix.socklen_t,
) void {
var vc: c.ngtcp2_version_cid = undefined;
const rv = c.ngtcp2_pkt_decode_version_cid(&vc, pkt.ptr, pkt.len, 8);
if (rv != 0) return;
// Match on ANY Connection ID this endpoint has advertised, not just the
// one the connection was created with: a migrating client switches to
// another CID we gave it (RFC 9000 §9.5), and matching only the first
// sends those packets to `accept`, which drops them.
for (&self.conns) |*slot| {
const cn = slot.* orelse continue;
if (!cn.matches(vc.dcid[0..vc.dcidlen])) continue;
self.feed(slot, pkt, from, from_len);
return;
}
self.accept(pkt, from, from_len);
}
fn feed(
self: *Listener,
slot: *?*Conn,
pkt: []const u8,
from: *std.posix.sockaddr.storage,
from_len: std.posix.socklen_t,
) void {
const cn = slot.* orelse return;
const conn = cn.conn orelse return;
var path = quic.pathFrom(&cn.local_storage, cn.local_len, from, from_len);
const rv = quic.readPkt(conn, &path, pkt, &self.ngtcp2_depth);
// Reaped before anything else touches the table: a connection the
// handler closed mid-callback is gone from here on.
self.reapClosing();
// Recomputed AFTER the reap and re-read from the SLOT, not the `cn`
// above: `reapClosing` calls `onClose` at depth zero, so a handler that
// closes another connection frees it immediately — and that other one
// may be this one.
const closed = slot.* == null or slot.*.?.close_state != .open;
if (rv != 0) {
if (!closed) self.kill(slot);
return;
}
if (closed) return;
self.drain(slot.*.?);
}
/// Drive egress until ngtcp2 has nothing more to send. Runs after every
/// event, which is the contract the library expects.
pub fn drain(self: *Listener, cn: *Conn) void {
// The invariant the queue-only send exists to keep, stated where it
// can be enforced. A future caller that drains from inside an ngtcp2
// callback fails here deterministically in Debug instead of aborting
// one run in a few hundred somewhere else entirely.
std.debug.assert(self.ngtcp2_depth == 0);
const conn = cn.conn orelse return;
// No refusal check, unlike `quic.Client`'s: this socket is UNCONNECTED,
// so the kernel has no peer to attribute an ICMP unreachable to. The
// asymmetry is the sockets', which is why `sendPkt` is the seam.
const To = struct { l: *Listener, cn: *Conn };
const sendPkt = struct {
fn f(ctx: To, pkt: []const u8) bool {
_ = std.posix.sendto(
ctx.l.fd,
pkt,
0,
@ptrCast(&ctx.cn.remote),
ctx.cn.remote_len,
) catch return false;
return true;
}
}.f;
_ = quic.drainConn(conn, &cn.out, cn.stream_id, To{ .l = self, .cn = cn }, sendPkt);
}
fn kill(self: *Listener, slot: *?*Conn) void {
const cn = slot.* orelse return;
// Same hazard as closeConn, and reached the same way: ngtcp2 is
// still using this connection. Marked to be reaped WITH its
// callback, since a kill is the listener giving up rather than the
// owner asking.
if (self.inNgtcp2()) {
cn.close_state = .closing_notify;
return;
}
const id = cn.id;
const opened = cn.opened;
cn.deinit(self.alloc);
self.alloc.destroy(cn);
slot.* = null;
if (opened) self.handler.onClose(self.handler.ctx, id);
}
};
// ---------------------------------------------------------------------------
// Tests: a real handshake against a real client, in one process. A QUIC
// handshake either completes against a real peer or it does not, so these dial
// the SHIPPING client, wrapped for the two conveniences a test wants.
// ---------------------------------------------------------------------------
/// The shipping `quic.Client`, plus the two things only a test wants: a backlog
/// it re-offers as the ring drains — `send` takes what fits, and a payload over
/// `egress_cap` needs somebody to hold the rest — and a value a stack `defer`
/// can deinit.
pub const TestPeer = struct {
cl: *quic.Client,
out: []const u8 = &.{},
out_sent: usize = 0,
/// Idle 30s, not the listener's: a peer that PINGed on its own
/// schedule would answer the keepalive test's question for it.
pub fn init(addr: std.net.Address, key: quic.Key) !TestPeer {
return .{ .cl = try quic.Client.connect(std.testing.allocator, addr, key, 30_000) };
}
pub fn deinit(self: *TestPeer) void {
self.cl.deinit();
}
/// Bytes to hand the ring, from the top, as room appears.
pub fn offer(self: *TestPeer, bytes: []const u8) void {
self.out = bytes;
self.out_sent = 0;
}
/// One service pass: top the ring up, then read, tick and write.
pub fn drain(self: *TestPeer) void {
if (self.out_sent < self.out.len) self.out_sent += self.cl.send(self.out[self.out_sent..]);
self.cl.pump();
}
/// Nothing consumes, so this counts and `cl.in.items` is the bytes.
pub fn echoed(self: *const TestPeer) usize {
return self.cl.in.items.len;
}
};
/// Test handler: echoes whatever arrives straight back through the listener,
/// which is enough to prove bytes cross the seam in both directions.
const EchoOwner = struct {
listener: *Listener = undefined,
alloc: std.mem.Allocator = std.testing.allocator,
id: u64 = 0,
opened: usize = 0,
closed: usize = 0,
received: usize = 0,
/// What the ring would not take yet. The daemon keeps exactly this
/// queue for exactly this reason: `send` takes what fits and the owner
/// holds — and bounds — the rest.
backlog: std.ArrayList(u8) = .empty,
fn deinit(self: *EchoOwner) void {
self.backlog.deinit(self.alloc);
}
fn onOpen(ctx: *anyopaque, id: u64) void {
const self: *EchoOwner = @ptrCast(@alignCast(ctx));
self.opened += 1;
self.id = id;
}
fn onData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
const self: *EchoOwner = @ptrCast(@alignCast(ctx));
self.received += bytes.len;
self.id = id;
self.backlog.appendSlice(self.alloc, bytes) catch return;
self.flush();
}
fn onClose(ctx: *anyopaque, _: u64) void {
const self: *EchoOwner = @ptrCast(@alignCast(ctx));
self.closed += 1;
}
/// Offer the backlog again. Called on every pump iteration as well as on
/// arrival, because the room to accept it comes from acks, which arrive
/// on their own schedule.
fn flush(self: *EchoOwner) void {
if (self.backlog.items.len == 0) return;
const n = self.listener.send(self.id, self.backlog.items) catch return;
if (n == 0) return;
self.backlog.replaceRangeAssumeCapacity(0, n, &.{});
}
fn handler(self: *EchoOwner) Handler {
return .{ .ctx = self, .onOpen = onOpen, .onData = onData, .onClose = onClose };
}
};
/// Drive both ends until `done` or the deadline. Single-threaded on purpose:
/// a test that needs threads to make a handshake happen is a test that will
/// wedge one day.
fn pump(l: *Listener, cl: *TestPeer, ms: u64, done: *const fn (*EchoOwner, *TestPeer) bool, owner: *EchoOwner) bool {
var waited: u64 = 0;
while (waited < ms) {
if (done(owner, cl)) return true;
var fds = [_]std.posix.pollfd{
.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&fds, 10) catch return false;
if (ready > 0) {
if (fds[0].revents != 0) l.readable();
} else {
waited += 10;
}
l.tick();
// The owner re-offers its backlog every iteration: acks free ring
// space on their own schedule, and nothing else would notice.
owner.flush();
cl.drain();
for (l.conns) |slot| {
if (slot) |cn| l.drain(cn);
}
}
return done(owner, cl);
}
fn loopbackListener(
alloc: std.mem.Allocator,
key: quic.Key,
owner: *EchoOwner,
idle_ms: u64,
) !struct { l: *Listener, addr: std.net.Address } {
const bind = try std.net.Address.parseIp("127.0.0.1", 0);
const l = try Listener.init(alloc, bind, key, owner.handler(), idle_ms);
owner.listener = l;
var actual: std.posix.sockaddr.storage = undefined;
var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
try std.posix.getsockname(l.fd, @ptrCast(&actual), &len);
return .{ .l = l, .addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual))) };
}
test "Listener: PSK handshake, Retry, and a payload larger than the initial window" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x5A} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, 5000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
// Both sides, not just the client: the client declares the handshake
// complete one flight before the server does, so waiting only on the
// client stops the pump while the server's final flight is still in the
// air.
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened >= 1;
}
}.f, &owner));
// The handshake only completed because the PSK matched on both sides:
// there is no certificate anywhere in this exchange to fall back on.
try std.testing.expectEqual(@as(usize, 1), owner.opened);
// A Retry happened before any of it. The client's original DCID is not
// what the server ended up using, which is the observable trace of
// address validation having taken place.
try std.testing.expect(setup.l.conns[0] != null);
// The freeze-shaped case. The size must exceed the windows this listener
// advertises (256KB per stream, 1MB per connection), because those are what
// the peer stops at when nobody extends them — 200KB fits inside both and
// passes with flow control disabled. Without the extends this does not
// fail, it STOPS, which is the shape of the bug in production.
const payload = try alloc.alloc(u8, 3 * 1024 * 1024);
defer alloc.free(payload);
// A repeating pattern, compared byte for byte and not merely counted:
// counting cannot tell a working transport from one delivering the right
// NUMBER of the wrong bytes, out of a buffer recycled underneath it.
for (payload, 0..) |*b, i| b.* = @truncate(i);
cl.offer(payload);
cl.drain();
const ok = pump(setup.l, &cl, 30_000, struct {
fn f(_: *EchoOwner, t: *TestPeer) bool {
return t.echoed() >= 3 * 1024 * 1024;
}
}.f, &owner);
if (!ok) {
std.debug.print(
"stalled: server saw {d} bytes, client got {d} of {d} back\n",
.{ owner.received, cl.echoed(), payload.len },
);
}
try std.testing.expect(ok);
try std.testing.expectEqual(@as(usize, 3 * 1024 * 1024), owner.received);
try std.testing.expectEqualSlices(u8, payload, cl.cl.in.items);
// The one place a LIVE connection with a finite expiry exists: negative is
// poll(2)'s "wait forever" and the fold must hand it back rather than
// `@intCast` it. It needs a real connection, or the loop never reaches the cast.
try std.testing.expectEqual(@as(i32, -1), setup.l.timeoutMs(-1));
// The ordinary case still folds as before.
try std.testing.expect(setup.l.timeoutMs(100) <= 100);
}
test "Listener: a client holding the wrong key never completes a handshake" {
const alloc = std.testing.allocator;
const server_key: quic.Key = .{ .bytes = [_]u8{0x11} ** quic.key_len };
const wrong_key: quic.Key = .{ .bytes = [_]u8{0x22} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, server_key, &owner, 5000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, wrong_key);
defer cl.deinit();
// Deliberately generous: the point is that it never succeeds, not that
// it fails fast. A shorter window could pass for the wrong reason.
_ = pump(setup.l, &cl, 3000, struct {
fn f(_: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done;
}
}.f, &owner);
try std.testing.expect(!cl.cl.handshake_done);
try std.testing.expectEqual(@as(usize, 0), owner.opened);
try std.testing.expectEqual(@as(usize, 0), owner.received);
// ...and it failed on the KEY, which is stronger than "it failed": a client
// refused earlier would satisfy every assertion above while proving nothing
// about authentication. An id is only allocated once a token-bearing Initial
// is accepted, so one having been handed out is the witness.
try std.testing.expect(setup.l.next_id > 1);
}
test "Listener: keepalive carries an idle connection past its idle timeout" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x3C} ** quic.key_len };
// Short enough that the test is quick, long enough that a handshake
// fits inside it: the keepalive lands at 300ms, the timeout at 900ms.
const idle_ms = 900;
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, idle_ms);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f, &owner));
// Now say nothing at all for three times the idle timeout. `pump` only
// counts quiet polls toward its budget, so the wall-clock silence is at
// least this long — the keepalive PINGs it does carry are the whole
// point, and they are not application traffic.
_ = pump(setup.l, &cl, idle_ms * 3, struct {
fn f(_: *EchoOwner, _: *TestPeer) bool {
return false;
}
}.f, &owner);
// Nothing timed out...
try std.testing.expectEqual(@as(usize, 0), owner.closed);
// ...and the connection is not merely un-reaped but still usable, which
// a live conn struct on its own would not prove.
cl.offer("still-here");
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(_: *EchoOwner, t: *TestPeer) bool {
return t.echoed() >= "still-here".len;
}
}.f, &owner));
try std.testing.expectEqual(@as(usize, 0), owner.closed);
}
/// Get the peer's stream on the record. The server learns a stream id only when
/// data ARRIVES on it, so a test that wants the server to send first must make
/// the client speak first. Waits for the ack, so the ring is empty again.
fn openStream(l: *Listener, cl: *TestPeer, owner: *EchoOwner) !void {
cl.offer("hi");
cl.drain();
const ok = pumpUntil(l, cl, owner, 10_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return o.received >= 2 and t.echoed() >= 2 and l_pending(o) == 0;
}
fn l_pending(o: *EchoOwner) usize {
return o.listener.pendingBytes(o.id);
}
}.f);
if (!ok) return error.StreamNeverOpened;
cl.offer(&.{});
cl.cl.in.clearRetainingCapacity();
}
/// `pump` counts only quiet iterations toward its deadline, which is wrong
/// for a bulk transfer that is busy throughout.
fn pumpUntil(
l: *Listener,
cl: *TestPeer,
owner: *EchoOwner,
wall_ms: i64,
done: *const fn (*EchoOwner, *TestPeer) bool,
) bool {
const deadline = std.time.milliTimestamp() + wall_ms;
while (std.time.milliTimestamp() < deadline) {
if (done(owner, cl)) return true;
var fds = [_]std.posix.pollfd{
.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&fds, 5) catch return false;
if (ready > 0) {
if (fds[0].revents != 0) l.readable();
}
l.tick();
owner.flush();
cl.drain();
for (l.conns) |slot| {
if (slot) |cn| l.drain(cn);
}
}
return done(owner, cl);
}
test "Listener: bytes survive retransmission, which is what the buffer is for" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x6E} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, 10_000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
// A deliberately tiny receive buffer, which is how loopback is made to lose
// packets. Loss is the ONLY way into ngtcp2's retransmission path, which
// re-reads bytes through a pointer handed over packets ago — so it is the
// only way to catch a buffer that recycled them underneath it.
try std.posix.setsockopt(
cl.cl.fd,
std.posix.SOL.SOCKET,
std.posix.SO.RCVBUF,
&std.mem.toBytes(@as(c_int, 64 * 1024)),
);
try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f));
// Twice the egress ring, so the ring wraps and every byte's release
// depends on an ack that may itself have been for a retransmission.
const size = 2 * quic.egress_cap;
const payload = try alloc.alloc(u8, size);
defer alloc.free(payload);
for (payload, 0..) |*b, i| b.* = @truncate(i *% 31 +% 7);
cl.offer(payload);
const ok = pumpUntil(setup.l, &cl, &owner, 60_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
_ = o;
return t.echoed() >= t.out.len;
}
}.f);
try std.testing.expect(ok);
// Count AND content. A transport that lost a retransmission and carried
// on would land the right number of bytes in the wrong order, which no
// length assertion can see.
try std.testing.expectEqual(size, cl.echoed());
try std.testing.expectEqualSlices(u8, payload, cl.cl.in.items);
try std.testing.expectEqual(size, owner.received);
try std.testing.expectEqual(@as(usize, 0), owner.closed);
}
test "Listener.send: takes what fits, refuses when full, and recovers on acks" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x21} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, 10_000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f));
try openStream(setup.l, &cl, &owner);
const big = try alloc.alloc(u8, quic.egress_cap + 4096);
defer alloc.free(big);
@memset(big, 0x5A);
// Nobody is pumping the client here, so nothing is acknowledged or released:
// the first send fills the ring exactly and reports the short count, and the
// daemon keeps the remainder where `pending_cap` can see it.
const first = try setup.l.send(owner.id, big);
try std.testing.expectEqual(quic.egress_cap, first);
try std.testing.expectEqual(@as(usize, 0), try setup.l.send(owner.id, "x"));
try std.testing.expectEqual(quic.egress_cap, setup.l.pendingBytes(owner.id));
// Space comes back from ACKS, not from having handed bytes to ngtcp2. The
// last byte reaching the client is not the same event as the ring being free
// to reuse it, and conflating them is what this ring exists to stop.
try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 20_000, struct {
fn f(o: *EchoOwner, _: *TestPeer) bool {
return o.listener.pendingBytes(o.id) == 0;
}
}.f));
try std.testing.expectEqual(quic.egress_cap, cl.echoed());
try std.testing.expect(try setup.l.send(owner.id, "room again") > 0);
// An id nobody owns is an error, not a silent success.
try std.testing.expectError(error.NoSuchConn, setup.l.send(owner.id + 999, "x"));
}
test "Listener: a full peer window blocks the stream without stopping the ACKs" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x4B} ** quic.key_len };
// Short enough that a connection nobody is servicing dies inside the
// test, which is exactly the failure being guarded against: treating
// the blocked-stream return as fatal abandons the egress loop, and the
// ACKs and keepalives that share it never leave either.
const idle_ms = 1500;
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, idle_ms);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f));
try openStream(setup.l, &cl, &owner);
// The client consumes but never grants more window, so the peer's
// stream credit runs out and stays out.
cl.cl.extend_windows = false;
const big = try alloc.alloc(u8, 4 * quic.egress_cap);
defer alloc.free(big);
@memset(big, 0x33);
try owner.backlog.appendSlice(alloc, big);
owner.flush();
// Well past the idle timeout, with the stream blocked the entire time.
_ = pumpUntil(setup.l, &cl, &owner, idle_ms * 3, struct {
fn f(_: *EchoOwner, _: *TestPeer) bool {
return false;
}
}.f);
// The stream really did block — the client cannot have taken it all.
try std.testing.expect(cl.echoed() < big.len);
// ...and the connection is alive, which it can only be if packets that
// are not stream data kept flowing while it was blocked.
try std.testing.expectEqual(@as(usize, 0), owner.closed);
try std.testing.expect(setup.l.find(owner.id) != null);
}
test "Listener: a packet addressed to any advertised CID reaches its connection" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x9C} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, 10_000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f));
const cn = setup.l.find(owner.id).?;
cn.refreshCids();
// Non-vacuity first: with only one CID there is no second one to route
// by, and everything below would pass for the wrong reason.
try std.testing.expect(cn.ncids > 1);
var secondary: ?*c.ngtcp2_cid = null;
for (cn.cids[0..cn.ncids]) |*cid| {
if (!cidEql(cid, cn.scid.data[0..cn.scid.datalen])) {
secondary = cid;
break;
}
}
try std.testing.expect(secondary != null);
const sec = secondary.?;
try std.testing.expect(cn.matches(sec.data[0..sec.datalen]));
// A CID nobody advertised belongs to nobody.
const stranger_cid = [_]u8{0xEE} ** 8;
try std.testing.expect(!cn.matches(&stranger_cid));
// The observable is what happens to a packet that MISSES: it falls through
// to `accept`, which answers a token-less Initial with a Retry. A probe that
// receives nothing is proof of delivery to the connection; the same probe
// receiving a Retry for an unknown CID proves the packet was well-formed.
const probe = try std.posix.socket(
std.posix.AF.INET,
std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
0,
);
defer std.posix.close(probe);
const probe_bind = try std.net.Address.parseIp("127.0.0.1", 0);
try std.posix.bind(probe, &probe_bind.any, probe_bind.getOsSockLen());
var probe_addr: std.posix.sockaddr.storage = undefined;
var probe_len: std.posix.socklen_t = @sizeOf(@TypeOf(probe_addr));
try std.posix.getsockname(probe, @ptrCast(&probe_addr), &probe_len);
// The baseline: nothing has been routed, so nothing may arrive. It waits
// the same budget the real absence check does, so the two are the same
// question asked twice and a budget too short to matter would show up
// here as well.
var pkt: [1300]u8 = undefined;
try std.testing.expect(!probeGotAnything(probe, probe_absence_ms));
// Addressed to a CID this connection advertised: delivered, not
// answered. This is the test's central negative and it has to be a WAIT,
// not a glance — a read that returns before any datagram could have
// crossed loopback is true of a listener that answered wrongly.
buildInitial(&pkt, sec.data[0..sec.datalen]);
setup.l.route(&pkt, &probe_addr, probe_len);
try std.testing.expect(!probeGotAnything(probe, probe_absence_ms));
// Addressed to nobody: `accept` answers with a Retry, which is what
// every migrated packet used to get. The budget here is generous because
// a MISSING reply is the failure this checks for; how long the reply
// really takes is what `probe_absence_ms` is sized against.
buildInitial(&pkt, &stranger_cid);
setup.l.route(&pkt, &probe_addr, probe_len);
try std.testing.expect(probeGotAnything(probe, 2000));
}
/// A syntactically valid, cryptographically meaningless Initial packet
/// addressed to `dcid`. Enough for `ngtcp2_accept` to recognise it and
/// answer, which is all this needs to tell routing hit from routing miss.
fn buildInitial(pkt: *[1300]u8, dcid: []const u8) void {
@memset(pkt, 0);
var i: usize = 0;
pkt[i] = 0xC3; // long header, fixed bit, Initial, 4-byte packet number
i += 1;
std.mem.writeInt(u32, pkt[i..][0..4], 1, .big); // version 1
i += 4;
pkt[i] = @intCast(dcid.len);
i += 1;
@memcpy(pkt[i..][0..dcid.len], dcid);
i += dcid.len;
pkt[i] = 8; // source CID length
i += 1;
@memset(pkt[i..][0..8], 0x5A);
i += 8;
pkt[i] = 0; // token length, varint 0
i += 1;
// Length: a two-byte varint covering everything left.
const rest: u16 = @intCast(pkt.len - i - 2);
std.mem.writeInt(u16, pkt[i..][0..2], rest | 0x4000, .big);
}
/// True when a datagram arrives within `budget_ms`. Every caller waits,
/// including the ones asserting ABSENCE: loopback UDP is not synchronous
/// everywhere — Linux hands the datagram over inside the `sendto`, so an
/// immediate recv finds it, and Darwin does not. A read with no budget
/// therefore CANNOT SEE a Retry on a Mac, which would make "nothing came
/// back" true of a listener that answered wrongly. The absence budget is
/// what makes the negative a claim rather than a race the Mac always wins;
/// `probe_absence_ms` says how it was chosen.
fn probeGotAnything(fd: std.posix.fd_t, budget_ms: i32) bool {
var buf: [2048]u8 = undefined;
var fds = [_]std.posix.pollfd{.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }};
if (budget_ms > 0) _ = std.posix.poll(&fds, budget_ms) catch return false;
const n = std.posix.recv(fd, &buf, 0) catch return false;
return n > 0;
}
/// How long "nothing came back" waits before it is believed. Sized against
/// the presence check at the end of the same test, on the same socket:
/// timed on both systems 2026-09-03, a Retry this listener DOES send is
/// readable in under a millisecond once the reader yields to a poll. Half a
/// second is three orders of magnitude of slack, so a wrongly-emitted one has
/// had every chance to arrive. What it is NOT sized against is `sendto`
/// itself: Linux delivers inside the call and Darwin does not, which is why
/// a glance with no budget at all could never fail on a Mac.
const probe_absence_ms: i32 = 500;
test "Listener: a reply queued just before a close still reaches the peer" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x5C} ** quic.key_len };
// The session-full refusal in miniature: the owner answers and then shuts
// the connection from inside one callback. Since `send` only QUEUES, those
// bytes have one chance to leave — the drain `reapClosing` does before it
// frees the connection. Bounded by the peer's window: a refusal always fits.
const AnswerThenClose = struct {
listener: *Listener = undefined,
id: u64 = 0,
opened: usize = 0,
closed: usize = 0,
queued: usize = 0,
fn onOpen(ctx: *anyopaque, id: u64) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
self.opened += 1;
self.id = id;
}
fn onData(ctx: *anyopaque, id: u64, _: []const u8) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
if (self.queued > 0) return;
self.id = id;
self.queued = self.listener.send(id, "REFUSED-FULL") catch 0;
self.listener.closeConn(id);
}
fn onClose(ctx: *anyopaque, _: u64) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
self.closed += 1;
}
fn handler(self: *@This()) Handler {
return .{ .ctx = self, .onOpen = onOpen, .onData = onData, .onClose = onClose };
}
};
var owner: AnswerThenClose = .{};
const bind = try std.net.Address.parseIp("127.0.0.1", 0);
const l = try Listener.init(alloc, bind, key, owner.handler(), 10_000);
defer l.deinit();
owner.listener = l;
var actual: std.posix.sockaddr.storage = undefined;
var alen: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);
const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));
var cl = try TestPeer.init(addr, key);
defer cl.deinit();
cl.offer("attach");
var waited: u64 = 0;
while (waited < 10_000) : (waited += 5) {
var fds = [_]std.posix.pollfd{
.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
_ = std.posix.poll(&fds, 5) catch break;
if (fds[0].revents != 0) l.readable();
l.tick();
cl.drain();
for (l.conns) |slot| {
if (slot) |cn| l.drain(cn);
}
if (cl.echoed() >= "REFUSED-FULL".len) break;
}
try std.testing.expectEqual(@as(usize, "REFUSED-FULL".len), owner.queued);
// The bytes are on the peer, not merely in a ring that was then freed.
try std.testing.expect(std.mem.indexOf(
u8,
cl.cl.in.items,
"REFUSED-FULL",
) != null);
// ...and the connection really did go.
try std.testing.expect(l.find(owner.id) == null);
}
test "Listener: closing a connection from inside a receive callback is deferred" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x7E} ** quic.key_len };
// An owner that closes the connection from inside `onData`, as the daemon
// does on goodbye. ngtcp2 is on the stack and uses the connection after the
// callback returns, so freeing there is a use-after-free. This pins the
// DEFERRAL: still valid when the callback returns, gone by the next pump.
const CloseOnData = struct {
listener: *Listener = undefined,
id: u64 = 0,
opened: usize = 0,
closed: usize = 0,
received: usize = 0,
alive_after_callback: bool = false,
fn onOpen(ctx: *anyopaque, id: u64) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
self.opened += 1;
self.id = id;
}
fn onData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
self.received += bytes.len;
self.listener.closeConn(id);
// Still addressable: the free was deferred, not done. Reading
// this through the listener is exactly what ngtcp2 does next.
self.alive_after_callback = self.listener.find(id) != null;
}
fn onClose(ctx: *anyopaque, _: u64) void {
const self: *@This() = @ptrCast(@alignCast(ctx));
self.closed += 1;
}
fn handler(self: *@This()) Handler {
return .{ .ctx = self, .onOpen = onOpen, .onData = onData, .onClose = onClose };
}
};
var owner: CloseOnData = .{};
const bind = try std.net.Address.parseIp("127.0.0.1", 0);
const l = try Listener.init(alloc, bind, key, owner.handler(), 10_000);
defer l.deinit();
owner.listener = l;
var actual: std.posix.sockaddr.storage = undefined;
var alen: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
try std.posix.getsockname(l.fd, @ptrCast(&actual), &alen);
const addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual)));
var cl = try TestPeer.init(addr, key);
defer cl.deinit();
var waited: u64 = 0;
cl.offer("goodbye");
while (waited < 10_000 and owner.received == 0) : (waited += 5) {
var fds = [_]std.posix.pollfd{
.{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
_ = std.posix.poll(&fds, 5) catch break;
if (fds[0].revents != 0) l.readable();
l.tick();
cl.drain();
for (l.conns) |slot| {
if (slot) |cn| l.drain(cn);
}
}
try std.testing.expect(owner.received > 0);
// The connection outlived the callback...
try std.testing.expect(owner.alive_after_callback);
// ...and did not outlive the pump.
try std.testing.expect(l.find(owner.id) == null);
// closeConn is the owner asking, so it gets no callback back.
try std.testing.expectEqual(@as(usize, 0), owner.closed);
}
test "initFromFd: a pipe is not a listener" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x5A} ** quic.key_len };
const fds = try std.posix.pipe();
defer {
std.posix.close(fds[0]);
std.posix.close(fds[1]);
}
try std.testing.expectError(
error.NotAUdpSocket,
Listener.initFromFd(alloc, fds[0], key, .{
.ctx = undefined,
.onOpen = ignoreOpen,
.onData = ignoreData,
.onClose = ignoreClose,
}, 5000),
);
}
test "initFromFd: an adopted bound fd serves a handshake" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x5A} ** quic.key_len };
// Bind a UDP socket by hand — the same steps init() does internally,
// minus the wolfSSL/TLS setup that initFromFd shares.
const fd = try std.posix.socket(
std.posix.AF.INET,
std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
0,
);
errdefer std.posix.close(fd);
const bind_addr = try std.net.Address.parseIp("127.0.0.1", 0);
try std.posix.bind(fd, &bind_addr.any, bind_addr.getOsSockLen());
var owner: EchoOwner = .{};
defer owner.deinit();
const l = try Listener.initFromFd(alloc, fd, key, owner.handler(), 5000);
defer l.deinit();
owner.listener = l;
// The manifest records the RUNTIME bound address; a kernel-assigned
// port (port 0 above) is the case that proves boundAddr reads the fd,
// not a flag.
var storage: std.posix.sockaddr.storage = undefined;
var slen: std.posix.socklen_t = @sizeOf(@TypeOf(storage));
try std.posix.getsockname(fd, @ptrCast(&storage), &slen);
const expected_port = std.net.Address.initPosix(@ptrCast(@alignCast(&storage))).getPort();
try std.testing.expectEqual(expected_port, l.boundAddr().getPort());
var cl = try TestPeer.init(l.boundAddr(), key);
defer cl.deinit();
try std.testing.expect(pump(l, &cl, 5000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened >= 1;
}
}.f, &owner));
try std.testing.expectEqual(@as(usize, 1), owner.opened);
// The key the manifest carries came from the module global, not a flag.
try std.testing.expectEqual(@as(?quic.Key, key), Listener.currentKey());
}
test "closeAll: a connected client hears the goodbye" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x5A} ** quic.key_len };
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, 5000);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
defer cl.deinit();
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened >= 1;
}
}.f, &owner));
// The goodbye: CONNECTION_CLOSE on every live conn, drained once so the
// packet leaves before the caller proceeds to exec.
setup.l.closeAll();
// The oracle is the client's observed close, not bytes the server queued
// — and specifically the draining period, which only a CONNECTION_CLOSE
// it actually read can enter. `dead` would also be set by any other read
// failure, and would pass for the wrong reason.
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(_: *EchoOwner, t: *TestPeer) bool {
return c.ngtcp2_conn_in_draining_period(t.cl.conn.?) != 0;
}
}.f, &owner));
try std.testing.expect(c.ngtcp2_conn_in_draining_period(cl.cl.conn.?) != 0);
}
// Plain, not recursive: this module reaches the QUIC stack's @cImport, and
// a recursive walk would force-analyze the entire wolfSSL/ngtcp2 namespace.
test {
std.testing.refAllDecls(@This());
}
test "Listener: a peer's deinit frees its slot at once, not at the idle timeout" {
const alloc = std.testing.allocator;
const key: quic.Key = .{ .bytes = [_]u8{0x3D} ** quic.key_len };
// Long enough that an idle-timer reap inside this test's budget is
// impossible: the close observed below has to be the peer's goodbye.
const idle_ms = 5000;
var owner: EchoOwner = .{};
defer owner.deinit();
const setup = try loopbackListener(alloc, key, &owner, idle_ms);
defer setup.l.deinit();
var cl = try TestPeer.init(setup.addr, key);
try std.testing.expect(pump(setup.l, &cl, 5000, struct {
fn f(o: *EchoOwner, t: *TestPeer) bool {
return t.cl.handshake_done and o.opened > 0;
}
}.f, &owner));
try std.testing.expectEqual(@as(usize, 0), owner.closed);
// The peer leaves the way every wall poll does: deinit, nothing else.
// Without CONNECTION_CLOSE the listener would hold this slot for
// `idle_ms`, so a poll a second would spend the daemon's whole client
// table in `max_clients` seconds and refuse every attach after that.
cl.deinit();
var waited: u64 = 0;
while (waited < 1000 and owner.closed == 0) : (waited += 10) {
var fds = [_]std.posix.pollfd{
.{ .fd = setup.l.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
_ = std.posix.poll(&fds, 10) catch break;
if (fds[0].revents != 0) setup.l.readable();
setup.l.tick();
for (setup.l.conns) |slot| {
if (slot) |cn| setup.l.drain(cn);
}
}
try std.testing.expectEqual(@as(usize, 1), owner.closed);
}