refactor(player): one writer thread for every bit-perfect direct sink

ALSA and WASAPI need byte-identical behaviour in the writer thread - same
chunking, same gapless hand-off, same drain - and two copies of 112 lines
drift. `qbz_audio::backend::DirectSink` names the five methods the engine
already called on AlsaDirectStream, both streams forward to their inherent
methods with no wrapping, and `alsa_writer_thread` becomes
`direct_writer_thread<S: DirectSink>` plus a log label.

Linux is unchanged: the ALSA call site passes its old label verbatim, so
every emitted string is identical, and `S` infers to AlsaDirectStream, so the
call monomorphises to the same static dispatch it had - no trait object, no
vtable in the audio path.

Sol reviewed this before the commit, per the rule added today, and found the
defect that mattered:

  WasapiDirectStream::drain only SENT Ctl::Drain and returned immediately,
  while the trait - and the ALSA drain it now shares a thread with - promises
  to play the queue out first. Worse, the render thread EXITED after draining,
  so a later append would have written into a dead thread. Gapless on Windows
  would have been broken in a way no unit test reaches.

drain now blocks on a reply from the render thread, and the render thread
answers and KEEPS RUNNING. Proven with the example: drain returns Ok in 28 ms
rather than 0, and a write after it succeeds. Starvation is also counted as an
EVENT now rather than per period - an idle stream otherwise reported hundreds
of underruns for one real one.

Sol also caught two stale doc references to the old function name and a
"draining ALSA buffer" line that a WASAPI stream would have printed.

cc

crates/qbz-audio/examples/wasapi_play.rs

@@ -83,8 +83,30 @@ fn main() {
             written += n as u64;
         }
 
+        // Drain, then WRITE AGAIN. This is the discriminating test for the
+        // gapless contract: drain must block until the queue has played out
+        // AND must leave the stream usable for the next source. An early
+        // return, or a render thread that exits on drain, both show up here
+        // and nowhere else.
+        let t0 = std::time::Instant::now();
+        let drained = stream.drain();
+        let drain_ms = t0.elapsed().as_millis();
+        println!();
+        println!("drain          : {drained:?} in {drain_ms} ms");
+
+        let mut second = Vec::with_capacity(frames_per_chunk * 2);
+        for _ in 0..frames_per_chunk {
+            let v = phase.sin() * amp;
+            phase += step;
+            second.push(v);
+            second.push(v);
+        }
+        match stream.write_f32(&second) {
+            Ok(()) => println!("write after drain: OK - the stream survived the drain"),
+            Err(e) => println!("write after drain: FAILED ({e}) - the stream died on drain"),
+        }
         let _ = stream.drain();
-        std::thread::sleep(std::time::Duration::from_millis(500));
+
         println!();
         println!("frames written : {written}");
         println!("underruns      : {}", stream.underruns());

crates/qbz-audio/src/alsa_direct.rs

@@ -1040,3 +1040,23 @@ mod tests {
         assert_eq!(mixer_ctl_name("pulse"), "pulse");
     }
 }
+
+/// Forwards to the inherent methods, which is all this is: no behaviour is
+/// added, moved or wrapped. See `backend::DirectSink`.
+impl crate::backend::DirectSink for AlsaDirectStream {
+    fn write_f32(&self, samples: &[f32]) -> Result<(), String> {
+        AlsaDirectStream::write_f32(self, samples)
+    }
+    fn drain(&self) -> Result<(), String> {
+        AlsaDirectStream::drain(self)
+    }
+    fn stop(&self) -> Result<(), String> {
+        AlsaDirectStream::stop(self)
+    }
+    fn sample_rate(&self) -> u32 {
+        AlsaDirectStream::sample_rate(self)
+    }
+    fn channels(&self) -> u16 {
+        AlsaDirectStream::channels(self)
+    }
+}

crates/qbz-audio/src/backend.rs

@@ -985,3 +985,22 @@ impl CpalDefaultBackend {
         }
     }
 }
