273232d6
refactor: dial.ask asks the Link; askOn's loop becomes awaitFrame
a73x 2026-08-31 18:47
Commit message
build.zig
| Old | New | ||
|---|---|---|---|
| @@ -163,10 +163,10 @@ const mod_table = [_]ModSpec{ | |||
| 163 | .{ .name = "spawn", .path = "src/cli/spawn.zig", .link_libc = true }, | 163 | .{ .name = "spawn", .path = "src/cli/spawn.zig", .link_libc = true }, |
| 164 | // ---- single-hop over the leaves ---- | 164 | // ---- single-hop over the leaves ---- |
| 165 | // The client side of a daemon's socket: dial it, and say hello. `term` | 165 | // The client side of a daemon's socket: dial it, and say hello. `term` |
| 166 | // is the attach encoders and is all it takes, which is the point — an | 166 | // is the attach encoders, `link` is the round trip's wait — an embedder |
| 167 | // embedder reaches a daemon by linking this and the wire contract | 167 | // reaches a daemon by linking those two and this, instead of the whole |
| 168 | // instead of the whole client module. | 168 | // client module. |
| 169 | .{ .name = "dial", .path = "src/dial.zig", .imports = &.{"term"} }, | 169 | .{ .name = "dial", .path = "src/dial.zig", .link_libc = true, .imports = &.{ "term", "link" }, .quic_tests = true }, |
| 170 | // The live connection itself — fd, pipe or QUIC — and the one wait-for-a- | 170 | // The live connection itself — fd, pipe or QUIC — and the one wait-for-a- |
| 171 | // frame loop. `term` for frames, `quic` for the third arm; policy stays | 171 | // frame loop. `term` for frames, `quic` for the third arm; policy stays |
| 172 | // with the rows that import this one. | 172 | // with the rows that import this one. |
src/dial.zig
| Old | New | ||
|---|---|---|---|
| @@ -4,11 +4,13 @@ | |||
| 4 | //! | 4 | //! |
| 5 | //! Connecting a client to a daemon is the operation this product exists to | 5 | //! Connecting a client to a daemon is the operation this product exists to |
| 6 | //! perform, so it is a callable primitive rather than four lines every caller | 6 | //! perform, so it is a callable primitive rather than four lines every caller |
| 7 | //! writes again. It imports `term` for the attach encoders and nothing else, | 7 | //! writes again. It imports `term` for the attach encoders and `link` for the |
| 8 | //! which is the point: an embedder that wants to reach a daemon links this | 8 | //! one round trip's wait, and nothing else: an embedder that wants to reach a |
| 9 | //! and `term`, not the client module's transports, hosts file and pane tree. | 9 | //! daemon links this, `term` and `link`, not the client module's transports, |
| 10 | //! hosts file and pane tree. | ||
| 10 | const std = @import("std"); | 11 | const std = @import("std"); |
| 11 | const proto = @import("term").protocol; | 12 | const proto = @import("term").protocol; |
| 13 | const link_mod = @import("link"); | ||
| 12 | 14 | ||
| 13 | /// The connection alone, with no frame sent. What an observer verb, a probe | 15 | /// The connection alone, with no frame sent. What an observer verb, a probe |
| 14 | /// or a client resuming from a watermark wants: the first bytes on the | 16 | /// or a client resuming from a watermark wants: the first bytes on the |
| @@ -91,50 +93,24 @@ pub fn ask( | |||
| 91 | deadline_ms: ?u32, | 93 | deadline_ms: ?u32, |
| 92 | ) !?proto.Frame { | 94 | ) !?proto.Frame { |
| 93 | const s = dial(sock_path) catch return error.NoDaemon; | 95 | const s = dial(sock_path) catch return error.NoDaemon; |
| 94 | defer s.close(); | 96 | // The Link owns the fd from here: its close() is the one that runs. |
| 95 | return askOn(alloc, s.handle, req, payload, want, deadline_ms); | 97 | var l: link_mod.Link = .{ .fd = s.handle }; |
| 96 | } | 98 | defer l.close(); |
| 97 | |||
| 98 | /// `ask` once the connection exists. Private on purpose: this loop DROPS every | ||
| 99 | /// frame that is not the one it was told to wait for, which is right for a | ||
| 100 | /// socket opened to ask one question and wrong for a client connection | ||
| 101 | /// carrying a snapshot and its deltas. The tests below reach it through a | ||
| 102 | /// socketpair — the one fd of this shape that no dial produced. | ||
| 103 | fn askOn( | ||
| 104 | alloc: std.mem.Allocator, | ||
| 105 | fd: std.posix.fd_t, | ||
| 106 | req: proto.MsgType, | ||
| 107 | payload: []const u8, | ||
| 108 | want: proto.MsgType, | ||
| 109 | deadline_ms: ?u32, | ||
| 110 | ) !?proto.Frame { | ||
| 111 | // A request that could not be delivered is its own answer: the caller | 99 | // A request that could not be delivered is its own answer: the caller |
| 112 | // reports a daemon that never heard the question differently from one | 100 | // reports a daemon that never heard the question differently from one |
| 113 | // that heard it and said nothing. | 101 | // that heard it and said nothing. |
| 114 | proto.writeFrame(fd, req, payload) catch return error.RequestNotSent; | 102 | l.sendFrame(req, payload) catch return error.RequestNotSent; |
| 115 | const deadline: ?i64 = if (deadline_ms) |ms| std.time.milliTimestamp() + ms else null; | 103 | // A peer that closed without answering is the same "no answer" as a |
| 116 | while (true) { | 104 | // deadline that ran out — dial.ask's contract predates the Link and |
| 117 | if (deadline) |end| { | 105 | // keeps it; callers that need the distinction hold a Link themselves. |
| 118 | const left = end - std.time.milliTimestamp(); | 106 | // The null sink is the observer-verb policy: every frame that is not the |
| 119 | if (left <= 0) return null; | 107 | // wanted one is dropped, which is right for a socket opened to ask one |
| 120 | var fds = [_]std.posix.pollfd{ | 108 | // question and wrong for a client connection carrying a snapshot and its |
| 121 | .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, | 109 | // deltas. |
| 122 | }; | 110 | return l.awaitFrame(alloc, want, deadline_ms, .{}) catch |e| switch (e) { |
| 123 | if ((std.posix.poll(&fds, @intCast(left)) catch return null) == 0) return null; | 111 | error.Closed => null, |
| 124 | if (fds[0].revents == 0) continue; | 112 | else => e, |
| 125 | } | 113 | }; |
| 126 | // The frame read may extend past the poll deadline if a peer writes | ||
| 127 | // only part of a frame. A malformed or partial frame is an error | ||
| 128 | // rather than an absent reply. | ||
| 129 | const frame = (try proto.readFrame(alloc, fd)) orelse return null; | ||
| 130 | // An empty payload of the wanted type IS the answer here. What an | ||
| 131 | // empty reply means belongs to that verb's decoder — `stats_reply` | ||
| 132 | // says "nothing to report" with one, `parseUpgradeReply` calls one no | ||
| 133 | // answer at all — and a transport that guessed for them would have to | ||
| 134 | // be told the verb's policy by every caller. | ||
| 135 | if (frame.type == want) return frame; | ||
| 136 | frame.deinit(alloc); | ||
| 137 | } | ||
| 138 | } | 114 | } |
| 139 | 115 | ||
| 140 | // A daemon of our own is the server suite's business, not this module's: | 116 | // A daemon of our own is the server suite's business, not this module's: |
| @@ -178,79 +154,3 @@ test "ask: a path nothing is bound at is error.NoDaemon, not a connect errno" { | |||
| 178 | ask(std.testing.allocator, "/nonexistent-dir/mux-dial-test.sock", .stats_req, "", .stats_reply, 100), | 154 | ask(std.testing.allocator, "/nonexistent-dir/mux-dial-test.sock", .stats_req, "", .stats_reply, 100), |
| 179 | ); | 155 | ); |
| 180 | } | 156 | } |
| 181 | |||
| 182 | test "askOn: a deadline gives up on silence, and no deadline waits out a late reply" { | ||
| 183 | const alloc = std.testing.allocator; | ||
| 184 | |||
| 185 | // A socket pair models the daemon and controls whether and when a reply | ||
| 186 | // arrives. | ||
| 187 | { | ||
| 188 | var pair: [2]i32 = undefined; | ||
| 189 | try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair)); | ||
| 190 | defer std.posix.close(pair[0]); | ||
| 191 | defer std.posix.close(pair[1]); | ||
| 192 | const t0 = std.time.milliTimestamp(); | ||
| 193 | try std.testing.expect((try askOn(alloc, pair[0], .stats_req, "", .stats_reply, 100)) == null); | ||
| 194 | // Verify that silence consumes the deadline before returning null. | ||
| 195 | try std.testing.expect(std.time.milliTimestamp() - t0 >= 100); | ||
| 196 | } | ||
| 197 | |||
| 198 | // Without a deadline, wait for a delayed reply. | ||
| 199 | { | ||
| 200 | var pair: [2]i32 = undefined; | ||
| 201 | try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair)); | ||
| 202 | defer std.posix.close(pair[0]); | ||
| 203 | const Late = struct { | ||
| 204 | fn run(fd: std.posix.fd_t) void { | ||
| 205 | std.Thread.sleep(150 * std.time.ns_per_ms); | ||
| 206 | // Ignore an unrelated frame before returning the requested type. | ||
| 207 | proto.writeFrame(fd, .stats_req, "") catch {}; | ||
| 208 | proto.writeFrame(fd, .stats_reply, "late") catch {}; | ||
| 209 | std.posix.close(fd); | ||
| 210 | } | ||
| 211 | }; | ||
| 212 | const th = try std.Thread.spawn(.{}, Late.run, .{pair[1]}); | ||
| 213 | defer th.join(); | ||
| 214 | const t0 = std.time.milliTimestamp(); | ||
| 215 | const frame = (try askOn(alloc, pair[0], .stats_req, "", .stats_reply, null)).?; | ||
| 216 | defer frame.deinit(alloc); | ||
| 217 | try std.testing.expectEqualStrings("late", frame.payload); | ||
| 218 | try std.testing.expect(std.time.milliTimestamp() - t0 >= 150); | ||
| 219 | } | ||
| 220 | } | ||
| 221 | |||
| 222 | test "askOn: an empty payload of the wanted type is the answer, not a frame to skip" { | ||
| 223 | // The verb's decoder decides what an empty reply means. This loop used to | ||
| 224 | // carry a per-caller policy so that `upgrade_reply` could keep waiting | ||
| 225 | // through one; `protocol.parseUpgradeReply` now answers that question | ||
| 226 | // where the rest of the upgrade wire is read. | ||
| 227 | const alloc = std.testing.allocator; | ||
| 228 | var pair: [2]i32 = undefined; | ||
| 229 | try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair)); | ||
| 230 | defer std.posix.close(pair[0]); | ||
| 231 | defer std.posix.close(pair[1]); | ||
| 232 | try proto.writeFrame(pair[1], .upgrade_reply, ""); | ||
| 233 | try proto.writeFrame(pair[1], .upgrade_reply, &.{0}); | ||
| 234 | |||
| 235 | const frame = (try askOn(alloc, pair[0], .upgrade_req, "", .upgrade_reply, 500)).?; | ||
| 236 | defer frame.deinit(alloc); | ||
| 237 | try std.testing.expectEqual(@as(usize, 0), frame.payload.len); | ||
| 238 | try std.testing.expect(proto.parseUpgradeReply(frame.payload) == null); | ||
| 239 | } | ||
| 240 | |||
| 241 | test "askOn: a frame this side cannot read is an error, never silence" { | ||
| 242 | // Preserve corrupt-frame errors so dump and stats distinguish a broken | ||
| 243 | // daemon response from an absent daemon. | ||
| 244 | var pair: [2]i32 = undefined; | ||
| 245 | try std.testing.expectEqual(@as(usize, 0), std.os.linux.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair)); | ||
| 246 | defer std.posix.close(pair[0]); | ||
| 247 | defer std.posix.close(pair[1]); | ||
| 248 | var hdr: [5]u8 = undefined; | ||
| 249 | hdr[0] = @intFromEnum(proto.MsgType.stats_reply); | ||
| 250 | std.mem.writeInt(u32, hdr[1..5], proto.max_payload + 1, .little); | ||
| 251 | try proto.writeAllFd(pair[1], &hdr); | ||
| 252 | try std.testing.expectError( | ||
| 253 | error.FrameTooLarge, | ||
| 254 | askOn(std.testing.allocator, pair[0], .stats_req, "", .stats_reply, null), | ||
| 255 | ); | ||
| 256 | } | ||