1 files changed, 194 insertions, 0 deletions
diff --git a/src/Server.zig b/src/Server.zig
new file mode 100644
index 0000000..a652f99
--- /dev/null
+++ b/src/Server.zig
@@ -0,0 +1,194 @@
+//! A bounded, caller-driven 9P2000 server connection.
+//! The backend owns fids, authentication, permissions, and filesystem operations.
+//! receive() borrows one input frame until release(). Pending operations may outlive
+//! that frame only if the backend copies their strings/data. reply() copies output.
+const std = @import("std");
+const assert = std.debug.assert;
+const wire = @import("wire.zig");
+const Server = @This();
+
+in: []u8,
+out: []u8,
+in_len: usize = 0,
+frame: u32 = 0,
+out_len: usize = 0,
+out_off: usize = 0,
+msize: u32 = 0,
+dead: bool = false,
+pending: [65]Pending = @splat(.{}),
+versioning: bool = false,
+
+const Pending = struct {
+    request: ?wire.Type = null,
+    tag: u16 = 0,
+    count: u32 = 0,
+    oldtag: u16 = wire.notag,
+};
+
+pub const Error = wire.Error || error{ Protocol, NoTags, UnknownTag, WrongReply, TooLarge };
+pub const Options = struct { in: []u8, out: []u8 };
+/// Local resource floor, not a minimum imposed by the 9P specification.
+pub const msize_min: u32 = 24;
+
+pub fn init(options: Options) Server {
+    assert(options.in.len >= msize_min);
+    assert(options.out.len >= msize_min);
+    return .{ .in = options.in, .out = options.out };
+}
+
+pub fn push(s: *Server, bytes: []const u8) usize {
+    if (s.dead) return 0;
+    const n = @min(bytes.len, s.in.len - s.in_len);
+    @memcpy(s.in[s.in_len..][0..n], bytes[0..n]);
+    s.in_len += n;
+    return n;
+}
+
+pub fn output(s: *const Server) []const u8 {
+    return s.out[s.out_off..s.out_len];
+}
+
+pub fn wrote(s: *Server, n: usize) void {
+    assert(n <= s.output().len);
+    s.out_off += n;
+    if (s.out_off == s.out_len) {
+        s.out_off = 0;
+        s.out_len = 0;
+    }
+}
+
+pub fn hasRoom(s: *Server) bool {
+    if (s.out_off != 0) {
+        const n = s.output().len;
+        std.mem.copyForwards(u8, s.out[0..n], s.output());
+        s.out_off = 0;
+        s.out_len = n;
+    }
+    return s.out.len - s.out_len >= @max(s.msize, msize_min);
+}
+
+/// A null result means more input or output drainage is needed.
+/// Any receive error is terminal: the stream cannot be safely resynchronized.
+pub fn receive(s: *Server) Error!?wire.Decoded {
+    assert(s.frame == 0);
+    if (s.dead) return null;
+    errdefer s.dead = true;
+    const len = wire.frameLen(s.in[0..s.in_len]) orelse return null;
+    if (len < wire.header_len or len > s.in.len) return error.Protocol;
+    if (len > s.in_len) return null;
+    const got = try wire.decode(s.in[0..len]);
+    const kind = got.msg.msgType();
+    if (!wire.isT(kind)) return error.Protocol;
+    if (!s.hasRoom()) return null;
+    if (kind == .tversion) {
+        if (got.tag != wire.notag) return error.Protocol;
+        // Finish any partially transmitted response before starting a new session.
+        if (s.output().len != 0) return null;
+        s.pending = @splat(.{});
+        s.versioning = true;
+        s.msize = 0;
+    } else {
+        if (s.msize == 0 or s.versioning) return error.Protocol;
+        if (len > s.msize or got.tag == wire.notag) return error.Protocol;
+        if (s.find(got.tag) != null) return error.Protocol;
+        const slot = s.free(kind) orelse return error.NoTags;
+        slot.* = .{ .request = kind, .tag = got.tag, .count = switch (got.msg) {
+            .tread => |m| m.count,
+            .twrite => |m| @intCast(m.data.len),
+            .twalk => |m| m.nwname,
+            else => 0,
+        }, .oldtag = if (got.msg == .tflush) got.msg.tflush.oldtag else wire.notag };
+    }
+    s.frame = len;
+    return got;
+}
+
+pub fn release(s: *Server) void {
+    assert(s.frame != 0 and s.frame <= s.in_len);
+    const n = s.in_len - s.frame;
+    std.mem.copyForwards(u8, s.in[0..n], s.in[s.frame..s.in_len]);
+    s.in_len = n;
+    s.frame = 0;
+}
+
+/// Call only for a received Tversion, after aborting backend work and releasing fids.
+/// Suffixes may fall back to base 9P2000; arbitrary strings beginning with 9P may not.
+pub fn negotiate(s: *Server, want: u32, version: []const u8) Error!void {
+    assert(s.versioning);
+    const size: u32 = @intCast(@min(want, s.in.len, s.out.len, std.math.maxInt(u32)));
+    if (size < msize_min) return error.TooLarge;
+    const base = std.mem.sliceTo(version, '.');
+    const known = std.mem.eql(u8, base, "9P2000");
+    try s.append(wire.notag, .{ .rversion = .{
+        .msize = size,
+        .version = if (known) "9P2000" else "unknown",
+    } }, size);
+    s.msize = if (known) size else 0;
+    s.versioning = false;
+}
+
+/// An Rflush is a backend promise: no further response for oldtag will be sent.
+/// A canceled backend operation must be retired before its tag can be reused.
+pub fn reply(s: *Server, tag: u16, msg: wire.Msg) Error!void {
+    if (s.dead) return error.Protocol;
+    const slot = s.find(tag) orelse return error.UnknownTag;
+    const request = slot.request.?;
+    const kind = msg.msgType();
+    if (request == .tflush and kind != .rflush) return error.WrongReply;
+    if (kind != .rerror and @intFromEnum(kind) != @intFromEnum(request) + 1)
+        return error.WrongReply;
+    switch (msg) {
+        .rread => |m| if (m.data.len > slot.count) return error.WrongReply,
+        .rwrite => |m| if (m.count > slot.count) return error.WrongReply,
+        .rwalk => |m| {
+            if (m.nwqid > slot.count) return error.WrongReply;
+            if (m.nwqid == 0 and slot.count != 0) return error.WrongReply;
+        },
+        else => {},
+    }
+    var response = msg;
+    if (response == .rerror) {
+        const cap = @min(s.msize - wire.header_len - 2, std.math.maxInt(u16));
+        response.rerror.ename = response.rerror.ename[0..@min(response.rerror.ename.len, cap)];
+    }
+    try s.append(tag, response, s.msize);
+    const oldtag = slot.oldtag;
+    slot.* = .{};
+    if (kind == .rflush) {
+        if (s.find(oldtag)) |old| old.* = .{};
+    }
+}
+
+fn append(s: *Server, tag: u16, msg: wire.Msg, limit: u32) Error!void {
+    if (try wire.encodedLen(msg) > limit) return error.TooLarge;
+    _ = s.hasRoom();
+    const bytes = try wire.encode(msg, tag, s.out[s.out_len..]);
+    s.out_len += bytes.len;
+}
+
+fn find(s: *Server, tag: u16) ?*Pending {
+    for (&s.pending) |*slot| {
+        if (slot.request != null and slot.tag == tag) return slot;
+    }
+    return null;
+}
+
+fn free(s: *Server, request: wire.Type) ?*Pending {
+    for (&s.pending, 0..) |*slot, i| {
+        // A full ordinary request window must still admit cancellation.
+        if (i == s.pending.len - 1 and request != .tflush) continue;
+        if (slot.request == null) return slot;
+    }
+    return null;
+}
+
+pub fn hangup(s: *Server) void {
+    s.dead = true;
+    s.pending = @splat(.{});
+    s.in_len = 0;
+    s.frame = 0;
+    s.out_len = 0;
+    s.out_off = 0;
+    s.msize = 0;
+    s.versioning = false;
+}