a73x

24397e0d

refactor: one owner for "the frame at the front of this buffer"

a73x   2026-08-29 10:01

Commit message
refactor: one owner for "the frame at the front of this buffer"

Three copies delimited a frame off a QUIC inbound buffer: the daemon's
`Server.takeFrame`, the quic arm of `client.Transport.readFrame` (which
hand-rolled the header check `delimitFrame` already is), and muxa's
`frameFrom` paired with `quic_client.consume`. The daemon's fn moves
verbatim to `protocol.zig` as `proto.takeFrame(alloc, *ArrayList(u8))`
and the other two call it; `quic_client.inbound`/`consume` had no caller
left and are gone — a reader reaches the pub `in` field as it reaches
`dead`.

Transport stays dumb: `protocol` still never enters quic_client,
quic_server or quic.zig.

muxa's three `frameFrom` tests move to protocol.zig against the new
shape, and gain the claim the old ones could not make — a null answer
consumes nothing, so the partial tail survives for the bytes that
complete it.

Product lines 16544 -> 16517.
src/cli/muxa.zig
Old New
@@ -526,18 +526,17 @@ const Conn = struct {
526 const q = &self.link.quic; 526 const q = &self.link.quic;
527 while (true) { 527 while (true) {
528 q.cl.pump(); 528 q.cl.pump();
529 while (try frameFrom(alloc, q.cl.inbound())) |got| { 529 while (try proto.takeFrame(alloc, &q.cl.in)) |frame| {
530 q.cl.consume(got.consumed); 530 if (frame.type == want) return frame;
531 if (got.frame.type == want) return got.frame; 531 defer frame.deinit(alloc);
532 defer got.frame.deinit(alloc); 532 // The socket arm's classification again rather than a shared
533 // The socket arm's rule again rather than a shared helper: 533 // helper: that arm answers `DaemonGone` on a closed stream
534 // the two arms own their frame differently (`got.frame` 534 // where this one answers `ConnectionLost`, so the walk that
535 // paired with `consume`, against a plain `frame`), so there 535 // reaches these two lines is not the same walk.
536 // is no cursor for one reader to hand the other. 536 if (frame.type == .snapshot) self.saw_snapshot = true;
537 if (got.frame.type == .snapshot) self.saw_snapshot = true; 537 if (frame.type == .exit_status) {
538 if (got.frame.type == .exit_status) {
539 if (!self.saw_snapshot) return error.AttachRefused; 538 if (!self.saw_snapshot) return error.AttachRefused;
540 self.session_exit = if (got.frame.payload.len >= 1) got.frame.payload[0] else null; 539 self.session_exit = if (frame.payload.len >= 1) frame.payload[0] else null;
541 return error.SessionExited; 540 return error.SessionExited;
542 } 541 }
543 } 542 }
@@ -574,84 +573,6 @@ const Conn = struct {
574 } 573 }
575 }; 574 };
576 575
577 /// Copies: the caller consumes the bytes it points at.
578 fn frameFrom(
579 alloc: std.mem.Allocator,
580 buf: []const u8,
581 ) !?struct { frame: proto.Frame, consumed: usize } {
582 const d = try proto.delimitFrame(buf) orelse return null;
583 const payload = try alloc.alloc(u8, d.payload.len);
584 errdefer alloc.free(payload);
585 @memcpy(payload, d.payload);
586 return .{
587 .frame = .{ .type = d.type, .payload = payload },
588 .consumed = d.consumed,
589 };
590 }
591
592 test "frameFrom: a partial tail is not a frame and not an error" {
593 const alloc = std.testing.allocator;
594
595 // Nothing, and less than a header: the two shapes a datagram that
596 // carried the start of a frame leaves behind.
597 try std.testing.expect(try frameFrom(alloc, "") == null);
598 try std.testing.expect(try frameFrom(alloc, &[_]u8{ 0x0a, 1, 0 }) == null);
599
600 // A whole header whose payload is still in flight. This is the case a
601 // blocking read would have sat on: the length is known, the bytes are
602 // not here, and the answer is to wait rather than to read.
603 const partial = [_]u8{ @intFromEnum(proto.MsgType.input), 4, 0, 0, 0, 'a', 'b' };
604 try std.testing.expect(try frameFrom(alloc, &partial) == null);
605
606 // The same bytes, completed.
607 const whole = [_]u8{ @intFromEnum(proto.MsgType.input), 4, 0, 0, 0, 'a', 'b', 'c', 'd' };
608 const got = (try frameFrom(alloc, &whole)).?;
609 defer got.frame.deinit(alloc);
610 try std.testing.expectEqual(proto.MsgType.input, got.frame.type);
611 try std.testing.expectEqualStrings("abcd", got.frame.payload);
612 try std.testing.expectEqual(@as(usize, 9), got.consumed);
613 }
614
615 test "frameFrom: two frames in one buffer, walked by consumed" {
616 const alloc = std.testing.allocator;
617 // What a single datagram routinely carries: the push we skip and the
618 // reply we asked for. A walk that stopped after one would leave the
619 // answer sitting in the buffer while the deadline ran out.
620 var buf: std.ArrayList(u8) = .empty;
621 defer buf.deinit(alloc);
622 try proto.appendFrame(&buf, alloc, .pty_mode, &[_]u8{0});
623 try proto.appendFrame(&buf, alloc, .status_reply, "xy");
624
625 const first = (try frameFrom(alloc, buf.items)).?;
626 defer first.frame.deinit(alloc);
627 try std.testing.expectEqual(proto.MsgType.pty_mode, first.frame.type);
628
629 const second = (try frameFrom(alloc, buf.items[first.consumed..])).?;
630 defer second.frame.deinit(alloc);
631 try std.testing.expectEqual(proto.MsgType.status_reply, second.frame.type);
632 try std.testing.expectEqualStrings("xy", second.frame.payload);
633 try std.testing.expectEqual(buf.items.len, first.consumed + second.consumed);
634
635 // An empty payload is a frame like any other — `status_req` and
636 // `detach` are nothing else — and must not read as "nothing yet".
637 var empty: std.ArrayList(u8) = .empty;
638 defer empty.deinit(alloc);
639 try proto.appendFrame(&empty, alloc, .detach, "");
640 const none = (try frameFrom(alloc, empty.items)).?;
641 defer none.frame.deinit(alloc);
642 try std.testing.expectEqual(@as(usize, proto.frame_header_len), none.consumed);
643 }
644
645 test "frameFrom: a length no frame can carry is refused, not allocated" {
646 const alloc = std.testing.allocator;
647 // The peer chose this number. Reading on would mean allocating against
648 // it; the daemon's own walk refuses the same bound the same way.
649 var hdr: [proto.frame_header_len]u8 = undefined;
650 hdr[0] = @intFromEnum(proto.MsgType.input);
651 std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
652 try std.testing.expectError(error.FrameTooLarge, frameFrom(alloc, &hdr));
653 }
654
655 test "graceMs: flat over a socket, RTT-derived over QUIC, and capped" { 576 test "graceMs: flat over a socket, RTT-derived over QUIC, and capped" {
656 const alloc = std.testing.allocator; 577 const alloc = std.testing.allocator;
657 const local = Conn{ .link = .{ .fd = -1 }, .alloc = alloc }; 578 const local = Conn{ .link = .{ .fd = -1 }, .alloc = alloc };
src/client/client.zig
Old New
@@ -657,19 +657,13 @@ pub const Transport = struct {
657 .quic => |cl| { 657 .quic => |cl| {
658 // Death is checked after the pump, so bytes that arrived in the 658 // Death is checked after the pump, so bytes that arrived in the
659 // same pass as the close are still delivered before the tear. 659 // same pass as the close are still delivered before the tear.
660 const buf = cl.inbound(); 660 const got = proto.takeFrame(alloc, &cl.in) catch |err| switch (err) {
661 if (buf.len < 5) return if (cl.dead) .closed else .incomplete; 661 // Not a transport event, and it stays loud.
662 const len = std.mem.readInt(u32, buf[1..5], .little); 662 error.OutOfMemory => return err,
663 if (len > proto.max_payload) return .closed; 663 else => return .closed,
664 if (buf.len < 5 + len) return if (cl.dead) .closed else .incomplete;
665 const payload = try alloc.alloc(u8, len);
666 @memcpy(payload, buf[5 .. 5 + len]);
667 const frame: proto.Frame = .{
668 .type = @enumFromInt(buf[0]),
669 .payload = payload,
670 }; 664 };
671 cl.consume(5 + len); 665 if (got) |frame| return .{ .frame = frame };
672 return .{ .frame = frame }; 666 return if (cl.dead) .closed else .incomplete;
673 }, 667 },
674 .fd, .pipe => { 668 .fd, .pipe => {
675 const frame = (proto.readFrame(alloc, self.conn.r) catch |err| switch (err) { 669 const frame = (proto.readFrame(alloc, self.conn.r) catch |err| switch (err) {
src/client/quic_client.zig
Old New
@@ -410,35 +410,8 @@ pub const Client = struct {
410 if (n > 0) self.drain(); 410 if (n > 0) self.drain();
411 return n; 411 return n;
412 } 412 }
413
414 /// Stream bytes received and not yet consumed.
415 pub fn inbound(self: *const Client) []const u8 {
416 return self.in.items;
417 }
418
419 /// Drop `n` bytes off the front of the inbound buffer.
420 pub fn consume(self: *Client, n: usize) void {
421 self.in.replaceRangeAssumeCapacity(0, @min(n, self.in.items.len), &.{});
422 }
423 }; 413 };
424 414
425 test "Client.consume: takes from the front and keeps the rest" {
426 const alloc = std.testing.allocator;
427 var cl: Client = .{ .alloc = alloc, .fd = -1, .out = .{ .buf = &.{} } };
428 defer cl.in.deinit(alloc);
429
430 try cl.in.appendSlice(alloc, "abcdefgh");
431 cl.consume(3);
432 try std.testing.expectEqualStrings("defgh", cl.inbound());
433 cl.consume(0);
434 try std.testing.expectEqualStrings("defgh", cl.inbound());
435 // Consuming more than is there is not an error: a frame walk that asked
436 // for a payload it had already been handed would otherwise corrupt the
437 // buffer rather than say so.
438 cl.consume(99);
439 try std.testing.expectEqualStrings("", cl.inbound());
440 }
441
442 // Plain, not recursive: this module reaches the QUIC stack's @cImport, and 415 // Plain, not recursive: this module reaches the QUIC stack's @cImport, and
443 // a recursive walk would force-analyze the entire wolfSSL/ngtcp2 namespace. 416 // a recursive walk would force-analyze the entire wolfSSL/ngtcp2 namespace.
444 test { 417 test {
src/engine/protocol.zig
Old New
@@ -103,6 +103,21 @@ pub fn delimitFrame(buf: []const u8) !?Delimited {
103 }; 103 };
104 } 104 }
105 105
106 /// The frame at the front of `buf`, or null while only part of one is
107 /// here — the ordinary state of a byte stream, never an error. An error
108 /// is a length no frame can carry, and every caller answers it the same
109 /// way: it has a connection, and drops it.
110 pub fn takeFrame(alloc: std.mem.Allocator, buf: *std.ArrayList(u8)) !?Frame {
111 const d = (try delimitFrame(buf.items)) orelse return null;
112 // Copied out and `buf` shifted BEFORE the caller dispatches, because
113 // a handler can reallocate `buf` under a slice into it; the payload
114 // is then the caller's frame to free.
115 const payload = try alloc.alloc(u8, d.payload.len);
116 @memcpy(payload, d.payload);
117 buf.replaceRangeAssumeCapacity(0, d.consumed, &.{});
118 return .{ .type = d.type, .payload = payload };
119 }
120
106 pub fn writeFrame(fd: std.posix.fd_t, t: MsgType, payload: []const u8) !void { 121 pub fn writeFrame(fd: std.posix.fd_t, t: MsgType, payload: []const u8) !void {
107 var hdr: [5]u8 = undefined; 122 var hdr: [5]u8 = undefined;
108 hdr[0] = @intFromEnum(t); 123 hdr[0] = @intFromEnum(t);
@@ -1203,6 +1218,81 @@ test "appendFrame concatenates frames the way a queue would" {
1203 try std.testing.expectEqualSlices(u8, &.{7}, f2.payload); 1218 try std.testing.expectEqualSlices(u8, &.{7}, f2.payload);
1204 } 1219 }
1205 1220
1221 test "takeFrame: a partial tail is not a frame and not an error" {
1222 const alloc = std.testing.allocator;
1223 var buf: std.ArrayList(u8) = .empty;
1224 defer buf.deinit(alloc);
1225
1226 // Nothing, and less than a header: the two shapes a datagram that
1227 // carried the start of a frame leaves behind.
1228 try std.testing.expect(try takeFrame(alloc, &buf) == null);
1229 try buf.appendSlice(alloc, &[_]u8{ 0x0a, 1, 0 });
1230 try std.testing.expect(try takeFrame(alloc, &buf) == null);
1231
1232 // A whole header whose payload is still in flight. This is the case a
1233 // blocking read would have sat on: the length is known, the bytes are
1234 // not here, and the answer is to wait rather than to read.
1235 buf.clearRetainingCapacity();
1236 const partial = [_]u8{ @intFromEnum(MsgType.input), 4, 0, 0, 0, 'a', 'b' };
1237 try buf.appendSlice(alloc, &partial);
1238 try std.testing.expect(try takeFrame(alloc, &buf) == null);
1239 // A null answer consumed nothing: the partial tail is still there,
1240 // byte for byte, for the bytes that complete it.
1241 try std.testing.expectEqualSlices(u8, &partial, buf.items);
1242
1243 // The same bytes, completed.
1244 try buf.appendSlice(alloc, "cd");
1245 const got = (try takeFrame(alloc, &buf)).?;
1246 defer got.deinit(alloc);
1247 try std.testing.expectEqual(MsgType.input, got.type);
1248 try std.testing.expectEqualStrings("abcd", got.payload);
1249 try std.testing.expectEqual(@as(usize, 0), buf.items.len);
1250 }
1251
1252 test "takeFrame: two frames in one buffer, walked one call at a time" {
1253 const alloc = std.testing.allocator;
1254 // What a single datagram routinely carries: the push we skip and the
1255 // reply we asked for. A walk that stopped after one would leave the
1256 // answer sitting in the buffer while the deadline ran out.
1257 var buf: std.ArrayList(u8) = .empty;
1258 defer buf.deinit(alloc);
1259 try appendFrame(&buf, alloc, .pty_mode, &[_]u8{0});
1260 try appendFrame(&buf, alloc, .status_reply, "xy");
1261
1262 const first = (try takeFrame(alloc, &buf)).?;
1263 defer first.deinit(alloc);
1264 try std.testing.expectEqual(MsgType.pty_mode, first.type);
1265
1266 const second = (try takeFrame(alloc, &buf)).?;
1267 defer second.deinit(alloc);
1268 try std.testing.expectEqual(MsgType.status_reply, second.type);
1269 try std.testing.expectEqualStrings("xy", second.payload);
1270 try std.testing.expectEqual(@as(usize, 0), buf.items.len);
1271
1272 // An empty payload is a frame like any other — `status_req` and
1273 // `detach` are nothing else — and must not read as "nothing yet".
1274 var empty: std.ArrayList(u8) = .empty;
1275 defer empty.deinit(alloc);
1276 try appendFrame(&empty, alloc, .detach, "");
1277 const none = (try takeFrame(alloc, &empty)).?;
1278 defer none.deinit(alloc);
1279 try std.testing.expectEqual(MsgType.detach, none.type);
1280 try std.testing.expectEqual(@as(usize, 0), empty.items.len);
1281 }
1282
1283 test "takeFrame: a length no frame can carry is refused, not allocated" {
1284 const alloc = std.testing.allocator;
1285 // The peer chose this number. Reading on would mean allocating against
1286 // it; every caller answers the refusal by dropping the connection.
1287 var buf: std.ArrayList(u8) = .empty;
1288 defer buf.deinit(alloc);
1289 try buf.append(alloc, @intFromEnum(MsgType.input));
1290 var len: [4]u8 = undefined;
1291 std.mem.writeInt(u32, &len, max_payload + 1, .little);
1292 try buf.appendSlice(alloc, &len);
1293 try std.testing.expectError(error.FrameTooLarge, takeFrame(alloc, &buf));
1294 }
1295
1206 test "scrollback request encode/decode round trip" { 1296 test "scrollback request encode/decode round trip" {
1207 const req = try decodeScrollbackReq(&encodeScrollbackReq(70000, 24)); 1297 const req = try decodeScrollbackReq(&encodeScrollbackReq(70000, 24));
1208 try std.testing.expectEqual(@as(u32, 70000), req.start); 1298 try std.testing.expectEqual(@as(u32, 70000), req.start);
src/server/server.zig
Old New
@@ -1868,21 +1868,6 @@ pub const Server = struct {
1868 } 1868 }
1869 } 1869 }
1870 1870
1871 /// The frame at the front of `buf`, or null while only part of one is
1872 /// here — the ordinary state of a byte stream, never an error. An error
1873 /// is a length no frame can carry, and every caller answers it the same
1874 /// way: it has a connection, and drops it.
1875 fn takeFrame(alloc: std.mem.Allocator, buf: *std.ArrayList(u8)) !?proto.Frame {
1876 const d = (try proto.delimitFrame(buf.items)) orelse return null;
1877 // Copied out and `buf` shifted BEFORE the caller dispatches, because
1878 // a handler can reallocate `buf` under a slice into it; the payload
1879 // is then the caller's frame to free.
1880 const payload = try alloc.alloc(u8, d.payload.len);
1881 @memcpy(payload, d.payload);
1882 buf.replaceRangeAssumeCapacity(0, d.consumed, &.{});
1883 return .{ .type = d.type, .payload = payload };
1884 }
1885
1886 /// Feed bytes that arrived for client `i`, extracting whole frames as 1871 /// Feed bytes that arrived for client `i`, extracting whole frames as
1887 /// they complete. Every transport lands here — a socket read and a QUIC 1872 /// they complete. Every transport lands here — a socket read and a QUIC
1888 /// stream chunk from the shared UDP socket alike — so that from the 1873 /// stream chunk from the shared UDP socket alike — so that from the
@@ -1903,7 +1888,7 @@ pub const Server = struct {
1903 1888
1904 while (true) { 1889 while (true) {
1905 if (self.clients[i] == null) return; // a handler dropped it 1890 if (self.clients[i] == null) return; // a handler dropped it
1906 const frame = takeFrame(self.alloc, &self.clients[i].?.inbound) catch { 1891 const frame = proto.takeFrame(self.alloc, &self.clients[i].?.inbound) catch {
1907 self.dropClient(i); 1892 self.dropClient(i);
1908 return; 1893 return;
1909 } orelse return; // header or payload still coming 1894 } orelse return; // header or payload still coming
@@ -2499,7 +2484,7 @@ pub const Server = struct {
2499 fn drainObserver(self: *Server, i: usize) void { 2484 fn drainObserver(self: *Server, i: usize) void {
2500 for (0..max_observer_frames_per_pump) |_| { 2485 for (0..max_observer_frames_per_pump) |_| {
2501 const o = if (self.observers[i]) |*p| p else return; 2486 const o = if (self.observers[i]) |*p| p else return;
2502 const frame = takeFrame(self.alloc, &o.inbound) catch { 2487 const frame = proto.takeFrame(self.alloc, &o.inbound) catch {
2503 self.dropObserver(i); 2488 self.dropObserver(i);
2504 return; 2489 return;
2505 } orelse return; 2490 } orelse return;