a73x

c6e21ed4

test: server_test_session rides the harness awaits

a73x   2026-08-31 21:58

Commit message
test: server_test_session rides the harness awaits

`pumpUntilReplicaSees` becomes the pumping twin of `awaitReplicaText`:
same `ReplicaFeed` sink (now public in the harness), and a wait that
drives the daemon because this file's daemon has no thread. The reborn
attach's snapshot wait carries its "no delta may cross" rule as a sink
instead of a switch arm.

Two waits here are about the CLOSE, not about a frame, so the harness
gains the pair that says so: `awaitClosed` and its pumping twin
`pumpUntilClosed`. `Link.awaitFrame`'s `error.Closed` is folded into a
null by the other helpers — for them a gone peer and a quiet one are the
same non-answer — and these two exist because for a test asserting the
daemon hung up they are opposites.

`awaitFrame` gains `awaitFrameSink`, the same split `awaitFrameOn`
already has.

Four counted waits become `pumpUntil` on named conditions: `Dead` (asked
of the OS with kill(2), because a session table that has forgotten a
shell says nothing about the process) and `Freed`. The nagging loop in
the repeated-end_req test keeps its own shape and says why: the nag is
work done every round, and a predicate that wrote to a socket to answer
"is it dead" would read worse than the loop.

The reborn attach's `RebornAttachRefused` merges into
`NoRebornSnapshot`: a refused attach closes the connection and a silent
daemon leaves it open, and the assertion is about neither — it is about
no snapshot arriving.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TWxBL1HBULH1ZwTNzzKTja
src/server/server_test_harness.zig
Old New
@@ -341,6 +341,20 @@ pub fn awaitFrame(
341 want: proto.MsgType, 341 want: proto.MsgType,
342 iters: usize, 342 iters: usize,
343 ) !?proto.Frame { 343 ) !?proto.Frame {
344 return awaitFrameSink(alloc, srv, fd, want, iters, .{});
345 }
346
347 /// `awaitFrame` for the callers that assert on what arrived BEFORE the
348 /// match — a delta where only a snapshot may cross, a frame after a
349 /// goodbye. Same borrow rule as `awaitFrameOnSink`.
350 pub fn awaitFrameSink(
351 alloc: std.mem.Allocator,
352 srv: *Server,
353 fd: std.posix.fd_t,
354 want: proto.MsgType,
355 iters: usize,
356 sink: link.Sink,
357 ) !?proto.Frame {
344 // Not `l.close()` on any path: the caller owns this fd and closes it. 358 // Not `l.close()` on any path: the caller owns this fd and closes it.
345 var l: link.Link = .{ .fd = fd }; 359 var l: link.Link = .{ .fd = fd };
346 var i: usize = 0; 360 var i: usize = 0;
@@ -348,7 +362,7 @@ pub fn awaitFrame(
348 try srv.pumpOnce(5); 362 try srv.pumpOnce(5);
349 // Anything else on the way (a snapshot for the attached half of this 363 // Anything else on the way (a snapshot for the attached half of this
350 // test) is not what was asked for; the default sink drops it. 364 // test) is not what was asked for; the default sink drops it.
351 const got = l.awaitFrame(alloc, want, 2, .{}) catch |err| switch (err) { 365 const got = l.awaitFrame(alloc, want, 2, sink) catch |err| switch (err) {
352 error.Closed => return null, 366 error.Closed => return null,
353 else => return err, 367 else => return err,
354 }; 368 };
@@ -429,6 +443,31 @@ pub const ReplicaWait = struct {
429 reject: []const proto.MsgType = &.{}, 443 reject: []const proto.MsgType = &.{},
430 }; 444 };
431 445
446 /// The sink behind `awaitReplicaText`, public because the daemon is
447 /// sometimes threaded and sometimes pumped by the test, and only the wait
448 /// around this differs between those two.
449 pub const ReplicaFeed = struct {
450 alloc: std.mem.Allocator,
451 w: ReplicaWait,
452
453 pub fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
454 const self: *@This() = @ptrCast(@alignCast(ctx.?));
455 for (self.w.reject) |bad| {
456 if (frame.type == bad) return error.RejectedFrameArrived;
457 }
458 // applyFrame ignores everything that is not state, so the stream's
459 // replies and marks pass through untouched.
460 try applyFrame(self.alloc, self.w.replica, frame);
461 const plain = try self.w.replica.dumpPlain(self.alloc);
462 defer self.alloc.free(plain);
463 if (std.mem.indexOf(u8, plain, self.w.needle) != null) return error.TextArrived;
464 }
465
466 pub fn sink(self: *ReplicaFeed) link.Sink {
467 return .{ .ctx = self, .on = ReplicaFeed.on };
468 }
469 };
470
432 /// Replay one connection's frames into a replica until its grid shows the 471 /// Replay one connection's frames into a replica until its grid shows the
433 /// text, or the budget runs out. The loop this replaces was written out 472 /// text, or the budget runs out. The loop this replaces was written out
434 /// eight times in server_test_attach.zig alone, each copy re-deciding what 473 /// eight times in server_test_attach.zig alone, each copy re-deciding what
@@ -439,33 +478,48 @@ pub fn awaitReplicaText(
439 budget_ms: i64, 478 budget_ms: i64,
440 w: ReplicaWait, 479 w: ReplicaWait,
441 ) !bool { 480 ) !bool {
442 const Feed = struct { 481 var feed: ReplicaFeed = .{ .alloc = alloc, .w = w };
443 alloc: std.mem.Allocator, 482 _ = awaitFrameOnSink(alloc, fd, never_from_daemon, budget_ms, feed.sink()) catch |err| switch (err) {
444 w: ReplicaWait,
445 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
446 const self: *@This() = @ptrCast(@alignCast(ctx.?));
447 for (self.w.reject) |bad| {
448 if (frame.type == bad) return error.RejectedFrameArrived;
449 }
450 // applyFrame ignores everything that is not state, so the
451 // stream's replies and marks pass through untouched.
452 try applyFrame(self.alloc, self.w.replica, frame);
453 const plain = try self.w.replica.dumpPlain(self.alloc);
454 defer self.alloc.free(plain);
455 if (std.mem.indexOf(u8, plain, self.w.needle) != null) return error.TextArrived;
456 }
457 };
458 var feed: Feed = .{ .alloc = alloc, .w = w };
459 _ = awaitFrameOnSink(alloc, fd, never_from_daemon, budget_ms, .{
460 .ctx = &feed,
461 .on = Feed.on,
462 }) catch |err| switch (err) {
463 error.TextArrived => return true, 483 error.TextArrived => return true,
464 else => return err, 484 else => return err,
465 }; 485 };
466 return false; 486 return false;
467 } 487 }
468 488
489 /// True when the peer closed inside the budget. The CLOSE is the answer
490 /// here, not a frame — `Link.awaitFrame`'s `error.Closed`, which the
491 /// helpers above fold into a null because for them a gone peer and a quiet
492 /// one are the same non-answer. For a test asserting that the daemon hung
493 /// up, they are the opposite of each other. Anything that arrives first
494 /// goes to `sink`, which is where a caller says whether a frame after the
495 /// goodbye is a failure.
496 pub fn awaitClosed(alloc: std.mem.Allocator, fd: std.posix.fd_t, budget_ms: i64, sink: link.Sink) !bool {
497 var l: link.Link = .{ .fd = fd };
498 _ = l.awaitFrame(alloc, never_from_daemon, @intCast(@max(budget_ms, 0)), sink) catch |err| switch (err) {
499 error.Closed => return true,
500 else => return err,
501 };
502 return false;
503 }
504
505 /// `awaitClosed` for a daemon with no thread of its own: nothing reaches
506 /// the fd, close included, unless this loop pumps it.
507 pub fn pumpUntilClosed(
508 alloc: std.mem.Allocator,
509 srv: *Server,
510 fd: std.posix.fd_t,
511 deadline_ms: u64,
512 sink: link.Sink,
513 ) !bool {
514 var left = deadline_ms;
515 while (true) {
516 if (try awaitClosed(alloc, fd, 2, sink)) return true;
517 if (left == 0) return false;
518 try srv.pumpOnce(5);
519 left -|= 5;
520 }
521 }
522
469 /// The liveness half: "no mark arrived" is worthless against a shell that 523 /// The liveness half: "no mark arrived" is worthless against a shell that
470 /// never started. 524 /// never started.
471 pub fn awaitGridText( 525 pub fn awaitGridText(
src/server/server_test_session.zig
Old New
@@ -524,20 +524,13 @@ fn pumpUntilReplicaSees(
524 needle: []const u8, 524 needle: []const u8,
525 iters: usize, 525 iters: usize,
526 ) !bool { 526 ) !bool {
527 var i: usize = 0; 527 // The pumping twin of `h.awaitReplicaText`: same sink, and a wait that
528 while (i < iters) : (i += 1) { 528 // drives the daemon because this test's daemon has no thread.
529 try srv.pumpOnce(5); 529 var feed: h.ReplicaFeed = .{ .alloc = alloc, .w = .{ .replica = rep, .needle = needle } };
530 var pfd = [_]std.posix.pollfd{ 530 _ = h.awaitFrameSink(alloc, srv, fd, h.never_from_daemon, iters, feed.sink()) catch |err| switch (err) {
531 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, 531 error.TextArrived => return true,
532 }; 532 else => return err,
533 if ((std.posix.poll(&pfd, 1) catch 0) == 0) continue; 533 };
534 const frame = (try proto.readFrame(alloc, fd)) orelse return false;
535 defer frame.deinit(alloc);
536 try applyFrame(alloc, rep, frame);
537 const plain = try rep.dumpPlain(alloc);
538 defer alloc.free(plain);
539 if (std.mem.indexOf(u8, plain, needle) != null) return true;
540 }
541 return false; 534 return false;
542 } 535 }
543 536
@@ -900,14 +893,12 @@ test "Server: one session's shell exiting drops only its clients; the daemon car
900 } 893 }
901 // exit_status was the last frame queued before the drop, so what a's 894 // exit_status was the last frame queued before the drop, so what a's
902 // connection reads next is EOF — the daemon-side close, not a reset. 895 // connection reads next is EOF — the daemon-side close, not a reset.
903 var pfd = [_]std.posix.pollfd{ 896 const Stray = struct {
904 .{ .fd = ca.handle, .events = std.posix.POLL.IN, .revents = 0 }, 897 fn on(_: ?*anyopaque, _: proto.Frame) anyerror!void {
898 return error.FramesAfterExitStatus;
899 }
905 }; 900 };
906 try std.testing.expect((try std.posix.poll(&pfd, 1000)) > 0); 901 try std.testing.expect(try h.awaitClosed(alloc, ca.handle, 1000, .{ .on = Stray.on }));
907 if (try proto.readFrame(alloc, ca.handle)) |stray| {
908 stray.deinit(alloc);
909 return error.FramesAfterExitStatus;
910 }
911 902
912 // And b never noticed: still attached, still echoing. 903 // And b never noticed: still attached, still echoing.
913 try proto.writeFrame(cb.handle, .input, "post\n"); 904 try proto.writeFrame(cb.handle, .input, "post\n");
@@ -1029,11 +1020,14 @@ test "Server: a dead name re-attaches as a fresh session with a new epoch" {
1029 1020
1030 // Kill it, and pump until the name is free. 1021 // Kill it, and pump until the name is free.
1031 try proto.writeFrame(c1.handle, .input, "die 0\n"); 1022 try proto.writeFrame(c1.handle, .input, "die 0\n");
1032 var rounds: usize = 0; 1023 const Freed = struct {
1033 while (rounds < 500 and td.srv.sessions.find("a") != null) : (rounds += 1) { 1024 srv: *Server,
1034 try td.srv.pumpOnce(5); 1025 name: []const u8,
1035 } 1026 fn yes(self: @This()) bool {
1036 try std.testing.expect(td.srv.sessions.find("a") == null); 1027 return self.srv.sessions.find(self.name) == null;
1028 }
1029 };
1030 try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Freed{ .srv = &td.srv, .name = "a" }, Freed.yes));
1037 1031
1038 // Re-attach the dead name quoting the DEAD instance's seq and epoch — the 1032 // Re-attach the dead name quoting the DEAD instance's seq and epoch — the
1039 // reconnect a client that missed the death sends. A fresh session cannot 1033 // reconnect a client that missed the death sends. A fresh session cannot
@@ -1046,24 +1040,20 @@ test "Server: a dead name re-attaches as a fresh session with a new epoch" {
1046 .attach, 1040 .attach,
1047 proto.encodeAttachNamed(&abuf, 80, 24, p1.seq, p1.epoch, "a"), 1041 proto.encodeAttachNamed(&abuf, 80, 24, p1.seq, p1.epoch, "a"),
1048 ); 1042 );
1049 var second: ?proto.SnapshotPrefix = null; 1043 // A delta here would mean the reborn session kept the old instance's
1050 rounds = 0; 1044 // seq space, which is the bug this whole test exists for.
1051 while (rounds < 400 and second == null) : (rounds += 1) { 1045 const NoDelta = struct {
1052 try td.srv.pumpOnce(5); 1046 fn on(_: ?*anyopaque, frame: proto.Frame) anyerror!void {
1053 var pfd = [_]std.posix.pollfd{ 1047 if (frame.type == .delta) return error.DeltaAcrossInstances;
1054 .{ .fd = c2.handle, .events = std.posix.POLL.IN, .revents = 0 },
1055 };
1056 if ((std.posix.poll(&pfd, 1) catch 0) == 0) continue;
1057 const frame = (try proto.readFrame(alloc, c2.handle)) orelse
1058 return error.RebornAttachRefused;
1059 defer frame.deinit(alloc);
1060 switch (frame.type) {
1061 .snapshot => second = try proto.readSnapshotPrefix(frame.payload),
1062 .delta => return error.DeltaAcrossInstances,
1063 else => {},
1064 } 1048 }
1065 } 1049 };
1066 const p2 = second orelse return error.NoRebornSnapshot; 1050 // A refused attach closes the connection and a daemon that never
1051 // answered leaves it open; both reach here as "no snapshot arrived",
1052 // which is the fact the assertion is about either way.
1053 const f2 = (try h.awaitFrameSink(alloc, &td.srv, c2.handle, .snapshot, 400, .{ .on = NoDelta.on })) orelse
1054 return error.NoRebornSnapshot;
1055 defer f2.deinit(alloc);
1056 const p2 = try proto.readSnapshotPrefix(f2.payload);
1067 try std.testing.expect(p2.epoch != 0); 1057 try std.testing.expect(p2.epoch != 0);
1068 try std.testing.expect(p2.epoch != p1.epoch); 1058 try std.testing.expect(p2.epoch != p1.epoch);
1069 try std.testing.expect(td.srv.sessions.find("a") != null); 1059 try std.testing.expect(td.srv.sessions.find("a") != null);
@@ -1181,19 +1171,7 @@ test "Server: an observer's status_req names a session by tail" {
1181 defer refusal.deinit(alloc); 1171 defer refusal.deinit(alloc);
1182 try std.testing.expectEqualSlices(u8, &.{1}, refusal.payload); 1172 try std.testing.expectEqualSlices(u8, &.{1}, refusal.payload);
1183 1173
1184 var saw_eof = false; 1174 try std.testing.expect(try h.pumpUntilClosed(alloc, &td.srv, obs.handle, 2400, .{}));
1185 var iters: usize = 0;
1186 while (iters < 400 and !saw_eof) : (iters += 1) {
1187 try td.srv.pumpOnce(5);
1188 var pfd = [_]std.posix.pollfd{
1189 .{ .fd = obs.handle, .events = std.posix.POLL.IN, .revents = 0 },
1190 };
1191 if ((std.posix.poll(&pfd, 1) catch 0) == 0) continue;
1192 var drain: [64]u8 = undefined;
1193 const n = try std.posix.read(obs.handle, &drain);
1194 if (n == 0) saw_eof = true;
1195 }
1196 try std.testing.expect(saw_eof);
1197 } 1175 }
1198 1176
1199 test "Server: stats names every live session" { 1177 test "Server: stats names every live session" {
@@ -1363,6 +1341,15 @@ fn alive(pid: std.posix.pid_t) bool {
1363 return true; 1341 return true;
1364 } 1342 }
1365 1343
1344 /// Asked of the OS, not of the daemon: a session table that has forgotten a
1345 /// shell says nothing about whether the process is gone.
1346 const Dead = struct {
1347 pid: std.posix.pid_t,
1348 fn yes(self: Dead) bool {
1349 return !alive(self.pid);
1350 }
1351 };
1352
1366 test "Server: end_req with another client attached is refused with the count; forced, both see the exit" { 1353 test "Server: end_req with another client attached is refused with the count; forced, both see the exit" {
1367 const alloc = std.testing.allocator; 1354 const alloc = std.testing.allocator;
1368 1355
@@ -1406,12 +1393,7 @@ test "Server: end_req with another client attached is refused with the count; fo
1406 try std.testing.expect((proto.parseEndReply(r3.payload) orelse return error.BadEndReply).accepted); 1393 try std.testing.expect((proto.parseEndReply(r3.payload) orelse return error.BadEndReply).accepted);
1407 const x2 = (try awaitFrame(alloc, &td.srv, a2.handle, .exit_status, 400)) orelse return error.NoExitOnSibling; 1394 const x2 = (try awaitFrame(alloc, &td.srv, a2.handle, .exit_status, 400)) orelse return error.NoExitOnSibling;
1408 x2.deinit(alloc); 1395 x2.deinit(alloc);
1409 var waited: u32 = 0; 1396 try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Dead{ .pid = pid_a }, Dead.yes));
1410 while (alive(pid_a) and waited < 3000) : (waited += 50) {
1411 try td.srv.pumpOnce(20);
1412 std.Thread.sleep(30 * std.time.ns_per_ms);
1413 }
1414 try std.testing.expect(!alive(pid_a));
1415 try std.testing.expect(alive(shellPidOf(&td.srv, "b"))); 1397 try std.testing.expect(alive(shellPidOf(&td.srv, "b")));
1416 } 1398 }
1417 1399
@@ -1431,12 +1413,7 @@ test "Server: end_req alone on a session ends it at once, and an unknown name is
1431 const r = (try awaitFrame(alloc, &td.srv, c.handle, .end_reply, 200)) orelse return error.NoEndReply; 1413 const r = (try awaitFrame(alloc, &td.srv, c.handle, .end_reply, 200)) orelse return error.NoEndReply;
1432 defer r.deinit(alloc); 1414 defer r.deinit(alloc);
1433 try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted); 1415 try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted);
1434 var waited: u32 = 0; 1416 try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Dead{ .pid = pid }, Dead.yes));
1435 while (alive(pid) and waited < 3000) : (waited += 50) {
1436 try td.srv.pumpOnce(20);
1437 std.Thread.sleep(30 * std.time.ns_per_ms);
1438 }
1439 try std.testing.expect(!alive(pid));
1440 1417
1441 const obs = try dial.dial(td.sock_path); 1418 const obs = try dial.dial(td.sock_path);
1442 defer obs.close(); 1419 defer obs.close();
@@ -1516,12 +1493,7 @@ test "Server: an accepted end kills a shell that ignores TERM and HUP" {
1516 defer r.deinit(alloc); 1493 defer r.deinit(alloc);
1517 try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted); 1494 try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted);
1518 1495
1519 var waited: u32 = 0; 1496 try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Dead{ .pid = pid }, Dead.yes));
1520 while (alive(pid) and waited < 3000) : (waited += 50) {
1521 try td.srv.pumpOnce(20);
1522 std.Thread.sleep(30 * std.time.ns_per_ms);
1523 }
1524 try std.testing.expect(!alive(pid));
1525 } 1497 }
1526 1498
1527 test "Server: a keystroke into an ending session does not cost that client its exit_status" { 1499 test "Server: a keystroke into an ending session does not cost that client its exit_status" {
@@ -1577,6 +1549,9 @@ test "Server: a repeated end_req does not push the SIGKILL deadline out" {
1577 // clock, the shell would outlive every deadline it was ever given. 1549 // clock, the shell would outlive every deadline it was ever given.
1578 const obs = try dial.dial(td.sock_path); 1550 const obs = try dial.dial(td.sock_path);
1579 defer obs.close(); 1551 defer obs.close();
1552 // Its own loop rather than `h.pumpUntil`: the nag is WORK done every
1553 // round, and a predicate that wrote to a socket to answer "is it dead"
1554 // would be a worse thing to read than this.
1580 var waited: u32 = 0; 1555 var waited: u32 = 0;
1581 while (alive(pid) and waited < 3000) : (waited += 80) { 1556 while (alive(pid) and waited < 3000) : (waited += 80) {
1582 proto.writeFrame(obs.handle, .end_req, proto.encodeEndReq(&rq, true, "nag")) catch {}; 1557 proto.writeFrame(obs.handle, .end_req, proto.encodeEndReq(&rq, true, "nag")) catch {};