From 68317f5a342578d8dc62c7e25c11a34501140775 Mon Sep 17 00:00:00 2001 From: Afonso Jorge Ramos Date: Sun, 12 Jul 2026 23:59:04 +0200 Subject: [PATCH] fix(dsd): treat demux I/O errors as sticky failures, not clean EOF Only Ok(0) ends the stream cleanly; mid-file read errors set io_error. --- crates/qbz-dsd/src/dop.rs | 78 ++++++++++++++++++- crates/qbz-dsd/src/lib.rs | 22 ++++++ crates/qbz-dsd/src/native.rs | 15 +++- crates/qbz-player/src/player/mod.rs | 112 +++++++++++++++++++++++++++- 4 files changed, 221 insertions(+), 6 deletions(-) diff --git a/crates/qbz-dsd/src/dop.rs b/crates/qbz-dsd/src/dop.rs index 9ca11611a..f688a750a 100644 --- a/crates/qbz-dsd/src/dop.rs +++ b/crates/qbz-dsd/src/dop.rs @@ -82,6 +82,8 @@ pub struct DopStream { buf: Vec, idx: usize, done: bool, + /// Set when demux I/O fails mid-stream (not clean EOF). + io_error: Option, } /// 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![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], + max_bytes_per_ch: usize, + ) -> Result { + 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")); + } +} diff --git a/crates/qbz-dsd/src/lib.rs b/crates/qbz-dsd/src/lib.rs index 75f73545b..089dc6acf 100644 --- a/crates/qbz-dsd/src/lib.rs +++ b/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` 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 + 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; diff --git a/crates/qbz-dsd/src/native.rs b/crates/qbz-dsd/src/native.rs index 54f85aaa8..6d05f5f91 100644 --- a/crates/qbz-dsd/src/native.rs +++ b/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; 2], done: bool, + /// Set when demux I/O fails mid-stream (not clean EOF). + io_error: Option, } 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() { diff --git a/crates/qbz-player/src/player/mod.rs b/crates/qbz-player/src/player/mod.rs index 37b78d7f5..3d71ce4d9 100644 --- a/crates/qbz-player/src/player/mod.rs +++ b/crates/qbz-player/src/player/mod.rs @@ -2591,7 +2591,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; @@ -2706,7 +2709,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; @@ -2759,7 +2765,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 + Send, >, @@ -2773,7 +2782,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 + Send, >, @@ -4901,6 +4913,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`, 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 { + inner: S, + state: Arc, + reported: bool, +} + +#[cfg_attr(not(target_os = "linux"), allow(dead_code))] +impl DsdErrorReport { + fn new(inner: S, state: Arc) -> Self { + Self { + inner, + state, + reported: false, + } + } +} + +impl Iterator for DsdErrorReport { + type Item = i32; + fn next(&mut self) -> Option { + 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 @@ -4922,6 +4972,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, + error: Option, + } + + impl Iterator for FakeDsdSource { + type Item = i32; + fn next(&mut self) -> Option { + 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() {