audio(win): tighten the cpal start-handshake (Codex re-review B1/B2/B4)

Codex's xhigh re-review of the prior cpal RT fixes confirmed W2/W3/W7/W4-diag
addressed (and validated the reserve-first ring-publish ordering), but found the
W1/W6 start-handshake fixes were partial. This closes the holes:

- B1 (P1): wait_for_stream_start checked the liveness flag before the error code,
  so a callback that ran then failed in the same WASAPI cycle could still report
  Ok on a dead stream. Readiness now (a) treats the error as terminal — checked
  first each loop AND re-checked before returning Ok — and (b) requires
  MIN_START_CALLBACKS (2) completed callbacks, not one, so a fire-once-then-die
  stream is caught by the error/timeout path. The liveness signal is now a
  callback counter (AtomicUsize) instead of a one-shot bool.
- B2 (P2): on the inner STREAM_START_TIMEOUT the owner sent Err and THEN dropped
  the stream; since cpal Stream::drop joins its (wedged) WASAPI worker and
  finish_start joins the owner on that Err, start_*/stop could still hang past the
  backstop. The owner now drops the stream BEFORE reporting Err, so a wedged drop
  withholds the Err and lets finish_start's timeout branch detach.
- B4 (P3): the two timeouts didn't compose — a slow-but-valid setup plus a slow
  first callback could exceed the 6s backstop and be falsely failed. Raised
  FINISH_START_TIMEOUT to 10s (setup budget + callback wait + cleanup slack) and
  corrected the comment.

Deferred (logged in review-2026-06-19-cpal-rt-audit.md): B3 (orphan-thread
tombstone accounting on a permanent >10s driver wedge — rare, non-crashing, needs
a slot-state redesign) and B5 (choose_config picking a bounded supported rate for
an oddball sub-8k/over-384k default-rate device — rare; the safety validation
already prevents the panic/spin).

Verified: Linux cargo test --lib 326/0, clippy --all-targets clean; windows-gnu
cargo check --lib --tests --bins clean; windows-gnu release peerspeak.exe links.

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

src/audio/cpal_impl.rs

@@ -88,16 +88,23 @@ const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS;
 /// How often a blocked playback worker re-checks its `running` flag, bounding how
 /// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`).
 const WORKER_POLL: Duration = Duration::from_millis(100);
-/// How long a freshly-played stream has to deliver its first RT callback (proof
+/// How long a freshly-played stream has to prove itself (deliver its first RT
-/// WASAPI actually `Start()`ed it) before the start is treated as failed. cpal's
+/// callbacks) before the start is treated as failed. cpal's `play()` only *queues*
-/// `play()` only *queues* the WASAPI start, so a queued-but-failed start would
+/// the WASAPI `Start()`, so a queued-but-failed start would otherwise masquerade as
-/// otherwise masquerade as success and join the UI into a silent room (review W1).
+/// success and join the UI into a silent room (review W1).
 const STREAM_START_TIMEOUT: Duration = Duration::from_secs(3);
-/// Backstop for [`finish_start`]: covers the whole owner setup (resolve + build +
+/// Number of completed RT callbacks the owner waits for before declaring the stream
-/// play + first-callback wait). Larger than [`STREAM_START_TIMEOUT`] so the owner's
+/// live. One callback isn't proof: a stream can fire once and immediately fail in
-/// own timeout normally fires first with a precise error; this only trips if the
+/// the same processing cycle, so requiring a couple of cycles (plus the terminal
-/// owner itself wedges in a driver call before it can report readiness (review W6).
+/// error check) keeps a one-shot-then-dead stream from being reported Ok (B1).
-const FINISH_START_TIMEOUT: Duration = Duration::from_secs(6);
+const MIN_START_CALLBACKS: usize = 2;
+/// Backstop for [`finish_start`]: bounds the WHOLE owner path (resolve + build +
+/// play + the [`STREAM_START_TIMEOUT`] callback wait + any wedged-stream cleanup).
+/// Sized as a generous setup budget plus the callback wait plus slack so a slow but
+/// valid device (e.g. a Bluetooth endpoint that takes seconds to spin up) is not
+/// falsely failed, while a driver that wedges before the owner can report is still
+/// released eventually (review W6, B4).
+const FINISH_START_TIMEOUT: Duration = Duration::from_secs(10);
 /// Lowest / highest device sample rate the backend will drive. The floor bounds
 /// the playback pull-resampler's input-pulls-per-output-frame (≈48000/rate) so a
 /// pathological low rate can't blow the RT callback's deadline; the ceiling and a
@@ -132,24 +139,36 @@ fn stream_err_text(code: u8) -> &'static str {
     }
 }
 
