fix(audio): playback worker re-checks running flag so stop() can't hang (A7)

The playback worker thread looped `while running { rx.recv() }`. A blocking
recv() never re-checks the `running` flag — it only wakes on a new frame or the
sender being dropped. So when stop() set running=false and called
worker_handle.join(), the worker stayed parked in recv() and join() hung until
the frame Sender happened to be dropped. audio_probe reproduced this every run
(it calls backend.stop() while its tx is still in scope), hanging on exit; the
GUI could hang on shutdown on any teardown path that stops audio before dropping
the sender.

Fix: extract a `drain_loop` seam that uses recv_timeout(WORKER_POLL=100ms) so
the loop re-checks `running` at least every 100ms even when idle, and returns
promptly on Disconnected. stop() now joins within one poll interval regardless
of the sender's lifetime. +3 unit tests (151 lib): exits on running-flip with
the sender still alive (the exact hang case, asserted via is_finished), returns
on disconnect, and delivers frames. Verified: audio_probe now self-exits cleanly
(exit 0, "done.", no lingering process). clippy --all-targets clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/audio/pipewire_impl.rs

@@ -1,5 +1,5 @@
 use crate::audio::{AudioBackend, AudioError};
-use std::sync::mpsc::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver, RecvTimeoutError};
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::thread::{self, JoinHandle};
@@ -238,6 +238,31 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>, target_n
 ///   never the whole slice, because over-pulling past the ring depth is exactly
 ///   the sustained crackle this replaces. Under-filling a cycle is a harmless
 ///   brief glitch (PipeWire pads).
+// How often the playback worker wakes to re-check its `running` flag while idle,
+// bounding how long `stop` can block joining the worker (bug A7). A plain
+// blocking receive would never re-check the flag, waking only on a new frame or
+// the sender dropping, so shutdown would hang if the sender outlives `stop`.
+const WORKER_POLL: Duration = Duration::from_millis(100);
+
+/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
+/// disconnects. Uses a timed receive so the loop re-checks `running` at least
+/// every `WORKER_POLL` even when no frames arrive — this is what lets `stop()`
+/// join the worker promptly instead of hanging on a parked blocking `recv()`
+/// (bug A7). Pure w.r.t. its inputs (no PipeWire), so it's unit-testable.
+fn drain_loop(
+    rx: &Receiver<Vec<i16>>,
+    running: &AtomicBool,
+    mut on_frame: impl FnMut(Vec<i16>),
+) {
+    while running.load(Ordering::Relaxed) {
+        match rx.recv_timeout(WORKER_POLL) {
+            Ok(frame) => on_frame(frame),
+            Err(RecvTimeoutError::Timeout) => continue, // re-check `running`
+            Err(RecvTimeoutError::Disconnected) => return, // sender gone for good
+        }
+    }
+}
+
 fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize {
     /// Safe per-cycle fallback when the graph doesn't report a quantum.
     const FALLBACK_FRAMES: usize = 1024;
@@ -483,24 +508,20 @@ fn run_playback(
     let worker_dropped = dropped_frames.clone();
     let worker_fill = fill_gauge.clone();
     let worker_handle = thread::spawn(move || {
-        while running_clone.load(Ordering::Relaxed) {
-            if let Ok(frame) = rx.recv() {
-                // Drop the whole frame (rather than tearing it) only if it truly
-                // won't fit — checked against the exact occupancy counter, not
-                // the ring's stale length observer. With clock-paced production
-                // this should never fire.
-                if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
-                    worker_dropped.fetch_add(1, Ordering::Relaxed);
-                    continue;
-                }
-                for &sample in &frame {
-                    let _ = producer.try_push(sample);
-                }
-                worker_fill.fetch_add(frame.len(), Ordering::Relaxed);
-            } else {
+        drain_loop(&rx, &running_clone, |frame| {
+            // Drop the whole frame (rather than tearing it) only if it truly
+            // won't fit — checked against the exact occupancy counter, not
+            // the ring's stale length observer. With clock-paced production
+            // this should never fire.
+            if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
+                worker_dropped.fetch_add(1, Ordering::Relaxed);
                 return;
             }
-        }
+            for &sample in &frame {
+                let _ = producer.try_push(sample);
+            }
+            worker_fill.fetch_add(frame.len(), Ordering::Relaxed);
+        });
     });
 
     // Diagnostic logger: once per second, report the ring fill and the delta in
@@ -551,7 +572,11 @@ fn run_playback(
 
 #[cfg(test)]
 mod tests {
-    use super::frames_to_produce;
+    use super::{drain_loop, frames_to_produce};
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::{Arc, Mutex};
+    use std::time::Duration;
+    use std::{sync::mpsc, thread};
 
     #[test]
     fn requested_in_range_is_honored() {
@@ -583,4 +608,56 @@ mod tests {
         assert_eq!(frames_to_produce(0, 0), 0);
         assert_eq!(frames_to_produce(1024, 0), 0);
     }
+
+    // --- drain_loop (A7: worker must not hang shutdown) ---
+
+    #[test]
+    fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
+        // The exact A7 hang scenario: the frame sender is STILL alive (never
+        // dropped) when `running` goes false. A blocking `recv()` would park
+        // forever; `drain_loop` must wake within a poll interval and return.
+        let (tx, rx) = mpsc::channel::<Vec<i16>>();
+        let running = Arc::new(AtomicBool::new(true));
+        let r2 = running.clone();
+        let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
+        // Let it park in recv_timeout, then signal stop.
+        thread::sleep(Duration::from_millis(50));
+        running.store(false, Ordering::Relaxed);
+        // Wait comfortably longer than one poll interval; it must have exited.
+        thread::sleep(super::WORKER_POLL + Duration::from_millis(150));
+        assert!(
+            h.is_finished(),
+            "drain_loop must exit after running=false even while the sender is alive"
+        );
+        drop(tx); // keep tx alive until here so the test really covers the case
+        h.join().unwrap();
+    }
+
+    #[test]
+    fn drain_loop_returns_on_disconnect() {
+        // Sender dropped before we start → recv errors Disconnected → return.
+        // If this hangs, the test runner times out (i.e. it would fail loudly).
+        let (tx, rx) = mpsc::channel::<Vec<i16>>();
+        let running = Arc::new(AtomicBool::new(true));
+        drop(tx);
+        drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
+    }
+
+    #[test]
+    fn drain_loop_delivers_frames_to_callback() {
+        let (tx, rx) = mpsc::channel::<Vec<i16>>();
+        let running = Arc::new(AtomicBool::new(true));
+        let r2 = running.clone();
+        let got = Arc::new(Mutex::new(Vec::new()));
+        let g2 = got.clone();
+        let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
+        tx.send(vec![1, 2, 3]).unwrap();
+        tx.send(vec![4, 5]).unwrap();
+        thread::sleep(Duration::from_millis(50));
+        running.store(false, Ordering::Relaxed);
+        drop(tx);
+        h.join().unwrap();
+        let got = got.lock().unwrap();
+        assert_eq!(*got, vec![vec![1, 2, 3], vec![4, 5]]);
+    }
 }