a73x

0fffc3cf

test: the harness gains pumpUntil and Link-backed awaits

a73x   2026-08-31 21:58

Commit message
test: the harness gains pumpUntil and Link-backed awaits

`pumpUntil` gives the sibling test files one way to say "pump the daemon
until this is true", so a wait asserts the CONDITION and names it on
failure instead of guessing a round count. Its false return is the
deadline made assertable: a wedged zig test prints nothing at all, so a
loop that can never finish has to end in a failure message rather than a
hang.

`awaitFrameOn`, `awaitFrame` and `firstStateFrame` become wrappers on
`Link.awaitFrame` — the primitive every hand-rolled poll+readFrame loop
in the tree was a copy of. All three keep their exact signatures; the
eight sibling files calling them are the proof, and `awaitFrameOnSink`
is the new spelling for the callers that assert on what arrived BEFORE
the match. `firstStateFrame` wants either state frame, so it names
`.snapshot` and catches `.delta` in a sink that copies the header ints
out and ends the wait — a Sink borrows its frame.

`awaitFrame` keeps its own loop around the primitive rather than
becoming one call: its daemon has no thread, so nothing reaches the fd
unless this loop pumps between waits. Its per-turn deadline is 2 ms
rather than 1 because awaitFrame re-reads the clock before polling, and
a 1 ms budget can already be spent by then.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TWxBL1HBULH1ZwTNzzKTja
build.zig
Old New
@@ -196,7 +196,7 @@ const mod_table = [_]ModSpec{
196 // daemon itself when nobody handed it a --key — and for the shim 196 // daemon itself when nobody handed it a --key — and for the shim
197 // directory shell integration writes under the same 0700 policy. 197 // directory shell integration writes under the same 0700 policy.
198 // `pty` stays a row of its own: the ptyclient fixture consumes it. 198 // `pty` stays a row of its own: the ptyclient fixture consumes it.
199 .{ .name = "daemon", .path = "src/server/server.zig", .link_libc = true, .imports = &.{ "term", "pty", "sockpath", "serve", "quic", "xdg", "proxy" }, .test_imports = &.{ "testtmp", "dial" }, .quic_tests = true }, 199 .{ .name = "daemon", .path = "src/server/server.zig", .link_libc = true, .imports = &.{ "term", "pty", "sockpath", "serve", "quic", "xdg", "proxy" }, .test_imports = &.{ "testtmp", "dial", "link" }, .quic_tests = true },
200 // The agent-facing client. It speaks frames and owns no terminal, which 200 // The agent-facing client. It speaks frames and owns no terminal, which
201 // is the whole point — it attaches at 0x0 and never claims the grid. 201 // is the whole point — it attaches at 0x0 and never claims the grid.
202 // The transport modules are the CLI client's, minus everything that 202 // The transport modules are the CLI client's, minus everything that
src/server/server_test_harness.zig
Old New
@@ -11,6 +11,10 @@ const TmpDir = @import("testtmp").TmpDir;
11 /// CLI client and `mux a` dial through, so these tests hold the socket the 11 /// CLI client and `mux a` dial through, so these tests hold the socket the
12 /// way the product does rather than hand-rolling the pair of calls. 12 /// way the product does rather than hand-rolling the pair of calls.
13 pub const dial = @import("dial"); 13 pub const dial = @import("dial");
14 /// The awaits below are wrappers on `Link.awaitFrame` rather than their own
15 /// poll+readFrame loops, so a test waiting on a daemon frame waits the way
16 /// the client and `mux a` do.
17 const link_mod = @import("link");
14 const srv_mod = @import("server.zig"); 18 const srv_mod = @import("server.zig");
15 const Server = srv_mod.Server; 19 const Server = srv_mod.Server;
16 20
@@ -20,6 +24,27 @@ pub fn serverThread(srv: *Server, stop: *std.atomic.Value(bool)) void {
20 } 24 }
21 } 25 }
22 26
27 /// Pump the daemon until pred says the world arrived, or the deadline says
28 /// it never will. Returns whether pred fired, so a test asserts the
29 /// CONDITION and its failure names what didn't happen — not a guessed
30 /// round count. A false return is an assertable value: the deadline turns
31 /// a wedge into a legible failure instead of a silent hang (a wedged zig
32 /// test prints nothing).
33 pub fn pumpUntil(
34 srv: *Server,
35 deadline_ms: u64,
36 ctx: anytype,
37 comptime pred: fn (@TypeOf(ctx)) bool,
38 ) !bool {
39 var left = deadline_ms;
40 while (true) {
41 if (pred(ctx)) return true;
42 if (left == 0) return false;
43 try srv.pumpOnce(5);
44 left -|= 5;
45 }
46 }
47
23 /// A daemon on a socket of its own, which is what nearly every test in this 48 /// A daemon on a socket of its own, which is what nearly every test in this
24 /// folder opens with: a short-path temp directory (`testtmp`, because a unix 49 /// folder opens with: a short-path temp directory (`testtmp`, because a unix
25 /// socket path caps at 108 bytes), a socket named inside it, and a `Server` 50 /// socket path caps at 108 bytes), a socket named inside it, and a `Server`
@@ -172,30 +197,38 @@ pub fn connectedPair() !SockPair {
172 /// carry the session epoch. 197 /// carry the session epoch.
173 pub const StateFrame = struct { type: proto.MsgType, seq: u64, epoch: u64 }; 198 pub const StateFrame = struct { type: proto.MsgType, seq: u64, epoch: u64 };
174 199
200 /// Either state frame answers, so the wait names `.snapshot` as its `want`
201 /// and catches `.delta` in the sink. The sink copies the header's ints out
202 /// and ends the wait with `error.StateFrameSeen`, which is this function's
203 /// success rather than a failure — a Sink borrows its frame, so nothing may
204 /// be kept but the decoded numbers.
175 pub fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !?StateFrame { 205 pub fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !?StateFrame {
176 var deadline_ms = timeout_ms; 206 const Catch = struct {
177 while (deadline_ms > 0) { 207 got: StateFrame = undefined,
178 var pfd = [_]std.posix.pollfd{ 208 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
179 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, 209 if (frame.type != .delta) return;
180 }; 210 const self: *@This() = @ptrCast(@alignCast(ctx.?));
181 const ready = try std.posix.poll(&pfd, 100); 211 const hdr = try proto.readDeltaHeader(frame.payload);
182 deadline_ms -|= 100; 212 self.got = .{ .type = .delta, .seq = hdr.seq, .epoch = 0 };
183 if (ready == 0) continue; 213 return error.StateFrameSeen;
184 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
185 defer frame.deinit(alloc);
186 switch (frame.type) {
187 .snapshot => {
188 const p = try proto.readSnapshotPrefix(frame.payload);
189 return .{ .type = .snapshot, .seq = p.seq, .epoch = p.epoch };
190 },
191 .delta => {
192 const h = try proto.readDeltaHeader(frame.payload);
193 return .{ .type = .delta, .seq = h.seq, .epoch = 0 };
194 },
195 else => {},
196 } 214 }
197 } 215 };
198 return null; 216 var caught: Catch = .{};
217 // Not `l.close()` on any path: the caller owns this fd and closes it.
218 var l: link_mod.Link = .{ .fd = fd };
219 const frame = l.awaitFrame(alloc, .snapshot, @intCast(timeout_ms), .{
220 .ctx = &caught,
221 .on = Catch.on,
222 }) catch |err| switch (err) {
223 error.StateFrameSeen => return caught.got,
224 // A peer that went away before any state frame reads the same as one
225 // that never sent one, which is what every caller asserts on.
226 error.Closed => return null,
227 else => return err,
228 } orelse return null;
229 defer frame.deinit(alloc);
230 const p = try proto.readSnapshotPrefix(frame.payload);
231 return .{ .type = .snapshot, .seq = p.seq, .epoch = p.epoch };
199 } 232 }
200 233
201 // --------------------------------------------------------------------------- 234 // ---------------------------------------------------------------------------
@@ -259,8 +292,11 @@ pub fn findFrame(bytes: []const u8, want: proto.MsgType) ?[]const u8 {
259 // and fewer tests is fewer places that can fail to. 292 // and fewer tests is fewer places that can fail to.
260 // --------------------------------------------------------------------------- 293 // ---------------------------------------------------------------------------
261 294
262 /// The read after the poll is blocking: safe only because these replies land 295 /// The daemon here has no thread of its own: nothing arrives on `fd` unless
263 /// in one write. 296 /// this loop pumps it, so the wait is `iters` short `Link.awaitFrame` waits
297 /// with a pump between them rather than one long one. 2 ms rather than 1 so
298 /// that each turn always reaches its poll — `awaitFrame` re-reads the clock
299 /// before polling, and a 1 ms deadline can already be spent by then.
264 pub fn awaitFrame( 300 pub fn awaitFrame(
265 alloc: std.mem.Allocator, 301 alloc: std.mem.Allocator,
266 srv: *Server, 302 srv: *Server,
@@ -268,18 +304,18 @@ pub fn awaitFrame(
268 want: proto.MsgType, 304 want: proto.MsgType,
269 iters: usize, 305 iters: usize,
270 ) !?proto.Frame { 306 ) !?proto.Frame {
307 // Not `l.close()` on any path: the caller owns this fd and closes it.
308 var l: link_mod.Link = .{ .fd = fd };
271 var i: usize = 0; 309 var i: usize = 0;
272 while (i < iters) : (i += 1) { 310 while (i < iters) : (i += 1) {
273 try srv.pumpOnce(5); 311 try srv.pumpOnce(5);
274 var pfd = [_]std.posix.pollfd{
275 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
276 };
277 if ((std.posix.poll(&pfd, 1) catch 0) == 0) continue;
278 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
279 if (frame.type == want) return frame;
280 // Anything else on the way (a snapshot for the attached half of this 312 // Anything else on the way (a snapshot for the attached half of this
281 // test) is not what was asked for; drop it and keep pumping. 313 // test) is not what was asked for; the default sink drops it.
282 frame.deinit(alloc); 314 const got = l.awaitFrame(alloc, want, 2, .{}) catch |err| switch (err) {
315 error.Closed => return null,
316 else => return err,
317 };
318 if (got) |frame| return frame;
283 } 319 }
284 return null; 320 return null;
285 } 321 }
@@ -298,30 +334,40 @@ pub fn awaitFrame(
298 /// giving them more time. 334 /// giving them more time.
299 /// 335 ///
300 /// Non-matching frames are dropped, the same as `awaitFrame` drops them, so 336 /// Non-matching frames are dropped, the same as `awaitFrame` drops them, so
301 /// this is only right for "the first `want` to arrive" — not for a caller that 337 /// this is only right for "the first `want` to arrive"; a caller that asserts
302 /// asserts on what came before it. Null means the budget ran out or the peer 338 /// on what came before it wants `awaitFrameOnSink`. Null means the budget ran
303 /// closed; the caller is the one that knows which of those is a failure. 339 /// out or the peer closed; the caller is the one that knows which of those is
340 /// a failure.
304 pub fn awaitFrameOn( 341 pub fn awaitFrameOn(
305 alloc: std.mem.Allocator, 342 alloc: std.mem.Allocator,
306 fd: std.posix.fd_t, 343 fd: std.posix.fd_t,
307 want: proto.MsgType, 344 want: proto.MsgType,
308 timeout_ms: i64, 345 timeout_ms: i64,
309 ) !?proto.Frame { 346 ) !?proto.Frame {
310 const deadline = std.time.milliTimestamp() + timeout_ms; 347 return awaitFrameOnSink(alloc, fd, want, timeout_ms, .{});
311 while (true) { 348 }
312 const left = deadline - std.time.milliTimestamp(); 349
313 if (left <= 0) return null; 350 /// `awaitFrameOn` for the callers that DO assert on what came before the
314 var pfd = [_]std.posix.pollfd{ 351 /// match — a count of scrollback frames, the size of a snapshot that should
315 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, 352 /// not have arrived. The sink BORROWS each non-matching frame (see
316 }; 353 /// `link.Sink`), so anything kept has to be copied out of it, and an error
317 // Capped so the poll cannot outlast the budget, and so a long budget 354 /// out of the sink ends the wait with that error.
318 // still wakes often enough to notice it has expired. 355 pub fn awaitFrameOnSink(
319 const ready = try std.posix.poll(&pfd, @intCast(@min(left, 100))); 356 alloc: std.mem.Allocator,
320 if (ready == 0) continue; 357 fd: std.posix.fd_t,
321 const frame = (try proto.readFrame(alloc, fd)) orelse return null; 358 want: proto.MsgType,
322 if (frame.type == want) return frame; 359 timeout_ms: i64,
323 frame.deinit(alloc); 360 sink: link_mod.Sink,
324 } 361 ) !?proto.Frame {
362 // Not `l.close()` on any path: the caller owns this fd and closes it.
363 var l: link_mod.Link = .{ .fd = fd };
364 return l.awaitFrame(alloc, want, @intCast(@max(timeout_ms, 0)), sink) catch |err| switch (err) {
365 // The budget running out and the peer going away were both null in
366 // the hand-rolled loop this replaces, and every caller reads null as
367 // "no such frame arrived".
368 error.Closed => null,
369 else => err,
370 };
325 } 371 }
326 372
327 /// The liveness half: "no mark arrived" is worthless against a shell that 373 /// The liveness half: "no mark arrived" is worthless against a shell that
@@ -407,3 +453,42 @@ pub fn writeStubbornShell(alloc: std.mem.Allocator, tmp: *TmpDir) ![:0]u8 {
407 }); 453 });
408 return std.fmt.allocPrintSentinel(alloc, "{s}/stubborn.sh", .{tmp.path()}, 0); 454 return std.fmt.allocPrintSentinel(alloc, "{s}/stubborn.sh", .{tmp.path()}, 0);
409 } 455 }
456
457 // ---------------------------------------------------------------------------
458 // The harness's own primitive. `pumpUntil` is what the sibling files assert
459 // their daemon-side conditions through, so its two outcomes are pinned here
460 // rather than inferred from a suite that happens to pass.
461 // ---------------------------------------------------------------------------
462
463 const Preds = struct {
464 fn hasAnyClient(srv: *Server) bool {
465 for (srv.clients) |slot| if (slot != null) return true;
466 return false;
467 }
468 fn never(srv: *Server) bool {
469 _ = srv;
470 return false;
471 }
472 };
473
474 test "pumpUntil: a condition the daemon reaches returns true, one it never does returns false" {
475 const alloc = std.testing.allocator;
476 var td = try TestDaemon.init(alloc, "pumpuntil", .{ .shell = "/bin/cat" });
477 defer td.deinit();
478
479 // Nothing has dialled yet, so the predicate must be false BEFORE the
480 // dial — otherwise the true below would prove nothing about pumping.
481 try std.testing.expect(!Preds.hasAnyClient(&td.srv));
482 const c = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
483 defer c.close();
484 try std.testing.expect(try pumpUntil(&td.srv, 2000, &td.srv, Preds.hasAnyClient));
485
486 // The deadline is the whole point: a predicate that can never fire has to
487 // come back as an assertable false, in about the time asked for, rather
488 // than wedging the runner. Only the lower bound is asserted — a loaded
489 // box makes the upper one flaky, and overshooting is not the failure
490 // this guards against.
491 const t0 = std.time.milliTimestamp();
492 try std.testing.expect(!try pumpUntil(&td.srv, 200, &td.srv, Preds.never));
493 try std.testing.expect(std.time.milliTimestamp() - t0 >= 200);
494 }