Hold a read that has nothing yet and answer it when its file has news

A following log, event, pty/data or a pty/run before its answer used to
answer .again and wait for a wakeAll, which only the parked-write path
asked for, so the band-aid had every queue push set turn.parked. Now the
core keeps such a read (ctlfs.hold) and, as the turn is given up after
anything that queued a record, ran a command out or closed a pane,
answers it on its own connection, the way factotum answers the log reads
it keeps and acme an event read. Only a read the engine still holds
parked is answered, because cloud9 tells the backend nothing of a
Tflush, so a flushed read spends no record.

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

3 files changed, 45 insertions, 14 deletions

diff --git a/src/ninep/events.zig b/src/ninep/events.zig
index bd5fecf3..6b8c744e 100644
--- a/src/ninep/events.zig
+++ b/src/ninep/events.zig
@@ -50,10 +50,6 @@ pub const Queue = struct {
             q.buf.shrinkRetainingCapacity(q.buf.items.len - 4);
             return;
         };
-        // A read parked on this queue is retried only when the turn is next
-        // given up with `parked` set; without it a follower sleeps until some
-        // unrelated request happens to park.
-        pardes.turn.parked = true;
     }
 
     pub fn peek(q: *const Queue) ?[]const u8 {
@@ -105,7 +101,7 @@ pub fn pending(q: *const Queue) u64 {
     return record.len;
 }
 
-/// One record per read; `.again` parks the read until a record arrives.
+/// One record per read; `.again` holds the read until a record arrives.
 pub fn readQueue(p: *Pardes, req: Req, q: *Queue) Reply {
     const record = q.peek() orelse return .{ .tag = req.tag, .status = .again };
     if (req.size < record.len) return Reply.fail(req.tag, E.INVAL);
@@ -129,6 +125,7 @@ pub fn noteInstall(p: *Pardes, id: usize) void {
 pub fn noteRetire(p: *Pardes, id: usize, pane: *Pane) void {
     if (id >= MAX_PANES) return;
     tree.pty.shellGone(p, id);
+    p.fs.news = true; // a read held on its event or pty/data hears it went
     if (p.fs.panes[id].unannounced) {
         p.fs.panes[id].unannounced = false;
         return;
@@ -174,6 +171,7 @@ fn pushLog(p: *Pardes, record: []u8) void {
     while (end > 0 and record[end] & 0xC0 == 0x80) end -= 1;
     record[end] = '\n';
     p.fs.log.push(p.gpa, record[0 .. end + 1]);
+    p.fs.news = true;
 }
 
 /// An open freezes the ring's text, so `cat log` answers what happened lately
@@ -384,6 +382,7 @@ pub fn noteAction(
     var buf: [max_record_text + 64]u8 = undefined;
     const record = formatRecord(&buf, p.fs.origin, action, q0, q1, flag, text);
     p.fs.panes[id].events.push(p.gpa, record);
+    p.fs.news = true;
     return true;
 }
 
@@ -397,6 +396,7 @@ pub fn notePtyOutput(p: *Pardes, id: usize, bytes: []const u8) void {
         pf.pty_out.push(p.gpa, bytes[off..][0..n]);
         off += n;
     }
+    p.fs.news = true;
 }
 
 const EventRecord = struct { action: Action, q0: u32, q1: u32 };
diff --git a/src/ninep/pty.zig b/src/ninep/pty.zig
index a689d87f..2a38f1f0 100644
--- a/src/ninep/pty.zig
+++ b/src/ninep/pty.zig
@@ -162,10 +162,10 @@ pub fn openRun(p: *Pardes, req: Req, serial: u32) Reply {
     return Reply.fail(req.tag, E.NFILE);
 }
 
-fn answer(slot: *Run, comptime fmt: []const u8, args: anytype) void {
+fn answer(p: *Pardes, slot: *Run, comptime fmt: []const u8, args: anytype) void {
     slot.len = @intCast((std.fmt.bufPrint(&slot.answer, fmt ++ "\n", args) catch unreachable).len);
     slot.phase = .done;
-    pardes.turn.parked = true; // wake the read waiting on it
+    p.fs.news = true; // the read held on it can be answered
 }
 
 pub fn writeRun(p: *Pardes, req: Req, id: usize, pane: *Pane) Reply {
@@ -180,7 +180,7 @@ pub fn writeRun(p: *Pardes, req: Req, id: usize, pane: *Pane) Reply {
     const state = pane.terminal orelse return tree.failText(req.tag, E.INVAL, e_bad_line);
     const marks = &state.stream.handler;
     if (pf.unmarked) {
-        answer(slot, "error no prompt marks", .{});
+        answer(p, slot, "error no prompt marks", .{});
     } else if (pf.run != null or marks.phase != .input or
         !pardes.panes.Terminal.promptInputEmpty(pane) or p.hostTtyTaken(id))
     {
@@ -189,7 +189,7 @@ pub fn writeRun(p: *Pardes, req: Req, id: usize, pane: *Pane) Reply {
         // the middle of it. The phase is pardes's own marks: a nested
         // shell's prompt (ssh, a shell with its own integration) looks like
         // an empty prompt to ghostty but is not the shell this run knows.
-        answer(slot, "busy", .{});
+        answer(p, slot, "busy", .{});
     } else {
         slot.want = marks.started +% 1;
         slot.prompts = marks.prompts;
@@ -285,7 +285,7 @@ pub fn noteMarks(p: *Pardes, id: usize, pane: *Pane) void {
             .kept_line => {},
         }
         if (!empty) return;
-        answer(slot, "error not run", .{});
+        answer(p, slot, "error not run", .{});
         pf.run = null;
         return;
     }
@@ -305,11 +305,11 @@ pub fn noteMarks(p: *Pardes, id: usize, pane: *Pane) void {
     // The header is the whole first line, so a count there can never be
     // mistaken for output; `cut` with no count: its start scrolled away.
     if (printed == null or scrolled_out)
-        answer(slot, "exit {d} cut", .{status})
+        answer(p, slot, "exit {d} cut", .{status})
     else if (cut > 0)
-        answer(slot, "exit {d} cut {d}", .{ status, cut })
+        answer(p, slot, "exit {d} cut {d}", .{ status, cut })
     else
-        answer(slot, "exit {d}", .{status});
+        answer(p, slot, "exit {d}", .{status});
     if (keep.len > 0) {
         const nl = @intFromBool(keep[keep.len - 1] != '\n');
         if (p.gpa.alloc(u8, keep.len + nl)) |owned| {
@@ -326,7 +326,7 @@ pub fn noteMarks(p: *Pardes, id: usize, pane: *Pane) void {
 pub fn shellGone(p: *Pardes, id: usize) void {
     const pf = &p.fs.panes[id];
     const idx = pf.run orelse return;
-    answer(&p.fs.runs[idx], "error shell gone", .{});
+    answer(p, &p.fs.runs[idx], "error shell gone", .{});
     pf.run = null;
 }
 
diff --git a/src/ninep/tree.zig b/src/ninep/tree.zig
index e8bcad2f..3ebea669 100644
--- a/src/ninep/tree.zig
+++ b/src/ninep/tree.zig
@@ -107,6 +107,32 @@ pub fn changesPane(req: Req) bool {
 
 pub const out_reserve = 4 * 1024;
 
+/// A read that found nothing yet (`.again`), kept to be answered when what it
+/// waits on has something, the way factotum keeps its log's waiting reads
+/// and answers them on append (security/auth/factotum/log.c:4-52) and acme
+/// an event read (editors/acme/xfid.c:994). `asker` is the connection it
+/// came on, which only the listener knows (src/9p_io.zig, `answerHeld`).
+pub const Held = struct { asker: *anyopaque, req: Req };
+
+/// Keeps a read that answered `.again`. An open waits with one read at a
+/// time, as acme's window keeps one `eventx`: a newer read on the same open
+/// takes the place of the older.
+pub fn hold(p: *Pardes, req: Req, asker: *anyopaque) void {
+    var free: ?*?Held = null;
+    for (&p.fs.held) |*slot| {
+        const held = slot.* orelse {
+            if (free == null) free = slot;
+            continue;
+        };
+        if (held.req.node == req.node and held.req.handle == req.handle) {
+            slot.* = .{ .asker = asker, .req = req };
+            return;
+        }
+    }
+    // One slot per open that can wait, so a free one is always there.
+    if (free) |slot| slot.* = .{ .asker = asker, .req = req };
+}
+
 // ---- nodes ----
 
 pub const TopFile = enum(u4) {
@@ -559,6 +585,11 @@ fn remove(p: *Pardes, req: Req) Reply {
 }
 
 fn release(p: *Pardes, req: Req) Reply {
+    // A read held on this open goes with it: a hangup pays its releases
+    // before the connection's slot is reused, so none outlives its asker.
+    for (&p.fs.held) |*slot| if (slot.*) |held| {
+        if (held.req.node == req.node and held.req.handle == req.handle) slot.* = null;
+    };
     // The handle's bookkeeping runs whether or not the removal is allowed.
     const done = releaseHandle(p, req);
     return if (req.remove) remove(p, req) else done;