A log follower hears every record queued before a Restore: the read it holds takes all that fit, before the hangup cuts it

A follower reads one record per read, and a Restore hangs every client
up, so records queued just before it -- the ones a script most wants,
what led up to the Restore -- were lost: the read after the first was
cut. Before the cut, each held read of /log is answered with every
whole record queued that fits it.

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

3 files changed, 42 insertions, 1 deletions

diff --git a/src/9p_io.zig b/src/9p_io.zig
index d80d7b30..b11a4997 100644
--- a/src/9p_io.zig
+++ b/src/9p_io.zig
@@ -566,6 +566,12 @@ pub const Listener = struct {
     /// answer nothing, their connections being hung up. The connections pay their releases on their own tasks, so
     /// the editor rests while they do.
     pub fn reset(l: *Listener, replacement: *pardes.Pardes) void {
+        // A follower of the log hears every record queued before the cut, in
+        // the answer to the read it holds; the read after it is cut.
+        l.core.fs.hanging_up = true;
+        l.core.fs.news = true;
+        answerHeld(l);
+        l.core.fs.hanging_up = false;
         for (0..quic_slots) |i| l.drop(@intCast(i));
         if (comptime quic_enabled) while (l.tick().pending) {};
         l.core = replacement;
@@ -1831,6 +1837,27 @@ test "a held read is answered when the log has news, and a flushed one spends no
     s.drop(1);
     s.drop(2);
     try heldNow.count(p, 0);
+
+    // Records queued as a Restore comes reach a follower in the answer to
+    // the read it holds, all of them, before the hangup cuts it. (The
+    // interrupted read above answered first: one clunk's answer is unread.)
+    _ = try s.settle();
+    _ = try s.ask(.{ .walk = .{ .fid = 0, .newfid = 3, .names = &.{"log"} } }, &remote);
+    _ = try s.ask(.{ .open = .{ .fid = 3, .mode = ninep.ordwr } }, &remote);
+    _ = try s.ask(.{ .write = .{ .fid = 3, .offset = 0, .data = "follow new\n" } }, &remote);
+    const last = try s.cl.submit(.{ .read = .{ .fid = 3, .offset = 0, .count = 4096 } });
+    try s.flush();
+    try heldNow.count(p, 1);
+    pardes.turn.wake();
+    p.setMessage(0, "queued one");
+    p.setMessage(0, "queued two");
+    p.setMessage(0, "queued three");
+    l.reset(p);
+    pardes.turn.rest();
+    const cut = try s.settle();
+    try testing.expectEqual(last, cut.tag);
+    for ([_][]const u8{ "queued one", "queued two", "queued three" }) |said|
+        try testing.expect(std.mem.indexOf(u8, cut.result.read, said) != null);
 }
 
 extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
diff --git a/src/fs.zig b/src/fs.zig
index 91edd429..965eb47c 100644
--- a/src/fs.zig
+++ b/src/fs.zig
@@ -1506,6 +1506,9 @@ pub const Namespace = struct {
     /// Something a held read may be waiting on changed since they were last
     /// answered: a record queued, a run answered, a pane gone.
     news: bool = false,
+    /// A Restore is about to hang every client up: a follower's held read
+    /// of /log takes every queued record that fits, not one (events.readLog).
+    hanging_up: bool = false,
     /// The editor-wide event ring: panes made, renamed, saved and closed, and
     /// what the editor said. Recorded whether or not anyone reads /log.
     log: tree.events.Queue = .{ .cap = limits.log_bytes },
diff --git a/src/ninep/events.zig b/src/ninep/events.zig
index 8fae6eab..d60c07ad 100644
--- a/src/ninep/events.zig
+++ b/src/ninep/events.zig
@@ -492,7 +492,18 @@ pub fn readLog(p: *Pardes, req: Req) Reply {
                 slot.next += 1;
                 slot.part = 0;
             }
-            return .{ .tag = req.tag, .payload = .{ .staged = @intCast(n) } };
+            // This is the last read a Restore lets through: it takes every
+            // whole record that fits, or they were lost with the hangup.
+            if (p.fs.hanging_up and slot.part == 0) {
+                at += 4 + len;
+                while (at + 4 <= q.buf.items.len) : (slot.next += 1) {
+                    const more = std.mem.readInt(u32, q.buf.items[at..][0..4], .little);
+                    if (out.items.len + more > req.size) break;
+                    out.appendSlice(p.gpa, q.buf.items[at + 4 ..][0..more]) catch break;
+                    at += 4 + more;
+                }
+            }
+            return .{ .tag = req.tag, .payload = .{ .staged = @intCast(out.items.len) } };
         }
         at += 4 + len;
     }