a73x

63169650

test: server_test_attach rides the harness awaits

a73x   2026-08-31 21:58

Commit message
test: server_test_attach rides the harness awaits

Thirty-eight of this file's poll+readFrame loops were the same loop with
different bodies, each re-deciding what a poll timeout meant and each
charging its budget 100 ms per turn whether it waited or not. They become
`Link.awaitFrame` waits with sinks, and the shapes that recurred get one
name each:

  - `awaitReplicaText` (harness) replays a connection into a replica until
    the grid shows the text — eight copies here alone. Its `reject` list
    is the "and no snapshot arrived" half those copies carried inline.
  - `awaitHistoryRows` for the two waits on a state frame's history count,
    which both kinds of state frame carry.
  - `ApplyEach`, a sink that keeps a replica current while the wait is for
    one particular reply underneath the broadcast.
  - `never_from_daemon` (harness) names `.attach` for the waits whose
    answer comes out of the SINK: nothing can match, so every frame
    reaches it.

`awaitMarkerWithoutSnapshot` is gone, replaced by `awaitReplicaText` with
a blank replica at each of its two call sites. `awaitSnapshotSize` KEEPS
its name against the brief's "delete it": ten call sites would each have
grown the same sink, and its body is now one `awaitFrameOnSink`. It has
to sink every frame rather than want `.snapshot`, because being lenient
about snapshots at other sizes is the whole point of it.

`Lead` moves to the harness — this file and server_test_agent both wait
on the latest-active rule now.

