a73x

ae0b1138

feat: muxd speaks QUIC — one UDP fd, PSK, and the protocol never noticed

a73x   2026-08-08 14:08

Commit message
feat: muxd speaks QUIC — one UDP fd, PSK, and the protocol never noticed
build.zig
Old New
@@ -99,6 +99,7 @@ pub fn build(b: *std.Build) void {
99 server_mod.addImport("engine", engine_mod); 99 server_mod.addImport("engine", engine_mod);
100 server_mod.addImport("pty", pty_mod); 100 server_mod.addImport("pty", pty_mod);
101 server_mod.addImport("protocol", protocol_mod); 101 server_mod.addImport("protocol", protocol_mod);
102 server_mod.addImport("quic", quic_mod);
102 103
103 const client_mod = b.createModule(.{ 104 const client_mod = b.createModule(.{
104 .root_source_file = b.path("src/client.zig"), 105 .root_source_file = b.path("src/client.zig"),
@@ -135,6 +136,9 @@ pub fn build(b: *std.Build) void {
135 exe_mod.addImport("server", server_mod); 136 exe_mod.addImport("server", server_mod);
136 exe_mod.addImport("protocol", protocol_mod); 137 exe_mod.addImport("protocol", protocol_mod);
137 exe_mod.addImport("proxy", proxy_mod); 138 exe_mod.addImport("proxy", proxy_mod);
139 // The daemon entrypoint loads the key and constructs the listener, so it
140 // needs the module directly rather than through the server.
141 exe_mod.addImport("quic", quic_mod);
138 142
139 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod }); 143 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod });
140 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe 144 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe
@@ -152,9 +156,12 @@ pub fn build(b: *std.Build) void {
152 b.installArtifact(mux_exe); 156 b.installArtifact(mux_exe);
153 157
154 const test_step = b.step("test", "Run unit tests"); 158 const test_step = b.step("test", "Run unit tests");
155 // mux_mod is an executable root, but it carries the argument parser, and 159 // mux_mod and exe_mod are executable roots, but they carry the argument
156 // a test that is never built is not a test. 160 // parsers, and a test that is never built is not a test. exe_mod's
157 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod }) |mod| { 161 // absence here was a live hazard recorded in decisions.md — muxd's
162 // entrypoint could grow tests that silently never ran, exactly as
163 // mux_main.zig's five did before it was added.
164 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, exe_mod }) |mod| {
158 const t = b.addTest(.{ .root_module = mod }); 165 const t = b.addTest(.{ .root_module = mod });
159 t.use_llvm = true; 166 t.use_llvm = true;
160 t.use_lld = true; 167 t.use_lld = true;
@@ -164,7 +171,7 @@ pub fn build(b: *std.Build) void {
164 // deps when they are absent. Without it the dependency reached only 171 // deps when they are absent. Without it the dependency reached only
165 // `muxd`, and a clean checkout running `make test` first would have 172 // `muxd`, and a clean checkout running `make test` first would have
166 // found no libraries and no explanation. 173 // found no libraries and no explanation.
167 if (mod == server_mod or mod == quic_mod) linkQuic(b, t, quic); 174 if (mod == server_mod or mod == quic_mod or mod == exe_mod) linkQuic(b, t, quic);
168 test_step.dependOn(&b.addRunArtifact(t).step); 175 test_step.dependOn(&b.addRunArtifact(t).step);
169 } 176 }
170 177
src/main.zig
Old New
@@ -5,104 +5,312 @@ const std = @import("std");
5 const Server = @import("server").Server; 5 const Server = @import("server").Server;
6 const proto = @import("protocol"); 6 const proto = @import("protocol");
7 const proxy = @import("proxy"); 7 const proxy = @import("proxy");
8 const quic = @import("quic");
8 9
9 const usage = 10 const usage =
10 \\usage: 11 \\usage:
11 \\ muxd run [--sock PATH] [--shell PATH] [--cols N] [--rows N] 12 \\ muxd run [--sock PATH] [--shell PATH] [--cols N] [--rows N]
13 \\ [--quic HOST:PORT --key FILE] [--quic-idle-ms N]
12 \\ muxd dump [--vt] [--sock PATH] 14 \\ muxd dump [--vt] [--sock PATH]
13 \\ muxd stats [--sock PATH] 15 \\ muxd stats [--sock PATH]
14 \\ muxd proxy [--sock PATH] (byte pump: stdio <-> session socket) 16 \\ muxd proxy [--sock PATH] (byte pump: stdio <-> session socket)
15 \\ 17 \\
16 ; 18 ;
17 19
18 pub fn main() !u8 { 20 /// Long enough that a quiet terminal is not a suspicious one, short enough
19 var gpa: std.heap.DebugAllocator(.{}) = .init; 21 /// that a client which has genuinely vanished stops being served within a
20 defer _ = gpa.deinit(); 22 /// few seconds of keepalives failing. Tunable because the reconnect tests
21 const alloc = gpa.allocator(); 23 /// need death declared on a schedule they can wait for.
24 const default_quic_idle_ms: u32 = 15_000;
22 25
23 const args = try std.process.argsAlloc(alloc); 26 const Cmd = enum { run, dump, stats, proxy };
24 defer std.process.argsFree(alloc, args);
25 27
26 if (args.len < 2) { 28 /// Everything the command line can say, once. Parsed away from `main` so it
27 std.debug.print("{s}", .{usage}); 29 /// can be tested without a process to exit from — the same reason
28 return 2; 30 /// mux_main.zig's parser is its own function.
29 } 31 const Opts = struct {
32 cmd: Cmd,
33 sock: ?[]const u8 = null,
34 shell: ?[]const u8 = null,
35 cols: u16 = 80,
36 rows: u16 = 24,
37 vt: bool = false,
38 /// `--quic` and `--key` are both-or-neither, enforced here, which is
39 /// what makes `quic != null` licence to unwrap `key`.
40 quic: ?[]const u8 = null,
41 key: ?[]const u8 = null,
42 quic_idle_ms: u32 = default_quic_idle_ms,
43 };
44
45 /// A refusal, carrying whatever `main` needs to print one line about it.
46 /// None of these is a daemon bug, so none of them gets a stack trace.
47 const Usage = union(enum) {
48 no_command,
49 unknown_command: []const u8,
50 unknown_arg: []const u8,
51 /// A flag at the end of argv with nothing left to consume.
52 missing_value: []const u8,
53 /// The flag whose value would not parse as the number it wants.
54 bad_number: []const u8,
55 quic_without_key,
56 key_without_quic,
57 };
58
59 const ParseResult = union(enum) { ok: Opts, err: Usage };
60
61 fn parseArgs(args: []const [:0]const u8) ParseResult {
62 if (args.len < 2) return .{ .err = .no_command };
63 const cmd: Cmd = if (std.mem.eql(u8, args[1], "run"))
64 .run
65 else if (std.mem.eql(u8, args[1], "dump"))
66 .dump
67 else if (std.mem.eql(u8, args[1], "stats"))
68 .stats
69 else if (std.mem.eql(u8, args[1], "proxy"))
70 .proxy
71 else
72 return .{ .err = .{ .unknown_command = args[1] } };
30 73
31 var sock_arg: ?[]const u8 = null; 74 var o: Opts = .{ .cmd = cmd };
32 var shell_arg: ?[]const u8 = null;
33 var vt_mode = false;
34 var cols: u16 = 80;
35 var rows: u16 = 24;
36 var i: usize = 2; 75 var i: usize = 2;
37 while (i < args.len) : (i += 1) { 76 while (i < args.len) : (i += 1) {
38 const a = args[i]; 77 const a = args[i];
39 if (std.mem.eql(u8, a, "--sock") and i + 1 < args.len) { 78 if (std.mem.eql(u8, a, "--vt")) {
40 i += 1; 79 o.vt = true;
41 sock_arg = args[i]; 80 continue;
42 } else if (std.mem.eql(u8, a, "--shell") and i + 1 < args.len) {
43 i += 1;
44 shell_arg = args[i];
45 } else if (std.mem.eql(u8, a, "--cols") and i + 1 < args.len) {
46 i += 1;
47 cols = try std.fmt.parseInt(u16, args[i], 10);
48 } else if (std.mem.eql(u8, a, "--rows") and i + 1 < args.len) {
49 i += 1;
50 rows = try std.fmt.parseInt(u16, args[i], 10);
51 } else if (std.mem.eql(u8, a, "--vt")) {
52 vt_mode = true;
53 } else {
54 std.debug.print("unknown argument: {s}\n{s}", .{ a, usage });
55 return 2;
56 } 81 }
82 // Every remaining flag takes a value, so the missing-value case is
83 // answered once here rather than at each arm — a flag with nothing
84 // after it used to fall through to "unknown argument", which named
85 // the wrong mistake.
86 const takes_value = std.mem.eql(u8, a, "--sock") or
87 std.mem.eql(u8, a, "--shell") or
88 std.mem.eql(u8, a, "--cols") or
89 std.mem.eql(u8, a, "--rows") or
90 std.mem.eql(u8, a, "--quic") or
91 std.mem.eql(u8, a, "--key") or
92 std.mem.eql(u8, a, "--quic-idle-ms");
93 if (!takes_value) return .{ .err = .{ .unknown_arg = a } };
94 if (i + 1 >= args.len) return .{ .err = .{ .missing_value = a } };
95 i += 1;
96 const v = args[i];
97 if (std.mem.eql(u8, a, "--sock")) {
98 o.sock = v;
99 } else if (std.mem.eql(u8, a, "--shell")) {
100 o.shell = v;
101 } else if (std.mem.eql(u8, a, "--quic")) {
102 o.quic = v;
103 } else if (std.mem.eql(u8, a, "--key")) {
104 o.key = v;
105 } else if (std.mem.eql(u8, a, "--cols")) {
106 o.cols = std.fmt.parseInt(u16, v, 10) catch return .{ .err = .{ .bad_number = a } };
107 } else if (std.mem.eql(u8, a, "--rows")) {
108 o.rows = std.fmt.parseInt(u16, v, 10) catch return .{ .err = .{ .bad_number = a } };
109 } else if (std.mem.eql(u8, a, "--quic-idle-ms")) {
110 // u32 rather than u64 so that an absurd value is a parse
111 // failure here instead of an overflow where it is multiplied
112 // out to nanoseconds. Zero is refused because ngtcp2 reads it
113 // as "no idle timeout", the opposite of what the flag says.
114 const n = std.fmt.parseInt(u32, v, 10) catch return .{ .err = .{ .bad_number = a } };
115 if (n == 0) return .{ .err = .{ .bad_number = a } };
116 o.quic_idle_ms = n;
117 }
118 }
119
120 // Both or neither. A key with nowhere to listen is as much a mistake as
121 // a listener with nothing to authenticate against — and there is no
122 // unauthenticated mode to fall back to, so neither can be a default.
123 if (o.quic != null and o.key == null) return .{ .err = .quic_without_key };
124 if (o.key != null and o.quic == null) return .{ .err = .key_without_quic };
125
126 return .{ .ok = o };
127 }
128
129 fn usageExit(u: Usage) u8 {
130 switch (u) {
131 .no_command => std.debug.print("{s}", .{usage}),
132 .unknown_command => std.debug.print("{s}", .{usage}),
133 .unknown_arg => |a| std.debug.print("unknown argument: {s}\n{s}", .{ a, usage }),
134 .missing_value => |f| std.debug.print("muxd: {s} needs a value\n{s}", .{ f, usage }),
135 .bad_number => |f| std.debug.print("muxd: {s} needs a positive number\n{s}", .{ f, usage }),
136 .quic_without_key => std.debug.print(
137 "muxd: --quic needs --key; there is no unauthenticated mode\n",
138 .{},
139 ),
140 .key_without_quic => std.debug.print(
141 "muxd: --key without --quic has nothing to listen on; name both or neither\n",
142 .{},
143 ),
57 } 144 }
145 return 2;
146 }
147
148 /// `HOST:PORT` where HOST is a literal address — `127.0.0.1:4433`,
149 /// `0.0.0.0:4433`, `[::]:4433`. Deliberately no DNS: this is the address to
150 /// bind, and a name resolving to several is a question, not an answer.
151 fn splitHostPort(s: []const u8) !struct { host: []const u8, port: u16 } {
152 if (s.len > 0 and s[0] == '[') {
153 const close = std.mem.indexOfScalar(u8, s, ']') orelse return error.MalformedAddress;
154 if (close + 1 >= s.len or s[close + 1] != ':') return error.MalformedAddress;
155 return .{ .host = s[1..close], .port = try parsePort(s[close + 2 ..]) };
156 }
157 const colon = std.mem.lastIndexOfScalar(u8, s, ':') orelse return error.MalformedAddress;
158 // An unbracketed IPv6 literal carries colons of its own, and splitting
159 // on the last one would quietly take its final group as a port:
160 // `fe80::1:4433` reads equally well as host `fe80::1` port 4433 and as
161 // host `fe80::1:4433` with the port left off. Brackets are how that
162 // ambiguity is spelled out, so without them it is refused rather than
163 // guessed at.
164 if (std.mem.indexOfScalar(u8, s[0..colon], ':') != null) return error.MalformedAddress;
165 return .{ .host = s[0..colon], .port = try parsePort(s[colon + 1 ..]) };
166 }
58 167
59 const sock_path = if (sock_arg) |s| 168 fn parsePort(s: []const u8) !u16 {
169 return std.fmt.parseInt(u16, s, 10) catch error.MalformedAddress;
170 }
171
172 fn parseBindAddr(s: []const u8) !std.net.Address {
173 const hp = try splitHostPort(s);
174 return std.net.Address.parseIp(hp.host, hp.port);
175 }
176
177 pub fn main() !u8 {
178 var gpa: std.heap.DebugAllocator(.{}) = .init;
179 defer _ = gpa.deinit();
180 const alloc = gpa.allocator();
181
182 const args = try std.process.argsAlloc(alloc);
183 defer std.process.argsFree(alloc, args);
184
185 const o = switch (parseArgs(args)) {
186 .err => |u| return usageExit(u),
187 .ok => |o| o,
188 };
189
190 const sock_path = if (o.sock) |s|
60 try alloc.dupe(u8, s) 191 try alloc.dupe(u8, s)
61 else 192 else
62 try defaultSockPath(alloc); 193 try defaultSockPath(alloc);
63 defer alloc.free(sock_path); 194 defer alloc.free(sock_path);
64 195
65 if (std.mem.eql(u8, args[1], "run")) { 196 switch (o.cmd) {
66 const shell_z: [:0]const u8 = if (shell_arg) |s| 197 .run => return run(alloc, o, sock_path),
67 try alloc.dupeZ(u8, s) 198 .dump => return dump(alloc, sock_path, o.vt),
68 else 199 .stats => return stats(alloc, sock_path),
69 try alloc.dupeZ(u8, std.posix.getenv("SHELL") orelse "/bin/sh"); 200 .proxy => return proxy.run(sock_path),
70 defer alloc.free(shell_z); 201 }
71 202 }
72 var srv = Server.init(alloc, .{ 203
73 .sock_path = sock_path, 204 fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {
74 .shell = shell_z, 205 // Address and key are settled before anything binds: a mistyped address
75 .cols = cols, 206 // or an unreadable key must not first leave a session socket and a live
76 .rows = rows, 207 // shell behind. Same discipline as the key loader's own refusals.
77 }) catch |err| switch (err) { 208 var quic_bind: ?std.net.Address = null;
78 // All of these mean "that path is not ours to take", and all 209 var quic_key: quic.Key = undefined;
79 // are ordinary operator mistakes rather than daemon bugs: say 210 if (o.quic) |hostport| {
80 // so in one line and exit, no stack trace. 211 quic_bind = parseBindAddr(hostport) catch {
81 // 212 std.debug.print(
82 // AddressInUse is the same situation found one syscall later: 213 "muxd: --quic wants HOST:PORT with a literal address, got {s}\n",
83 // two daemons starting at once can both see an empty path and 214 .{hostport},
84 // both try to bind it. The loser has simply lost a dead heat, 215 );
85 // and telling it "a daemon is already running" is exactly 216 return 1;
86 // right — by the time it reads the message, one is. 217 };
87 error.DaemonAlreadyRunning, error.AddressInUse => { 218 quic_key = quic.Key.load(o.key.?) catch |err| switch (err) {
88 std.debug.print("muxd: a daemon is already running on {s}\n", .{sock_path}); 219 error.KeyFileMissing => {
220 std.debug.print("muxd: no such key file: {s}\n", .{o.key.?});
221 return 1;
222 },
223 error.KeyFilePermissive => {
224 std.debug.print(
225 "muxd: {s} is readable by group or other; chmod 600 it\n",
226 .{o.key.?},
227 );
89 return 1; 228 return 1;
90 }, 229 },
91 error.SockPathNotASocket => { 230 error.KeyFileMalformed => {
92 std.debug.print("muxd: {s} exists and is not a socket\n", .{sock_path}); 231 std.debug.print(
232 "muxd: {s} is not a key: want 32 raw bytes or 64 hex characters\n",
233 .{o.key.?},
234 );
93 return 1; 235 return 1;
94 }, 236 },
95 else => return err, 237 else => return err,
96 }; 238 };
97 defer srv.deinit();
98 @import("server").installSignalHandlers();
99 return try srv.run();
100 } 239 }
101 if (std.mem.eql(u8, args[1], "dump")) return dump(alloc, sock_path, vt_mode); 240
102 if (std.mem.eql(u8, args[1], "stats")) return stats(alloc, sock_path); 241 // The UDP socket is bound BEFORE the session socket, so a port that is
103 if (std.mem.eql(u8, args[1], "proxy")) return proxy.run(sock_path); 242 // already taken costs nothing: no shell has been started and no socket
104 std.debug.print("{s}", .{usage}); 243 // left on disk. It is the same discipline as loading the key first, one
105 return 2; 244 // syscall further along.
245 var listener: ?*quic.Listener = null;
246 if (quic_bind) |addr| {
247 listener = quic.Listener.bind(alloc, addr, quic_key, o.quic_idle_ms) catch |err| switch (err) {
248 // The QUIC edition of "a daemon is already running", refused for
249 // the same reason: the listener sets no SO_REUSEADDR, so rather
250 // than silently splitting a port's datagrams with the daemon
251 // already there, the second one says so and stops.
252 error.AddressInUse => {
253 std.debug.print(
254 "muxd: a daemon is already listening on udp {s}\n",
255 .{o.quic.?},
256 );
257 return 1;
258 },
259 else => {
260 std.debug.print(
261 "muxd: cannot listen on udp {s}: {s}\n",
262 .{ o.quic.?, @errorName(err) },
263 );
264 return 1;
265 },
266 };
267 }
268 // Registered before the server's, so it runs after it: a client slot
269 // backed by QUIC closes through the listener, and the server must finish
270 // tearing its slots down before the listener is freed.
271 defer if (listener) |l| l.deinit();
272
273 const shell_z: [:0]const u8 = if (o.shell) |s|
274 try alloc.dupeZ(u8, s)
275 else
276 try alloc.dupeZ(u8, std.posix.getenv("SHELL") orelse "/bin/sh");
277 defer alloc.free(shell_z);
278
279 var srv = Server.init(alloc, .{
280 .sock_path = sock_path,
281 .shell = shell_z,
282 .cols = o.cols,
283 .rows = o.rows,
284 }) catch |err| switch (err) {
285 // All of these mean "that path is not ours to take", and all
286 // are ordinary operator mistakes rather than daemon bugs: say
287 // so in one line and exit, no stack trace.
288 //
289 // AddressInUse is the same situation found one syscall later:
290 // two daemons starting at once can both see an empty path and
291 // both try to bind it. The loser has simply lost a dead heat,
292 // and telling it "a daemon is already running" is exactly
293 // right — by the time it reads the message, one is.
294 error.DaemonAlreadyRunning, error.AddressInUse => {
295 std.debug.print("muxd: a daemon is already running on {s}\n", .{sock_path});
296 return 1;
297 },
298 error.SockPathNotASocket => {
299 std.debug.print("muxd: {s} exists and is not a socket\n", .{sock_path});
300 return 1;
301 },
302 else => return err,
303 };
304
305 defer srv.deinit();
306
307 if (listener) |l| {
308 l.setHandler(srv.quicHandler());
309 srv.attachQuic(l);
310 }
311
312 @import("server").installSignalHandlers();
313 return try srv.run();
106 } 314 }
107 315
108 pub fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 { 316 pub fn defaultSockPath(alloc: std.mem.Allocator) ![]const u8 {
@@ -147,3 +355,129 @@ fn stats(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
147 } 355 }
148 return 1; 356 return 1;
149 } 357 }
358
359 // ---------------------------------------------------------------------------
360 // Tests. These run because `exe_mod` was added to build.zig's test loop in
361 // the same commit; before that, a test written here would have compiled and
362 // silently never executed (the hazard recorded in decisions.md, which cost
363 // mux_main.zig five invisible tests).
364 // ---------------------------------------------------------------------------
365
366 /// parseArgs takes what argsAlloc produces, so the tests must speak the same
367 /// type: a slice of sentinel-terminated strings.
368 fn parse(comptime argv: []const [:0]const u8) ParseResult {
369 return parseArgs(argv);
370 }
371
372 test "parseArgs: subcommands and their existing flags" {
373 const r = parse(&.{ "muxd", "run" });
374 try std.testing.expect(r == .ok);
375 try std.testing.expect(r.ok.cmd == .run);
376 try std.testing.expect(r.ok.sock == null);
377 try std.testing.expectEqual(@as(u16, 80), r.ok.cols);
378 try std.testing.expectEqual(@as(u16, 24), r.ok.rows);
379
380 const d = parse(&.{ "muxd", "dump", "--vt", "--sock", "/tmp/x.sock" });
381 try std.testing.expect(d.ok.cmd == .dump);
382 try std.testing.expect(d.ok.vt);
383 try std.testing.expectEqualStrings("/tmp/x.sock", d.ok.sock.?);
384
385 const g = parse(&.{ "muxd", "run", "--cols", "120", "--rows", "40", "--shell", "/bin/dash" });
386 try std.testing.expectEqual(@as(u16, 120), g.ok.cols);
387 try std.testing.expectEqual(@as(u16, 40), g.ok.rows);
388 try std.testing.expectEqualStrings("/bin/dash", g.ok.shell.?);
389
390 try std.testing.expect(parse(&.{"muxd"}).err == .no_command);
391 try std.testing.expect(parse(&.{ "muxd", "wat" }).err == .unknown_command);
392 try std.testing.expect(parse(&.{ "muxd", "run", "--wat" }).err == .unknown_arg);
393 }
394
395 test "parseArgs: --quic and --key are both or neither" {
396 const both = parse(&.{ "muxd", "run", "--quic", "0.0.0.0:4433", "--key", "/k" });
397 try std.testing.expect(both == .ok);
398 try std.testing.expectEqualStrings("0.0.0.0:4433", both.ok.quic.?);
399 try std.testing.expectEqualStrings("/k", both.ok.key.?);
400
401 // Naming one is not a request to guess the other: there is no
402 // unauthenticated listener and no default key path.
403 try std.testing.expect(parse(&.{ "muxd", "run", "--quic", "0.0.0.0:4433" }).err == .quic_without_key);
404 try std.testing.expect(parse(&.{ "muxd", "run", "--key", "/k" }).err == .key_without_quic);
405
406 // Neither is the ordinary case and must stay silent.
407 const neither = parse(&.{ "muxd", "run" });
408 try std.testing.expect(neither.ok.quic == null);
409 try std.testing.expect(neither.ok.key == null);
410 }
411
412 test "parseArgs: --quic-idle-ms defaults, parses, and refuses nonsense" {
413 const dflt = parse(&.{ "muxd", "run", "--quic", "127.0.0.1:1", "--key", "/k" });
414 try std.testing.expectEqual(default_quic_idle_ms, dflt.ok.quic_idle_ms);
415
416 const set = parse(&.{ "muxd", "run", "--quic", "127.0.0.1:1", "--key", "/k", "--quic-idle-ms", "2500" });
417 try std.testing.expectEqual(@as(u32, 2500), set.ok.quic_idle_ms);
418
419 // Zero means "no idle timeout" to ngtcp2 — the opposite of what anyone
420 // typing a timeout of zero is asking for, so it is refused rather than
421 // silently inverted.
422 try std.testing.expect(parse(&.{ "muxd", "run", "--quic-idle-ms", "0" }).err == .bad_number);
423 try std.testing.expect(parse(&.{ "muxd", "run", "--quic-idle-ms", "soon" }).err == .bad_number);
424 try std.testing.expect(parse(&.{ "muxd", "run", "--quic-idle-ms", "-5" }).err == .bad_number);
425 // Wider than u32: refused at the parse rather than overflowing where it
426 // is multiplied out to nanoseconds.
427 try std.testing.expect(parse(&.{ "muxd", "run", "--quic-idle-ms", "99999999999" }).err == .bad_number);
428 // The idle flag alone does not turn QUIC on, and must not smuggle the
429 // both-or-neither rule past the check.
430 try std.testing.expect(parse(&.{ "muxd", "run", "--quic-idle-ms", "2500" }) == .ok);
431
432 // Same treatment for the numbers that were already here.
433 try std.testing.expect(parse(&.{ "muxd", "run", "--cols", "wide" }).err == .bad_number);
434 try std.testing.expect(parse(&.{ "muxd", "run", "--rows", "99999" }).err == .bad_number);
435 }
436
437 test "parseArgs: a value-taking flag at the end of argv names itself" {
438 // This used to report "unknown argument: --quic", which blames the flag
439 // rather than the missing value.
440 inline for (.{ "--sock", "--shell", "--cols", "--rows", "--quic", "--key", "--quic-idle-ms" }) |flag| {
441 const r = parse(&.{ "muxd", "run", flag });
442 try std.testing.expect(r.err == .missing_value);
443 try std.testing.expectEqualStrings(flag, r.err.missing_value);
444 }
445 }
446
447 test "splitHostPort: literal addresses, bracketed and not" {
448 const v4 = try splitHostPort("127.0.0.1:4433");
449 try std.testing.expectEqualStrings("127.0.0.1", v4.host);
450 try std.testing.expectEqual(@as(u16, 4433), v4.port);
451
452 const v6 = try splitHostPort("[::1]:4433");
453 try std.testing.expectEqualStrings("::1", v6.host);
454 try std.testing.expectEqual(@as(u16, 4433), v6.port);
455
456 const any6 = try splitHostPort("[::]:1");
457 try std.testing.expectEqualStrings("::", any6.host);
458 try std.testing.expectEqual(@as(u16, 1), any6.port);
459
460 try std.testing.expectError(error.MalformedAddress, splitHostPort("127.0.0.1"));
461 try std.testing.expectError(error.MalformedAddress, splitHostPort("127.0.0.1:"));
462 try std.testing.expectError(error.MalformedAddress, splitHostPort("127.0.0.1:99999"));
463 try std.testing.expectError(error.MalformedAddress, splitHostPort("[::1]4433"));
464 try std.testing.expectError(error.MalformedAddress, splitHostPort("[::1]"));
465
466 // An IPv6 literal without brackets is ambiguous about where the address
467 // stops, so it is refused instead of being read either way.
468 try std.testing.expectError(error.MalformedAddress, splitHostPort("::1:4433"));
469 try std.testing.expectError(error.MalformedAddress, splitHostPort("fe80::1:4433"));
470 }
471
472 test "parseBindAddr: a hostname is refused, not resolved" {
473 const a = try parseBindAddr("127.0.0.1:4433");
474 try std.testing.expectEqual(@as(u16, 4433), a.getPort());
475
476 const six = try parseBindAddr("[::1]:4433");
477 try std.testing.expectEqual(@as(u16, 4433), six.getPort());
478 try std.testing.expect(six.any.family == std.posix.AF.INET6);
479
480 // No DNS at bind time, deliberately: this is the address to bind, and a
481 // name resolving to several is a question rather than an answer.
482 try std.testing.expect(std.meta.isError(parseBindAddr("localhost:4433")));
483 }
src/quic_server.zig
Old New
@@ -207,10 +207,32 @@ pub const Handler = struct {
207 onOpen: *const fn (ctx: *anyopaque, id: u64) void, 207 onOpen: *const fn (ctx: *anyopaque, id: u64) void,
208 /// Stream bytes arrived. Arbitrary chunking: the owner reassembles. 208 /// Stream bytes arrived. Arbitrary chunking: the owner reassembles.
209 onData: *const fn (ctx: *anyopaque, id: u64, bytes: []const u8) void, 209 onData: *const fn (ctx: *anyopaque, id: u64, bytes: []const u8) void,
210 /// A connection the LISTENER gave up on: an idle timeout, a protocol
211 /// error, a peer that went away.
212 ///
213 /// These two do not pair, and an owner that assumes they do will be
214 /// wrong in both directions. A handshake that never completes — a wrong
215 /// key, a client that vanishes mid-flight — produces neither: the
216 /// connection is torn down having never been announced, because there
217 /// was never anything to announce. And a close the owner itself asked
218 /// for via `closeConn` produces no callback either. `onClose` means
219 /// "this ended without you asking", nothing more.
210 onClose: *const fn (ctx: *anyopaque, id: u64) void, 220 onClose: *const fn (ctx: *anyopaque, id: u64) void,
211 }; 221 };
212 222
223 /// Connections the listener will hold at once. Deliberately larger than the
224 /// daemon's `max_clients`: a connection exists from the moment its handshake
225 /// completes, and only then asks for a client slot — so the extra room is
226 /// headroom for handshakes in flight, not for sessions. A connection that
227 /// finds no slot is answered and closed (see the daemon's onOpen), which is
228 /// why the two numbers do not need to agree.
213 const max_conns = 16; 229 const max_conns = 16;
230
231 /// Source Connection IDs cached per connection. ngtcp2 caps its own pool at
232 /// NGTCP2_MAX_SCID_POOL_SIZE (8), so this is headroom; and because the
233 /// primary CID is matched separately, a cache that ever did fill would
234 /// degrade to the old primary-only behaviour rather than misroute.
235 const max_cids = 16;
214 const max_udp = 1452; // conservative IPv4 datagram that avoids fragmenting 236 const max_udp = 1452; // conservative IPv4 datagram that avoids fragmenting
215 237
216 /// wolfSSL's PSK callbacks carry no user pointer, so the key has to be 238 /// wolfSSL's PSK callbacks carry no user pointer, so the key has to be
@@ -244,6 +266,93 @@ const psk_identity: [*:0]const u8 = "mux";
244 const psk_ciphersuite: [*:0]const u8 = "TLS13-AES128-GCM-SHA256"; 266 const psk_ciphersuite: [*:0]const u8 = "TLS13-AES128-GCM-SHA256";
245 const alpn = "\x03mux"; 267 const alpn = "\x03mux";
246 268
269 /// How many outbound bytes one connection may hold. Sized to the stream
270 /// window the peer advertises, because holding much more than the peer will
271 /// let us send buys nothing: past this the daemon's own `pending_cap` is the
272 /// right place for the backlog to sit and be judged.
273 const egress_cap = 256 * 1024;
274
275 /// Outbound stream bytes, in a ring that never moves a byte once written.
276 ///
277 /// A ring rather than a growable buffer, because ngtcp2 does NOT copy stream
278 /// payload: `ngtcp2_conn_writev_stream` stores the *vector* it is handed —
279 /// `ngtcp2_vec_copy` is a memcpy of base+len, not of the bytes — in its
280 /// retransmission queue, and re-reads those bytes if the packet is lost. A
281 /// buffer that reallocates on append, or that clears once the last byte has
282 /// been handed over, therefore leaves ngtcp2 holding a freed or recycled
283 /// pointer, and it dies inside `ngtcp2_pkt_encode_stream_frame`. It did:
284 /// rarely, only under loss, and only in the one test big enough to overflow
285 /// a socket buffer — which is the worst shape a bug can have.
286 ///
287 /// So the invariant, and everything here exists to hold it: **a byte handed
288 /// to ngtcp2 does not move or get overwritten until the peer acknowledges
289 /// it.** `head` advances only from `acked_stream_data_offset`; a write can
290 /// only land in the free space that leaves. Being fixed in size is the other
291 /// half of the point — a full ring is the backpressure signal.
292 const Egress = struct {
293 buf: []u8,
294 /// The oldest byte the peer has not acknowledged.
295 head: usize = 0,
296 /// Bytes from `head` still owed: handed to ngtcp2 and unacknowledged,
297 /// plus not yet handed over.
298 held: usize = 0,
299 /// The tail of `held` that ngtcp2 has not taken yet.
300 unsent: usize = 0,
301
302 fn deinit(self: *Egress, alloc: std.mem.Allocator) void {
303 alloc.free(self.buf);
304 self.* = .{ .buf = &.{} };
305 }
306
307 fn freeSpace(self: *const Egress) usize {
308 return self.buf.len - self.held;
309 }
310
311 /// Take what fits and report how much that was. A short return is not an
312 /// error, it is the whole mechanism: the caller keeps the remainder and
313 /// is thereby the one holding — and bounding — the backlog.
314 fn push(self: *Egress, bytes: []const u8) usize {
315 const n = @min(bytes.len, self.freeSpace());
316 if (n == 0) return 0;
317 const start = (self.head + self.held) % self.buf.len;
318 const first = @min(n, self.buf.len - start);
319 @memcpy(self.buf[start..][0..first], bytes[0..first]);
320 if (first < n) @memcpy(self.buf[0 .. n - first], bytes[first..n]);
321 self.held += n;
322 self.unsent += n;
323 return n;
324 }
325
326 /// The unsent region as up to two vectors — two when it wraps, which is
327 /// the price of never moving a byte, and ngtcp2 takes a vector array
328 /// precisely so that price is payable.
329 fn vecs(self: *const Egress, out: *[2]c.ngtcp2_vec) usize {
330 if (self.unsent == 0) return 0;
331 const start = (self.head + (self.held - self.unsent)) % self.buf.len;
332 const first = @min(self.unsent, self.buf.len - start);
333 out[0] = .{ .base = self.buf.ptr + start, .len = first };
334 if (first == self.unsent) return 1;
335 out[1] = .{ .base = self.buf.ptr, .len = self.unsent - first };
336 return 2;
337 }
338
339 /// ngtcp2 took `n` bytes off the unsent region. They stay exactly where
340 /// they are — it now has pointers to them.
341 fn took(self: *Egress, n: usize) void {
342 self.unsent -= @min(n, self.unsent);
343 }
344
345 /// The peer acknowledged `n` more bytes. ngtcp2 documents this callback
346 /// as arriving "sequentially in increasing order of offset without any
347 /// overlap", so a running count IS the acknowledged prefix, and this is
348 /// the only thing that ever frees space.
349 fn ack(self: *Egress, n: usize) void {
350 const taken = @min(n, self.held - self.unsent);
351 self.head = (self.head + taken) % self.buf.len;
352 self.held -= taken;
353 }
354 };
355
247 /// One authenticated peer: an ngtcp2 connection, its TLS object, and the 356 /// One authenticated peer: an ngtcp2 connection, its TLS object, and the
248 /// single bidirectional stream that carries everything. 357 /// single bidirectional stream that carries everything.
249 const Conn = struct { 358 const Conn = struct {
@@ -257,13 +366,43 @@ const Conn = struct {
257 local_storage: std.posix.sockaddr.storage = undefined, 366 local_storage: std.posix.sockaddr.storage = undefined,
258 local_len: std.posix.socklen_t = 0, 367 local_len: std.posix.socklen_t = 0,
259 scid: c.ngtcp2_cid = undefined, 368 scid: c.ngtcp2_cid = undefined,
369 /// Every Connection ID this endpoint has advertised and not retired.
370 /// A peer is entitled to address us by any of them — that is what makes
371 /// connection migration work — so matching only `scid` meant a client
372 /// that switched CID had its packets fall through to `accept`, where
373 /// they were dropped as noise.
374 cids: [max_cids]c.ngtcp2_cid = undefined,
375 ncids: usize = 0,
376 cids_dirty: bool = true,
260 stream_id: i64 = -1, 377 stream_id: i64 = -1,
261 /// Bytes the owner asked us to send that ngtcp2 has not taken yet. 378 /// Bytes owed to this peer. See Egress: they do not move until acked.
262 out: std.ArrayList(u8) = .empty, 379 out: Egress,
263 out_sent: usize = 0,
264 opened: bool = false, 380 opened: bool = false,
265 381
266 fn deinit(self: *Conn, alloc: std.mem.Allocator) void { 382 /// Re-read the advertised CIDs from ngtcp2. Lazy, because the answer
383 /// only changes when one is issued or retired and both of those tell us.
384 fn refreshCids(self: *Conn) void {
385 self.cids_dirty = false;
386 self.ncids = 0;
387 const conn = self.conn orelse return;
388 const n = c.ngtcp2_conn_get_scid2(conn, null);
389 if (n == 0 or n > max_cids) return;
390 self.ncids = c.ngtcp2_conn_get_scid2(conn, &self.cids);
391 }
392
393 /// Is this connection addressed by `dcid`? The primary is checked first
394 /// and without the cache, so this can only ever be more permissive than
395 /// the CID the connection was created with — never less.
396 fn matches(self: *Conn, dcid: []const u8) bool {
397 if (cidEql(&self.scid, dcid)) return true;
398 if (self.cids_dirty) self.refreshCids();
399 for (self.cids[0..self.ncids]) |*cid| {
400 if (cidEql(cid, dcid)) return true;
401 }
402 return false;
403 }
404
405 pub fn deinit(self: *Conn, alloc: std.mem.Allocator) void {
267 self.out.deinit(alloc); 406 self.out.deinit(alloc);
268 if (self.conn) |cn| c.ngtcp2_conn_del(cn); 407 if (self.conn) |cn| c.ngtcp2_conn_del(cn);
269 if (self.ssl) |s| c.wolfSSL_free(s); 408 if (self.ssl) |s| c.wolfSSL_free(s);
@@ -281,6 +420,45 @@ fn randCb(dest: [*c]u8, destlen: usize, _: [*c]const c.ngtcp2_rand_ctx) callconv
281 std.crypto.random.bytes(dest[0..destlen]); 420 std.crypto.random.bytes(dest[0..destlen]);
282 } 421 }
283 422
423 /// The handler a listener carries between `bind` and `setHandler`: it
424 /// exists so that window has no null to check on every packet.
425 fn ignoreOpen(_: *anyopaque, _: u64) void {}
426 fn ignoreData(_: *anyopaque, _: u64, _: []const u8) void {}
427 fn ignoreClose(_: *anyopaque, _: u64) void {}
428
429 fn cidEql(cid: *const c.ngtcp2_cid, bytes: []const u8) bool {
430 if (cid.datalen != bytes.len) return false;
431 return std.mem.eql(u8, cid.data[0..cid.datalen], bytes);
432 }
433
434 /// The server's variant: mints the CID exactly as the shared callback does,
435 /// and marks the connection's cache stale. Separate from `getNewCidCb`
436 /// because the test client shares that one and its user_data is not a Conn.
437 fn serverGetNewCidCb(
438 conn: ?*c.ngtcp2_conn,
439 cid: [*c]c.ngtcp2_cid,
440 token: [*c]c.ngtcp2_stateless_reset_token,
441 cidlen: usize,
442 user_data: ?*anyopaque,
443 ) callconv(.c) c_int {
444 const rv = getNewCidCb(conn, cid, token, cidlen, user_data);
445 const cn: *Conn = @ptrCast(@alignCast(user_data.?));
446 cn.cids_dirty = true;
447 return rv;
448 }
449
450 /// A CID was retired: the cache must stop matching it, or this listener
451 /// would keep answering to a name the peer has been told to forget.
452 fn serverRemoveCidCb(
453 _: ?*c.ngtcp2_conn,
454 _: [*c]const c.ngtcp2_cid,
455 user_data: ?*anyopaque,
456 ) callconv(.c) c_int {
457 const cn: *Conn = @ptrCast(@alignCast(user_data.?));
458 cn.cids_dirty = true;
459 return 0;
460 }
461
284 fn getNewCidCb( 462 fn getNewCidCb(
285 _: ?*c.ngtcp2_conn, 463 _: ?*c.ngtcp2_conn,
286 cid: [*c]c.ngtcp2_cid, 464 cid: [*c]c.ngtcp2_cid,
@@ -301,6 +479,22 @@ fn handshakeCompletedCb(_: ?*c.ngtcp2_conn, user_data: ?*anyopaque) callconv(.c)
301 return 0; 479 return 0;
302 } 480 }
303 481
482 /// The peer acknowledged stream bytes, so the space they occupy can be
483 /// reused. This callback is the ONLY thing that frees egress space, and
484 /// registering it is what makes the buffer's no-move invariant affordable.
485 fn ackedStreamDataCb(
486 _: ?*c.ngtcp2_conn,
487 _: i64,
488 _: u64,
489 datalen: u64,
490 user_data: ?*anyopaque,
491 _: ?*anyopaque,
492 ) callconv(.c) c_int {
493 const cn: *Conn = @ptrCast(@alignCast(user_data.?));
494 cn.out.ack(@intCast(datalen));
495 return 0;
496 }
497
304 fn streamOpenCb(_: ?*c.ngtcp2_conn, stream_id: i64, user_data: ?*anyopaque) callconv(.c) c_int { 498 fn streamOpenCb(_: ?*c.ngtcp2_conn, stream_id: i64, user_data: ?*anyopaque) callconv(.c) c_int {
305 const cn: *Conn = @ptrCast(@alignCast(user_data.?)); 499 const cn: *Conn = @ptrCast(@alignCast(user_data.?));
306 // One stream per connection: the first the client opens is the session. 500 // One stream per connection: the first the client opens is the session.
@@ -329,6 +523,14 @@ fn recvStreamDataCb(
329 _ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen); 523 _ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen);
330 _ = c.ngtcp2_conn_extend_max_offset(conn, datalen); 524 _ = c.ngtcp2_conn_extend_max_offset(conn, datalen);
331 525
526 // Note the order: the window is extended BEFORE the owner has done
527 // anything with these bytes, so the credit is granted against a consume
528 // that has not happened yet. That is the right trade here and not an
529 // oversight — the owner's `onData` is synchronous and bounded (it
530 // appends to a queue the daemon's own cap governs), so there is no
531 // second buffer to overflow by being generous. It would stop being the
532 // right trade the moment an owner could hold bytes indefinitely, which
533 // is exactly what the egress ring now refuses to let it do.
332 if (datalen > 0) { 534 if (datalen > 0) {
333 cn.listener.handler.onData(cn.listener.handler.ctx, cn.id, data[0..datalen]); 535 cn.listener.handler.onData(cn.listener.handler.ctx, cn.id, data[0..datalen]);
334 } 536 }
@@ -348,6 +550,32 @@ pub const Listener = struct {
348 retry_secret: [32]u8, 550 retry_secret: [32]u8,
349 idle_ms: u64, 551 idle_ms: u64,
350 552
553 /// Bind the socket and stand up TLS: everything that can fail for
554 /// reasons outside this process. Split from the handler because the
555 /// daemon that will own these connections does not exist until its own
556 /// socket is bound, and this one has to be bound FIRST — otherwise a
557 /// refused UDP port has already cost a session socket and a live shell.
558 ///
559 /// A listener returned from here drops anything that arrives until
560 /// `setHandler` is called. Nothing polls it before then.
561 pub fn bind(
562 alloc: std.mem.Allocator,
563 bind_addr: std.net.Address,
564 key: Key,
565 idle_ms: u64,
566 ) !*Listener {
567 return init(alloc, bind_addr, key, .{
568 .ctx = undefined,
569 .onOpen = ignoreOpen,
570 .onData = ignoreData,
571 .onClose = ignoreClose,
572 }, idle_ms);
573 }
574
575 pub fn setHandler(self: *Listener, h: Handler) void {
576 self.handler = h;
577 }
578
351 pub fn init( 579 pub fn init(
352 alloc: std.mem.Allocator, 580 alloc: std.mem.Allocator,
353 bind_addr: std.net.Address, 581 bind_addr: std.net.Address,
@@ -363,7 +591,14 @@ pub const Listener = struct {
363 0, 591 0,
364 ); 592 );
365 errdefer std.posix.close(fd); 593 errdefer std.posix.close(fd);
366 try std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.REUSEADDR, &std.mem.toBytes(@as(c_int, 1))); 594 // Deliberately NO SO_REUSEADDR. On UDP it lets a second daemon bind
595 // the same address, and the kernel then hands each datagram to one
596 // of them: two sessions silently splitting one port, with packets
597 // going to whichever process the kernel picked. The unix socket has
598 // a whole story for "a daemon is already running" precisely because
599 // taking over another daemon's endpoint by accident is unacceptable;
600 // this is that story's QUIC edition, and the answer is the same —
601 // fail the bind, loudly, and let the operator decide.
367 try std.posix.bind(fd, &bind_addr.any, bind_addr.getOsSockLen()); 602 try std.posix.bind(fd, &bind_addr.any, bind_addr.getOsSockLen());
368 603
369 if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit; 604 if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit;
@@ -420,14 +655,57 @@ pub const Listener = struct {
420 return null; 655 return null;
421 } 656 }
422 657
423 /// Queue bytes for a peer's stream. Egress is drained here and again on 658 /// Queue what fits of `bytes` for a peer's stream and report how many
424 /// every event, so a caller never has to think about flushing. 659 /// were taken. Egress is drained here and again on every event, so a
425 pub fn send(self: *Listener, id: u64, bytes: []const u8) !void { 660 /// caller never has to think about flushing.
661 ///
662 /// A short return is the backpressure signal, and it has to exist: while
663 /// this accepted everything unconditionally, a peer that stopped reading
664 /// grew an unbounded buffer down here, where nothing watches it, instead
665 /// of tripping the daemon's `pending_cap` up where something does. The
666 /// contract now matches the socket sink's — take what you can, tell the
667 /// truth about how much — so one rule bounds both kinds of client.
668 pub fn send(self: *Listener, id: u64, bytes: []const u8) !usize {
426 const cn = self.find(id) orelse return error.NoSuchConn; 669 const cn = self.find(id) orelse return error.NoSuchConn;
427 try cn.out.appendSlice(self.alloc, bytes); 670 const n = cn.out.push(bytes);
671 self.drain(cn);
672 return n;
673 }
674
675 /// Bytes accepted from the owner that the peer has not acknowledged.
676 /// Zero means this connection owes nothing — the only honest answer to
677 /// "has it all gone out", since a QUIC send is not done when the syscall
678 /// returns but when the ack arrives.
679 pub fn pendingBytes(self: *Listener, id: u64) usize {
680 const cn = self.find(id) orelse return 0;
681 return cn.out.held;
682 }
683
684 /// Push a connection's egress along without an event to hang it off —
685 /// what the daemon calls when it is draining on the way out, or when
686 /// acks have just freed room that queued bytes are waiting for.
687 pub fn kick(self: *Listener, id: u64) void {
688 const cn = self.find(id) orelse return;
428 self.drain(cn); 689 self.drain(cn);
429 } 690 }
430 691
692 /// Close ONE connection and nothing else — never the socket it shares.
693 ///
694 /// The distinction is the whole reason a client slot stopped being a
695 /// descriptor: every QUIC peer on this daemon is multiplexed over one
696 /// UDP socket, so closing "the client's transport" because one session
697 /// ended would take down every other session with it.
698 ///
699 /// Two things this deliberately does NOT do, both of which callers have
700 /// to know. It does not invoke `onClose`: a close the owner asked for
701 /// needs no callback telling the owner what it just did, and calling
702 /// back into a handler mid-teardown is how re-entrancy bugs start. Only
703 /// `kill` — the listener deciding a connection is finished — calls back.
704 /// And it does not send CONNECTION_CLOSE: the peer finds out when its
705 /// idle timer expires. That is a real cost (a client learns of a
706 /// deliberate close no faster than of a crash) and it is accepted for
707 /// now rather than unnoticed; a graceful close belongs with the client
708 /// transport, which is the side that would act on it.
431 pub fn closeConn(self: *Listener, id: u64) void { 709 pub fn closeConn(self: *Listener, id: u64) void {
432 for (&self.conns) |*slot| { 710 for (&self.conns) |*slot| {
433 if (slot.*) |cn| { 711 if (slot.*) |cn| {
@@ -486,12 +764,17 @@ pub const Listener = struct {
486 } else return; // full: drop, the peer will retry 764 } else return; // full: drop, the peer will retry
487 765
488 const cn = self.alloc.create(Conn) catch return; 766 const cn = self.alloc.create(Conn) catch return;
489 cn.* = .{ .id = self.next_id, .listener = self }; 767 const ring = self.alloc.alloc(u8, egress_cap) catch {
768 self.alloc.destroy(cn);
769 return;
770 };
771 cn.* = .{ .id = self.next_id, .listener = self, .out = .{ .buf = ring } };
490 self.next_id += 1; 772 self.next_id += 1;
491 @memcpy(std.mem.asBytes(&cn.remote)[0..from_len], std.mem.asBytes(from)[0..from_len]); 773 @memcpy(std.mem.asBytes(&cn.remote)[0..from_len], std.mem.asBytes(from)[0..from_len]);
492 cn.remote_len = from_len; 774 cn.remote_len = from_len;
493 cn.local_len = @sizeOf(std.posix.sockaddr.storage); 775 cn.local_len = @sizeOf(std.posix.sockaddr.storage);
494 std.posix.getsockname(self.fd, @ptrCast(&cn.local_storage), &cn.local_len) catch { 776 std.posix.getsockname(self.fd, @ptrCast(&cn.local_storage), &cn.local_len) catch {
777 cn.out.deinit(self.alloc);
495 self.alloc.destroy(cn); 778 self.alloc.destroy(cn);
496 return; 779 return;
497 }; 780 };
@@ -504,6 +787,7 @@ pub const Listener = struct {
504 cn.scid = hd.dcid; 787 cn.scid = hd.dcid;
505 788
506 const ssl = c.wolfSSL_new(self.ssl_ctx) orelse { 789 const ssl = c.wolfSSL_new(self.ssl_ctx) orelse {
790 cn.out.deinit(self.alloc);
507 self.alloc.destroy(cn); 791 self.alloc.destroy(cn);
508 return; 792 return;
509 }; 793 };
@@ -523,10 +807,12 @@ pub const Listener = struct {
523 cbs.get_path_challenge_data = c.ngtcp2_crypto_get_path_challenge_data_cb; 807 cbs.get_path_challenge_data = c.ngtcp2_crypto_get_path_challenge_data_cb;
524 cbs.version_negotiation = c.ngtcp2_crypto_version_negotiation_cb; 808 cbs.version_negotiation = c.ngtcp2_crypto_version_negotiation_cb;
525 cbs.rand = randCb; 809 cbs.rand = randCb;
526 cbs.get_new_connection_id2 = getNewCidCb; 810 cbs.get_new_connection_id2 = serverGetNewCidCb;
811 cbs.remove_connection_id = serverRemoveCidCb;
527 cbs.handshake_completed = handshakeCompletedCb; 812 cbs.handshake_completed = handshakeCompletedCb;
528 cbs.stream_open = streamOpenCb; 813 cbs.stream_open = streamOpenCb;
529 cbs.recv_stream_data = recvStreamDataCb; 814 cbs.recv_stream_data = recvStreamDataCb;
815 cbs.acked_stream_data_offset = ackedStreamDataCb;
530 816
531 var settings: c.ngtcp2_settings = undefined; 817 var settings: c.ngtcp2_settings = undefined;
532 c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings); 818 c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings);
@@ -570,11 +856,20 @@ pub const Listener = struct {
570 cn, 856 cn,
571 ) != 0) { 857 ) != 0) {
572 c.wolfSSL_free(ssl); 858 c.wolfSSL_free(ssl);
859 cn.out.deinit(self.alloc);
573 self.alloc.destroy(cn); 860 self.alloc.destroy(cn);
574 return; 861 return;
575 } 862 }
576 cn.conn = conn; 863 cn.conn = conn;
577 c.ngtcp2_conn_set_tls_native_handle(conn, ssl); 864 c.ngtcp2_conn_set_tls_native_handle(conn, ssl);
865 // Without this, a terminal nobody is typing into is
866 // indistinguishable from a peer that has gone away, and the
867 // connection is dropped after idle_ms of quiet — which for a
868 // default of 15s means a session dies while its owner reads the
869 // screen. The keep-alive PING is ack-eliciting, so its ACK restarts
870 // the idle timer at both ends; a peer that has genuinely vanished
871 // answers nothing and still times out on schedule.
872 c.ngtcp2_conn_set_keep_alive_timeout(conn, keepAliveNs(self.idle_ms));
578 slot.* = cn; 873 slot.* = cn;
579 874
580 self.feed(slot, pkt, from, from_len); 875 self.feed(slot, pkt, from, from_len);
@@ -687,11 +982,16 @@ pub const Listener = struct {
687 const rv = c.ngtcp2_pkt_decode_version_cid(&vc, pkt.ptr, pkt.len, 8); 982 const rv = c.ngtcp2_pkt_decode_version_cid(&vc, pkt.ptr, pkt.len, 8);
688 if (rv != 0) return; 983 if (rv != 0) return;
689 984
690 // Known connection? Match on the destination CID we handed out. 985 // Known connection? Match on ANY Connection ID this endpoint has
986 // advertised, not merely the one the connection was created with.
987 // A client that migrates — a new network, a NAT rebind, the exact
988 // case QUIC exists to survive — switches to another CID we gave it
989 // (RFC 9000 section 9.5). Matching only the first one sent those
990 // packets to `accept`, which has no token for them and drops them:
991 // the migration presents as a dead connection.
691 for (&self.conns) |*slot| { 992 for (&self.conns) |*slot| {
692 const cn = slot.* orelse continue; 993 const cn = slot.* orelse continue;
693 if (cn.scid.datalen != vc.dcidlen) continue; 994 if (!cn.matches(vc.dcid[0..vc.dcidlen])) continue;
694 if (!std.mem.eql(u8, cn.scid.data[0..cn.scid.datalen], vc.dcid[0..vc.dcidlen])) continue;
695 self.feed(slot, pkt, from, from_len); 995 self.feed(slot, pkt, from, from_len);
696 return; 996 return;
697 } 997 }
@@ -723,25 +1023,24 @@ pub const Listener = struct {
723 1023
724 /// Drive egress until ngtcp2 has nothing more to send. Runs after every 1024 /// Drive egress until ngtcp2 has nothing more to send. Runs after every
725 /// event, which is the contract the library expects. 1025 /// event, which is the contract the library expects.
726 fn drain(self: *Listener, cn: *Conn) void { 1026 pub fn drain(self: *Listener, cn: *Conn) void {
727 const conn = cn.conn orelse return; 1027 const conn = cn.conn orelse return;
728 var buf: [max_udp]u8 = undefined; 1028 var buf: [max_udp]u8 = undefined;
1029 // Set when the peer's stream window is full. Everything that is not
1030 // stream data still has to leave.
1031 var stream_blocked = false;
729 while (true) { 1032 while (true) {
730 var ps: c.ngtcp2_path_storage = undefined; 1033 var ps: c.ngtcp2_path_storage = undefined;
731 c.ngtcp2_path_storage_zero(&ps); 1034 c.ngtcp2_path_storage_zero(&ps);
732 var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; 1035 var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
733 var wrote: c.ngtcp2_ssize = 0; 1036 var wrote: c.ngtcp2_ssize = 0;
734 1037
735 var vec: c.ngtcp2_vec = undefined; 1038 var vecs: [2]c.ngtcp2_vec = undefined;
736 var vcnt: usize = 0; 1039 var vcnt: usize = 0;
737 var sid: i64 = -1; 1040 var sid: i64 = -1;
738 if (cn.stream_id != -1 and cn.out_sent < cn.out.items.len) { 1041 if (!stream_blocked and cn.stream_id != -1) {
739 vec = .{ 1042 vcnt = cn.out.vecs(&vecs);
740 .base = cn.out.items.ptr + cn.out_sent, 1043 if (vcnt > 0) sid = cn.stream_id;
741 .len = cn.out.items.len - cn.out_sent,
742 };
743 vcnt = 1;
744 sid = cn.stream_id;
745 } 1044 }
746 1045
747 const n = c.ngtcp2_conn_writev_stream_versioned( 1046 const n = c.ngtcp2_conn_writev_stream_versioned(
@@ -754,12 +1053,22 @@ pub const Listener = struct {
754 &wrote, 1053 &wrote,
755 0, 1054 0,
756 sid, 1055 sid,
757 if (vcnt > 0) &vec else null, 1056 if (vcnt > 0) &vecs else null,
758 vcnt, 1057 vcnt,
759 timestampNs(), 1058 timestampNs(),
760 ); 1059 );
1060 if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) {
1061 // A documented return, not a failure: the peer has no window
1062 // left for this stream. Treating it as fatal abandoned the
1063 // whole egress loop — including the ACKs and the flow-control
1064 // updates that are how the peer's window reopens, which turns
1065 // a moment of backpressure into a stall that never resolves.
1066 // Retry the same iteration carrying no stream data.
1067 stream_blocked = true;
1068 continue;
1069 }
761 if (n < 0) return; 1070 if (n < 0) return;
762 if (wrote > 0) cn.out_sent += @intCast(wrote); 1071 if (wrote > 0) cn.out.took(@intCast(wrote));
763 if (n == 0) break; 1072 if (n == 0) break;
764 1073
765 _ = std.posix.sendto( 1074 _ = std.posix.sendto(
@@ -770,13 +1079,6 @@ pub const Listener = struct {
770 cn.remote_len, 1079 cn.remote_len,
771 ) catch break; 1080 ) catch break;
772 } 1081 }
773
774 // Reclaim once the queue is fully handed over, so a long session
775 // does not grow an unbounded outbound buffer.
776 if (cn.out_sent > 0 and cn.out_sent == cn.out.items.len) {
777 cn.out.clearRetainingCapacity();
778 cn.out_sent = 0;
779 }
780 } 1082 }
781 1083
782 fn kill(self: *Listener, slot: *?*Conn) void { 1084 fn kill(self: *Listener, slot: *?*Conn) void {
@@ -790,6 +1092,14 @@ pub const Listener = struct {
790 } 1092 }
791 }; 1093 };
792 1094
1095 /// A third of the idle timeout, so two keepalives can go unanswered before
1096 /// the connection is called dead. Never zero: ngtcp2 reads a zero timeout as
1097 /// "disabled", exactly as it reads UINT64_MAX, so a small idle_ms rounding
1098 /// down would silently restore the behaviour the keepalive exists to prevent.
1099 fn keepAliveNs(idle_ms: u64) u64 {
1100 return @max(1, idle_ms / 3) * 1_000_000;
1101 }
1102
793 fn timestampNs() u64 { 1103 fn timestampNs() u64 {
794 const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch return 0; 1104 const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch return 0;
795 return @as(u64, @intCast(ts.sec)) * 1_000_000_000 + @as(u64, @intCast(ts.nsec)); 1105 return @as(u64, @intCast(ts.sec)) * 1_000_000_000 + @as(u64, @intCast(ts.nsec));
@@ -823,7 +1133,10 @@ fn pskClientCb(
823 return key_len; 1133 return key_len;
824 } 1134 }
825 1135
826 const TestClient = struct { 1136 /// Test scaffolding, exported so the daemon's own tests can drive a real
1137 /// QUIC peer. Not a preview of Task 3's client transport: no reconnect, no
1138 /// resumption, and a fixed receive buffer.
1139 pub const TestClient = struct {
827 fd: std.posix.fd_t, 1140 fd: std.posix.fd_t,
828 ssl_ctx: ?*c.WOLFSSL_CTX, 1141 ssl_ctx: ?*c.WOLFSSL_CTX,
829 ssl: ?*c.WOLFSSL, 1142 ssl: ?*c.WOLFSSL,
@@ -836,8 +1149,20 @@ const TestClient = struct {
836 stream_id: i64 = -1, 1149 stream_id: i64 = -1,
837 handshake_done: bool = false, 1150 handshake_done: bool = false,
838 echoed: usize = 0, 1151 echoed: usize = 0,
1152 /// When set, every arriving byte is compared against this at the offset
1153 /// it arrives at. A transport that echoes the right COUNT of the wrong
1154 /// bytes satisfies a length assertion and fails this one.
1155 verify: ?[]const u8 = null,
1156 mismatch: bool = false,
1157 /// Whether to give the peer more stream window as data is consumed.
1158 /// Turning it off is how a test reaches the blocked-stream path, which
1159 /// is a documented return value rather than a failure.
1160 extend: bool = true,
839 out: []const u8 = &.{}, 1161 out: []const u8 = &.{},
840 out_sent: usize = 0, 1162 out_sent: usize = 0,
1163 /// Everything the peer sent, so a test can walk frames out of it.
1164 recv_buf: [128 * 1024]u8 = undefined,
1165 recv_len: usize = 0,
841 1166
842 fn getConn(ref: [*c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?*c.ngtcp2_conn { 1167 fn getConn(ref: [*c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?*c.ngtcp2_conn {
843 const s: *TestClient = @ptrCast(@alignCast(ref.*.user_data)); 1168 const s: *TestClient = @ptrCast(@alignCast(ref.*.user_data));
@@ -864,7 +1189,7 @@ const TestClient = struct {
864 _: u32, 1189 _: u32,
865 stream_id: i64, 1190 stream_id: i64,
866 _: u64, 1191 _: u64,
867 _: [*c]const u8, 1192 data: [*c]const u8,
868 datalen: usize, 1193 datalen: usize,
869 ud: ?*anyopaque, 1194 ud: ?*anyopaque,
870 _: ?*anyopaque, 1195 _: ?*anyopaque,
@@ -873,13 +1198,27 @@ const TestClient = struct {
873 // The client must extend too: without this the SERVER stalls once 1198 // The client must extend too: without this the SERVER stalls once
874 // it has echoed a window's worth back, and the large-payload test 1199 // it has echoed a window's worth back, and the large-payload test
875 // below would hang rather than fail. 1200 // below would hang rather than fail.
876 _ = c.ngtcp2_conn_extend_max_stream_offset(cn, stream_id, datalen); 1201 if (s.extend) {
877 _ = c.ngtcp2_conn_extend_max_offset(cn, datalen); 1202 _ = c.ngtcp2_conn_extend_max_stream_offset(cn, stream_id, datalen);
1203 _ = c.ngtcp2_conn_extend_max_offset(cn, datalen);
1204 }
1205 if (s.verify) |exp| {
1206 const off = s.echoed;
1207 if (off + datalen > exp.len or
1208 !std.mem.eql(u8, exp[off..][0..datalen], data[0..datalen]))
1209 {
1210 s.mismatch = true;
1211 }
1212 }
878 s.echoed += datalen; 1213 s.echoed += datalen;
1214 if (s.recv_len + datalen <= s.recv_buf.len) {
1215 @memcpy(s.recv_buf[s.recv_len..][0..datalen], data[0..datalen]);
1216 s.recv_len += datalen;
1217 }
879 return 0; 1218 return 0;
880 } 1219 }
881 1220
882 fn init(server_addr: std.net.Address, key: Key) !TestClient { 1221 pub fn init(server_addr: std.net.Address, key: Key) !TestClient {
883 g_client_key = key; 1222 g_client_key = key;
884 const fd = try std.posix.socket( 1223 const fd = try std.posix.socket(
885 server_addr.any.family, 1224 server_addr.any.family,
@@ -903,7 +1242,7 @@ const TestClient = struct {
903 return self; 1242 return self;
904 } 1243 }
905 1244
906 fn start(self: *TestClient) !void { 1245 pub fn start(self: *TestClient) !void {
907 if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit; 1246 if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit;
908 const ctx = c.wolfSSL_CTX_new(c.wolfTLSv1_3_client_method()) orelse return error.TlsInit; 1247 const ctx = c.wolfSSL_CTX_new(c.wolfTLSv1_3_client_method()) orelse return error.TlsInit;
909 self.ssl_ctx = ctx; 1248 self.ssl_ctx = ctx;
@@ -976,7 +1315,7 @@ const TestClient = struct {
976 c.ngtcp2_conn_set_tls_native_handle(conn, ssl); 1315 c.ngtcp2_conn_set_tls_native_handle(conn, ssl);
977 } 1316 }
978 1317
979 fn drain(self: *TestClient) void { 1318 pub fn drain(self: *TestClient) void {
980 const conn = self.conn orelse return; 1319 const conn = self.conn orelse return;
981 var buf: [max_udp]u8 = undefined; 1320 var buf: [max_udp]u8 = undefined;
982 while (true) { 1321 while (true) {
@@ -1013,7 +1352,7 @@ const TestClient = struct {
1013 } 1352 }
1014 } 1353 }
1015 1354
1016 fn readable(self: *TestClient) void { 1355 pub fn readable(self: *TestClient) void {
1017 var buf: [65536]u8 = undefined; 1356 var buf: [65536]u8 = undefined;
1018 while (true) { 1357 while (true) {
1019 const n = std.posix.recv(self.fd, &buf, 0) catch return; 1358 const n = std.posix.recv(self.fd, &buf, 0) catch return;
@@ -1029,7 +1368,7 @@ const TestClient = struct {
1029 } 1368 }
1030 } 1369 }
1031 1370
1032 fn deinit(self: *TestClient) void { 1371 pub fn deinit(self: *TestClient) void {
1033 if (self.conn) |cn| c.ngtcp2_conn_del(cn); 1372 if (self.conn) |cn| c.ngtcp2_conn_del(cn);
1034 if (self.ssl) |s| c.wolfSSL_free(s); 1373 if (self.ssl) |s| c.wolfSSL_free(s);
1035 if (self.ssl_ctx) |x| c.wolfSSL_CTX_free(x); 1374 if (self.ssl_ctx) |x| c.wolfSSL_CTX_free(x);
@@ -1041,23 +1380,47 @@ const TestClient = struct {
1041 /// which is enough to prove bytes cross the seam in both directions. 1380 /// which is enough to prove bytes cross the seam in both directions.
1042 const EchoOwner = struct { 1381 const EchoOwner = struct {
1043 listener: *Listener = undefined, 1382 listener: *Listener = undefined,
1383 alloc: std.mem.Allocator = std.testing.allocator,
1384 id: u64 = 0,
1044 opened: usize = 0, 1385 opened: usize = 0,
1045 closed: usize = 0, 1386 closed: usize = 0,
1046 received: usize = 0, 1387 received: usize = 0,
1388 /// What the ring would not take yet. The daemon keeps exactly this
1389 /// queue for exactly this reason: `send` takes what fits and the owner
1390 /// holds — and bounds — the rest.
1391 backlog: std.ArrayList(u8) = .empty,
1047 1392
1048 fn onOpen(ctx: *anyopaque, _: u64) void { 1393 fn deinit(self: *EchoOwner) void {
1394 self.backlog.deinit(self.alloc);
1395 }
1396
1397 fn onOpen(ctx: *anyopaque, id: u64) void {
1049 const self: *EchoOwner = @ptrCast(@alignCast(ctx)); 1398 const self: *EchoOwner = @ptrCast(@alignCast(ctx));
1050 self.opened += 1; 1399 self.opened += 1;
1400 self.id = id;
1051 } 1401 }
1052 fn onData(ctx: *anyopaque, id: u64, bytes: []const u8) void { 1402 fn onData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
1053 const self: *EchoOwner = @ptrCast(@alignCast(ctx)); 1403 const self: *EchoOwner = @ptrCast(@alignCast(ctx));
1054 self.received += bytes.len; 1404 self.received += bytes.len;
1055 self.listener.send(id, bytes) catch {}; 1405 self.id = id;
1406 self.backlog.appendSlice(self.alloc, bytes) catch return;
1407 self.flush();
1056 } 1408 }
1057 fn onClose(ctx: *anyopaque, _: u64) void { 1409 fn onClose(ctx: *anyopaque, _: u64) void {
1058 const self: *EchoOwner = @ptrCast(@alignCast(ctx)); 1410 const self: *EchoOwner = @ptrCast(@alignCast(ctx));
1059 self.closed += 1; 1411 self.closed += 1;
1060 } 1412 }
1413 /// Offer the backlog again. Called on every pump iteration as well as on
1414 /// arrival, because the room to accept it comes from acks, which arrive
1415 /// on their own schedule.
1416 fn flush(self: *EchoOwner) void {
1417 if (self.backlog.items.len == 0) return;
1418 const n = self.listener.send(self.id, self.backlog.items) catch return;
1419 if (n == 0) return;
1420 const rest = self.backlog.items.len - n;
1421 std.mem.copyForwards(u8, self.backlog.items[0..rest], self.backlog.items[n..]);
1422 self.backlog.shrinkRetainingCapacity(rest);
1423 }
1061 fn handler(self: *EchoOwner) Handler { 1424 fn handler(self: *EchoOwner) Handler {
1062 return .{ .ctx = self, .onOpen = onOpen, .onData = onData, .onClose = onClose }; 1425 return .{ .ctx = self, .onOpen = onOpen, .onData = onData, .onClose = onClose };
1063 } 1426 }
@@ -1082,6 +1445,9 @@ fn pump(l: *Listener, cl: *TestClient, ms: u64, done: *const fn (*EchoOwner, *Te
1082 waited += 10; 1445 waited += 10;
1083 } 1446 }
1084 l.tick(); 1447 l.tick();
1448 // The owner re-offers its backlog every iteration: acks free ring
1449 // space on their own schedule, and nothing else would notice.
1450 owner.flush();
1085 cl.drain(); 1451 cl.drain();
1086 for (l.conns) |slot| { 1452 for (l.conns) |slot| {
1087 if (slot) |cn| l.drain(cn); 1453 if (slot) |cn| l.drain(cn);
@@ -1090,9 +1456,14 @@ fn pump(l: *Listener, cl: *TestClient, ms: u64, done: *const fn (*EchoOwner, *Te
1090 return done(owner, cl); 1456 return done(owner, cl);
1091 } 1457 }
1092 1458
1093 fn loopbackListener(alloc: std.mem.Allocator, key: Key, owner: *EchoOwner) !struct { l: *Listener, addr: std.net.Address } { 1459 fn loopbackListener(
1460 alloc: std.mem.Allocator,
1461 key: Key,
1462 owner: *EchoOwner,
1463 idle_ms: u64,
1464 ) !struct { l: *Listener, addr: std.net.Address } {
1094 const bind = try std.net.Address.parseIp("127.0.0.1", 0); 1465 const bind = try std.net.Address.parseIp("127.0.0.1", 0);
1095 const l = try Listener.init(alloc, bind, key, owner.handler(), 5000); 1466 const l = try Listener.init(alloc, bind, key, owner.handler(), idle_ms);
1096 owner.listener = l; 1467 owner.listener = l;
1097 var actual: std.posix.sockaddr.storage = undefined; 1468 var actual: std.posix.sockaddr.storage = undefined;
1098 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual)); 1469 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
@@ -1105,7 +1476,8 @@ test "Listener: PSK handshake, Retry, and a payload larger than the initial wind
1105 const key: Key = .{ .bytes = [_]u8{0x5A} ** key_len }; 1476 const key: Key = .{ .bytes = [_]u8{0x5A} ** key_len };
1106 1477
1107 var owner: EchoOwner = .{}; 1478 var owner: EchoOwner = .{};
1108 const setup = try loopbackListener(alloc, key, &owner); 1479 defer owner.deinit();
1480 const setup = try loopbackListener(alloc, key, &owner, 5000);
1109 defer setup.l.deinit(); 1481 defer setup.l.deinit();
1110 1482
1111 var cl = try TestClient.init(setup.addr, key); 1483 var cl = try TestClient.init(setup.addr, key);
@@ -1170,7 +1542,8 @@ test "Listener: a client holding the wrong key never completes a handshake" {
1170 const wrong_key: Key = .{ .bytes = [_]u8{0x22} ** key_len }; 1542 const wrong_key: Key = .{ .bytes = [_]u8{0x22} ** key_len };
1171 1543
1172 var owner: EchoOwner = .{}; 1544 var owner: EchoOwner = .{};
1173 const setup = try loopbackListener(alloc, server_key, &owner); 1545 defer owner.deinit();
1546 const setup = try loopbackListener(alloc, server_key, &owner, 5000);
1174 defer setup.l.deinit(); 1547 defer setup.l.deinit();
1175 1548
1176 var cl = try TestClient.init(setup.addr, wrong_key); 1549 var cl = try TestClient.init(setup.addr, wrong_key);
@@ -1190,3 +1563,431 @@ test "Listener: a client holding the wrong key never completes a handshake" {
1190 try std.testing.expectEqual(@as(usize, 0), owner.opened); 1563 try std.testing.expectEqual(@as(usize, 0), owner.opened);
1191 try std.testing.expectEqual(@as(usize, 0), owner.received); 1564 try std.testing.expectEqual(@as(usize, 0), owner.received);
1192 } 1565 }
1566
1567 test "Listener: keepalive carries an idle connection past its idle timeout" {
1568 const alloc = std.testing.allocator;
1569 const key: Key = .{ .bytes = [_]u8{0x3C} ** key_len };
1570
1571 // Short enough that the test is quick, long enough that a handshake
1572 // fits inside it: the keepalive lands at 300ms, the timeout at 900ms.
1573 const idle_ms = 900;
1574 var owner: EchoOwner = .{};
1575 defer owner.deinit();
1576 const setup = try loopbackListener(alloc, key, &owner, idle_ms);
1577 defer setup.l.deinit();
1578
1579 var cl = try TestClient.init(setup.addr, key);
1580 defer cl.deinit();
1581 try cl.start();
1582 cl.drain();
1583 try std.testing.expect(pump(setup.l, &cl, 5000, struct {
1584 fn f(o: *EchoOwner, t: *TestClient) bool {
1585 return t.handshake_done and o.opened > 0;
1586 }
1587 }.f, &owner));
1588
1589 // Now say nothing at all for three times the idle timeout. `pump` only
1590 // counts quiet polls toward its budget, so the wall-clock silence is at
1591 // least this long — the keepalive PINGs it does carry are the whole
1592 // point, and they are not application traffic.
1593 _ = pump(setup.l, &cl, idle_ms * 3, struct {
1594 fn f(_: *EchoOwner, _: *TestClient) bool {
1595 return false;
1596 }
1597 }.f, &owner);
1598
1599 // Nothing timed out...
1600 try std.testing.expectEqual(@as(usize, 0), owner.closed);
1601 // ...and the connection is not merely un-reaped but still usable, which
1602 // a live conn struct on its own would not prove.
1603 cl.out = "still-here";
1604 cl.out_sent = 0;
1605 try std.testing.expect(pump(setup.l, &cl, 5000, struct {
1606 fn f(_: *EchoOwner, t: *TestClient) bool {
1607 return t.echoed >= "still-here".len;
1608 }
1609 }.f, &owner));
1610 try std.testing.expectEqual(@as(usize, 0), owner.closed);
1611 }
1612
1613 test "Egress: a byte does not move until it is acked, and the ring wraps" {
1614 const alloc = std.testing.allocator;
1615 var e: Egress = .{ .buf = try alloc.alloc(u8, 8) };
1616 defer e.deinit(alloc);
1617
1618 try std.testing.expectEqual(@as(usize, 5), e.push("hello"));
1619 var v: [2]c.ngtcp2_vec = undefined;
1620 try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
1621 try std.testing.expectEqual(@as(usize, 5), v[0].len);
1622 const base = e.buf.ptr;
1623 try std.testing.expectEqual(base, v[0].base);
1624
1625 // ngtcp2 takes three. Those three now have a pointer pointing at them
1626 // and must not move; the two behind them are still ours to offer.
1627 e.took(3);
1628 try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
1629 try std.testing.expectEqual(@as(usize, 2), v[0].len);
1630 try std.testing.expectEqual(base + 3, v[0].base);
1631
1632 // Room is what is left after everything HELD, sent or not — the three
1633 // in ngtcp2's hands are not free space just because they left the box.
1634 try std.testing.expectEqual(@as(usize, 3), e.freeSpace());
1635 try std.testing.expectEqual(@as(usize, 3), e.push("world"));
1636 try std.testing.expectEqual(@as(usize, 0), e.push("x"));
1637
1638 // An acknowledgement is the only thing that frees anything.
1639 e.ack(3);
1640 try std.testing.expectEqual(@as(usize, 3), e.freeSpace());
1641
1642 // ...and the write that follows wraps rather than shifting a byte.
1643 try std.testing.expectEqual(@as(usize, 3), e.push("abc"));
1644 try std.testing.expectEqual(@as(usize, 2), e.vecs(&v));
1645 try std.testing.expectEqualStrings("lowor", v[0].base[0..v[0].len]);
1646 try std.testing.expectEqualStrings("abc", v[1].base[0..v[1].len]);
1647 }
1648
1649 test "Egress: an ack can never free more than is outstanding" {
1650 const alloc = std.testing.allocator;
1651 var e: Egress = .{ .buf = try alloc.alloc(u8, 8) };
1652 defer e.deinit(alloc);
1653
1654 _ = e.push("abcd");
1655 e.took(2);
1656 // Two are in flight and two are still unsent. A callback claiming more
1657 // than is outstanding must not consume the unsent ones — they have
1658 // never been on the wire and cannot have been acknowledged.
1659 e.ack(99);
1660 try std.testing.expectEqual(@as(usize, 2), e.held);
1661 try std.testing.expectEqual(@as(usize, 2), e.unsent);
1662 var v: [2]c.ngtcp2_vec = undefined;
1663 try std.testing.expectEqual(@as(usize, 1), e.vecs(&v));
1664 try std.testing.expectEqualStrings("cd", v[0].base[0..v[0].len]);
1665 }
1666
1667 /// Get the peer's stream on the record. The server learns a stream id only
1668 /// when data arrives on it — the client opening one locally tells the server
1669 /// nothing — so a test that wants the server to SEND first has to make the
1670 /// client say something first. Waits until the round trip is acknowledged,
1671 /// so the egress ring is empty again and the next assertion is about the
1672 /// test's own bytes.
1673 fn openStream(l: *Listener, cl: *TestClient, owner: *EchoOwner) !void {
1674 cl.out = "hi";
1675 cl.out_sent = 0;
1676 cl.drain();
1677 const ok = pumpUntil(l, cl, owner, 10_000, struct {
1678 fn f(o: *EchoOwner, t: *TestClient) bool {
1679 return o.received >= 2 and t.echoed >= 2 and l_pending(o) == 0;
1680 }
1681 fn l_pending(o: *EchoOwner) usize {
1682 return o.listener.pendingBytes(o.id);
1683 }
1684 }.f);
1685 if (!ok) return error.StreamNeverOpened;
1686 cl.out = &.{};
1687 cl.out_sent = 0;
1688 cl.echoed = 0;
1689 }
1690
1691 /// Drive both ends for a wall-clock budget, unconditionally. `pump` counts
1692 /// only quiet iterations toward its deadline, which is right for waiting on
1693 /// an event and wrong for a bulk transfer that is busy the whole time and
1694 /// must still be prevented from hanging a suite.
1695 fn pumpUntil(
1696 l: *Listener,
1697 cl: *TestClient,
1698 owner: *EchoOwner,
1699 wall_ms: i64,
1700 done: *const fn (*EchoOwner, *TestClient) bool,
1701 ) bool {
1702 const deadline = std.time.milliTimestamp() + wall_ms;
1703 while (std.time.milliTimestamp() < deadline) {
1704 if (done(owner, cl)) return true;
1705 var fds = [_]std.posix.pollfd{
1706 .{ .fd = l.fd, .events = std.posix.POLL.IN, .revents = 0 },
1707 .{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
1708 };
1709 const ready = std.posix.poll(&fds, 5) catch return false;
1710 if (ready > 0) {
1711 if (fds[0].revents != 0) l.readable();
1712 if (fds[1].revents != 0) cl.readable();
1713 }
1714 l.tick();
1715 owner.flush();
1716 cl.drain();
1717 for (l.conns) |slot| {
1718 if (slot) |cn| l.drain(cn);
1719 }
1720 }
1721 return done(owner, cl);
1722 }
1723
1724 test "Listener: bytes survive retransmission, which is what the buffer is for" {
1725 const alloc = std.testing.allocator;
1726 const key: Key = .{ .bytes = [_]u8{0x6E} ** key_len };
1727
1728 var owner: EchoOwner = .{};
1729 defer owner.deinit();
1730 const setup = try loopbackListener(alloc, key, &owner, 10_000);
1731 defer setup.l.deinit();
1732
1733 var cl = try TestClient.init(setup.addr, key);
1734 defer cl.deinit();
1735
1736 // A deliberately tiny receive buffer, which is how loopback is made to
1737 // lose packets on purpose. Loss is the ONLY way into ngtcp2's
1738 // retransmission path, and that path re-reads the original bytes
1739 // through a pointer the listener handed over packets ago — so it is
1740 // also the only way to catch a buffer that recycled them underneath it.
1741 // Without the loss this test is just another echo; the 3MB test above
1742 // reached the same path by accident, about one run in twenty, and
1743 // segfaulted when it did.
1744 try std.posix.setsockopt(
1745 cl.fd,
1746 std.posix.SOL.SOCKET,
1747 std.posix.SO.RCVBUF,
1748 &std.mem.toBytes(@as(c_int, 64 * 1024)),
1749 );
1750
1751 try cl.start();
1752 cl.drain();
1753 try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1754 fn f(o: *EchoOwner, t: *TestClient) bool {
1755 return t.handshake_done and o.opened > 0;
1756 }
1757 }.f));
1758
1759 // Twice the egress ring, so the ring wraps and every byte's release
1760 // depends on an ack that may itself have been for a retransmission.
1761 const size = 2 * egress_cap;
1762 const payload = try alloc.alloc(u8, size);
1763 defer alloc.free(payload);
1764 for (payload, 0..) |*b, i| b.* = @truncate(i *% 31 +% 7);
1765
1766 cl.verify = payload;
1767 cl.out = payload;
1768 cl.out_sent = 0;
1769
1770 const ok = pumpUntil(setup.l, &cl, &owner, 60_000, struct {
1771 fn f(o: *EchoOwner, t: *TestClient) bool {
1772 _ = o;
1773 return t.echoed >= t.verify.?.len;
1774 }
1775 }.f);
1776 try std.testing.expect(ok);
1777
1778 // Count AND content. A transport that lost a retransmission and carried
1779 // on would land the right number of bytes in the wrong order, which no
1780 // length assertion can see.
1781 try std.testing.expectEqual(size, cl.echoed);
1782 try std.testing.expect(!cl.mismatch);
1783 try std.testing.expectEqual(size, owner.received);
1784 try std.testing.expectEqual(@as(usize, 0), owner.closed);
1785 }
1786
1787 test "Listener.send: takes what fits, refuses when full, and recovers on acks" {
1788 const alloc = std.testing.allocator;
1789 const key: Key = .{ .bytes = [_]u8{0x21} ** key_len };
1790
1791 var owner: EchoOwner = .{};
1792 defer owner.deinit();
1793 const setup = try loopbackListener(alloc, key, &owner, 10_000);
1794 defer setup.l.deinit();
1795
1796 var cl = try TestClient.init(setup.addr, key);
1797 defer cl.deinit();
1798 try cl.start();
1799 cl.drain();
1800 try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1801 fn f(o: *EchoOwner, t: *TestClient) bool {
1802 return t.handshake_done and o.opened > 0;
1803 }
1804 }.f));
1805
1806 try openStream(setup.l, &cl, &owner);
1807
1808 const big = try alloc.alloc(u8, egress_cap + 4096);
1809 defer alloc.free(big);
1810 @memset(big, 0x5A);
1811
1812 // Nobody is pumping the client from here, so nothing is acknowledged
1813 // and nothing is released. The first send fills the ring exactly and
1814 // reports the short count; the daemon keeps the remainder, which is the
1815 // whole point — an accept-everything send moved that backlog down here
1816 // where pending_cap could never see it.
1817 const first = try setup.l.send(owner.id, big);
1818 try std.testing.expectEqual(egress_cap, first);
1819 try std.testing.expectEqual(@as(usize, 0), try setup.l.send(owner.id, "x"));
1820 try std.testing.expectEqual(egress_cap, setup.l.pendingBytes(owner.id));
1821
1822 // Now let the client read and acknowledge: space comes back, and it
1823 // comes back from acks rather than from having handed bytes to ngtcp2.
1824 // Waiting on the ACK, not on the arrival: the last byte reaching the
1825 // client is not the same event as the ring being free to reuse it, and
1826 // conflating them is what this buffer exists to stop anyone doing.
1827 try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 20_000, struct {
1828 fn f(o: *EchoOwner, _: *TestClient) bool {
1829 return o.listener.pendingBytes(o.id) == 0;
1830 }
1831 }.f));
1832 try std.testing.expectEqual(egress_cap, cl.echoed);
1833 try std.testing.expect(try setup.l.send(owner.id, "room again") > 0);
1834
1835 // An id nobody owns is an error, not a silent success.
1836 try std.testing.expectError(error.NoSuchConn, setup.l.send(owner.id + 999, "x"));
1837 }
1838
1839 test "Listener: a full peer window blocks the stream without stopping the ACKs" {
1840 const alloc = std.testing.allocator;
1841 const key: Key = .{ .bytes = [_]u8{0x4B} ** key_len };
1842
1843 // Short enough that a connection nobody is servicing dies inside the
1844 // test, which is exactly the failure being guarded against: treating
1845 // the blocked-stream return as fatal abandons the egress loop, and the
1846 // ACKs and keepalives that share it never leave either.
1847 const idle_ms = 1500;
1848 var owner: EchoOwner = .{};
1849 defer owner.deinit();
1850 const setup = try loopbackListener(alloc, key, &owner, idle_ms);
1851 defer setup.l.deinit();
1852
1853 var cl = try TestClient.init(setup.addr, key);
1854 defer cl.deinit();
1855 try cl.start();
1856 cl.drain();
1857 try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1858 fn f(o: *EchoOwner, t: *TestClient) bool {
1859 return t.handshake_done and o.opened > 0;
1860 }
1861 }.f));
1862
1863 try openStream(setup.l, &cl, &owner);
1864
1865 // The client consumes but never grants more window, so the peer's
1866 // stream credit runs out and stays out.
1867 cl.extend = false;
1868
1869 const big = try alloc.alloc(u8, 4 * egress_cap);
1870 defer alloc.free(big);
1871 @memset(big, 0x33);
1872 try owner.backlog.appendSlice(alloc, big);
1873 owner.flush();
1874
1875 // Well past the idle timeout, with the stream blocked the entire time.
1876 _ = pumpUntil(setup.l, &cl, &owner, idle_ms * 3, struct {
1877 fn f(_: *EchoOwner, _: *TestClient) bool {
1878 return false;
1879 }
1880 }.f);
1881
1882 // The stream really did block — the client cannot have taken it all.
1883 try std.testing.expect(cl.echoed < big.len);
1884 // ...and the connection is alive, which it can only be if packets that
1885 // are not stream data kept flowing while it was blocked.
1886 try std.testing.expectEqual(@as(usize, 0), owner.closed);
1887 try std.testing.expect(setup.l.find(owner.id) != null);
1888 }
1889
1890 test "Listener: a packet addressed to any advertised CID reaches its connection" {
1891 const alloc = std.testing.allocator;
1892 const key: Key = .{ .bytes = [_]u8{0x9C} ** key_len };
1893
1894 var owner: EchoOwner = .{};
1895 defer owner.deinit();
1896 const setup = try loopbackListener(alloc, key, &owner, 10_000);
1897 defer setup.l.deinit();
1898
1899 var cl = try TestClient.init(setup.addr, key);
1900 defer cl.deinit();
1901 try cl.start();
1902 cl.drain();
1903 try std.testing.expect(pumpUntil(setup.l, &cl, &owner, 10_000, struct {
1904 fn f(o: *EchoOwner, t: *TestClient) bool {
1905 return t.handshake_done and o.opened > 0;
1906 }
1907 }.f));
1908
1909 const cn = setup.l.find(owner.id).?;
1910 cn.refreshCids();
1911 // Non-vacuity first: with only one CID there is no second one to route
1912 // by, and everything below would pass for the wrong reason.
1913 try std.testing.expect(cn.ncids > 1);
1914
1915 var secondary: ?*c.ngtcp2_cid = null;
1916 for (cn.cids[0..cn.ncids]) |*cid| {
1917 if (!cidEql(cid, cn.scid.data[0..cn.scid.datalen])) {
1918 secondary = cid;
1919 break;
1920 }
1921 }
1922 try std.testing.expect(secondary != null);
1923 const sec = secondary.?;
1924 try std.testing.expect(cn.matches(sec.data[0..sec.datalen]));
1925
1926 // A CID nobody advertised belongs to nobody.
1927 const stranger_cid = [_]u8{0xEE} ** 8;
1928 try std.testing.expect(!cn.matches(&stranger_cid));
1929
1930 // Now the routing itself. The observable is what happens to a packet
1931 // that MISSES: it falls through to `accept`, which answers a token-less
1932 // Initial with a Retry. So a probe socket that receives nothing is proof
1933 // the packet was delivered to the connection instead — and the same
1934 // probe receiving a Retry for an unknown CID proves the packet was
1935 // well-formed enough for `accept` to have answered it.
1936 const probe = try std.posix.socket(
1937 std.posix.AF.INET,
1938 std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
1939 0,
1940 );
1941 defer std.posix.close(probe);
1942 const probe_bind = try std.net.Address.parseIp("127.0.0.1", 0);
1943 try std.posix.bind(probe, &probe_bind.any, probe_bind.getOsSockLen());
1944 var probe_addr: std.posix.sockaddr.storage = undefined;
1945 var probe_len: std.posix.socklen_t = @sizeOf(@TypeOf(probe_addr));
1946 try std.posix.getsockname(probe, @ptrCast(&probe_addr), &probe_len);
1947
1948 var pkt: [1300]u8 = undefined;
1949 try std.testing.expect(!probeGotAnything(probe));
1950
1951 // Addressed to a CID this connection advertised: delivered, not
1952 // answered.
1953 buildInitial(&pkt, sec.data[0..sec.datalen]);
1954 setup.l.route(&pkt, &probe_addr, probe_len);
1955 try std.testing.expect(!probeGotAnything(probe));
1956
1957 // Addressed to nobody: `accept` answers with a Retry, which is what
1958 // every migrated packet used to get.
1959 buildInitial(&pkt, &stranger_cid);
1960 setup.l.route(&pkt, &probe_addr, probe_len);
1961 try std.testing.expect(probeGotAnything(probe));
1962 }
1963
1964 /// A syntactically valid, cryptographically meaningless Initial packet
1965 /// addressed to `dcid`. Enough for `ngtcp2_accept` to recognise it and
1966 /// answer, which is all this needs to tell routing hit from routing miss.
1967 fn buildInitial(pkt: *[1300]u8, dcid: []const u8) void {
1968 @memset(pkt, 0);
1969 var i: usize = 0;
1970 pkt[i] = 0xC3; // long header, fixed bit, Initial, 4-byte packet number
1971 i += 1;
1972 std.mem.writeInt(u32, pkt[i..][0..4], 1, .big); // version 1
1973 i += 4;
1974 pkt[i] = @intCast(dcid.len);
1975 i += 1;
1976 @memcpy(pkt[i..][0..dcid.len], dcid);
1977 i += dcid.len;
1978 pkt[i] = 8; // source CID length
1979 i += 1;
1980 @memset(pkt[i..][0..8], 0x5A);
1981 i += 8;
1982 pkt[i] = 0; // token length, varint 0
1983 i += 1;
1984 // Length: a two-byte varint covering everything left.
1985 const rest: u16 = @intCast(pkt.len - i - 2);
1986 std.mem.writeInt(u16, pkt[i..][0..2], rest | 0x4000, .big);
1987 }
1988
1989 fn probeGotAnything(fd: std.posix.fd_t) bool {
1990 var buf: [2048]u8 = undefined;
1991 const n = std.posix.recv(fd, &buf, 0) catch return false;
1992 return n > 0;
1993 }
src/server.zig
Old New
@@ -9,6 +9,7 @@ const std = @import("std");
9 const Engine = @import("engine").Engine; 9 const Engine = @import("engine").Engine;
10 const Pty = @import("pty").Pty; 10 const Pty = @import("pty").Pty;
11 const proto = @import("protocol"); 11 const proto = @import("protocol");
12 const quic = @import("quic");
12 13
13 const max_clients = 8; 14 const max_clients = 8;
14 const max_observers = 4; 15 const max_observers = 4;
@@ -212,6 +213,9 @@ pub fn installSignalHandlers() void {
212 /// need to change when it does. 213 /// need to change when it does.
213 const Sink = union(enum) { 214 const Sink = union(enum) {
214 socket: std.posix.fd_t, 215 socket: std.posix.fd_t,
216 /// A QUIC peer: the listener that owns the shared UDP socket, plus the
217 /// connection id within it. Note what is NOT here — a descriptor.
218 quic: struct { listener: *quic.Listener, id: u64 },
215 219
216 /// The descriptor to poll for this client, or -1 for "nothing of its 220 /// The descriptor to poll for this client, or -1 for "nothing of its
217 /// own" — poll(2) ignores negative fds, which is exactly the behaviour 221 /// own" — poll(2) ignores negative fds, which is exactly the behaviour
@@ -220,12 +224,20 @@ const Sink = union(enum) {
220 fn pollFd(self: Sink) std.posix.fd_t { 224 fn pollFd(self: Sink) std.posix.fd_t {
221 return switch (self) { 225 return switch (self) {
222 .socket => |fd| fd, 226 .socket => |fd| fd,
227 .quic => -1,
223 }; 228 };
224 } 229 }
225 230
226 /// Hand bytes to the kernel without blocking. Same contract as the 231 /// Hand bytes onward without blocking, returning how many were taken.
227 /// send(2) it replaces: returns what was accepted, WouldBlock when the 232 ///
228 /// buffer is full, any other error means the peer is unusable. 233 /// For a socket that is send(2) straight to the kernel. For QUIC it is a
234 /// queue-and-drain into the connection's bounded egress ring, which
235 /// takes what it has room for and says so. Both therefore obey the same
236 /// contract — a short return means the rest stays in `pending` and the
237 /// cap gets to judge it — and that symmetry is load-bearing: while the
238 /// QUIC arm accepted everything, `pending_cap` could never trip for a
239 /// QUIC client, and the unbounded growth simply moved down into the
240 /// listener where nothing was watching for it.
229 fn send(self: Sink, bytes: []const u8) !usize { 241 fn send(self: Sink, bytes: []const u8) !usize {
230 return switch (self) { 242 return switch (self) {
231 .socket => |fd| std.posix.send( 243 .socket => |fd| std.posix.send(
@@ -233,12 +245,33 @@ const Sink = union(enum) {
233 bytes, 245 bytes,
234 std.posix.MSG.DONTWAIT | std.posix.MSG.NOSIGNAL, 246 std.posix.MSG.DONTWAIT | std.posix.MSG.NOSIGNAL,
235 ), 247 ),
248 .quic => |q| q.listener.send(q.id, bytes),
249 };
250 }
251
252 /// Bytes this sink has taken but not yet got off the box. Zero for a
253 /// socket: once the kernel has it, it is the kernel's problem. For QUIC
254 /// the send is not finished until the peer acknowledges, so the listener
255 /// is the only thing that knows.
256 fn inFlight(self: Sink) usize {
257 return switch (self) {
258 .socket => 0,
259 .quic => |q| q.listener.pendingBytes(q.id),
236 }; 260 };
237 } 261 }
238 262
263 /// Close THIS CLIENT'S channel — never the transport it shares.
264 ///
265 /// For a socket those are the same object, which is exactly why the
266 /// distinction has to be written down before QUIC exists: a QUIC client
267 /// shares one UDP socket with every other session on this daemon, so
268 /// closing that socket because one client left would take down every
269 /// other client with it. `closeConn` tears down the one connection and
270 /// leaves the listener's socket alone. Pinned by a test.
239 fn close(self: Sink) void { 271 fn close(self: Sink) void {
240 switch (self) { 272 switch (self) {
241 .socket => |fd| std.posix.close(fd), 273 .socket => |fd| std.posix.close(fd),
274 .quic => |q| q.listener.closeConn(q.id),
242 } 275 }
243 } 276 }
244 }; 277 };
@@ -294,6 +327,10 @@ pub const Server = struct {
294 /// have_seq must be snapshotted, not deltaed. Never 0 — that value is 327 /// have_seq must be snapshotted, not deltaed. Never 0 — that value is
295 /// reserved for a client saying "I hold nothing". 328 /// reserved for a client saying "I hold nothing".
296 epoch: u64, 329 epoch: u64,
330 /// The QUIC listener, when `--quic` was given. Optional by design:
331 /// QUIC is opt-in per invocation and the unix socket is unaffected by
332 /// its presence or absence.
333 quic_listener: ?*quic.Listener = null,
297 /// Row-level change tracking behind the delta stream. 334 /// Row-level change tracking behind the delta stream.
298 tracker: DeltaTracker = .{}, 335 tracker: DeltaTracker = .{},
299 stats: Stats = .{}, 336 stats: Stats = .{},
@@ -453,7 +490,7 @@ pub const Server = struct {
453 } 490 }
454 491
455 const obs_base = 2 + max_clients; 492 const obs_base = 2 + max_clients;
456 var fds: [2 + max_clients + max_observers]std.posix.pollfd = undefined; 493 var fds: [2 + max_clients + max_observers + 1]std.posix.pollfd = undefined;
457 fds[0] = .{ .fd = self.pty.master, .events = std.posix.POLL.IN, .revents = 0 }; 494 fds[0] = .{ .fd = self.pty.master, .events = std.posix.POLL.IN, .revents = 0 };
458 fds[1] = .{ .fd = self.listener.stream.handle, .events = std.posix.POLL.IN, .revents = 0 }; 495 fds[1] = .{ .fd = self.listener.stream.handle, .events = std.posix.POLL.IN, .revents = 0 };
459 for (&self.clients, 0..) |*slot, i| { 496 for (&self.clients, 0..) |*slot, i| {
@@ -471,7 +508,35 @@ pub const Server = struct {
471 for (self.observers, 0..) |slot, i| { 508 for (self.observers, 0..) |slot, i| {
472 fds[obs_base + i] = .{ .fd = slot orelse -1, .events = std.posix.POLL.IN, .revents = 0 }; 509 fds[obs_base + i] = .{ .fd = slot orelse -1, .events = std.posix.POLL.IN, .revents = 0 };
473 } 510 }
474 _ = try std.posix.poll(&fds, timeout_ms); 511 // One extra descriptor for every QUIC client there will ever be:
512 // they share it, which is the whole reason a client slot cannot be
513 // a descriptor.
514 const quic_idx = fds.len - 1;
515 fds[quic_idx] = .{
516 .fd = if (self.quic_listener) |q| q.pollFd() else -1,
517 .events = std.posix.POLL.IN,
518 .revents = 0,
519 };
520
521 // ngtcp2's timers, folded in: the listener's earliest deadline
522 // shortens this poll, so retransmits and idle timeouts happen on
523 // time without a timerfd or a second loop.
524 var wait_ms = timeout_ms;
525 if (self.quic_listener) |q| wait_ms = q.timeoutMs(timeout_ms);
526 _ = try std.posix.poll(&fds, wait_ms);
527
528 if (self.quic_listener) |q| {
529 if (fds[quic_idx].revents != 0) q.readable();
530 // Unconditional: expiry work is due whether or not a packet
531 // arrived, and this is the only place it can happen.
532 q.tick();
533 // Acks that just arrived are what frees room in a connection's
534 // egress ring, and a QUIC client has no descriptor of its own to
535 // go writable. Without this, a client whose ring filled would
536 // sit on its backlog until the next frame happened to be queued
537 // for it — progress by coincidence rather than by design.
538 self.flushQuicClients();
539 }
475 540
476 if (fds[0].revents != 0) { 541 if (fds[0].revents != 0) {
477 var buf: [64 * 1024]u8 = undefined; 542 var buf: [64 * 1024]u8 = undefined;
@@ -583,39 +648,79 @@ pub const Server = struct {
583 fn drainPending(self: *Server, budget_ms: i64) void { 648 fn drainPending(self: *Server, budget_ms: i64) void {
584 const deadline = std.time.milliTimestamp() + budget_ms; 649 const deadline = std.time.milliTimestamp() + budget_ms;
585 while (true) { 650 while (true) {
586 var fds: [max_clients]std.posix.pollfd = undefined; 651 // One extra slot for the QUIC listener: its readability is how
652 // acknowledgements arrive, and acknowledgements are the only
653 // thing that frees room in a connection's egress ring.
654 var fds: [max_clients + 1]std.posix.pollfd = undefined;
587 var slots: [max_clients]usize = undefined; 655 var slots: [max_clients]usize = undefined;
588 var n: usize = 0; 656 var n_sock: usize = 0;
589 var owed: usize = 0; 657 var owed: usize = 0;
658 var quic_owed: usize = 0;
590 for (&self.clients, 0..) |*slot, i| { 659 for (&self.clients, 0..) |*slot, i| {
591 if (slot.*) |*c| { 660 if (slot.* == null) continue;
592 if (c.pending.items.len == 0) continue; 661 if (slot.*.?.sink == .quic) {
593 owed += c.pending.items.len; 662 // A QUIC client has no descriptor of its own to wait on,
594 slots[n] = i; 663 // so its egress is driven directly. It owes in two
595 fds[n] = .{ .fd = c.sink.pollFd(), .events = std.posix.POLL.OUT, .revents = 0 }; 664 // places, and only counting the first is how this arm
596 n += 1; 665 // used to look like it worked: bytes still queued in the
666 // slot, AND bytes the ring took but the peer has not
667 // acknowledged. A send is not finished when it is
668 // accepted, it is finished when it is acked.
669 self.flushClient(i);
670 if (self.clients[i]) |*c| {
671 const still = c.pending.items.len + c.sink.inFlight();
672 owed += still;
673 quic_owed += still;
674 }
675 continue;
597 } 676 }
677 const c = &self.clients[i].?;
678 if (c.pending.items.len == 0) continue;
679 owed += c.pending.items.len;
680 slots[n_sock] = i;
681 fds[n_sock] = .{ .fd = c.sink.pollFd(), .events = std.posix.POLL.OUT, .revents = 0 };
682 n_sock += 1;
598 } 683 }
599 if (n == 0) return; // everything delivered 684 // Give QUIC egress a chance to leave the box before deciding
685 // there is nothing left to wait for.
686 if (self.quic_listener) |q| q.tick();
687 if (owed == 0) return;
600 688
601 const remaining = deadline - std.time.milliTimestamp(); 689 const remaining = deadline - std.time.milliTimestamp();
602 if (remaining <= 0) return; 690 if (remaining <= 0) return;
691
692 var n = n_sock;
693 var quic_idx: ?usize = null;
694 if (quic_owed > 0) {
695 if (self.quic_listener) |q| {
696 quic_idx = n;
697 fds[n] = .{ .fd = q.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
698 n += 1;
699 }
700 }
701 if (n == 0) return; // owed, but nothing left that could pay
702
603 const ready = std.posix.poll(fds[0..n], @intCast(remaining)) catch return; 703 const ready = std.posix.poll(fds[0..n], @intCast(remaining)) catch return;
604 if (ready == 0) return; // out of time with bytes still owed 704 if (ready == 0) return; // out of time with bytes still owed
605 705
606 for (fds[0..n], slots[0..n]) |pfd, i| { 706 if (quic_idx) |qi| {
707 if (fds[qi].revents != 0) {
708 if (self.quic_listener) |q| q.readable();
709 }
710 }
711 for (fds[0..n_sock], slots[0..n_sock]) |pfd, i| {
607 // Non-POLLOUT revents (the peer hung up) reach flushClient 712 // Non-POLLOUT revents (the peer hung up) reach flushClient
608 // too: its send fails and the client is dropped, which is 713 // too: its send fails and the client is dropped, which is
609 // also what stops this loop from retrying a dead fd. 714 // also what stops this loop from retrying a dead fd.
610 if (pfd.revents != 0) self.flushClient(i); 715 if (pfd.revents != 0) self.flushClient(i);
611 } 716 }
612 717
613 // Guard against a poll that keeps reporting writable while send 718 // Guard against a poll that keeps reporting ready while nothing
614 // keeps refusing: without this the loop would spin hot until the 719 // actually moves: without this the loop would spin hot until the
615 // deadline instead of giving up on a peer making no progress. 720 // deadline instead of giving up on a peer making no progress.
616 var still: usize = 0; 721 var still: usize = 0;
617 for (&self.clients) |*slot| { 722 for (&self.clients) |*slot| {
618 if (slot.*) |*c| still += c.pending.items.len; 723 if (slot.*) |*c| still += c.pending.items.len + c.sink.inFlight();
619 } 724 }
620 if (still >= owed) return; 725 if (still >= owed) return;
621 } 726 }
@@ -661,6 +766,96 @@ pub const Server = struct {
661 } 766 }
662 } 767 }
663 768
769 /// Offer every QUIC client's backlog to its ring again. Cheap when there
770 /// is nothing owed, which is the common case.
771 fn flushQuicClients(self: *Server) void {
772 for (&self.clients, 0..) |*slot, i| {
773 const cl = &(slot.* orelse continue);
774 if (cl.sink != .quic) continue;
775 if (cl.pending.items.len > 0) self.flushClient(i);
776 }
777 }
778
779 // ---- QUIC glue -------------------------------------------------------
780 //
781 // Three callbacks, and between them they contain the entire difference
782 // between a QUIC client and a socket client. Everything past pushInbound
783 // is code that has never heard of either.
784
785 fn quicOnOpen(ctx: *anyopaque, id: u64) void {
786 const self: *Server = @ptrCast(@alignCast(ctx));
787 const listener = self.quic_listener orelse return;
788 const slot = self.freeClientSlot() orelse {
789 // Session full: the same answer the socket path gives, sent the
790 // same way, then the connection goes.
791 // A short accept would be a truncated refusal, which reads as a
792 // corrupt frame rather than a "no". The ring is empty on a
793 // connection that has just opened and this frame is a handful of
794 // bytes, so a partial take here means something is very wrong;
795 // either way, closing is the answer.
796 _ = listener.send(id, &frameBytes(.exit_status, &.{1})) catch {};
797 listener.closeConn(id);
798 return;
799 };
800 self.clients[slot] = .{ .sink = .{ .quic = .{ .listener = listener, .id = id } } };
801 }
802
803 fn quicOnData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
804 const self: *Server = @ptrCast(@alignCast(ctx));
805 const i = self.slotForQuic(id) orelse return;
806 self.pushInbound(i, bytes);
807 }
808
809 fn quicOnClose(ctx: *anyopaque, id: u64) void {
810 const self: *Server = @ptrCast(@alignCast(ctx));
811 const i = self.slotForQuic(id) orelse return;
812 // The connection is already gone; drop the slot without asking the
813 // sink to close it again.
814 if (self.clients[i]) |*slot| {
815 slot.pending.deinit(self.alloc);
816 slot.inbound.deinit(self.alloc);
817 }
818 self.clients[i] = null;
819 }
820
821 fn slotForQuic(self: *Server, id: u64) ?usize {
822 for (self.clients, 0..) |slot, i| {
823 const cs = slot orelse continue;
824 switch (cs.sink) {
825 .quic => |q| if (q.id == id) return i,
826 else => {},
827 }
828 }
829 return null;
830 }
831
832 pub fn quicHandler(self: *Server) quic.Handler {
833 return .{
834 .ctx = self,
835 .onOpen = quicOnOpen,
836 .onData = quicOnData,
837 .onClose = quicOnClose,
838 };
839 }
840
841 /// Adopt a listener built by the caller (main.zig, or a test). The
842 /// server does not own the socket, only the reference — deinit leaves
843 /// the listener to its creator, which keeps the ownership story the
844 /// same as the unix listener's.
845 pub fn attachQuic(self: *Server, listener: *quic.Listener) void {
846 self.quic_listener = listener;
847 }
848
849 /// One frame, on the stack, for the two places the QUIC path has to
850 /// speak before a client slot exists.
851 fn frameBytes(t: proto.MsgType, payload: []const u8) [6]u8 {
852 var buf: [6]u8 = undefined;
853 buf[0] = @intFromEnum(t);
854 std.mem.writeInt(u32, buf[1..5], @intCast(payload.len), .little);
855 buf[5] = if (payload.len > 0) payload[0] else 0;
856 return buf;
857 }
858
664 fn hasClients(self: *const Server) bool { 859 fn hasClients(self: *const Server) bool {
665 for (self.clients) |slot| { 860 for (self.clients) |slot| {
666 if (slot != null) return true; 861 if (slot != null) return true;
@@ -686,6 +881,11 @@ pub const Server = struct {
686 fn serviceClient(self: *Server, i: usize) void { 881 fn serviceClient(self: *Server, i: usize) void {
687 const fd = switch (self.clients[i].?.sink) { 882 const fd = switch (self.clients[i].?.sink) {
688 .socket => |fd| fd, 883 .socket => |fd| fd,
884 // A QUIC client has no descriptor of its own to read: its bytes
885 // arrive from the shared UDP socket and enter through
886 // pushInbound. Poll never reports it readable (pollFd is -1), so
887 // reaching here would mean the poll bookkeeping had drifted.
888 .quic => return,
689 }; 889 };
690 const frame = proto.readFrame(self.alloc, fd) catch { 890 const frame = proto.readFrame(self.alloc, fd) catch {
691 self.dropClient(i); 891 self.dropClient(i);
@@ -3157,3 +3357,328 @@ test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-r
3157 else => return error.ExpectedDiscontinuity, 3357 else => return error.ExpectedDiscontinuity,
3158 } 3358 }
3159 } 3359 }
3360
3361 // ---------------------------------------------------------------------------
3362 // QUIC integration tests.
3363 //
3364 // These two exist because the claims they check — that a QUIC client still
3365 // gets the shell's exit code, and that one client leaving does not take the
3366 // others with it — were, until they were written, assertions of mine rather
3367 // than evidence.
3368 // ---------------------------------------------------------------------------
3369
3370 /// Bring a Server up with a QUIC listener bound to an ephemeral loopback
3371 /// port, and hand back the address a client should dial.
3372 fn quicTestServer(srv: *Server, key: quic.Key) !struct { l: *quic.Listener, addr: std.net.Address } {
3373 const bind = try std.net.Address.parseIp("127.0.0.1", 0);
3374 const l = try quic.Listener.init(srv.alloc, bind, key, srv.quicHandler(), 5000);
3375 srv.attachQuic(l);
3376 var actual: std.posix.sockaddr.storage = undefined;
3377 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
3378 try std.posix.getsockname(l.pollFd(), @ptrCast(&actual), &len);
3379 return .{ .l = l, .addr = std.net.Address.initPosix(@ptrCast(@alignCast(&actual))) };
3380 }
3381
3382 /// Drive the daemon and up to two QUIC clients until `done`, or the budget
3383 /// runs out. Single-threaded: the daemon's own pump is what services the
3384 /// listener, which is the integration under test.
3385 fn quicPump(
3386 srv: *Server,
3387 clients: []*quic.TestClient,
3388 budget_ms: u64,
3389 ctx: anytype,
3390 done: *const fn (@TypeOf(ctx)) bool,
3391 ) !?u8 {
3392 var waited: u64 = 0;
3393 var exit_code: ?u8 = null;
3394 while (waited < budget_ms) {
3395 if (done(ctx)) return exit_code;
3396 if (exit_code == null) {
3397 exit_code = try srv.pumpOnce(5);
3398 }
3399 for (clients) |cl| {
3400 cl.drain();
3401 var pfd = [_]std.posix.pollfd{.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
3402 if ((std.posix.poll(&pfd, 1) catch 0) > 0) cl.readable();
3403 }
3404 waited += 6;
3405 }
3406 return exit_code;
3407 }
3408
3409 /// Find a frame of `want` in a client's received bytes, returning its
3410 /// payload. The client is handed a byte stream, so this does the same
3411 /// framing walk a real client would.
3412 fn findFrame(bytes: []const u8, want: proto.MsgType) ?[]const u8 {
3413 var off: usize = 0;
3414 while (off + 5 <= bytes.len) {
3415 const len = std.mem.readInt(u32, bytes[off + 1 ..][0..4], .little);
3416 if (off + 5 + len > bytes.len) return null;
3417 if (@as(proto.MsgType, @enumFromInt(bytes[off])) == want) {
3418 return bytes[off + 5 ..][0..len];
3419 }
3420 off += 5 + len;
3421 }
3422 return null;
3423 }
3424
3425 fn attachOver(cl: *quic.TestClient, buf: *std.ArrayList(u8), alloc: std.mem.Allocator) !void {
3426 try proto.appendFrame(buf, alloc, .attach, &proto.encodeAttach(80, 24, 0, 0));
3427 cl.out = buf.items;
3428 cl.drain();
3429 }
3430
3431 test "Server: a QUIC client still receives the shell's exit status" {
3432 const alloc = std.testing.allocator;
3433 var tmp = std.testing.tmpDir(.{});
3434 defer tmp.cleanup();
3435 var path_buf: [256]u8 = undefined;
3436 const dir_path = try tmp.dir.realpath(".", &path_buf);
3437 const sock_path = try std.fmt.allocPrint(alloc, "{s}/qexit.sock", .{dir_path});
3438 defer alloc.free(sock_path);
3439
3440 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
3441 defer srv.deinit();
3442 const key: quic.Key = .{ .bytes = [_]u8{0x31} ** quic.key_len };
3443 const q = try quicTestServer(&srv, key);
3444 defer q.l.deinit();
3445
3446 var cl = try quic.TestClient.init(q.addr, key);
3447 defer cl.deinit();
3448 try cl.start();
3449 cl.drain();
3450
3451 var out: std.ArrayList(u8) = .empty;
3452 defer out.deinit(alloc);
3453 try attachOver(&cl, &out, alloc);
3454
3455 // Attached and carrying state: the frame path works before we start
3456 // asking about the harder one.
3457 var only = [_]*quic.TestClient{&cl};
3458 _ = try quicPump(&srv, &only, 8000, &cl, struct {
3459 fn f(t: *quic.TestClient) bool {
3460 return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;
3461 }
3462 }.f);
3463 try std.testing.expect(findFrame(cl.recv_buf[0..cl.recv_len], .snapshot) != null);
3464
3465 // Now the case drainPending gets wrong when a QUIC slot contributes a
3466 // -1 descriptor: the shell exits, the exit_status is queued, and the
3467 // 250ms budget must actually deliver it rather than sleeping through it.
3468 try proto.appendFrame(&out, alloc, .input, "exit 7\n");
3469 cl.out = out.items;
3470 cl.drain();
3471
3472 const code = try quicPump(&srv, &only, 15000, &cl, struct {
3473 fn f(t: *quic.TestClient) bool {
3474 return findFrame(t.recv_buf[0..t.recv_len], .exit_status) != null;
3475 }
3476 }.f);
3477 _ = code;
3478
3479 const status = findFrame(cl.recv_buf[0..cl.recv_len], .exit_status);
3480 try std.testing.expect(status != null);
3481 try std.testing.expectEqual(@as(usize, 1), status.?.len);
3482 try std.testing.expectEqual(@as(u8, 7), status.?[0]);
3483 }
3484
3485 test "Server: one QUIC client leaving does not disturb the other" {
3486 const alloc = std.testing.allocator;
3487 var tmp = std.testing.tmpDir(.{});
3488 defer tmp.cleanup();
3489 var path_buf: [256]u8 = undefined;
3490 const dir_path = try tmp.dir.realpath(".", &path_buf);
3491 const sock_path = try std.fmt.allocPrint(alloc, "{s}/qtwo.sock", .{dir_path});
3492 defer alloc.free(sock_path);
3493
3494 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
3495 defer srv.deinit();
3496 const key: quic.Key = .{ .bytes = [_]u8{0x77} ** quic.key_len };
3497 const q = try quicTestServer(&srv, key);
3498 defer q.l.deinit();
3499
3500 var a = try quic.TestClient.init(q.addr, key);
3501 defer a.deinit();
3502 try a.start();
3503 a.drain();
3504 var b = try quic.TestClient.init(q.addr, key);
3505 defer b.deinit();
3506 try b.start();
3507 b.drain();
3508
3509 var abuf: std.ArrayList(u8) = .empty;
3510 defer abuf.deinit(alloc);
3511 var bbuf: std.ArrayList(u8) = .empty;
3512 defer bbuf.deinit(alloc);
3513 try attachOver(&a, &abuf, alloc);
3514 try attachOver(&b, &bbuf, alloc);
3515
3516 var both = [_]*quic.TestClient{ &a, &b };
3517 _ = try quicPump(&srv, &both, 10000, &b, struct {
3518 fn f(t: *quic.TestClient) bool {
3519 return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;
3520 }
3521 }.f);
3522 try std.testing.expect(findFrame(a.recv_buf[0..a.recv_len], .snapshot) != null);
3523 try std.testing.expect(findFrame(b.recv_buf[0..b.recv_len], .snapshot) != null);
3524
3525 // A leaves. Sink.close() must tear down A's connection and NOT the UDP
3526 // socket every other client is reached through — for a socket client
3527 // those are the same object, which is exactly why this needs a test.
3528 try proto.appendFrame(&abuf, alloc, .detach, "");
3529 a.out = abuf.items;
3530 a.drain();
3531 _ = try quicPump(&srv, &both, 4000, &srv, struct {
3532 fn f(s: *Server) bool {
3533 var n: usize = 0;
3534 for (s.clients) |slot| {
3535 if (slot != null) n += 1;
3536 }
3537 return n == 1;
3538 }
3539 }.f);
3540
3541 // B is still served: a marker typed now must come back to it.
3542 const before = b.recv_len;
3543 try proto.appendFrame(&bbuf, alloc, .input, "echo quic-two-ok\n");
3544 b.out = bbuf.items;
3545 b.drain();
3546 var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
3547 defer replica.deinit();
3548 _ = try quicPump(&srv, &both, 15000, &b, struct {
3549 fn f(t: *quic.TestClient) bool {
3550 return t.recv_len > 0 and findFrame(t.recv_buf[0..t.recv_len], .delta) != null;
3551 }
3552 }.f);
3553 try std.testing.expect(b.recv_len > before);
3554 }
3555
3556 test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" {
3557 const alloc = std.testing.allocator;
3558 var tmp = std.testing.tmpDir(.{});
3559 defer tmp.cleanup();
3560 var path_buf: [256]u8 = undefined;
3561 const dir_path = try tmp.dir.realpath(".", &path_buf);
3562 const sock_path = try std.fmt.allocPrint(alloc, "{s}/qcap.sock", .{dir_path});
3563 defer alloc.free(sock_path);
3564
3565 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
3566 const key: quic.Key = .{ .bytes = [_]u8{0x5D} ** quic.key_len };
3567 const q = try quicTestServer(&srv, key);
3568 defer q.l.deinit();
3569 defer srv.deinit();
3570
3571 var cl = try quic.TestClient.init(q.addr, key);
3572 defer cl.deinit();
3573 try cl.start();
3574
3575 var bbuf: std.ArrayList(u8) = .empty;
3576 defer bbuf.deinit(alloc);
3577 try attachOver(&cl, &bbuf, alloc);
3578
3579 var only = [_]*quic.TestClient{&cl};
3580 _ = try quicPump(&srv, &only, 8000, &cl, struct {
3581 fn f(t: *quic.TestClient) bool {
3582 return t.handshake_done and t.echoed > 0;
3583 }
3584 }.f);
3585 try std.testing.expect(srv.clients[0] != null);
3586
3587 // From here the client acknowledges nothing and reads nothing, so the
3588 // connection's egress ring fills and stays full. Everything the daemon
3589 // queues after that has nowhere to go but the slot's own queue, which is
3590 // what pending_cap is for. While the QUIC sink accepted every byte
3591 // unconditionally, this backlog lived inside the listener where no cap
3592 // could see it, and the daemon would have grown without bound rather
3593 // than dropping one hopeless client.
3594 srv.pending_cap = 512 * 1024;
3595 const chunk = try alloc.alloc(u8, 32 * 1024);
3596 defer alloc.free(chunk);
3597 @memset(chunk, 'q');
3598
3599 var i: usize = 0;
3600 while (i < 200 and srv.clients[0] != null) : (i += 1) {
3601 _ = srv.queueFrame(0, .snapshot, chunk);
3602 // The daemon's own machinery, not a hand-rolled loop: expiry,
3603 // egress and the flush that follows an ack all live in pumpOnce.
3604 _ = srv.pumpOnce(1) catch null;
3605 }
3606
3607 try std.testing.expect(srv.clients[0] == null);
3608 // The listener let go of the connection with the slot: a dropped client
3609 // must not leave its conn behind holding a ring.
3610 try std.testing.expect(q.l.pendingBytes(1) == 0);
3611 }
3612
3613 test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
3614 const alloc = std.testing.allocator;
3615 var tmp = std.testing.tmpDir(.{});
3616 defer tmp.cleanup();
3617 var path_buf: [256]u8 = undefined;
3618 const dir_path = try tmp.dir.realpath(".", &path_buf);
3619 const sock_path = try std.fmt.allocPrint(alloc, "{s}/qdrain.sock", .{dir_path});
3620 defer alloc.free(sock_path);
3621
3622 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
3623 const key: quic.Key = .{ .bytes = [_]u8{0x6D} ** quic.key_len };
3624 const q = try quicTestServer(&srv, key);
3625 defer q.l.deinit();
3626 defer srv.deinit();
3627
3628 var cl = try quic.TestClient.init(q.addr, key);
3629 defer cl.deinit();
3630 try cl.start();
3631
3632 var bbuf: std.ArrayList(u8) = .empty;
3633 defer bbuf.deinit(alloc);
3634 try attachOver(&cl, &bbuf, alloc);
3635
3636 var only = [_]*quic.TestClient{&cl};
3637 _ = try quicPump(&srv, &only, 8000, &cl, struct {
3638 fn f(t: *quic.TestClient) bool {
3639 return t.handshake_done and t.echoed > 0;
3640 }
3641 }.f);
3642 try std.testing.expect(srv.clients[0] != null);
3643 const before = cl.echoed;
3644
3645 // Queue more than one ring's worth in one go, so the drain cannot
3646 // possibly finish by handing everything over once: some of it is still
3647 // in the slot's queue, and the rest is in ngtcp2's hands unacknowledged.
3648 // Both have to be waited out, which is the arm that used to return
3649 // immediately and call the job done.
3650 const payload = try alloc.alloc(u8, 700 * 1024);
3651 defer alloc.free(payload);
3652 @memset(payload, 'D');
3653 _ = srv.queueFrame(0, .snapshot, payload);
3654 try std.testing.expect(srv.clients[0] != null);
3655 try std.testing.expect(srv.clients[0].?.pending.items.len > 0);
3656
3657 // drainPending has to do the waiting itself. Nothing else is driving the
3658 // daemon here — the only thing servicing the listener for the rest of
3659 // this test is the loop inside drainPending.
3660 var done = false;
3661 const t0 = std.time.milliTimestamp();
3662 const th = try std.Thread.spawn(.{}, struct {
3663 fn f(client: *quic.TestClient, flag: *bool) void {
3664 // The peer: reads and acknowledges until the daemon says stop.
3665 while (!flag.*) {
3666 client.drain();
3667 var pfd = [_]std.posix.pollfd{
3668 .{ .fd = client.fd, .events = std.posix.POLL.IN, .revents = 0 },
3669 };
3670 if ((std.posix.poll(&pfd, 5) catch 0) > 0) client.readable();
3671 }
3672 }
3673 }.f, .{ &cl, &done });
3674 srv.drainPending(15_000);
3675 done = true;
3676 th.join();
3677 const elapsed = std.time.milliTimestamp() - t0;
3678
3679 // Everything owed actually left, and it left inside the budget.
3680 try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.pending.items.len);
3681 try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.sink.inFlight());
3682 try std.testing.expect(cl.echoed >= before + payload.len);
3683 try std.testing.expect(elapsed < 15_000);
3684 }
test/e2e.sh
Old New
@@ -14,6 +14,14 @@ D2PID=""
14 # the same path, so it cannot share the long-lived one. 14 # the same path, so it cannot share the long-lived one.
15 SOCK3="${TMPDIR:-/tmp}/muxd-e2e-restart-$$.sock" 15 SOCK3="${TMPDIR:-/tmp}/muxd-e2e-restart-$$.sock"
16 D3PID="" 16 D3PID=""
17 # Fourth daemon, for the M8 --quic scenario: it is the only one holding a UDP
18 # port, so it gets its own path rather than sharing the long-lived one.
19 SOCK4="${TMPDIR:-/tmp}/muxd-e2e-quic-$$.sock"
20 QKEY="${TMPDIR:-/tmp}/mux-e2e-key-$$"
21 D4PID=""
22 # A port out of the way of the ephemeral range, made per-run so two suites can
23 # overlap. Collisions surface as a loud bind failure, never as a silent pass.
24 QPORT=$(( 21000 + ($$ % 4000) ))
17 25
18 # Wait until PATTERN shows up in FILE (default 15s). Timing that keys off the 26 # Wait until PATTERN shows up in FILE (default 15s). Timing that keys off the
19 # session's own output instead of a fixed sleep: the marker is proof the 27 # session's own output instead of a fixed sleep: the marker is proof the
@@ -42,8 +50,11 @@ cleanup() {
42 # failing a suite that had actually passed. 50 # failing a suite that had actually passed.
43 [ -n "$D2PID" ] && kill "$D2PID" 2>/dev/null || true 51 [ -n "$D2PID" ] && kill "$D2PID" 2>/dev/null || true
44 [ -n "$D3PID" ] && kill "$D3PID" 2>/dev/null || true 52 [ -n "$D3PID" ] && kill "$D3PID" 2>/dev/null || true
45 rm -f "$SOCK" "$SOCK2" "$SOCK3" "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" \ 53 [ -n "$D4PID" ] && kill "$D4PID" 2>/dev/null || true
46 "$OUT.b" "$OUT.via" "$OUT.dead" "$OUT.abort" "$OUT.m7" "$OUT.m7b" 54 rm -f "$SOCK" "$SOCK2" "$SOCK3" "$SOCK4" "$SOCK4.second" "$QKEY" "$QKEY.bad" \
55 "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" \
56 "$OUT.b" "$OUT.via" "$OUT.dead" "$OUT.abort" "$OUT.m7" "$OUT.m7b" \
57 "$OUT.q"
47 } 58 }
48 trap cleanup EXIT INT TERM 59 trap cleanup EXIT INT TERM
49 60
@@ -288,4 +299,102 @@ SNAPS_NEW=$("$MUXD" stats --sock "$SOCK3" | sed -n 's/.*snapshots=\([0-9]*\).*/\
288 } 299 }
289 rm -f "$OUT.m7b" 300 rm -f "$OUT.m7b"
290 301
302 # --- M8: the --quic flags. Every refusal must cost nothing — no session
303 # socket, no shell, no stack trace — and the accepted case must leave a
304 # daemon that is both listening on UDP and still an ordinary muxd.
305 #
306 # The client half of this (attaching over quic://) is Task 3; what is proven
307 # here is the daemon's side of the command line.
308
309 # A refusal that leaves a socket behind has already started a session, which
310 # is the failure this ordering exists to prevent.
311 refuse() {
312 _want="$1"; shift
313 set +e
314 "$MUXD" run --sock "$SOCK4" --shell /bin/sh "$@" > "$OUT.q" 2>&1
315 _rc=$?
316 set -e
317 [ "$_rc" -eq "$_want" ] || {
318 echo "e2e FAIL: muxd run $* exited $_rc (want $_want)"; cat "$OUT.q"; exit 1;
319 }
320 [ ! -e "$SOCK4" ] || {
321 echo "e2e FAIL: muxd run $* was refused but left $SOCK4 behind"; exit 1;
322 }
323 # One line, and not a stack trace: a Zig panic runs to dozens of lines
324 # and names a source file, which is what this is guarding against.
325 [ "$(wc -l < "$OUT.q")" -le 8 ] || {
326 echo "e2e FAIL: muxd run $* answered with more than a message:"; cat "$OUT.q"; exit 1;
327 }
328 }
329
330 head -c 32 /dev/urandom > "$QKEY"
331 chmod 600 "$QKEY"
332 cp "$QKEY" "$QKEY.bad"
333 chmod 644 "$QKEY.bad"
334
335 # Both or neither, and a usage mistake exits 2 like every other one.
336 refuse 2 --quic "127.0.0.1:$QPORT"
337 refuse 2 --key "$QKEY"
338 refuse 2 --quic "127.0.0.1:$QPORT" --key "$QKEY" --quic-idle-ms 0
339 refuse 2 --quic "127.0.0.1:$QPORT" --key "$QKEY" --quic-idle-ms soon
340 # Refusals that are about the world rather than the spelling exit 1.
341 refuse 1 --quic "127.0.0.1" --key "$QKEY"
342 refuse 1 --quic "localhost:$QPORT" --key "$QKEY"
343 refuse 1 --quic "127.0.0.1:$QPORT" --key "$QKEY.bad"
344 refuse 1 --quic "127.0.0.1:$QPORT" --key "$QKEY.missing"
345
346 # The accepted case.
347 "$MUXD" run --sock "$SOCK4" --shell /bin/sh \
348 --quic "127.0.0.1:$QPORT" --key "$QKEY" --quic-idle-ms 3000 &
349 D4PID=$!
350 i=0
351 while [ ! -S "$SOCK4" ] && [ "$i" -lt 50 ]; do sleep 0.1; i=$((i+1)); done
352 [ -S "$SOCK4" ] || { echo "e2e FAIL: --quic daemon never bound its session socket"; exit 1; }
353
354 # The UDP port is actually held. /proc/net/udp rather than ss or lsof: it is
355 # always there on the platform this daemon runs on, and needs no privileges.
356 # 127.0.0.1 is 0100007F in the little-endian hex the file uses.
357 QHEX=$(printf '0100007F:%04X' "$QPORT")
358 grep -qi " $QHEX " /proc/net/udp || {
359 echo "e2e FAIL: no UDP socket bound at 127.0.0.1:$QPORT ($QHEX)"
360 grep -i "0100007F" /proc/net/udp || true
361 exit 1
362 }
363
364 # A second daemon must NOT be able to take a share of that port. UDP with
365 # SO_REUSEADDR would let it bind alongside the first and the kernel would
366 # hand each datagram to one of them — two sessions splitting one port, with
367 # no error anywhere. This is the QUIC edition of the stale-socket story, and
368 # it needs two processes to test, which is why it lives here.
369 set +e
370 "$MUXD" run --sock "$SOCK4.second" --shell /bin/sh \
371 --quic "127.0.0.1:$QPORT" --key "$QKEY" > "$OUT.q" 2>&1
372 RC=$?
373 set -e
374 [ "$RC" -eq 1 ] || {
375 echo "e2e FAIL: a second daemon took udp $QPORT (exit $RC, want 1)"; cat "$OUT.q"; exit 1;
376 }
377 grep -q "already listening" "$OUT.q" || {
378 echo "e2e FAIL: second daemon refused, but not with the already-listening message:"
379 cat "$OUT.q"; exit 1;
380 }
381 [ ! -e "$SOCK4.second" ] || {
382 echo "e2e FAIL: refused second daemon left $SOCK4.second behind"; exit 1;
383 }
384
385 # ...and the daemon is still an ordinary daemon: the session runs and the
386 # unix-socket path is unaffected by the listener sharing its poll loop.
387 { printf 'printf "quic-%%s\\n" flags-ok\n'; sleep 2; printf '\034'; } | \
388 "$MUX" --sock "$SOCK4" > "$OUT.q"
389 grep -q "quic-flags-ok" "$OUT.q" || {
390 echo "e2e FAIL: --quic daemon did not serve an ordinary socket client"; cat "$OUT.q"; exit 1;
391 }
392 "$MUXD" dump --sock "$SOCK4" | grep -q "quic-flags-ok" || {
393 echo "e2e FAIL: --quic daemon's grid missing output"; exit 1;
394 }
395 kill -0 "$D4PID" || { echo "e2e FAIL: --quic daemon died"; exit 1; }
396 kill "$D4PID" 2>/dev/null || true
397 D4PID=""
398 rm -f "$OUT.q" "$QKEY" "$QKEY.bad"
399
291 echo "e2e OK" 400 echo "e2e OK"