a73x

e955b4f0

fix: a link.readFrame framing violation is an error, not a closed peer

a73x   2026-08-31 18:47

Commit message
fix: a link.readFrame framing violation is an error, not a closed peer

Every error but OutOfMemory folded into `.closed`, which is right for
client.Transport (a bad frame there means tear and redial) and wrong for
the caller Task 2 puts on top: dial.ask reports `.closed` as "no answer"
(null), so a daemon that replied to `mux d dump` with a corrupt header
would have read as a daemon that is not there. FrameTooLarge now comes
back as itself, and the two pins it protects move here from dial.zig,
where the await loop they describe no longer lives.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TWxBL1HBULH1ZwTNzzKTja
src/link.zig
Old New
@@ -145,29 +145,38 @@ pub const Link = union(enum) {
145 /// The next whole frame, if there is one. (Moved from 145 /// The next whole frame, if there is one. (Moved from
146 /// client.Transport.readFrame; see Incoming for why a missing frame is 146 /// client.Transport.readFrame; see Incoming for why a missing frame is
147 /// not automatically a dead transport.) 147 /// not automatically a dead transport.)
148 ///
149 /// Two errors are NOT transport events and are returned as themselves.
150 /// `error.OutOfMemory` is this side's failure, not the peer's. And a
151 /// header claiming more than `max_payload` is a peer that spoke and got
152 /// the framing wrong, which is a different fact from a peer that went
153 /// away: `dial.ask` reports a closed peer as "no answer" (null) and every
154 /// other error as an error, precisely so a corrupt reply to `mux d dump`
155 /// or `stats` is never reported as an absent daemon. Everything else —
156 /// a read errno, a frame cut short mid-payload — is the peer gone.
148 pub fn readFrame(self: *Link, alloc: std.mem.Allocator) !Incoming { 157 pub fn readFrame(self: *Link, alloc: std.mem.Allocator) !Incoming {
149 switch (self.*) { 158 switch (self.*) {
150 .quic => |*q| { 159 .quic => |*q| {
151 // Death is checked after the pump, so bytes that arrived in 160 // Death is checked after the pump, so bytes that arrived in
152 // the same pass as the close are still delivered before the 161 // the same pass as the close are still delivered before the
153 // tear. 162 // tear.
154 const got = proto.takeFrame(alloc, &q.cl.in) catch |err| switch (err) { 163 // takeFrame reads a buffer rather than a socket, so
155 error.OutOfMemory => return err, // not a transport event 164 // OutOfMemory and FrameTooLarge are its ONLY failures: there
156 else => return .closed, 165 // is no errno arm left to fold into .closed here.
157 }; 166 const got = try proto.takeFrame(alloc, &q.cl.in);
158 if (got) |frame| return .{ .frame = frame }; 167 if (got) |frame| return .{ .frame = frame };
159 return if (q.cl.dead) .closed else .incomplete; 168 return if (q.cl.dead) .closed else .incomplete;
160 }, 169 },
161 .fd => |fd| { 170 .fd => |fd| {
162 const frame = (proto.readFrame(alloc, fd) catch |err| switch (err) { 171 const frame = (proto.readFrame(alloc, fd) catch |err| switch (err) {
163 error.OutOfMemory => return err, 172 error.OutOfMemory, error.FrameTooLarge => return err,
164 else => return .closed, 173 else => return .closed,
165 }) orelse return .closed; 174 }) orelse return .closed;
166 return .{ .frame = frame }; 175 return .{ .frame = frame };
167 }, 176 },
168 .pipe => |p| { 177 .pipe => |p| {
169 const frame = (proto.readFrame(alloc, p.r) catch |err| switch (err) { 178 const frame = (proto.readFrame(alloc, p.r) catch |err| switch (err) {
170 error.OutOfMemory => return err, 179 error.OutOfMemory, error.FrameTooLarge => return err,
171 else => return .closed, 180 else => return .closed,
172 }) orelse return .closed; 181 }) orelse return .closed;
173 return .{ .frame = frame }; 182 return .{ .frame = frame };
@@ -196,7 +205,8 @@ pub const Link = union(enum) {
196 // in userspace. A BOUNDED fd or pipe may not: readFrame there is a 205 // in userspace. A BOUNDED fd or pipe may not: readFrame there is a
197 // read(2) on a stream, which on a quiet peer returns only when that 206 // read(2) on a stream, which on a quiet peer returns only when that
198 // peer speaks or goes away — the wait has to be spent in poll, where 207 // peer speaks or goes away — the wait has to be spent in poll, where
199 // the deadline can end it, which is why dial.askOn polls first too. 208 // the deadline can end it, which is why dial.ask has always polled
209 // before reading too.
200 // With no deadline the blocking read IS the wait, so they read 210 // With no deadline the blocking read IS the wait, so they read
201 // straight away and never reach the poll below. 211 // straight away and never reach the poll below.
202 const eager = switch (self.*) { 212 const eager = switch (self.*) {
@@ -251,7 +261,7 @@ pub const Link = union(enum) {
251 ready = eager or n != 0; 261 ready = eager or n != 0;
252 // Once poll HAS announced bytes, the fd arms still block in 262 // Once poll HAS announced bytes, the fd arms still block in
253 // readFrame until that frame is whole, even past the deadline — 263 // readFrame until that frame is whole, even past the deadline —
254 // the price of frames over a stream, same as dial.askOn always 264 // the price of frames over a stream, same as dial.ask always
255 // paid; a malformed frame is an error rather than an absent 265 // paid; a malformed frame is an error rather than an absent
256 // reply. 266 // reply.
257 } 267 }
@@ -437,3 +447,44 @@ test "awaitFrame: no deadline waits out a late reply" {
437 defer got.deinit(alloc); 447 defer got.deinit(alloc);
438 try testing.expectEqualStrings("late", got.payload); 448 try testing.expectEqualStrings("late", got.payload);
439 } 449 }
450
451 // The two pins below moved here from dial.zig when `dial.ask` stopped
452 // carrying its own await loop: both are properties of THIS loop, and the
453 // contract sentence they protect ("a corrupt frame is never reported as
454 // silence") is still dial.ask's doc comment.
455
456 test "awaitFrame: an empty payload of the wanted type is the answer, not a frame to skip" {
457 // The verb's decoder decides what an empty reply means. This loop used to
458 // carry a per-caller policy so that `upgrade_reply` could keep waiting
459 // through one; `protocol.parseUpgradeReply` now answers that question
460 // where the rest of the upgrade wire is read.
461 const alloc = testing.allocator;
462 const pair = try mkPair();
463 defer std.posix.close(pair[1]);
464 var l: Link = .{ .fd = pair[0] };
465 defer l.close();
466 try proto.writeFrame(pair[1], .upgrade_reply, "");
467 try proto.writeFrame(pair[1], .upgrade_reply, &.{0});
468
469 const frame = (try l.awaitFrame(alloc, .upgrade_reply, 500, .{})).?;
470 defer frame.deinit(alloc);
471 try testing.expectEqual(@as(usize, 0), frame.payload.len);
472 try testing.expect(proto.parseUpgradeReply(frame.payload) == null);
473 }
474
475 test "awaitFrame: a frame this side cannot read is an error, never silence" {
476 // Preserve corrupt-frame errors so dump and stats distinguish a broken
477 // daemon response from an absent daemon.
478 const pair = try mkPair();
479 defer std.posix.close(pair[1]);
480 var l: Link = .{ .fd = pair[0] };
481 defer l.close();
482 var hdr: [5]u8 = undefined;
483 hdr[0] = @intFromEnum(proto.MsgType.stats_reply);
484 std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little);
485 try proto.writeAllFd(pair[1], &hdr);
486 try testing.expectError(
487 error.FrameTooLarge,
488 l.awaitFrame(testing.allocator, .stats_reply, null, .{}),
489 );
490 }