Two poll loops stay, and say why in place: the quiet-drain in the
same-size-join test and `drainHeld`. Their condition is that the socket
went SILENT, and `Link.awaitFrame` deliberately does not report that — a
poll that timed out inside it is indistinguishable from the deadline
running out.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TWxBL1HBULH1ZwTNzzKTja
src/server/server_test_agent.zig
Old New
@@ -9,6 +9,7 @@ const Server = srv_mod.Server;
9 const max_agent_chans = srv_mod.max_agent_chans; 9 const max_agent_chans = srv_mod.max_agent_chans;
10 const attachNamed = h.attachNamed; 10 const attachNamed = h.attachNamed;
11 const ClientSlot = h.ClientSlot; 11 const ClientSlot = h.ClientSlot;
12 const Lead = h.Lead;
12 const pumpUntil = h.pumpUntil; 13 const pumpUntil = h.pumpUntil;
13 const awaitFrame = h.awaitFrame; 14 const awaitFrame = h.awaitFrame;
14 const awaitGridText = h.awaitGridText; 15 const awaitGridText = h.awaitGridText;
@@ -145,21 +146,6 @@ fn pumpUntilReadable(srv: *Server, fd: std.posix.fd_t, buf: []u8, budget_ms: u64
145 return std.posix.read(fd, buf) catch 0; 146 return std.posix.read(fd, buf) catch 0;
146 } 147 }
147 148
148 /// Which of two client slots is the latest-active one. STRICTLY ahead, not
149 /// merely tied: the rule under test is "the most recently active client
150 /// answers", and equal activity would let a wait finish before the lead had
151 /// actually changed hands.
152 const Lead = struct {
153 srv: *Server,
154 ahead: usize,
155 behind: usize,
156 fn taken(self: Lead) bool {
157 const a = self.srv.clients[self.ahead] orelse return false;
158 const b = self.srv.clients[self.behind] orelse return false;
159 return a.activity > b.activity;
160 }
161 };
162
163 /// Test helper: seat a client in `slot` and have it volunteer an agent, 149 /// Test helper: seat a client in `slot` and have it volunteer an agent,
164 /// pumping until the daemon holds both facts. Attaches one at a time by 150 /// pumping until the daemon holds both facts. Attaches one at a time by
165 /// contract — freeClientSlot hands out the lowest free slot, so the caller's 151 /// contract — freeClientSlot hands out the lowest free slot, so the caller's
src/server/server_test_attach.zig
Old New
@@ -12,8 +12,61 @@ const applyFrame = h.applyFrame;
12 const awaitFrame = h.awaitFrame; 12 const awaitFrame = h.awaitFrame;
13 const connectedPair = h.connectedPair; 13 const connectedPair = h.connectedPair;
14 const firstStateFrame = h.firstStateFrame; 14 const firstStateFrame = h.firstStateFrame;
15 const awaitReplicaText = h.awaitReplicaText;
16 const ClientSlot = h.ClientSlot;
17 const Lead = h.Lead;
18 const pumpUntil = h.pumpUntil;
15 const serverThread = h.serverThread; 19 const serverThread = h.serverThread;
16 20
21 /// A sink that replays every frame it is handed into a replica and lets the
22 /// wait run on. For the waits whose answer is one particular reply while the
23 /// session keeps broadcasting underneath it.
24 const ApplyEach = struct {
25 alloc: std.mem.Allocator,
26 replica: *Engine,
27
28 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
29 const self: *@This() = @ptrCast(@alignCast(ctx.?));
30 try applyFrame(self.alloc, self.replica, frame);
31 }
32
33 fn sink(self: *ApplyEach) h.link.Sink {
34 return .{ .ctx = self, .on = ApplyEach.on };
35 }
36 };
37
38 /// Wait until a state frame reports at least `want` rows of history. Both
39 /// frame kinds carry the count, and only the attach is a snapshot, so a
40 /// wait for one kind alone would miss it on the other.
41 fn awaitHistoryRows(
42 alloc: std.mem.Allocator,
43 fd: std.posix.fd_t,
44 want: u32,
45 budget_ms: i64,
46 ) !bool {
47 const Count = struct {
48 want: u32,
49 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
50 const self: *@This() = @ptrCast(@alignCast(ctx.?));
51 const rows: u32 = switch (frame.type) {
52 .snapshot => (try proto.readSnapshotPrefix(frame.payload)).history_rows,
53 .delta => (try proto.readDeltaHeader(frame.payload)).history_rows,
54 else => return,
55 };
56 if (rows >= self.want) return error.HistoryReached;
57 }
58 };
59 var count: Count = .{ .want = want };
60 _ = h.awaitFrameOnSink(alloc, fd, h.never_from_daemon, budget_ms, .{
61 .ctx = &count,
62 .on = Count.on,
63 }) catch |err| switch (err) {
64 error.HistoryReached => return true,
65 else => return err,
66 };
67 return false;
68 }
69
17 test "Server: survives a client that dies without detaching; next attach works" { 70 test "Server: survives a client that dies without detaching; next attach works" {
18 const alloc = std.testing.allocator; 71 const alloc = std.testing.allocator;
19 72
@@ -58,24 +111,7 @@ test "Server: serves scrollback chunks on request" {
58 // Wait until an update reports enough history, then fetch the oldest 111 // Wait until an update reports enough history, then fetch the oldest
59 // page. Only the attach is a snapshot now; the rest are deltas, whose 112 // page. Only the attach is a snapshot now; the rest are deltas, whose
60 // header carries the same history count. 113 // header carries the same history count.
61 var history: u32 = 0; 114 try std.testing.expect(try awaitHistoryRows(alloc, c.handle, 50, 10_000));
62 var deadline_ms: u64 = 10_000;
63 while (deadline_ms > 0 and history < 50) {
64 var pfd = [_]std.posix.pollfd{
65 .{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },
66 };
67 const ready = try std.posix.poll(&pfd, 100);
68 deadline_ms -|= 100;
69 if (ready == 0) continue;
70 const frame = (try proto.readFrame(alloc, c.handle)) orelse break;
71 defer frame.deinit(alloc);
72 switch (frame.type) {
73 .snapshot => history = (try proto.readSnapshotPrefix(frame.payload)).history_rows,
74 .delta => history = (try proto.readDeltaHeader(frame.payload)).history_rows,
75 else => {},
76 }
77 }
78 try std.testing.expect(history >= 50);
79 115
80 try proto.writeFrame(c.handle, .fetch_scrollback, &proto.encodeScrollbackReq(0, 24)); 116 try proto.writeFrame(c.handle, .fetch_scrollback, &proto.encodeScrollbackReq(0, 24));
81 var chunk: ?[]u8 = null; 117 var chunk: ?[]u8 = null;
@@ -127,24 +163,10 @@ test "Server: a full session refuses the next attach instead of displacing anyon
127 try proto.writeFrame(streams[0].handle, .input, "echo still-here\n"); 163 try proto.writeFrame(streams[0].handle, .input, "echo still-here\n");
128 var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 }); 164 var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
129 defer replica.deinit(); 165 defer replica.deinit();
130 var alive = false; 166 try std.testing.expect(try awaitReplicaText(alloc, streams[0].handle, 10_000, .{
131 var deadline_ms: u64 = 10_000; 167 .replica = replica,
132 while (deadline_ms > 0 and !alive) { 168 .needle = "still-here",
133 var pfd = [_]std.posix.pollfd{ 169 }));
134 .{ .fd = streams[0].handle, .events = std.posix.POLL.IN, .revents = 0 },
135 };
136 const ready = try std.posix.poll(&pfd, 100);
137 deadline_ms -|= 100;
138 if (ready == 0) continue;
139 const frame = (try proto.readFrame(alloc, streams[0].handle)) orelse break;
140 defer frame.deinit(alloc);
141 if (frame.type != .snapshot and frame.type != .delta) continue;
142 try applyFrame(alloc, replica, frame);
143 const plain = try replica.dumpPlain(alloc);
144 defer alloc.free(plain);
145 if (std.mem.indexOf(u8, plain, "still-here") != null) alive = true;
146 }
147 try std.testing.expect(alive);
148 } 170 }
149 171
150 test "Server: two clients converge on one session" { 172 test "Server: two clients converge on one session" {
@@ -172,44 +194,19 @@ test "Server: two clients converge on one session" {
172 .{ .fd = a.handle, .rep = replica_a }, 194 .{ .fd = a.handle, .rep = replica_a },
173 .{ .fd = b.handle, .rep = replica_b }, 195 .{ .fd = b.handle, .rep = replica_b },
174 }) |side| { 196 }) |side| {
175 var deadline_ms: u64 = 10_000; 197 try std.testing.expect(try awaitReplicaText(alloc, side.fd, 10_000, .{
176 var seen = false; 198 .replica = side.rep,
177 while (deadline_ms > 0 and !seen) { 199 .needle = "both-see-this",
178 var pfd = [_]std.posix.pollfd{ 200 }));
179 .{ .fd = side.fd, .events = std.posix.POLL.IN, .revents = 0 },
180 };
181 const ready = try std.posix.poll(&pfd, 100);
182 deadline_ms -|= 100;
183 if (ready == 0) continue;
184 const frame = (try proto.readFrame(alloc, side.fd)) orelse break;
185 defer frame.deinit(alloc);
186 try applyFrame(alloc, side.rep, frame);
187 const plain = try side.rep.dumpPlain(alloc);
188 defer alloc.free(plain);
189 if (std.mem.indexOf(u8, plain, "both-see-this") != null) seen = true;
190 }
191 try std.testing.expect(seen);
192 } 201 }
193 202
194 // Byte-level convergence: both replicas match the daemon exactly. 203 // Byte-level convergence: both replicas match the daemon exactly.
195 try proto.writeFrame(a.handle, .debug_dump, &.{1}); 204 try proto.writeFrame(a.handle, .debug_dump, &.{1});
196 var daemon_vt: ?[]u8 = null; 205 var daemon_vt: ?[]u8 = null;
197 defer if (daemon_vt) |d| alloc.free(d); 206 defer if (daemon_vt) |d| alloc.free(d);
198 var deadline_ms: u64 = 5000; 207 var feed_a: ApplyEach = .{ .alloc = alloc, .replica = replica_a };
199 while (deadline_ms > 0 and daemon_vt == null) { 208 if (try h.awaitFrameOnSink(alloc, a.handle, .dump_reply, 5000, feed_a.sink())) |frame| {
200 var pfd = [_]std.posix.pollfd{ 209 daemon_vt = frame.payload; // ownership taken
201 .{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
202 };
203 const ready = try std.posix.poll(&pfd, 100);
204 deadline_ms -|= 100;
205 if (ready == 0) continue;
206 const frame = (try proto.readFrame(alloc, a.handle)) orelse break;
207 if (frame.type == .dump_reply) {
208 daemon_vt = frame.payload; // ownership taken
209 } else {
210 defer frame.deinit(alloc);
211 try applyFrame(alloc, replica_a, frame);
212 }
213 } 210 }
214 try std.testing.expect(daemon_vt != null); 211 try std.testing.expect(daemon_vt != null);
215 212
@@ -219,26 +216,30 @@ test "Server: two clients converge on one session" {
219 216
220 // B saw the same broadcasts but may still have some in its socket 217 // B saw the same broadcasts but may still have some in its socket
221 // buffer: drain until it agrees with the dump A already fetched. 218 // buffer: drain until it agrees with the dump A already fetched.
222 var vb = try replica_b.dumpVt(alloc); 219 const Converge = struct {
220 alloc: std.mem.Allocator,
221 replica: *Engine,
222 want: []const u8,
223 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
224 const self: *@This() = @ptrCast(@alignCast(ctx.?));
225 try applyFrame(self.alloc, self.replica, frame);
226 const vt = try self.replica.dumpVt(self.alloc);
227 defer self.alloc.free(vt);
228 if (std.mem.eql(u8, self.want, vt)) return error.Converged;
229 }
230 };
231 var conv: Converge = .{ .alloc = alloc, .replica = replica_b, .want = daemon_vt.? };
232 _ = h.awaitFrameOnSink(alloc, b.handle, h.never_from_daemon, 5000, .{
233 .ctx = &conv,
234 .on = Converge.on,
235 }) catch |err| switch (err) {
236 error.Converged => {},
237 else => return err,
238 };
239 // Dumped again rather than kept from the sink: the sink BORROWS its
240 // frame, and the comparison it made is the one this reproduces.
241 const vb = try replica_b.dumpVt(alloc);
223 defer alloc.free(vb); 242 defer alloc.free(vb);
224 deadline_ms = 5000;
225 while (deadline_ms > 0 and !std.mem.eql(u8, daemon_vt.?, vb)) {
226 var pfd = [_]std.posix.pollfd{
227 .{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },
228 };
229 const ready = try std.posix.poll(&pfd, 100);
230 deadline_ms -|= 100;
231 if (ready == 0) continue;
232 const frame = (try proto.readFrame(alloc, b.handle)) orelse break;
233 defer frame.deinit(alloc);
234 try applyFrame(alloc, replica_b, frame);
235 // New dump first: freeing vb before the call that replaces it
236 // would leave a dangling pointer for the defer to free if the
237 // dump itself failed.
238 const next = try replica_b.dumpVt(alloc);
239 alloc.free(vb);
240 vb = next;
241 }
242 try std.testing.expectEqualStrings(daemon_vt.?, vb); 243 try std.testing.expectEqualStrings(daemon_vt.?, vb);
243 } 244 }
244 245
@@ -258,6 +259,11 @@ test "Server: a same-size join snapshots the joiner only" {
258 259
259 // Drain A's own join snapshot and everything the shell's startup 260 // Drain A's own join snapshot and everything the shell's startup
260 // produces, so that anything arriving after B joins is B's doing. 261 // produces, so that anything arriving after B joins is B's doing.
262 //
263 // Still its own poll loop, and has to be: the condition is that the
264 // socket went QUIET, and `Link.awaitFrame` does not report that — a
265 // poll that timed out inside it is indistinguishable from the deadline
266 // running out, which is the one fact this loop needs per turn.
261 var a_snapshots: usize = 0; 267 var a_snapshots: usize = 0;
262 var quiet_ms: u64 = 0; 268 var quiet_ms: u64 = 0;
263 var deadline_ms: u64 = 10_000; 269 var deadline_ms: u64 = 10_000;
@@ -294,47 +300,19 @@ test "Server: a same-size join snapshots the joiner only" {
294 // A stays live across the join, by delta: the rebuild B triggered 300 // A stays live across the join, by delta: the rebuild B triggered
295 // bumps the seq A will see next, which A neither notices nor needs. 301 // bumps the seq A will see next, which A neither notices nor needs.
296 try proto.writeFrame(a.handle, .input, "echo join-unicast\n"); 302 try proto.writeFrame(a.handle, .input, "echo join-unicast\n");
297 var a_live = false; 303 try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
298 deadline_ms = 10_000; 304 .replica = replica_a,
299 while (deadline_ms > 0 and !a_live) { 305 .needle = "join-unicast",
300 var pfd = [_]std.posix.pollfd{ 306 .reject = &.{.snapshot},
301 .{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 }, 307 }));
302 };
303 const ready = try std.posix.poll(&pfd, 100);
304 deadline_ms -|= 100;
305 if (ready == 0) continue;
306 const frame = (try proto.readFrame(alloc, a.handle)) orelse break;
307 defer frame.deinit(alloc);
308 try std.testing.expect(frame.type != .snapshot);
309 if (frame.type != .delta) continue;
310 try applyFrame(alloc, replica_a, frame);
311 const plain = try replica_a.dumpPlain(alloc);
312 defer alloc.free(plain);
313 if (std.mem.indexOf(u8, plain, "join-unicast") != null) a_live = true;
314 }
315 try std.testing.expect(a_live);
316 308
317 // ...and B, the joiner, sees the same input. 309 // ...and B, the joiner, sees the same input.
318 var replica_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 }); 310 var replica_b = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
319 defer replica_b.deinit(); 311 defer replica_b.deinit();
320 var b_live = false; 312 try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{
321 deadline_ms = 10_000; 313 .replica = replica_b,
322 while (deadline_ms > 0 and !b_live) { 314 .needle = "join-unicast",
323 var pfd = [_]std.posix.pollfd{ 315 }));
324 .{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },
325 };
326 const ready = try std.posix.poll(&pfd, 100);
327 deadline_ms -|= 100;
328 if (ready == 0) continue;
329 const frame = (try proto.readFrame(alloc, b.handle)) orelse break;
330 defer frame.deinit(alloc);
331 if (frame.type != .snapshot and frame.type != .delta) continue;
332 try applyFrame(alloc, replica_b, frame);
333 const plain = try replica_b.dumpPlain(alloc);
334 defer alloc.free(plain);
335 if (std.mem.indexOf(u8, plain, "join-unicast") != null) b_live = true;
336 }
337 try std.testing.expect(b_live);
338 } 316 }
339 317
340 test "Server: latest attacher's size wins; earlier client is resnapshotted at the new size" { 318 test "Server: latest attacher's size wins; earlier client is resnapshotted at the new size" {
@@ -371,22 +349,7 @@ test "Server: latest attacher's size wins; earlier client is resnapshotted at th
371 defer b.close(); 349 defer b.close();
372 350
373 // The grid follows the newest attacher, and A is told about it. 351 // The grid follows the newest attacher, and A is told about it.
374 var a_resized = false; 352 try std.testing.expect(try awaitSnapshotSize(alloc, a.handle, 100, 30, 5000));
375 var deadline_ms: u64 = 5000;
376 while (deadline_ms > 0 and !a_resized) {
377 var pfd = [_]std.posix.pollfd{
378 .{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
379 };
380 const ready = try std.posix.poll(&pfd, 100);
381 deadline_ms -|= 100;
382 if (ready == 0) continue;
383 const frame = (try proto.readFrame(alloc, a.handle)) orelse break;
384 defer frame.deinit(alloc);
385 if (frame.type != .snapshot) continue;
386 const p = try proto.readSnapshotPrefix(frame.payload);
387 if (p.cols == 100 and p.rows == 30) a_resized = true;
388 }
389 try std.testing.expect(a_resized);
390 353
391 stop.store(true, .release); 354 stop.store(true, .release);
392 th.join(); 355 th.join();
@@ -404,20 +367,27 @@ fn awaitSnapshotSize(
404 rows: u16, 367 rows: u16,
405 timeout_ms: u64, 368 timeout_ms: u64,
406 ) !bool { 369 ) !bool {
407 var deadline_ms = timeout_ms; 370 const Sized = struct {
408 while (deadline_ms > 0) { 371 cols: u16,
409 var pfd = [_]std.posix.pollfd{ 372 rows: u16,
410 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, 373 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
411 }; 374 if (frame.type != .snapshot) return;
412 const ready = try std.posix.poll(&pfd, 100); 375 const self: *@This() = @ptrCast(@alignCast(ctx.?));
413 deadline_ms -|= 100; 376 const p = try proto.readSnapshotPrefix(frame.payload);
414 if (ready == 0) continue; 377 if (p.cols == self.cols and p.rows == self.rows) return error.SizeArrived;
415 const frame = (try proto.readFrame(alloc, fd)) orelse return false; 378 }
416 defer frame.deinit(alloc); 379 };
417 if (frame.type != .snapshot) continue; 380 var sized: Sized = .{ .cols = cols, .rows = rows };
418 const p = try proto.readSnapshotPrefix(frame.payload); 381 // Every frame has to reach the sink: a snapshot at some OTHER size is
419 if (p.cols == cols and p.rows == rows) return true; 382 // exactly what this is lenient about, and `want = .snapshot` would
420 } 383 // return the first one and call it the answer.
384 _ = h.awaitFrameOnSink(alloc, fd, h.never_from_daemon, @intCast(timeout_ms), .{
385 .ctx = &sized,
386 .on = Sized.on,
387 }) catch |err| switch (err) {
388 error.SizeArrived => return true,
389 else => return err,
390 };
421 return false; 391 return false;
422 } 392 }
423 393
@@ -473,38 +443,6 @@ test "Server: typing claims the grid for the typist (latest-wins on input)" {
473 try std.testing.expectEqual(proto.MsgType.delta, first.?.type); 443 try std.testing.expectEqual(proto.MsgType.delta, first.?.type);
474 } 444 }
475 445
476 /// The replica starts blank on purpose — a delta carries every row it changed,
477 /// so the row the marker lands on arrives whole.
478 fn awaitMarkerWithoutSnapshot(
479 alloc: std.mem.Allocator,
480 fd: std.posix.fd_t,
481 cols: u16,
482 rows: u16,
483 marker: []const u8,
484 timeout_ms: u64,
485 ) !bool {
486 var replica = try Engine.init(alloc, .{ .cols = cols, .rows = rows });
487 defer replica.deinit();
488 var deadline_ms = timeout_ms;
489 while (deadline_ms > 0) {
490 var pfd = [_]std.posix.pollfd{
491 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
492 };
493 const ready = try std.posix.poll(&pfd, 100);
494 deadline_ms -|= 100;
495 if (ready == 0) continue;
496 const frame = (try proto.readFrame(alloc, fd)) orelse return false;
497 defer frame.deinit(alloc);
498 if (frame.type == .snapshot) return error.UnexpectedSnapshotBroadcast;
499 if (frame.type != .delta) continue;
500 try applyFrame(alloc, replica, frame);
501 const plain = try replica.dumpPlain(alloc);
502 defer alloc.free(plain);
503 if (std.mem.indexOf(u8, plain, marker) != null) return true;
504 }
505 return false;
506 }
507
508 test "Server: a size the grid refuses never becomes a claim" { 446 test "Server: a size the grid refuses never becomes a claim" {
509 const alloc = std.testing.allocator; 447 const alloc = std.testing.allocator;
510 448
@@ -539,14 +477,18 @@ test "Server: a size the grid refuses never becomes a claim" {
539 // arrives at all is the other half — a refused size must not cost the 477 // arrives at all is the other half — a refused size must not cost the
540 // client its keystrokes. 478 // client its keystrokes.
541 try proto.writeFrame(d.handle, .input, "echo typed-by-refused\n"); 479 try proto.writeFrame(d.handle, .input, "echo typed-by-refused\n");
542 try std.testing.expect(try awaitMarkerWithoutSnapshot( 480 {
543 alloc, 481 // The replica starts blank on purpose — a delta carries every row it
544 a.handle, 482 // changed, so the row the marker lands on arrives whole, and a
545 80, 483 // snapshot is the failure rather than the way the text gets here.
546 24, 484 var seen = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
547 "typed-by-refused", 485 defer seen.deinit();
548 10_000, 486 try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
549 )); 487 .replica = seen,
488 .needle = "typed-by-refused",
489 .reject = &.{.snapshot},
490 }));
491 }
550 492
551 // The reviewer's repro: a real client moves the grid, and then D types 493 // The reviewer's repro: a real client moves the grid, and then D types
552 // again. If D's slot had been stamped with the grid it found at attach 494 // again. If D's slot had been stamped with the grid it found at attach
@@ -558,14 +500,17 @@ test "Server: a size the grid refuses never becomes a claim" {
558 _ = try drainHeld(alloc, d.handle, 10_000); 500 _ = try drainHeld(alloc, d.handle, 10_000);
559 501
560 try proto.writeFrame(d.handle, .input, "echo typed-again\n"); 502 try proto.writeFrame(d.handle, .input, "echo typed-again\n");
561 try std.testing.expect(try awaitMarkerWithoutSnapshot( 503 {
562 alloc, 504 // Blank again, at the size the grid has moved to; a snapshot here
563 a.handle, 505 // would mean D's keystroke claimed a size it was refused.
564 100, 506 var seen = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });
565 30, 507 defer seen.deinit();
566 "typed-again", 508 try std.testing.expect(try awaitReplicaText(alloc, a.handle, 10_000, .{
567 10_000, 509 .replica = seen,
568 )); 510 .needle = "typed-again",
511 .reject = &.{.snapshot},
512 }));
513 }
569 514
570 stop.store(true, .release); 515 stop.store(true, .release);
571 th.join(); 516 th.join();
@@ -597,24 +542,7 @@ test "Server: scrollback fetch is per-client and independent" {
597 try proto.writeFrame(b.handle, .input, "seq 1 100\n"); 542 try proto.writeFrame(b.handle, .input, "seq 1 100\n");
598 543
599 for ([_]std.posix.fd_t{ a.handle, b.handle }) |fd| { 544 for ([_]std.posix.fd_t{ a.handle, b.handle }) |fd| {
600 var history: u32 = 0; 545 try std.testing.expect(try awaitHistoryRows(alloc, fd, 50, 10_000));
601 var deadline_ms: u64 = 10_000;
602 while (deadline_ms > 0 and history < 50) {
603 var pfd = [_]std.posix.pollfd{
604 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
605 };
606 const ready = try std.posix.poll(&pfd, 100);
607 deadline_ms -|= 100;
608 if (ready == 0) continue;
609 const frame = (try proto.readFrame(alloc, fd)) orelse break;
610 defer frame.deinit(alloc);
611 switch (frame.type) {
612 .snapshot => history = (try proto.readSnapshotPrefix(frame.payload)).history_rows,
613 .delta => history = (try proto.readDeltaHeader(frame.payload)).history_rows,
614 else => {},
615 }
616 }
617 try std.testing.expect(history >= 50);
618 } 546 }
619 547
620 // Selection is another client-local read, but unlike a scrollback page 548 // Selection is another client-local read, but unlike a scrollback page
@@ -647,64 +575,65 @@ test "Server: scrollback fetch is per-client and independent" {
647 try proto.writeFrame(a.handle, .selection_req, &one); 575 try proto.writeFrame(a.handle, .selection_req, &one);
648 try proto.writeFrame(a.handle, .selection_req, &sentinel); 576 try proto.writeFrame(a.handle, .selection_req, &sentinel);
649 577
650 var reply_count: usize = 0; 578 const Replies = struct {
651 var deadline_ms: u64 = 5000; 579 n: usize = 0,
652 while (deadline_ms > 0 and reply_count < 4) { 580 fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
653 var pfd = [_]std.posix.pollfd{ 581 if (frame.type != .selection_reply) return;
654 .{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 }, 582 const self: *@This() = @ptrCast(@alignCast(ctx.?));
655 }; 583 const reply = try proto.decodeSelectionReply(frame.payload);
656 const ready = try std.posix.poll(&pfd, 100); 584 switch (self.n) {
657 deadline_ms -|= 100; 585 0 => {
658 if (ready == 0) continue; 586 try std.testing.expectEqual(@as(u32, 77), reply.id);
659 const frame = (try proto.readFrame(alloc, a.handle)) orelse break; 587 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
660 defer frame.deinit(alloc); 588 try std.testing.expectEqualStrings("seq 1 100", reply.text);
661 if (frame.type != .selection_reply) continue; 589 },
662 const reply = try proto.decodeSelectionReply(frame.payload); 590 1 => {
663 switch (reply_count) { 591 try std.testing.expectEqual(@as(u32, 78), reply.id);
664 0 => { 592 try std.testing.expectEqual(proto.SelectionStatus.invalid, reply.status);
665 try std.testing.expectEqual(@as(u32, 77), reply.id); 593 try std.testing.expectEqual(@as(usize, 0), reply.text.len);
666 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status); 594 },
667 try std.testing.expectEqualStrings("seq 1 100", reply.text); 595 2 => {
668 }, 596 try std.testing.expectEqual(@as(u32, 79), reply.id);
669 1 => { 597 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
670 try std.testing.expectEqual(@as(u32, 78), reply.id); 598 try std.testing.expectEqualStrings("1", reply.text);
671 try std.testing.expectEqual(proto.SelectionStatus.invalid, reply.status); 599 },
672 try std.testing.expectEqual(@as(usize, 0), reply.text.len); 600 3 => {
673 }, 601 // Reaching this request proves the malformed frame before
674 2 => { 602 // id 79 was processed. Exactly four replies, in order,
675 try std.testing.expectEqual(@as(u32, 79), reply.id); 603 // proves it generated none of its own without a timeout-only
676 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status); 604 // assertion.
677 try std.testing.expectEqualStrings("1", reply.text); 605 try std.testing.expectEqual(@as(u32, 80), reply.id);
678 }, 606 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
679 3 => { 607 try std.testing.expectEqualStrings("seq 1 100", reply.text);
680 // Reaching this request proves the malformed frame before 608 },
681 // id 79 was processed. Exactly four replies, in order, 609 else => unreachable,
682 // proves it generated none of its own without a timeout-only 610 }
683 // assertion. 611 self.n += 1;
684 try std.testing.expectEqual(@as(u32, 80), reply.id); 612 if (self.n == 4) return error.AllRepliesSeen;
685 try std.testing.expectEqual(proto.SelectionStatus.ok, reply.status);
686 try std.testing.expectEqualStrings("seq 1 100", reply.text);
687 },
688 else => unreachable,
689 } 613 }
690 reply_count += 1; 614 };
691 } 615 var replies: Replies = .{};
692 try std.testing.expectEqual(@as(usize, 4), reply_count); 616 _ = h.awaitFrameOnSink(alloc, a.handle, h.never_from_daemon, 5000, .{
617 .ctx = &replies,
618 .on = Replies.on,
619 }) catch |err| switch (err) {
620 error.AllRepliesSeen => {},
621 else => return err,
622 };
623 try std.testing.expectEqual(@as(usize, 4), replies.n);
693 624
694 try proto.writeFrame(a.handle, .status_req, ""); 625 try proto.writeFrame(a.handle, .status_req, "");
626 const NoSelection = struct {
627 fn on(_: ?*anyopaque, frame: proto.Frame) anyerror!void {
628 // A fifth reply to four requests would mean the malformed frame
629 // generated one of its own.
630 if (frame.type == .selection_reply) return error.LateSelectionReply;
631 }
632 };
695 var status_reply: ?proto.StatusReply = null; 633 var status_reply: ?proto.StatusReply = null;
696 deadline_ms = 5000; 634 if (try h.awaitFrameOnSink(alloc, a.handle, .status_reply, 5000, .{ .on = NoSelection.on })) |frame| {
697 while (deadline_ms > 0 and status_reply == null) {
698 var pfd = [_]std.posix.pollfd{
699 .{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
700 };
701 const ready = try std.posix.poll(&pfd, 100);
702 deadline_ms -|= 100;
703 if (ready == 0) continue;
704 const frame = (try proto.readFrame(alloc, a.handle)) orelse break;
705 defer frame.deinit(alloc); 635 defer frame.deinit(alloc);
706 try std.testing.expect(frame.type != .selection_reply); 636 status_reply = try proto.decodeStatusReply(frame.payload);
707 if (frame.type == .status_reply) status_reply = try proto.decodeStatusReply(frame.payload);
708 } 637 }
709 try std.testing.expect(status_reply != null); 638 try std.testing.expect(status_reply != null);
710 try std.testing.expectEqual(@as(u16, 100), status_reply.?.cols); 639 try std.testing.expectEqual(@as(u16, 100), status_reply.?.cols);
@@ -725,28 +654,13 @@ test "Server: scrollback fetch is per-client and independent" {
725 try proto.writeFrame(b.handle, .input, "echo b-still-live\n"); 654 try proto.writeFrame(b.handle, .input, "echo b-still-live\n");
726 var replica_b = try Engine.init(alloc, .{ .cols = 100, .rows = 30 }); 655 var replica_b = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });
727 defer replica_b.deinit(); 656 defer replica_b.deinit();
728 var b_live = false; 657 // The replica starts empty, so only a snapshot or the deltas after it
729 deadline_ms = 10_000; 658 // can produce the echoed text — and neither of A's answers may appear.
730 while (deadline_ms > 0 and !b_live) { 659 try std.testing.expect(try awaitReplicaText(alloc, b.handle, 10_000, .{
731 var pfd = [_]std.posix.pollfd{ 660 .replica = replica_b,
732 .{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 }, 661 .needle = "b-still-live",
733 }; 662 .reject = &.{ .scrollback_chunk, .selection_reply },
734 const ready = try std.posix.poll(&pfd, 100); 663 }));
735 deadline_ms -|= 100;
736 if (ready == 0) continue;
737 const frame = (try proto.readFrame(alloc, b.handle)) orelse break;
738 defer frame.deinit(alloc);
739 try std.testing.expect(frame.type != .scrollback_chunk);
740 try std.testing.expect(frame.type != .selection_reply);
741 if (frame.type != .snapshot and frame.type != .delta) continue;
742 // The replica starts empty, so only a snapshot or the deltas after
743 // it can produce the echoed text.
744 try applyFrame(alloc, replica_b, frame);
745 const plain = try replica_b.dumpPlain(alloc);
746 defer alloc.free(plain);
747 if (std.mem.indexOf(u8, plain, "b-still-live") != null) b_live = true;
748 }
749 try std.testing.expect(b_live);
750 } 664 }
751 665
752 test "Server: replica rebuilt from snapshots matches the authoritative grid" { 666 test "Server: replica rebuilt from snapshots matches the authoritative grid" {
@@ -768,46 +682,20 @@ test "Server: replica rebuilt from snapshots matches the authoritative grid" {
768 682
769 // Consume the attach snapshot and the deltas that follow it until the 683 // Consume the attach snapshot and the deltas that follow it until the
770 // replica shows the command output. 684 // replica shows the command output.
771 var deadline_ms: u64 = 10_000; 685 try std.testing.expect(try awaitReplicaText(alloc, fd, 10_000, .{
772 var converged = false; 686 .replica = replica,
773 while (deadline_ms > 0 and !converged) { 687 .needle = "fidelity-ok",
774 var pfd = [_]std.posix.pollfd{ 688 }));
775 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
776 };
777 const ready = try std.posix.poll(&pfd, 100);
778 deadline_ms -|= 100;
779 if (ready == 0) continue;
780 const frame = (try proto.readFrame(alloc, fd)) orelse break;
781 defer frame.deinit(alloc);
782 if (frame.type != .snapshot and frame.type != .delta) continue;
783 try applyFrame(alloc, replica, frame);
784 const plain = try replica.dumpPlain(alloc);
785 defer alloc.free(plain);
786 if (std.mem.indexOf(u8, plain, "fidelity-ok") != null) converged = true;
787 }
788 try std.testing.expect(converged);
789 689
790 // Byte-compare replica vs authoritative daemon grid. 690 // Byte-compare replica vs authoritative daemon grid.
791 try proto.writeFrame(fd, .debug_dump, &.{1}); 691 try proto.writeFrame(fd, .debug_dump, &.{1});
792 var daemon_vt: ?[]u8 = null; 692 var daemon_vt: ?[]u8 = null;
793 defer if (daemon_vt) |d| alloc.free(d); 693 defer if (daemon_vt) |d| alloc.free(d);
794 deadline_ms = 10_000; 694 // Late updates may arrive before the reply; the sink applies them so
795 while (deadline_ms > 0 and daemon_vt == null) { 695 // the replica stays current with what the dump will show.
796 var pfd = [_]std.posix.pollfd{ 696 var feed: ApplyEach = .{ .alloc = alloc, .replica = replica };
797 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }, 697 if (try h.awaitFrameOnSink(alloc, fd, .dump_reply, 10_000, feed.sink())) |frame| {
798 }; 698 daemon_vt = frame.payload; // ownership taken
799 const ready = try std.posix.poll(&pfd, 100);
800 deadline_ms -|= 100;
801 if (ready == 0) continue;
802 const frame = (try proto.readFrame(alloc, fd)) orelse break;
803 if (frame.type == .dump_reply) {
804 daemon_vt = frame.payload; // ownership taken
805 } else {
806 // Late updates may arrive before the reply; apply them so the
807 // replica stays current with what the dump will show.
808 try applyFrame(alloc, replica, frame);
809 frame.deinit(alloc);
810 }
811 } 699 }
812 try std.testing.expect(daemon_vt != null); 700 try std.testing.expect(daemon_vt != null);
813 const replica_vt = try replica.dumpVt(alloc); 701 const replica_vt = try replica.dumpVt(alloc);
@@ -837,28 +725,20 @@ test "Server: typing produces deltas, not snapshots; stats track both" {
837 // From here on every update must be a delta: a snapshot means the 725 // From here on every update must be a delta: a snapshot means the
838 // tracker isn't diffing, which is the regression this test guards. 726 // tracker isn't diffing, which is the regression this test guards.
839 try proto.writeFrame(c.handle, .input, "x"); 727 try proto.writeFrame(c.handle, .input, "x");
728 const NoSnapshot = struct {
729 fn on(_: ?*anyopaque, frame: proto.Frame) anyerror!void {
730 if (frame.type != .snapshot) return;
731 std.debug.print(
732 "post-attach snapshot ({d} bytes) where a delta was due\n",
733 .{frame.payload.len},
734 );
735 return error.SnapshotInsteadOfDelta;
736 }
737 };
840 var hdr: ?proto.DeltaHeader = null; 738 var hdr: ?proto.DeltaHeader = null;
841 var deadline_ms: u64 = 10_000; 739 if (try h.awaitFrameOnSink(alloc, c.handle, .delta, 10_000, .{ .on = NoSnapshot.on })) |frame| {
842 while (deadline_ms > 0 and hdr == null) {
843 var pfd = [_]std.posix.pollfd{
844 .{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },
845 };
846 const ready = try std.posix.poll(&pfd, 100);
847 deadline_ms -|= 100;
848 if (ready == 0) continue;
849 const frame = (try proto.readFrame(alloc, c.handle)) orelse break;
850 defer frame.deinit(alloc); 740 defer frame.deinit(alloc);
851 switch (frame.type) { 741 hdr = try proto.readDeltaHeader(frame.payload);
852 .snapshot => {
853 std.debug.print(
854 "post-attach snapshot ({d} bytes) where a delta was due\n",
855 .{frame.payload.len},
856 );
857 return error.SnapshotInsteadOfDelta;
858 },
859 .delta => hdr = try proto.readDeltaHeader(frame.payload),
860 else => {},
861 }
862 } 742 }
863 try std.testing.expect(hdr != null); 743 try std.testing.expect(hdr != null);
864 try std.testing.expect(hdr.?.seq > attach_seq); 744 try std.testing.expect(hdr.?.seq > attach_seq);
@@ -1158,6 +1038,10 @@ const Held = struct { seq: u64, epoch: u64 };
1158 /// Test helper: read an attached connection until it goes quiet, then report 1038 /// Test helper: read an attached connection until it goes quiet, then report
1159 /// what it holds. "Quiet" rather than a frame count because a shell's 1039 /// what it holds. "Quiet" rather than a frame count because a shell's
1160 /// startup output arrives as an unpredictable number of frames. 1040 /// startup output arrives as an unpredictable number of frames.
1041 ///
1042 /// One of the two poll loops left in this file, for the reason the other one
1043 /// carries: `Link.awaitFrame` hides whether a turn ended in a timeout or in
1044 /// a frame, and silence is exactly what this measures.
1161 fn drainHeld(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !Held { 1045 fn drainHeld(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !Held {
1162 var held: Held = .{ .seq = 0, .epoch = 0 }; 1046 var held: Held = .{ .seq = 0, .epoch = 0 };
1163 var quiet_ms: u64 = 0; 1047 var quiet_ms: u64 = 0;
@@ -1315,15 +1199,13 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1315 1199
1316 const ca = try dial.dialAttachNamed(td.sock_path, 80, 24, ""); 1200 const ca = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
1317 defer ca.close(); 1201 defer ca.close();
1318 var spun: usize = 0; 1202 try std.testing.expect(try pumpUntil(&td.srv, 2000, ClientSlot{ .srv = &td.srv, .n = 0 }, ClientSlot.seated));
1319 while (spun < 200 and td.srv.clients[0] == null) : (spun += 1) try td.srv.pumpOnce(5);
1320 1203
1321 // Seated one at a time so the slot indices below are the attach order: 1204 // Seated one at a time so the slot indices below are the attach order:
1322 // freeClientSlot hands out the lowest free slot. 1205 // freeClientSlot hands out the lowest free slot.
1323 const cb = try dial.dialAttachNamed(td.sock_path, 80, 24, ""); 1206 const cb = try dial.dialAttachNamed(td.sock_path, 80, 24, "");
1324 defer cb.close(); 1207 defer cb.close();
1325 spun = 0; 1208 try std.testing.expect(try pumpUntil(&td.srv, 2000, ClientSlot{ .srv = &td.srv, .n = 1 }, ClientSlot.seated));
1326 while (spun < 200 and td.srv.clients[1] == null) : (spun += 1) try td.srv.pumpOnce(5);
1327 1209
1328 const a_attached = (td.srv.clients[0] orelse return error.ClientANeverSeated).activity; 1210 const a_attached = (td.srv.clients[0] orelse return error.ClientANeverSeated).activity;
1329 const b_attached = (td.srv.clients[1] orelse return error.ClientBNeverSeated).activity; 1211 const b_attached = (td.srv.clients[1] orelse return error.ClientBNeverSeated).activity;
@@ -1332,21 +1214,12 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1332 1214
1333 // Typing is activity too, and it takes the lead back. 1215 // Typing is activity too, and it takes the lead back.
1334 try proto.writeFrame(ca.handle, .input, "x"); 1216 try proto.writeFrame(ca.handle, .input, "x");
1335 spun = 0; 1217 try std.testing.expect(try pumpUntil(&td.srv, 2000, Lead{ .srv = &td.srv, .ahead = 0, .behind = 1 }, Lead.taken));
1336 while (spun < 200 and td.srv.clients[0].?.activity == a_attached) : (spun += 1) {
1337 try td.srv.pumpOnce(5);
1338 }
1339 try std.testing.expect(td.srv.clients[0].?.activity > td.srv.clients[1].?.activity);
1340 1218
1341 // Resizing is the third activity verb, and B dragging its window back is 1219 // Resizing is the third activity verb, and B dragging its window back is
1342 // enough to take the lead without B typing a character. 1220 // enough to take the lead without B typing a character.
1343 const a_typed = td.srv.clients[0].?.activity;
1344 try proto.writeFrame(cb.handle, .resize, &proto.encodeSize(100, 30)); 1221 try proto.writeFrame(cb.handle, .resize, &proto.encodeSize(100, 30));
1345 spun = 0; 1222 try std.testing.expect(try pumpUntil(&td.srv, 2000, Lead{ .srv = &td.srv, .ahead = 1, .behind = 0 }, Lead.taken));
1346 while (spun < 200 and td.srv.clients[1].?.activity < a_typed) : (spun += 1) {
1347 try td.srv.pumpOnce(5);
1348 }
1349 try std.testing.expect(td.srv.clients[1].?.activity > td.srv.clients[0].?.activity);
1350 } 1223 }
1351 1224
1352 test "Server: a slot promoted before it attached enters the activity order at its own attach" { 1225 test "Server: a slot promoted before it attached enters the activity order at its own attach" {
src/server/server_test_harness.zig
Old New
@@ -13,8 +13,9 @@ const TmpDir = @import("testtmp").TmpDir;
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 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 15 /// poll+readFrame loops, so a test waiting on a daemon frame waits the way
16 /// the client and `mux a` do. 16 /// the client and `mux a` do. Re-exported like `dial` above: a sibling
17 const link_mod = @import("link"); 17 /// spelling its own `Sink` needs the type, not a second import edge.
18 pub const link = @import("link");
18 const srv_mod = @import("server.zig"); 19 const srv_mod = @import("server.zig");
19 const Server = srv_mod.Server; 20 const Server = srv_mod.Server;
20 21
@@ -65,6 +66,22 @@ pub const ClientSlot = struct {
65 } 66 }
66 }; 67 };
67 68
69 /// Which of two client slots is the latest-active one. STRICTLY ahead, not
70 /// merely tied: the rule under test is "the most recently active client
71 /// answers", and equal activity would let a wait finish before the lead had
72 /// actually changed hands.
73 pub const Lead = struct {
74 srv: *Server,
75 ahead: usize,
76 behind: usize,
77
78 pub fn taken(self: Lead) bool {
79 const a = self.srv.clients[self.ahead] orelse return false;
80 const b = self.srv.clients[self.behind] orelse return false;
81 return a.activity > b.activity;
82 }
83 };
84
68 /// A daemon on a socket of its own, which is what nearly every test in this 85 /// A daemon on a socket of its own, which is what nearly every test in this
69 /// folder opens with: a short-path temp directory (`testtmp`, because a unix 86 /// folder opens with: a short-path temp directory (`testtmp`, because a unix
70 /// socket path caps at 108 bytes), a socket named inside it, and a `Server` 87 /// socket path caps at 108 bytes), a socket named inside it, and a `Server`
@@ -235,7 +252,7 @@ pub fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms:
235 }; 252 };
236 var caught: Catch = .{}; 253 var caught: Catch = .{};
237 // Not `l.close()` on any path: the caller owns this fd and closes it. 254 // Not `l.close()` on any path: the caller owns this fd and closes it.
238 var l: link_mod.Link = .{ .fd = fd }; 255 var l: link.Link = .{ .fd = fd };
239 const frame = l.awaitFrame(alloc, .snapshot, @intCast(timeout_ms), .{ 256 const frame = l.awaitFrame(alloc, .snapshot, @intCast(timeout_ms), .{
240 .ctx = &caught, 257 .ctx = &caught,
241 .on = Catch.on, 258 .on = Catch.on,
@@ -325,7 +342,7 @@ pub fn awaitFrame(
325 iters: usize, 342 iters: usize,
326 ) !?proto.Frame { 343 ) !?proto.Frame {
327 // Not `l.close()` on any path: the caller owns this fd and closes it. 344 // Not `l.close()` on any path: the caller owns this fd and closes it.
328 var l: link_mod.Link = .{ .fd = fd }; 345 var l: link.Link = .{ .fd = fd };
329 var i: usize = 0; 346 var i: usize = 0;
330 while (i < iters) : (i += 1) { 347 while (i < iters) : (i += 1) {
331 try srv.pumpOnce(5); 348 try srv.pumpOnce(5);
@@ -377,10 +394,10 @@ pub fn awaitFrameOnSink(
377 fd: std.posix.fd_t, 394 fd: std.posix.fd_t,
378 want: proto.MsgType, 395 want: proto.MsgType,
379 timeout_ms: i64, 396 timeout_ms: i64,
380 sink: link_mod.Sink, 397 sink: link.Sink,
381 ) !?proto.Frame { 398 ) !?proto.Frame {
382 // Not `l.close()` on any path: the caller owns this fd and closes it. 399 // Not `l.close()` on any path: the caller owns this fd and closes it.
383 var l: link_mod.Link = .{ .fd = fd }; 400 var l: link.Link = .{ .fd = fd };
384 return l.awaitFrame(alloc, want, @intCast(@max(timeout_ms, 0)), sink) catch |err| switch (err) { 401 return l.awaitFrame(alloc, want, @intCast(@max(timeout_ms, 0)), sink) catch |err| switch (err) {
385 // The budget running out and the peer going away were both null in 402 // The budget running out and the peer going away were both null in
386 // the hand-rolled loop this replaces, and every caller reads null as 403 // the hand-rolled loop this replaces, and every caller reads null as
@@ -390,6 +407,65 @@ pub fn awaitFrameOnSink(
390 }; 407 };
391 } 408 }
392 409
410 /// A frame type the daemon never SENDS. Naming it as `want` is how a wait
411 /// whose answer comes out of the SINK says so: nothing can match, so every
412 /// frame reaches the sink and the wait ends on the sink's error or on the
413 /// deadline. `.attach` is a client's first word on the wire — the daemon
414 /// reads them and writes none.
415 pub const never_from_daemon: proto.MsgType = .attach;
416
417 /// What `awaitReplicaText` is waiting for on one connection.
418 pub const ReplicaWait = struct {
419 /// Replayed through `applyFrame`, so through the production replica.
420 replica: *Engine,
421 /// The text the grid must show. Found in `dumpPlain`, not in the frame
422 /// bytes: a row can arrive spread over several deltas, and the grid is
423 /// what a user would see.
424 needle: []const u8,
425 /// Frame types whose mere ARRIVAL fails the wait — a snapshot where the
426 /// tracker owed a delta, a scrollback chunk on a client that asked for
427 /// none. `error.RejectedFrameArrived` rather than a per-caller error
428 /// name, because the type is in the trace either way.
429 reject: []const proto.MsgType = &.{},
430 };
431
432 /// 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
434 /// eight times in server_test_attach.zig alone, each copy re-deciding what
435 /// a poll timeout meant.
436 pub fn awaitReplicaText(
437 alloc: std.mem.Allocator,
438 fd: std.posix.fd_t,
439 budget_ms: i64,
440 w: ReplicaWait,
441 ) !bool {
442 const Feed = struct {
443 alloc: std.mem.Allocator,
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,
464 else => return err,
465 };
466 return false;
467 }
468
393 /// The liveness half: "no mark arrived" is worthless against a shell that 469 /// The liveness half: "no mark arrived" is worthless against a shell that
394 /// never started. 470 /// never started.
395 pub fn awaitGridText( 471 pub fn awaitGridText(