Merge branch 'pr/599' into integration/pr-review-smoke-2026-07-13

crates/qbz-dsd/src/dop.rs

@@ -82,6 +82,8 @@ pub struct DopStream {
     buf: Vec<i32>,
     idx: usize,
     done: bool,
+    /// Set when demux I/O fails mid-stream (not clean EOF).
+    io_error: Option<String>,
 }
 
 /// DSD bytes pulled from the demuxer per refill, per channel.
@@ -102,9 +104,15 @@ impl DopStream {
             buf: Vec::new(),
             idx: 0,
             done: false,
+            io_error: None,
         })
     }
 
+    /// Mid-stream demux I/O error, if any. Clean EOF leaves this `None`.
+    pub fn io_error(&self) -> Option<&str> {
+        self.io_error.as_deref()
+    }
+
     pub fn carrier_rate(&self) -> u32 {
         dop_carrier_rate(self.dsd_rate)
     }
@@ -119,7 +127,13 @@ impl DopStream {
     fn refill(&mut self) -> bool {
         let mut planar: Vec<Vec<u8>> = vec![Vec::new(), Vec::new()];
         match self.demux.read_planar(&mut planar, REFILL_BYTES_PER_CH) {
-            Ok(0) | Err(_) => {
+            Ok(0) => {
+                self.done = true;
+                false
+            }
+            Err(e) => {
+                log::error!("[DSD/DoP] demux I/O error (not clean EOF): {e}");
+                self.io_error = Some(e.to_string());
                 self.done = true;
                 false
             }
@@ -188,3 +202,65 @@ mod tests {
         assert_eq!(out[2], ((0xFA << 16) | 0x6969) << 8);
     }
 }
+
+#[cfg(test)]
+mod io_error_tests {
+    use super::*;
+    use crate::demux::{DsdDemuxer, DsdError, DsdStreamInfo};
+    use std::io;
+
+    struct FailAfter {
+        left: usize,
+        info: DsdStreamInfo,
+    }
+
+    impl DsdDemuxer for FailAfter {
+        fn info(&self) -> &DsdStreamInfo {
+            &self.info
+        }
+        fn read_planar(
+            &mut self,
+            out: &mut [Vec<u8>],
+            max_bytes_per_ch: usize,
+        ) -> Result<usize, DsdError> {
+            if self.left == 0 {
+                return Err(DsdError::Io(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "simulated mid-file I/O error",
+                )));
+            }
+            self.left -= 1;
+            let n = max_bytes_per_ch.min(4);
+            for ch in out.iter_mut().take(2) {
+                ch.extend(std::iter::repeat(0x69).take(n));
+            }
+            Ok(n)
+        }
+    }
+
+    #[test]
+    fn demux_io_error_sets_sticky_flag_not_clean_eof() {
+        let info = DsdStreamInfo {
+            channels: 2,
+            dsd_rate: 2_822_400,
+            sample_count: 1_000_000,
+            lsb_first: false,
+            tags: Default::default(),
+        };
+        let demux = FailAfter { left: 1, info };
+        let mut stream = DopStream::new(Box::new(demux)).unwrap();
+        // First refill succeeds; drain some samples.
+        let _ = stream.next();
+        // Force more refills until error.
+        for _ in 0..100_000 {
+            if stream.next().is_none() {
+                break;
+            }
+        }
+        assert!(stream.io_error().is_some(), "I/O error must be sticky");
+        assert!(stream
+            .io_error()
+            .unwrap()
+            .contains("simulated mid-file I/O error"));
+    }
+}

crates/qbz-dsd/src/lib.rs

@@ -21,6 +21,28 @@ pub use dop::{dop_carrier_rate, DopPacker, DopStream};
 pub use native::{native_u32_rate, NativeDsdStream, NATIVE_DSD_SILENCE_U32};
 pub use wav::{frames_to_pcm24, wav_header, wav_total_size};
 
+/// Common surface of the boxed DSD word streams (DoP and native): an
+/// `Iterator<Item = i32>` that can also report a mid-stream demux I/O
+/// error once it ends, since the iterator surface itself can only express
+/// "no more items" and would otherwise make a read failure look like a
+/// clean end of track.
+pub trait DsdWordSource: Iterator<Item = i32> + Send {
+    /// Mid-stream demux I/O error, if any. Clean EOF leaves this `None`.
+    fn io_error(&self) -> Option<&str>;
+}
+
+impl DsdWordSource for DopStream {
+    fn io_error(&self) -> Option<&str> {
+        DopStream::io_error(self)
+    }
+}
+
+impl DsdWordSource for NativeDsdStream {
+    fn io_error(&self) -> Option<&str> {
+        NativeDsdStream::io_error(self)
+    }
+}
+
 /// DSD64 base rate (bits per second per channel). Multiples of this are the
 /// only valid DSD rates: DSD64/128/256/512.
 pub const DSD64_RATE: u32 = 2_822_400;

crates/qbz-dsd/src/native.rs

@@ -39,6 +39,8 @@ pub struct NativeDsdStream {
     /// Carry-over bytes per channel when a demux read isn't a multiple of 4.
     carry: [Vec<u8>; 2],
     done: bool,
+    /// Set when demux I/O fails mid-stream (not clean EOF).
+    io_error: Option<String>,
 }
 
 const REFILL_BYTES_PER_CH: usize = 32 * 1024;
@@ -60,9 +62,15 @@ impl NativeDsdStream {
             idx: 0,
             carry: [Vec::new(), Vec::new()],
             done: false,
+            io_error: None,
         })
     }
 
