src/quic.zig
Ref: Size: 52.4 KiB History
//! 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. The
//! dialling end lives here because the listener's own tests need a real peer
//! and the folder rule forbids the server naming a client module.
//!
//! THE OWNERSHIP RULE: this file holds the one and only `@cImport` of the QUIC
//! stack. Two blocks over the same headers produce two DISTINCT Zig types, and
//! the egress ring hands `ngtcp2_vec`s across the seam between them.
const std = @import("std");
/// The C view of the QUIC stack. Exported because the client transport
/// shares it: two @cImport blocks over the same headers produce two
/// *distinct* Zig types. One import, one type universe.
pub const c = @cImport({
@cInclude("ngtcp2/ngtcp2.h");
@cInclude("ngtcp2/ngtcp2_crypto.h");
@cInclude("ngtcp2/ngtcp2_crypto_wolfssl.h");
@cInclude("wolfssl/options.h");
@cInclude("wolfssl/ssl.h");
});
// ---------------------------------------------------------------------------
// Wire constants: the numbers and strings both ends must spell alike
// ---------------------------------------------------------------------------
/// mux's conventional QUIC port. Both parsers reach for it when the user
/// names no port; it lives here because this module is the one thing both
/// binaries already import.
pub const default_port: u16 = 4433;
/// How long a QUIC connection tolerates silence before declaring the peer gone.
/// Keepalives run at a third of it, so an idle session never trips it.
/// `--quic-idle-ms` tunes it wherever a human dials or listens, because the
/// reconnect tests need death declared on a schedule they can wait for; `mux a`
/// has no such flag, since `--timeout` already bounds an agent's wait.
pub const default_idle_ms: u32 = 15_000;
/// Zero means NO idle timeout to ngtcp2, the inverse of what anyone typing
/// zero means, so `--quic-idle-ms`'s own type refuses it — once, rather
/// than in a post-check each binary copies.
pub const IdleMs = struct {
ms: u32 = default_idle_ms,
pub fn parseCLI(s: []const u8) error{Invalid}!IdleMs {
const ms = std.fmt.parseInt(u32, s, 10) catch return error.Invalid;
if (ms == 0) return error.Invalid;
return .{ .ms = ms };
}
};
pub const max_udp = 1452; // conservative IPv4 datagram that avoids fragmenting
pub const psk_identity: [*:0]const u8 = "mux";
pub const psk_ciphersuite: [*:0]const u8 = "TLS13-AES128-GCM-SHA256";
pub const alpn = "\x03mux";
// ---------------------------------------------------------------------------
// The dial address grammar
// ---------------------------------------------------------------------------
/// A name is resolved rather than refused: `--quic` names an address to
/// BIND, this names a box to reach, and a box is normally spelled with a
/// name.
pub fn parseAddr(alloc: std.mem.Allocator, host_port: []const u8) !std.net.Address {
const hp = try splitHostPort(host_port);
return resolveHost(alloc, hp.host, hp.port);
}
/// The grammar half, nothing resolved: `HOST[:PORT]` as a client types it,
/// where an omitted port means `default_port` and an empty one is a mistake.
/// Every spelling of `--quic` reads this — an agent, a human and the box they
/// point at must type the same thing, and two grammars are two dialects.
pub fn splitHostPort(s: []const u8) !struct { host: []const u8, port: u16 } {
if (s.len > 0 and s[0] == '[') {
// Bracketed: the brackets say where the address stops, so the port
// can default — and anything but a `:` after them is a typo, not a
// spelling to guess at.
const close = std.mem.indexOfScalar(u8, s, ']') orelse return error.MalformedAddress;
if (close + 1 == s.len) return .{ .host = s[1..close], .port = default_port };
if (s[close + 1] != ':') return error.MalformedAddress;
return .{ .host = s[1..close], .port = try parsePort(s[close + 2 ..]) };
}
const colon = std.mem.lastIndexOfScalar(u8, s, ':') orelse
return .{ .host = s, .port = default_port };
// An unbracketed IPv6 literal carries its own colons: `fe80::1:4433` reads
// equally well as a host with a port and as a host without one. Brackets
// spell the ambiguity out, so without them it is refused, not guessed.
if (std.mem.indexOfScalar(u8, s[0..colon], ':') != null) return error.MalformedAddress;
return .{ .host = s[0..colon], .port = try parsePort(s[colon + 1 ..]) };
}
fn parsePort(s: []const u8) !u16 {
return std.fmt.parseInt(u16, s, 10) catch error.MalformedAddress;
}
/// A host that is already known to be unambiguous, plus the port it goes
/// with: literal if it parses as one, resolved if it does not.
pub fn resolveHost(alloc: std.mem.Allocator, host: []const u8, port: u16) !std.net.Address {
if (host.len == 0) return error.MalformedAddress;
if (std.net.Address.parseIp(host, port)) |addr| return addr else |_| {}
// Not a literal: resolve it. A remote host is normally a name.
const list = try std.net.getAddressList(alloc, host, port);
defer list.deinit();
if (list.addrs.len == 0) return error.UnknownHostName;
return list.addrs[0];
}
test "splitHostPort: literal addresses, bracketed and not" {
const v4 = try splitHostPort("127.0.0.1:4433");
try std.testing.expectEqualStrings("127.0.0.1", v4.host);
try std.testing.expectEqual(@as(u16, 4433), v4.port);
const v6 = try splitHostPort("[::1]:4433");
try std.testing.expectEqualStrings("::1", v6.host);
try std.testing.expectEqual(@as(u16, 4433), v6.port);
const any6 = try splitHostPort("[::]:1");
try std.testing.expectEqualStrings("::", any6.host);
try std.testing.expectEqual(@as(u16, 1), any6.port);
// No port names the default. Spelled out rather than written
// `quic.default_port`: comparing the parse's answer against the constant the
// parse reads holds for ANY value, and 4433 is what both ends must agree on.
const dflt = try splitHostPort("127.0.0.1");
try std.testing.expectEqualStrings("127.0.0.1", dflt.host);
try std.testing.expectEqual(@as(u16, 4433), dflt.port);
const dflt6 = try splitHostPort("[::1]");
try std.testing.expectEqualStrings("::1", dflt6.host);
try std.testing.expectEqual(@as(u16, 4433), dflt6.port);
try std.testing.expectError(error.MalformedAddress, splitHostPort("127.0.0.1:"));
try std.testing.expectError(error.MalformedAddress, splitHostPort("127.0.0.1:99999"));
try std.testing.expectError(error.MalformedAddress, splitHostPort("[::1]4433"));
// An IPv6 literal without brackets is ambiguous about where the address
// stops, so it is refused instead of being read either way.
try std.testing.expectError(error.MalformedAddress, splitHostPort("::1:4433"));
try std.testing.expectError(error.MalformedAddress, splitHostPort("fe80::1:4433"));
}
test "parseAddr: literals, brackets, and the spellings that are refused" {
const alloc = std.testing.allocator;
// Literals only here: a name would send this test to a resolver, and
// what it answered would depend on the machine running it.
try std.testing.expectEqual(
@as(u16, 4433),
(try parseAddr(alloc, "127.0.0.1:4433")).getPort(),
);
try std.testing.expectEqual(@as(u16, 9), (try parseAddr(alloc, "127.0.0.1:9")).getPort());
// An omitted port means mux's own. Spelled out rather than written
// `default_port`, because comparing against the constant the parse reads
// would hold for any value — and this is the number the daemon must agree on.
try std.testing.expectEqual(@as(u16, 4433), (try parseAddr(alloc, "10.0.0.2")).getPort());
const six = try parseAddr(alloc, "[::1]:9999");
try std.testing.expectEqual(@as(u16, 9999), six.getPort());
try std.testing.expect(six.any.family == std.posix.AF.INET6);
// Bracketed and portless: the brackets say where the address stops, so
// the port can default.
try std.testing.expectEqual(@as(u16, 4433), (try parseAddr(alloc, "[::1]")).getPort());
// An unbracketed IPv6 literal would have its last group read as a
// port. Refused rather than guessed at — the same refusal the daemon's
// `parseBindAddr` inherits from the shared split.
try std.testing.expectError(error.MalformedAddress, parseAddr(alloc, "fe80::1:4433"));
try std.testing.expectError(error.MalformedAddress, parseAddr(alloc, "127.0.0.1:"));
try std.testing.expectError(error.MalformedAddress, parseAddr(alloc, "127.0.0.1:99999"));
try std.testing.expectError(error.MalformedAddress, parseAddr(alloc, ""));
try std.testing.expectError(error.MalformedAddress, parseAddr(alloc, ":4433"));
}
// ---------------------------------------------------------------------------
// The pre-shared key
// ---------------------------------------------------------------------------
pub const key_len = 32;
/// The pre-shared key, and the rules for getting one off disk. A key file is as
/// sensitive as an ssh private key and held to the same standard: readable by
/// nobody but its owner. A daemon that starts anyway has authenticated nothing.
pub const Key = struct {
bytes: [key_len]u8,
pub const LoadError = error{
KeyFileMissing,
KeyFilePermissive,
KeyFileMalformed,
};
/// Accepts either 32 raw bytes or 64 hex characters (trailing
/// whitespace ignored, so `xxd -p` and a text editor both work).
pub fn load(path: []const u8) !Key {
const file = std.fs.cwd().openFile(path, .{}) catch |err| switch (err) {
error.FileNotFound => return error.KeyFileMissing,
else => return err,
};
defer file.close();
const st = try file.stat();
// Group or other bits set = refuse, exactly as ssh does for a
// private key. Checked before the contents are read, so a bad-mode
// key is never even loaded into memory.
if (st.mode & 0o077 != 0) return error.KeyFilePermissive;
var buf: [128]u8 = undefined;
const n = try file.readAll(&buf);
// A file of exactly 32 bytes is a raw key, taken WITHOUT trimming: a
// random key ends in one of " \t\r\n" about one time in 64, and trimming
// shortens those to 31 bytes and rejects them as malformed.
if (n == key_len) {
var k: Key = undefined;
@memcpy(&k.bytes, buf[0..key_len]);
return k;
}
// Anything else may carry a text editor's trailing newline: trim
// and retry, which is the case the trimming actually exists for.
const raw = std.mem.trimRight(u8, buf[0..n], " \t\r\n");
if (raw.len == key_len) {
var k: Key = undefined;
@memcpy(&k.bytes, raw);
return k;
}
if (raw.len == key_len * 2) {
var k: Key = undefined;
_ = std.fmt.hexToBytes(&k.bytes, raw) catch return error.KeyFileMalformed;
return k;
}
return error.KeyFileMalformed;
}
};
/// A buffer for one refusal body: PATH_MAX for the path, 128 for the rest. The
/// 128 is chosen, not derived, because the catch-all appends an `@errorName`
/// whose only bound is the longest error name in the binary. `keyRefusalBody`'s
/// clipping is what makes choosing rather than deriving safe.
pub const key_refusal_len = std.fs.max_path_bytes + 128;
/// The middle sentence of every key refusal, in every binary — one owner so the
/// catch-alls cannot drift. `err` is `anyerror` rather than `Key.LoadError`
/// because `load` widens past its own set. Truncating rather than failing:
/// this line is the user's only account of the refusal.
pub fn keyRefusalBody(buf: []u8, err: anyerror, path: []const u8) []const u8 {
var w: std.Io.Writer = .fixed(buf);
switch (err) {
error.KeyFileMissing => w.print("no such key file: {s}", .{path}) catch {},
error.KeyFilePermissive => w.print(
"{s} is readable by group or other; chmod 600 it",
.{path},
) catch {},
error.KeyFileMalformed => w.print(
"{s} is not a key: want 32 raw bytes or 64 hex characters",
.{path},
) catch {},
else => w.print("cannot read {s}: {s}", .{ path, @errorName(err) }) catch {},
}
return w.buffered();
}
/// Test helper: chmod a file inside a Dir. `Dir.chmod` applies to the
/// directory itself, not to an entry in it.
fn chmodAt(dir: std.fs.Dir, sub: []const u8, mode: std.posix.mode_t) !void {
const f = try dir.openFile(sub, .{});
defer f.close();
try f.chmod(mode);
}
test "Key.load: accepts 32 raw bytes and 64 hex chars, owner-only" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const raw = [_]u8{0xAB} ** key_len;
try tmp.dir.writeFile(.{ .sub_path = "raw.key", .data = &raw });
try chmodAt(tmp.dir, "raw.key", 0o600);
var hex: [key_len * 2]u8 = undefined;
_ = try std.fmt.bufPrint(&hex, "{x}", .{&raw});
// A key pasted by a human ends in a newline; that must not change it.
try tmp.dir.writeFile(.{ .sub_path = "hex.key", .data = hex ++ "\n" });
try chmodAt(tmp.dir, "hex.key", 0o600);
var path_buf: [256]u8 = undefined;
const dir = try tmp.dir.realpath(".", &path_buf);
var jb: [512]u8 = undefined;
const from_raw = try Key.load(try std.fmt.bufPrint(&jb, "{s}/raw.key", .{dir}));
try std.testing.expectEqualSlices(u8, &raw, &from_raw.bytes);
const from_hex = try Key.load(try std.fmt.bufPrint(&jb, "{s}/hex.key", .{dir}));
try std.testing.expectEqualSlices(u8, &raw, &from_hex.bytes);
}
test "Key.load: a raw key whose last byte is whitespace is still a key" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var path_buf: [256]u8 = undefined;
const dir = try tmp.dir.realpath(".", &path_buf);
var jb: [512]u8 = undefined;
// Found by a review probe, not by me. Kept because the probe was
// transient: a bug with no test is a bug with a return ticket.
inline for (.{ '\n', '\r', '\t', ' ' }) |last| {
var raw = [_]u8{0x7F} ** key_len;
raw[key_len - 1] = last;
try tmp.dir.writeFile(.{ .sub_path = "ws.key", .data = &raw });
try chmodAt(tmp.dir, "ws.key", 0o600);
const k = try Key.load(try std.fmt.bufPrint(&jb, "{s}/ws.key", .{dir}));
try std.testing.expectEqualSlices(u8, &raw, &k.bytes);
}
// ...while 32 key bytes plus an editor's newline (33 on disk) still
// loads, which is what the trimming is for.
const clean = [_]u8{0x42} ** key_len;
try tmp.dir.writeFile(.{ .sub_path = "nl.key", .data = clean ++ "\n" });
try chmodAt(tmp.dir, "nl.key", 0o600);
const k2 = try Key.load(try std.fmt.bufPrint(&jb, "{s}/nl.key", .{dir}));
try std.testing.expectEqualSlices(u8, &clean, &k2.bytes);
}
test "Key.load: refuses a permissive mode, a missing file, and a bad length" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var path_buf: [256]u8 = undefined;
const dir = try tmp.dir.realpath(".", &path_buf);
var jb: [512]u8 = undefined;
const raw = [_]u8{0xCD} ** key_len;
// Group-readable: refused, like an ssh private key.
try tmp.dir.writeFile(.{ .sub_path = "group.key", .data = &raw });
try chmodAt(tmp.dir, "group.key", 0o640);
try std.testing.expectError(
error.KeyFilePermissive,
Key.load(try std.fmt.bufPrint(&jb, "{s}/group.key", .{dir})),
);
// World-readable: same.
try tmp.dir.writeFile(.{ .sub_path = "world.key", .data = &raw });
try chmodAt(tmp.dir, "world.key", 0o604);
try std.testing.expectError(
error.KeyFilePermissive,
Key.load(try std.fmt.bufPrint(&jb, "{s}/world.key", .{dir})),
);
try std.testing.expectError(
error.KeyFileMissing,
Key.load(try std.fmt.bufPrint(&jb, "{s}/nope.key", .{dir})),
);
// Right mode, wrong content: neither 32 raw nor 64 hex.
try tmp.dir.writeFile(.{ .sub_path = "short.key", .data = "too short" });
try chmodAt(tmp.dir, "short.key", 0o600);
try std.testing.expectError(
error.KeyFileMalformed,
Key.load(try std.fmt.bufPrint(&jb, "{s}/short.key", .{dir})),
);
// 64 characters, but not hex.
try tmp.dir.writeFile(.{ .sub_path = "nothex.key", .data = "z" ** 64 });
try chmodAt(tmp.dir, "nothex.key", 0o600);
try std.testing.expectError(
error.KeyFileMalformed,
Key.load(try std.fmt.bufPrint(&jb, "{s}/nothex.key", .{dir})),
);
}
test "keyRefusalBody: the words three binaries print, byte for byte" {
// These bytes ARE the contract: every key refusal any binary prints is a
// prefix, this body, and at most a suffix, so a change here changes every
// caller at once.
var buf: [key_refusal_len]u8 = undefined;
try std.testing.expectEqualStrings(
"no such key file: /etc/mux/key",
keyRefusalBody(&buf, error.KeyFileMissing, "/etc/mux/key"),
);
try std.testing.expectEqualStrings(
"/etc/mux/key is readable by group or other; chmod 600 it",
keyRefusalBody(&buf, error.KeyFilePermissive, "/etc/mux/key"),
);
try std.testing.expectEqualStrings(
"/etc/mux/key is not a key: want 32 raw bytes or 64 hex characters",
keyRefusalBody(&buf, error.KeyFileMalformed, "/etc/mux/key"),
);
// Anything `load` failed with that is not one of the three: the body
// names the error rather than inventing a cause for it, and the path
// still comes first so the line reads the same way as the other three.
try std.testing.expectEqualStrings(
"cannot read /etc/mux/key: AccessDenied",
keyRefusalBody(&buf, error.AccessDenied, "/etc/mux/key"),
);
}
// ---------------------------------------------------------------------------
// Egress: the ring, and what one writev_stream return means for it
// ---------------------------------------------------------------------------
/// How many outbound bytes one connection may hold. Sized to the stream
/// window the peer advertises, because holding much more than the peer will
/// let us send buys nothing: past this the daemon's own `pending_cap` is the
/// right place for the backlog to sit and be judged.
pub const egress_cap = 256 * 1024;
/// Outbound stream bytes, in a ring that never moves a byte once written.
///
/// A ring rather than a growable buffer, because ngtcp2 does NOT copy stream
/// payload: `ngtcp2_conn_writev_stream` stores the VECTOR it is handed and
/// re-reads those bytes if the packet is lost. A buffer that reallocates on
/// append, or clears once the last byte is handed over, leaves ngtcp2 holding a
/// recycled pointer and it dies inside `ngtcp2_pkt_encode_stream_frame`.
///
/// The invariant everything here exists to hold: a byte handed to ngtcp2 does
/// not move or get overwritten until the peer acknowledges it. `head` advances
/// only from `acked_stream_data_offset`. Fixed in size is the other half — a
/// full ring is the backpressure signal.
pub const Egress = struct {
buf: []u8,
/// The oldest byte the peer has not acknowledged.
head: usize = 0,
/// Bytes from `head` still owed: handed to ngtcp2 and unacknowledged,
/// plus not yet handed over.
held: usize = 0,
/// The tail of `held` that ngtcp2 has not taken yet.
unsent: usize = 0,
pub fn deinit(self: *Egress, alloc: std.mem.Allocator) void {
alloc.free(self.buf);
self.* = .{ .buf = &.{} };
}
pub fn freeSpace(self: *const Egress) usize {
return self.buf.len - self.held;
}
/// Take what fits and report how much that was. A short return is not an
/// error, it is the whole mechanism: the caller keeps the remainder and
/// is thereby the one holding — and bounding — the backlog.
pub fn push(self: *Egress, bytes: []const u8) usize {
const n = @min(bytes.len, self.freeSpace());
if (n == 0) return 0;
const start = (self.head + self.held) % self.buf.len;
const first = @min(n, self.buf.len - start);
@memcpy(self.buf[start..][0..first], bytes[0..first]);
if (first < n) @memcpy(self.buf[0 .. n - first], bytes[first..n]);
self.held += n;
self.unsent += n;
return n;
}
/// The unsent region as up to two vectors — two when it wraps, which is
/// the price of never moving a byte, and ngtcp2 takes a vector array
/// precisely so that price is payable.
pub fn vecs(self: *const Egress, out: *[2]c.ngtcp2_vec) usize {
if (self.unsent == 0) return 0;
const start = (self.head + (self.held - self.unsent)) % self.buf.len;
const first = @min(self.unsent, self.buf.len - start);
out[0] = .{ .base = self.buf.ptr + start, .len = first };
if (first == self.unsent) return 1;
out[1] = .{ .base = self.buf.ptr, .len = self.unsent - first };
return 2;
}
/// The bytes stay exactly where they are — ngtcp2 now holds pointers to them.
pub fn took(self: *Egress, n: usize) void {
self.unsent -= @min(n, self.unsent);
}
/// The peer acknowledged `n` more bytes. ngtcp2 documents this callback
/// as arriving "sequentially in increasing order of offset without any
/// overlap", so a running count IS the acknowledged prefix, and this is
/// the only thing that ever frees space.
pub fn ack(self: *Egress, n: usize) void {
// A deinit'd ring has a zero-length buffer, and the modulo below
// would divide by zero. Unreachable while owners delete the conn
// before freeing the ring; the guard keeps a teardown reordering
// from turning into one.
if (self.buf.len == 0) return;
const taken = @min(n, self.held - self.unsent);
self.head = (self.head + taken) % self.buf.len;
self.held -= taken;
}
};
/// One `writev_stream` return, in ORDER: ngtcp2 can commit `ndatalen` and
/// still return an error afterwards, so account for the bytes first or they
/// are re-offered at an offset the peer has moved past.
pub const WriteAction = enum {
/// The call failed. Stop draining; the bytes are already accounted.
stop,
/// Nothing more to send right now.
brk,
/// A packet was produced; send it and go round again.
cont,
};
pub fn accountWrite(out: *Egress, wrote: c.ngtcp2_ssize, n: c.ngtcp2_ssize) WriteAction {
if (wrote > 0) out.took(@intCast(wrote));
if (n < 0) return .stop;
if (n == 0) return .brk;
return .cont;
}
// ---------------------------------------------------------------------------
// The per-connection core. `Client` and `Listener.Conn` are different objects —
// one dials and dies of a refusal, the other accepts and cannot be told — but
// the ngtcp2 they drive is the same library run the same way. What follows is
// the parts where a difference would be a BUG, not a design.
// ---------------------------------------------------------------------------
/// Milliseconds until ngtcp2 next wants servicing on `conn`, capped. A
/// connection with nothing due, or none at all, wants the cap. Feeds a
/// poll timeout at both ends: no timerfd anywhere, deliberately.
pub fn expiryMs(conn: ?*c.ngtcp2_conn, cap_ms: i32, now: u64) i32 {
const cn = conn orelse return cap_ms;
const expiry = c.ngtcp2_conn_get_expiry(cn);
if (expiry == std.math.maxInt(u64)) return cap_ms;
if (expiry <= now) return 0;
return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000));
}
/// One packet in, re-entrancy counter held across it. Nonzero is dead.
pub fn readPkt(
conn: *c.ngtcp2_conn,
path: *c.ngtcp2_path,
pkt: []const u8,
depth: *u8,
) c.ngtcp2_ssize {
var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
depth.* += 1;
defer depth.* -= 1;
return c.ngtcp2_conn_read_pkt(conn, path, &pi, pkt.ptr, pkt.len, timestampNs());
}
/// How a drain pass ended. `failed` is ngtcp2 refusing the connection: the
/// client marks itself dead, the listener discards it and leaves the verdict
/// to `tick`'s expiry, as it always has. A socket that would not take a
/// datagram ends the pass as `done`, because the next pass offers it again.
pub const DrainEnd = enum { done, failed };
/// The egress loop: from the ring, into ngtcp2, onto the wire, repeat.
/// `sendPkt` is the seam where the two sockets differ.
pub fn drainConn(
conn: *c.ngtcp2_conn,
out: *Egress,
stream_id: i64,
ctx: anytype,
comptime sendPkt: fn (@TypeOf(ctx), []const u8) bool,
) DrainEnd {
var buf: [max_udp]u8 = undefined;
// Set when the peer's stream window is full. Everything that is not
// stream data still has to leave.
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 stream_id != -1) {
vcnt = out.vecs(&vecs);
if (vcnt > 0) sid = stream_id;
}
const n = c.ngtcp2_conn_writev_stream_versioned(
conn,
&ps.path,
c.NGTCP2_PKT_INFO_VERSION,
&pi,
&buf,
buf.len,
&wrote,
0,
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) {
// A documented return, not a failure: the peer has no window left.
// Treating it as fatal abandons the ACKs and flow-control updates
// that reopen that window, turning backpressure into a stall.
stream_blocked = true;
continue;
}
switch (accountWrite(out, wrote, n)) {
.stop => return .failed,
.brk => return .done,
.cont => {},
}
if (!sendPkt(ctx, buf[0..@intCast(n)])) return .done;
}
}
test "Egress: a byte does not move until it is acked, and the ring wraps" {
const alloc = std.testing.allocator;
var e: Egress = .{ .buf = try alloc.alloc(u8, 8) };
defer e.deinit(alloc);
try std.testing.expectEqual(@as(usize, 5), e.push("hello"));
var v: [2]c.ngtcp2_vec = undefined;
try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
try std.testing.expectEqual(@as(usize, 5), v[0].len);
const base = e.buf.ptr;
try std.testing.expectEqual(base, v[0].base);
// ngtcp2 takes three. Those three now have a pointer pointing at them
// and must not move; the two behind them are still ours to offer.
e.took(3);
try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
try std.testing.expectEqual(@as(usize, 2), v[0].len);
try std.testing.expectEqual(base + 3, v[0].base);
// Room is what is left after everything HELD, sent or not — the three
// in ngtcp2's hands are not free space just because they left the box.
try std.testing.expectEqual(@as(usize, 3), e.freeSpace());
try std.testing.expectEqual(@as(usize, 3), e.push("world"));
try std.testing.expectEqual(@as(usize, 0), e.push("x"));
// An acknowledgement is the only thing that frees anything.
e.ack(3);
try std.testing.expectEqual(@as(usize, 3), e.freeSpace());
// ...and the write that follows wraps rather than shifting a byte.
try std.testing.expectEqual(@as(usize, 3), e.push("abc"));
try std.testing.expectEqual(@as(usize, 2), e.vecs(&v));
try std.testing.expectEqualStrings("lowor", v[0].base[0..v[0].len]);
try std.testing.expectEqualStrings("abc", v[1].base[0..v[1].len]);
}
test "Egress: an ack can never free more than is outstanding" {
const alloc = std.testing.allocator;
var e: Egress = .{ .buf = try alloc.alloc(u8, 8) };
defer e.deinit(alloc);
_ = e.push("abcd");
e.took(2);
// Two are in flight and two are still unsent. A callback claiming more
// than is outstanding must not consume the unsent ones — they have
// never been on the wire and cannot have been acknowledged.
e.ack(99);
try std.testing.expectEqual(@as(usize, 2), e.held);
try std.testing.expectEqual(@as(usize, 2), e.unsent);
var v: [2]c.ngtcp2_vec = undefined;
try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
try std.testing.expectEqualStrings("cd", v[0].base[0..v[0].len]);
}
test "accountWrite: bytes ngtcp2 committed are accounted even when the call failed" {
const alloc = std.testing.allocator;
var e: Egress = .{ .buf = try alloc.alloc(u8, 16) };
defer e.deinit(alloc);
_ = e.push("abcdefgh");
try std.testing.expectEqual(@as(usize, 8), e.unsent);
// ngtcp2 committed four bytes of stream data — its offset has moved — and
// THEN returned an error. Accounting after the check leaves those four
// counted as unsent, and the next drain offers them at an offset the peer
// is already past.
try std.testing.expectEqual(WriteAction.stop, accountWrite(&e, 4, -1));
try std.testing.expectEqual(@as(usize, 4), e.unsent);
// The ordinary returns, for completeness of the contract.
try std.testing.expectEqual(WriteAction.brk, accountWrite(&e, 0, 0));
try std.testing.expectEqual(WriteAction.cont, accountWrite(&e, 4, 120));
try std.testing.expectEqual(@as(usize, 0), e.unsent);
}
test "Egress: an ack against a torn-down ring is ignored, not a division by zero" {
const alloc = std.testing.allocator;
var e: Egress = .{ .buf = try alloc.alloc(u8, 8) };
_ = e.push("abcd");
e.took(4);
// Teardown leaves a zero-length buffer behind. No live path acks it today,
// but the modulo in `ack` divides by `buf.len`, so a teardown reordering
// would be a division by zero on a path nobody would look at.
e.deinit(alloc);
try std.testing.expectEqual(@as(usize, 0), e.buf.len);
e.ack(4);
try std.testing.expectEqual(@as(usize, 0), e.held);
}
// ---------------------------------------------------------------------------
// Time, and the keepalive derived from it
// ---------------------------------------------------------------------------
pub fn timestampNs() u64 {
const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch return 0;
return @as(u64, @intCast(ts.sec)) * 1_000_000_000 + @as(u64, @intCast(ts.nsec));
}
/// Never zero: ngtcp2 reads a zero timeout as "disabled", like UINT64_MAX.
pub fn keepAliveNs(idle_ms: u64) u64 {
return @max(1, idle_ms / 3) * 1_000_000;
}
test "keepAlive: a third of the idle timeout, and never disabled" {
try std.testing.expectEqual(@as(u64, 5_000_000_000), keepAliveNs(15_000));
try std.testing.expectEqual(@as(u64, 500_000_000), keepAliveNs(1500));
// Zero would mean "no keepalive" to ngtcp2 — the opposite of what a
// small idle timeout is asking for.
try std.testing.expectEqual(@as(u64, 1_000_000), keepAliveNs(1));
try std.testing.expectEqual(@as(u64, 1_000_000), keepAliveNs(2));
}
// ---------------------------------------------------------------------------
// ngtcp2 callbacks with no per-endpoint variation
// ---------------------------------------------------------------------------
pub fn randCb(dest: [*c]u8, destlen: usize, _: [*c]const c.ngtcp2_rand_ctx) callconv(.c) void {
std.crypto.random.bytes(dest[0..destlen]);
}
pub fn getNewCidCb(
_: ?*c.ngtcp2_conn,
cid: [*c]c.ngtcp2_cid,
token: [*c]c.ngtcp2_stateless_reset_token,
cidlen: usize,
_: ?*anyopaque,
) callconv(.c) c_int {
std.crypto.random.bytes(cid.*.data[0..cidlen]);
cid.*.datalen = cidlen;
std.crypto.random.bytes(&token.*.data);
return 0;
}
// ---------------------------------------------------------------------------
// The client side of the handshake, kept separate from `Client` so the
// listener's tests can dial with it rather than a second copy: a drift between
// two copies fails the handshake with nothing to read but a TLS alert.
// ---------------------------------------------------------------------------
/// Identity "mux", the key's bytes, our one ciphersuite: none is a choice.
pub fn answerPsk(
key: Key,
identity: [*c]u8,
id_max: c_uint,
key_out: [*c]u8,
key_max: c_uint,
ciphersuite: [*c][*c]const u8,
) c_uint {
if (id_max < 4 or key_max < key_len) return 0;
@memcpy(identity[0..4], "mux\x00");
@memcpy(key_out[0..key_len], &key.bytes);
if (ciphersuite) |cs| cs.* = psk_ciphersuite;
return key_len;
}
/// PSK over TLS 1.3, our ciphersuite, our ALPN.
pub fn clientTls(
psk_cb: c.wc_psk_client_tls13_callback,
conn_ref: *c.ngtcp2_crypto_conn_ref,
) !struct { ctx: *c.WOLFSSL_CTX, ssl: *c.WOLFSSL } {
if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit;
const ctx = c.wolfSSL_CTX_new(c.wolfTLSv1_3_client_method()) orelse return error.TlsInit;
// The caller records `ctx` only on success, so every failure past here
// owns the free. Both callers used to set their field first and lean on
// deinit for it; a shared owner cannot, having no field to set.
errdefer c.wolfSSL_CTX_free(ctx);
if (c.ngtcp2_crypto_wolfssl_configure_client_context(ctx) != 0) return error.TlsInit;
c.wolfSSL_CTX_set_psk_client_tls13_callback(ctx, psk_cb);
_ = c.wolfSSL_CTX_set_cipher_list(ctx, psk_ciphersuite);
const ssl = c.wolfSSL_new(ctx) orelse return error.TlsInit;
_ = c.wolfSSL_set_app_data(ssl, conn_ref);
_ = c.wolfSSL_UseALPN(
ssl,
@constCast(alpn[1..].ptr),
alpn.len - 1,
c.WOLFSSL_ALPN_FAILED_ON_MISMATCH,
);
return .{ .ctx = ctx, .ssl = ssl };
}
/// The ngtcp2 callbacks that are the library's own, not the caller's.
pub fn clientCallbacks() c.ngtcp2_callbacks {
var cbs: c.ngtcp2_callbacks = std.mem.zeroes(c.ngtcp2_callbacks);
cbs.client_initial = c.ngtcp2_crypto_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;
// The listener answers every fresh Initial with a Retry, so a client
// that cannot process one never gets past its first flight.
cbs.recv_retry = c.ngtcp2_crypto_recv_retry_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 = randCb;
cbs.get_new_connection_id2 = getNewCidCb;
return cbs;
}
/// The flow-control window a mux client opens with. `max_idle_timeout` stays
/// at ngtcp2's default: only the reconnecting transport tunes it, and a
/// fixture with no reconnect loop must not inherit a number chosen for one.
pub fn clientParams() c.ngtcp2_transport_params {
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;
return params;
}
// ---------------------------------------------------------------------------
// Paths
// ---------------------------------------------------------------------------
/// The ngtcp2_path literal, spelled once. Six call sites built this
/// by hand and the seventh would have drifted.
pub fn pathFrom(
local: *std.posix.sockaddr.storage,
local_len: std.posix.socklen_t,
remote: *std.posix.sockaddr.storage,
remote_len: std.posix.socklen_t,
) c.ngtcp2_path {
return .{
.local = .{ .addr = @ptrCast(local), .addrlen = local_len },
.remote = .{ .addr = @ptrCast(remote), .addrlen = remote_len },
.user_data = null,
};
}
// ---------------------------------------------------------------------------
// The client transport: one UDP socket, one connection, one bidirectional
// stream of opaque bytes. It knows NOTHING about the frame protocol it carries.
//
// Three ngtcp2 constraints shape it, and the listener obeys the same three:
// 1. ngtcp2 does not copy stream payload — outbound bytes must not move or
// be reused until the peer acknowledges them.
// 2. A blocked stream is not a dead one: abandoning the egress loop on
// STREAM_DATA_BLOCKED also abandons the ACKs that reopen the window.
// 3. BOTH flow-control windows get extended on consume, or the connection
// window closes and the daemon stalls a few hundred kilobytes in.
// ---------------------------------------------------------------------------
/// wolfSSL's PSK callback carries no user pointer, so the key must be reachable
/// without one. Process-global, and a mux client is NOT one dial at a time:
/// every poller and tile pump dials on its own thread, so two concurrent dials
/// with different keys race between `connect` writing this and the ClientHello
/// its `drain()` builds. The symptom is a handshake that never completes.
/// Fix tracked as issue 92d0a0e2: a per-connection key via `wolfSSL_set_app_data`.
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.?));
// A forwarding wire has a deliberately small opaque-byte budget. Check
// before append OR extending flow-control: credit must never be granted
// for bytes its owner cannot retain. Terminal clients leave this null.
if (self.inbound_cap) |cap| if (datalen > cap -| self.in.items.len) {
self.dead = true;
return -1;
};
// 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;
}
test "forwarding receive cap accepts one legal frame exactly and rejects the next fragment" {
const alloc = std.testing.allocator;
// quic.zig is tested as its own root, so it cannot import the client
// protocol module here. A forwarding data frame is its five-byte wire
// header, u32 channel id, and the protocol's 16 KiB maximum data body.
// The dial installs its larger transport queue cap before waitReady.
const cap = 5 + 4 + 16 * 1024;
var cl: Client = .{
.alloc = alloc,
.fd = -1,
.out = .{ .buf = &.{} },
.inbound_cap = cap,
.extend_windows = false,
};
defer cl.in.deinit(alloc);
var header: [5]u8 = undefined;
var data: [4 + 16 * 1024]u8 = undefined;
const first = recvStreamDataCb(null, 0, 0, 0, &header, header.len, &cl, null);
const second = recvStreamDataCb(null, 0, 0, header.len, &data, data.len, &cl, null);
try std.testing.expectEqual(@as(c_int, 0), first);
try std.testing.expectEqual(@as(c_int, 0), second);
try std.testing.expectEqual(cap, cl.in.items.len);
try std.testing.expect(!cl.dead);
const extra = [_]u8{0};
try std.testing.expectEqual(@as(c_int, -1), recvStreamDataCb(null, 0, 0, cap, &extra, extra.len, &cl, null));
try std.testing.expect(cl.dead);
try std.testing.expectEqual(cap, cl.in.items.len);
}
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 — its callbacks only set flags and append to `in` —
/// so `send` may still drain, which keeps a keystroke off the next poll
/// cycle. The assert in `drain` is what makes that checked.
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,
/// Optional opaque receive budget selected by a role-specific owner.
/// Null preserves the terminal transport's existing unbounded framing
/// limit; no forwarding frame policy belongs in this transport.
inbound_cap: ?usize = null,
/// 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,
0,
);
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,
¶ms,
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.sayGoodbye();
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);
}
/// CONNECTION_CLOSE on the way out, so the daemon frees this
/// connection's client slot NOW rather than when its idle timer expires
/// (15 s by default). The wall polls every QUIC host once a second on a
/// connection of its own; a teardown that just dropped the socket left
/// those polls holding one client slot each, so the table filled at a
/// slot a second and every real attach after that was refused — found
/// on a laptop whose only clients were another wall's polls, when the
/// daemon had eight slots and so took eight seconds (2026-09-02). The
/// table is `max_clients` deep now, which buys time and fixes nothing:
/// without the goodbye a long-lived wall still fills it. Best
/// effort: a peer that never handshook or already closed gets nothing.
fn sayGoodbye(self: *Client) void {
const conn = self.conn orelse return;
if (self.dead) return;
var ccerr: c.ngtcp2_ccerr = undefined;
c.ngtcp2_ccerr_default(&ccerr);
var buf: [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,
timestampNs(),
);
if (n > 0) _ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch {};
}
/// 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; expiryMs's @intCast would panic
// on it instead of honouring it.
if (cap_ms < 0) return cap_ms;
return expiryMs(self.conn, cap_ms, timestampNs());
}
/// 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 sends the listener a packet, which carries the drain they 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);
if (readPkt(conn, &path, buf[0..n], &self.ngtcp2_depth) != 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 write half of the invariant `readPkt` counts. A future caller
// that drains from inside an ngtcp2 callback fails here
// deterministically in Debug rather than corrupting loss detection.
std.debug.assert(self.ngtcp2_depth == 0);
const conn = self.conn orelse return;
const sendPkt = struct {
fn f(cl: *Client, pkt: []const u8) bool {
_ = std.posix.send(cl.fd, pkt, 0) catch |err| {
cl.sendRecvFailed(err);
return false;
};
return true;
}
}.f;
if (drainConn(conn, &self.out, self.stream_id, self, sendPkt) == .failed) self.dead = true;
}
/// 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 {
std.testing.refAllDecls(@This());
}