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]>

1 files changed, 27 insertions, 0 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;