753c63d8 — mux

src/cli/main.zig

Old		New
@@ -543,6 +543,13 @@ fn resumeRun(alloc: std.mem.Allocator, o: Opts, resume_fd: std.posix.fd_t) !u8 {
543	return try srv.run();	543	return try srv.run();
544	}	544	}
545		545
		546	/// `muxd run`: the daemon, in the foreground. Every code this function picks
		547	/// is 1, and every one of them is a boot failure — an operator mistake caught
		548	/// before anything bound. Reaching `srv.run()` means the daemon served, and
		549	/// it answers 0 whenever something asks it to stop. No session's exit is ever
		550	/// reported here: a shell's code goes to that shell's own clients
		551	/// (`exit_status`), and an emptied daemon is one with nothing on it rather
		552	/// than one that is leaving.
546	fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {	553	fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {
547	// Address and key are settled before anything binds: a mistyped address	554	// Address and key are settled before anything binds: a mistyped address
548	// or an unreadable key must not first leave a session socket and a live	555	// or an unreadable key must not first leave a session socket and a live

src/server.zig

Old		New
@@ -998,11 +998,12 @@ pub const Server = struct {
998	}	998	}
999		999
1000	/// One poll iteration. A session whose shell has exited is torn down	1000	/// One poll iteration. A session whose shell has exited is torn down
1001	/// first — its clients told and dropped — and the pump carries on; only	1001	/// first — its clients told and dropped — and the pump carries on. It
1002	/// the LAST session's exit ends it, returning that shell's code. Null	1002	/// answers nothing: no session's death ends the daemon, not even the
1003	/// while any session lives.	1003	/// last one's, so an emptied daemon keeps serving its socket and a
1004	pub fn pumpOnce(self: *Server, timeout_ms: i32) !?u8 {	1004	/// fresh attach is born into it.
1005	if (self.sessions.reap(self)) \|code\| return code;	1005	pub fn pumpOnce(self: *Server, timeout_ms: i32) !void {
		1006	self.sessions.reap(self);
1006		1007
1007	const listener_idx = max_sessions;	1008	const listener_idx = max_sessions;
1008	const client_base = max_sessions + 1;	1009	const client_base = max_sessions + 1;
@@ -1186,10 +1187,13 @@ pub const Server = struct {
1186	// handling above, or everything queued during this pump would wait	1187	// handling above, or everything queued during this pump would wait
1187	// for the next one and every reply would cost a poll cycle.	1188	// for the next one and every reply would cost a poll cycle.
1188	if (self.quicListener()) \|q\| q.drainAll();	1189	if (self.quicListener()) \|q\| q.drainAll();
1189
1190	return null;
1191	}	1190	}
1192		1191
		1192	/// Pumps until something asks the daemon to stop. The only nonzero exit
		1193	/// `muxd run` has is a boot failure in main; getting here means the
		1194	/// daemon served, so the answer is 0 whether the ask was `muxd stop` or
		1195	/// a SIGTERM from a supervisor — which reads a nonzero exit on a clean
		1196	/// shutdown as a crash.
1193	pub fn run(self: *Server) !u8 {	1197	pub fn run(self: *Server) !u8 {
1194	// The pump cannot start on a listener whose handler points anywhere	1198	// The pump cannot start on a listener whose handler points anywhere
1195	// but here. An adopted listener (initFromManifest) is built before	1199	// but here. An adopted listener (initFromManifest) is built before
@@ -1197,7 +1201,7 @@ pub const Server = struct {
1197	// caller already wired is the same assignment twice.	1201	// caller already wired is the same assignment twice.
1198	if (self.quicListener()) \|l\| l.setHandler(self.quicHandler());	1202	if (self.quicListener()) \|l\| l.setHandler(self.quicHandler());
1199	while (true) {	1203	while (true) {
1200	if (shutdown_flag.load(.acquire)) return 130;	1204	if (shutdown_flag.load(.acquire)) return 0;
1201	// An upgrade was accepted: the reply has drained (pumpOnce	1205	// An upgrade was accepted: the reply has drained (pumpOnce
1202	// ran the observer handler), so exec now. Shaped like	1206	// ran the observer handler), so exec now. Shaped like
1203	// shutdown_flag but per-instance because the exec carries the	1207	// shutdown_flag but per-instance because the exec carries the
@@ -1207,7 +1211,7 @@ pub const Server = struct {
1207	// execUpgrade only returns on failure; the daemon carries on.	1211	// execUpgrade only returns on failure; the daemon carries on.
1208	continue;	1212	continue;
1209	}	1213	}
1210	if (try self.pumpOnce(100)) \|code\| return code;	1214	try self.pumpOnce(100);
1211	}	1215	}
1212	}	1216	}
1213		1217

src/server_sessions.zig

Old		New
@@ -184,11 +184,11 @@ pub const SessionTable = struct {
184	}	184	}
185		185
186	/// Tear down every session whose shell has exited: its clients told and	186	/// Tear down every session whose shell has exited: its clients told and
187	/// dropped, its slot nulled, its name freed. Returns the daemon's exit	187	/// dropped, its slot nulled, its name freed. Answers nothing, because a
188	/// code — the last dead shell's — once no session remains; null while	188	/// shell's exit code is a fact about that shell and never about the
189	/// any lives.	189	/// daemon: an emptied table is a daemon with nothing on it, not a daemon
190	pub fn reap(self: SessionTable, srv: Server) ?u8 {	190	/// that is leaving. `muxd stop` is the end (decisions.md).
191	var last_code: ?u8 = null;	191	pub fn reap(self: SessionTable, srv: Server) void {
192	for (&self.table, 0..) \|*slot, si\| {	192	for (&self.table, 0..) \|*slot, si\| {
193	const s = if (slot.) \|sp\| sp else continue;	193	const s = if (slot.) \|sp\| sp else continue;
194	const exited = s.pty.checkExited();	194	const exited = s.pty.checkExited();
@@ -263,23 +263,8 @@ pub const SessionTable = struct {
263	// to be told.	263	// to be told.
264	srv.agents.closeOfSession(srv, si);	264	srv.agents.closeOfSession(srv, si);
265	slot.* = null;	265	slot.* = null;
266	last_code = @intCast(code & 0xff);
267	}	266	}
268	}	267	}
269	// Recomputed, never counted as we went. `drainPending` above runs
270	// the QUIC read path, which reaches handleFrame → resolve →
271	// create: a session can therefore be BORN inside this loop,
272	// and if it lands in a slot the loop has already walked, an
273	// incremental counter never sees it. The daemon would then exit
274	// with a shell it forked moments earlier still running. Re-reading
275	// the table costs nothing and cannot be wrong.
276	if (self.live() == 0) {
277	// Nested rather than one `and`: no live sessions and no code
278	// means there were never any — unreachable after init, but this
279	// shape keeps it from ever being misread as exit 0.
280	if (last_code) \|code\| return code;
281	}
282	return null;
283	}	268	}
284		269
285	/// The `sessions_reply` payload: live names, '\n'-separated, slot order.	270	/// The `sessions_reply` payload: live names, '\n'-separated, slot order.

src/server_test_agent.zig

