a73x

a88c3250

feat: mux quic://host — the transport swap, third pass

a73x   2026-08-08 14:08

Commit message
feat: mux quic://host — the transport swap, third pass

`mux quic://HOST:PORT --key FILE` (or MUX_KEY_FILE) attaches straight to a
muxd running with a matching `--quic`. The parse arm counts with the others,
so naming it alongside HOST, `--sock` or `--via` is the same conflict as any
other pairing. A key with no `quic://` is ignored rather than refused —
unlike muxd, where `--key` alone means a listener was meant, here it is one
exported variable away from breaking every local attach in a shell.

src/quic_client.zig is the mirror of the listener and carries the three
things the daemon side already paid for: the egress ring, because ngtcp2
keeps pointers into outbound bytes until they are acked; the
STREAM_DATA_BLOCKED retry, because abandoning the egress loop also abandons
the ACKs that reopen the window; and extending BOTH flow-control windows on
consume, since extending only the stream's stalls the DAEMON a few hundred
kilobytes in. It shares the listener's @cImport rather than making its own —
two imports of the same headers are two distinct Zig types, and the ring
hands `ngtcp2_vec`s across the boundary.

Three things this cost that the plan did not predict, each found by a gate
rather than by reading:

**A QUIC client could never delta-resume.** The daemon's client-slot attach
arm snapshotted unconditionally, on the recorded reasoning that a client
attaching twice over one connection was too rare to deserve a resync path.
That was true while every client began as an observer and did its first
attach from there. A QUIC connection becomes a client slot when its
handshake completes, so its first attach lands on that arm — and every QUIC
attach was snapshot-served. This is kill-criterion leg 1, and only the e2e
counter assertion could see it: the rendering is identical either way.

**An incomplete frame must not skip the rest of the loop.** Over a socket a
readable descriptor always carried a frame, so `continue` was safe. Over
QUIC most passes have no complete frame, and skipping meant the detach
chord and every keystroke went unread — the client hung with its output
correct on screen.

**A QUIC transport has to be serviced whether or not its socket is
readable.** Its timers are the only thing that notices a peer that stopped
answering; polling it only when readable means a connection that goes quiet
never runs its idle timer and never dies. `Transport.open` also waits for
the handshake now, so "open succeeded" means the same thing it does for
every other transport — without that, reconnect counted dead attempts as
live ones and fired attaches into connections that never completed.