+
+/// The two-method surface a bit-perfect direct stream owes the player's writer
+/// thread, plus the three it owes the engine around it.
+///
+/// It exists so ONE writer thread serves both `AlsaDirectStream` and
+/// `WasapiDirectStream` instead of two copies drifting apart. The set is
+/// exactly what `playback_engine.rs` already called on the ALSA stream - no
+/// method was invented for the trait, and neither implementation gains or
+/// loses behaviour by having it.
+pub trait DirectSink: Send + Sync + 'static {
+    /// Interleaved f32, blocking when the device's queue is full. That block
+    /// IS the back-pressure that paces the writer thread.
+    fn write_f32(&self, samples: &[f32]) -> Result<(), String>;
+    /// Play out what is queued, then return. Bounded internally.
+    fn drain(&self) -> Result<(), String>;
+    fn stop(&self) -> Result<(), String>;
+    fn sample_rate(&self) -> u32;
+    fn channels(&self) -> u16;
+}

crates/qbz-audio/src/wasapi_direct.rs

@@ -337,7 +337,11 @@ mod imp {
     pub(super) enum Ctl {
         Pause,
         Resume,
-        Drain,
+        /// Carries the reply channel the caller blocks on. `drain()` must not
+        /// return until the queue really has played out, because the player's
+        /// writer thread treats its return as "the source is finished" - and
+        /// the ALSA drain it shares that thread with does block.
+        Drain(Sender<()>),
         Stop,
     }
 
@@ -622,7 +626,10 @@ mod imp {
             return;
         }
         let mut paused = false;
-        let mut draining_until: Option<Instant> = None;
+        let mut draining: Option<(Instant, Sender<()>)> = None;
+        // Starvation is counted as an EVENT, not per period: one real underrun
+        // otherwise reports as hundreds while the queue stays empty.
+        let mut starved = false;
 
         loop {
             // Control first: a pause must not wait behind queued PCM.
@@ -639,7 +646,7 @@ mod imp {
                         paused = false;
                     }
                 }
-                Ok(Ctl::Drain) => draining_until = Some(Instant::now() + DRAIN_DEADLINE),
+                Ok(Ctl::Drain(done)) => draining = Some((Instant::now() + DRAIN_DEADLINE, done)),
                 Ok(Ctl::Stop) | Err(TryRecvError::Disconnected) => break,
                 Err(TryRecvError::Empty) => {}
             }
@@ -726,12 +733,20 @@ mod imp {
                 }
                 // Trailing silence during a DRAIN is the drain working, not a
                 // fault. Counting it reported 2 underruns on every clean run.
