review: loopd prints every startup notice before it binds; tests wait for "serving on"

The serve tests waited for the socket file, then killed `loopd` and read its stderr, but the
socket is bound before its mode is set and before "serving on" and the broker notice are printed,
so a kill in between lost those lines (`pointers` failed once under load in M5c-1 task 15). The
broker notice now comes before the bind, so "serving on" is the last startup line, and the tests
wait for it (`tests/support/serving.rs`).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

crates/loopd/tests/support/serving.rs

@@ -0,0 +1,63 @@
+//! Waiting for `loopd serve` to be up, by the line it prints last before it serves.
+//!
+//! The socket appears before `loopd` has set its mode and printed "serving on", so a test that
+//! waits for the socket and then stops `loopd` can lose those lines (M5c-1 review, finding 7).
+//! Every startup notice comes before "serving on", so once it is read, the rest has been printed.
+
+use std::io::{BufRead, BufReader, Read};
+use std::process::Child;
+use std::sync::mpsc;
+use std::time::{Duration, Instant};
+
+/// How long `loopd` may take to pass its self-test and bind, on a loaded machine.
+const UP: Duration = Duration::from_secs(60);
+
+/// Reads `child`'s stderr until the "serving on" line, stops it, and returns whether that line came
+/// and everything it printed. `child` must have been spawned with `stderr(Stdio::piped())`.
+pub fn stderr_once_serving(mut child: Child) -> (bool, String) {
+    let stderr = child.stderr.take().expect("stderr is piped");
+    let (tx, rx) = mpsc::channel::<String>();
+    let reader = std::thread::spawn(move || {
+        let mut lines = BufReader::new(stderr);
+        let mut line = String::new();
+        loop {
+            line.clear();
+            match lines.read_line(&mut line) {
+                Ok(0) | Err(_) => break,
+                Ok(_) => {
+                    if tx.send(line.clone()).is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+        let mut rest = String::new();
+        let _ = lines.read_to_string(&mut rest);
+        rest
+    });
+    let deadline = Instant::now() + UP;
+    let mut text = String::new();
+    let mut up = false;
+    while let Some(left) = deadline.checked_duration_since(Instant::now()) {
+        match rx.recv_timeout(left) {
+            Ok(line) => {
+                up = line.starts_with("loopd: serving on ");
+                text.push_str(&line);
+                if up {
+                    break;
+                }
+            }
+            Err(_) => break,
+        }
+    }
+    let _ = child.kill();
+    let _ = child.wait();
+    // The reader ends at the pipe's end once `loopd` is gone; join it before draining, so no line
+    // it sends after the drain is lost.
+    let rest = reader.join().unwrap_or_default();
+    while let Ok(line) = rx.try_recv() {
+        text.push_str(&line);
+    }
+    text.push_str(&rest);
+    (up, text)
+}