Wake reads waiting on /log, event and pty/data; one reader per consuming file; pty/ctl reads back

A read parked on a queue was retried only when some unrelated write had
to wait, so a follower of /log (and a reader of event or pty/data) slept
until then. Pushing a record now marks the turn parked, and giving the
turn up wakes them (measured: stuck past 3 s before, 0 s after).

event and pty/data consume what they read, so a second open for reading
is refused with rio's "file in use" (EBUSY through 9ns); writers still
get in, and pty/data queues output only for an actual reader. pty/ctl
reads back "winsize C R", in the words it takes.

/log fixes from review: a record longer than a read comes in pieces (a
shell read loop failed on long lines), every repeated message is logged,
a record bigger than the ring is cut to fit instead of emptying it, the
ring is reserved at boot so recording never allocates, and panes present
at boot are recorded first.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

1 files changed, 29 insertions, 4 deletions

diff --git a/src/ninep/tree.zig b/src/ninep/tree.zig
index f98429ab..a85a1873 100644
--- a/src/ninep/tree.zig
+++ b/src/ninep/tree.zig
@@ -61,6 +61,20 @@ pub fn failText(tag: u64, errno: u16, text: []const u8) Reply {
 pub const e_bad_addr = "bad address syntax";
 pub const e_bad_ctl = "ill-formed control message";
 pub const e_bad_event = "bad event syntax";
+/// A second open reading a file whose reads consume: rio's word for it
+/// (windows/rio/xfid.c:25), which 9ns turns into EBUSY.
+pub const e_in_use = "file in use";
+
+/// Handles that mark an open as the one reading `event` or `pty/data`, so the
+/// release knows to give the file up; other opens get `open_handle`.
+const open_handle: u32 = 1;
+// ponytail: EBUSY spelled locally; use E.BUSY once cloud9 names it and pardes bumps the pin.
+const e_busy: u16 = 16;
+const reader_handle: u32 = 2;
+
+fn reads(omode: u8) bool {
+    return omode & 3 != 1; // OWRITE is the one access mode that never reads
+}
 
 /// Tremove closes a pane; nothing else in the tree is created or destroyed by
 /// the protocol, and wstat stays a truncation. Making a pane is an open of
@@ -167,7 +181,7 @@ pub const PaneFile = enum(u5) {
     pub fn mode(f: PaneFile) u16 {
         return switch (f) {
             .dir, .pty => 0o755,
-            .errors, .pty_ctl => 0o222,
+            .errors => 0o222,
             .pty_status => 0o444,
             else => 0o666,
         };
@@ -507,16 +521,26 @@ fn open(p: *Pardes, req: Req, target: Target) Reply {
             if (t.file.inPty() and !pn.isTerminal()) return Reply.fail(req.tag, E.NOENT);
             switch (t.file) {
                 .body => if (pn.isTerminal()) return screen.openSnapshot(p, req, false),
+                // Reads of these consume, so two readers would each see half
+                // the stream; the second is refused rather than robbed.
                 .event => {
+                    const reader = reads(req.omode);
+                    if (reader and pf.event_reader) return failText(req.tag, e_busy, e_in_use);
                     pf.readers +|= 1;
                     p.fs.listeners +|= 1;
+                    pf.event_reader = pf.event_reader or reader;
+                    if (reader) return .{ .tag = req.tag, .handle = reader_handle };
+                },
+                .pty_data => if (reads(req.omode)) {
+                    if (pf.pty_readers > 0) return failText(req.tag, e_busy, e_in_use);
+                    pf.pty_readers = 1;
+                    return .{ .tag = req.tag, .handle = reader_handle };
                 },
-                .pty_data => pf.pty_readers +|= 1,
                 else => {},
             }
         },
     }
-    return .{ .tag = req.tag, .handle = 1 };
+    return .{ .tag = req.tag, .handle = open_handle };
 }
 
 /// Tremove closes a pane. Nothing else in the tree can be removed, and this
@@ -549,12 +573,13 @@ fn releaseHandle(p: *Pardes, req: Req) Reply {
             const id = p.paneBySerial(t.serial) orelse return .{ .tag = req.tag };
             const pf = &p.fs.panes[id];
             if (t.file == .pty_data) {
-                if (pf.pty_readers == 0) return .{ .tag = req.tag };
+                if (req.handle != reader_handle or pf.pty_readers == 0) return .{ .tag = req.tag };
                 pf.pty_readers -= 1;
                 if (pf.pty_readers == 0) pf.pty_out.clearAndFree(p.gpa);
                 return .{ .tag = req.tag };
             }
             if (pf.readers == 0) return .{ .tag = req.tag };
+            if (req.handle == reader_handle) pf.event_reader = false;
             pf.readers -= 1;
             p.fs.listeners -|= 1;
             if (pf.readers == 0) pf.tag_snap.clearAndFree(p.gpa);