path: root/src/9p_io.zig

const std = @import("std");
const libc = std.c;
const builtin = @import("builtin");
const ninep = @import("9p.zig");
const transport = @import("cloud9").transport;
const pardes = @import("pardes.zig");
const limits = @import("memory.zig").limits;
pub const quic_enabled = @import("9p_options").quic;
const quic = if (quic_enabled) @import("9p_quic.zig") else struct {};

const log = std.log.scoped(.ninep);

pub const darwin = switch (builtin.os.tag) {
    .macos, .ios, .tvos, .watchos, .visionos => true,
    else => false,
};
pub const supported = builtin.os.tag == .linux or darwin;
pub const sun_path_len = @typeInfo(@FieldType(libc.sockaddr.un, "path")).array.len;

pub fn setCloexec(fd: c_int) void {
    _ = libc.fcntl(fd, libc.F.SETFD, @as(c_int, 1));
}

pub fn socketDir(buf: *[sun_path_len:0]u8) ?[:0]const u8 {
    if (libc.getenv("XDG_RUNTIME_DIR")) |path|
        return std.fmt.bufPrintSentinel(buf, "{s}", .{std.mem.span(path)}, 0) catch null;
    const home = libc.getenv("HOME") orelse return null;
    return std.fmt.bufPrintSentinel(buf, "{s}/.local/state/pardes", .{std.mem.span(home)}, 0) catch null;
}

pub const FileFacts = struct { mode: u32, uid: libc.uid_t };

pub fn statNoFollow(path: [:0]const u8) ?FileFacts {
    if (comptime darwin) {
        var stat: libc.Stat = undefined;
        if (libc.fstatat(libc.AT.FDCWD, path, &stat, libc.AT.SYMLINK_NOFOLLOW) != 0) return null;
        return .{ .mode = stat.mode, .uid = stat.uid };
    } else {
        const linux = std.os.linux;
        var stat: linux.Statx = undefined;
        const fields: linux.STATX = .{ .TYPE = true, .MODE = true, .UID = true };
        if (libc.statx(linux.AT.FDCWD, path, linux.AT.SYMLINK_NOFOLLOW, fields, &stat) != 0) return null;
        return .{ .mode = stat.mode, .uid = stat.uid };
    }
}

pub fn ensureSocketDir(dir: [:0]const u8) bool {
    if (dir.len == 0) return false;
    var partial: [sun_path_len:0]u8 = undefined;
    @memcpy(partial[0 .. dir.len + 1], dir[0 .. dir.len + 1]);
    for (1..dir.len) |i| {
        if (dir[i] != '/') continue;
        partial[i] = 0;
        _ = libc.mkdir(partial[0..i :0], 0o700);
        partial[i] = '/';
    }
    _ = libc.mkdir(dir, 0o700);
    const stat = statNoFollow(dir) orelse return false;
    return stat.mode & 0o170000 == 0o040000 and stat.uid == libc.getuid() and stat.mode & 0o077 == 0;
}

const prefix = "pardes-9p-";

pub const msize: u32 = 8192;

pub const max_conns = 4;

extern "c" fn inet_pton(family: c_int, src: [*:0]const u8, dst: *anyopaque) c_int;

pub fn networkAddress(dial: []const u8, allow_zero_port: bool) error{BadDial}!std.Io.net.IpAddress {
    if (comptime !supported) return error.BadDial;
    const host_start: usize = if (std.mem.startsWith(u8, dial, "tcp!")) 4 else if (std.mem.startsWith(u8, dial, "quic!")) 5 else return error.BadDial;
    const split = std.mem.lastIndexOfScalar(u8, dial, '!') orelse return error.BadDial;
    if (split <= host_start or split + 1 == dial.len) return error.BadDial;
    const port_text = dial[split + 1 ..];
    for (port_text) |c| if (c < '0' or c > '9') return error.BadDial;
    const port = std.fmt.parseInt(u16, port_text, 10) catch return error.BadDial;
    if (port == 0 and !allow_zero_port) return error.BadDial;
    const host = dial[host_start..split];
    if (std.mem.indexOfScalar(u8, host, 0) != null) return error.BadDial;
    var host_buf: [46]u8 = undefined;
    const host_z = std.fmt.bufPrintSentinel(&host_buf, "{s}", .{host}, 0) catch return error.BadDial;
    var ip4: std.Io.net.Ip4Address = .{ .port = port, .bytes = undefined };
    if (inet_pton(libc.AF.INET, host_z, &ip4.bytes) == 1) return .{ .ip4 = ip4 };
    var ip6: std.Io.net.Ip6Address = .{ .port = port, .bytes = undefined };
    if (inet_pton(libc.AF.INET6, host_z, &ip6.bytes) == 1) return canonicalIp(.{ .ip6 = ip6 });
    return error.BadDial;
}

fn canonicalIp(address: std.Io.net.IpAddress) std.Io.net.IpAddress {
    if (address == .ip6) {
        if (std.Io.net.Ip4Address.fromIp6(address.ip6)) |ip4| return .{ .ip4 = ip4 };
    }
    return address;
}

fn sockaddrIp(address: *const libc.sockaddr) ?std.Io.net.IpAddress {
    return switch (address.family) {
        libc.AF.INET => blk: {
            const addr: *const libc.sockaddr.in = @ptrCast(@alignCast(address));
            break :blk .{ .ip4 = .{ .port = std.mem.bigToNative(u16, addr.port), .bytes = @bitCast(addr.addr) } };
        },
        libc.AF.INET6 => blk: {
            const addr: *const libc.sockaddr.in6 = @ptrCast(@alignCast(address));
            break :blk canonicalIp(.{ .ip6 = .{ .port = std.mem.bigToNative(u16, addr.port), .bytes = addr.addr } });
        },
        else => null,
    };
}

const IfAddr = extern struct {
    next: ?*IfAddr,
    name: ?[*:0]u8,
    flags: c_uint,
    address: ?*libc.sockaddr,
    netmask: ?*libc.sockaddr,
    destination: ?*libc.sockaddr,
    data: ?*anyopaque,
};

extern "c" fn getifaddrs(out: *?*IfAddr) c_int;
extern "c" fn freeifaddrs(first: *IfAddr) void;

fn localIp(address: std.Io.net.IpAddress) bool {
    switch (address) {
        .ip4 => |ip| if (ip.bytes[0] == 127 or std.mem.allEqual(u8, &ip.bytes, 0)) return true,
        .ip6 => |ip| if (ip.isLoopBack() or std.mem.allEqual(u8, &ip.bytes, 0)) return true,
    }
    var first: ?*IfAddr = null;
    if (getifaddrs(&first) != 0) return false;
    defer if (first) |head| freeifaddrs(head);
    var next = first;
    while (next) |entry| : (next = entry.next) {
        var local = sockaddrIp(entry.address orelse continue) orelse continue;
        local.setPort(address.getPort());
        if (address.eql(&local)) return true;
    }
    return false;
}

const Srv = ninep.Server(pardes.filesystem, ninep.max_fids);