Measured on loopback, five tears: cont-to-resume 2-3ms, all delta-served.
That is the protocol cost with nothing else in it; what it does against a
real RTT is Task 5's question. 0-RTT is not part of this and cannot be:
early data is absent from this wolfSSL build, unreachable through its cmake
options, and layout-affecting to force — reported separately.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -111,6 +111,18 @@ pub fn build(b: *std.Build) void {
111 server_mod.addImport("quic", quic_mod); 111 server_mod.addImport("quic", quic_mod);
112 server_mod.addImport("testtmp", testtmp_mod); 112 server_mod.addImport("testtmp", testtmp_mod);
113 113
114 // The client's QUIC transport. Imports the listener's module for the
115 // pieces both ends must agree on (the key, the egress ring's lifetime
116 // discipline, the PSK identity and ALPN) — duplicating those would make
117 // a handshake failure the first sign they had drifted.
118 const quic_client_mod = b.createModule(.{
119 .root_source_file = b.path("src/quic_client.zig"),
120 .target = target,
121 .optimize = optimize,
122 .link_libc = true,
123 });
124 quic_client_mod.addImport("quic", quic_mod);
125
114 const client_mod = b.createModule(.{ 126 const client_mod = b.createModule(.{
115 .root_source_file = b.path("src/client.zig"), 127 .root_source_file = b.path("src/client.zig"),
116 .target = target, 128 .target = target,
@@ -120,6 +132,7 @@ pub fn build(b: *std.Build) void {
120 client_mod.addImport("engine", engine_mod); 132 client_mod.addImport("engine", engine_mod);
121 client_mod.addImport("protocol", protocol_mod); 133 client_mod.addImport("protocol", protocol_mod);
122 client_mod.addImport("testtmp", testtmp_mod); 134 client_mod.addImport("testtmp", testtmp_mod);
135 client_mod.addImport("quic_client", quic_client_mod);
123 136
124 const mux_mod = b.createModule(.{ 137 const mux_mod = b.createModule(.{
125 .root_source_file = b.path("src/mux_main.zig"), 138 .root_source_file = b.path("src/mux_main.zig"),
@@ -167,6 +180,8 @@ pub fn build(b: *std.Build) void {
167 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod }); 180 const mux_exe = b.addExecutable(.{ .name = "mux", .root_module = mux_mod });
168 mux_exe.use_llvm = true; 181 mux_exe.use_llvm = true;
169 mux_exe.use_lld = true; 182 mux_exe.use_lld = true;
183 // The client speaks QUIC now, so it carries the stack too.
184 linkQuic(b, mux_exe, quic);
170 b.installArtifact(mux_exe); 185 b.installArtifact(mux_exe);
171 186
172 const test_step = b.step("test", "Run unit tests"); 187 const test_step = b.step("test", "Run unit tests");
@@ -175,7 +190,7 @@ pub fn build(b: *std.Build) void {
175 // absence here was a live hazard recorded in decisions.md — muxd's 190 // absence here was a live hazard recorded in decisions.md — muxd's
176 // entrypoint could grow tests that silently never ran, exactly as 191 // entrypoint could grow tests that silently never ran, exactly as
177 // mux_main.zig's five did before it was added. 192 // mux_main.zig's five did before it was added.
178 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, exe_mod, testtmp_mod }) |mod| { 193 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, exe_mod, testtmp_mod, quic_client_mod }) |mod| {
179 const t = b.addTest(.{ .root_module = mod }); 194 const t = b.addTest(.{ .root_module = mod });
180 t.use_llvm = true; 195 t.use_llvm = true;
181 t.use_lld = true; 196 t.use_lld = true;
@@ -185,7 +200,8 @@ pub fn build(b: *std.Build) void {
185 // deps when they are absent. Without it the dependency reached only 200 // deps when they are absent. Without it the dependency reached only
186 // `muxd`, and a clean checkout running `make test` first would have 201 // `muxd`, and a clean checkout running `make test` first would have
187 // found no libraries and no explanation. 202 // found no libraries and no explanation.
188 if (mod == server_mod or mod == quic_mod or mod == exe_mod) linkQuic(b, t, quic); 203 if (mod == server_mod or mod == quic_mod or mod == exe_mod or
204 mod == client_mod or mod == mux_mod or mod == quic_client_mod) linkQuic(b, t, quic);
189 test_step.dependOn(&b.addRunArtifact(t).step); 205 test_step.dependOn(&b.addRunArtifact(t).step);
190 } 206 }
191 207
src/client.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 proto = @import("protocol"); 10 const proto = @import("protocol");
11 const TmpDir = @import("testtmp").TmpDir; 11 const TmpDir = @import("testtmp").TmpDir;
12 const quic_client = @import("quic_client");
12 13
13 var winch_flag = std.atomic.Value(bool).init(false); 14 var winch_flag = std.atomic.Value(bool).init(false);
14 15
@@ -48,15 +49,68 @@ const reconnect_grace_ms: i64 = 5000;
48 /// struct is that it can be closed and opened again from the same recipe 49 /// struct is that it can be closed and opened again from the same recipe
49 /// (`sock_path` or `via`), which is what lets a session outlive its 50 /// (`sock_path` or `via`), which is what lets a session outlive its
50 /// transport instead of exiting with it. 51 /// transport instead of exiting with it.
52 /// How long a QUIC connection tolerates silence before declaring the peer
53 /// gone. Keepalives run at a third of this, so an idle session is never the
54 /// thing that trips it — only a peer that has actually stopped answering.
55 pub const quic_idle_ms_default: u32 = 15_000;
56
57 /// What a QUIC attach was asked for: where to dial and what key to prove
58 /// ourselves with.
59 pub const QuicTarget = struct {
60 host_port: []const u8,
61 key_path: []const u8,
62 idle_ms: u32 = quic_idle_ms_default,
63 };
64
51 const Transport = struct { 65 const Transport = struct {
52 conn: Conn, 66 conn: Conn,
53 /// Only set for `--via`: the command whose stdio *is* the transport. 67 /// Only set for `--via`: the command whose stdio *is* the transport.
54 child: ?std.process.Child = null, 68 child: ?std.process.Child = null,
69 /// Only set for `quic://`: the connection that IS the transport. When
70 /// this is non-null `conn` holds the UDP socket in `.r` so the poll
71 /// path needs no special case, and `.w` is -1 because there is nothing
72 /// to write(2) to — bytes go through the stream layer.
73 quic: ?*quic_client.Client = null,
74 /// Bytes the QUIC ring would not take yet. Frames are appended whole and
75 /// handed over a prefix at a time, so a short accept can never split one
76 /// on the wire — it just means the rest waits here. Keystrokes and
77 /// attach frames are all this ever holds.
78 qout: std.ArrayList(u8) = .empty,
79 alloc: std.mem.Allocator = undefined,
55 80
56 /// Exactly one of `sock_path`/`via` is non-null; mux_main.zig enforces 81 /// Exactly one of `sock_path`/`via` is non-null; mux_main.zig enforces
57 /// that. Errors are returned rather than reported here — the caller 82 /// that. Errors are returned rather than reported here — the caller
58 /// knows which recipe it handed us and so which message fits. 83 /// knows which recipe it handed us and so which message fits.
59 fn open(alloc: std.mem.Allocator, sock_path: ?[]const u8, via: ?[]const u8) !Transport { 84 /// `carry` collects any non-abort bytes typed while a QUIC handshake is
85 /// in flight; null means drop them. See waitReady.
86 fn open(
87 alloc: std.mem.Allocator,
88 sock_path: ?[]const u8,
89 via: ?[]const u8,
90 quic: ?QuicTarget,
91 carry: ?*std.ArrayList(u8),
92 ) !Transport {
93 if (quic) |q| {
94 const key = try quic_client.Key.load(q.key_path);
95 const addr = try parseQuicAddr(q.host_port);
96 const cl = try quic_client.Client.connect(alloc, addr, key, q.idle_ms);
97 errdefer cl.deinit();
98 // Wait for the handshake here rather than letting the caller
99 // discover it later. `connect` only creates state — the first
100 // flight has not been answered yet — so returning now would make
101 // "open succeeded" mean something different for QUIC than for
102 // every other transport, and the reconnect loop believes it: it
103 // would count a dead attempt as a live one, fire an attach into
104 // a connection that never completes, and do it again on the next
105 // pass. Against a daemon that was merely paused, that left a
106 // trail of half-open connections and duplicate attaches.
107 try waitReady(cl, q.idle_ms, alloc, carry);
108 return .{
109 .conn = .{ .r = cl.pollFd(), .w = -1 },
110 .quic = cl,
111 .alloc = alloc,
112 };
113 }
60 if (via) |cmd| { 114 if (via) |cmd| {
61 var child = std.process.Child.init(&.{ "/bin/sh", "-c", cmd }, alloc); 115 var child = std.process.Child.init(&.{ "/bin/sh", "-c", cmd }, alloc);
62 child.stdin_behavior = .Pipe; 116 child.stdin_behavior = .Pipe;
@@ -94,12 +148,80 @@ const Transport = struct {
94 } 148 }
95 149
96 fn writeFrame(self: *Transport, t: proto.MsgType, payload: []const u8) !void { 150 fn writeFrame(self: *Transport, t: proto.MsgType, payload: []const u8) !void {
151 if (self.quic) |_| {
152 // Appended whole, handed over a prefix at a time. A frame can
153 // therefore never be split across a refusal — the ring takes
154 // what it takes and the remainder is offered again next pass.
155 try proto.appendFrame(&self.qout, self.alloc, t, payload);
156 self.flushQuic();
157 return;
158 }
97 return proto.writeFrame(self.conn.w, t, payload); 159 return proto.writeFrame(self.conn.w, t, payload);
98 } 160 }
99 161
162 /// One service pass, called every loop iteration whether or not the
163 /// socket was readable.
164 ///
165 /// It has to be unconditional, and finding out why cost an e2e run: a
166 /// QUIC connection's timers are the ONLY thing that notices a peer which
167 /// has stopped answering. Servicing it just from the readable path means
168 /// a transport that goes quiet — a wrong key, a daemon that stopped —
169 /// never runs its idle timer, never declares itself dead, and the client
170 /// polls a corpse forever instead of exiting or reconnecting.
171 fn service(self: *Transport) void {
172 const cl = self.quic orelse return;
173 cl.pump();
174 self.flushQuic();
175 }
176
177 /// How long the caller may sleep before this transport needs attention.
178 /// Folds ngtcp2's next deadline in, so retransmits and idle timeouts
179 /// happen on time without a second timer.
180 fn timeoutMs(self: *Transport, cap_ms: i32) i32 {
181 const cl = self.quic orelse return cap_ms;
182 return cl.timeoutMs(cap_ms);
183 }
184
185 /// Whether frames can arrive from somewhere other than the descriptor.
186 /// True only for QUIC, where bytes already sit in a buffer and the next
187 /// whole frame may be there without the socket ever going readable.
188 fn buffersFrames(self: *const Transport) bool {
189 return self.quic != null;
190 }
191
192 /// Offer the outbound queue to the ring again. Called after every write
193 /// and on every service pass, because the room to accept comes from
194 /// acknowledgements, which arrive on their own schedule.
195 fn flushQuic(self: *Transport) void {
196 const cl = self.quic orelse return;
197 if (self.qout.items.len == 0) return;
198 const n = cl.send(self.qout.items);
199 if (n == 0) return;
200 const rest = self.qout.items.len - n;
201 std.mem.copyForwards(u8, self.qout.items[0..rest], self.qout.items[n..]);
202 self.qout.shrinkRetainingCapacity(rest);
203 }
204
100 /// The next whole frame, if there is one. See `Incoming` for why a 205 /// The next whole frame, if there is one. See `Incoming` for why a
101 /// missing frame is not automatically a dead transport. 206 /// missing frame is not automatically a dead transport.
102 fn readFrame(self: *Transport, alloc: std.mem.Allocator) !Incoming { 207 fn readFrame(self: *Transport, alloc: std.mem.Allocator) !Incoming {
208 if (self.quic) |cl| {
209 // Death is checked after the pump, so bytes that arrived in the
210 // same pass as the close are still delivered before the tear.
211 const buf = cl.inbound();
212 if (buf.len < 5) return if (cl.dead) .closed else .incomplete;
213 const len = std.mem.readInt(u32, buf[1..5], .little);
214 if (len > proto.max_payload) return .closed;
215 if (buf.len < 5 + len) return if (cl.dead) .closed else .incomplete;
216 const payload = try alloc.alloc(u8, len);
217 @memcpy(payload, buf[5 .. 5 + len]);
218 const frame: proto.Frame = .{
219 .type = @enumFromInt(buf[0]),
220 .payload = payload,
221 };
222 cl.consume(5 + len);
223 return .{ .frame = frame };
224 }
103 const frame = (proto.readFrame(alloc, self.conn.r) catch |err| switch (err) { 225 const frame = (proto.readFrame(alloc, self.conn.r) catch |err| switch (err) {
104 // Not a transport event, and it stays loud. 226 // Not a transport event, and it stays loud.
105 error.OutOfMemory => return err, 227 error.OutOfMemory => return err,
@@ -111,6 +233,15 @@ const Transport = struct {
111 fn close(self: *Transport) void { 233 fn close(self: *Transport) void {
112 if (self.conn.r == -1) return; // already released 234 if (self.conn.r == -1) return; // already released
113 defer self.conn = .{ .r = -1, .w = -1 }; 235 defer self.conn = .{ .r = -1, .w = -1 };
236 if (self.quic) |cl| {
237 defer self.quic = null;
238 self.qout.deinit(self.alloc);
239 self.qout = .empty;
240 // Closes the UDP socket with it, so the fd this returns through
241 // `conn.r` must not be closed again below.
242 cl.deinit();
243 return;
244 }
114 if (self.child) |*c| { 245 if (self.child) |*c| {
115 defer self.child = null; 246 defer self.child = null;
116 // Close stdin first so the command sees EOF and can wind down its 247 // Close stdin first so the command sees EOF and can wind down its
@@ -127,12 +258,153 @@ const Transport = struct {
127 } 258 }
128 }; 259 };
129 260
261 /// Drive a fresh connection until it can carry bytes, or give up.
262 ///
263 /// Bounded by the same idle timeout the connection itself uses: a peer that
264 /// will not answer a handshake is the same peer that will not answer
265 /// anything, and one knob for both is one fewer thing to explain. An
266 /// unreachable UDP port usually produces no error at all — no ICMP, no
267 /// refusal — so this bound is the only thing that ends the wait.
268 fn waitReady(
269 cl: *quic_client.Client,
270 idle_ms: u32,
271 alloc: std.mem.Allocator,
272 carry: ?*std.ArrayList(u8),
273 ) !void {
274 const deadline = std.time.milliTimestamp() + idle_ms;
275 // A closed stdin stays readable forever, so once it reports EOF it has to
276 // stop being polled or this loop spins hot for the rest of the bound
277 // instead of waiting on the socket. Same hazard drainStdinForQuit
278 // documents, reached from the other direction.
279 var watch_stdin = true;
280 while (std.time.milliTimestamp() < deadline) {
281 cl.pump();
282 if (cl.isReady()) return;
283 if (cl.dead) return error.QuicHandshakeFailed;
284 // stdin is watched alongside the socket, and it has to be. This wait
285 // runs INSIDE Transport.open, after drainStdinForQuit has already
286 // returned, so for its whole length nothing else is looking for the
287 // abort key — and during a reconnect the terminal is in raw mode, so
288 // Ctrl-C is just a byte and Ctrl-\ is the only way out. Watching only
289 // the socket left the user with no way to stop for as long as the
290 // handshake bound allows: measured at 14.6s on the default idle
291 // timeout, against an endpoint that was never going to answer. M7's
292 // uncapped retry loop is justified by the user having an abort key,
293 // so an abort key that stops working mid-handshake takes the
294 // justification with it.
295 var fds = [_]std.posix.pollfd{
296 .{ .fd = cl.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
297 .{
298 .fd = if (watch_stdin) std.posix.STDIN_FILENO else -1,
299 .events = std.posix.POLL.IN,
300 .revents = 0,
301 },
302 };
303 _ = std.posix.poll(&fds, cl.timeoutMs(50)) catch break;
304 if (fds[1].revents != 0) {
305 var buf: [1024]u8 = undefined;
306 const n = std.posix.read(std.posix.STDIN_FILENO, &buf) catch 0;
307 if (n == 0) watch_stdin = false;
308 if (n > 0) {
309 if (std.mem.indexOfScalar(u8, buf[0..n], 0x1c) != null) return error.UserAbort;
310 // Not the abort key. Whether these bytes are kept or dropped
311 // is the caller's policy, not this function's: on a first
312 // attach they are the user's first keystrokes and are owed to
313 // the shell, while during a reconnect input is dropped by
314 // long-standing policy — replaying a burst of stale
315 // keystrokes on resume is worse than losing them.
316 if (carry) |q| q.appendSlice(alloc, buf[0..n]) catch {};
317 }
318 }
319 }
320 cl.pump();
321 if (cl.isReady()) return;
322 return error.QuicHandshakeFailed;
323 }
324
325 /// `HOST:PORT` for a `quic://` target. Literal addresses only on the muxd
326 /// side because a bind address that resolves to several is a question; here
327 /// a NAME is exactly what a user types, so this one does resolve.
328 fn parseQuicAddr(host_port: []const u8) !std.net.Address {
329 const colon = std.mem.lastIndexOfScalar(u8, host_port, ':') orelse
330 return error.MalformedAddress;
331 var host = host_port[0..colon];
332 const port_s = host_port[colon + 1 ..];
333 // `[::1]:4433` — brackets are how an IPv6 literal says where it stops.
334 if (host.len >= 2 and host[0] == '[' and host[host.len - 1] == ']') {
335 host = host[1 .. host.len - 1];
336 } else if (std.mem.indexOfScalar(u8, host, ':') != null) {
337 // Unbracketed and full of colons: an IPv6 literal missing its
338 // brackets, which would otherwise have its last group taken as a
339 // port. Refused rather than guessed at.
340 return error.MalformedAddress;
341 }
342 if (host.len == 0) return error.MalformedAddress;
343 const port = std.fmt.parseInt(u16, port_s, 10) catch return error.MalformedAddress;
344 if (std.net.Address.parseIp(host, port)) |addr| return addr else |_| {}
345 // Not a literal: resolve it. A remote host is normally a name.
346 const list = try std.net.getAddressList(std.heap.page_allocator, host, port);
347 defer list.deinit();
348 if (list.addrs.len == 0) return error.UnknownHostName;
349 return list.addrs[0];
350 }
351
130 /// Attach over a local socket (`sock_path`) or over an arbitrary command's 352 /// Attach over a local socket (`sock_path`) or over an arbitrary command's
131 /// stdio (`via`, typically `ssh host muxd proxy ...`). Exactly one is 353 /// stdio (`via`, typically `ssh host muxd proxy ...`). Exactly one is
132 /// non-null; mux_main.zig enforces that. 354 /// non-null; mux_main.zig enforces that.
133 pub fn attach(alloc: std.mem.Allocator, sock_path: ?[]const u8, via: ?[]const u8) !u8 { 355 pub fn attach(
134 var transport = Transport.open(alloc, sock_path, via) catch { 356 alloc: std.mem.Allocator,
135 if (via) |cmd| { 357 sock_path: ?[]const u8,
358 via: ?[]const u8,
359 quic: ?QuicTarget,
360 ) !u8 {
361 // Anything typed while the first handshake is in flight belongs to the
362 // shell, so it is held here rather than dropped — and handed over once
363 // there is a session to hand it to.
364 var carry: std.ArrayList(u8) = .empty;
365 defer carry.deinit(alloc);
366
367 var transport = Transport.open(alloc, sock_path, via, quic, &carry) catch |err| {
368 if (quic) |q| {
369 // A key the daemon would also have refused, said in the same
370 // words, because the user's mistake is the same one.
371 switch (err) {
372 error.KeyFileMissing => std.debug.print(
373 "mux: no such key file: {s}\n",
374 .{q.key_path},
375 ),
376 error.KeyFilePermissive => std.debug.print(
377 "mux: {s} is readable by group or other; chmod 600 it\n",
378 .{q.key_path},
379 ),
380 error.KeyFileMalformed => std.debug.print(
381 "mux: {s} is not a key: want 32 raw bytes or 64 hex characters\n",
382 .{q.key_path},
383 ),
384 error.MalformedAddress, error.UnknownHostName => std.debug.print(
385 "mux: cannot resolve quic://{s}\n",
386 .{q.host_port},
387 ),
388 // The handshake is also where a wrong key lands: an external
389 // PSK that does not match produces no distinguishable
390 // rejection, just a handshake that never completes. Saying
391 // both is more honest than guessing which it was.
392 error.QuicHandshakeFailed => std.debug.print(
393 "mux: quic://{s} did not answer (wrong key, or no muxd --quic there)\n",
394 .{q.host_port},
395 ),
396 // The user pressed Ctrl-\ while we were still dialling.
397 // Nothing failed, so nothing is reported as a failure.
398 error.UserAbort => {
399 std.debug.print("mux: aborted before attaching\n", .{});
400 return 0;
401 },
402 else => std.debug.print(
403 "mux: cannot reach quic://{s}: {s}\n",
404 .{ q.host_port, @errorName(err) },
405 ),
406 }
407 } else if (via) |cmd| {
136 std.debug.print("mux: cannot start --via command: {s}\n", .{cmd}); 408 std.debug.print("mux: cannot start --via command: {s}\n", .{cmd});
137 } else { 409 } else {
138 std.debug.print("mux: cannot connect to {s} (is muxd running?)\n", .{sock_path.?}); 410 std.debug.print("mux: cannot connect to {s} (is muxd running?)\n", .{sock_path.?});
@@ -140,7 +412,7 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: ?[]const u8, via: ?[]const u8
140 return 1; 412 return 1;
141 }; 413 };
142 defer transport.close(); 414 defer transport.close();
143 return session(alloc, &transport, sock_path, via); 415 return session(alloc, &transport, sock_path, via, quic, &carry);
144 } 416 }
145 417
146 /// `sock_path` and `via` are the recipe this session's transport was built 418 /// `sock_path` and `via` are the recipe this session's transport was built
@@ -152,6 +424,10 @@ fn session(
152 transport: *Transport, 424 transport: *Transport,
153 sock_path: ?[]const u8, 425 sock_path: ?[]const u8,
154 via: ?[]const u8, 426 via: ?[]const u8,
427 quic: ?QuicTarget,
428 /// Keystrokes that arrived during the opening handshake, owed to the
429 /// shell as soon as there is an attach to send them after.
430 carry: *std.ArrayList(u8),
155 ) !u8 { 431 ) !u8 {
156 // A daemon that dies mid-write must surface as an error return from 432 // A daemon that dies mid-write must surface as an error return from
157 // write(), not a fatal SIGPIPE. Zig's start.zig already installs a noop 433 // write(), not a fatal SIGPIPE. Zig's start.zig already installs a noop
@@ -222,6 +498,18 @@ fn session(
222 return 1; 498 return 1;
223 }; 499 };
224 500
501 // Whatever was typed while the handshake was in flight, in the order it
502 // was typed, now that there is somewhere for it to go. Dropping it would
503 // silently eat the first command of a piped session, which is how a test
504 // harness types.
505 if (carry.items.len > 0) {
506 transport.writeFrame(.input, carry.items) catch {
507 exit_msg = "mux: connection to muxd lost";
508 return 1;
509 };
510 carry.clearRetainingCapacity();
511 }
512
225 var stdin_open = true; 513 var stdin_open = true;
226 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live. 514 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live.
227 var scroll_pages: u32 = 0; 515 var scroll_pages: u32 = 0;
@@ -285,6 +573,7 @@ fn session(
285 transport, 573 transport,
286 sock_path, 574 sock_path,
287 via, 575 via,
576 quic,
288 size, 577 size,
289 last_seq, 578 last_seq,
290 session_epoch, 579 session_epoch,
@@ -328,9 +617,16 @@ fn session(
328 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 }, 617 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
329 .{ .fd = if (stdin_open) stdin_fd else -1, .events = std.posix.POLL.IN, .revents = 0 }, 618 .{ .fd = if (stdin_open) stdin_fd else -1, .events = std.posix.POLL.IN, .revents = 0 },
330 }; 619 };
331 _ = try std.posix.poll(&fds, 100); 620 _ = try std.posix.poll(&fds, transport.timeoutMs(100));
332 621 transport.service();
333 if (fds[0].revents != 0) { 622
623 // Labelled, because "no whole frame yet" must leave the REST of this
624 // iteration running. Everything below — the detach chord, keystrokes,
625 // resize — lives in the same pass, so jumping straight back to the
626 // poll would drop the user's input on the floor. It never came up
627 // over a socket, where a readable descriptor always carried a frame;
628 // over QUIC it is the ordinary case, and it cost an e2e run to find.
629 if (fds[0].revents != 0 or transport.buffersFrames()) frames: {
334 // A clean EOF and a frame torn in half are the same event to us: 630 // A clean EOF and a frame torn in half are the same event to us:
335 // the transport is gone. Over ssh the tear is the ordinary case — 631 // the transport is gone. Over ssh the tear is the ordinary case —
336 // the channel drops mid-frame — so it has to read as a message 632 // the channel drops mid-frame — so it has to read as a message
@@ -339,8 +635,8 @@ fn session(
339 const frame = switch (try transport.readFrame(alloc)) { 635 const frame = switch (try transport.readFrame(alloc)) {
340 .frame => |f| f, 636 .frame => |f| f,
341 // Bytes arrived that do not complete a frame. Nothing is 637 // Bytes arrived that do not complete a frame. Nothing is
342 // wrong; go back to the poll and wait for the rest. 638 // wrong; skip the frame handling and carry on.
343 .incomplete => continue, 639 .incomplete => break :frames,
344 .closed => { 640 .closed => {
345 // Not fatal any more: rebuild the transport and 641 // Not fatal any more: rebuild the transport and
346 // re-attach with what we hold. The replica stays exactly 642 // re-attach with what we hold. The replica stays exactly
@@ -660,6 +956,7 @@ fn reconnect(
660 transport: *Transport, 956 transport: *Transport,
661 sock_path: ?[]const u8, 957 sock_path: ?[]const u8,
662 via: ?[]const u8, 958 via: ?[]const u8,
959 quic: ?QuicTarget,
663 size: proto.Size, 960 size: proto.Size,
664 last_seq: u64, 961 last_seq: u64,
665 session_epoch: u64, 962 session_epoch: u64,
@@ -693,7 +990,12 @@ fn reconnect(
693 if (drainStdinForQuit(stdin_fd, backoff_ms)) return false; 990 if (drainStdinForQuit(stdin_fd, backoff_ms)) return false;
694 backoff_ms = if (backoff_ms == 0) 200 else @min(backoff_ms * 2, 2000); 991 backoff_ms = if (backoff_ms == 0) 200 else @min(backoff_ms * 2, 2000);
695 992
696 var fresh = Transport.open(alloc, sock_path, via) catch continue; 993 var fresh = Transport.open(alloc, sock_path, via, quic, null) catch |err| {
994 // Ctrl-\ during the handshake is the same answer as Ctrl-\
995 // during the backoff: the user is done waiting.
996 if (err == error.UserAbort) return false;
997 continue;
998 };
697 fresh.writeFrame( 999 fresh.writeFrame(
698 .attach, 1000 .attach,
699 &proto.encodeAttach(size.cols, size.rows, last_seq, session_epoch), 1001 &proto.encodeAttach(size.cols, size.rows, last_seq, session_epoch),
@@ -744,7 +1046,7 @@ test "Transport.close is idempotent: the abort path closes what reconnect alread
744 var listener = try addr.listen(.{}); 1046 var listener = try addr.listen(.{});
745 defer listener.deinit(); 1047 defer listener.deinit();
746 1048
747 var transport = try Transport.open(alloc, sock_path, null); 1049 var transport = try Transport.open(alloc, sock_path, null, null, null);
748 1050
749 // reconnect() closes the dead transport at entry; if the user then aborts, 1051 // reconnect() closes the dead transport at entry; if the user then aborts,
750 // attach()'s `defer transport.close()` closes it a second time. Without a 1052 // attach()'s `defer transport.close()` closes it a second time. Without a
src/mux_main.zig
Old New
@@ -6,8 +6,11 @@ const std = @import("std");
6 const client = @import("client"); 6 const client = @import("client");
7 7
8 const usage = 8 const usage =
9 \\usage: mux [HOST | --sock PATH | --via CMD] 9 \\usage: mux [HOST | --sock PATH | --via CMD | quic://HOST:PORT]
10 \\ HOST attaches over "ssh HOST muxd proxy" (muxd must be on HOST's PATH) 10 \\ HOST attaches over "ssh HOST muxd proxy" (muxd must be on HOST's PATH)
11 \\ quic://HOST:PORT needs --key FILE (or MUX_KEY_FILE); muxd must be
12 \\ running with a matching --quic and --key
13 \\ [--quic-idle-ms N] tunes how fast a dead link is noticed
11 \\ 14 \\
12 ; 15 ;
13 16
@@ -20,16 +23,31 @@ const ParseResult = union(enum) {
20 /// A bare hostname: the ssh recipe is built from it in main, where there 23 /// A bare hostname: the ssh recipe is built from it in main, where there
21 /// is an allocator to build it with. 24 /// is an allocator to build it with.
22 host: []const u8, 25 host: []const u8,
26 /// A direct QUIC attach. The key is resolved in main, where the
27 /// environment can be consulted.
28 quic: struct { host_port: []const u8, key: ?[]const u8, idle_ms: u32 },
29 /// `quic://` named without a key anywhere. Its own result rather than a
30 /// usage error: nothing is misspelled, something is missing, and the
31 /// message that helps says which.
32 quic_without_key,
23 /// More than one transport named — a request that cannot be honoured 33 /// More than one transport named — a request that cannot be honoured
24 /// rather than one to reconcile. 34 /// rather than one to reconcile.
25 conflict, 35 conflict,
26 usage_error, 36 usage_error,
27 }; 37 };
28 38
29 fn parseArgs(args: []const [:0]const u8) ParseResult { 39 /// The environment variable consulted when `--key` is absent. Named rather
40 /// than inlined because the parse cannot read it — the parse stays pure so
41 /// it stays testable — and `main` has to use exactly the same name.
42 pub const key_env = "MUX_KEY_FILE";
43
44 fn parseArgs(args: []const [:0]const u8, env_key: ?[]const u8) ParseResult {
30 var sock: ?[]const u8 = null; 45 var sock: ?[]const u8 = null;
31 var via: ?[]const u8 = null; 46 var via: ?[]const u8 = null;
32 var host: ?[]const u8 = null; 47 var host: ?[]const u8 = null;
48 var quic: ?[]const u8 = null;
49 var key: ?[]const u8 = null;
50 var idle_ms: u32 = client.quic_idle_ms_default;
33 51
34 var i: usize = 1; 52 var i: usize = 1;
35 while (i < args.len) : (i += 1) { 53 while (i < args.len) : (i += 1) {
@@ -42,6 +60,22 @@ fn parseArgs(args: []const [:0]const u8) ParseResult {
42 i += 1; 60 i += 1;
43 if (via != null) return .conflict; 61 if (via != null) return .conflict;
44 via = args[i]; 62 via = args[i];
63 } else if (std.mem.eql(u8, a, "--key") and i + 1 < args.len) {
64 i += 1;
65 key = args[i];
66 } else if (std.mem.eql(u8, a, "--quic-idle-ms") and i + 1 < args.len) {
67 i += 1;
68 const n = std.fmt.parseInt(u32, args[i], 10) catch return .usage_error;
69 // Zero means "no idle timeout" to ngtcp2, the inverse of what
70 // anyone typing a timeout of zero is asking for.
71 if (n == 0) return .usage_error;
72 idle_ms = n;
73 } else if (std.mem.startsWith(u8, a, "quic://")) {
74 // Counted with the others, so `mux quic://a:1 --sock /x` is the
75 // same conflict as naming any other two transports.
76 if (quic != null) return .conflict;
77 quic = a["quic://".len..];
78 if (quic.?.len == 0) return .usage_error;
45 } else if (a.len > 0 and a[0] != '-') { 79 } else if (a.len > 0 and a[0] != '-') {
46 // A bare word is a host to hop to. Two of them is as ambiguous 80 // A bare word is a host to hop to. Two of them is as ambiguous
47 // as naming two transports, so it lands in the same place. 81 // as naming two transports, so it lands in the same place.
@@ -54,11 +88,24 @@ fn parseArgs(args: []const [:0]const u8) ParseResult {
54 } 88 }
55 } 89 }
56 90
57 // Every pairing of the three is two transports for one session. 91 // Every pairing of the four is two transports for one session.
58 const named: u8 = @as(u8, @intFromBool(sock != null)) + 92 const named: u8 = @as(u8, @intFromBool(sock != null)) +
59 @intFromBool(via != null) + @intFromBool(host != null); 93 @intFromBool(via != null) + @intFromBool(host != null) +
94 @intFromBool(quic != null);
60 if (named > 1) return .conflict; 95 if (named > 1) return .conflict;
61 96
97 if (quic) |hp| {
98 // --key wins over the environment; the environment exists so a
99 // shell can set it once rather than repeating it per invocation.
100 const k = key orelse env_key orelse return .quic_without_key;
101 if (k.len == 0) return .quic_without_key;
102 return .{ .quic = .{ .host_port = hp, .key = k, .idle_ms = idle_ms } };
103 }
104 // A key with no quic:// has nothing to authenticate and is ignored
105 // rather than refused: unlike muxd, where --key without --quic means a
106 // listener was meant, here it is one env var away from being set for
107 // every invocation in a shell, and refusing `mux --sock ...` because
108 // MUX_KEY_FILE happens to be exported would be absurd.
62 if (host) |h| return .{ .host = h }; 109 if (host) |h| return .{ .host = h };
63 return .{ .attach = .{ .sock = sock, .via = via } }; 110 return .{ .attach = .{ .sock = sock, .via = via } };
64 } 111 }
@@ -71,25 +118,40 @@ pub fn main() !u8 {
71 const args = try std.process.argsAlloc(alloc); 118 const args = try std.process.argsAlloc(alloc);
72 defer std.process.argsFree(alloc, args); 119 defer std.process.argsFree(alloc, args);
73 120
74 const parsed = parseArgs(args); 121 const parsed = parseArgs(args, std.posix.getenv(key_env));
75 switch (parsed) { 122 switch (parsed) {
76 .usage_error => { 123 .usage_error => {
77 std.debug.print("{s}", .{usage}); 124 std.debug.print("{s}", .{usage});
78 return 2; 125 return 2;
79 }, 126 },
80 .conflict => { 127 .conflict => {
81 std.debug.print("mux: name one transport: HOST, --sock or --via\n{s}", .{usage}); 128 std.debug.print(
129 "mux: name one transport: HOST, --sock, --via or quic://\n{s}",
130 .{usage},
131 );
132 return 2;
133 },
134 .quic_without_key => {
135 std.debug.print(
136 "mux: quic:// needs --key FILE (or {s}); there is no unauthenticated mode\n",
137 .{key_env},
138 );
82 return 2; 139 return 2;
83 }, 140 },
141 .quic => |q| return client.attach(alloc, null, null, .{
142 .host_port = q.host_port,
143 .key_path = q.key.?,
144 .idle_ms = q.idle_ms,
145 }),
84 .host => |h| { 146 .host => |h| {
85 // muxd on the far side exposes the session over its stdio; ssh 147 // muxd on the far side exposes the session over its stdio; ssh
86 // carries the bytes and nothing here knows the difference. 148 // carries the bytes and nothing here knows the difference.
87 const cmd = try std.fmt.allocPrint(alloc, "ssh {s} muxd proxy", .{h}); 149 const cmd = try std.fmt.allocPrint(alloc, "ssh {s} muxd proxy", .{h});
88 defer alloc.free(cmd); 150 defer alloc.free(cmd);
89 return client.attach(alloc, null, cmd); 151 return client.attach(alloc, null, cmd, null);
90 }, 152 },
91 .attach => |t| { 153 .attach => |t| {
92 if (t.via) |cmd| return client.attach(alloc, null, cmd); 154 if (t.via) |cmd| return client.attach(alloc, null, cmd, null);
93 const sock_path = if (t.sock) |s| 155 const sock_path = if (t.sock) |s|
94 try alloc.dupe(u8, s) 156 try alloc.dupe(u8, s)
95 else if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| 157 else if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir|
@@ -97,7 +159,7 @@ pub fn main() !u8 {
97 else 159 else
98 try std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()}); 160 try std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()});
99 defer alloc.free(sock_path); 161 defer alloc.free(sock_path);
100 return client.attach(alloc, sock_path, null); 162 return client.attach(alloc, sock_path, null, null);
101 }, 163 },
102 } 164 }
103 } 165 }
@@ -105,7 +167,12 @@ pub fn main() !u8 {
105 /// Test helper: parseArgs takes what argsAlloc produces, so the tests have to 167 /// Test helper: parseArgs takes what argsAlloc produces, so the tests have to
106 /// speak the same type — a slice of sentinel-terminated strings. 168 /// speak the same type — a slice of sentinel-terminated strings.
107 fn parse(comptime argv: []const [:0]const u8) ParseResult { 169 fn parse(comptime argv: []const [:0]const u8) ParseResult {
108 return parseArgs(argv); 170 return parseArgs(argv, null);
171 }
172
173 /// The same, with `MUX_KEY_FILE` set to `env`.
174 fn parseEnv(comptime argv: []const [:0]const u8, env: ?[]const u8) ParseResult {
175 return parseArgs(argv, env);
109 } 176 }
110 177
111 test "parseArgs: no arguments means the default local socket" { 178 test "parseArgs: no arguments means the default local socket" {
@@ -157,3 +224,55 @@ test "parseArgs: unknown flags and valueless flags are usage errors" {
157 try std.testing.expect(parse(&.{ "mux", "--sock" }) == .usage_error); 224 try std.testing.expect(parse(&.{ "mux", "--sock" }) == .usage_error);
158 try std.testing.expect(parse(&.{ "mux", "--via" }) == .usage_error); 225 try std.testing.expect(parse(&.{ "mux", "--via" }) == .usage_error);
159 } 226 }
227
228 test "parseArgs: quic:// is a transport like any other" {
229 const q = parse(&.{ "mux", "quic://box:4433", "--key", "/k" });
230 try std.testing.expect(q == .quic);
231 try std.testing.expectEqualStrings("box:4433", q.quic.host_port);
232 try std.testing.expectEqualStrings("/k", q.quic.key.?);
233 try std.testing.expectEqual(client.quic_idle_ms_default, q.quic.idle_ms);
234
235 // Counted with the rest: naming it alongside another transport is the
236 // same ambiguity as any other pairing, whichever order they arrive in.
237 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--sock", "/x" }) == .conflict);
238 try std.testing.expect(parse(&.{ "mux", "--sock", "/x", "quic://a:1", "--key", "/k" }) == .conflict);
239 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--via", "ssh h" }) == .conflict);
240 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "vm1" }) == .conflict);
241 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "quic://b:2", "--key", "/k" }) == .conflict);
242
243 // The scheme with nothing after it names no host.
244 try std.testing.expect(parse(&.{ "mux", "quic://", "--key", "/k" }) == .usage_error);
245 }
246
247 test "parseArgs: a quic attach without a key is refused, by whichever route" {
248 // No --key and no environment.
249 try std.testing.expect(parse(&.{ "mux", "quic://a:1" }) == .quic_without_key);
250 // An environment variable set to nothing is not a key.
251 try std.testing.expect(parseEnv(&.{ "mux", "quic://a:1" }, "") == .quic_without_key);
252
253 // The environment supplies it when the flag does not...
254 const e = parseEnv(&.{ "mux", "quic://a:1" }, "/env.key");
255 try std.testing.expect(e == .quic);
256 try std.testing.expectEqualStrings("/env.key", e.quic.key.?);
257
258 // ...and the flag wins when both are there, because it is the more
259 // specific statement of intent.
260 const both = parseEnv(&.{ "mux", "quic://a:1", "--key", "/flag.key" }, "/env.key");
261 try std.testing.expectEqualStrings("/flag.key", both.quic.key.?);
262
263 // A key with no quic:// is ignored rather than refused: MUX_KEY_FILE
264 // exported in a shell must not break an ordinary local attach.
265 try std.testing.expect(parseEnv(&.{"mux"}, "/env.key") == .attach);
266 try std.testing.expect(parse(&.{ "mux", "--key", "/k" }) == .attach);
267 try std.testing.expect(parse(&.{ "mux", "--key", "/k", "vm1" }) == .host);
268 }
269
270 test "parseArgs: --quic-idle-ms parses, and refuses what ngtcp2 would invert" {
271 const t = parse(&.{ "mux", "quic://a:1", "--key", "/k", "--quic-idle-ms", "1500" });
272 try std.testing.expectEqual(@as(u32, 1500), t.quic.idle_ms);
273
274 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--quic-idle-ms", "0" }) == .usage_error);
275 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--quic-idle-ms", "soon" }) == .usage_error);
276 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--quic-idle-ms", "99999999999" }) == .usage_error);
277 try std.testing.expect(parse(&.{ "mux", "quic://a:1", "--key", "/k", "--quic-idle-ms" }) == .usage_error);
278 }
src/quic_client.zig
Old New
@@ -0,0 +1,483 @@
1 //! mux's QUIC client transport: one UDP socket, one connection, one
2 //! bidirectional stream of opaque bytes.
3 //!
4 //! The mirror of `quic_server.zig` and held to the same discipline: it knows
5 //! NOTHING about the frame protocol it carries. It hands the caller a byte
6 //! stream and takes bytes back; `client.zig` does the framing. That is what
7 //! keeps "the transport is a swap" a checkable claim rather than a slogan,
8 //! and it is why there is no `proto` import here.
9 //!
10 //! Three things in here are carried over from the daemon's side rather than
11 //! rediscovered, because they were paid for once already:
12 //!
13 //! 1. **ngtcp2 does not copy stream payload.** It keeps the vector it is
14 //! handed and re-reads those bytes to retransmit, so outbound bytes
15 //! must not move or be reused until the peer acknowledges them. Same
16 //! `Egress` ring as the listener, same invariant, same reason.
17 //! 2. **A blocked stream is not a dead one.** NGTCP2_ERR_STREAM_DATA_BLOCKED
18 //! is a documented return; abandoning the egress loop on it also
19 //! abandons the ACKs that would reopen the window.
20 //! 3. **Both flow-control windows get extended on consume.** Extending
21 //! only the stream's leaves the connection window to close instead,
22 //! and the DAEMON stalls a few hundred kilobytes in — a freeze with no
23 //! error on either side.
24 const std = @import("std");
25 const quic = @import("quic");
26
27 /// Shared with the listener rather than imported again: two @cImport blocks
28 /// over the same headers are two distinct type universes, and the egress
29 /// ring hands `ngtcp2_vec`s across this boundary.
30 const c = quic.c;
31
32 pub const Key = quic.Key;
33 pub const key_len = quic.key_len;
34
35 /// wolfSSL's PSK callback carries no user pointer, so the key has to be
36 /// reachable without one. A client process runs one connection at a time,
37 /// which makes this correct rather than merely convenient.
38 var g_key: ?Key = null;
39
40 fn pskClientCb(
41 _: ?*c.WOLFSSL,
42 _: [*c]const u8,
43 identity: [*c]u8,
44 id_max: c_uint,
45 key_out: [*c]u8,
46 key_max: c_uint,
47 ciphersuite: [*c][*c]const u8,
48 ) callconv(.c) c_uint {
49 const k = g_key orelse return 0;
50 if (id_max < 4 or key_max < key_len) return 0;
51 @memcpy(identity[0..4], "mux\x00");
52 @memcpy(key_out[0..key_len], &k.bytes);
53 if (ciphersuite) |cs| cs.* = quic.psk_ciphersuite;
54 return key_len;
55 }
56
57 fn getConnCb(ref: [*c]c.ngtcp2_crypto_conn_ref) callconv(.c) ?*c.ngtcp2_conn {
58 const self: *Client = @ptrCast(@alignCast(ref.*.user_data));
59 return self.conn;
60 }
61
62 fn randCb(dest: [*c]u8, destlen: usize, _: [*c]const c.ngtcp2_rand_ctx) callconv(.c) void {
63 std.crypto.random.bytes(dest[0..destlen]);
64 }
65
66 fn getNewCidCb(
67 _: ?*c.ngtcp2_conn,
68 cid: [*c]c.ngtcp2_cid,
69 token: [*c]c.ngtcp2_stateless_reset_token,
70 cidlen: usize,
71 _: ?*anyopaque,
72 ) callconv(.c) c_int {
73 std.crypto.random.bytes(cid.*.data[0..cidlen]);
74 cid.*.datalen = cidlen;
75 std.crypto.random.bytes(&token.*.data);
76 return 0;
77 }
78
79 fn handshakeCompletedCb(_: ?*c.ngtcp2_conn, ud: ?*anyopaque) callconv(.c) c_int {
80 const self: *Client = @ptrCast(@alignCast(ud.?));
81 self.handshake_done = true;
82 return 0;
83 }
84
85 /// The peer granted stream credit: open the one stream this transport uses.
86 /// It cannot be opened before the handshake, which is why this is a callback
87 /// rather than a line in `connect`.
88 fn extendStreamsCb(conn: ?*c.ngtcp2_conn, _: u64, ud: ?*anyopaque) callconv(.c) c_int {
89 const self: *Client = @ptrCast(@alignCast(ud.?));
90 if (self.stream_id == -1) {
91 var sid: i64 = -1;
92 if (c.ngtcp2_conn_open_bidi_stream(conn, &sid, null) == 0) self.stream_id = sid;
93 }
94 return 0;
95 }
96
97 fn recvStreamDataCb(
98 conn: ?*c.ngtcp2_conn,
99 _: u32,
100 stream_id: i64,
101 _: u64,
102 data: [*c]const u8,
103 datalen: usize,
104 ud: ?*anyopaque,
105 _: ?*anyopaque,
106 ) callconv(.c) c_int {
107 const self: *Client = @ptrCast(@alignCast(ud.?));
108
109 // BOTH windows, and the connection-level one is the half that is easy to
110 // forget: extend only the stream and the daemon stops sending a few
111 // hundred kilobytes into a session — a scrollback fetch, a big
112 // snapshot — with no error anywhere. It presents as a freeze.
113 _ = c.ngtcp2_conn_extend_max_stream_offset(conn, stream_id, datalen);
114 _ = c.ngtcp2_conn_extend_max_offset(conn, datalen);
115
116 if (datalen > 0) {
117 self.in.appendSlice(self.alloc, data[0..datalen]) catch {
118 self.dead = true;
119 return 0;
120 };
121 }
122 return 0;
123 }
124
125 fn ackedStreamDataCb(
126 _: ?*c.ngtcp2_conn,
127 _: i64,
128 _: u64,
129 datalen: u64,
130 ud: ?*anyopaque,
131 _: ?*anyopaque,
132 ) callconv(.c) c_int {
133 const self: *Client = @ptrCast(@alignCast(ud.?));
134 self.out.ack(@intCast(datalen));
135 return 0;
136 }
137
138 /// One connection to a muxd's QUIC listener.
139 pub const Client = struct {
140 alloc: std.mem.Allocator,
141 fd: std.posix.fd_t,
142 ssl_ctx: ?*c.WOLFSSL_CTX = null,
143 ssl: ?*c.WOLFSSL = null,
144 conn: ?*c.ngtcp2_conn = null,
145 conn_ref: c.ngtcp2_crypto_conn_ref = undefined,
146 remote: std.posix.sockaddr.storage = undefined,
147 remote_len: std.posix.socklen_t = 0,
148 local: std.posix.sockaddr.storage = undefined,
149 local_len: std.posix.socklen_t = 0,
150 stream_id: i64 = -1,
151 handshake_done: bool = false,
152 /// Set once this connection can never carry another byte: an idle
153 /// timeout, a protocol error, a peer that stopped answering. The caller
154 /// reads it as "the transport is gone" and reconnects.
155 dead: bool = false,
156 /// Outbound bytes, in the ring that does not move them until they are
157 /// acknowledged. See the module comment for why that matters.
158 out: quic.Egress,
159 /// Stream bytes the caller has not consumed yet.
160 in: std.ArrayList(u8) = .empty,
161
162 pub fn connect(
163 alloc: std.mem.Allocator,
164 addr: std.net.Address,
165 key: Key,
166 idle_ms: u32,
167 ) !*Client {
168 const fd = try std.posix.socket(
169 addr.any.family,
170 std.posix.SOCK.DGRAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC,
171 0,
172 );
173 errdefer std.posix.close(fd);
174
175 const self = try alloc.create(Client);
176 errdefer alloc.destroy(self);
177 const ring = try alloc.alloc(u8, quic.egress_cap);
178 errdefer alloc.free(ring);
179
180 self.* = .{
181 .alloc = alloc,
182 .fd = fd,
183 .out = .{ .buf = ring },
184 .remote_len = addr.getOsSockLen(),
185 .local_len = @sizeOf(std.posix.sockaddr.storage),
186 };
187 @memcpy(
188 std.mem.asBytes(&self.remote)[0..self.remote_len],
189 std.mem.asBytes(&addr.any)[0..self.remote_len],
190 );
191 // Connected UDP: the kernel filters out anything from another
192 // address, which is one fewer thing this code has to check.
193 try std.posix.connect(fd, &addr.any, self.remote_len);
194 try std.posix.getsockname(fd, @ptrCast(&self.local), &self.local_len);
195
196 g_key = key;
197 try self.startTls();
198 try self.startConn(idle_ms);
199 self.drain();
200 return self;
201 }
202
203 fn startTls(self: *Client) !void {
204 if (c.wolfSSL_Init() != c.WOLFSSL_SUCCESS) return error.TlsInit;
205 const ctx = c.wolfSSL_CTX_new(c.wolfTLSv1_3_client_method()) orelse return error.TlsInit;
206 self.ssl_ctx = ctx;
207 if (c.ngtcp2_crypto_wolfssl_configure_client_context(ctx) != 0) return error.TlsInit;
208 c.wolfSSL_CTX_set_psk_client_tls13_callback(ctx, pskClientCb);
209 _ = c.wolfSSL_CTX_set_cipher_list(ctx, quic.psk_ciphersuite);
210
211 const ssl = c.wolfSSL_new(ctx) orelse return error.TlsInit;
212 self.ssl = ssl;
213 self.conn_ref = .{ .get_conn = getConnCb, .user_data = self };
214 _ = c.wolfSSL_set_app_data(ssl, &self.conn_ref);
215 _ = c.wolfSSL_UseALPN(
216 ssl,
217 @constCast(quic.alpn[1..].ptr),
218 quic.alpn.len - 1,
219 c.WOLFSSL_ALPN_FAILED_ON_MISMATCH,
220 );
221 }
222
223 fn startConn(self: *Client, idle_ms: u32) !void {
224 var dcid: c.ngtcp2_cid = undefined;
225 dcid.datalen = 16;
226 std.crypto.random.bytes(dcid.data[0..16]);
227 var scid: c.ngtcp2_cid = undefined;
228 scid.datalen = 8;
229 std.crypto.random.bytes(scid.data[0..8]);
230
231 var cbs: c.ngtcp2_callbacks = std.mem.zeroes(c.ngtcp2_callbacks);
232 cbs.client_initial = c.ngtcp2_crypto_client_initial_cb;
233 cbs.recv_crypto_data = c.ngtcp2_crypto_recv_crypto_data_cb;
234 cbs.encrypt = c.ngtcp2_crypto_encrypt_cb;
235 cbs.decrypt = c.ngtcp2_crypto_decrypt_cb;
236 cbs.hp_mask = c.ngtcp2_crypto_hp_mask_cb;
237 // The listener answers every fresh Initial with a Retry, so a client
238 // that cannot process one never gets past its first flight.
239 cbs.recv_retry = c.ngtcp2_crypto_recv_retry_cb;
240 cbs.update_key = c.ngtcp2_crypto_update_key_cb;
241 cbs.delete_crypto_aead_ctx = c.ngtcp2_crypto_delete_crypto_aead_ctx_cb;
242 cbs.delete_crypto_cipher_ctx = c.ngtcp2_crypto_delete_crypto_cipher_ctx_cb;
243 cbs.get_path_challenge_data = c.ngtcp2_crypto_get_path_challenge_data_cb;
244 cbs.version_negotiation = c.ngtcp2_crypto_version_negotiation_cb;
245 cbs.rand = randCb;
246 cbs.get_new_connection_id2 = getNewCidCb;
247 cbs.handshake_completed = handshakeCompletedCb;
248 cbs.extend_max_local_streams_bidi = extendStreamsCb;
249 cbs.recv_stream_data = recvStreamDataCb;
250 cbs.acked_stream_data_offset = ackedStreamDataCb;
251
252 var settings: c.ngtcp2_settings = undefined;
253 c.ngtcp2_settings_default_versioned(c.NGTCP2_SETTINGS_VERSION, &settings);
254 settings.initial_ts = quic.timestampNs();
255
256 var params: c.ngtcp2_transport_params = undefined;
257 c.ngtcp2_transport_params_default_versioned(c.NGTCP2_TRANSPORT_PARAMS_VERSION, &params);
258 params.initial_max_streams_bidi = 4;
259 params.initial_max_stream_data_bidi_local = 256 * 1024;
260 params.initial_max_stream_data_bidi_remote = 256 * 1024;
261 params.initial_max_data = 1024 * 1024;
262 // Tunable for the same reason as the daemon's: the reconnect loop
263 // has to see a dead transport on a schedule a test can wait for.
264 params.max_idle_timeout = @as(u64, idle_ms) * 1_000_000;
265
266 var path: c.ngtcp2_path = .{
267 .local = .{ .addr = @ptrCast(&self.local), .addrlen = self.local_len },
268 .remote = .{ .addr = @ptrCast(&self.remote), .addrlen = self.remote_len },
269 .user_data = null,
270 };
271 var conn: ?*c.ngtcp2_conn = null;
272 if (c.ngtcp2_conn_client_new_versioned(
273 &conn,
274 &dcid,
275 &scid,
276 &path,
277 c.NGTCP2_PROTO_VER_V1,
278 c.NGTCP2_CALLBACKS_VERSION,
279 &cbs,
280 c.NGTCP2_SETTINGS_VERSION,
281 &settings,
282 c.NGTCP2_TRANSPORT_PARAMS_VERSION,
283 &params,
284 null,
285 self,
286 ) != 0) return error.ConnInit;
287 self.conn = conn;
288 c.ngtcp2_conn_set_tls_native_handle(conn, self.ssl);
289 // Silence is not death: a terminal nobody is typing into must not be
290 // dropped at the idle timeout. A third of it, matching the daemon.
291 c.ngtcp2_conn_set_keep_alive_timeout(conn, keepAliveNs(idle_ms));
292 }
293
294 pub fn deinit(self: *Client) void {
295 self.in.deinit(self.alloc);
296 self.out.deinit(self.alloc);
297 if (self.conn) |cn| c.ngtcp2_conn_del(cn);
298 if (self.ssl) |s| c.wolfSSL_free(s);
299 if (self.ssl_ctx) |x| c.wolfSSL_CTX_free(x);
300 std.posix.close(self.fd);
301 self.alloc.destroy(self);
302 }
303
304 /// The descriptor to poll. Readable does NOT mean "a frame is waiting":
305 /// it means a datagram arrived, which may be an ack, a handshake flight,
306 /// or part of a frame. See `Incoming` in client.zig.
307 pub fn pollFd(self: *const Client) std.posix.fd_t {
308 return self.fd;
309 }
310
311 /// Ready to carry the caller's bytes: handshake finished AND the stream
312 /// open. Both, because a handshake without a stream has nowhere to put
313 /// them and would silently hold everything in the ring.
314 pub fn isReady(self: *const Client) bool {
315 return self.handshake_done and self.stream_id != -1 and !self.dead;
316 }
317
318 /// Milliseconds until ngtcp2 next wants servicing, capped. Feeds the
319 /// client's existing poll timeout — no timer fd, same as the daemon.
320 pub fn timeoutMs(self: *Client, cap_ms: i32) i32 {
321 const conn = self.conn orelse return cap_ms;
322 const expiry = c.ngtcp2_conn_get_expiry(conn);
323 if (expiry == std.math.maxInt(u64)) return cap_ms;
324 const now = quic.timestampNs();
325 if (expiry <= now) return 0;
326 return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000));
327 }
328
329 /// One service pass: take what the socket has, run whatever timers are
330 /// due, push whatever egress is ready. Safe to call at any time.
331 pub fn pump(self: *Client) void {
332 if (self.dead) return;
333 self.readable();
334 self.tick();
335 self.drain();
336 }
337
338 fn readable(self: *Client) void {
339 var buf: [65536]u8 = undefined;
340 while (true) {
341 const n = std.posix.recv(self.fd, &buf, 0) catch |err| switch (err) {
342 error.WouldBlock => return,
343 // ECONNREFUSED on a connected UDP socket means the port is
344 // not there: an ICMP unreachable came back. That is a dead
345 // transport, not a blip to keep polling.
346 error.ConnectionRefused => {
347 self.dead = true;
348 return;
349 },
350 else => return,
351 };
352 if (n == 0) continue;
353 const conn = self.conn orelse return;
354 var path: c.ngtcp2_path = .{
355 .local = .{ .addr = @ptrCast(&self.local), .addrlen = self.local_len },
356 .remote = .{ .addr = @ptrCast(&self.remote), .addrlen = self.remote_len },
357 .user_data = null,
358 };
359 var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
360 if (c.ngtcp2_conn_read_pkt(conn, &path, &pi, &buf, n, quic.timestampNs()) != 0) {
361 self.dead = true;
362 return;
363 }
364 }
365 }
366
367 fn tick(self: *Client) void {
368 const conn = self.conn orelse return;
369 const now = quic.timestampNs();
370 if (c.ngtcp2_conn_get_expiry(conn) > now) return;
371 // An idle timeout arrives here, which is how the reconnect loop
372 // learns the daemon stopped answering.
373 if (c.ngtcp2_conn_handle_expiry(conn, now) != 0) self.dead = true;
374 }
375
376 fn drain(self: *Client) void {
377 const conn = self.conn orelse return;
378 var buf: [quic.max_udp]u8 = undefined;
379 var stream_blocked = false;
380 while (true) {
381 var ps: c.ngtcp2_path_storage = undefined;
382 c.ngtcp2_path_storage_zero(&ps);
383 var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 };
384 var wrote: c.ngtcp2_ssize = 0;
385
386 var vecs: [2]c.ngtcp2_vec = undefined;
387 var vcnt: usize = 0;
388 var sid: i64 = -1;
389 if (!stream_blocked and self.stream_id != -1) {
390 vcnt = self.out.vecs(&vecs);
391 if (vcnt > 0) sid = self.stream_id;
392 }
393
394 const n = c.ngtcp2_conn_writev_stream_versioned(
395 conn,
396 &ps.path,
397 c.NGTCP2_PKT_INFO_VERSION,
398 &pi,
399 &buf,
400 buf.len,
401 &wrote,
402 0,
403 sid,
404 if (vcnt > 0) &vecs else null,
405 vcnt,
406 quic.timestampNs(),
407 );
408 if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) {
409 // Documented, not fatal. Retry the same iteration carrying
410 // no stream data so ACKs and keepalives still leave: they
411 // are how the peer's window reopens, and a client that
412 // stopped here would stall exactly as the daemon did.
413 stream_blocked = true;
414 continue;
415 }
416 if (n < 0) {
417 self.dead = true;
418 return;
419 }
420 if (wrote > 0) self.out.took(@intCast(wrote));
421 if (n == 0) return;
422 _ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch return;
423 }
424 }
425
426 /// Take what fits and report how much. A short return is the caller's
427 /// signal to keep the rest and offer it again — the same contract the
428 /// daemon's sink obeys, and for the same reason: the ring is bounded, so
429 /// somebody has to hold the backlog and it should be somebody who can
430 /// see how big it is.
431 pub fn send(self: *Client, bytes: []const u8) usize {
432 if (self.dead or self.stream_id == -1) return 0;
433 const n = self.out.push(bytes);
434 if (n > 0) self.drain();
435 return n;
436 }
437
438 /// Stream bytes received and not yet consumed.
439 pub fn inbound(self: *const Client) []const u8 {
440 return self.in.items;
441 }
442
443 /// Drop `n` bytes off the front of the inbound buffer.
444 pub fn consume(self: *Client, n: usize) void {
445 const take = @min(n, self.in.items.len);
446 const rest = self.in.items.len - take;
447 std.mem.copyForwards(u8, self.in.items[0..rest], self.in.items[take..]);
448 self.in.shrinkRetainingCapacity(rest);
449 }
450 };
451
452 /// A third of the idle timeout, so two keepalives can go unanswered before
453 /// the connection is called dead. Never zero: ngtcp2 reads zero as
454 /// "disabled", which would restore the behaviour this exists to prevent.
455 fn keepAliveNs(idle_ms: u32) u64 {
456 return @max(1, @as(u64, idle_ms) / 3) * 1_000_000;
457 }
458
459 test "keepAlive: a third of the idle timeout, and never disabled" {
460 try std.testing.expectEqual(@as(u64, 5_000_000_000), keepAliveNs(15_000));
461 try std.testing.expectEqual(@as(u64, 500_000_000), keepAliveNs(1500));
462 // Zero would mean "no keepalive" to ngtcp2 — the opposite of what a
463 // small idle timeout is asking for.
464 try std.testing.expectEqual(@as(u64, 1_000_000), keepAliveNs(1));
465 try std.testing.expectEqual(@as(u64, 1_000_000), keepAliveNs(2));
466 }
467
468 test "Client.consume: takes from the front and keeps the rest" {
469 const alloc = std.testing.allocator;
470 var cl: Client = .{ .alloc = alloc, .fd = -1, .out = .{ .buf = &.{} } };
471 defer cl.in.deinit(alloc);
472
473 try cl.in.appendSlice(alloc, "abcdefgh");
474 cl.consume(3);
475 try std.testing.expectEqualStrings("defgh", cl.inbound());
476 cl.consume(0);
477 try std.testing.expectEqualStrings("defgh", cl.inbound());
478 // Consuming more than is there is not an error: a frame walk that asked
479 // for a payload it had already been handed would otherwise corrupt the
480 // buffer rather than say so.
481 cl.consume(99);
482 try std.testing.expectEqualStrings("", cl.inbound());
483 }
src/quic_server.zig
Old New
@@ -13,7 +13,11 @@
13 //! for the integration assessment this implements. 13 //! for the integration assessment this implements.
14 const std = @import("std"); 14 const std = @import("std");
15 15
16 const c = @cImport({ 16 /// The C view of the QUIC stack. Exported because the client transport
17 /// shares it: two @cImport blocks over the same headers produce two
18 /// *distinct* Zig types, so a client with its own would find that
19 /// `ngtcp2_vec` is not `ngtcp2_vec`. One import, one type universe.
20 pub const c = @cImport({
17 @cInclude("ngtcp2/ngtcp2.h"); 21 @cInclude("ngtcp2/ngtcp2.h");
18 @cInclude("ngtcp2/ngtcp2_crypto.h"); 22 @cInclude("ngtcp2/ngtcp2_crypto.h");
19 @cInclude("ngtcp2/ngtcp2_crypto_wolfssl.h"); 23 @cInclude("ngtcp2/ngtcp2_crypto_wolfssl.h");
@@ -233,7 +237,7 @@ const max_conns = 16;
233 /// primary CID is matched separately, a cache that ever did fill would 237 /// primary CID is matched separately, a cache that ever did fill would
234 /// degrade to the old primary-only behaviour rather than misroute. 238 /// degrade to the old primary-only behaviour rather than misroute.
235 const max_cids = 16; 239 const max_cids = 16;
236 const max_udp = 1452; // conservative IPv4 datagram that avoids fragmenting 240 pub const max_udp = 1452; // conservative IPv4 datagram that avoids fragmenting
237 241
238 /// wolfSSL's PSK callbacks carry no user pointer, so the key has to be 242 /// wolfSSL's PSK callbacks carry no user pointer, so the key has to be
239 /// reachable without one. A daemon runs a single listener, which makes a 243 /// reachable without one. A daemon runs a single listener, which makes a
@@ -262,15 +266,15 @@ fn pskServerCb(
262 return key_len; 266 return key_len;
263 } 267 }
264 268
265 const psk_identity: [*:0]const u8 = "mux"; 269 pub const psk_identity: [*:0]const u8 = "mux";
266 const psk_ciphersuite: [*:0]const u8 = "TLS13-AES128-GCM-SHA256"; 270 pub const psk_ciphersuite: [*:0]const u8 = "TLS13-AES128-GCM-SHA256";
267 const alpn = "\x03mux"; 271 pub const alpn = "\x03mux";
268 272
269 /// How many outbound bytes one connection may hold. Sized to the stream 273 /// 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 274 /// 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 275 /// 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. 276 /// right place for the backlog to sit and be judged.
273 const egress_cap = 256 * 1024; 277 pub const egress_cap = 256 * 1024;
274 278
275 /// Outbound stream bytes, in a ring that never moves a byte once written. 279 /// Outbound stream bytes, in a ring that never moves a byte once written.
276 /// 280 ///
@@ -289,7 +293,7 @@ const egress_cap = 256 * 1024;
289 /// it.** `head` advances only from `acked_stream_data_offset`; a write can 293 /// 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 294 /// 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. 295 /// half of the point — a full ring is the backpressure signal.
292 const Egress = struct { 296 pub const Egress = struct {
293 buf: []u8, 297 buf: []u8,
294 /// The oldest byte the peer has not acknowledged. 298 /// The oldest byte the peer has not acknowledged.
295 head: usize = 0, 299 head: usize = 0,
@@ -299,19 +303,19 @@ const Egress = struct {
299 /// The tail of `held` that ngtcp2 has not taken yet. 303 /// The tail of `held` that ngtcp2 has not taken yet.
300 unsent: usize = 0, 304 unsent: usize = 0,
301 305
302 fn deinit(self: *Egress, alloc: std.mem.Allocator) void { 306 pub fn deinit(self: *Egress, alloc: std.mem.Allocator) void {
303 alloc.free(self.buf); 307 alloc.free(self.buf);
304 self.* = .{ .buf = &.{} }; 308 self.* = .{ .buf = &.{} };
305 } 309 }
306 310
307 fn freeSpace(self: *const Egress) usize { 311 pub fn freeSpace(self: *const Egress) usize {
308 return self.buf.len - self.held; 312 return self.buf.len - self.held;
309 } 313 }
310 314
311 /// Take what fits and report how much that was. A short return is not an 315 /// 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 316 /// error, it is the whole mechanism: the caller keeps the remainder and
313 /// is thereby the one holding — and bounding — the backlog. 317 /// is thereby the one holding — and bounding — the backlog.
314 fn push(self: *Egress, bytes: []const u8) usize { 318 pub fn push(self: *Egress, bytes: []const u8) usize {
315 const n = @min(bytes.len, self.freeSpace()); 319 const n = @min(bytes.len, self.freeSpace());
316 if (n == 0) return 0; 320 if (n == 0) return 0;
317 const start = (self.head + self.held) % self.buf.len; 321 const start = (self.head + self.held) % self.buf.len;
@@ -326,7 +330,7 @@ const Egress = struct {
326 /// The unsent region as up to two vectors — two when it wraps, which is 330 /// 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 331 /// the price of never moving a byte, and ngtcp2 takes a vector array
328 /// precisely so that price is payable. 332 /// precisely so that price is payable.
329 fn vecs(self: *const Egress, out: *[2]c.ngtcp2_vec) usize { 333 pub fn vecs(self: *const Egress, out: *[2]c.ngtcp2_vec) usize {
330 if (self.unsent == 0) return 0; 334 if (self.unsent == 0) return 0;
331 const start = (self.head + (self.held - self.unsent)) % self.buf.len; 335 const start = (self.head + (self.held - self.unsent)) % self.buf.len;
332 const first = @min(self.unsent, self.buf.len - start); 336 const first = @min(self.unsent, self.buf.len - start);
@@ -338,7 +342,7 @@ const Egress = struct {
338 342
339 /// ngtcp2 took `n` bytes off the unsent region. They stay exactly where 343 /// ngtcp2 took `n` bytes off the unsent region. They stay exactly where
340 /// they are — it now has pointers to them. 344 /// they are — it now has pointers to them.
341 fn took(self: *Egress, n: usize) void { 345 pub fn took(self: *Egress, n: usize) void {
342 self.unsent -= @min(n, self.unsent); 346 self.unsent -= @min(n, self.unsent);
343 } 347 }
344 348
@@ -346,7 +350,7 @@ const Egress = struct {
346 /// as arriving "sequentially in increasing order of offset without any 350 /// as arriving "sequentially in increasing order of offset without any
347 /// overlap", so a running count IS the acknowledged prefix, and this is 351 /// overlap", so a running count IS the acknowledged prefix, and this is
348 /// the only thing that ever frees space. 352 /// the only thing that ever frees space.
349 fn ack(self: *Egress, n: usize) void { 353 pub fn ack(self: *Egress, n: usize) void {
350 const taken = @min(n, self.held - self.unsent); 354 const taken = @min(n, self.held - self.unsent);
351 self.head = (self.head + taken) % self.buf.len; 355 self.head = (self.head + taken) % self.buf.len;
352 self.held -= taken; 356 self.held -= taken;
@@ -1100,7 +1104,7 @@ fn keepAliveNs(idle_ms: u64) u64 {
1100 return @max(1, idle_ms / 3) * 1_000_000; 1104 return @max(1, idle_ms / 3) * 1_000_000;
1101 } 1105 }
1102 1106
1103 fn timestampNs() u64 { 1107 pub fn timestampNs() u64 {
1104 const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch return 0; 1108 const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch return 0;
1105 return @as(u64, @intCast(ts.sec)) * 1_000_000_000 + @as(u64, @intCast(ts.nsec)); 1109 return @as(u64, @intCast(ts.sec)) * 1_000_000_000 + @as(u64, @intCast(ts.nsec));
1106 } 1110 }
src/server.zig
Old New
@@ -991,14 +991,23 @@ pub const Server = struct {
991 fn handleFrame(self: *Server, i: usize, frame: proto.Frame) void { 991 fn handleFrame(self: *Server, i: usize, frame: proto.Frame) void {
992 switch (frame.type) { 992 switch (frame.type) {
993 .attach => { 993 .attach => {
994 // Re-attach over a connection that is already attached. Its 994 // This used to snapshot unconditionally, on the reasoning
995 // have_seq would be serviceable, but a client attaching 995 // that a client attaching twice over one connection is rare
996 // twice over one connection is rare enough that a second 996 // enough not to deserve a second resync path. That was true
997 // resync path isn't worth carrying — snapshot it (which, 997 // while every client began life as an observer and did its
998 // like any resize, everyone else hears about too). 998 // FIRST attach from there — and it stopped being true the
999 // moment QUIC clients existed, because a QUIC connection is
1000 // promoted to a client slot when its handshake completes, so
1001 // its first attach arrives HERE. Every QUIC attach was
1002 // therefore snapshot-served, and a reconnecting QUIC client
1003 // could never resume from a delta no matter what it held.
1004 // Found by the reconnect scenario in e2e, which asserts the
1005 // counter rather than the rendering.
999 const req = proto.decodeAttach(frame.payload) catch return; 1006 const req = proto.decodeAttach(frame.payload) catch return;
1000 if (self.applySize(req.cols, req.rows)) self.recordSize(i); 1007 const size_changed = (req.cols != self.colsNow() or req.rows != self.rowsNow());
1001 self.resyncSnapshot(); 1008 const applied = self.applySize(req.cols, req.rows);
1009 if (applied) self.recordSize(i);
1010 self.sendResync(i, req.have_seq, req.have_epoch, size_changed and applied);
1002 }, 1011 },
1003 .resize => { 1012 .resize => {
1004 // Latest wins: whoever resized last sets the grid, and the 1013 // Latest wins: whoever resized last sets the grid, and the
test/e2e.sh
Old New
@@ -54,7 +54,7 @@ cleanup() {
54 rm -f "$SOCK" "$SOCK2" "$SOCK3" "$SOCK4" "$SOCK4.second" "$QKEY" "$QKEY.bad" \ 54 rm -f "$SOCK" "$SOCK2" "$SOCK3" "$SOCK4" "$SOCK4.second" "$QKEY" "$QKEY.bad" \
55 "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" \ 55 "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" \
56 "$OUT.b" "$OUT.via" "$OUT.dead" "$OUT.abort" "$OUT.m7" "$OUT.m7b" \ 56 "$OUT.b" "$OUT.via" "$OUT.dead" "$OUT.abort" "$OUT.m7" "$OUT.m7b" \
57 "$OUT.q" 57 "$OUT.q" "$OUT.qc" "$OUT.qr" "$OUT.qa" "$QKEY.wrong"
58 } 58 }
59 trap cleanup EXIT INT TERM 59 trap cleanup EXIT INT TERM
60 60
@@ -393,8 +393,130 @@ grep -q "quic-flags-ok" "$OUT.q" || {
393 echo "e2e FAIL: --quic daemon's grid missing output"; exit 1; 393 echo "e2e FAIL: --quic daemon's grid missing output"; exit 1;
394 } 394 }
395 kill -0 "$D4PID" || { echo "e2e FAIL: --quic daemon died"; exit 1; } 395 kill -0 "$D4PID" || { echo "e2e FAIL: --quic daemon died"; exit 1; }
396
397 # --- M8 Task 3: the client speaks quic://. Same daemon, new transport.
398
399 # 1. Attach, run something, detach. The whole point of the milestone in one
400 # scenario: the wire protocol did not change, so this must behave exactly
401 # as the socket client does.
402 set +e
403 { printf 'printf "quic-%%s\\n" attach-ok\n'; sleep 2; printf '\034'; } | \
404 timeout 30 "$MUX" "quic://127.0.0.1:$QPORT" --key "$QKEY" > "$OUT.qc" 2>&1
405 RC=$?
406 set -e
407 [ "$RC" -eq 0 ] || {
408 echo "e2e FAIL: quic:// client exited $RC (want 0)"; cat "$OUT.qc"; exit 1;
409 }
410 grep -q "quic-attach-ok" "$OUT.qc" || {
411 echo "e2e FAIL: quic:// client render missing output"; cat "$OUT.qc"; exit 1;
412 }
413 "$MUXD" dump --sock "$SOCK4" | grep -q "quic-attach-ok" || {
414 echo "e2e FAIL: daemon grid missing the quic:// client's output"; exit 1;
415 }
416
417 # 2. The key is checked, and a wrong one is refused loudly rather than
418 # retried forever. Nothing was ever established, so the reconnect loop
419 # must not engage — that is the never-established gate, over QUIC.
420 head -c 32 /dev/urandom > "$QKEY.wrong"
421 chmod 600 "$QKEY.wrong"
422 set +e
423 : | timeout 30 "$MUX" "quic://127.0.0.1:$QPORT" --key "$QKEY.wrong" \
424 --quic-idle-ms 1500 > "$OUT.qc" 2>&1
425 RC=$?
426 set -e
427 [ "$RC" -eq 1 ] || {
428 echo "e2e FAIL: wrong-key quic client exited $RC (want 1; 124 means it retried)"
429 cat "$OUT.qc"; exit 1;
430 }
431 grep -q "did not answer" "$OUT.qc" || {
432 echo "e2e FAIL: wrong-key quic client said nothing useful:"; cat "$OUT.qc"; exit 1;
433 }
434
435 # 3. The abort key works DURING a handshake, not just after it. waitReady
436 # runs inside Transport.open, after drainStdinForQuit has returned, so
437 # watching only the socket left nothing looking for Ctrl-\ for as long as
438 # the handshake bound allows — and during a reconnect the terminal is in
439 # raw mode, where Ctrl-\ is the only way out. A deliberately unreachable
440 # port makes the handshake run its full length if nothing interrupts it.
441 QDEAD=$(( QPORT + 1 ))
442 # A fifo rather than a pipeline, so what is timed is the CLIENT's exit and
443 # not how long the writer happened to hang around afterwards.
444 QFIFO="${TMPDIR:-/tmp}/mux-e2e-abort-fifo-$$"
445 mkfifo "$QFIFO"
446 ( sleep 0.3; printf '\034'; sleep 20 ) > "$QFIFO" &
447 QWPID=$!
448 QT0=$(date +%s%N)
449 set +e
450 timeout 30 "$MUX" "quic://127.0.0.1:$QDEAD" --key "$QKEY" \
451 --quic-idle-ms 15000 < "$QFIFO" > "$OUT.qa" 2>&1
452 RC=$?
453 set -e
454 QT1=$(date +%s%N)
455 kill "$QWPID" 2>/dev/null || true
456 rm -f "$QFIFO"
457 QMS=$(( (QT1 - QT0) / 1000000 ))
458 [ "$RC" -eq 0 ] || {
459 echo "e2e FAIL: aborted quic handshake exited $RC (want 0; 124 means Ctrl-\ went unheard)"
460 cat "$OUT.qa"; exit 1;
461 }
462 # Generous against the 15000ms bound but far below it: the point is that the
463 # abort is answered on the user's schedule, not the handshake's. Measured at
464 # ~200ms; anything under 5s proves stdin was being watched.
465 [ "$QMS" -lt 5000 ] || {
466 echo "e2e FAIL: abort during handshake took ${QMS}ms (want well under the 15000ms bound)"
467 exit 1;
468 }
469
470 # 4. Reconnect over QUIC, with the resume kind asserted rather than assumed.
471 # The tear is a SIGSTOP held past the client's idle timeout: the daemon
472 # stops answering, the client declares the transport dead and reconnects.
473 # Deterministic, needs no privileges, and unlike kill -9 it leaves the
474 # session alive so the resume can be a DELTA rather than a fresh snapshot.
475 SNAPS_BEFORE=$("$MUXD" stats --sock "$SOCK4" | sed -n 's/.*snapshots=\([0-9]*\).*/\1/p')
476 [ -n "$SNAPS_BEFORE" ] || { echo "e2e FAIL: could not read snapshots before the tear"; exit 1; }
477
478 set +e
479 { printf 'printf "quic-%%s\\n" pre-tear\n'; sleep 12; printf '\034'; } | \
480 timeout 40 "$MUX" "quic://127.0.0.1:$QPORT" --key "$QKEY" \
481 --quic-idle-ms 1500 > "$OUT.qr" 2>&1 &
482 QRPID=$!
483 set -e
484 wait_for "$OUT.qr" "quic-pre-tear" 20 || {
485 echo "e2e FAIL: quic reconnect client never got its pre-tear marker"
486 cat "$OUT.qr"; exit 1;
487 }
488
489 # Held well past the 1500ms idle timeout, so the client cannot mistake it
490 # for a slow moment.
491 kill -STOP "$D4PID"
492 sleep 4
493 kill -CONT "$D4PID"
494
495 set +e
496 wait "$QRPID"
497 RC=$?
498 set -e
499 [ "$RC" -eq 0 ] || {
500 echo "e2e FAIL: quic client exited $RC across a tear (want 0)"; cat "$OUT.qr"; exit 1;
501 }
502 grep -q "quic-pre-tear" "$OUT.qr" || {
503 echo "e2e FAIL: quic client lost its session across the tear"; cat "$OUT.qr"; exit 1;
504 }
505
506 # The counter is the only witness to HOW the resume was served: a snapshot
507 # renders identically to a delta, so markers cannot tell them apart. One
508 # more snapshot is the fresh attach at the start of this scenario; a second
509 # would mean the reconnect was resynced from scratch instead of resumed.
510 SNAPS_AFTER=$("$MUXD" stats --sock "$SOCK4" | sed -n 's/.*snapshots=\([0-9]*\).*/\1/p')
511 [ -n "$SNAPS_AFTER" ] || { echo "e2e FAIL: could not read snapshots after the tear"; exit 1; }
512 [ "$((SNAPS_AFTER - SNAPS_BEFORE))" -eq 1 ] || {
513 echo "e2e FAIL: quic reconnect served $((SNAPS_AFTER - SNAPS_BEFORE)) snapshots (want 1:"
514 echo " the attach only; the resume itself must be delta-served)"
515 exit 1;
516 }
517
396 kill "$D4PID" 2>/dev/null || true 518 kill "$D4PID" 2>/dev/null || true
397 D4PID="" 519 D4PID=""
398 rm -f "$OUT.q" "$QKEY" "$QKEY.bad" 520 rm -f "$OUT.q" "$OUT.qc" "$OUT.qr" "$OUT.qa" "$QKEY" "$QKEY.bad" "$QKEY.wrong"
399 521
400 echo "e2e OK" 522 echo "e2e OK"