a73x

a518671e

feat: mux --via arbitrary-command transport; muxd proxy byte pump

a73x   2026-08-08 14:08

Commit message
feat: mux --via arbitrary-command transport; muxd proxy byte pump

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -58,6 +58,15 @@ pub fn build(b: *std.Build) void {
58 }); 58 });
59 mux_mod.addImport("client", client_mod); 59 mux_mod.addImport("client", client_mod);
60 60
61 // No imports, deliberately: the proxy is a byte pump that knows nothing
62 // about the protocol it carries.
63 const proxy_mod = b.createModule(.{
64 .root_source_file = b.path("src/proxy.zig"),
65 .target = target,
66 .optimize = optimize,
67 .link_libc = true,
68 });
69
61 const exe_mod = b.createModule(.{ 70 const exe_mod = b.createModule(.{
62 .root_source_file = b.path("src/main.zig"), 71 .root_source_file = b.path("src/main.zig"),
63 .target = target, 72 .target = target,
@@ -66,6 +75,7 @@ pub fn build(b: *std.Build) void {
66 }); 75 });
67 exe_mod.addImport("server", server_mod); 76 exe_mod.addImport("server", server_mod);
68 exe_mod.addImport("protocol", protocol_mod); 77 exe_mod.addImport("protocol", protocol_mod);
78 exe_mod.addImport("proxy", proxy_mod);
69 79
70 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod }); 80 const exe = b.addExecutable(.{ .name = "muxd", .root_module = exe_mod });
71 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe 81 // Zig 0.15's self-hosted x86_64 linker can't handle the .sframe
@@ -80,7 +90,7 @@ pub fn build(b: *std.Build) void {
80 b.installArtifact(mux_exe); 90 b.installArtifact(mux_exe);
81 91
82 const test_step = b.step("test", "Run unit tests"); 92 const test_step = b.step("test", "Run unit tests");
83 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod }) |mod| { 93 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, server_mod, client_mod, proxy_mod }) |mod| {
84 const t = b.addTest(.{ .root_module = mod }); 94 const t = b.addTest(.{ .root_module = mod });
85 t.use_llvm = true; 95 t.use_llvm = true;
86 t.use_lld = true; 96 t.use_lld = true;
src/client.zig
Old New
@@ -15,16 +15,57 @@ fn onWinch(_: c_int) callconv(.c) void {
15 winch_flag.store(true, .release); 15 winch_flag.store(true, .release);
16 } 16 }
17 17
18 pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 { 18 /// The client's transport: a read fd and a write fd. For a unix socket they
19 const stream = std.net.connectUnixSocket(sock_path) catch { 19 /// are one and the same; under `--via` they are the child command's stdout
20 std.debug.print("mux: cannot connect to {s} (is muxd running?)\n", .{sock_path}); 20 /// and stdin. Nothing below the transport setup knows which it is — that
21 /// blindness is the point of the M6 spike.
22 const Conn = struct { r: std.posix.fd_t, w: std.posix.fd_t };
23
24 /// Attach over a local socket (`sock_path`) or over an arbitrary command's
25 /// stdio (`via`, typically `ssh host muxd proxy ...`). Exactly one is
26 /// non-null; mux_main.zig enforces that.
27 pub fn attach(alloc: std.mem.Allocator, sock_path: ?[]const u8, via: ?[]const u8) !u8 {
28 if (via) |cmd| {
29 var child = std.process.Child.init(&.{ "/bin/sh", "-c", cmd }, alloc);
30 child.stdin_behavior = .Pipe;
31 child.stdout_behavior = .Pipe;
32 // Inherited, not piped: ssh's diagnostics (auth failure, unknown host,
33 // connection refused) are the user's only clue when the transport
34 // never comes up, and we would otherwise swallow them.
35 child.stderr_behavior = .Inherit;
36 child.spawn() catch {
37 std.debug.print("mux: cannot start --via command: {s}\n", .{cmd});
38 return 1;
39 };
40 defer {
41 // Close stdin first so the command sees EOF and can wind down its
42 // remote end cleanly; then TERM it. kill() waitpid()s internally,
43 // so this also reaps — no zombie is left behind.
44 if (child.stdin) |*in| {
45 in.close();
46 child.stdin = null;
47 }
48 _ = child.kill() catch {};
49 }
50 return session(alloc, .{ .r = child.stdout.?.handle, .w = child.stdin.?.handle });
51 }
52
53 const stream = std.net.connectUnixSocket(sock_path.?) catch {
54 std.debug.print("mux: cannot connect to {s} (is muxd running?)\n", .{sock_path.?});
21 return 1; 55 return 1;
22 }; 56 };
23 defer stream.close(); 57 defer stream.close();
24 const sock = stream.handle; 58 return session(alloc, .{ .r = stream.handle, .w = stream.handle });
59 }
25 60
61 fn session(alloc: std.mem.Allocator, conn: Conn) !u8 {
26 // A daemon that dies mid-write must surface as an error return from 62 // A daemon that dies mid-write must surface as an error return from
27 // write(), not deliver a fatal SIGPIPE. 63 // write(), not a fatal SIGPIPE. Zig's start.zig already installs a noop
64 // SIGPIPE handler, so this is defence in depth rather than the thing that
65 // makes EPIPE reachable. The one difference that matters: SIG_IGN survives
66 // exec while a handler does not, so this must stay *after* the `--via`
67 // child is spawned (it is — attach() spawns, then calls us), or ssh and
68 // the remote proxy would inherit an ignored SIGPIPE they never asked for.
28 var ign: std.posix.Sigaction = .{ 69 var ign: std.posix.Sigaction = .{
29 .handler = .{ .handler = std.posix.SIG.IGN }, 70 .handler = .{ .handler = std.posix.SIG.IGN },
30 .mask = std.posix.sigemptyset(), 71 .mask = std.posix.sigemptyset(),
@@ -46,8 +87,10 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
46 var exit_msg: ?[]const u8 = null; 87 var exit_msg: ?[]const u8 = null;
47 defer if (exit_msg) |m| std.debug.print("{s}\n", .{m}); 88 defer if (exit_msg) |m| std.debug.print("{s}\n", .{m});
48 89
49 // Raw mode + alternate screen when we own a terminal. 90 // Raw mode when we own a terminal. The alternate screen is NOT entered
91 // here — see the first-frame gate in the loop below.
50 var orig_termios: ?std.posix.termios = null; 92 var orig_termios: ?std.posix.termios = null;
93 var alt_screen = false;
51 if (is_tty) { 94 if (is_tty) {
52 const orig = try std.posix.tcgetattr(stdin_fd); 95 const orig = try std.posix.tcgetattr(stdin_fd);
53 orig_termios = orig; 96 orig_termios = orig;
@@ -58,9 +101,6 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
58 raw.iflag.IXON = false; 101 raw.iflag.IXON = false;
59 raw.iflag.ICRNL = false; 102 raw.iflag.ICRNL = false;
60 try std.posix.tcsetattr(stdin_fd, .FLUSH, raw); 103 try std.posix.tcsetattr(stdin_fd, .FLUSH, raw);
61 // Autowrap off: an oversized grid row must clip at the right edge
62 // rather than wrap onto the next line and shift the whole paint.
63 try proto.writeAllFd(stdout_fd, "\x1b[?1049h\x1b[?25l\x1b[?7l");
64 104
65 var sa: std.posix.Sigaction = .{ 105 var sa: std.posix.Sigaction = .{
66 .handler = .{ .handler = onWinch }, 106 .handler = .{ .handler = onWinch },
@@ -69,14 +109,21 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
69 }; 109 };
70 std.posix.sigaction(std.posix.SIG.WINCH, &sa, null); 110 std.posix.sigaction(std.posix.SIG.WINCH, &sa, null);
71 } 111 }
72 defer if (orig_termios) |t| { 112 defer {
73 proto.writeAllFd(stdout_fd, "\x1b[?7h\x1b[?25h\x1b[?1049l") catch {}; 113 // Only undo what was actually done: leaving the alternate screen we
74 std.posix.tcsetattr(stdin_fd, .FLUSH, t) catch {}; 114 // never entered would wipe the user's own scrollback.
75 }; 115 if (alt_screen) proto.writeAllFd(stdout_fd, "\x1b[?7h\x1b[?25h\x1b[?1049l") catch {};
116 if (orig_termios) |t| std.posix.tcsetattr(stdin_fd, .FLUSH, t) catch {};
117 }
76 118
77 // A fresh client process holds no state, so it asks for a full snapshot: 119 // A fresh client process holds no state, so it asks for a full snapshot:
78 // no seq, and no epoch to interpret one in. 120 // no seq, and no epoch to interpret one in. A transport that died between
79 try proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0)); 121 // spawn and here (ssh refused, host unreachable) makes this a broken pipe;
122 // that is a message, not a crash.
123 proto.writeFrame(conn.w, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0)) catch {
124 exit_msg = "mux: connection to muxd lost";
125 return 1;
126 };
80 127
81 var stdin_open = true; 128 var stdin_open = true;
82 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live. 129 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live.
@@ -105,23 +152,47 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
105 // follows the daemon, which answers the resize frame 152 // follows the daemon, which answers the resize frame
106 // with a snapshot carrying the new grid size. 153 // with a snapshot carrying the new grid size.
107 size = new_size; 154 size = new_size;
108 try proto.writeFrame(sock, .resize, &proto.encodeSize(size.cols, size.rows)); 155 // Same rule as every other transport write: a dead
156 // transport is reported, never thrown.
157 proto.writeFrame(conn.w, .resize, &proto.encodeSize(size.cols, size.rows)) catch {
158 exit_msg = "mux: connection to muxd lost";
159 return 1;
160 };
109 } 161 }
110 } 162 }
111 } 163 }
112 164
113 var fds = [_]std.posix.pollfd{ 165 var fds = [_]std.posix.pollfd{
114 .{ .fd = sock, .events = std.posix.POLL.IN, .revents = 0 }, 166 .{ .fd = conn.r, .events = std.posix.POLL.IN, .revents = 0 },
115 .{ .fd = if (stdin_open) stdin_fd else -1, .events = std.posix.POLL.IN, .revents = 0 }, 167 .{ .fd = if (stdin_open) stdin_fd else -1, .events = std.posix.POLL.IN, .revents = 0 },
116 }; 168 };
117 _ = try std.posix.poll(&fds, 100); 169 _ = try std.posix.poll(&fds, 100);
118 170
119 if (fds[0].revents != 0) { 171 if (fds[0].revents != 0) {
120 const frame = (try proto.readFrame(alloc, sock)) orelse { 172 // A clean EOF and a frame torn in half are the same event to us:
173 // the transport is gone. Over ssh the tear is the ordinary case —
174 // the channel drops mid-frame — so it has to read as a message
175 // rather than a stack trace. OutOfMemory is not a transport event
176 // and stays loud.
177 const frame = (proto.readFrame(alloc, conn.r) catch |err| switch (err) {
178 error.OutOfMemory => return err,
179 else => null,
180 }) orelse {
121 exit_msg = "mux: connection to muxd lost"; 181 exit_msg = "mux: connection to muxd lost";
122 return 1; 182 return 1;
123 }; 183 };
124 defer frame.deinit(alloc); 184 defer frame.deinit(alloc);
185 // The alternate screen waits for proof that the transport works.
186 // Entering at setup would erase whatever the `--via` command wrote
187 // to its inherited stderr (ssh reports auth and connection failures
188 // hundreds of ms after spawn), and would blank the screen for the
189 // whole of a hang like `mux --via "sleep 30"`. Autowrap goes off
190 // with it: an oversized grid row must clip at the right edge rather
191 // than wrap and shift the whole paint.
192 if (is_tty and !alt_screen) {
193 try proto.writeAllFd(stdout_fd, "\x1b[?1049h\x1b[?25l\x1b[?7l");
194 alt_screen = true;
195 }
125 switch (frame.type) { 196 switch (frame.type) {
126 .snapshot => { 197 .snapshot => {
127 const prefix = proto.readSnapshotPrefix(frame.payload) catch continue; 198 const prefix = proto.readSnapshotPrefix(frame.payload) catch continue;
@@ -149,7 +220,7 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
149 // silently skipping it and desyncing for good. By 220 // silently skipping it and desyncing for good. By
150 // latest-wins this re-attach also re-asserts our size 221 // latest-wins this re-attach also re-asserts our size
151 // onto the shared session — accepted. 222 // onto the shared session — accepted.
152 proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0)) catch {}; 223 proto.writeFrame(conn.w, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0)) catch {};
153 continue; 224 continue;
154 }; 225 };
155 defer alloc.free(composed.bytes); 226 defer alloc.free(composed.bytes);
@@ -188,7 +259,7 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
188 } else { 259 } else {
189 if (std.mem.indexOfScalar(u8, buf[0..n], 0x1c) != null) { 260 if (std.mem.indexOfScalar(u8, buf[0..n], 0x1c) != null) {
190 // Ctrl-\: detach and leave the session running. 261 // Ctrl-\: detach and leave the session running.
191 proto.writeFrame(sock, .detach, "") catch {}; 262 proto.writeFrame(conn.w, .detach, "") catch {};
192 exit_msg = "mux: detached (session still running; run mux to reattach)"; 263 exit_msg = "mux: detached (session still running; run mux to reattach)";
193 return 0; 264 return 0;
194 } 265 }
@@ -198,21 +269,30 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
198 if (history_rows > 0) { 269 if (history_rows > 0) {
199 const max_pages: u32 = (history_rows + size.rows - 1) / size.rows; 270 const max_pages: u32 = (history_rows + size.rows - 1) / size.rows;
200 if (scroll_pages < max_pages) scroll_pages += 1; 271 if (scroll_pages < max_pages) scroll_pages += 1;
201 try requestScrollPage(sock, scroll_pages, history_rows, size); 272 requestScrollPage(conn.w, scroll_pages, history_rows, size) catch {
273 exit_msg = "mux: connection to muxd lost";
274 return 1;
275 };
202 } 276 }
203 } else if (std.mem.eql(u8, buf[0..n], scroll_dn)) { 277 } else if (std.mem.eql(u8, buf[0..n], scroll_dn)) {
204 if (scroll_pages > 0) scroll_pages -= 1; 278 if (scroll_pages > 0) scroll_pages -= 1;
205 if (scroll_pages == 0) { 279 if (scroll_pages == 0) {
206 try renderClipped(alloc, replica, size, stdout_fd); 280 try renderClipped(alloc, replica, size, stdout_fd);
207 } else { 281 } else {
208 try requestScrollPage(sock, scroll_pages, history_rows, size); 282 requestScrollPage(conn.w, scroll_pages, history_rows, size) catch {
283 exit_msg = "mux: connection to muxd lost";
284 return 1;
285 };
209 } 286 }
210 } else if (scroll_pages > 0) { 287 } else if (scroll_pages > 0) {
211 // Any other key exits scroll mode (swallowed, not forwarded). 288 // Any other key exits scroll mode (swallowed, not forwarded).
212 scroll_pages = 0; 289 scroll_pages = 0;
213 try renderClipped(alloc, replica, size, stdout_fd); 290 try renderClipped(alloc, replica, size, stdout_fd);
214 } else { 291 } else {
215 try proto.writeFrame(sock, .input, buf[0..n]); 292 proto.writeFrame(conn.w, .input, buf[0..n]) catch {
293 exit_msg = "mux: connection to muxd lost";
294 return 1;
295 };
216 } 296 }
217 } 297 }
218 } 298 }
@@ -287,7 +367,7 @@ fn paintDeltaClipped(alloc: std.mem.Allocator, payload: []const u8, tty: proto.S
287 } 367 }
288 368
289 fn requestScrollPage( 369 fn requestScrollPage(
290 sock: std.posix.fd_t, 370 w: std.posix.fd_t,
291 pages_up: u32, 371 pages_up: u32,
292 history_rows: u32, 372 history_rows: u32,
293 size: proto.Size, 373 size: proto.Size,
@@ -296,7 +376,7 @@ fn requestScrollPage(
296 // top (screen-space row index history_rows). 376 // top (screen-space row index history_rows).
297 const rows: u32 = size.rows; 377 const rows: u32 = size.rows;
298 const start = history_rows -| (pages_up * rows); 378 const start = history_rows -| (pages_up * rows);
299 try proto.writeFrame(sock, .fetch_scrollback, &proto.encodeScrollbackReq(start, size.rows)); 379 try proto.writeFrame(w, .fetch_scrollback, &proto.encodeScrollbackReq(start, size.rows));
300 } 380 }
301 381
302 /// Paint a fetched history page: clear, rows, and an inverse [scroll] 382 /// Paint a fetched history page: clear, rows, and an inverse [scroll]
src/main.zig
Old New
@@ -1,14 +1,17 @@
1 //! muxd — daemon entrypoint. `run` hosts the session; `dump` prints the 1 //! muxd — daemon entrypoint. `run` hosts the session; `dump` prints the
2 //! authoritative grid over the protocol (debug aid, also used by e2e). 2 //! authoritative grid over the protocol (debug aid, also used by e2e);
3 //! `proxy` exposes the session socket over stdio for `mux --via`.
3 const std = @import("std"); 4 const std = @import("std");
4 const Server = @import("server").Server; 5 const Server = @import("server").Server;
5 const proto = @import("protocol"); 6 const proto = @import("protocol");
7 const proxy = @import("proxy");
6 8
7 const usage = 9 const usage =
8 \\usage: 10 \\usage:
9 \\ muxd run [--sock PATH] [--shell PATH] [--cols N] [--rows N] 11 \\ muxd run [--sock PATH] [--shell PATH] [--cols N] [--rows N]
10 \\ muxd dump [--vt] [--sock PATH] 12 \\ muxd dump [--vt] [--sock PATH]
11 \\ muxd stats [--sock PATH] 13 \\ muxd stats [--sock PATH]
14 \\ muxd proxy [--sock PATH] (byte pump: stdio <-> session socket)
12 \\ 15 \\
13 ; 16 ;
14 17
@@ -78,6 +81,7 @@ pub fn main() !u8 {
78 } 81 }
79 if (std.mem.eql(u8, args[1], "dump")) return dump(alloc, sock_path, vt_mode); 82 if (std.mem.eql(u8, args[1], "dump")) return dump(alloc, sock_path, vt_mode);
80 if (std.mem.eql(u8, args[1], "stats")) return stats(alloc, sock_path); 83 if (std.mem.eql(u8, args[1], "stats")) return stats(alloc, sock_path);
84 if (std.mem.eql(u8, args[1], "proxy")) return proxy.run(sock_path);
81 std.debug.print("{s}", .{usage}); 85 std.debug.print("{s}", .{usage});
82 return 2; 86 return 2;
83 } 87 }
src/mux_main.zig
Old New
@@ -1,8 +1,10 @@
1 //! mux — client binary. `mux [--sock PATH]` attaches to the running muxd. 1 //! mux — client binary. `mux [--sock PATH]` attaches to the local muxd;
2 //! `mux --via CMD` attaches over CMD's stdio instead (any command that
3 //! exposes a session socket as a byte pipe, e.g. `ssh host muxd proxy`).
2 const std = @import("std"); 4 const std = @import("std");
3 const client = @import("client"); 5 const client = @import("client");
4 6
5 const usage = "usage: mux [--sock PATH]\n"; 7 const usage = "usage: mux [--sock PATH | --via CMD]\n";
6 8
7 pub fn main() !u8 { 9 pub fn main() !u8 {
8 var gpa: std.heap.DebugAllocator(.{}) = .init; 10 var gpa: std.heap.DebugAllocator(.{}) = .init;
@@ -13,17 +15,29 @@ pub fn main() !u8 {
13 defer std.process.argsFree(alloc, args); 15 defer std.process.argsFree(alloc, args);
14 16
15 var sock_arg: ?[]const u8 = null; 17 var sock_arg: ?[]const u8 = null;
18 var via_arg: ?[]const u8 = null;
16 var i: usize = 1; 19 var i: usize = 1;
17 while (i < args.len) : (i += 1) { 20 while (i < args.len) : (i += 1) {
18 if (std.mem.eql(u8, args[i], "--sock") and i + 1 < args.len) { 21 if (std.mem.eql(u8, args[i], "--sock") and i + 1 < args.len) {
19 i += 1; 22 i += 1;
20 sock_arg = args[i]; 23 sock_arg = args[i];
24 } else if (std.mem.eql(u8, args[i], "--via") and i + 1 < args.len) {
25 i += 1;
26 via_arg = args[i];
21 } else { 27 } else {
22 std.debug.print("{s}", .{usage}); 28 std.debug.print("{s}", .{usage});
23 return 2; 29 return 2;
24 } 30 }
25 } 31 }
26 32
33 // Two transports, one session: naming both is a request we cannot honour
34 // rather than one to reconcile.
35 if (sock_arg != null and via_arg != null) {
36 std.debug.print("mux: --sock and --via are mutually exclusive\n{s}", .{usage});
37 return 2;
38 }
39 if (via_arg) |cmd| return client.attach(alloc, null, cmd);
40
27 const sock_path = if (sock_arg) |s| 41 const sock_path = if (sock_arg) |s|
28 try alloc.dupe(u8, s) 42 try alloc.dupe(u8, s)
29 else if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir| 43 else if (std.posix.getenv("XDG_RUNTIME_DIR")) |dir|
@@ -32,5 +46,5 @@ pub fn main() !u8 {
32 try std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()}); 46 try std.fmt.allocPrint(alloc, "/tmp/muxd-{d}.sock", .{std.os.linux.getuid()});
33 defer alloc.free(sock_path); 47 defer alloc.free(sock_path);
34 48
35 return client.attach(alloc, sock_path); 49 return client.attach(alloc, sock_path, null);
36 } 50 }
src/proxy.zig
Old New
@@ -0,0 +1,363 @@
1 //! `muxd proxy`: a bidirectional byte pump between stdio and the local
2 //! daemon socket. Deliberately frame-agnostic — it contains no protocol
3 //! knowledge at all. That is the M6 transport thesis: if an opaque byte
4 //! pipe suffices to carry the protocol over SSH, transport is a swap,
5 //! not a redesign. Keep this file's import list empty of `protocol`.
6 const std = @import("std");
7
8 /// `muxd proxy` proper: pump between this process's stdio and `sock_path`.
9 pub fn run(sock_path: []const u8) !u8 {
10 return pump(std.posix.STDIN_FILENO, std.posix.STDOUT_FILENO, sock_path);
11 }
12
13 /// Copy bytes in both directions until either side hangs up. `in_fd`/`out_fd`
14 /// are the transport (stdio under `run`, pipes under test); the socket is the
15 /// local daemon. Exits 0 when either side hangs up cleanly, 1 if the daemon
16 /// socket cannot be reached or a read or write fails.
17 pub fn pump(in_fd: std.posix.fd_t, out_fd: std.posix.fd_t, sock_path: []const u8) !u8 {
18 const stream = std.net.connectUnixSocket(sock_path) catch {
19 std.debug.print("muxd proxy: cannot connect to {s}\n", .{sock_path});
20 return 1;
21 };
22 defer stream.close();
23 const sock = stream.handle;
24
25 // Defence in depth, not a fix: Zig's start.zig already installs a noop
26 // SIGPIPE handler, so a hangup surfaces as EPIPE from write() rather than
27 // killing us, and the error returns below are reachable without this. What
28 // this pins is that the reachability belongs to this file instead of to a
29 // std default (`std.options.keep_sigpipe`) another module could flip. The
30 // one real difference between the two: SIG_IGN survives exec, a handler
31 // does not — which is why client.zig installs its ignore only after
32 // spawning the transport child, and why the order matters there and not
33 // here (the proxy spawns nothing).
34 var ign: std.posix.Sigaction = .{
35 .handler = .{ .handler = std.posix.SIG.IGN },
36 .mask = std.posix.sigemptyset(),
37 .flags = 0,
38 };
39 std.posix.sigaction(std.posix.SIG.PIPE, &ign, null);
40
41 var buf: [64 * 1024]u8 = undefined;
42 while (true) {
43 var fds = [_]std.posix.pollfd{
44 .{ .fd = in_fd, .events = std.posix.POLL.IN, .revents = 0 },
45 .{ .fd = sock, .events = std.posix.POLL.IN, .revents = 0 },
46 };
47 _ = try std.posix.poll(&fds, -1);
48
49 if (fds[0].revents != 0) {
50 const n = std.posix.read(in_fd, &buf) catch return 1;
51 if (n == 0) return 0; // client hung up: nothing left to carry
52 writeAll(sock, buf[0..n]) catch return 1;
53 }
54 if (fds[1].revents != 0) {
55 const n = std.posix.read(sock, &buf) catch return 1;
56 // The daemon hung up: exit, never reconnect. A delta stream cannot
57 // outlive the daemon instance that opened it (see the epoch note at
58 // src/protocol.zig), so a reconnecting proxy would silently rebind a
59 // client to a different session — or to a restarted one — instead of
60 // letting it learn its session died. Honest failure beats a pipe
61 // that heals into a lie.
62 if (n == 0) return 0;
63 writeAll(out_fd, buf[0..n]) catch return 1;
64 }
65 }
66 }
67
68 fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
69 var i: usize = 0;
70 while (i < data.len) i += try std.posix.write(fd, data[i..]);
71 }
72
73 /// Runs `pump` off the test's main thread so a pump that fails to exit fails
74 /// an assertion instead of wedging the suite. That matters specifically here:
75 /// break the EOF-exit invariant and the loop spins on a readable-but-empty
76 /// socket, which nothing outside the pump can interrupt.
77 const PumpRunner = struct {
78 in_fd: std.posix.fd_t,
79 out_fd: std.posix.fd_t,
80 sock_path: []const u8,
81 rc: u8 = 0xff,
82 failed: bool = false,
83 done: std.atomic.Value(bool) = .init(false),
84
85 fn run(self: *PumpRunner) void {
86 self.rc = pump(self.in_fd, self.out_fd, self.sock_path) catch blk: {
87 self.failed = true;
88 break :blk 0xff;
89 };
90 self.done.store(true, .release);
91 }
92
93 /// Wait for the pump to return. On timeout the caller must NOT join: a
94 /// spinning pump never returns, so the thread is abandoned to process exit
95 /// and the test fails rather than hanging.
96 fn wait(self: *PumpRunner, ms: u32) !void {
97 var waited: u32 = 0;
98 while (waited < ms) : (waited += 20) {
99 if (self.done.load(.acquire)) return;
100 std.Thread.sleep(20 * std.time.ns_per_ms);
101 }
102 return error.PumpDidNotExit;
103 }
104 };
105
106 /// A one-shot peer on a unix socket: accepts, reads `want` bytes, then sends
107 /// `reply` and closes. Closing is what ends the pump under test.
108 const EchoPeer = struct {
109 listener: *std.net.Server,
110 alloc: std.mem.Allocator,
111 want: usize,
112 reply: []const u8,
113 got: std.ArrayList(u8) = .empty,
114
115 fn run(self: *EchoPeer) void {
116 const conn = self.listener.accept() catch return;
117 defer conn.stream.close();
118 var buf: [4096]u8 = undefined;
119 // Idle deadline rather than a blocking read: a pump that delivers only
120 // some of the bytes must fail the byte-for-byte assertion below, not
121 // wedge the test suite waiting for the rest.
122 var idle: i32 = 0;
123 while (self.got.items.len < self.want and idle < 3000) {
124 var fds = [_]std.posix.pollfd{
125 .{ .fd = conn.stream.handle, .events = std.posix.POLL.IN, .revents = 0 },
126 };
127 const ready = std.posix.poll(&fds, 100) catch return;
128 if (ready == 0) {
129 idle += 100;
130 continue;
131 }
132 const n = std.posix.read(conn.stream.handle, &buf) catch return;
133 if (n == 0) break;
134 self.got.appendSlice(self.alloc, buf[0..n]) catch return;
135 idle = 0;
136 }
137 writeAll(conn.stream.handle, self.reply) catch return;
138 }
139 };
140
141 /// Feeds a pipe from its own thread: more than a pipe buffer's worth of input
142 /// cannot be queued up front.
143 const PipeWriter = struct {
144 fd: std.posix.fd_t,
145 data: []const u8,
146 fn run(self: *PipeWriter) void {
147 writeAll(self.fd, self.data) catch return;
148 }
149 };
150
151 /// Drains a pipe concurrently with the pump, stopping once it has been idle
152 /// for a while. Idle-based rather than EOF-based so the test never has to
153 /// close the pump's output fd at a particular moment.
154 const PipeDrainer = struct {
155 fd: std.posix.fd_t,
156 alloc: std.mem.Allocator,
157 out: std.ArrayList(u8) = .empty,
158 fn run(self: *PipeDrainer, idle_limit_ms: i32) void {
159 var idle: i32 = 0;
160 var buf: [8192]u8 = undefined;
161 while (idle < idle_limit_ms) {
162 var fds = [_]std.posix.pollfd{
163 .{ .fd = self.fd, .events = std.posix.POLL.IN, .revents = 0 },
164 };
165 const ready = std.posix.poll(&fds, 100) catch return;
166 if (ready == 0) {
167 idle += 100;
168 continue;
169 }
170 const n = std.posix.read(self.fd, &buf) catch return;
171 if (n == 0) return;
172 self.out.appendSlice(self.alloc, buf[0..n]) catch return;
173 idle = 0;
174 }
175 }
176 };
177
178 /// Shrink a socket's buffers so a transfer larger than them cannot be swallowed
179 /// whole by the kernel. Set on the listener, it is inherited by the accepted
180 /// connection. Linux doubles and clamps the request, so this is a floor
181 /// request, not a promise — the tests depend only on it being small.
182 fn shrinkBufs(fd: std.posix.fd_t) void {
183 const v: c_int = 1024;
184 std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.RCVBUF, std.mem.asBytes(&v)) catch {};
185 std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.SNDBUF, std.mem.asBytes(&v)) catch {};
186 }
187
188 fn readExactly(fd: std.posix.fd_t, buf: []u8) !void {
189 var i: usize = 0;
190 while (i < buf.len) {
191 const n = try std.posix.read(fd, buf[i..]);
192 if (n == 0) return error.UnexpectedEof;
193 i += n;
194 }
195 }
196
197 test "pump carries bytes both ways verbatim" {
198 const alloc = std.testing.allocator;
199
200 var tmp = std.testing.tmpDir(.{});
201 defer tmp.cleanup();
202 var path_buf: [256]u8 = undefined;
203 const dir_path = try tmp.dir.realpath(".", &path_buf);
204 var sock_buf: [280]u8 = undefined;
205 const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/proxy.sock", .{dir_path});
206
207 const addr = try std.net.Address.initUnix(sock_path);
208 var listener = try addr.listen(.{});
209 defer listener.deinit();
210
211 // Frame-shaped payloads with embedded zeros and 0xFF: the pump must be
212 // byte-transparent, not text-safe.
213 const to_daemon = [_]u8{ 0x01, 0x08, 0x00, 0x00, 0x00, 0xff, 0x00, 0x1b, 0x5c, 0x00, 0x7f, 0xfe, 0x0a };
214 const to_client = [_]u8{ 0x80, 0x06, 0x00, 0x00, 0x00, 0x00, 0xff, 0x00, 0xff, 0x1c, 0x00 };
215
216 const in_pipe = try std.posix.pipe();
217 defer std.posix.close(in_pipe[0]);
218 defer std.posix.close(in_pipe[1]);
219 const out_pipe = try std.posix.pipe();
220 defer std.posix.close(out_pipe[0]);
221 defer std.posix.close(out_pipe[1]);
222
223 // Queued before the pump starts; the pipe buffer holds it, so nothing here
224 // depends on thread scheduling.
225 try writeAll(in_pipe[1], &to_daemon);
226
227 var peer = EchoPeer{
228 .listener = &listener,
229 .alloc = alloc,
230 .want = to_daemon.len,
231 .reply = &to_client,
232 };
233 defer peer.got.deinit(alloc);
234 const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
235
236 var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
237 const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
238 // Returns when the peer closes — the daemon-hangup path.
239 try runner.wait(5000);
240 pump_th.join();
241 peer_th.join();
242
243 try std.testing.expect(!runner.failed);
244 try std.testing.expectEqual(@as(u8, 0), runner.rc);
245 try std.testing.expectEqualSlices(u8, &to_daemon, peer.got.items);
246
247 // Read a known length rather than to EOF: the write end is still open.
248 var out: [to_client.len]u8 = undefined;
249 try readExactly(out_pipe[0], &out);
250 try std.testing.expectEqualSlices(u8, &to_client, &out);
251 }
252
253 test "pump carries a large transfer verbatim under backpressure" {
254 const alloc = std.testing.allocator;
255
256 var tmp = std.testing.tmpDir(.{});
257 defer tmp.cleanup();
258 var path_buf: [256]u8 = undefined;
259 const dir_path = try tmp.dir.realpath(".", &path_buf);
260 var sock_buf: [280]u8 = undefined;
261 const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/big.sock", .{dir_path});
262
263 const addr = try std.net.Address.initUnix(sock_path);
264 var listener = try addr.listen(.{});
265 defer listener.deinit();
266 // Backpressure: far more data than either buffer can hold, so the copy
267 // loops in both directions instead of completing in one syscall.
268 shrinkBufs(listener.stream.handle);
269
270 const size = 300 * 1024;
271 const up = try alloc.alloc(u8, size);
272 defer alloc.free(up);
273 const down = try alloc.alloc(u8, size);
274 defer alloc.free(down);
275 for (up, 0..) |*b, i| b.* = @truncate(i *% 31 +% (i >> 8));
276 for (down, 0..) |*b, i| b.* = @truncate(i *% 37 +% (i >> 7) +% 11);
277
278 const in_pipe = try std.posix.pipe();
279 defer std.posix.close(in_pipe[0]);
280 defer std.posix.close(in_pipe[1]);
281 const out_pipe = try std.posix.pipe();
282 defer std.posix.close(out_pipe[0]);
283 defer std.posix.close(out_pipe[1]);
284
285 var peer = EchoPeer{ .listener = &listener, .alloc = alloc, .want = size, .reply = down };
286 defer peer.got.deinit(alloc);
287 const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
288
289 // Both ends need their own thread: 300KB exceeds a pipe buffer, so a
290 // single-threaded test would deadlock against its own pump.
291 var writer = PipeWriter{ .fd = in_pipe[1], .data = up };
292 const writer_th = try std.Thread.spawn(.{}, PipeWriter.run, .{&writer});
293 var drainer = PipeDrainer{ .fd = out_pipe[0], .alloc = alloc };
294 defer drainer.out.deinit(alloc);
295 const drain_th = try std.Thread.spawn(.{}, PipeDrainer.run, .{ &drainer, 1500 });
296
297 var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
298 const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
299 try runner.wait(30_000);
300 pump_th.join();
301 peer_th.join();
302 writer_th.join();
303 drain_th.join();
304
305 try std.testing.expect(!runner.failed);
306 try std.testing.expectEqual(@as(u8, 0), runner.rc);
307 try std.testing.expectEqualSlices(u8, up, peer.got.items);
308 try std.testing.expectEqualSlices(u8, down, drainer.out.items);
309 }
310
311 test "pump exits when the far side of its output pipe is gone" {
312 const alloc = std.testing.allocator;
313
314 var tmp = std.testing.tmpDir(.{});
315 defer tmp.cleanup();
316 var path_buf: [256]u8 = undefined;
317 const dir_path = try tmp.dir.realpath(".", &path_buf);
318 var sock_buf: [280]u8 = undefined;
319 const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/gone.sock", .{dir_path});
320
321 const addr = try std.net.Address.initUnix(sock_path);
322 var listener = try addr.listen(.{});
323 defer listener.deinit();
324
325 const in_pipe = try std.posix.pipe();
326 defer std.posix.close(in_pipe[0]);
327 defer std.posix.close(in_pipe[1]);
328 const out_pipe = try std.posix.pipe();
329 defer std.posix.close(out_pipe[1]);
330 // The reader is gone before a byte is written: writing to out_fd can only
331 // fail. The pump must report that as a failed transfer (1), not die of
332 // SIGPIPE and not spin.
333 std.posix.close(out_pipe[0]);
334
335 var peer = EchoPeer{ .listener = &listener, .alloc = alloc, .want = 0, .reply = "x" };
336 defer peer.got.deinit(alloc);
337 const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
338
339 var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
340 const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
341 try runner.wait(5000);
342 pump_th.join();
343 peer_th.join();
344
345 try std.testing.expect(!runner.failed);
346 try std.testing.expectEqual(@as(u8, 1), runner.rc);
347 }
348
349 test "pump reports a missing daemon socket instead of hanging" {
350 var tmp = std.testing.tmpDir(.{});
351 defer tmp.cleanup();
352 var path_buf: [256]u8 = undefined;
353 const dir_path = try tmp.dir.realpath(".", &path_buf);
354 var sock_buf: [280]u8 = undefined;
355 const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/absent.sock", .{dir_path});
356
357 // No fds are touched on this path, so -1 would do; use the real stdio
358 // constants to keep the call shape honest.
359 try std.testing.expectEqual(
360 @as(u8, 1),
361 try pump(std.posix.STDIN_FILENO, std.posix.STDOUT_FILENO, sock_path),
362 );
363 }
test/e2e.sh
Old New
@@ -7,7 +7,7 @@ MUX="$2"
7 SOCK="${TMPDIR:-/tmp}/muxd-e2e-$$.sock" 7 SOCK="${TMPDIR:-/tmp}/muxd-e2e-$$.sock"
8 OUT="${TMPDIR:-/tmp}/mux-e2e-out-$$" 8 OUT="${TMPDIR:-/tmp}/mux-e2e-out-$$"
9 9
10 cleanup() { kill "$DPID" 2>/dev/null || true; rm -f "$SOCK" "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" "$OUT.b"; } 10 cleanup() { kill "$DPID" 2>/dev/null || true; rm -f "$SOCK" "$OUT" "$OUT.kill" "$OUT.re" "$OUT.a" "$OUT.b" "$OUT.via"; }
11 trap cleanup EXIT INT TERM 11 trap cleanup EXIT INT TERM
12 12
13 "$MUXD" run --sock "$SOCK" --shell /bin/sh & 13 "$MUXD" run --sock "$SOCK" --shell /bin/sh &
@@ -69,4 +69,23 @@ grep -q "m5-after-a-left" "$OUT.b" || {
69 kill -0 "$DPID" || { echo "e2e FAIL: daemon died in two-client scenario"; exit 1; } 69 kill -0 "$DPID" || { echo "e2e FAIL: daemon died in two-client scenario"; exit 1; }
70 rm -f "$OUT.a" "$OUT.b" 70 rm -f "$OUT.a" "$OUT.b"
71 71
72 # --- M6: the same protocol over an arbitrary byte pipe. `muxd proxy` is a
73 # frame-agnostic stdio<->socket pump; if the session works through it, the
74 # transport really is a swap. The string is one shell word here and is split
75 # by the /bin/sh -c that mux spawns, so $MUXD must contain no spaces — it is
76 # the build tree's artifact path, which does not.
77 #
78 # XDG_RUNTIME_DIR is pointed at nothing so the test cannot pass by environment
79 # luck: if --via ever silently fell back to the default socket path, that path
80 # would resolve to a directory that does not exist and the session would fail
81 # instead of quietly attaching over the local socket. `|| true` keeps set -e
82 # from aborting before the labelled diagnostic below runs.
83 { sleep 0.5; printf 'printf "m6-%%s\\n" via-pipe\n'; sleep 2; printf '\034'; } | \
84 XDG_RUNTIME_DIR=/nonexistent-mux-e2e "$MUX" --via "$MUXD proxy --sock $SOCK" > "$OUT.via" || true
85 grep -q "m6-via-pipe" "$OUT.via" || {
86 echo "e2e FAIL: --via transport"; cat "$OUT.via"; exit 1;
87 }
88 kill -0 "$DPID" || { echo "e2e FAIL: daemon died in --via scenario"; exit 1; }
89 rm -f "$OUT.via"
90
72 echo "e2e OK" 91 echo "e2e OK"