const Conn = struct {
    fd: c_int = -1,
    quic: if (quic_enabled) ?quic.Connection else void = if (quic_enabled) null else {},
    draining: bool = false,
    accepted_ms: i64 = 0,
    srv: Srv = undefined,
    in: [msize]u8 = undefined,
    out: [2 * msize]u8 = undefined,

    fn step(c: *Conn, core: *pardes.Pardes, req: pardes.filesystem.Req) void {
        core.update(.{ .fs_req = req });
        var answered = false;
        while (core.nextEffect()) |effect| {
            if (effect == .fs_reply) {
                const reply = effect.fs_reply;
                if (reply.tag == req.tag) answered = true;
                c.srv.reply(&reply, core.fsPayload(reply));
            } else core.perform(effect);
        }
        if (!answered) {
            const reply = pardes.filesystem.Reply.fail(req.tag, pardes.filesystem.E.IO);
            c.srv.reply(&reply, "");
        }
    }
};

const accept_pause_ms: i64 = 100;

pub const Listener = struct {
    fd: c_int = -1,
    tcp_fd: c_int = -1,
    tcp_address: ?std.Io.net.IpAddress = null,
    quic: if (quic_enabled) ?quic.Listener else void = if (quic_enabled) null else {},
    quic_address: ?std.Io.net.IpAddress = null,
    paused_ms: i64 = 0,
    path_buf: [sun_path_len]u8 = undefined,
    path_len: usize = 0,
    conns: [max_conns]Conn = @splat(.{}),
    control: [2]c_int = .{ -1, -1 },
    watcher: ?std.Thread = null,
    stopping: std.atomic.Value(bool) = .init(false),
    watch_lock: std.atomic.Mutex = .unlocked,
    watch_fds: [max_conns + 3 + @as(usize, @intFromBool(quic_enabled))]libc.pollfd = undefined,
    watch_len: usize = 0,
    watch_timeout: c_int = -1,
    wake_ctx: ?*anyopaque = null,
    wake: ?*const fn (?*anyopaque) void = null,

    pub fn path(l: *const Listener) []const u8 {
        return l.path_buf[0..l.path_len];
    }

    fn listenTcp(l: *Listener, address: std.Io.net.IpAddress) !void {
        const fd = try transport.listenFd(.{ .tcp = address }, max_conns);
        errdefer transport.close(fd);
        var addr: libc.sockaddr.storage = undefined;
        var len: libc.socklen_t = @sizeOf(@TypeOf(addr));
        if (libc.getsockname(fd, @ptrCast(&addr), &len) != 0) return error.SocketAddressFailed;
        l.tcp_address = sockaddrIp(@ptrCast(&addr)) orelse return error.SocketAddressFailed;
        l.tcp_fd = fd;
        log.info("serving 9P2000 over TCP on {f}", .{l.tcp_address.?});
    }

    pub fn accept(l: *Listener) void {
        if (comptime !supported) return;
        for ([_]c_int{ l.fd, l.tcp_fd }) |listener_fd| {
            if (listener_fd < 0) continue;
            for (0..max_conns + 1) |_| {
                const fd = (transport.acceptFd(listener_fd, listener_fd == l.tcp_fd) catch {
                    l.paused_ms = nowMs() +| accept_pause_ms;
                    log.warn("accept failed; pausing the listener for {d} ms", .{accept_pause_ms});
                    return;
                }) orelse break;
                const c = for (&l.conns, 0..) |*cand, i| {
                    if (!l.live(@intCast(i)) and !cand.draining) break cand;
                } else {
                    log.debug("refusing a connection, all {d} slots busy", .{max_conns});
                    _ = libc.close(fd);
                    continue;
                };
                c.fd = fd;
                c.draining = false;
                c.accepted_ms = nowMs();
                c.srv = .init(.{
                    .in = &c.in,
                    .out = &c.out,
                    .root = pardes.filesystem.namespace_root,
                });
            }
        }
        if (comptime quic_enabled) {
            if (l.quic) |*listener| for (0..max_conns + 1) |_| {
                var connection = (listener.accept() catch |err| {
                    log.warn("QUIC accept failed: {s}", .{@errorName(err)});
                    return;
                }) orelse break;
                const c = for (&l.conns, 0..) |*cand, i| {
                    if (!l.live(@intCast(i)) and !cand.draining) break cand;
                } else {
                    connection.deinit();
                    continue;
                };
                c.fd = -1;
                c.quic = connection;
                c.draining = false;
                c.accepted_ms = nowMs();
                c.srv = .init(.{ .in = &c.in, .out = &c.out, .root = pardes.filesystem.namespace_root });
            };
        }
    }

    pub const greet_deadline_ms: i64 = 5000;

    pub fn expire(l: *Listener) void {
        if (comptime !supported) return;
        const now = nowMs();
        if (now == 0) return;
        for (&l.conns, 0..) |*c, i| {
            if (!l.live(@intCast(i)) or c.srv.protocol.msize != 0) continue;
            if (now - c.accepted_ms < greet_deadline_ms) continue;
            log.debug("slot {d} never sent Tversion; taking it back", .{i});
            l.drop(@intCast(i));
        }
    }

    pub fn accepting(l: *const Listener) bool {
        if (comptime !supported) return false;
        if (l.fd < 0 and l.tcp_fd < 0) return false;
        if (l.paused_ms == 0) return true;
        const now = nowMs();
        return now == 0 or now >= l.paused_ms;
    }

    pub fn nextDue(l: *const Listener) ?i32 {
        if (comptime !supported) return null;
        const now = nowMs();
        if (now == 0) return null;
        var due: ?i64 = null;
        if (l.paused_ms > now) due = l.paused_ms;
        if (comptime quic_enabled) {
            if (l.quic) |*listener| if (listener.nextDue()) |ms| {
                const at = now + ms;
                due = if (due) |d| @min(d, at) else at;
            };
        }
        for (&l.conns, 0..) |*c, i| {
            if (!l.live(@intCast(i))) continue;
            if (comptime quic_enabled) {
                if (c.quic) |*connection| if (connection.pending()) return 0;
            }
            if (c.srv.protocol.msize != 0) continue;
            const at = c.accepted_ms + greet_deadline_ms;
            due = if (due) |d| @min(d, at) else at;
        }
        const at = due orelse return null;
        return @intCast(@max(0, at - now));
    }

    const nowMs = transport.nowMs;

    pub fn fill(l: *Listener, i: u8) void {
        if (comptime !supported) return;
        const c = &l.conns[i];
        if (c.srv.protocol.dead) return l.drop(i);
        const room = c.srv.protocol.in.len - c.srv.protocol.in_len;
        if (room == 0) return;
        var buf: [msize]u8 = undefined;
        const got = if (quic_enabled and c.quic != null)
            (c.quic.?.read(buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return
        else
            (transport.read(c.fd, buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return;
        if (got == 0) return l.drop(i);
        const n = c.srv.push(buf[0..@intCast(got)]);
        if (c.srv.protocol.dead) return l.drop(i);
        std.debug.assert(n == @as(usize, @intCast(got)));
    }

    pub fn flush(l: *Listener, i: u8) void {
        if (comptime !supported) return;
        const c = &l.conns[i];
        if (!l.live(i)) return;
        while (true) {
            const bytes = c.srv.output();
            if (bytes.len == 0) return;
            const n = if (quic_enabled and c.quic != null)
                c.quic.?.write(bytes) catch return l.drop(i)
            else
                (transport.write(c.fd, bytes) catch return l.drop(i)) orelse return;
            if (n == 0) return;
            c.srv.wrote(@intCast(n));
        }
    }

    pub fn owes(l: *const Listener, i: u8) bool {
        return l.conns[i].srv.output().len != 0;
    }

    pub fn live(l: *const Listener, i: u8) bool {
        return l.conns[i].fd >= 0 or (quic_enabled and l.conns[i].quic != null);
    }

    pub fn drop(l: *Listener, i: u8) void {
        const c = &l.conns[i];
        if (c.fd >= 0) {
            _ = libc.close(c.fd);
            c.fd = -1;
        }
        if (comptime quic_enabled) {
            if (c.quic) |*connection| {
                connection.deinit();
                c.quic = null;
            }
        }
        if (c.draining or c.accepted_ms == 0) return;
        c.srv.hangup();
        c.draining = true;
    }

    pub const Drained = struct { count: usize = 0, pending: bool = false };

    pub fn drain(l: *Listener, core: *pardes.Pardes) Drained {
        core.fs.socket_path = l.path();
        core.fs.tcp_address = l.tcp_address;
        core.fs.quic_address = l.quic_address;
        l.expire();
        var result: Drained = .{};
        for (&l.conns, 0..) |*conn, i| {
            if (!l.live(@intCast(i)) and !conn.draining) continue;
            var count: usize = 0;
            while (conn.srv.retry()) |req| {
                conn.step(core, req);
                l.collectOs(core);
                count += 1;
            }
            while (count < 64) {
                const req = conn.srv.next() orelse break;
                conn.step(core, req);
                l.collectOs(core);
                count += 1;
            }
            result.count += count;
            result.pending = result.pending or count >= 64;
            if (conn.draining) {
                if (count == 0) conn.draining = false else result.pending = true;
            } else l.flush(@intCast(i));
            if (comptime quic_enabled) {
                if (conn.quic) |*connection| result.pending = result.pending or connection.pending();
            }
        }
        return result;
    }

    pub fn tick(l: *Listener, core: *pardes.Pardes) Drained {
        if (comptime quic_enabled) {
            if (l.quic) |*listener| listener.events() catch |err| {
                log.warn("QUIC listener stopped: {s}", .{@errorName(err)});
                for (&l.conns, 0..) |*conn, i| if (conn.quic != null) l.drop(@intCast(i));
                listener.deinit();
                l.quic = null;
                l.quic_address = null;
            };
        }
        if (l.accepting()) l.accept();
        for (0..max_conns) |i| if (l.live(@intCast(i))) l.fill(@intCast(i));
        const result = l.drain(core);
        l.arm();
        return result;
    }

    pub fn reset(l: *Listener, core: *pardes.Pardes) void {
        for (0..max_conns) |i| l.drop(@intCast(i));
        while (l.drain(core).pending) {}
        l.collectOs(core);
        for (&l.conns) |*conn| conn.accepted_ms = 0;
        l.arm();
    }

    fn collectOs(l: *Listener, core: *pardes.Pardes) void {
        var i: usize = 0;
        while (i < core.fs.os_paths.items.len) {
            const entry = core.fs.os_paths.items[i];
            var held = false;
            for (&l.conns, 0..) |*conn, j| {
                if (!l.live(@intCast(j)) and !conn.draining) continue;
                if (conn.srv.references(entry.node)) {
                    held = true;
                    break;
                }
            }
            if (held) {
                i += 1;
            } else {
                core.gpa.free(entry.path);
                _ = core.fs.os_paths.swapRemove(i);
            }
        }
    }

    pub fn wakeThread(l: *Listener, ctx: ?*anyopaque, wake: *const fn (?*anyopaque) void) !void {
        if (l.watcher != null) return;
        if (libc.pipe(&l.control) != 0) return error.PipeFailed;
        errdefer {
            _ = libc.close(l.control[0]);
            _ = libc.close(l.control[1]);
            l.control = .{ -1, -1 };
        }
        for (l.control) |fd| {
            setCloexec(fd);
            setNonblock(fd);
        }
        l.wake_ctx = ctx;
        l.wake = wake;
        l.arm();
        l.watcher = try std.Thread.spawn(.{}, watch, .{l});
    }

    fn arm(l: *Listener) void {
        if (l.control[1] < 0) return;
        while (!l.watch_lock.tryLock()) std.atomic.spinLoopHint();
        l.watch_fds[0] = .{ .fd = l.control[0], .events = @intCast(libc.POLL.IN), .revents = 0 };
        l.watch_len = 1;
        if (l.accepting()) {
            for ([_]c_int{ l.fd, l.tcp_fd }) |fd| {
                if (fd < 0) continue;
                l.watch_fds[l.watch_len] = .{ .fd = fd, .events = @intCast(libc.POLL.IN), .revents = 0 };
                l.watch_len += 1;
            }
        }
        if (comptime quic_enabled) {
            if (l.quic) |*listener| {
                l.watch_fds[l.watch_len] = listener.poll();
                l.watch_len += 1;
            }
        }
        for (0..max_conns) |i| {
            if (l.conns[i].fd < 0) continue;
            l.watch_fds[l.watch_len] = .{
                .fd = l.conns[i].fd,
                .events = @as(i16, @intCast(libc.POLL.IN)) | if (l.owes(@intCast(i))) @as(i16, @intCast(libc.POLL.OUT)) else 0,
                .revents = 0,
            };
            l.watch_len += 1;
        }
        l.watch_timeout = l.nextDue() orelse -1;
        l.watch_lock.unlock();
        _ = libc.write(l.control[1], "w", 1);
    }

    fn watch(l: *Listener) void {
        var notified = false;
        while (!l.stopping.load(.acquire)) {
            var fds: [max_conns + 3 + @as(usize, @intFromBool(quic_enabled))]libc.pollfd = undefined;
            while (!l.watch_lock.tryLock()) std.atomic.spinLoopHint();
            const len = if (notified) 1 else l.watch_len;
            @memcpy(fds[0..len], l.watch_fds[0..len]);
            const timeout = if (notified) -1 else l.watch_timeout;
            l.watch_lock.unlock();
            const ready = libc.poll(&fds, @intCast(len), timeout);
            if (ready < 0) continue;
            if (fds[0].revents != 0) {
                var buf: [64]u8 = undefined;
                while (libc.read(l.control[0], &buf, buf.len) > 0) {}
                notified = false;
                continue;
            }
            if (!l.stopping.load(.acquire)) l.wake.?(l.wake_ctx);
            notified = true;
        }
    }

    pub fn deinit(l: *Listener, gpa: std.mem.Allocator) void {
        if (l.watcher) |thread| {
            l.stopping.store(true, .release);
            _ = libc.write(l.control[1], "q", 1);
            thread.join();
            for (l.control) |fd| _ = libc.close(fd);
        }
        for (0..max_conns) |i| l.drop(@intCast(i));
        if (comptime quic_enabled) {
            if (l.quic) |*listener| listener.deinit();
        }
        if (l.tcp_fd >= 0) _ = libc.close(l.tcp_fd);
        if (l.fd >= 0) {
            _ = libc.close(l.fd);
            l.fd = -1;
            var z: [sun_path_len:0]u8 = undefined;
            @memcpy(z[0..l.path_len], l.path_buf[0..l.path_len]);
            z[l.path_len] = 0;
            _ = libc.unlink(z[0..l.path_len :0]);
        }
        gpa.destroy(l);
    }
};

pub fn socketPath(buf: *[sun_path_len]u8, dir: []const u8, name: []const u8) ?[:0]const u8 {
    if (name.len == 0) return null;
    if (std.mem.indexOfAny(u8, name, "/\x00") != null) return null;
    return std.fmt.bufPrintSentinel(buf, "{s}/" ++ prefix ++ "{s}.sock", .{ dir, name }, 0) catch null;
}

pub fn listen(gpa: std.mem.Allocator, named: []const u8, fallback: []const u8, tcp_dial: ?[]const u8, quic_dial: ?[]const u8) ?*Listener {
    if (comptime !supported) return null;
    var dir_buf: [sun_path_len:0]u8 = undefined;
    const dir = socketDir(&dir_buf) orelse {
        log.warn("no runtime directory for the socket", .{});
        return null;
    };
    if (!ensureSocketDir(dir)) return null;
    const l = gpa.create(Listener) catch return null;
    l.* = .{};
    const p = socketPath(&l.path_buf, dir, if (named.len != 0) named else fallback) orelse {
        gpa.destroy(l);
        return null;
    };
    const fd = transport.listenFd(.{ .unix = p }, max_conns) catch |err| retry: {
        const io = std.Io.Threaded.global_single_threaded.io();
        const existing = std.Io.Dir.cwd().statFile(io, p, .{ .follow_symlinks = false }) catch null;
        if (err != error.Bind or existing == null or existing.?.kind != .unix_domain_socket or alive(p)) {
            log.warn("something is already listening on {s}", .{p});
            gpa.destroy(l);
            return null;
        }
        if (libc.unlink(p) != 0) {
            gpa.destroy(l);
            return null;
        }
        break :retry transport.listenFd(.{ .unix = p }, max_conns) catch {
            gpa.destroy(l);
            return null;
        };
    };
    if (libc.chmod(p, 0o600) != 0) {
        transport.close(fd);
        _ = libc.unlink(p);
        gpa.destroy(l);
        return null;
    }
    l.fd = fd;
    l.path_len = p.len;
    if (tcp_dial) |dial| {
        if (!std.mem.startsWith(u8, dial, "tcp!")) {
            l.deinit(gpa);
            return null;
        }
        const address = networkAddress(dial, true) catch {
            log.warn("invalid TCP address {s}", .{dial});
            l.deinit(gpa);
            return null;
        };
        l.listenTcp(address) catch |err| {
            log.warn("cannot listen on {s}: {s}", .{ dial, @errorName(err) });
            l.deinit(gpa);
            return null;
        };
    }
    if (quic_dial) |dial| {
        if (comptime quic_enabled) {
            if (!std.mem.startsWith(u8, dial, "quic!")) {
                l.deinit(gpa);
                return null;
            }
            const address = networkAddress(dial, true) catch {
                log.warn("invalid QUIC address {s}", .{dial});
                l.deinit(gpa);
                return null;
            };
            l.quic = quic.Listener.init(address) catch |err| {
                log.warn("cannot listen on {s}: {s}", .{ dial, @errorName(err) });
                l.deinit(gpa);
                return null;
            };
            l.quic_address = l.quic.?.address;
            log.info("serving 9P2000 over QUIC on {f}", .{l.quic_address.?});
        } else {
            log.warn("QUIC is unavailable in this build", .{});
            l.deinit(gpa);
            return null;
        }
    }
    log.info("serving 9P2000 on {s}", .{p});
    return l;
}

const alive = transport.isListening;

fn setNonblock(fd: c_int) void {
    const flags = libc.fcntl(fd, libc.F.GETFL, @as(c_int, 0));
    if (flags < 0) return;
    var o: libc.O = @bitCast(@as(u32, @bitCast(flags)));
    o.NONBLOCK = true;
    _ = libc.fcntl(fd, libc.F.SETFL, @as(c_int, @bitCast(@as(u32, @bitCast(o)))));
}

const testing = std.testing;

test "the socket name is a third prefix in the shared directory" {
    var buf: [sun_path_len]u8 = undefined;
    const p = socketPath(&buf, "/run/user/1000", "t9srv").?;
    try testing.expectEqualStrings("/run/user/1000/pardes-9p-t9srv.sock", p);
    try testing.expect(!std.mem.startsWith(u8, std.fs.path.basename(p), "pardes-detached-"));
}

test "a name that is not one path component is no address at all" {
    var buf: [sun_path_len]u8 = undefined;
    try testing.expect(socketPath(&buf, "/run", "") == null);
    try testing.expect(socketPath(&buf, "/run", "a/b") == null);
    try testing.expect(socketPath(&buf, "/run", "a\x00b") == null);
}

test "one connection's buffers are sized from the one msize constant" {
    try testing.expect(msize >= ninep.min_msize);
    const c: Conn = .{};
    try testing.expectEqual(@as(usize, msize), c.in.len);
    try testing.expectEqual(@as(usize, 2 * msize), c.out.len);
}

test "TCP addresses are numeric and normalize mapped IPv4" {
    const loopback = try networkAddress("tcp!127.0.0.1!5640", false);
    try testing.expectEqualDeep(loopback, try networkAddress("tcp!::ffff:127.0.0.1!5640", false));
    try testing.expectEqualDeep(loopback, try networkAddress("tcp!::ffff:7f00:1!5640", false));
    try testing.expectEqualDeep(try networkAddress("tcp!::1!5640", false), try networkAddress("tcp!0:0:0:0:0:0:0:1!5640", false));
    try testing.expectEqual(@as(u16, 0), (try networkAddress("tcp!127.0.0.1!0", true)).getPort());
    for ([_][]const u8{ "tcp!localhost!5640", "tcp!127.0.0.1!0", "tcp!127.0.0.1!-1", "tcp!127.0.0.1!65536", "tcp!127.0.0.1!", "tcp!!5640" }) |dial|
        try testing.expectError(error.BadDial, networkAddress(dial, false));
    try Client.validateDial("tcp!127.0.0.1!5640");
    try Client.validateDial("/tmp/pardes-owned.sock");
    try Client.validateDial("unix!/tmp/pardes-owned.sock");
    try testing.expectError(error.BadDial, Client.validateDial("unix!work"));
    try testing.expectError(error.BadDial, Client.validateDial("unix!"));
    try testing.expectError(error.BadDial, Client.validateDial("unix!/tmp/a\x00b"));
    try testing.expectError(error.BadDial, Client.validateDial("tcp!localhost!5640"));
    try testing.expectError(error.BadDial, Client.validateDial("/tmp/a\x00b"));
    if (quic_enabled) {
        try Client.validateDial("quic!127.0.0.1!5640");
    } else try testing.expectError(error.QuicUnavailable, Client.validateDial("quic!127.0.0.1!5640"));
}

test "same-session TCP mounts compare canonical endpoints and local wildcard destinations" {
    if (comptime !supported) return error.SkipZigTest;
    const Case = struct { bound: []const u8, dial: []const u8, same: bool };
    for ([_]Case{
        .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!::ffff:127.0.0.1!5640", .same = true },
        .{ .bound = "tcp!::ffff:127.0.0.1!5640", .dial = "tcp!127.0.0.1!5640", .same = true },
        .{ .bound = "tcp!::1!5640", .dial = "tcp!0:0:0:0:0:0:0:1!5640", .same = true },
        .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!0.0.0.0!5640", .same = true },
        .{ .bound = "tcp!::1!5640", .dial = "tcp!::!5640", .same = true },
        .{ .bound = "tcp!0.0.0.0!5640", .dial = "tcp!127.0.0.2!5640", .same = true },
        .{ .bound = "tcp!::!5640", .dial = "tcp!::1!5640", .same = true },
        .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!127.0.0.1!5641", .same = false },
        .{ .bound = "tcp!0.0.0.0!5640", .dial = "tcp!192.0.2.1!5640", .same = false },
        .{ .bound = "tcp!::!5640", .dial = "tcp!2001:db8::1!5640", .same = false },
        .{ .bound = "tcp!::!5640", .dial = "tcp!127.0.0.1!5640", .same = false },
    }) |c| try testing.expectEqual(c.same, Client.sameSession(c.dial, "", try networkAddress(c.bound, true), null));
    try testing.expect(Client.sameSession("/tmp/pardes-owned.sock", "/tmp/pardes-owned.sock", null, null));
    try testing.expect(Client.sameSession("unix!/tmp/pardes-owned.sock", "/tmp/pardes-owned.sock", null, null));
    try testing.expect(!Client.sameSession("tcp!127.0.0.1!5640", "/tmp/pardes-owned.sock", null, null));
    if (quic_enabled) {
        const endpoint = try networkAddress("quic!127.0.0.1!5640", false);
        try testing.expect(Client.sameSession("quic!::ffff:127.0.0.1!5640", "", null, endpoint));
        try testing.expect(!Client.sameSession("tcp!127.0.0.1!5640", "", null, endpoint));
        try testing.expect(!Client.sameSession("quic!127.0.0.1!5640", "", endpoint, null));
    }
}

extern "c" fn mkdtemp(template: [*:0]u8) ?[*:0]u8;
extern "c" fn rmdir(path: [*:0]const u8) c_int;

test "Unix TCP and QUIC share one listener through reads writes reconnects and reset" {
    if (comptime !supported) return error.SkipZigTest;
    const gpa = testing.allocator;
    var directory: [64:0]u8 = undefined;
    _ = try std.fmt.bufPrintSentinel(&directory, "/tmp/pardes-tcp-XXXXXX", .{}, 0);
    if (mkdtemp(&directory) == null) return error.TempDirectoryFailed;
    defer _ = rmdir(&directory);
    const old_runtime = if (libc.getenv("XDG_RUNTIME_DIR")) |v| try gpa.dupeZ(u8, std.mem.span(v)) else null;
    defer {
        if (old_runtime) |v| {
            _ = setenv("XDG_RUNTIME_DIR", v, 1);
            gpa.free(v);
        } else _ = unsetenv("XDG_RUNTIME_DIR");
    }
    try testing.expectEqual(@as(c_int, 0), setenv("XDG_RUNTIME_DIR", &directory, 1));
    const replacement = try gpa.alloc(u8, 3 * msize + 27);
    defer gpa.free(replacement);
    @memset(replacement, 'x');
    @memcpy(replacement[0.."changed café λ\n".len], "changed café λ\n");
    replacement[replacement.len - 1] = '\n';

    const Worker = struct {
        dial: []const u8,
        body_path: []const u8,
        expected: []const u8,
        replacement: []const u8,
        done: std.atomic.Value(bool) = .init(false),
        failure: ?anyerror = null,

        fn run(w: *@This()) void {
            defer w.done.store(true, .release);
            w.check() catch |err| {
                w.failure = err;
            };
        }

        fn check(w: *@This()) !void {
            const before = try Client.readLimit(testing.allocator, w.dial, w.body_path, w.body_path, w.expected.len);
            defer testing.allocator.free(before);
            try testing.expectEqualStrings(w.expected, before);
            try testing.expectError(error.FileTooLarge, Client.readLimit(testing.allocator, w.dial, w.body_path, w.body_path, w.expected.len - 1));
            const listing = try Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", 128);
            defer testing.allocator.free(listing);
            const exact_listing = try Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", listing.len);
            defer testing.allocator.free(exact_listing);
            try testing.expectEqualStrings(listing, exact_listing);
            try testing.expectError(error.FileTooLarge, Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", listing.len - 1));
            try Client.write(testing.allocator, w.dial, w.body_path, w.replacement);
            const after = try Client.read(testing.allocator, w.dial, w.body_path, w.body_path);
            defer testing.allocator.free(after);
            try testing.expectEqualStrings(w.replacement, after);
            const screen = try Client.read(testing.allocator, w.dial, "/self/screen", "/self/screen");
            defer testing.allocator.free(screen);
            const parsed = try std.json.parseFromSlice(struct { cols: u16, rows: u16 }, testing.allocator, screen, .{ .ignore_unknown_fields = true });
            defer parsed.deinit();
            try testing.expectEqual(@as(u16, 40), parsed.value.cols);
            try testing.expectEqual(@as(u16, 12), parsed.value.rows);
        }
    };

    const protocols: []const []const u8 = if (quic_enabled) &.{ "tcp", "quic" } else &.{"tcp"};
    for (protocols) |protocol| for ([_][]const u8{ "127.0.0.1", "::1" }) |host| {
        const p = try pardes.Pardes.init(gpa, .{ .tty_only = true, .cols = 40, .rows = 12 });
        defer p.deinit();
        const pane = try p.setTestFile("initial\n");
        while (p.nextEffect()) |_| {}
        var bind_buf: [64]u8 = undefined;
        const bind = try std.fmt.bufPrint(&bind_buf, "{s}!{s}!0", .{ protocol, host });
        const tcp = std.mem.eql(u8, protocol, "tcp");
        const l = listen(gpa, "roundtrip", "", if (tcp) bind else null, if (tcp) null else bind) orelse return error.ListenFailed;
        defer {
            l.reset(p);
            l.deinit(gpa);
        }
        try testing.expect(l.fd >= 0);
        try testing.expectEqual(tcp, l.tcp_fd >= 0);
        try testing.expectEqual(@as(usize, 4), l.conns.len);
        try testing.expect(l.watcher == null);
        const port = (if (tcp) l.tcp_address else l.quic_address).?.getPort();
        try testing.expect(port != 0);
        var dial_buf: [64]u8 = undefined;
        const network_dial = try std.fmt.bufPrint(&dial_buf, "{s}!{s}!{d}", .{ protocol, host, port });
        var body_buf: [64]u8 = undefined;
        const body = try std.fmt.bufPrint(&body_buf, "/self/pane/{d}/body", .{pane.serial});
        for ([_][]const u8{ network_dial, l.path(), network_dial }, 0..) |dial, attempt| {
            var worker: Worker = .{ .dial = dial, .body_path = body, .expected = if (attempt == 0) "initial\n" else replacement, .replacement = replacement };
            const thread = try std.Thread.spawn(.{}, Worker.run, .{&worker});
            defer thread.join();
            const deadline = Client.nowMs() + 3 * Client.budget_ms;
            while (!worker.done.load(.acquire) and Client.nowMs() < deadline) {
                _ = l.tick(p);
                Client.nap(1);
            }
            try testing.expect(worker.done.load(.acquire));
            if (worker.failure) |err| return err;
            const unix_fd = l.fd;
            const tcp_fd = l.tcp_fd;
            l.reset(p);
            try testing.expectEqual(unix_fd, l.fd);
            try testing.expectEqual(tcp_fd, l.tcp_fd);
            for (&l.conns, 0..) |conn, i| try testing.expect(!l.live(@intCast(i)) and !conn.draining);
            for (p.fs.snapshots) |snapshot| try testing.expect(snapshot.node == 0);
        }
    };
}

extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
extern "c" fn unsetenv(name: [*:0]const u8) c_int;

pub fn start(gpa: std.mem.Allocator, core: *pardes.Pardes) ?*Listener {
    var name: [16]u8 = undefined;
    const fallback = std.fmt.bufPrint(&name, "{d}", .{@as(u32, @intCast(libc.getpid()))}) catch unreachable;
    const listener = listen(gpa, core.opts.ninep_name, fallback, core.opts.ninep_tcp, core.opts.ninep_quic) orelse {
        core.reportError(0, "9p listener", error.ListenFailed);
        return null;
    };
    core.fs.socket_path = listener.path();
    core.fs.tcp_address = listener.tcp_address;
    core.fs.quic_address = listener.quic_address;
    return listener;
}

pub fn exportPaneEnv(listener: ?*const Listener, serial: u32, forward_look: bool) void {
    _ = unsetenv("PARDES_FORWARD_LOOK");
    if (listener) |l| exporting: {
        var sock: [sun_path_len]u8 = undefined;
        const path = std.fmt.bufPrintSentinel(&sock, "{s}", .{l.path()}, 0) catch break :exporting;
        var buf: [16]u8 = undefined;
        const id = std.fmt.bufPrintSentinel(&buf, "{d}", .{serial}, 0) catch break :exporting;
        if (setenv("PARDES_9P", path, 1) != 0) break :exporting;
        if (setenv("PARDES_PANE", id, 1) != 0) break :exporting;
        if (setenv("PARDES_FORWARD_LOOK", if (forward_look) "1" else "0", 1) == 0) return;
    }
    _ = unsetenv("PARDES_9P");
    _ = unsetenv("PARDES_PANE");
    _ = setenv("PARDES_FORWARD_LOOK", "0", 1);
}

test "9P shell environment preserves identity when nested Look forwarding is disabled" {
    const names = [_][*:0]const u8{ "XDG_RUNTIME_DIR", "HOME", "PARDES_9P", "PARDES_PANE", "PARDES_FORWARD_LOOK" };
    var saved: [names.len]?[:0]u8 = @splat(null);
    for (names, &saved) |name, *value| {
        if (libc.getenv(name)) |old| value.* = try testing.allocator.dupeZ(u8, std.mem.span(old));
    }
    defer for (names, saved) |name, value| {
        if (value) |old| {
            _ = setenv(name, old, 1);
            testing.allocator.free(old);
        } else _ = unsetenv(name);
    };

    var dir: [sun_path_len:0]u8 = undefined;
    _ = setenv("XDG_RUNTIME_DIR", "/run/user/1000", 1);
    try testing.expectEqualStrings("/run/user/1000", socketDir(&dir).?);
    _ = unsetenv("XDG_RUNTIME_DIR");
    _ = setenv("HOME", "/home/example", 1);
    try testing.expectEqualStrings("/home/example/.local/state/pardes", socketDir(&dir).?);
    _ = unsetenv("HOME");
    try testing.expect(socketDir(&dir) == null);

    const listener = try testing.allocator.create(Listener);
    defer testing.allocator.destroy(listener);
    listener.* = .{};
    const path = "/tmp/pardes-example.sock";
    @memcpy(listener.path_buf[0..path.len], path);
    listener.path_len = path.len;
    exportPaneEnv(listener, 7, false);
    try testing.expectEqualStrings(path, std.mem.span(libc.getenv("PARDES_9P").?));
    try testing.expectEqualStrings("7", std.mem.span(libc.getenv("PARDES_PANE").?));
    try testing.expectEqualStrings("0", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?));
    exportPaneEnv(listener, 8, true);
    try testing.expectEqualStrings("8", std.mem.span(libc.getenv("PARDES_PANE").?));
    try testing.expectEqualStrings("1", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?));
    exportPaneEnv(null, 0, false);
    try testing.expect(libc.getenv("PARDES_9P") == null);
    try testing.expect(libc.getenv("PARDES_PANE") == null);
    try testing.expectEqualStrings("0", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?));
}

pub const Client = struct {
    pub const budget_ms: i64 = 2000;

    pub const max_depth: usize = 2 * ninep.max_welem;

    const uname = "pardes";

    const Dial = union(enum) { unix: [:0]const u8, tcp: std.Io.net.IpAddress, quic: std.Io.net.IpAddress };

    pub const Error = error{
        PathTooDeep,
        BadDial,
        Dial,
        Hangup,
        Timeout,
        Botch,
        Remote,
        IsDirectory,
        NotFound,
        FileTooLarge,
        QuicUnavailable,
    };

    pub fn read(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, display_path: []const u8) ![]u8 {
        return readLimit(gpa, dial, path, display_path, limits.max_file_bytes);
    }

    pub fn readLimit(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, display_path: []const u8, max_bytes: usize) ![]u8 {
        if (comptime !supported) return error.Unsupported;
        var names: [max_depth][]const u8 = undefined;
        const n = try elements(path, &names);
        var sock_buf: [sun_path_len]u8 = undefined;
        const sock = try resolve(&sock_buf, dial);
        var remote: RemoteError = .{};
        return fetchBytes(gpa, sock, names[0..n], &remote, null, display_path, @min(max_bytes, limits.max_file_bytes));
    }

    pub fn write(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, bytes: []const u8) !void {
        if (comptime !supported) return error.Unsupported;
        var names: [max_depth][]const u8 = undefined;
        const n = try elements(path, &names);
        var sock_buf: [sun_path_len]u8 = undefined;
        const sock = try resolve(&sock_buf, dial);
        var remote: RemoteError = .{};
        const result = try fetchBytes(gpa, sock, names[0..n], &remote, bytes, path, limits.max_file_bytes);
        gpa.free(result);
    }

    const RemoteError = struct {
        buf: [ninep.errmax]u8 = undefined,
        len: usize = 0,

        fn set(r: *RemoteError, msg: []const u8) error{Remote} {
            r.len = @min(msg.len, r.buf.len);
            @memcpy(r.buf[0..r.len], msg[0..r.len]);
            return error.Remote;
        }
    };

    fn elements(path: []const u8, out: *[max_depth][]const u8) Error!usize {
        var n: usize = 0;
        var it = std.mem.tokenizeScalar(u8, path, '/');
        while (it.next()) |name| {
            if (n == out.len) return Error.PathTooDeep;
            out[n] = name;
            n += 1;
        }
        return n;
    }

    fn resolve(buf: *[sun_path_len]u8, dial: []const u8) error{ BadDial, QuicUnavailable }!Dial {
        if (dial.len == 0) return error.BadDial;
        if (std.mem.startsWith(u8, dial, "tcp!")) return .{ .tcp = try networkAddress(dial, false) };
        if (std.mem.startsWith(u8, dial, "quic!")) {
            if (comptime !quic_enabled) return error.QuicUnavailable;
            return .{ .quic = try networkAddress(dial, false) };
        }
        const explicit_unix = std.mem.startsWith(u8, dial, "unix!");
        const path = if (explicit_unix) dial[5..] else dial;
        if (explicit_unix and !std.mem.startsWith(u8, path, "/")) return error.BadDial;
        if (std.mem.indexOfScalar(u8, path, '/') != null) {
            if (std.mem.indexOfScalar(u8, path, 0) != null) return error.BadDial;
            return .{ .unix = std.fmt.bufPrintSentinel(buf, "{s}", .{path}, 0) catch return error.BadDial };
        }
        var dir_buf: [sun_path_len:0]u8 = undefined;
        const dir = socketDir(&dir_buf) orelse return error.BadDial;
        return .{ .unix = socketPath(buf, dir, dial) orelse return error.BadDial };
    }

    pub fn validateDial(dial: []const u8) error{ BadDial, QuicUnavailable }!void {
        var buf: [sun_path_len]u8 = undefined;
        _ = try resolve(&buf, dial);
    }

    pub fn sameSession(dial: []const u8, socket_path: []const u8, tcp_address: ?std.Io.net.IpAddress, quic_address: ?std.Io.net.IpAddress) bool {
        if (comptime !supported) return false;
        var buf: [sun_path_len]u8 = undefined;
        const address = resolve(&buf, dial) catch return false;
        switch (address) {
            .unix => |path| return socket_path.len != 0 and std.mem.eql(u8, path, socket_path),
            .tcp, .quic => |destination| {
                var ip = destination;
                switch (ip) {
                    .ip4 => |v4| if (std.mem.allEqual(u8, &v4.bytes, 0)) {
                        ip = .{ .ip4 = .loopback(v4.port) };
                    },
                    .ip6 => |v6| if (std.mem.allEqual(u8, &v6.bytes, 0)) {
                        ip = .{ .ip6 = .loopback(v6.port) };
                    },
                }
                const bound = canonicalIp((if (address == .tcp) tcp_address else quic_address) orelse return false);
                if (ip.getPort() != bound.getPort() or @as(std.Io.net.IpAddress.Family, ip) != @as(std.Io.net.IpAddress.Family, bound)) return false;
                if (ip.eql(&bound)) return true;
                const wildcard = switch (bound) {
                    .ip4 => |v4| std.mem.allEqual(u8, &v4.bytes, 0),
                    .ip6 => |v6| std.mem.allEqual(u8, &v6.bytes, 0),
                };
                if (wildcard) return localIp(ip);
                return false;
            },
        }
    }

    const Session = struct {
        fd: c_int,
        quic: if (quic_enabled) ?quic.Connection else void = if (quic_enabled) null else {},
        deadline: i64,
        display_path: []const u8,
        cl: ninep.Client = undefined,
        in: [msize]u8 = undefined,
        out: [msize]u8 = undefined,
        stage: [msize]u8 = undefined,

        fn wait(s: *Session, events: i16) Error!void {
            while (true) {
                const left = s.deadline - nowMs();
                if (left <= 0) return Error.Timeout;
                if (comptime quic_enabled) {
                    if (s.quic) |*connection| {
                        var fds = [1]libc.pollfd{connection.poll().?};
                        const timeout = @min(left, connection.nextDue() orelse budget_ms);
                        const ready = libc.poll(&fds, 1, @intCast(timeout));
                        if (ready < 0) {
                            if (libc.errno(ready) == .INTR) continue;
                            return Error.Hangup;
                        }
                        if (nowMs() >= s.deadline) return Error.Timeout;
                        if (fds[0].revents & @as(i16, @intCast(libc.POLL.NVAL)) != 0) return Error.Hangup;
                        connection.events() catch return Error.Hangup;
                        return;
                    }
                }
                return transport.wait(s.fd, events, s.deadline) catch |err| switch (err) {
                    error.Timeout => Error.Timeout,
                    else => Error.Hangup,
                };
            }
        }

        fn flush(s: *Session) Error!void {
            while (s.cl.output().len != 0) {
                if (nowMs() >= s.deadline) return Error.Timeout;
                const bytes = s.cl.output();
                if (comptime quic_enabled) {
                    if (s.quic) |*connection| {
                        const sent = connection.write(bytes) catch return Error.Hangup;
                        if (sent == 0) {
                            try s.wait(poll_out);
                            continue;
                        }
                        s.cl.wrote(sent);
                        continue;
                    }
                }
                try s.wait(poll_out);
                const sent = (transport.write(s.fd, bytes) catch return Error.Hangup) orelse continue;
                s.cl.wrote(@intCast(sent));
            }
        }

        fn settle(s: *Session) Error!ninep.Client.Done {
            while (true) {
                if (nowMs() >= s.deadline) return Error.Timeout;
                try s.flush();
                if (s.cl.take()) |done| return done;
                if (s.cl.dead) return Error.Botch;
                const room = s.cl.in.len - s.cl.in_len;
                if (room == 0) return Error.Botch;
                if (comptime quic_enabled) {
                    if (s.quic) |*connection| {
                        const got = (connection.read(s.stage[0..@min(room, s.stage.len)]) catch return Error.Hangup) orelse {
                            try s.wait(poll_in);
                            continue;
                        };
                        if (got == 0) return Error.Hangup;
                        const n = s.cl.push(s.stage[0..got]);
                        std.debug.assert(n == got);
                        continue;
                    }
                }
                try s.wait(poll_in);
                const got = (transport.read(s.fd, s.stage[0..@min(room, s.stage.len)]) catch return Error.Hangup) orelse continue;
                if (got == 0) return Error.Hangup;
                const n = s.cl.push(s.stage[0..@intCast(got)]);
                std.debug.assert(n == @as(usize, @intCast(got)));
            }
        }

        fn ask(s: *Session, req: ninep.Client.Request, remote: *RemoteError) Error!ninep.Client.Result {
            _ = s.cl.submit(req) catch return Error.Botch;
            const done = try s.settle();
            if (done.result == .fail) return remote.set(done.result.fail);
            if (std.mem.eql(u8, @tagName(done.result), @tagName(std.meta.activeTag(req)))) return done.result;
            return Error.Botch;
        }

        fn drop(s: *Session, fid: u32) void {
            _ = s.cl.submit(.{ .clunk = .{ .fid = fid } }) catch return;
            _ = s.settle() catch {};
        }

        fn dropNoWait(s: *Session, fid: u32) void {
            _ = s.cl.submit(.{ .clunk = .{ .fid = fid } }) catch return;
            s.flush() catch {};
        }
    };

    fn transact(
        s: *Session,
        names: []const []const u8,
        out: *std.Io.Writer.Allocating,
        remote: *RemoteError,
        write_bytes: ?[]const u8,
        read_limit: usize,
    ) !void {
        _ = try s.ask(.{ .version = .{} }, remote);
        if (s.cl.msize == 0) return Error.Botch;

        const root: u32 = 0;
        var here = (try s.ask(.{ .attach = .{ .fid = root, .uname = uname } }, remote)).attach;
        var cur: u32 = root;
        var next: u32 = 1;

        var i: usize = 0;
        while (i < names.len) {
            const n = @min(ninep.max_welem, names.len - i);
            const w = (try s.ask(.{ .walk = .{
                .fid = cur,
                .newfid = next,
                .names = names[i..][0..n],
            } }, remote)).walk;
            if (w.nwqid != n) return Error.NotFound;
            here = w.wqid[n - 1];
            if (cur != root) s.drop(cur);
            cur = next;
            next = if (next == 1) 2 else 1;
            i += n;
        }
        defer s.dropNoWait(cur);

        const directory = here.type & ninep.qtdir != 0;

        const mode: u8 = if (write_bytes != null) ninep.owrite else ninep.oread;
        const truncate = write_bytes != null and names.len > 0 and
            (std.mem.eql(u8, names[0], "os") or std.mem.eql(u8, names[names.len - 1], "body"));
        _ = try s.ask(.{ .open = .{ .fid = cur, .mode = mode | if (truncate) ninep.otrunc else 0 } }, remote);
        if (write_bytes) |bytes| {
            if (directory) return Error.IsDirectory;
            var written: usize = 0;
            while (written < bytes.len) {
                const chunk = bytes[written..][0..@min(bytes.len - written, s.cl.maxWrite())];
                const count = (try s.ask(.{ .write = .{ .fid = cur, .offset = written, .data = chunk } }, remote)).write;
                if (count == 0 or count > chunk.len) return Error.Botch;
                written += count;
            }
            return;
        }

        const max_bytes: u64 = @min(read_limit, @as(usize, if (directory) limits.max_stream_bytes else limits.max_file_bytes));
        const wire_limit: u64 = if (directory) limits.max_stream_bytes else max_bytes;
        var off: u64 = 0;
        while (true) {
            const want: u32 = @intCast(@min(@as(u64, s.cl.maxRead()), wire_limit + 1 - off));
            const data = (try s.ask(.{ .read = .{ .fid = cur, .offset = off, .count = want } }, remote)).read;
            if (data.len == 0) return;
            if (off + data.len > wire_limit) return Error.FileTooLarge;
            if (directory) {
                var pos: usize = 0;
                while (pos < data.len) {
                    if (data.len - pos < 2) return Error.Botch;
                    const len: usize = 2 + @as(usize, std.mem.readInt(u16, data[pos..][0..2], .little));
                    if (len > data.len - pos) return Error.Botch;
                    const entry = ninep.Stat.decode(data[pos..][0..len]) catch return Error.Botch;
                    const display_dir = std.mem.trimEnd(u8, s.display_path, "/");
                    const row_len = display_dir.len + entry.name.len + 2 + @as(usize, @intFromBool(entry.qid.type & ninep.qtdir != 0));
                    if (row_len > max_bytes - out.written().len) return Error.FileTooLarge;
                    try out.writer.print("{s}/{s}", .{ display_dir, entry.name });
                    if (entry.qid.type & ninep.qtdir != 0) try out.writer.writeByte('/');
                    try out.writer.writeByte('\n');
                    pos += len;
                }
            } else try out.writer.writeAll(data);
            off += data.len;
        }
    }

    fn fetchBytes(
        gpa: std.mem.Allocator,
        sock: Dial,
        names: []const []const u8,
        remote: *RemoteError,
        write_bytes: ?[]const u8,
        display_path: []const u8,
        read_limit: usize,
    ) ![]u8 {
        if (comptime !supported) return Error.Dial;
        const deadline = nowMs() +| budget_ms;
        const s = try gpa.create(Session);
        s.* = .{ .fd = -1, .deadline = deadline, .display_path = display_path };
        defer {
            if (quic_enabled and s.quic != null) {
                s.quic.?.deinit();
            } else if (s.fd >= 0) _ = libc.close(s.fd);
            gpa.destroy(s);
        }
        if (sock == .quic) {
            if (comptime quic_enabled) {
                s.quic = quic.Connection.dial(sock.quic) catch return Error.Dial;
                s.fd = s.quic.?.fd;
            } else return Error.QuicUnavailable;
        } else s.fd = try connect(sock, deadline);
        s.cl = .init(.{ .in = &s.in, .out = &s.out });

        var out: std.Io.Writer.Allocating = .init(gpa);
        errdefer out.deinit();
        try transact(s, names, &out, remote, write_bytes, read_limit);
        return out.toOwnedSlice();
    }

    fn connect(sock: Dial, deadline: i64) Error!c_int {
        const address: transport.Address = switch (sock) {
            .unix => |path| .{ .unix = path },
            .tcp => |ip| .{ .tcp = ip },
            .quic => return Error.QuicUnavailable,
        };
        return transport.connectFd(address, deadline) catch return Error.Dial;
    }

    const nowMs = transport.nowMs;

    const poll_in: i16 = @intCast(libc.POLL.IN);
    const poll_out: i16 = @intCast(libc.POLL.OUT);

    fn nap(ms: c_int) void {
        _ = libc.poll(&[0]libc.pollfd{}, 0, ms);
    }

    test "a path becomes walk elements, normalised the way a shell would" {
        var out: [max_depth][]const u8 = undefined;
        try testing.expectEqual(@as(usize, 4), try elements("/self/pane/1/body", &out));
        try testing.expectEqualStrings("self", out[0]);
        try testing.expectEqualStrings("pane", out[1]);
        try testing.expectEqualStrings("1", out[2]);
        try testing.expectEqualStrings("body", out[3]);

        try testing.expectEqual(@as(usize, 4), try elements("self/pane/1/body", &out));
        try testing.expectEqual(@as(usize, 4), try elements("//self//pane//1//body//", &out));
        try testing.expectEqual(@as(usize, 2), try elements("/self/index", &out));

        try testing.expectEqual(@as(usize, 0), try elements("/", &out));

        var deep: [8 * max_depth]u8 = @splat('/');
        for (0..max_depth + 1) |i| deep[i * 2 + 1] = 'a';
        try testing.expectError(Error.PathTooDeep, elements(deep[0 .. (max_depth + 1) * 2], &out));
    }

    test "a bare dial resolves to the socket --9p binds, and a path is taken as given" {
        if (comptime !supported) return error.SkipZigTest;
        var buf: [sun_path_len]u8 = undefined;

        const named = (try resolve(&buf, "work")).unix;
        try testing.expect(std.mem.endsWith(u8, named, "/pardes-9p-work.sock"));
        var expect: [sun_path_len]u8 = undefined;
        var dir_buf: [sun_path_len:0]u8 = undefined;
        const dir = socketDir(&dir_buf).?;
        try testing.expectEqualStrings(socketPath(&expect, dir, "work").?, named);

        const path = (try resolve(&buf, "/tmp/somewhere.sock")).unix;
        try testing.expectEqualStrings("/tmp/somewhere.sock", path);
        try testing.expectEqualStrings("/tmp/somewhere.sock", (try resolve(&buf, "unix!/tmp/somewhere.sock")).unix);

        try testing.expectError(error.BadDial, resolve(&buf, ""));
        try testing.expectError(error.BadDial, resolve(&buf, "/tmp/a\x00b"));
    }

    test "a dial with nothing listening is one error and not a wait" {
        if (comptime !supported) return error.SkipZigTest;
        var names: [max_depth][]const u8 = undefined;
        const n = try elements("/self/pane/1/body", &names);
        var remote: RemoteError = .{};
        const before = nowMs();
        try testing.expectError(
            Error.Dial,
            fetchBytes(testing.allocator, .{ .unix = "/tmp/pardes-9p-no-such-socket.sock" }, names[0..n], &remote, null, "/self/pane/1/body", limits.max_file_bytes),
        );
        try testing.expect(nowMs() - before < budget_ms);
    }

    test "one fetch has three msize buffers and bounded transport metadata" {
        const transport_bytes = if (quic_enabled) @sizeOf(?quic.Connection) else 0;
        try testing.expectEqual(@as(usize, msize), @as(usize, (Session{ .fd = -1, .deadline = 0, .display_path = "" }).in.len));
        try testing.expect(transport_bytes <= 64);
        try testing.expect(@sizeOf(Session) <= 3 * msize + @sizeOf(ninep.Client) + 128 + transport_bytes);
        try testing.expect(@sizeOf(ninep.Client) <= 512);
    }

    test "expired sessions do not send or consume buffered protocol work" {
        var session: Session = .{ .fd = -1, .deadline = 0, .display_path = "" };
        session.cl = .init(.{ .in = &session.in, .out = &session.out });
        _ = try session.cl.submit(.{ .version = .{} });
        const queued = session.cl.output().len;
        try testing.expect(queued > 0);
        try testing.expectError(Error.Timeout, session.flush());
        try testing.expectEqual(queued, session.cl.output().len);
        try testing.expectError(Error.Timeout, session.settle());
        try testing.expectError(Error.Timeout, session.wait(poll_in));
    }
};