fix(player): surface stream feeder failures to BufferWriter

CMAF and remote download tasks now error the streaming buffer on failure
so readers do not hang waiting for complete/error forever.
This commit is contained in:
Afonso Jorge Ramos committed 2026-07-13 16:16:14 +02:00
1 parent a9949b8cf2
commit b4450965df
3 files changed
+118 -6

No files matched your search

+41 -5
View File
@@ -2349,13 +2349,22 @@ impl Player {
};
while !buffer_sufficient(&source)
&& source.download_error().is_none()
&& start_wait.elapsed() < max_wait
{
std::thread::sleep(Duration::from_millis(50));
}
if !source.has_min_buffer() {
log::error!("Streaming: timeout waiting for initial buffer");
match source.download_error() {
Some(err) => log::error!(
"Streaming: feeder failed before initial buffer: {}",
err
),
None => log::error!(
"Streaming: timeout waiting for initial buffer"
),
}
return;
}
if resume_buffer_target > 0
@@ -4274,6 +4283,24 @@ impl Player {
cache: Arc<qbz_cache::AudioCache>,
skip_cache: bool,
) -> Result<(), String> {
struct FailGuard {
writer: BufferWriter,
armed: bool,
}
impl Drop for FailGuard {
fn drop(&mut self) {
if self.armed {
let _ = self
.writer
.error("CMAF stream aborted before completion".into());
}
}
}
let mut guard = FailGuard {
writer,
armed: true,
};
let writer = &guard.writer;
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(10))
.build()
@@ -4281,7 +4308,9 @@ impl Player {
// Write the FLAC header first so the decoder can identify the format.
if let Err(e) = writer.push_chunk(&flac_header) {
return Err(format!("Failed to write FLAC header to buffer: {}", e));
let msg = format!("Failed to write FLAC header to buffer: {e}");
let _ = writer.error(msg.clone());
return Err(msg);
}
let mut total_written: u64 = flac_header.len() as u64;
@@ -4323,7 +4352,9 @@ impl Player {
}
// Write decrypted frame to the streaming buffer.
if let Err(e) = writer.push_chunk(&frame) {
log::error!("[CMAF-STREAM] Failed to push frame: {}", e);
let msg = format!("Failed to push frame: {e}");
let _ = writer.error(msg.clone());
return Err(msg);
}
if !skip_cache {
cache_data.extend_from_slice(&frame);
@@ -4336,7 +4367,9 @@ impl Player {
if data_pos < crypto.mdat_end && crypto.mdat_end <= seg_data.len() {
let trailing = &seg_data[data_pos..crypto.mdat_end];
if let Err(e) = writer.push_chunk(trailing) {
log::error!("[CMAF-STREAM] Failed to push trailing data: {}", e);
let msg = format!("Failed to push trailing data: {e}");
let _ = writer.error(msg.clone());
return Err(msg);
}
if !skip_cache {
cache_data.extend_from_slice(trailing);
@@ -4361,9 +4394,12 @@ impl Player {
}
}
// Signal end of stream.
// Signal end of stream (disarm fail-guard so Drop does not error).
guard.armed = false;
if let Err(e) = writer.complete() {
log::error!("[CMAF-STREAM] Failed to mark buffer complete: {}", e);
let _ = writer.error(format!("Failed to mark buffer complete: {e}"));
return Err(format!("Failed to mark buffer complete: {e}"));
}
log::info!(
@@ -352,6 +352,14 @@ impl BufferedMediaSource {
false
}
}
/// Error reported by the feeder, if any. Lets waiters (the initial
/// buffer fill loop) bail out immediately instead of sitting through
/// the full buffer timeout when the feeder has already died.
pub fn download_error(&self) -> Option<String> {
let (lock, _) = &*self.state;
lock.lock().ok().and_then(|s| s.download_error.clone())
}
}
impl Read for BufferedMediaSource {
@@ -518,11 +526,16 @@ impl BufferWriter {
/// Mark download as failed
///
/// After this is called, readers will receive the error on next read.
/// The first recorded error wins: it is the root cause, and the feeder
/// fail-guards fire a generic "aborted" error on drop after a specific
/// failure has already been recorded, which must not overwrite it.
pub fn error(&self, err: String) -> Result<(), String> {
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| "Failed to acquire buffer lock")?;
state.download_error = Some(err);
if state.download_error.is_none() {
state.download_error = Some(err);
}
cvar.notify_all();
Ok(())
@@ -991,6 +1004,28 @@ mod tests {
use std::thread;
use std::time::Duration;
#[test]
fn first_error_wins_over_later_generic_abort() {
let (source, writer) = BufferedMediaSource::new(StreamingConfig::from_seconds(1), None);
writer.error("root cause".to_string()).unwrap();
writer
.error("CMAF stream aborted before completion".to_string())
.unwrap();
assert_eq!(source.download_error().as_deref(), Some("root cause"));
}
#[test]
fn feeder_error_is_visible_to_waiters_before_min_buffer() {
let (source, writer) = BufferedMediaSource::new(StreamingConfig::from_seconds(1), None);
assert!(!source.has_min_buffer());
assert!(source.download_error().is_none());
writer.error("feeder died".to_string()).unwrap();
// The initial-buffer wait loop polls this instead of sleeping out
// the full buffer timeout.
assert_eq!(source.download_error().as_deref(), Some("feeder died"));
assert!(!source.has_min_buffer());
}
#[test]
fn raw_initial_buffer_for_speed_follows_documented_ladder() {
// Each band of the documented speed ladder produces its own size.
@@ -1136,4 +1171,20 @@ mod tests {
assert_eq!(n, 7);
assert_eq!(&buf, b"Delayed");
}
#[test]
fn error_unblocks_reader_with_io_error() {
use std::io::ErrorKind;
let config = StreamingConfig {
initial_buffer_bytes: 5,
max_buffer_bytes: 100,
};
let (mut source, writer) = BufferedMediaSource::new(config, None);
writer.error("cdn failed".into()).unwrap();
let mut buf = [0u8; 8];
let err = source.read(&mut buf).unwrap_err();
assert_eq!(err.kind(), ErrorKind::Other);
let msg = err.to_string();
assert!(msg.contains("cdn failed"), "{msg}");
}
}