src/proxy.zig
Ref: Size: 14.4 KiB History
//! `mux d proxy`: a bidirectional byte pump between stdio and the local daemon
//! socket. Deliberately frame-agnostic — the transport thesis is that if an
//! opaque byte pipe carries the protocol, transport is a swap and not a
//! redesign. Keep this file's import list empty of `protocol`.
//!
//! `sockpath` is the one module of ours here, and it carries no wire contract:
//! it owns the question "what is at this path", which the connect below has to
//! ask because the two kernels answer a non-socket differently and one of them
//! answers with a panic.
const std = @import("std");
const sockpath = @import("sockpath");
const TmpDir = @import("testtmp").TmpDir;
/// Make a hangup surface as EPIPE from write() instead of killing the
/// process. SIG_IGN survives exec where a handler does not: a caller that
/// SPAWNS must install this after the spawn or the child inherits it.
pub fn ignoreSigpipe() void {
var ign: std.posix.Sigaction = .{
.handler = .{ .handler = std.posix.SIG.IGN },
.mask = std.posix.sigemptyset(),
.flags = 0,
};
std.posix.sigaction(std.posix.SIG.PIPE, &ign, null);
}
/// `mux d proxy` proper: pump between this process's stdio and `sock_path`.
pub fn run(sock_path: []const u8) !u8 {
return pump(std.posix.STDIN_FILENO, std.posix.STDOUT_FILENO, sock_path);
}
/// Copy bytes in both directions until either side hangs up. `in_fd`/`out_fd`
/// are the transport (stdio under `run`, pipes under test); the socket is the
/// local daemon. Exits 0 when either side hangs up cleanly, 1 if the daemon
/// socket cannot be reached or a read or write fails.
pub fn pump(in_fd: std.posix.fd_t, out_fd: std.posix.fd_t, sock_path: []const u8) !u8 {
const stream = sockpath.connectSocket(sock_path) catch {
std.debug.print("mux d proxy: cannot connect to {s}\n", .{sock_path});
return 1;
};
defer stream.close();
const sock = stream.handle;
ignoreSigpipe();
var buf: [64 * 1024]u8 = undefined;
while (true) {
var fds = [_]std.posix.pollfd{
.{ .fd = in_fd, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = sock, .events = std.posix.POLL.IN, .revents = 0 },
};
_ = try std.posix.poll(&fds, -1);
if (fds[0].revents != 0) {
const n = std.posix.read(in_fd, &buf) catch return 1;
if (n == 0) return 0; // client hung up: nothing left to carry
writeAll(sock, buf[0..n]) catch return 1;
}
if (fds[1].revents != 0) {
const n = std.posix.read(sock, &buf) catch return 1;
// The daemon hung up: exit, never reconnect. A delta stream cannot
// outlive the daemon instance that opened it, so a reconnecting proxy
// would rebind a client to a different session instead of letting it
// learn its own died.
if (n == 0) return 0;
writeAll(out_fd, buf[0..n]) catch return 1;
}
}
}
/// Byte-identical to `protocol.writeAllFd`, and deliberately not shared: this
/// file's contract is that it imports nothing of the wire, and `protocol` is
/// the wire. The duplicate is the price of that ban, not an oversight.
fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
var i: usize = 0;
while (i < data.len) i += try std.posix.write(fd, data[i..]);
}
/// Runs `pump` off the test's main thread so a pump that fails to exit fails
/// an assertion instead of wedging the suite. That matters specifically here:
/// break the EOF-exit invariant and the loop spins on a readable-but-empty
/// socket, which nothing outside the pump can interrupt.
const PumpRunner = struct {
in_fd: std.posix.fd_t,
out_fd: std.posix.fd_t,
sock_path: []const u8,
rc: u8 = 0xff,
failed: bool = false,
done: std.atomic.Value(bool) = .init(false),
fn run(self: *PumpRunner) void {
self.rc = pump(self.in_fd, self.out_fd, self.sock_path) catch blk: {
self.failed = true;
break :blk 0xff;
};
self.done.store(true, .release);
}
/// Wait for the pump to return. On timeout the caller must NOT join: a
/// spinning pump never returns, so the thread is abandoned to process exit
/// and the test fails rather than hanging.
fn wait(self: *PumpRunner, ms: u32) !void {
var waited: u32 = 0;
while (waited < ms) : (waited += 20) {
if (self.done.load(.acquire)) return;
std.Thread.sleep(20 * std.time.ns_per_ms);
}
return error.PumpDidNotExit;
}
};
/// A one-shot peer on a unix socket: accepts, reads `want` bytes, then sends
/// `reply` and closes. Closing is what ends the pump under test.
const EchoPeer = struct {
listener: *std.net.Server,
alloc: std.mem.Allocator,
want: usize,
reply: []const u8,
got: std.ArrayList(u8) = .empty,
fn run(self: *EchoPeer) void {
const conn = self.listener.accept() catch return;
defer conn.stream.close();
var buf: [4096]u8 = undefined;
// Idle deadline rather than a blocking read: a pump that delivers only
// some of the bytes must fail the byte-for-byte assertion below, not
// wedge the test suite waiting for the rest.
var idle: i32 = 0;
while (self.got.items.len < self.want and idle < 3000) {
var fds = [_]std.posix.pollfd{
.{ .fd = conn.stream.handle, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&fds, 100) catch return;
if (ready == 0) {
idle += 100;
continue;
}
const n = std.posix.read(conn.stream.handle, &buf) catch return;
if (n == 0) break;
self.got.appendSlice(self.alloc, buf[0..n]) catch return;
idle = 0;
}
writeAll(conn.stream.handle, self.reply) catch return;
}
};
/// Feeds a pipe from its own thread: more than a pipe buffer's worth of input
/// cannot be queued up front.
const PipeWriter = struct {
fd: std.posix.fd_t,
data: []const u8,
fn run(self: *PipeWriter) void {
writeAll(self.fd, self.data) catch return;
}
};
/// Drains a pipe concurrently with the pump, stopping once it has been idle
/// for a while. Idle-based rather than EOF-based so the test never has to
/// close the pump's output fd at a particular moment.
const PipeDrainer = struct {
fd: std.posix.fd_t,
alloc: std.mem.Allocator,
out: std.ArrayList(u8) = .empty,
fn run(self: *PipeDrainer, idle_limit_ms: i32) void {
var idle: i32 = 0;
var buf: [8192]u8 = undefined;
while (idle < idle_limit_ms) {
var fds = [_]std.posix.pollfd{
.{ .fd = self.fd, .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&fds, 100) catch return;
if (ready == 0) {
idle += 100;
continue;
}
const n = std.posix.read(self.fd, &buf) catch return;
if (n == 0) return;
self.out.appendSlice(self.alloc, buf[0..n]) catch return;
idle = 0;
}
}
};
/// Shrink a socket's buffers so a transfer larger than them cannot be
/// swallowed whole by the kernel. Set on the listener, inherited by the
/// accepted connection. Linux doubles and clamps the request, so this is a
/// floor request, not a promise.
fn shrinkBufs(fd: std.posix.fd_t) void {
const v: c_int = 1024;
std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.RCVBUF, std.mem.asBytes(&v)) catch {};
std.posix.setsockopt(fd, std.posix.SOL.SOCKET, std.posix.SO.SNDBUF, std.mem.asBytes(&v)) catch {};
}
fn readExactly(fd: std.posix.fd_t, buf: []u8) !void {
var i: usize = 0;
while (i < buf.len) {
const n = try std.posix.read(fd, buf[i..]);
if (n == 0) return error.UnexpectedEof;
i += n;
}
}
test "pump carries bytes both ways verbatim" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const dir_path = tmp.path();
var sock_buf: [280]u8 = undefined;
const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/proxy.sock", .{dir_path});
const addr = try std.net.Address.initUnix(sock_path);
var listener = try addr.listen(.{});
defer listener.deinit();
// Frame-shaped payloads with embedded zeros and 0xFF: the pump must be
// byte-transparent, not text-safe.
const to_daemon = [_]u8{ 0x01, 0x08, 0x00, 0x00, 0x00, 0xff, 0x00, 0x1b, 0x5c, 0x00, 0x7f, 0xfe, 0x0a };
const to_client = [_]u8{ 0x80, 0x06, 0x00, 0x00, 0x00, 0x00, 0xff, 0x00, 0xff, 0x1c, 0x00 };
const in_pipe = try std.posix.pipe();
defer std.posix.close(in_pipe[0]);
defer std.posix.close(in_pipe[1]);
const out_pipe = try std.posix.pipe();
defer std.posix.close(out_pipe[0]);
defer std.posix.close(out_pipe[1]);
// Queued before the pump starts; the pipe buffer holds it, so nothing here
// depends on thread scheduling.
try writeAll(in_pipe[1], &to_daemon);
var peer = EchoPeer{
.listener = &listener,
.alloc = alloc,
.want = to_daemon.len,
.reply = &to_client,
};
defer peer.got.deinit(alloc);
const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
// Returns when the peer closes — the daemon-hangup path.
try runner.wait(5000);
pump_th.join();
peer_th.join();
try std.testing.expect(!runner.failed);
try std.testing.expectEqual(@as(u8, 0), runner.rc);
try std.testing.expectEqualSlices(u8, &to_daemon, peer.got.items);
// Read a known length rather than to EOF: the write end is still open.
var out: [to_client.len]u8 = undefined;
try readExactly(out_pipe[0], &out);
try std.testing.expectEqualSlices(u8, &to_client, &out);
}
test "pump carries a large transfer verbatim under backpressure" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const dir_path = tmp.path();
var sock_buf: [280]u8 = undefined;
const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/big.sock", .{dir_path});
const addr = try std.net.Address.initUnix(sock_path);
var listener = try addr.listen(.{});
defer listener.deinit();
// Backpressure: far more data than either buffer can hold, so the copy
// loops in both directions instead of completing in one syscall.
shrinkBufs(listener.stream.handle);
const size = 300 * 1024;
const up = try alloc.alloc(u8, size);
defer alloc.free(up);
const down = try alloc.alloc(u8, size);
defer alloc.free(down);
for (up, 0..) |*b, i| b.* = @truncate(i *% 31 +% (i >> 8));
for (down, 0..) |*b, i| b.* = @truncate(i *% 37 +% (i >> 7) +% 11);
const in_pipe = try std.posix.pipe();
defer std.posix.close(in_pipe[0]);
defer std.posix.close(in_pipe[1]);
const out_pipe = try std.posix.pipe();
defer std.posix.close(out_pipe[0]);
defer std.posix.close(out_pipe[1]);
var peer = EchoPeer{ .listener = &listener, .alloc = alloc, .want = size, .reply = down };
defer peer.got.deinit(alloc);
const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
// Both ends need their own thread: 300KB exceeds a pipe buffer, so a
// single-threaded test would deadlock against its own pump.
var writer = PipeWriter{ .fd = in_pipe[1], .data = up };
const writer_th = try std.Thread.spawn(.{}, PipeWriter.run, .{&writer});
var drainer = PipeDrainer{ .fd = out_pipe[0], .alloc = alloc };
defer drainer.out.deinit(alloc);
const drain_th = try std.Thread.spawn(.{}, PipeDrainer.run, .{ &drainer, 1500 });
var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
try runner.wait(30_000);
pump_th.join();
peer_th.join();
writer_th.join();
drain_th.join();
try std.testing.expect(!runner.failed);
try std.testing.expectEqual(@as(u8, 0), runner.rc);
try std.testing.expectEqualSlices(u8, up, peer.got.items);
try std.testing.expectEqualSlices(u8, down, drainer.out.items);
}
test "pump exits when the far side of its output pipe is gone" {
const alloc = std.testing.allocator;
var tmp = try TmpDir.make();
defer tmp.cleanup();
const dir_path = tmp.path();
var sock_buf: [280]u8 = undefined;
const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/gone.sock", .{dir_path});
const addr = try std.net.Address.initUnix(sock_path);
var listener = try addr.listen(.{});
defer listener.deinit();
const in_pipe = try std.posix.pipe();
defer std.posix.close(in_pipe[0]);
defer std.posix.close(in_pipe[1]);
const out_pipe = try std.posix.pipe();
defer std.posix.close(out_pipe[1]);
// The reader is gone before a byte is written: writing to out_fd can only
// fail. The pump must report that as a failed transfer (1), not die of
// SIGPIPE and not spin.
std.posix.close(out_pipe[0]);
var peer = EchoPeer{ .listener = &listener, .alloc = alloc, .want = 0, .reply = "x" };
defer peer.got.deinit(alloc);
const peer_th = try std.Thread.spawn(.{}, EchoPeer.run, .{&peer});
var runner = PumpRunner{ .in_fd = in_pipe[0], .out_fd = out_pipe[1], .sock_path = sock_path };
const pump_th = try std.Thread.spawn(.{}, PumpRunner.run, .{&runner});
try runner.wait(5000);
pump_th.join();
peer_th.join();
try std.testing.expect(!runner.failed);
try std.testing.expectEqual(@as(u8, 1), runner.rc);
}
test "pump reports a missing daemon socket instead of hanging" {
var tmp = try TmpDir.make();
defer tmp.cleanup();
const dir_path = tmp.path();
var sock_buf: [280]u8 = undefined;
const sock_path = try std.fmt.bufPrint(&sock_buf, "{s}/absent.sock", .{dir_path});
// No fds are touched on this path, so -1 would do; use the real stdio
// constants to keep the call shape honest.
try std.testing.expectEqual(
@as(u8, 1),
try pump(std.posix.STDIN_FILENO, std.posix.STDOUT_FILENO, sock_path),
);
}
// Forces semantic analysis of every pub decl under `zig build test`, so an
// unreferenced decl must at least compile (the silent-module-loss hazard,
// decisions.md). Pub decls only: std.meta.declarations sees nothing private.
test {
std.testing.refAllDeclsRecursive(@This());
}