+    /// Mid-stream demux I/O error, if any. Clean EOF leaves this `None`.
+    pub fn io_error(&self) -> Option<&str> {
+        self.io_error.as_deref()
+    }
+
     pub fn rate(&self) -> u32 {
         native_u32_rate(self.dsd_rate)
     }
@@ -91,7 +99,12 @@ impl NativeDsdStream {
         let pre = [planar[0].len(), planar[1].len()];
         let got = match self.demux.read_planar(&mut planar, REFILL_BYTES_PER_CH) {
             Ok(n) => n,
-            Err(_) => 0,
+            Err(e) => {
+                log::error!("[DSD/native] demux I/O error (not clean EOF): {e}");
+                self.io_error = Some(e.to_string());
+                self.done = true;
+                return false;
+            }
         };
         if self.lsb_first {
             for (i, chan) in planar.iter_mut().enumerate() {

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

@@ -2687,7 +2687,10 @@ impl Player {
                                 *current_streaming_source = None;
 
                                 let mut engine = PlaybackEngine::new_alsa_dop(stream, false);
-                                if let Err(e) = engine.append_dop(Box::new(dop)) {
+                                if let Err(e) = engine.append_dop(Box::new(DsdErrorReport::new(
+                                    dop,
+                                    thread_state.clone(),
+                                ))) {
                                     log::error!("DoP: append failed: {}", e);
                                     thread_state.set_stream_error(true);
                                     return;
@@ -2802,7 +2805,10 @@ impl Player {
                                 *current_streaming_source = None;
 
                                 let mut engine = PlaybackEngine::new_alsa_dop(stream, true);
-                                if let Err(e) = engine.append_dop(Box::new(native_src)) {
+                                if let Err(e) = engine.append_dop(Box::new(DsdErrorReport::new(
+                                    native_src,
+                                    thread_state.clone(),
+                                ))) {
                                     log::error!("Native DSD: append failed: {}", e);
                                     thread_state.set_stream_error(true);
                                     return;
@@ -2855,7 +2861,10 @@ impl Player {
                                                 let rate = st.carrier_rate();
                                                 let frames = st.total_frames();
                                                 (
-                                                    Box::new(st)
+                                                    Box::new(DsdErrorReport::new(
+                                                        st,
+                                                        thread_state.clone(),
+                                                    ))
                                                         as Box<
                                                             dyn Iterator<Item = i32> + Send,
                                                         >,
@@ -2869,7 +2878,10 @@ impl Player {
                                                 let rate = st.rate();
                                                 let frames = st.total_frames();
                                                 (
-                                                    Box::new(st)
+                                                    Box::new(DsdErrorReport::new(
+                                                        st,
+                                                        thread_state.clone(),
+                                                    ))
                                                         as Box<
                                                             dyn Iterator<Item = i32> + Send,
                                                         >,
@@ -5146,6 +5158,44 @@ pub struct PlaybackState {
     pub volume: f32,
 }
 
+/// Surfaces a DSD stream's mid-file demux I/O error as a stream error when
+/// the stream runs out. The DoP writer consumes these streams as boxed
+/// `Iterator<Item = i32>`, which can only say "no more words", so without
+/// this wrapper a read failure is indistinguishable from a clean end of
+/// track and the user gets a silent early stop.
+#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
+struct DsdErrorReport<S: qbz_dsd::DsdWordSource> {
+    inner: S,
+    state: Arc<SharedState>,
+    reported: bool,
+}
+
+#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
+impl<S: qbz_dsd::DsdWordSource> DsdErrorReport<S> {
+    fn new(inner: S, state: Arc<SharedState>) -> Self {
+        Self {
+            inner,
+            state,
+            reported: false,
+        }
+    }
+}
+
+impl<S: qbz_dsd::DsdWordSource> Iterator for DsdErrorReport<S> {
+    type Item = i32;
+    fn next(&mut self) -> Option<i32> {
+        let item = self.inner.next();
+        if item.is_none() && !self.reported {
+            self.reported = true;
+            if let Some(e) = self.inner.io_error() {
+                self.state
+                    .record_stream_error(format!("DSD playback failed: {e}"));
+            }
+        }
+        item
+    }
+}
+
 /// Pick the MIME to advertise to an external renderer for a legacy stream-URL
 /// download. Prefer the server-provided `mime_type`; when it is empty (Qobuz
 /// can return `""`), fall back by format id so the renderer is never handed an
@@ -5167,6 +5217,60 @@ pub fn external_content_type(mime: &str, format_id: u32) -> String {
 mod tests {
     use super::compute_needs_new_stream;
     use super::external_content_type;
+    use super::{DsdErrorReport, SharedState};
+    use std::sync::Arc;
+
+    struct FakeDsdSource {
+        words: std::vec::IntoIter<i32>,
+        error: Option<String>,
+    }
+
+    impl Iterator for FakeDsdSource {
+        type Item = i32;
+        fn next(&mut self) -> Option<i32> {
+            self.words.next()
+        }
+    }
+
+    impl qbz_dsd::DsdWordSource for FakeDsdSource {
+        fn io_error(&self) -> Option<&str> {
+            self.error.as_deref()
+        }
+    }
+
+    #[test]
+    fn dsd_error_report_surfaces_io_error_at_stream_end() {
+        let state = Arc::new(SharedState::new());
+        let source = FakeDsdSource {
+            words: vec![1, 2].into_iter(),
+            error: Some("read failed".to_string()),
+        };
+        let mut wrapped = DsdErrorReport::new(source, state.clone());
+        assert_eq!(wrapped.next(), Some(1));
+        assert!(!state.has_stream_error());
+        assert_eq!(wrapped.next(), Some(2));
+        assert_eq!(wrapped.next(), None);
+        assert!(state.has_stream_error());
+        let msg = state.take_stream_error_message().unwrap();
+        assert!(msg.contains("read failed"));
+        // Repeated polls after exhaustion must not re-record the message.
+        assert_eq!(wrapped.next(), None);
+        assert_eq!(state.take_stream_error_message(), None);
+    }
+
+    #[test]
+    fn dsd_error_report_clean_eof_is_not_an_error() {
+        let state = Arc::new(SharedState::new());
+        let source = FakeDsdSource {
+            words: vec![1].into_iter(),
+            error: None,
+        };
+        let mut wrapped = DsdErrorReport::new(source, state.clone());
+        assert_eq!(wrapped.next(), Some(1));
+        assert_eq!(wrapped.next(), None);
+        assert!(!state.has_stream_error());
+        assert_eq!(state.take_stream_error_message(), None);
+    }
 
     #[test]
     fn no_stream_always_needs_new() {