-/// Wait for a just-played stream to prove it actually started: its first RT
+/// Wait for a just-played stream to prove it actually started: its RT data
-/// callback sets `started`, or an error callback sets `err_code`. Returns `Ok` on
+/// callback bumps `callbacks`, or an error callback sets `err_code`. Returns `Ok`
-/// the first callback, `Err` on an error-callback code or [`STREAM_START_TIMEOUT`],
+/// once [`MIN_START_CALLBACKS`] cycles have run with no error, `Err` on an
-/// or a clean abort if `stop()` cleared `running` mid-start. Polls a couple of
+/// error-callback code or [`STREAM_START_TIMEOUT`], or a clean abort if `stop()`
-/// cheap atomics on the owner thread — never the RT thread (review W1).
+/// cleared `running` mid-start. Polls a few cheap atomics on the owner thread —
+/// never the RT thread (review W1).
+///
+/// The error is **terminal and wins any race** with `callbacks`: a stream can run a
+/// callback and then fail in the same processing cycle, so `err_code` is checked
+/// first each loop AND re-checked before declaring success (review B1).
 fn wait_for_stream_start(
-    started: &AtomicBool,
+    callbacks: &AtomicUsize,
     err_code: &AtomicU8,
     running: &AtomicBool,
 ) -> Result<(), AudioError> {
     let deadline = Instant::now() + STREAM_START_TIMEOUT;
+    let as_err = |code: u8| Err(AudioError::Stream(stream_err_text(code).to_string()));
     loop {
-        if started.load(Ordering::Relaxed) {
-            return Ok(());
-        }
         let code = err_code.load(Ordering::Relaxed);
         if code != STREAM_ERR_NONE {
-            return Err(AudioError::Stream(stream_err_text(code).to_string()));
+            return as_err(code);
+        }
+        if callbacks.load(Ordering::Relaxed) >= MIN_START_CALLBACKS {
+            // Re-check: a callback that pushed us to the threshold may have been the
+            // last before a same-cycle failure. Let a terminal error win.
+            let code = err_code.load(Ordering::Relaxed);
+            if code != STREAM_ERR_NONE {
+                return as_err(code);
+            }
+            return Ok(());
         }
         if !running.load(Ordering::Relaxed) {
             return Err(AudioError::Stream("stream start aborted".to_string()));
@@ -487,10 +506,10 @@ fn run_capture(
     let rb = HeapRb::<i16>::new(CAPTURE_RING_CAPACITY);
     let (producer, mut consumer) = rb.split();
     let overrun = Arc::new(AtomicU64::new(0));
-    // Stream-liveness signals read by `wait_for_stream_start`: the first RT data
+    // Stream-liveness signals read by `wait_for_stream_start`: each RT data callback
-    // callback sets `started`; the RT error callback sets `err_code` (it does NOT
+    // bumps `callbacks`; the RT error callback sets `err_code` (it does NOT log —
-    // log — that would allocate/syscall on the time-critical thread). See W1/W2.
+    // that would allocate/syscall on the time-critical thread). See W1/W2/B1.
-    let started = Arc::new(AtomicBool::new(false));
+    let callbacks = Arc::new(AtomicUsize::new(0));
     let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
 
     // Fallible device/stream setup: resolve, build, and *queue* the stream start.
@@ -500,15 +519,15 @@ fn run_capture(
         let device_rate = config.sample_rate.0;
         let stream = match sample_format {
             SampleFormat::F32 => build_input::<f32, _>(
-                &device, &config, producer, channels, overrun.clone(), started.clone(),
+                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                 err_code.clone(),
             ),
             SampleFormat::I16 => build_input::<i16, _>(
-                &device, &config, producer, channels, overrun.clone(), started.clone(),
+                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                 err_code.clone(),
             ),
             SampleFormat::U16 => build_input::<u16, _>(
-                &device, &config, producer, channels, overrun.clone(), started.clone(),
+                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                 err_code.clone(),
             ),
             other => Err(AudioError::Stream(format!(
@@ -526,14 +545,18 @@ fn run_capture(
     // reporting readiness. `play()` returning Ok only means WASAPI's `Start()` was
     // queued; a later Start failure would otherwise leave us joined-but-silent (W1).
     let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
-        Ok(v) => match wait_for_stream_start(&started, &err_code, &running) {
+        Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
             Ok(()) => {
                 let _ = ready.send(Ok(()));
                 v
             }
             Err(e) => {
+                // Drop the (possibly wedged) stream BEFORE reporting: cpal's
+                // Stream::drop joins its WASAPI worker, so if that wedges we want
+                // the Err withheld and finish_start's backstop to detach, rather
+                // than finish_start joining this owner forever (B2).
+                drop(v.0);
                 let _ = ready.send(Err(e));
-                drop(v.0); // the stream
                 return;
             }
         },
@@ -611,7 +634,7 @@ fn build_input<T, P>(
     mut producer: P,
     channels: usize,
     overrun: Arc<AtomicU64>,
-    started: Arc<AtomicBool>,
+    callbacks: Arc<AtomicUsize>,
     err_code: Arc<AtomicU8>,
 ) -> Result<Stream, AudioError>
 where
@@ -627,8 +650,9 @@ where
         .build_input_stream::<T, _, _>(
             config,
             move |data: &[T], _| {
-                // First callback proves WASAPI actually started the stream (W1).
+                // Count callbacks so the owner can confirm the stream is really
-                started.store(true, Ordering::Relaxed);
+                // running before reporting Ok (W1/B1).
+                callbacks.fetch_add(1, Ordering::Relaxed);
                 // RT-safe: downmix + wait-free push only. A full ring means the
                 // drain thread stalled; count the drop and keep going.
                 for frame in data.chunks_exact(channels) {
@@ -731,9 +755,9 @@ fn run_playback(
     // which would force an underrun every cycle (review W2). The callback only
     // does a wait-free fetch_max; the health logger reports/warns off the RT path.
     let max_cb = Arc::new(AtomicUsize::new(0));
-    // Stream-liveness signals (see the capture path / W1, W2): first RT callback
+    // Stream-liveness signals (see the capture path / W1, W2, B1): each RT callback
-    // sets `started`; the RT error callback sets `err_code` without logging.
+    // bumps `callbacks`; the RT error callback sets `err_code` without logging.
-    let started = Arc::new(AtomicBool::new(false));
+    let callbacks = Arc::new(AtomicUsize::new(0));
     let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));
 
     // Fallible device/stream setup. `consumer` is moved into the output callback.
@@ -744,15 +768,15 @@ fn run_playback(
         let stream = match sample_format {
             SampleFormat::F32 => build_output::<f32, _>(
                 &device, &config, consumer, ring_fill.clone(), underrun.clone(),
-                max_cb.clone(), started.clone(), err_code.clone(),
+                max_cb.clone(), callbacks.clone(), err_code.clone(),
             ),
             SampleFormat::I16 => build_output::<i16, _>(
                 &device, &config, consumer, ring_fill.clone(), underrun.clone(),
-                max_cb.clone(), started.clone(), err_code.clone(),
+                max_cb.clone(), callbacks.clone(), err_code.clone(),
             ),
             SampleFormat::U16 => build_output::<u16, _>(
                 &device, &config, consumer, ring_fill.clone(), underrun.clone(),
-                max_cb.clone(), started.clone(), err_code.clone(),
+                max_cb.clone(), callbacks.clone(), err_code.clone(),
             ),
             other => Err(AudioError::Stream(format!(
                 "unsupported playback sample format: {other:?}"
@@ -765,16 +789,18 @@ fn run_playback(
         Ok((stream, name, sample_format, channels, device_rate))
     };
 
-    // Build + queue, then wait for a real callback before reporting readiness (W1).
+    // Build + queue, then wait for real callbacks before reporting readiness (W1).
     let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
-        Ok(v) => match wait_for_stream_start(&started, &err_code, &running) {
+        Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
             Ok(()) => {
                 let _ = ready.send(Ok(()));
                 v
             }
             Err(e) => {
+                // Drop before reporting so a wedged Stream::drop withholds the Err
+                // and lets finish_start's backstop detach instead of hanging (B2).
+                drop(v.0);
                 let _ = ready.send(Err(e));
-                drop(v.0); // the stream
                 return;
             }
         },
@@ -835,7 +861,7 @@ fn build_output<T, C>(
     ring_fill: Arc<AtomicUsize>,
     underrun: Arc<AtomicU64>,
     max_cb: Arc<AtomicUsize>,
-    started: Arc<AtomicBool>,
+    callbacks: Arc<AtomicUsize>,
     err_code: Arc<AtomicU8>,
 ) -> Result<Stream, AudioError>
 where
@@ -851,7 +877,7 @@ where
             .build_output_stream::<T, _, _>(
                 config,
                 move |data: &mut [T], _| {
-                    started.store(true, Ordering::Relaxed);
+                    callbacks.fetch_add(1, Ordering::Relaxed);
                     // Record demand in INTERNAL 48 kHz-stereo samples (not raw
                     // device samples) so the health logger's prefill-target
                     // comparison is apples-to-apples for any rate/layout (W4).
@@ -880,7 +906,7 @@ where
             .build_output_stream::<T, _, _>(
                 config,
                 move |data: &mut [T], _| {
-                    started.store(true, Ordering::Relaxed);
+                    callbacks.fetch_add(1, Ordering::Relaxed);
                     max_cb.fetch_max(
                         internal_demand(data.len(), device_channels, device_rate),
                         Ordering::Relaxed,