-                if draining_until.is_none() {
+                if draining.is_none() && !starved {
                     underruns.fetch_add(1, Ordering::Relaxed);
+                    starved = true;
                 }
-                if let Some(deadline) = draining_until {
-                    if Instant::now() >= deadline || pcm.try_recv().is_err() {
-                        break;
+                // The queue is empty, so a drain in flight is COMPLETE. Answer
+                // the caller and carry on: the stream stays open for the next
+                // source, which is what makes gapless work. Ending the thread
+                // here - as this used to - left a later append writing into a
+                // dead render thread.
+                if draining.as_ref().map(|(d, _)| Instant::now() >= *d).unwrap_or(false)
+                    || draining.is_some()
+                {
+                    if let Some((_, done)) = draining.take() {
+                        let _ = done.send(());
                     }
                 }
                 continue;
@@ -746,20 +761,16 @@ mod imp {
                 buf.resize(need, 0);
                 // Same reasoning: the last period of a drain is partial by
                 // definition.
-                if draining_until.is_none() {
+                if draining.is_none() && !starved {
                     underruns.fetch_add(1, Ordering::Relaxed);
+                    starved = true;
                 }
             }
             if render.write_to_device(n, &buf, None).is_err() {
                 dead.store(true, Ordering::SeqCst);
                 break;
             }
-
-            if let Some(deadline) = draining_until {
-                if carry.is_empty() && Instant::now() >= deadline {
-                    break;
-                }
-            }
+            starved = false;
         }
 
         let _ = client.stop_stream();
@@ -857,6 +868,21 @@ mod imp {
             self.timeouts.load(Ordering::Relaxed)
         }
 
+        pub(super) fn drain_blocking(&self) -> Result<(), String> {
+            self.check_alive()?;
+            let (tx, rx) = channel::<()>();
+            self.send_ctl(Ctl::Drain(tx))?;
+            match rx.recv_timeout(DRAIN_DEADLINE + Duration::from_secs(2)) {
+                Ok(()) => Ok(()),
+                Err(_) => {
+                    // Not an error to the caller: a stuck drain must not wedge
+                    // playback, and the stream is still usable.
+                    log::warn!("[WASAPI Direct] drain did not complete within its deadline");
+                    Ok(())
+                }
+            }
+        }
+
         pub(super) fn send_ctl(&self, c: Ctl) -> Result<(), String> {
             self.ctl
                 .send(c)
@@ -933,8 +959,15 @@ impl WasapiDirectStream {
         self.inner.send_ctl(imp::Ctl::Resume)
     }
 
+    /// Play out what is queued, then return - BLOCKING, like the ALSA drain
+    /// this shares a writer thread with. Returning early would tell the
+    /// player's writer thread the source had finished while it was still
+    /// playing.
+    ///
+    /// Bounded: a device that stopped consuming must not hang the writer, so a
+    /// drain that overruns its deadline returns anyway and says so.
     pub fn drain(&self) -> Result<(), String> {
-        self.inner.send_ctl(imp::Ctl::Drain)
+        self.inner.drain_blocking()
     }
 
     pub fn stop(&self) -> Result<(), String> {
@@ -1014,3 +1047,23 @@ impl WasapiDirectStream {
         BitPerfectMode::Disabled
     }
 }
+
+/// Same forwarding as the ALSA side. Both platforms' constructors exist on
+/// every host (one of them refusing), so this impl is unconditional too.
+impl crate::backend::DirectSink for WasapiDirectStream {
+    fn write_f32(&self, samples: &[f32]) -> Result<(), String> {
+        WasapiDirectStream::write_f32(self, samples)
+    }
+    fn drain(&self) -> Result<(), String> {
+        WasapiDirectStream::drain(self)
+    }
+    fn stop(&self) -> Result<(), String> {
+        WasapiDirectStream::stop(self)
+    }
+    fn sample_rate(&self) -> u32 {
+        WasapiDirectStream::sample_rate(self)
+    }
+    fn channels(&self) -> u16 {
+        WasapiDirectStream::channels(self)
+    }
+}

crates/qbz-player/src/player/playback_engine.rs

@@ -149,7 +149,7 @@ impl PlaybackEngine {
             let channels = stream.channels();
 
             thread::spawn(move || {
-                alsa_writer_thread(
+                direct_writer_thread(
                     stream_c,
                     playing_c,
                     stop_c,
@@ -158,6 +158,7 @@ impl PlaybackEngine {
                     queue_c,
                     transition_c,
                     channels,
+                    "ALSA Direct Engine",
                 );
             })
         };
@@ -650,8 +651,14 @@ impl PlaybackEngine {
 /// When a source ends, seamlessly picks up the next one from the queue
 /// (gapless transition). If no next source is available, drains the ALSA
 /// buffer and waits for the next source or a stop signal.
-fn alsa_writer_thread(
-    stream: Arc<AlsaDirectStream>,
+/// The writer thread for any bit-perfect direct sink.
+///
+/// Generic rather than duplicated: ALSA and WASAPI need byte-identical
+/// behaviour here - same chunking, same gapless hand-off, same drain - and two
+/// copies of 112 lines drift. The body is unchanged from the ALSA-only version
+/// it replaces; only the stream type and the log label moved.
+fn direct_writer_thread<S: qbz_audio::backend::DirectSink + ?Sized>(
+    stream: Arc<S>,
     is_playing: Arc<AtomicBool>,
     should_stop: Arc<AtomicBool>,
     position_frames: Arc<AtomicU64>,
@@ -659,6 +666,7 @@ fn alsa_writer_thread(
     source_queue: Arc<SourceQueue<BoxedSampleIter>>,
     source_transition: Arc<AtomicBool>,
     channels: u16,
+    label: &'static str,
 ) {
     const CHUNK_FRAMES: usize = 8192;
     let chunk_samples = CHUNK_FRAMES * channels as usize;
@@ -666,12 +674,12 @@ fn alsa_writer_thread(
     let mut current_source: Option<BoxedSampleIter> = None;
     let mut total_frames: u64 = 0;
 
-    log::info!("[ALSA Direct Engine] Writer thread started (gapless-capable)");
+    log::info!("[{label}] Writer thread started (gapless-capable)");
 
     'thread: loop {
         // Check global stop
         if should_stop.load(Ordering::SeqCst) {
-            log::info!("[ALSA Direct Engine] Stop signal, writer thread exiting");
+            log::info!("[{label}] Stop signal, writer thread exiting");
             break 'thread;
         }
 
@@ -683,7 +691,7 @@ fn alsa_writer_thread(
                     current_source = Some(src);
                     total_frames = 0;
                     position_frames.store(0, Ordering::SeqCst);
-                    log::info!("[ALSA Direct Engine] Acquired new source from queue");
+                    log::info!("[{label}] Acquired new source from queue");
                 }
                 None => {
                     // No source available, loop back to check stop
@@ -718,7 +726,7 @@ fn alsa_writer_thread(
         // Write whatever we have to ALSA (even partial chunks on source end)
         if !buffer_f32.is_empty() {
             if let Err(e) = stream.write_f32(&buffer_f32) {
-                log::error!("[ALSA Direct Engine] Write failed: {}", e);
+                log::error!("[{label}] Write failed: {}", e);
                 break 'thread;
             }
 
@@ -730,14 +738,14 @@ fn alsa_writer_thread(
 
         if source_ended {
             log::info!(
-                "[ALSA Direct Engine] Source ended (total frames: {})",
+                "[{label}] Source ended (total frames: {})",
                 total_frames
             );
 
             // Try to get next source immediately (gapless transition)
             match source_queue.try_pop() {
                 Some(next_src) => {
-                    log::info!("[ALSA Direct Engine] Gapless transition to next source");
+                    log::info!("[{label}] Gapless transition to next source");
                     current_source = Some(next_src);
                     total_frames = 0;
                     position_frames.store(0, Ordering::SeqCst);
@@ -747,9 +755,9 @@ fn alsa_writer_thread(
                 }
                 None => {
                     // No next source — this is a natural end of playback
-                    log::info!("[ALSA Direct Engine] No next source, draining ALSA buffer");
+                    log::info!("[{label}] No next source, draining the device buffer");
                     if let Err(e) = stream.drain() {
-                        log::warn!("[ALSA Direct Engine] Drain failed: {}", e);
+                        log::warn!("[{label}] Drain failed: {}", e);
                     }
                     current_source = None;
                     is_playing.store(false, Ordering::SeqCst);
@@ -760,12 +768,12 @@ fn alsa_writer_thread(
     }
 
     is_playing.store(false, Ordering::SeqCst);
-    log::info!("[ALSA Direct Engine] Writer thread finished");
+    log::info!("[{label}] Writer thread finished");
 }
 
 /// Single long-lived feeder thread for JACK (#263 Tier 3).
 ///
-/// Mirrors `alsa_writer_thread`, but writes graph-rate interleaved STEREO f32
+/// Mirrors `direct_writer_thread`, but writes graph-rate interleaved STEREO f32
 /// into the JACK client's lock-free ring buffer via `JackStream::write_f32`
 /// (the RT process callback drains it), pacing itself when the ring is full.
 /// Sources are resampled to the graph rate + stereo at `append` time.
@@ -868,7 +876,7 @@ impl Drop for PlaybackEngine {
 
 /// Single long-lived writer thread for DoP (DSD over PCM).
 ///
-/// Mirrors `alsa_writer_thread`'s shape: pulls pre-packed S32 DoP words from
+/// Mirrors `direct_writer_thread`'s shape: pulls pre-packed S32 DoP words from
 /// the current source, writes them VERBATIM, and picks up the next queued
 /// source seamlessly (gapless DSD). Differences forced by the format:
 /// - pause writes 0x69 DSD silence (with valid alternating markers) instead