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, 16 insertions, 4 deletions

diff --git a/test/fs.py b/test/fs.py
index 48104e3f..1332c319 100644
--- a/test/fs.py
+++ b/test/fs.py
@@ -142,7 +142,7 @@ def walk_tree(client, path='/', skip=('/log', '/os', '/screen')):
 def discovery(binary, embedded=False):
     with tempfile.TemporaryDirectory(prefix='pardes-discovery-') as directory:
         root = Path(directory)
-        with session(binary, root, 'discovery') as (client, _):
+        with session(binary, root, 'discovery') as (client, address):
             top = client.list('/')
             assert top[:10] == ['README', 'index', 'status', 'look', 'exec', 'log', 'screen',
                                 'listeners', 'pane', 'os'], top
@@ -197,14 +197,26 @@ def discovery(binary, embedded=False):
             assert 'new' in client.list('/pane'), 'new should be visible to ls'
             client.stat('/pane/new')
             assert newest(client) == second, 'a stat of new made a pane'
-            # What the editor said is in the stream too, and the fixture's own
-            # `new` lands on whichever side of the open its first update did.
+            # What the editor said is in the stream too.
             def pane_event():
-                while (record := client.read_fid(log, frozen)).startswith((b'msg ', f'new {fixture} '.encode())):
+                while (record := client.read_fid(log, frozen)).startswith(b'msg '):
                     pass
                 return record
             assert pane_event() == f'new {first} {root}/+New\n'.encode()
             assert pane_event() == f'new {second} {root}/+New\n'.encode()
+            # A follower that is waiting wakes when the editor logs, not when
+            # some unrelated request next happens to wait.
+            with Client(address) as follower:
+                waiting = follower.open('/log', 2)
+                start = len(follower.read_fid(waiting))
+                follower.rpc(118, struct.pack('<IQI', waiting, 0, 7) + b'follow\n')
+                woke = []
+                reader = threading.Thread(target=lambda: woke.append(follower.read_fid(waiting, start)), daemon=True)
+                reader.start()
+                time.sleep(.2)
+                client.write('/exec', b'Msg woken\n')
+                reader.join(5)
+                assert woke and woke[0].endswith(b' woken\n'), woke
             assert set(client.list('/pane')) == {'new', str(fixture), str(first), str(second)}
             client.write(f'/pane/{first}/body', b'first pane', truncate=True)
             assert client.read(f'/pane/{first}/body') == b'first pane'