Old		New
@@ -82,7 +82,7 @@ test "Server: a reaped session takes its agent socket with it" {
82	try proto.writeAllFd(srv.ses(0).pty.master, "die\n");	82	try proto.writeAllFd(srv.ses(0).pty.master, "die\n");
83	var reaped = false;	83	var reaped = false;
84	for (0..600) \|_\| {	84	for (0..600) \|_\| {
85	_ = srv.pumpOnce(5) catch break;	85	srv.pumpOnce(5) catch break;
86	if (srv.sessions.table[0] == null) {	86	if (srv.sessions.table[0] == null) {
87	reaped = true;	87	reaped = true;
88	break;	88	break;
@@ -110,7 +110,7 @@ test "Server: agent_offer flags the slot, and an unknown type leaves the client
110	defer c.close();	110	defer c.close();
111	try attachNamed(c.handle, 80, 24, "");	111	try attachNamed(c.handle, 80, 24, "");
112	var spun: usize = 0;	112	var spun: usize = 0;
113	while (spun < 200 and srv.clients[0] == null) : (spun += 1) _ = try srv.pumpOnce(5);	113	while (spun < 200 and srv.clients[0] == null) : (spun += 1) try srv.pumpOnce(5);
114	try std.testing.expect(srv.clients[0] != null);	114	try std.testing.expect(srv.clients[0] != null);
115	// The offer is opt-in, so the flag must start false or every client	115	// The offer is opt-in, so the flag must start false or every client
116	// would look like it had offered.	116	// would look like it had offered.
@@ -118,7 +118,7 @@ test "Server: agent_offer flags the slot, and an unknown type leaves the client
118		118
119	try proto.writeFrame(c.handle, .agent_offer, "");	119	try proto.writeFrame(c.handle, .agent_offer, "");
120	spun = 0;	120	spun = 0;
121	while (spun < 200 and !srv.clients[0].?.agent_offer) : (spun += 1) _ = try srv.pumpOnce(5);	121	while (spun < 200 and !srv.clients[0].?.agent_offer) : (spun += 1) try srv.pumpOnce(5);
122	try std.testing.expect(srv.clients[0].?.agent_offer);	122	try std.testing.expect(srv.clients[0].?.agent_offer);
123		123
124	// 0x7e is unmapped in MsgType. Skipping it rather than dropping the	124	// 0x7e is unmapped in MsgType. Skipping it rather than dropping the
@@ -148,7 +148,7 @@ test "Server: agent_offer flags the slot, and an unknown type leaves the client
148	fn pumpUntilReadable(srv: *Server, fd: std.posix.fd_t, buf: []u8, iters: usize) !?usize {	148	fn pumpUntilReadable(srv: *Server, fd: std.posix.fd_t, buf: []u8, iters: usize) !?usize {
149	var i: usize = 0;	149	var i: usize = 0;
150	while (i < iters) : (i += 1) {	150	while (i < iters) : (i += 1) {
151	_ = try srv.pumpOnce(5);	151	try srv.pumpOnce(5);
152	var pfd = [_]std.posix.pollfd{	152	var pfd = [_]std.posix.pollfd{
153	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	153	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
154	};	154	};
@@ -165,11 +165,11 @@ fn pumpUntilReadable(srv: *Server, fd: std.posix.fd_t, buf: []u8, iters: usize)
165	fn attachOffering(srv: *Server, fd: std.posix.fd_t, slot: usize, name: []const u8) !void {	165	fn attachOffering(srv: *Server, fd: std.posix.fd_t, slot: usize, name: []const u8) !void {
166	try attachNamed(fd, 80, 24, name);	166	try attachNamed(fd, 80, 24, name);
167	var spun: usize = 0;	167	var spun: usize = 0;
168	while (spun < 200 and srv.clients[slot] == null) : (spun += 1) _ = try srv.pumpOnce(5);	168	while (spun < 200 and srv.clients[slot] == null) : (spun += 1) try srv.pumpOnce(5);
169	if (srv.clients[slot] == null) return error.ClientNeverSeated;	169	if (srv.clients[slot] == null) return error.ClientNeverSeated;
170	try proto.writeFrame(fd, .agent_offer, "");	170	try proto.writeFrame(fd, .agent_offer, "");
171	spun = 0;	171	spun = 0;
172	while (spun < 200 and !srv.clients[slot].?.agent_offer) : (spun += 1) _ = try srv.pumpOnce(5);	172	while (spun < 200 and !srv.clients[slot].?.agent_offer) : (spun += 1) try srv.pumpOnce(5);
173	if (!srv.clients[slot].?.agent_offer) return error.OfferNeverLanded;	173	if (!srv.clients[slot].?.agent_offer) return error.OfferNeverLanded;
174	}	174	}
175		175
@@ -225,14 +225,14 @@ test "Server: a session with no agent socket does not inherit the daemon's" {
225	defer c.close();	225	defer c.close();
226	try attachNamed(c.handle, 80, 24, "nosock");	226	try attachNamed(c.handle, 80, 24, "nosock");
227	var spun: usize = 0;	227	var spun: usize = 0;
228	while (spun < 200 and srv.clients[0] == null) : (spun += 1) _ = try srv.pumpOnce(5);	228	while (spun < 200 and srv.clients[0] == null) : (spun += 1) try srv.pumpOnce(5);
229	const si = srv.clients[0].?.session orelse return error.ClientNeverSeated;	229	const si = srv.clients[0].?.session orelse return error.ClientNeverSeated;
230	try std.testing.expect(srv.sessions.table[si].?.agent_path == null);	230	try std.testing.expect(srv.sessions.table[si].?.agent_path == null);
231		231
232	try proto.writeFrame(c.handle, .input, cmd);	232	try proto.writeFrame(c.handle, .input, cmd);
233	spun = 0;	233	spun = 0;
234	while (spun < 600) : (spun += 1) {	234	while (spun < 600) : (spun += 1) {
235	_ = try srv.pumpOnce(5);	235	try srv.pumpOnce(5);
236	std.fs.accessAbsolute(done, .{}) catch continue;	236	std.fs.accessAbsolute(done, .{}) catch continue;
237	break;	237	break;
238	}	238	}
@@ -269,7 +269,7 @@ test "Server: a full channel table refuses the newest dial and says so once" {
269	defer for (dials) \|d\| d.close();	269	defer for (dials) \|d\| d.close();
270		270
271	var spun: usize = 0;	271	var spun: usize = 0;
272	while (spun < 400 and srv.agents.refused_full == 0) : (spun += 1) _ = try srv.pumpOnce(5);	272	while (spun < 400 and srv.agents.refused_full == 0) : (spun += 1) try srv.pumpOnce(5);
273	for (srv.agents.chans) \|slot\| try std.testing.expect(slot != null);	273	for (srv.agents.chans) \|slot\| try std.testing.expect(slot != null);
274	try std.testing.expectEqual(@as(u32, 1), srv.agents.refused_full);	274	try std.testing.expectEqual(@as(u32, 1), srv.agents.refused_full);
275	try std.testing.expectEqual(@as(u32, 0), srv.agents.refused_no_offer);	275	try std.testing.expectEqual(@as(u32, 0), srv.agents.refused_no_offer);
@@ -296,7 +296,7 @@ test "Server: a full channel table refuses the newest dial and says so once" {
296	const again = try std.net.connectUnixSocket(path);	296	const again = try std.net.connectUnixSocket(path);
297	defer again.close();	297	defer again.close();
298	spun = 0;	298	spun = 0;
299	while (spun < 400 and srv.agents.refused_full < 2) : (spun += 1) _ = try srv.pumpOnce(5);	299	while (spun < 400 and srv.agents.refused_full < 2) : (spun += 1) try srv.pumpOnce(5);
300	try std.testing.expectEqual(@as(u32, 2), srv.agents.refused_full);	300	try std.testing.expectEqual(@as(u32, 2), srv.agents.refused_full);
301	try std.testing.expect(srv.agents.full_said);	301	try std.testing.expect(srv.agents.full_said);
302	srv.agents.closeChan(&srv, 0, .notify);	302	srv.agents.closeChan(&srv, 0, .notify);
@@ -318,7 +318,7 @@ test "Server: an agent connection with nobody offering is refused fast" {
318	defer c.close();	318	defer c.close();
319	try attachNamed(c.handle, 80, 24, "");	319	try attachNamed(c.handle, 80, 24, "");
320	var spun: usize = 0;	320	var spun: usize = 0;
321	while (spun < 200 and srv.clients[0] == null) : (spun += 1) _ = try srv.pumpOnce(5);	321	while (spun < 200 and srv.clients[0] == null) : (spun += 1) try srv.pumpOnce(5);
322	try std.testing.expect(srv.clients[0] != null);	322	try std.testing.expect(srv.clients[0] != null);
323	// Attached but never offered — the case this test is about. The socket	323	// Attached but never offered — the case this test is about. The socket
324	// exists for as long as the session does; only the answer comes and goes.	324	// exists for as long as the session does; only the answer comes and goes.
@@ -431,7 +431,7 @@ test "Server: an agent connection is routed to the latest-active offerer" {
431	try proto.writeFrame(cb.handle, .input, "x");	431	try proto.writeFrame(cb.handle, .input, "x");
432	var spun: usize = 0;	432	var spun: usize = 0;
433	while (spun < 200 and srv.clients[1].?.activity < srv.clients[0].?.activity) : (spun += 1) {	433	while (spun < 200 and srv.clients[1].?.activity < srv.clients[0].?.activity) : (spun += 1) {
434	_ = try srv.pumpOnce(5);	434	try srv.pumpOnce(5);
435	}	435	}
436	try std.testing.expect(srv.clients[1].?.activity > srv.clients[0].?.activity);	436	try std.testing.expect(srv.clients[1].?.activity > srv.clients[0].?.activity);
437		437
@@ -454,7 +454,7 @@ test "Server: an agent connection is routed to the latest-active offerer" {
454	try proto.writeFrame(ca.handle, .input, "y");	454	try proto.writeFrame(ca.handle, .input, "y");
455	spun = 0;	455	spun = 0;
456	while (spun < 200 and srv.clients[0].?.activity < srv.clients[1].?.activity) : (spun += 1) {	456	while (spun < 200 and srv.clients[0].?.activity < srv.clients[1].?.activity) : (spun += 1) {
457	_ = try srv.pumpOnce(5);	457	try srv.pumpOnce(5);
458	}	458	}
459	try std.testing.expect(srv.clients[0].?.activity > srv.clients[1].?.activity);	459	try std.testing.expect(srv.clients[0].?.activity > srv.clients[1].?.activity);
460		460
@@ -726,7 +726,7 @@ test "Server: a QUIC client's agent channels die with the client" {
726	cl.drain();	726	cl.drain();
727		727
728	var only = [_]*quic_server.TestClient{&cl};	728	var only = [_]*quic_server.TestClient{&cl};
729	_ = try quicPump(&srv, &only, 10000, &srv, struct {	729	try quicPump(&srv, &only, 10000, &srv, struct {
730	fn f(s: *Server) bool {	730	fn f(s: *Server) bool {
731	return s.clients[0] != null and s.clients[0].?.agent_offer;	731	return s.clients[0] != null and s.clients[0].?.agent_offer;
732	}	732	}
@@ -736,7 +736,7 @@ test "Server: a QUIC client's agent channels die with the client" {
736	const path = srv.ses(0).agent_path orelse return error.NoAgentSocket;	736	const path = srv.ses(0).agent_path orelse return error.NoAgentSocket;
737	const agent = try std.net.connectUnixSocket(path);	737	const agent = try std.net.connectUnixSocket(path);
738	defer agent.close();	738	defer agent.close();
739	_ = try quicPump(&srv, &only, 10000, &cl, struct {	739	try quicPump(&srv, &only, 10000, &cl, struct {
740	fn f(t: *quic_server.TestClient) bool {	740	fn f(t: *quic_server.TestClient) bool {
741	return findFrame(t.recv_buf[0..t.recv_len], .agent_open) != null;	741	return findFrame(t.recv_buf[0..t.recv_len], .agent_open) != null;
742	}	742	}

src/server_test_attach.zig

src/server_test_await.zig

Old		New
@@ -107,7 +107,7 @@ fn anyReturnWithin(
107	) !?proto.CmdState {	107	) !?proto.CmdState {
108	const deadline = std.time.milliTimestamp() + budget_ms;	108	const deadline = std.time.milliTimestamp() + budget_ms;
109	while (std.time.milliTimestamp() < deadline) {	109	while (std.time.milliTimestamp() < deadline) {
110	_ = try srv.pumpOnce(5);	110	try srv.pumpOnce(5);
111	var pfd = [_]std.posix.pollfd{	111	var pfd = [_]std.posix.pollfd{
112	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	112	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
113	};	113	};
@@ -555,7 +555,7 @@ test "Server: an await is held open, answered by a mark, and re-answered immedia
555	.settle_ms = 0,	555	.settle_ms = 0,
556	.timeout_ms = 5000,	556	.timeout_ms = 5000,
557	}));	557	}));
558	_ = try srv.pumpOnce(5);	558	try srv.pumpOnce(5);
559	const f2 = (try readQueued(alloc, c.handle, .await_reply)) orelse	559	const f2 = (try readQueued(alloc, c.handle, .await_reply)) orelse
560	return error.AwaitNotAnsweredOnTheSamePump;	560	return error.AwaitNotAnsweredOnTheSamePump;
561	defer f2.deinit(alloc);	561	defer f2.deinit(alloc);
@@ -625,7 +625,7 @@ test "Server: a return is still answerable once the next command is running" {
625	try proto.writeFrame(c.handle, .input, "go\n");	625	try proto.writeFrame(c.handle, .input, "go\n");
626	var spun: usize = 0;	626	var spun: usize = 0;
627	while (spun < 500 and srv.sessions.table[0].?.cmd.phase != .running) : (spun += 1) {	627	while (spun < 500 and srv.sessions.table[0].?.cmd.phase != .running) : (spun += 1) {
628	_ = try srv.pumpOnce(5);	628	try srv.pumpOnce(5);
629	}	629	}
630	try std.testing.expectEqual(proto.CmdPhase.running, srv.sessions.table[0].?.cmd.phase);	630	try std.testing.expectEqual(proto.CmdPhase.running, srv.sessions.table[0].?.cmd.phase);
631		631
@@ -863,7 +863,7 @@ test "Server: re-attaching to another session drops the await it left behind" {
863	defer c.close();	863	defer c.close();
864	try attachNamed(c.handle, 80, 24, "a");	864	try attachNamed(c.handle, 80, 24, "a");
865	var spun: usize = 0;	865	var spun: usize = 0;
866	while (spun < 200 and srv.sessions.find("a") == null) : (spun += 1) _ = try srv.pumpOnce(5);	866	while (spun < 200 and srv.sessions.find("a") == null) : (spun += 1) try srv.pumpOnce(5);
867	const si_a = srv.sessions.find("a") orelse return error.NoSessionA;	867	const si_a = srv.sessions.find("a") orelse return error.NoSessionA;
868		868
869	// An await is a question about ONE session's seq series: since_seq is a	869	// An await is a question about ONE session's seq series: since_seq is a
@@ -875,7 +875,7 @@ test "Server: re-attaching to another session drops the await it left behind" {
875	}));	875	}));
876	spun = 0;	876	spun = 0;
877	while (spun < 200) : (spun += 1) {	877	while (spun < 200) : (spun += 1) {
878	_ = try srv.pumpOnce(5);	878	try srv.pumpOnce(5);
879	if (slotAwaiting(&srv, si_a)) break;	879	if (slotAwaiting(&srv, si_a)) break;
880	}	880	}
881	try std.testing.expect(slotAwaiting(&srv, si_a));	881	try std.testing.expect(slotAwaiting(&srv, si_a));
@@ -885,7 +885,7 @@ test "Server: re-attaching to another session drops the await it left behind" {
885	// last_return — an await that answers instantly or never, arbitrarily.	885	// last_return — an await that answers instantly or never, arbitrarily.
886	try attachNamed(c.handle, 80, 24, "b");	886	try attachNamed(c.handle, 80, 24, "b");
887	spun = 0;	887	spun = 0;
888	while (spun < 200 and srv.sessions.find("b") == null) : (spun += 1) _ = try srv.pumpOnce(5);	888	while (spun < 200 and srv.sessions.find("b") == null) : (spun += 1) try srv.pumpOnce(5);
889	const si_b = srv.sessions.find("b") orelse return error.NoSessionB;	889	const si_b = srv.sessions.find("b") orelse return error.NoSessionB;
890	try std.testing.expect(si_a != si_b);	890	try std.testing.expect(si_a != si_b);
891		891
@@ -1025,7 +1025,7 @@ test "Server: an await resolves against the awaiting client's session" {
1025	try proto.writeFrame(b.handle, .input, "go\n");	1025	try proto.writeFrame(b.handle, .input, "go\n");
1026	var spun: usize = 0;	1026	var spun: usize = 0;
1027	while (spun < 500 and srv.sessions.table[si_b].?.last_return == null) : (spun += 1) {	1027	while (spun < 500 and srv.sessions.table[si_b].?.last_return == null) : (spun += 1) {
1028	_ = try srv.pumpOnce(5);	1028	try srv.pumpOnce(5);
1029	}	1029	}
1030	try std.testing.expect(srv.sessions.table[si_b].?.last_return != null);	1030	try std.testing.expect(srv.sessions.table[si_b].?.last_return != null);
1031		1031
@@ -1070,7 +1070,7 @@ test "Server: a promoted-but-unattached slot receives nothing" {
1070	// emits: grid updates from the echo, and the mode poll's first report.	1070	// emits: grid updates from the echo, and the mode poll's first report.
1071	try proto.writeAllFd(srv.sessions.table[0].?.pty.master, "say-something\n");	1071	try proto.writeAllFd(srv.sessions.table[0].?.pty.master, "say-something\n");
1072	var i: usize = 0;	1072	var i: usize = 0;
1073	while (i < 40) : (i += 1) _ = try srv.pumpOnce(5);	1073	while (i < 40) : (i += 1) try srv.pumpOnce(5);
1074		1074
1075	// Liveness: the grid really moved, so silence below is the filter at	1075	// Liveness: the grid really moved, so silence below is the filter at
1076	// work and not a session that never spoke.	1076	// work and not a session that never spoke.

src/server_test_clipboard.zig

Old		New
@@ -21,7 +21,7 @@ fn awaitSelectionReply(
21	) !?struct { id: u32, status: proto.SelectionStatus, text_len: usize } {	21	) !?struct { id: u32, status: proto.SelectionStatus, text_len: usize } {
22	var rounds: usize = 0;	22	var rounds: usize = 0;
23	while (rounds < 400) : (rounds += 1) {	23	while (rounds < 400) : (rounds += 1) {
24	_ = try srv.pumpOnce(5);	24	try srv.pumpOnce(5);
25	var pfd = [_]std.posix.pollfd{	25	var pfd = [_]std.posix.pollfd{
26	.{ .fd = peer, .events = std.posix.POLL.IN, .revents = 0 },	26	.{ .fd = peer, .events = std.posix.POLL.IN, .revents = 0 },
27	};	27	};
@@ -271,7 +271,7 @@ fn clipboardIntoGap(
271	a.close();	271	a.close();
272	var gone = false;	272	var gone = false;
273	for (0..400) \|_\| {	273	for (0..400) \|_\| {
274	_ = try srv.pumpOnce(5);	274	try srv.pumpOnce(5);
275	if (!srv.hasClientsIn(0)) {	275	if (!srv.hasClientsIn(0)) {
276	gone = true;	276	gone = true;
277	break;	277	break;
@@ -338,7 +338,7 @@ fn collectGapReplay(alloc: std.mem.Allocator, srv: *Server, fd: std.posix.fd_t)
338	var after: usize = 0;	338	var after: usize = 0;
339	while (i < 400 and after < 60) : (i += 1) {	339	while (i < 400 and after < 60) : (i += 1) {
340	if (out.content != null) after += 1;	340	if (out.content != null) after += 1;
341	_ = try srv.pumpOnce(5);	341	try srv.pumpOnce(5);
342	var pfd = [_]std.posix.pollfd{	342	var pfd = [_]std.posix.pollfd{
343	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	343	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
344	};	344	};
@@ -709,7 +709,7 @@ test "Server: a session that dies holding a pending event frees it" {
709	try proto.writeAllFd(srv.sessions.table[0].?.pty.master, "die\n");	709	try proto.writeAllFd(srv.sessions.table[0].?.pty.master, "die\n");
710	var reaped = false;	710	var reaped = false;
711	for (0..600) \|_\| {	711	for (0..600) \|_\| {
712	_ = srv.pumpOnce(5) catch break;	712	srv.pumpOnce(5) catch break;
713	if (srv.sessions.table[0] == null) {	713	if (srv.sessions.table[0] == null) {
714	reaped = true;	714	reaped = true;
715	break;	715	break;

src/server_test_deliver.zig

Old		New
@@ -178,7 +178,7 @@ test "Server: a writable backlog is flushed by poll, not mistaken for input" {
178	var drain: [16 * 1024]u8 = undefined;	178	var drain: [16 * 1024]u8 = undefined;
179	_ = try std.posix.read(c.peer, &drain);	179	_ = try std.posix.read(c.peer, &drain);
180		180
181	_ = try srv.pumpOnce(50);	181	try srv.pumpOnce(50);
182		182
183	// The other half: the pump used the POLLOUT for what it was.	183	// The other half: the pump used the POLLOUT for what it was.
184	try std.testing.expect(srv.clients[0] != null);	184	try std.testing.expect(srv.clients[0] != null);

src/server_test_harness.zig

Old		New
@@ -10,8 +10,7 @@ const Server = srv_mod.Server;
10		10
11	pub fn serverThread(srv: Server, stop: std.atomic.Value(bool)) void {	11	pub fn serverThread(srv: Server, stop: std.atomic.Value(bool)) void {
12	while (!stop.load(.acquire)) {	12	while (!stop.load(.acquire)) {
13	const code = srv.pumpOnce(50) catch break;	13	srv.pumpOnce(50) catch break;
14	if (code != null) break;
15	}	14	}
16	}	15	}
17		16
@@ -113,14 +112,11 @@ pub fn quicPump(
113	budget_ms: u64,	112	budget_ms: u64,
114	ctx: anytype,	113	ctx: anytype,
115	done: *const fn (@TypeOf(ctx)) bool,	114	done: *const fn (@TypeOf(ctx)) bool,
116	) !?u8 {	115	) !void {
117	var waited: u64 = 0;	116	var waited: u64 = 0;
118	var exit_code: ?u8 = null;
119	while (waited < budget_ms) {	117	while (waited < budget_ms) {
120	if (done(ctx)) return exit_code;	118	if (done(ctx)) return;
121	if (exit_code == null) {	119	try srv.pumpOnce(5);
122	exit_code = try srv.pumpOnce(5);
123	}
124	for (clients) \|cl\| {	120	for (clients) \|cl\| {
125	cl.drain();	121	cl.drain();
126	var pfd = [_]std.posix.pollfd{.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};	122	var pfd = [_]std.posix.pollfd{.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
@@ -128,7 +124,6 @@ pub fn quicPump(
128	}	124	}
129	waited += 6;	125	waited += 6;
130	}	126	}
131	return exit_code;
132	}	127	}
133		128
134	/// Find a frame of `want` in a client's received bytes, returning its	129	/// Find a frame of `want` in a client's received bytes, returning its
@@ -167,7 +162,7 @@ pub fn awaitFrame(
167	) !?proto.Frame {	162	) !?proto.Frame {
168	var i: usize = 0;	163	var i: usize = 0;
169	while (i < iters) : (i += 1) {	164	while (i < iters) : (i += 1) {
170	_ = try srv.pumpOnce(5);	165	try srv.pumpOnce(5);
171	var pfd = [_]std.posix.pollfd{	166	var pfd = [_]std.posix.pollfd{
172	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	167	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
173	};	168	};
@@ -191,7 +186,7 @@ pub fn awaitGridText(
191	) !bool {	186	) !bool {
192	const deadline = std.time.milliTimestamp() + budget_ms;	187	const deadline = std.time.milliTimestamp() + budget_ms;
193	while (std.time.milliTimestamp() < deadline) {	188	while (std.time.milliTimestamp() < deadline) {
194	_ = try srv.pumpOnce(5);	189	try srv.pumpOnce(5);
195	const grid = try srv.sessions.table[0].?.eng.dumpPlain(alloc);	190	const grid = try srv.sessions.table[0].?.eng.dumpPlain(alloc);
196	defer alloc.free(grid);	191	defer alloc.free(grid);
197	if (std.mem.indexOf(u8, grid, needle) != null) return true;	192	if (std.mem.indexOf(u8, grid, needle) != null) return true;

src/server_test_modes.zig

Old		New
@@ -7,17 +7,14 @@ const Server = srv_mod.Server;
7	const awaitFrame = h.awaitFrame;	7	const awaitFrame = h.awaitFrame;
8	const connectedPair = h.connectedPair;	8	const connectedPair = h.connectedPair;
9		9
10	/// Returns the exit code if the child died — a caller asserting about frames
11	/// wants to hear that rather than spin.
12	fn pumpAndCollectModes(	10	fn pumpAndCollectModes(
13	alloc: std.mem.Allocator,	11	alloc: std.mem.Allocator,
14	srv: *Server,	12	srv: *Server,
15	fd: std.posix.fd_t,	13	fd: std.posix.fd_t,
16	out: *std.ArrayList(proto.PtyModeFlags),	14	out: *std.ArrayList(proto.PtyModeFlags),
17	) !?u8 {	15	) !void {
18	const code = try srv.pumpOnce(20);	16	try srv.pumpOnce(20);
19	try drainModes(alloc, fd, out);	17	try drainModes(alloc, fd, out);
20	return code;
21	}	18	}
22		19
23	/// No pump: for paths that answer synchronously, where pumping would blur what	20	/// No pump: for paths that answer synchronously, where pumping would blur what
@@ -84,7 +81,7 @@ test "Server: an attach arriving on an established connection is answered with t
84	// And again once a pump has been round, where the daemon already knows	81	// And again once a pump has been round, where the daemon already knows
85	// the mode and the fallback is not what is being exercised. A reattach	82	// the mode and the fallback is not what is being exercised. A reattach
86	// over a live connection is a real client's reconnect path.	83	// over a live connection is a real client's reconnect path.
87	_ = try srv.pumpOnce(20);	84	try srv.pumpOnce(20);
88	modes.clearRetainingCapacity();	85	modes.clearRetainingCapacity();
89	srv.pushInbound(0, frame.items);	86	srv.pushInbound(0, frame.items);
90	try drainModes(alloc, c.peer, &modes);	87	try drainModes(alloc, c.peer, &modes);
@@ -141,10 +138,7 @@ test "Server: the pty's mode bits reach a client on attach, and again only when
141	// terminal is doing.	138	// terminal is doing.
142	var rounds: usize = 0;	139	var rounds: usize = 0;
143	while (rounds < 250 and modes.items.len == 0) : (rounds += 1) {	140	while (rounds < 250 and modes.items.len == 0) : (rounds += 1) {
144	try std.testing.expectEqual(	141	try pumpAndCollectModes(alloc, &srv, c.handle, &modes);
145	@as(?u8, null),
146	try pumpAndCollectModes(alloc, &srv, c.handle, &modes),
147	);
148	}	142	}
149	try std.testing.expectEqual(@as(usize, 1), modes.items.len);	143	try std.testing.expectEqual(@as(usize, 1), modes.items.len);
150	try std.testing.expect(modes.items[0].icanon);	144	try std.testing.expect(modes.items[0].icanon);
@@ -157,10 +151,7 @@ test "Server: the pty's mode bits reach a client on attach, and again only when
157		151
158	rounds = 0;	152	rounds = 0;
159	while (rounds < 500 and modes.items.len < 2) : (rounds += 1) {	153	while (rounds < 500 and modes.items.len < 2) : (rounds += 1) {
160	try std.testing.expectEqual(	154	try pumpAndCollectModes(alloc, &srv, c.handle, &modes);
161	@as(?u8, null),
162	try pumpAndCollectModes(alloc, &srv, c.handle, &modes),
163	);
164	}	155	}
165	try std.testing.expectEqual(@as(usize, 2), modes.items.len);	156	try std.testing.expectEqual(@as(usize, 2), modes.items.len);
166	try std.testing.expect(!modes.items[1].echo);	157	try std.testing.expect(!modes.items[1].echo);
@@ -180,10 +171,7 @@ test "Server: the pty's mode bits reach a client on attach, and again only when
180		171
181	rounds = 0;	172	rounds = 0;
182	while (rounds < 500 and modes.items.len < 3) : (rounds += 1) {	173	while (rounds < 500 and modes.items.len < 3) : (rounds += 1) {
183	try std.testing.expectEqual(	174	try pumpAndCollectModes(alloc, &srv, c.handle, &modes);
184	@as(?u8, null),
185	try pumpAndCollectModes(alloc, &srv, c.handle, &modes),
186	);
187	}	175	}
188	try std.testing.expectEqual(@as(usize, 3), modes.items.len);	176	try std.testing.expectEqual(@as(usize, 3), modes.items.len);
189	try std.testing.expect(!modes.items[2].icanon);	177	try std.testing.expect(!modes.items[2].icanon);
@@ -197,10 +185,7 @@ test "Server: the pty's mode bits reach a client on attach, and again only when
197	// every `make test` for evidence already in hand.	185	// every `make test` for evidence already in hand.
198	rounds = 0;	186	rounds = 0;
199	while (rounds < 25) : (rounds += 1) {	187	while (rounds < 25) : (rounds += 1) {
200	try std.testing.expectEqual(	188	try pumpAndCollectModes(alloc, &srv, c.handle, &modes);
201	@as(?u8, null),
202	try pumpAndCollectModes(alloc, &srv, c.handle, &modes),
203	);
204	}	189	}
205	try std.testing.expectEqual(@as(usize, 3), modes.items.len);	190	try std.testing.expectEqual(@as(usize, 3), modes.items.len);
206	}	191	}
@@ -432,7 +417,7 @@ test "Server: a session enabling bracketed paste tells its clients, and not agai
432	var i: usize = 0;	417	var i: usize = 0;
433	while (i < 400 and after < 60) : (i += 1) {	418	while (i < 400 and after < 60) : (i += 1) {
434	if (saw_content) after += 1;	419	if (saw_content) after += 1;
435	_ = try srv.pumpOnce(5);	420	try srv.pumpOnce(5);
436	var pfd = [_]std.posix.pollfd{	421	var pfd = [_]std.posix.pollfd{
437	.{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },	422	.{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },
438	};	423	};
@@ -563,7 +548,7 @@ test "Server: a joiner that resizes the grid is still told the session's modes"
563	var b_modes: ?bool = null;	548	var b_modes: ?bool = null;
564	var i: usize = 0;	549	var i: usize = 0;
565	while (i < 400 and !(a_resnapshotted and b_modes != null)) : (i += 1) {	550	while (i < 400 and !(a_resnapshotted and b_modes != null)) : (i += 1) {
566	_ = try srv.pumpOnce(5);	551	try srv.pumpOnce(5);
567	var pfds = [_]std.posix.pollfd{	552	var pfds = [_]std.posix.pollfd{
568	.{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },	553	.{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
569	.{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },	554	.{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },
@@ -643,7 +628,7 @@ test "Server: a window title reaches clients on change, and only on change" {
643	var i: usize = 0;	628	var i: usize = 0;
644	while (i < 400 and after < 60) : (i += 1) {	629	while (i < 400 and after < 60) : (i += 1) {
645	if (saw_content) after += 1;	630	if (saw_content) after += 1;
646	_ = try srv.pumpOnce(5);	631	try srv.pumpOnce(5);
647	var pfd = [_]std.posix.pollfd{	632	var pfd = [_]std.posix.pollfd{
648	.{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },	633	.{ .fd = c.handle, .events = std.posix.POLL.IN, .revents = 0 },
649	};	634	};
@@ -719,7 +704,7 @@ test "Server: a session that never set a title has none sent for it" {
719	}	704	}
720	after += 1;	705	after += 1;
721	}	706	}
722	_ = try srv.pumpOnce(5);	707	try srv.pumpOnce(5);
723	// Sliced to what exists rather than padded with a placeholder. The	708	// Sliced to what exists rather than padded with a placeholder. The
724	// padding this replaces put A's fd in slot 1 until B joined and	709	// padding this replaces put A's fd in slot 1 until B joined and
725	// relied on a `continue` to skip it — and in a test whose verdict is	710	// relied on a `continue` to skip it — and in a test whose verdict is
@@ -797,7 +782,7 @@ test "Server: a joiner that resizes the grid is still told the session's title"
797	defer if (b_title) \|t\| alloc.free(t);	782	defer if (b_title) \|t\| alloc.free(t);
798	var i: usize = 0;	783	var i: usize = 0;
799	while (i < 400 and !(a_resnapshotted and b_title != null)) : (i += 1) {	784	while (i < 400 and !(a_resnapshotted and b_title != null)) : (i += 1) {
800	_ = try srv.pumpOnce(5);	785	try srv.pumpOnce(5);
801	var pfds = [_]std.posix.pollfd{	786	var pfds = [_]std.posix.pollfd{
802	.{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },	787	.{ .fd = a.handle, .events = std.posix.POLL.IN, .revents = 0 },
803	.{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },	788	.{ .fd = b.handle, .events = std.posix.POLL.IN, .revents = 0 },
@@ -830,7 +815,7 @@ fn modesWithResync(
830	var modes: ?bool = null;	815	var modes: ?bool = null;
831	var i: usize = 0;	816	var i: usize = 0;
832	while (i < 400 and modes == null) : (i += 1) {	817	while (i < 400 and modes == null) : (i += 1) {
833	_ = try srv.pumpOnce(5);	818	try srv.pumpOnce(5);
834	var pfd = [_]std.posix.pollfd{	819	var pfd = [_]std.posix.pollfd{
835	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	820	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
836	};	821	};

src/server_test_quic.zig

src/server_test_session.zig

Old		New
@@ -376,7 +376,7 @@ test "Server: the stats buffer holds the widest reply its format can print" {
376	}	376	}
377	}	377	}
378		378
379	test "Server: stop_req from a bare connection requests shutdown; run returns 130" {	379	test "Server: stop_req from a bare connection requests shutdown; run returns 0" {
380	const alloc = std.testing.allocator;	380	const alloc = std.testing.allocator;
381		381
382	var tmp = try TmpDir.make();	382	var tmp = try TmpDir.make();
@@ -402,7 +402,7 @@ test "Server: stop_req from a bare connection requests shutdown; run returns 130
402	// suite: 100 iterations x 50ms is the deadline, the flag is the exit.	402	// suite: 100 iterations x 50ms is the deadline, the flag is the exit.
403	var i: usize = 0;	403	var i: usize = 0;
404	while (i < 100 and !shutdown_flag.load(.acquire)) : (i += 1) {	404	while (i < 100 and !shutdown_flag.load(.acquire)) : (i += 1) {
405	_ = try srv.pumpOnce(50);	405	try srv.pumpOnce(50);
406	}	406	}
407	try std.testing.expect(shutdown_flag.load(.acquire));	407	try std.testing.expect(shutdown_flag.load(.acquire));
408		408
@@ -411,7 +411,10 @@ test "Server: stop_req from a bare connection requests shutdown; run returns 130
411	// business, not run()'s. It is also the one unbounded call here, and it	411	// business, not run()'s. It is also the one unbounded call here, and it
412	// is safe only because the expect above guarantees the flag: an arm	412	// is safe only because the expect above guarantees the flag: an arm
413	// mutation fails there and never reaches this line.	413	// mutation fails there and never reaches this line.
414	try std.testing.expectEqual(@as(u8, 130), try srv.run());	414	//
		415	// 0: a daemon that served and was asked to stop did its job. A nonzero
		416	// exit here is what a supervisor reads as a crash-loop.
		417	try std.testing.expectEqual(@as(u8, 0), try srv.run());
415	}	418	}
416		419
417	test "Server: stop_req from an attached client is honored too" {	420	test "Server: stop_req from an attached client is honored too" {
@@ -441,7 +444,7 @@ test "Server: stop_req from an attached client is honored too" {
441		444
442	var i: usize = 0;	445	var i: usize = 0;
443	while (i < 100 and !shutdown_flag.load(.acquire)) : (i += 1) {	446	while (i < 100 and !shutdown_flag.load(.acquire)) : (i += 1) {
444	_ = try srv.pumpOnce(50);	447	try srv.pumpOnce(50);
445	}	448	}
446	try std.testing.expect(shutdown_flag.load(.acquire));	449	try std.testing.expect(shutdown_flag.load(.acquire));
447	}	450	}
@@ -609,7 +612,7 @@ fn pumpUntilReplicaSees(
609	) !bool {	612	) !bool {
610	var i: usize = 0;	613	var i: usize = 0;
611	while (i < iters) : (i += 1) {	614	while (i < iters) : (i += 1) {
612	_ = try srv.pumpOnce(5);	615	try srv.pumpOnce(5);
613	var pfd = [_]std.posix.pollfd{	616	var pfd = [_]std.posix.pollfd{
614	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },	617	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
615	};	618	};
@@ -822,7 +825,7 @@ test "Server: an attach past max_sessions is refused with exit_status, sessions
822	// The closes above are only visible to the daemon once it polls, and a	825	// The closes above are only visible to the daemon once it polls, and a
823	// client slot it still believes is live would refuse the probe below	826	// client slot it still believes is live would refuse the probe below
824	// before the session table ever got asked.	827	// before the session table ever got asked.
825	for (0..8) \|_\| _ = try srv.pumpOnce(1);	828	for (0..8) \|_\| try srv.pumpOnce(1);
826		829
827	// One name past the table gets the same honest no a full client table	830	// One name past the table gets the same honest no a full client table
828	// gives.	831	// gives.
@@ -1012,15 +1015,13 @@ test "Server: one session's shell exiting drops only its clients; the daemon car
1012	try pumpUntilReplicaSees(alloc, &srv, cb.handle, rep_b, "echo:pre", 400),	1015	try pumpUntilReplicaSees(alloc, &srv, cb.handle, rep_b, "echo:pre", 400),
1013	);	1016	);
1014		1017
1015	// Kill a's shell, and pump until its client hears so. Every pump must	1018	// Kill a's shell and pump until its client hears so. b is what makes
1016	// return null: one session dying is not the daemon's end while others	1019	// this a comparison: a's death must reach a's client and nobody else's.
1017	// live — this loop IS the regression the task exists to fix, since the
1018	// old exit arm returned a's code right here.
1019	try proto.writeFrame(ca.handle, .input, "die 0\n");	1020	try proto.writeFrame(ca.handle, .input, "die 0\n");
1020	var status: ?u8 = null;	1021	var status: ?u8 = null;
1021	var rounds: usize = 0;	1022	var rounds: usize = 0;
1022	while (rounds < 500 and status == null) : (rounds += 1) {	1023	while (rounds < 500 and status == null) : (rounds += 1) {
1023	try std.testing.expectEqual(@as(?u8, null), try srv.pumpOnce(5));	1024	try srv.pumpOnce(5);
1024	var pfd = [_]std.posix.pollfd{	1025	var pfd = [_]std.posix.pollfd{
1025	.{ .fd = ca.handle, .events = std.posix.POLL.IN, .revents = 0 },	1026	.{ .fd = ca.handle, .events = std.posix.POLL.IN, .revents = 0 },
1026	};	1027	};
@@ -1060,38 +1061,117 @@ test "Server: one session's shell exiting drops only its clients; the daemon car
1060	);	1061	);
1061	}	1062	}
1062		1063
1063	test "Server: the last session's exit code is the daemon's" {	1064	/// `run` on a thread; the code it answers is a fact only `run` holds.
		1065	fn runThread(srv: Server, out: ?u8) void {
		1066	out.* = srv.run() catch \|err\| {
		1067	std.debug.print("run() failed: {t}\n", .{err});
		1068	return;
		1069	};
		1070	}
		1071
		1072	/// Not `awaitFrame`: that one pumps the server, and here a thread does.
		1073	fn awaitFrameThreaded(
		1074	alloc: std.mem.Allocator,
		1075	fd: std.posix.fd_t,
		1076	want: proto.MsgType,
		1077	timeout_ms: u64,
		1078	) !?proto.Frame {
		1079	var left = timeout_ms;
		1080	while (left > 0) {
		1081	var pfd = [_]std.posix.pollfd{
		1082	.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
		1083	};
		1084	const ready = try std.posix.poll(&pfd, 50);
		1085	left -\|= 50;
		1086	if (ready == 0) continue;
		1087	const frame = (try proto.readFrame(alloc, fd)) orelse return null;
		1088	if (frame.type == want) return frame;
		1089	frame.deinit(alloc);
		1090	}
		1091	return null;
		1092	}
		1093
		1094	/// Ask the daemon for its stats until `want` appears in the text. The
		1095	/// counter is the daemon's own account of its table — what `muxd stats`
		1096	/// shows an operator looking at a box with nothing running on it.
		1097	fn waitStats(alloc: std.mem.Allocator, fd: std.posix.fd_t, want: []const u8) !void {
		1098	var tries: usize = 0;
		1099	while (tries < 40) : (tries += 1) {
		1100	try proto.writeFrame(fd, .stats_req, "");
		1101	const frame = (try awaitFrameThreaded(alloc, fd, .stats_reply, 200)) orelse continue;
		1102	defer frame.deinit(alloc);
		1103	if (std.mem.indexOf(u8, frame.payload, want) != null) return;
		1104	}
		1105	std.debug.print("stats never said {s}\n", .{want});
		1106	return error.StatsNeverSaid;
		1107	}
		1108
		1109	/// The socket half of the claim: every wait here runs while `run` pumps on
		1110	/// its own thread, so a daemon that RETURNED from `run` answers none of it
		1111	/// and each one times out.
		1112	fn probeEmptiedDaemon(alloc: std.mem.Allocator, sock_path: []const u8) !void {
		1113	const obs = try std.net.connectUnixSocket(sock_path);
		1114	defer obs.close();
		1115
		1116	// Attach the default session, then hang up its shell — the daemon's
		1117	// only one, so the table empties behind it.
		1118	const c1 = try std.net.connectUnixSocket(sock_path);
		1119	defer c1.close();
		1120	try attachNamed(c1.handle, 80, 24, proto.default_session);
		1121	const snap = (try awaitFrameThreaded(alloc, c1.handle, .snapshot, 4000)) orelse
		1122	return error.NoFirstSnapshot;
		1123	snap.deinit(alloc);
		1124	try proto.writeFrame(c1.handle, .input, "die 0\n");
		1125
		1126	try waitStats(alloc, obs.handle, "sessions=0");
		1127
		1128	// Still serving with nothing to serve: a birth on the emptied daemon
		1129	// takes the default name back, which is the attach `mux --sock PATH`
		1130	// sends.
		1131	const c2 = try std.net.connectUnixSocket(sock_path);
		1132	defer c2.close();
		1133	try attachNamed(c2.handle, 80, 24, proto.default_session);
		1134	const reborn = (try awaitFrameThreaded(alloc, c2.handle, .snapshot, 4000)) orelse
		1135	return error.EmptyDaemonRefusedABirth;
		1136	reborn.deinit(alloc);
		1137	try waitStats(alloc, obs.handle, "sessions=1");
		1138	}
		1139
		1140	test "Server: a daemon outlives its last session and ends only on stop" {
1064	const alloc = std.testing.allocator;	1141	const alloc = std.testing.allocator;
1065		1142
1066	var tmp = try TmpDir.make();	1143	var tmp = try TmpDir.make();
1067	defer tmp.cleanup();	1144	defer tmp.cleanup();
1068	const dir_path = tmp.path();	1145	const dir_path = tmp.path();
1069	const sock_path = try std.fmt.allocPrint(alloc, "{s}/lastcode.sock", .{dir_path});	1146	const sock_path = try std.fmt.allocPrint(alloc, "{s}/livesempty.sock", .{dir_path});
1070	defer alloc.free(sock_path);	1147	defer alloc.free(sock_path);
1071		1148
1072	// A shell that exits 7 on its own: the default session is the only one	1149	try writeMortalScript(&tmp);
1073	// this daemon ever holds, so its death is the last death.	1150	const script = try std.fmt.allocPrintSentinel(alloc, "{s}/mortal.sh", .{dir_path}, 0);
1074	try tmp.dir.writeFile(.{
1075	.sub_path = "exit7.sh",
1076	.data =
1077	\\#!/bin/sh
1078	\\exit 7
1079	\\
1080	,
1081	.flags = .{ .mode = 0o755 },
1082	});
1083	const script = try std.fmt.allocPrintSentinel(alloc, "{s}/exit7.sh", .{dir_path}, 0);
1084	defer alloc.free(script);	1151	defer alloc.free(script);
1085		1152
		1153	// The flag is global by necessity (a signal handler shares it); reset
		1154	// so this test neither inherits a stale request nor leaves one behind.
		1155	shutdown_flag.store(false, .release);
		1156	defer shutdown_flag.store(false, .release);
		1157
1086	var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = script });	1158	var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = script });
1087	defer srv.deinit();	1159	defer srv.deinit();
1088		1160
1089	var code: ?u8 = null;	1161	var code: ?u8 = null;
1090	var rounds: usize = 0;	1162	const th = try std.Thread.spawn(.{}, runThread, .{ &srv, &code });
1091	while (rounds < 500 and code == null) : (rounds += 1) {	1163	// Captured, not propagated: nothing may return between the spawn and
1092	code = try srv.pumpOnce(5);	1164	// the join, because `deinit` is a demolition list and must never run
1093	}	1165	// under a live pump thread.
1094	try std.testing.expectEqual(@as(?u8, 7), code);	1166	const probed = probeEmptiedDaemon(alloc, sock_path);
		1167	shutdown_flag.store(true, .release);
		1168	th.join();
		1169	try probed;
		1170
		1171	// Not the code of any shell: the one this daemon was born with was hung
		1172	// up with `die 0` two stats replies ago, and the one born after it is
		1173	// still alive. This is the shutdown's own code.
		1174	try std.testing.expectEqual(@as(?u8, 0), code);
1095	}	1175	}
1096		1176
1097	test "Server: a dead name re-attaches as a fresh session with a new epoch" {	1177	test "Server: a dead name re-attaches as a fresh session with a new epoch" {
@@ -1122,12 +1202,11 @@ test "Server: a dead name re-attaches as a fresh session with a new epoch" {
1122	const p1 = try proto.readSnapshotPrefix(f1.payload);	1202	const p1 = try proto.readSnapshotPrefix(f1.payload);
1123	try std.testing.expect(p1.epoch != 0);	1203	try std.testing.expect(p1.epoch != 0);
1124		1204
1125	// Kill it, and pump until the name is free. Null throughout: the	1205	// Kill it, and pump until the name is free.
1126	// default session still lives.
1127	try proto.writeFrame(c1.handle, .input, "die 0\n");	1206	try proto.writeFrame(c1.handle, .input, "die 0\n");
1128	var rounds: usize = 0;	1207	var rounds: usize = 0;
1129	while (rounds < 500 and srv.sessions.find("a") != null) : (rounds += 1) {	1208	while (rounds < 500 and srv.sessions.find("a") != null) : (rounds += 1) {
1130	try std.testing.expectEqual(@as(?u8, null), try srv.pumpOnce(5));	1209	try srv.pumpOnce(5);
1131	}	1210	}
1132	try std.testing.expect(srv.sessions.find("a") == null);	1211	try std.testing.expect(srv.sessions.find("a") == null);
1133		1212
@@ -1147,7 +1226,7 @@ test "Server: a dead name re-attaches as a fresh session with a new epoch" {
1147	var second: ?proto.SnapshotPrefix = null;	1226	var second: ?proto.SnapshotPrefix = null;
1148	rounds = 0;	1227	rounds = 0;
1149	while (rounds < 400 and second == null) : (rounds += 1) {	1228	while (rounds < 400 and second == null) : (rounds += 1) {
1150	try std.testing.expectEqual(@as(?u8, null), try srv.pumpOnce(5));	1229	try srv.pumpOnce(5);
1151	var pfd = [_]std.posix.pollfd{	1230	var pfd = [_]std.posix.pollfd{
1152	.{ .fd = c2.handle, .events = std.posix.POLL.IN, .revents = 0 },	1231	.{ .fd = c2.handle, .events = std.posix.POLL.IN, .revents = 0 },
1153	};	1232	};
@@ -1298,7 +1377,7 @@ test "Server: an observer's status_req names a session by tail" {
1298	var saw_eof = false;	1377	var saw_eof = false;
1299	var iters: usize = 0;	1378	var iters: usize = 0;
1300	while (iters < 400 and !saw_eof) : (iters += 1) {	1379	while (iters < 400 and !saw_eof) : (iters += 1) {
1301	_ = try srv.pumpOnce(5);	1380	try srv.pumpOnce(5);
1302	var pfd = [_]std.posix.pollfd{	1381	var pfd = [_]std.posix.pollfd{
1303	.{ .fd = obs.handle, .events = std.posix.POLL.IN, .revents = 0 },	1382	.{ .fd = obs.handle, .events = std.posix.POLL.IN, .revents = 0 },
1304	};	1383	};
@@ -1569,7 +1648,7 @@ test "Server: end_req with another client attached is refused with the count; fo
1569	x2.deinit(alloc);	1648	x2.deinit(alloc);
1570	var waited: u32 = 0;	1649	var waited: u32 = 0;
1571	while (alive(pid_a) and waited < 3000) : (waited += 50) {	1650	while (alive(pid_a) and waited < 3000) : (waited += 50) {
1572	_ = try srv.pumpOnce(20);	1651	try srv.pumpOnce(20);
1573	std.Thread.sleep(30 * std.time.ns_per_ms);	1652	std.Thread.sleep(30 * std.time.ns_per_ms);
1574	}	1653	}
1575	try std.testing.expect(!alive(pid_a));	1654	try std.testing.expect(!alive(pid_a));
@@ -1600,7 +1679,7 @@ test "Server: end_req alone on a session ends it at once, and an unknown name is
1600	try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted);	1679	try std.testing.expect((proto.parseEndReply(r.payload) orelse return error.BadEndReply).accepted);
1601	var waited: u32 = 0;	1680	var waited: u32 = 0;
1602	while (alive(pid) and waited < 3000) : (waited += 50) {	1681	while (alive(pid) and waited < 3000) : (waited += 50) {
1603	_ = try srv.pumpOnce(20);	1682	try srv.pumpOnce(20);
1604	std.Thread.sleep(30 * std.time.ns_per_ms);	1683	std.Thread.sleep(30 * std.time.ns_per_ms);
1605	}	1684	}
1606	try std.testing.expect(!alive(pid));	1685	try std.testing.expect(!alive(pid));
@@ -1657,7 +1736,7 @@ fn awaitGridIn(
1657	) !bool {	1736	) !bool {
1658	const deadline = std.time.milliTimestamp() + budget_ms;	1737	const deadline = std.time.milliTimestamp() + budget_ms;
1659	while (std.time.milliTimestamp() < deadline) {	1738	while (std.time.milliTimestamp() < deadline) {
1660	_ = try srv.pumpOnce(5);	1739	try srv.pumpOnce(5);
1661	const si = srv.sessions.find(proto.wireName(name)) orelse return false;	1740	const si = srv.sessions.find(proto.wireName(name)) orelse return false;
1662	const grid = try srv.ses(si).eng.dumpPlain(alloc);	1741	const grid = try srv.ses(si).eng.dumpPlain(alloc);
1663	defer alloc.free(grid);	1742	defer alloc.free(grid);
@@ -1698,7 +1777,7 @@ test "Server: an accepted end kills a shell that ignores TERM and HUP" {
1698		1777
1699	var waited: u32 = 0;	1778	var waited: u32 = 0;
1700	while (alive(pid) and waited < 3000) : (waited += 50) {	1779	while (alive(pid) and waited < 3000) : (waited += 50) {
1701	_ = try srv.pumpOnce(20);	1780	try srv.pumpOnce(20);
1702	std.Thread.sleep(30 * std.time.ns_per_ms);	1781	std.Thread.sleep(30 * std.time.ns_per_ms);
1703	}	1782	}
1704	try std.testing.expect(!alive(pid));	1783	try std.testing.expect(!alive(pid));
@@ -1768,7 +1847,7 @@ test "Server: a repeated end_req does not push the SIGKILL deadline out" {
1768	var waited: u32 = 0;	1847	var waited: u32 = 0;
1769	while (alive(pid) and waited < 3000) : (waited += 80) {	1848	while (alive(pid) and waited < 3000) : (waited += 80) {
1770	proto.writeFrame(obs.handle, .end_req, proto.encodeEndReq(&rq, true, "nag")) catch {};	1849	proto.writeFrame(obs.handle, .end_req, proto.encodeEndReq(&rq, true, "nag")) catch {};
1771	_ = try srv.pumpOnce(20);	1850	try srv.pumpOnce(20);
1772	std.Thread.sleep(60 * std.time.ns_per_ms);	1851	std.Thread.sleep(60 * std.time.ns_per_ms);
1773	}	1852	}
1774	try std.testing.expect(!alive(pid));	1853	try std.testing.expect(!alive(pid));

src/server_test_upgrade.zig

+    try std.testing.expectEqual(@as(u8, 1), reply.payload[0]);
+    try std.testing.expect(std.mem.indexOf(u8, reply.payload[1..], "session ending") != null);
+test "Server: an EMPTY daemon upgrades — the manifest names no session and the new image serves" {
+    const alloc = std.testing.allocator;
+    var tmp = try TmpDir.make();
+    defer tmp.cleanup();
+    const dir_path = tmp.path();
+    const sock_path = try std.fmt.allocPrint(alloc, "{s}/upempty.sock", .{dir_path});
+    defer alloc.free(sock_path);
+    // A shell that leaves the moment it is forked: the daemon's only
+    // session, so the table is empty a pump later. That state was
+    // unreachable while a daemon left with its last session; it is an
+    // ordinary one now, and the exec has to cross it.
+    try tmp.dir.writeFile(.{
+        .sub_path = "exit0.sh",
+        .data =
+        \\#!/bin/sh
+        \\exit 0
+        \\
+        ,
+        .flags = .{ .mode = 0o755 },
+    });
+    const script = try std.fmt.allocPrintSentinel(alloc, "{s}/exit0.sh", .{dir_path}, 0);
+    defer alloc.free(script);
+    var srv = try Server.init(alloc, .{
+        .sock_path = sock_path,
+        .shell = script,
+        .version = "0.0.1-1",
+    });
+    var rounds: usize = 0;
+    while (rounds < 500 and srv.sessions.live() != 0) : (rounds += 1) try srv.pumpOnce(5);
+    try std.testing.expectEqual(@as(usize, 0), srv.sessions.live());
+    // A candidate that passes every check, run against a daemon holding
+    // nothing: emptiness must not be mistaken for the one state that DOES
+    // refuse an upgrade, a session mid-hangup.
+    try tmp.dir.writeFile(.{
+        .sub_path = "fakemuxd.sh",
+        .data =
+        \\#!/bin/sh
+        \\case "$1" in
+        \\  --version) printf 'muxd 9.9.9\n' ;;
+        \\esac
+        \\exit 0
+        \\
+        ,
+        .flags = .{ .mode = 0o755 },
+    });
+    const cand = try std.fmt.allocPrint(alloc, "{s}/fakemuxd.sh", .{dir_path});
+    defer alloc.free(cand);
+    if (srv.validateUpgrade(.{
+        .allow_same_version = false,
+        .version = "9.9.9",
+        .path = cand,
+    }, "0.0.1-1")) |reason| {
+        defer alloc.free(reason);
+        std.debug.print("an empty daemon was refused an upgrade: {s}\n", .{reason});
+        return error.EmptyDaemonRefusedUpgrade;
+    }
+    const memfd = try std.posix.memfd_create("mux-empty-upgrade", 0);
+    defer std.posix.close(memfd);
+    try srv.writeManifestTo(memfd, "0.0.1-1");
+    // The first Server's MEMORY only, never deinit: deinit unlinks the
+    // socket and deleteTrees the dirs, and the adopting Server is about to
+    // inherit all of them. With no sessions there is nothing else to free.
+    if (srv.agents.dir) |d| alloc.free(d);
+    srv.shellint_arena.deinit();
+    var file = std.fs.File{ .handle = memfd };
+    try file.seekTo(0);
+    const buf = try file.readToEndAlloc(alloc, 4 * 1024 * 1024);
+    defer alloc.free(buf);
+    var parsed = try upgrade.parseManifest(alloc, buf);
+    defer parsed.deinit();
+    try std.testing.expectEqual(@as(usize, 0), parsed.sessions.len);
+    var srv2 = try Server.initFromManifest(alloc, &parsed, "0.0.1-2");
+    defer srv2.deinit();
+    try std.testing.expectEqual(@as(usize, 0), srv2.sessions.live());
+    // Serving, on the listener fd the old image bound: an empty daemon that
+    // came through an exec is still the box a client can be born on.
+    const obs = try std.net.connectUnixSocket(sock_path);
+    defer obs.close();
+    try proto.writeFrame(obs.handle, .stats_req, "");
+    const reply = (try awaitFrame(alloc, &srv2, obs.handle, .stats_reply, 400)) orelse
+        return error.NoStatsReply;
+    defer reply.deinit(alloc);
+    try std.testing.expect(std.mem.indexOf(u8, reply.payload, "sessions=0") != null);
+}

test/agent.sh

Old		New
@@ -848,9 +848,9 @@ scen_session_exit() {
848	return 0	848	return 0
849	}	849	}
850	run_scenario "quic: a session that exits 5 reports 5, not a lost connection" scen_session_exit	850	run_scenario "quic: a session that exits 5 reports 5, not a lost connection" scen_session_exit
851	# The tear daemon's session is gone with it; drop the pid so cleanup does not	851	# The pid is KEPT. The scenario ended this daemon's only session, and a
852	# chase one, and let the socket backstop cover the rest.	852	# daemon lives until `muxd stop` — so it is still running, and the sweep
853	D_TEAR=""	853	# below has to wait for its leak verdict like any other daemon's.
854		854
855	# --- 9: a destination that swallows still honours --timeout ----------------	855	# --- 9: a destination that swallows still honours --timeout ----------------
856	# Not a refusal: a never-listening port answers with ICMP and muxa fails	856	# Not a refusal: a never-listening port answers with ICMP and muxa fails
@@ -937,9 +937,10 @@ fi
937	# Every daemon this run started wrote its verdict into a log of its own (see	937	# Every daemon this run started wrote its verdict into a log of its own (see
938	# start_daemon), so this reads ALL of them rather than the most recent one's.	938	# start_daemon), so this reads ALL of them rather than the most recent one's.
939	# leakcheck already gated the three scenarios that stop their own daemon; what	939	# leakcheck already gated the three scenarios that stop their own daemon; what
940	# lands here is everything else — the TUI daemon its own session ended, the	940	# lands here is everything else — the TUI daemon whose session ended under
941	# QUIC daemon scenario 8 destroyed, and any daemon a FAILING scenario left	941	# it, the QUIC daemon scenario 8 emptied, and any daemon a FAILING scenario
942	# behind, which is exactly the case that used to go unswept.	942	# left behind, which is exactly the case that used to go unswept. Emptying a
		943	# daemon does not remove it from this list: nothing but `muxd stop` ends one.
943	#	944	#
944	# After the count pin on purpose: a leak is not a scenario, and folding it	945	# After the count pin on purpose: a leak is not a scenario, and folding it
945	# into PASSES+FAILS+SKIPS would make that pin's number stop meaning "nine	946	# into PASSES+FAILS+SKIPS would make that pin's number stop meaning "nine

test/e2e_05_session.sh

Old		New
@@ -139,7 +139,7 @@ if grep -q 'm18b-pin' "$OUT.m18mcap"; then
139	fi	139	fi
140		140
141	# Lifetime: a session ends when ITS shell exits, and the daemon outlives	141	# Lifetime: a session ends when ITS shell exits, and the daemon outlives
142	# every death but the last (decision 8). This client attaches to a session	142	# every one of those deaths (decision 8). This client attaches to a session
143	# that ALREADY EXISTS — attach-or-create's join arm, the other half of the	143	# that ALREADY EXISTS — attach-or-create's join arm, the other half of the
144	# two creations above — and types the exit that ends it.	144	# two creations above — and types the exit that ends it.
145	pipe_mux "$OUT.m18ax" "$OUT.m18ax.err" timeout 40 "$MUX" --sock "$SOCK21" --session a	145	pipe_mux "$OUT.m18ax" "$OUT.m18ax.err" timeout 40 "$MUX" --sock "$SOCK21" --session a
@@ -159,9 +159,43 @@ dump_session "$SOCK21" b \| grep -q "m18b-pin" \|\| {
159	echo "e2e FAIL: M18: session b lost its grid when session a died:"	159	echo "e2e FAIL: M18: session b lost its grid when session a died:"
160	dump_session "$SOCK21" b; exit 1; }	160	dump_session "$SOCK21" b; exit 1; }
161		161
		162	# ...and it outlives the LAST death too. `x` ends a session, never a box: b
		163	# and the default are exited in turn, and what is left is a daemon with
		164	# nothing on it — the state the host picker calls `no sessions`, and the
		165	# reason an emptied host is still there to be picked.
		166	pipe_mux "$OUT.m18bx" "$OUT.m18bx.err" timeout 40 "$MUX" --sock "$SOCK21" --session b
		167	pipe_send 'exit\n'
		168	pipe_waitexit "M18: the client of session b"
		169	pipe_mux "$OUT.m18dx" "$OUT.m18dx.err" timeout 40 "$MUX" --sock "$SOCK21"
		170	pipe_send 'exit\n'
		171	pipe_waitexit "M18: the client of the default session"
		172	wait_sessions "$SOCK21" 0 "M18: every one of the three shells has exited"
		173
		174	# The daemon is asked of the OS, not of itself: `stats` answering above
		175	# already needed a live daemon, but only a pid can say the PROCESS is still
		176	# here. The two seconds are not a wait for anything — the table emptied
		177	# before `stats` said so — they are the window the old exit-on-empty arm
		178	# used its grace on, so a daemon that still has one is caught here and not
		179	# by the next scenario finding a dead socket.
		180	sleep 2
		181	kill -0 "$D18PID" 2>/dev/null \|\| {
		182	echo "e2e FAIL: M18: the daemon left with its last session — only muxd stop ends one"
		183	cat "$OUT.m18.d"; exit 1; }
		184
		185	# And it can be born into again: a bare `mux --sock` names no session, so it
		186	# asks for the default, and an EMPTY daemon creates it rather than refusing.
		187	# This is `mux HOST` reaching a box someone emptied yesterday.
		188	pipe_mux "$OUT.m18rb" "$OUT.m18rb.err" timeout 40 "$MUX" --sock "$SOCK21"
		189	pipe_send 'printf "m18rb-%%s\\n" pin\n'
		190	await_out "$OUT.m18rb" "m18rb-pin" "m18rb-pin never reached the reborn default session"
		191	pipe_detach
		192	wait_sessions "$SOCK21" 1 "M18: the emptied daemon was born into again"
		193	wait_grid "$SOCK21" "m18rb-pin" "M18: the reborn default session's marker"
		194
162	assert_stopped "$SOCK21" "$D18PID" "M18 multi-session" "$OUT.m18stop"	195	assert_stopped "$SOCK21" "$D18PID" "M18 multi-session" "$OUT.m18stop"
163	D18PID=""	196	D18PID=""
164	ok "two sessions on one daemon are two shells; one dies without the other"	197	ok "two sessions on one daemon are two shells; one dies without the other"
		198	ok "a daemon outlives its last session and is born into again"
165		199
166	# --- M18 on the wall: one socket, two tiles, two sessions ---------------	200	# --- M18 on the wall: one socket, two tiles, two sessions ---------------
167	# What the milestone was FOR (decision 2): the wall showing the same host	201	# What the milestone was FOR (decision 2): the wall showing the same host

test/e2e_10_agent.sh

Old		New
@@ -165,11 +165,13 @@ ok "agent forwarding: -A with no agent is a usage error, not a silent no-op"
165		165
166	# --- ...and a session nobody offered an agent to refuses, fast -------------	166	# --- ...and a session nobody offered an agent to refuses, fast -------------
167	#	167	#
168	# A fresh daemon on the freed path, and the spawn is load-bearing: the leg	168	# A fresh daemon on the path, and both halves of that are load-bearing. The
169	# above took its daemon down with its last session, so a client dialling	169	# leg above ended its last session but not its DAEMON — a daemon lives until
170	# $SOCK48 now would AUTO-START one off $PATH instead — a daemon that need	170	# `muxd stop` — so the old one is stopped by name here rather than waited
171	# not be the one under test, and against a release with no agent code in it	171	# out. And the new one is started before anything dials, because a client
172	# every assertion below passes having witnessed nothing.	172	# dialling an unserved $SOCK48 would AUTO-START one off $PATH instead — a
		173	# daemon that need not be the one under test, and against a release with no
		174	# agent code in it every assertion below passes having witnessed nothing.
173	#	175	#
174	# The sibling case — a session beside one that HAS an answerer — is the unit	176	# The sibling case — a session beside one that HAS an answerer — is the unit
175	# suite's; what this leg witnesses is the plainer shape, a live daemon	177	# suite's; what this leg witnesses is the plainer shape, a live daemon
@@ -181,7 +183,7 @@ ok "agent forwarding: -A with no agent is a usage error, not a silent no-op"
181	# gap: ssh reads a closed agent socket as "agent refused operation" and	183	# gap: ssh reads a closed agent socket as "agent refused operation" and
182	# falls straight through to its other methods, where a connection accepted	184	# falls straight through to its other methods, where a connection accepted
183	# and left silent would make it wait out a timeout on every dial.	185	# and left silent would make it wait out a timeout on every dial.
184	wait_pid_gone "$D42PID" "agent forwarding: the positive leg's daemon outlived its last session"	186	assert_stopped "$SOCK48" "$D42PID" "agent forwarding: the positive leg's daemon" "$OUT.agtstop0"
185	start_daemon "$SOCK48" "$OUT.agtn.d" "agent-refusal daemon never bound" --shell /bin/sh	187	start_daemon "$SOCK48" "$OUT.agtn.d" "agent-refusal daemon never bound" --shell /bin/sh
186	D42PID=$DPID	188	D42PID=$DPID
187	AR0=$(date +%s%N)	189	AR0=$(date +%s%N)
@@ -461,13 +463,12 @@ set -e
461	# whatever routing did: they replicate one grid, so each capture holds both	463	# whatever routing did: they replicate one grid, so each capture holds both
462	# answers by the time the session ends.	464	# answers by the time the session ends.
463		465
464	# wait_pid_gone and not assert_stopped, because there is nothing left to	466	# A typed `exit` ended this daemon's only session and the daemon is still
465	# stop: A typed `exit`, that was this daemon's only session, and a daemon	467	# here: emptiness is not an exit, and only `muxd stop` ends one. So this is
466	# leaves with its last one (server.zig pumpOnce). The positive leg above	468	# assert_stopped like every other teardown — the kill -0 that says the
467	# ends on assert_stopped only because its second scenario auto-started a	469	# process really went is inside it (wait_pid_gone), and the empty daemon
468	# fresh daemon on the same path to stop. What is asserted here is the same	470	# had to answer the stop to get there.
469	# thing either way — the process this leg forked is gone.	471	assert_stopped "$SOCK49" "$D43PID" "agent-flip" "$OUT.flipstop"
470	wait_pid_gone "$D43PID" "agent-flip: the session ended and the daemon should follow"
471	D43PID=""	472	D43PID=""
472	# Ended here rather than in the trap, for the reason the leg above gives:	473	# Ended here rather than in the trap, for the reason the leg above gives:
473	# a green run should not leave two key-holding processes alive for the	474	# a green run should not leave two key-holding processes alive for the

test/e2e_11_select.sh

Old		New
@@ -432,6 +432,11 @@ MUTE2MS=$(sed -n 's/.mute2=[0-9] ms=$[0-9]$./\1/p' "$OUT.agtmute" \| head -
432	echo "e2e FAIL: agent-mute: the second ssh-add took ${MUTE2MS}ms — the mute"	432	echo "e2e FAIL: agent-mute: the second ssh-add took ${MUTE2MS}ms — the mute"
433	echo " client kept its offer and ssh paid the bound again"	433	echo " client kept its offer and ssh paid the bound again"
434	cat -v "$OUT.agtmute"; exit 1; }	434	cat -v "$OUT.agtmute"; exit 1; }
435	wait_pid_gone "$D42PID" "agent-mute: the session ended and the daemon should follow"	435	# assert_stopped, not wait_pid_gone: the typed `exit` ended this daemon's
		436	# only session and emptiness is not an exit — the daemon is still here and
		437	# has to be told. The kill -0 that says the process really went is inside
		438	# assert_stopped (wait_pid_gone), so nothing is given up by asking for the
		439	# stop first.
		440	assert_stopped "$SOCK48" "$D42PID" "agent-mute" "$OUT.agtmutestop"
436	D42PID=""	441	D42PID=""
437	ok "agent forwarding: a mute offerer is hung up on, then no longer offered (${MUTE2MS}ms)"	442	ok "agent forwarding: a mute offerer is hung up on, then no longer offered (${MUTE2MS}ms)"

Old		New
@@ -1347,7 +1347,7 @@ test "Server: a client that sends half a frame does not stall the pump" {
1347		1347
1348	// One pump with a 20ms poll must come back on its own — the readable	1348	// One pump with a 20ms poll must come back on its own — the readable
1349	// socket is serviced, the partial frame is held, and the loop returns.	1349	// socket is serviced, the partial frame is held, and the loop returns.
1350	_ = try srv.pumpOnce(20);	1350	try srv.pumpOnce(20);
1351	rescue.stop.store(true, .release);	1351	rescue.stop.store(true, .release);
1352	if (rescue.fired.load(.acquire)) {	1352	if (rescue.fired.load(.acquire)) {
1353	std.debug.print("pumpOnce blocked on a half frame; the rescue thread had to complete it\n", .{});	1353	std.debug.print("pumpOnce blocked on a half frame; the rescue thread had to complete it\n", .{});
@@ -1358,7 +1358,7 @@ test "Server: a client that sends half a frame does not stall the pump" {
1358		1358
1359	// The rest arrives; the frame completes on the next pump.	1359	// The rest arrives; the frame completes on the next pump.
1360	try proto.writeAllFd(c.peer, frame.items[3..]);	1360	try proto.writeAllFd(c.peer, frame.items[3..]);
1361	_ = try srv.pumpOnce(20);	1361	try srv.pumpOnce(20);
1362	try std.testing.expectEqual(@as(u16, 100), srv.colsNow(0));	1362	try std.testing.expectEqual(@as(u16, 100), srv.colsNow(0));
1363	try std.testing.expectEqual(@as(u16, 30), srv.rowsNow(0));	1363	try std.testing.expectEqual(@as(u16, 30), srv.rowsNow(0));
1364	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.inbound.items.len);	1364	try std.testing.expectEqual(@as(usize, 0), srv.clients[0].?.inbound.items.len);
@@ -1555,7 +1555,7 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1555	defer ca.close();	1555	defer ca.close();
1556	try attachNamed(ca.handle, 80, 24, "");	1556	try attachNamed(ca.handle, 80, 24, "");
1557	var spun: usize = 0;	1557	var spun: usize = 0;
1558	while (spun < 200 and srv.clients[0] == null) : (spun += 1) _ = try srv.pumpOnce(5);	1558	while (spun < 200 and srv.clients[0] == null) : (spun += 1) try srv.pumpOnce(5);
1559		1559
1560	// Seated one at a time so the slot indices below are the attach order:	1560	// Seated one at a time so the slot indices below are the attach order:
1561	// freeClientSlot hands out the lowest free slot.	1561	// freeClientSlot hands out the lowest free slot.
@@ -1563,7 +1563,7 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1563	defer cb.close();	1563	defer cb.close();
1564	try attachNamed(cb.handle, 80, 24, "");	1564	try attachNamed(cb.handle, 80, 24, "");
1565	spun = 0;	1565	spun = 0;
1566	while (spun < 200 and srv.clients[1] == null) : (spun += 1) _ = try srv.pumpOnce(5);	1566	while (spun < 200 and srv.clients[1] == null) : (spun += 1) try srv.pumpOnce(5);
1567		1567
1568	const a_attached = (srv.clients[0] orelse return error.ClientANeverSeated).activity;	1568	const a_attached = (srv.clients[0] orelse return error.ClientANeverSeated).activity;
1569	const b_attached = (srv.clients[1] orelse return error.ClientBNeverSeated).activity;	1569	const b_attached = (srv.clients[1] orelse return error.ClientBNeverSeated).activity;
@@ -1574,7 +1574,7 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1574	try proto.writeFrame(ca.handle, .input, "x");	1574	try proto.writeFrame(ca.handle, .input, "x");
1575	spun = 0;	1575	spun = 0;
1576	while (spun < 200 and srv.clients[0].?.activity == a_attached) : (spun += 1) {	1576	while (spun < 200 and srv.clients[0].?.activity == a_attached) : (spun += 1) {
1577	_ = try srv.pumpOnce(5);	1577	try srv.pumpOnce(5);
1578	}	1578	}
1579	try std.testing.expect(srv.clients[0].?.activity > srv.clients[1].?.activity);	1579	try std.testing.expect(srv.clients[0].?.activity > srv.clients[1].?.activity);
1580		1580
@@ -1584,7 +1584,7 @@ test "Server: attaching seats a client in activity order, and typing or resizing
1584	try proto.writeFrame(cb.handle, .resize, &proto.encodeSize(100, 30));	1584	try proto.writeFrame(cb.handle, .resize, &proto.encodeSize(100, 30));
1585	spun = 0;	1585	spun = 0;
1586	while (spun < 200 and srv.clients[1].?.activity < a_typed) : (spun += 1) {	1586	while (spun < 200 and srv.clients[1].?.activity < a_typed) : (spun += 1) {
1587	_ = try srv.pumpOnce(5);	1587	try srv.pumpOnce(5);
1588	}	1588	}
1589	try std.testing.expect(srv.clients[1].?.activity > srv.clients[0].?.activity);	1589	try std.testing.expect(srv.clients[1].?.activity > srv.clients[0].?.activity);
1590	}	1590	}
@@ -1646,7 +1646,7 @@ test "Server: an observer that sends one byte does not stall the pump, and finis
1646	// would seat `muxd stats`.	1646	// would seat `muxd stats`.
1647	const obs = try std.net.connectUnixSocket(sock_path);	1647	const obs = try std.net.connectUnixSocket(sock_path);
1648	defer obs.close();	1648	defer obs.close();
1649	_ = try srv.pumpOnce(20); // accept	1649	try srv.pumpOnce(20); // accept
1650	try std.testing.expect(srv.observers[0] != null);	1650	try std.testing.expect(srv.observers[0] != null);
1651		1651
1652	var frame: std.ArrayList(u8) = .empty;	1652	var frame: std.ArrayList(u8) = .empty;
@@ -1657,7 +1657,7 @@ test "Server: an observer that sends one byte does not stall the pump, and finis
1657	var rescue: HalfFrameRescue = .{ .peer = obs.handle, .rest = frame.items[1..] };	1657	var rescue: HalfFrameRescue = .{ .peer = obs.handle, .rest = frame.items[1..] };
1658	const th = try std.Thread.spawn(.{}, HalfFrameRescue.run, .{&rescue});	1658	const th = try std.Thread.spawn(.{}, HalfFrameRescue.run, .{&rescue});
1659	defer th.join();	1659	defer th.join();
1660	_ = try srv.pumpOnce(20);	1660	try srv.pumpOnce(20);
1661	rescue.stop.store(true, .release);	1661	rescue.stop.store(true, .release);
1662	if (rescue.fired.load(.acquire)) {	1662	if (rescue.fired.load(.acquire)) {
1663	std.debug.print("pumpOnce blocked on a one-byte observer; the rescue thread had to complete it\n", .{});	1663	std.debug.print("pumpOnce blocked on a one-byte observer; the rescue thread had to complete it\n", .{});
@@ -1700,13 +1700,13 @@ test "Server: stalled observers are dropped on the idle deadline, and the table
1700	for (&peers) \|*p\| {	1700	for (&peers) \|*p\| {
1701	p.* = try std.net.connectUnixSocket(sock_path);	1701	p.* = try std.net.connectUnixSocket(sock_path);
1702	try proto.writeAllFd(p.handle, &[_]u8{@intFromEnum(proto.MsgType.stats_req)});	1702	try proto.writeAllFd(p.handle, &[_]u8{@intFromEnum(proto.MsgType.stats_req)});
1703	_ = try srv.pumpOnce(20);	1703	try srv.pumpOnce(20);
1704	}	1704	}
1705	defer for (peers) \|p\| p.close();	1705	defer for (peers) \|p\| p.close();
1706	for (srv.observers) \|o\| try std.testing.expect(o != null);	1706	for (srv.observers) \|o\| try std.testing.expect(o != null);
1707		1707
1708	std.Thread.sleep(60 * std.time.ns_per_ms);	1708	std.Thread.sleep(60 * std.time.ns_per_ms);
1709	_ = try srv.pumpOnce(20);	1709	try srv.pumpOnce(20);
1710	for (srv.observers) \|o\| try std.testing.expect(o == null);	1710	for (srv.observers) \|o\| try std.testing.expect(o == null);
1711	// Dropped means closed: the peer reads EOF, not a hang.	1711	// Dropped means closed: the peer reads EOF, not a hang.
1712	var b: [1]u8 = undefined;	1712	var b: [1]u8 = undefined;
@@ -1750,7 +1750,7 @@ test "Server: bytes after an attach in the same write reach the promoted client"
1750	return error.NoSnapshotAfterAttach;	1750	return error.NoSnapshotAfterAttach;
1751	first.deinit(alloc);	1751	first.deinit(alloc);
1752	var i: usize = 0;	1752	var i: usize = 0;
1753	while (i < 100 and srv.colsNow(0) != 132) : (i += 1) _ = try srv.pumpOnce(5);	1753	while (i < 100 and srv.colsNow(0) != 132) : (i += 1) try srv.pumpOnce(5);
1754	try std.testing.expectEqual(@as(u16, 132), srv.colsNow(0));	1754	try std.testing.expectEqual(@as(u16, 132), srv.colsNow(0));
1755	try std.testing.expectEqual(@as(u16, 50), srv.rowsNow(0));	1755	try std.testing.expectEqual(@as(u16, 50), srv.rowsNow(0));
1756	// Nothing left behind on either side of the promotion.	1756	// Nothing left behind on either side of the promotion.

Old		New
@@ -115,7 +115,7 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
115	try attachOver(&cl, &out, alloc);	115	try attachOver(&cl, &out, alloc);
116		116
117	var only = [_]*quic_server.TestClient{&cl};	117	var only = [_]*quic_server.TestClient{&cl};
118	_ = try quicPump(&srv, &only, 8000, &cl, struct {	118	try quicPump(&srv, &only, 8000, &cl, struct {
119	fn f(t: *quic_server.TestClient) bool {	119	fn f(t: *quic_server.TestClient) bool {
120	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	120	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;
121	}	121	}
@@ -136,7 +136,7 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
136	var settle: usize = 0;	136	var settle: usize = 0;
137	while (settle < 400) : (settle += 1) {	137	while (settle < 400) : (settle += 1) {
138	if (q.l.timeoutMs(1000) > 100) break;	138	if (q.l.timeoutMs(1000) > 100) break;
139	_ = try srv.pumpOnce(5);	139	try srv.pumpOnce(5);
140	cl.drain();	140	cl.drain();
141	var pfd = [_]std.posix.pollfd{	141	var pfd = [_]std.posix.pollfd{
142	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },	142	.{ .fd = cl.fd, .events = std.posix.POLL.IN, .revents = 0 },
@@ -163,7 +163,7 @@ test "Server: output reaches a silent QUIC client without waiting for it to spea
163	//	163	//
164	// One pump answers the only question that isolates it: the frame was	164	// One pump answers the only question that isolates it: the frame was
165	// queued before this call, so did THIS call put it on the wire?	165	// queued before this call, so did THIS call put it on the wire?
166	_ = try srv.pumpOnce(5);	166	try srv.pumpOnce(5);
167		167
168	// Reading is not transmitting: the client takes whatever already	168	// Reading is not transmitting: the client takes whatever already
169	// arrived, and never gives the server an inbound packet to react to.	169	// arrived, and never gives the server an inbound packet to react to.
@@ -205,7 +205,7 @@ test "Server: a QUIC client still receives the shell's exit status" {
205	// Attached and carrying state: the frame path works before we start	205	// Attached and carrying state: the frame path works before we start
206	// asking about the harder one.	206	// asking about the harder one.
207	var only = [_]*quic_server.TestClient{&cl};	207	var only = [_]*quic_server.TestClient{&cl};
208	_ = try quicPump(&srv, &only, 8000, &cl, struct {	208	try quicPump(&srv, &only, 8000, &cl, struct {
209	fn f(t: *quic_server.TestClient) bool {	209	fn f(t: *quic_server.TestClient) bool {
210	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	210	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;
211	}	211	}
@@ -219,12 +219,11 @@ test "Server: a QUIC client still receives the shell's exit status" {
219	cl.out = out.items;	219	cl.out = out.items;
220	cl.drain();	220	cl.drain();
221		221
222	const code = try quicPump(&srv, &only, 15000, &cl, struct {	222	try quicPump(&srv, &only, 15000, &cl, struct {
223	fn f(t: *quic_server.TestClient) bool {	223	fn f(t: *quic_server.TestClient) bool {
224	return findFrame(t.recv_buf[0..t.recv_len], .exit_status) != null;	224	return findFrame(t.recv_buf[0..t.recv_len], .exit_status) != null;
225	}	225	}
226	}.f);	226	}.f);
227	_ = code;
228		227
229	const status = findFrame(cl.recv_buf[0..cl.recv_len], .exit_status);	228	const status = findFrame(cl.recv_buf[0..cl.recv_len], .exit_status);
230	try std.testing.expect(status != null);	229	try std.testing.expect(status != null);
@@ -263,7 +262,7 @@ test "Server: one QUIC client leaving does not disturb the other" {
263	try attachOver(&b, &bbuf, alloc);	262	try attachOver(&b, &bbuf, alloc);
264		263
265	var both = [_]*quic_server.TestClient{ &a, &b };	264	var both = [_]*quic_server.TestClient{ &a, &b };
266	_ = try quicPump(&srv, &both, 10000, &b, struct {	265	try quicPump(&srv, &both, 10000, &b, struct {
267	fn f(t: *quic_server.TestClient) bool {	266	fn f(t: *quic_server.TestClient) bool {
268	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;	267	return findFrame(t.recv_buf[0..t.recv_len], .snapshot) != null;
269	}	268	}
@@ -277,7 +276,7 @@ test "Server: one QUIC client leaving does not disturb the other" {
277	try proto.appendFrame(&abuf, alloc, .detach, "");	276	try proto.appendFrame(&abuf, alloc, .detach, "");
278	a.out = abuf.items;	277	a.out = abuf.items;
279	a.drain();	278	a.drain();
280	_ = try quicPump(&srv, &both, 4000, &srv, struct {	279	try quicPump(&srv, &both, 4000, &srv, struct {
281	fn f(s: *Server) bool {	280	fn f(s: *Server) bool {
282	var n: usize = 0;	281	var n: usize = 0;
283	for (s.clients) \|slot\| {	282	for (s.clients) \|slot\| {
@@ -294,7 +293,7 @@ test "Server: one QUIC client leaving does not disturb the other" {
294	b.drain();	293	b.drain();
295	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });	294	var replica = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
296	defer replica.deinit();	295	defer replica.deinit();
297	_ = try quicPump(&srv, &both, 15000, &b, struct {	296	try quicPump(&srv, &both, 15000, &b, struct {
298	fn f(t: *quic_server.TestClient) bool {	297	fn f(t: *quic_server.TestClient) bool {
299	return t.recv_len > 0 and findFrame(t.recv_buf[0..t.recv_len], .delta) != null;	298	return t.recv_len > 0 and findFrame(t.recv_buf[0..t.recv_len], .delta) != null;
300	}	299	}
@@ -325,7 +324,7 @@ test "Server: a QUIC client that stops reading is dropped by the cap, not tolera
325	try attachOver(&cl, &bbuf, alloc);	324	try attachOver(&cl, &bbuf, alloc);
326		325
327	var only = [_]*quic_server.TestClient{&cl};	326	var only = [_]*quic_server.TestClient{&cl};
328	_ = try quicPump(&srv, &only, 8000, &cl, struct {	327	try quicPump(&srv, &only, 8000, &cl, struct {
329	fn f(t: *quic_server.TestClient) bool {	328	fn f(t: *quic_server.TestClient) bool {
330	return t.handshake_done and t.echoed > 0;	329	return t.handshake_done and t.echoed > 0;
331	}	330	}
@@ -349,7 +348,7 @@ test "Server: a QUIC client that stops reading is dropped by the cap, not tolera
349	_ = srv.queueFrame(0, .snapshot, chunk);	348	_ = srv.queueFrame(0, .snapshot, chunk);
350	// The daemon's own machinery, not a hand-rolled loop: expiry,	349	// The daemon's own machinery, not a hand-rolled loop: expiry,
351	// egress and the flush that follows an ack all live in pumpOnce.	350	// egress and the flush that follows an ack all live in pumpOnce.
352	_ = srv.pumpOnce(1) catch null;	351	srv.pumpOnce(1) catch {};
353	}	352	}
354		353
355	try std.testing.expect(srv.clients[0] == null);	354	try std.testing.expect(srv.clients[0] == null);
@@ -381,7 +380,7 @@ test "Server: drainPending waits for a QUIC client's acks, not just its queue" {
381	try attachOver(&cl, &bbuf, alloc);	380	try attachOver(&cl, &bbuf, alloc);
382		381
383	var only = [_]*quic_server.TestClient{&cl};	382	var only = [_]*quic_server.TestClient{&cl};
384	_ = try quicPump(&srv, &only, 8000, &cl, struct {	383	try quicPump(&srv, &only, 8000, &cl, struct {
385	fn f(t: *quic_server.TestClient) bool {	384	fn f(t: *quic_server.TestClient) bool {
386	return t.handshake_done and t.echoed > 0;	385	return t.handshake_done and t.echoed > 0;
387	}	386	}