fix(qobuz): retry transient fetch failures with backoff instead of skipping (#467)

A single transient 5xx / timeout / reset on the next track's file/url or a CMAF
segment used to be terminal and skipped the track. Add a retry wrapper (3
attempts, exponential backoff + jitter) around get_file_url, the init segment,
and per-segment fetches. Retry only transient errors (network/timeout, 5xx, 429
honoring Retry-After); a real 404 is classified as TrackUnavailable and
propagates immediately to the bounded skip path.

cc

crates/qbz-qobuz/src/client.rs

@@ -2407,59 +2407,93 @@ impl QobuzClient {
         track_id: u64,
         quality: Quality,
     ) -> Result<TrackFileUrl> {
-        let (session_id, _infos) = self.ensure_cmaf_session().await?;
-
-        let timestamp = get_timestamp();
         let format_id = quality.id();
-        let sig = sign_file_url(track_id, format_id, timestamp);
-
         let url = endpoints::build_url(paths::FILE_URL);
 
-        let mut headers = self.authenticated_headers().await?;
-        headers.insert(
-            "X-Session-Id",
-            reqwest::header::HeaderValue::from_str(&session_id)
-                .map_err(|_| ApiError::ApiResponse("Invalid session ID format".into()))?,
-        );
+        // Retry transient failures (5xx / 429 / network blips) with backoff so
+        // a momentary hiccup on the next track's file/url is not turned into a
+        // queue skip. A real 404 → TrackUnavailable is terminal and returns
+        // immediately to the (bounded) skip path. Issue #467.
+        crate::retry::retry_transient(
+            crate::retry::DEFAULT_MAX_ATTEMPTS,
+            "CMAF file/url",
+            ApiError::is_transient,
+            |_attempt| {
+                let url = url.clone();
+                async move {
+                    let (session_id, _infos) = self.ensure_cmaf_session().await?;
 
-        let response = self
-            .http
-            .get(&url)
-            .headers(headers)
-            .query(&[
-                ("track_id", track_id.to_string()),
-                ("format_id", format_id.to_string()),
-                ("intent", "stream".to_string()),
-                ("request_ts", timestamp.to_string()),
-                ("request_sig", sig),
-            ])
-            .send()
-            .await?;
+                    // Fresh timestamp + signature per attempt — the request
+                    // signature is time-bound and would expire across retries.
+                    let timestamp = get_timestamp();
+                    let sig = sign_file_url(track_id, format_id, timestamp);
 
-        let status = response.status();
-        log::info!(
-            "[CMAF] file/url track_id={} format_id={} status={}",
-            track_id,
-            format_id,
-            status
-        );
+                    let mut headers = self.authenticated_headers().await?;
+                    headers.insert(
+                        "X-Session-Id",
+                        reqwest::header::HeaderValue::from_str(&session_id).map_err(|_| {
+                            ApiError::ApiResponse("Invalid session ID format".into())
+                        })?,
+                    );
 
-        if !status.is_success() {
-            return Err(ApiError::ApiResponse(format!(
-                "file/url failed with status {}",
-                status
-            )));
-        }
+                    let response = self
+                        .http
+                        .get(&url)
+                        .headers(headers)
+                        .query(&[
+                            ("track_id", track_id.to_string()),
+                            ("format_id", format_id.to_string()),
+                            ("intent", "stream".to_string()),
+                            ("request_ts", timestamp.to_string()),
+                            ("request_sig", sig),
+                        ])
+                        .send()
+                        .await?;
 
-        let file_url: TrackFileUrl = response.json().await?;
-        log::info!(
-            "[CMAF] file/url result: segments={}, mime={:?}, sampling_rate={:?}",
-            file_url.n_segments,
-            file_url.mime_type,
-            file_url.sampling_rate
-        );
+                    let status = response.status();
+                    log::info!(
+                        "[CMAF] file/url track_id={} format_id={} status={}",
+                        track_id,
+                        format_id,
+                        status
+                    );
 
-        Ok(file_url)
+                    if !status.is_success() {
+                        let code = status.as_u16();
+                        return Err(if code == 404 {
+                            ApiError::TrackUnavailable(track_id)
+                        } else if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
+                            // Honor Retry-After (seconds) when the server sends it.
+                            let retry_after = response
+                                .headers()
+                                .get(reqwest::header::RETRY_AFTER)
+                                .and_then(|v| v.to_str().ok())
+                                .and_then(|s| s.trim().parse::<u64>().ok())
+                                .unwrap_or(2);
+                            ApiError::RateLimited(retry_after)
+                        } else if status.is_server_error() {
+                            ApiError::ServerError(code)
+                        } else {
+                            ApiError::ApiResponse(format!(
+                                "file/url failed with status {}",
+                                status
+                            ))
+                        });
+                    }
+
+                    let file_url: TrackFileUrl = response.json().await?;
+                    log::info!(
+                        "[CMAF] file/url result: segments={}, mime={:?}, sampling_rate={:?}",
+                        file_url.n_segments,
+                        file_url.mime_type,
+                        file_url.sampling_rate
+                    );
+
+                    Ok(file_url)
+                }
+            },
+        )
+        .await
     }
 }
 

crates/qbz-qobuz/src/cmaf.rs

@@ -144,15 +144,9 @@ pub async fn setup_streaming(
     let init_start = std::time::Instant::now();
 
     log::info!("[CMAF] Fetching init segment for track {}", track_id);
-    let init_data = http
-        .get(&init_url)
-        .header("User-Agent", "Mozilla/5.0")
-        .send()
+    let init_data = fetch_bytes_with_retry(&http, &init_url, "CMAF init")
         .await
-        .map_err(|e| format!("Failed to fetch init segment: {}", e))?
-        .bytes()
-        .await
-        .map_err(|e| format!("Failed to read init segment: {}", e))?;
+        .map_err(|e| format!("Failed to fetch init segment: {}", e))?;
 
     let init_fetch_ms = init_start.elapsed().as_millis() as u64;
 
@@ -350,6 +344,44 @@ fn build_cdn_client() -> std::result::Result<reqwest::Client, String> {
         .map_err(|e| format!("CMAF client error: {}", e))
 }
 
+/// Fetch a CDN URL into bytes, retrying transient failures (network blips,
+/// 5xx, 429) with exponential backoff. A terminal status (404/403) fails
+/// immediately. Without this a single transient segment failure aborted the
+/// whole track download and the frontend skipped it — issue #467.
+async fn fetch_bytes_with_retry(
+    http: &reqwest::Client,
+    url: &str,
+    log_tag: &str,
+) -> std::result::Result<Vec<u8>, String> {
+    use crate::retry::{
+        classify_reqwest, classify_status, retry_transient, FetchError, DEFAULT_MAX_ATTEMPTS,
+    };
+    retry_transient(
+        DEFAULT_MAX_ATTEMPTS,
+        log_tag,
+        FetchError::is_transient,
+        |_attempt| async move {
+            let response = http
+                .get(url)
+                .header("User-Agent", "Mozilla/5.0")
+                .send()
+                .await
+                .map_err(|e| classify_reqwest(&e, "fetch"))?;
+            let status = response.status();
+            if !status.is_success() {
+                return Err(classify_status(status, "fetch"));
+            }
+            response
+                .bytes()
+                .await
+                .map(|b| b.to_vec())
+                .map_err(|e| classify_reqwest(&e, "read"))
+        },
+    )
+    .await
+    .map_err(|e| e.to_string())
+}
+
 /// Fetch segments 1..=n_segments concurrently with a semaphore cap and a
 /// cooldown per slot to stay under CDN rate limits.
 ///
@@ -379,15 +411,10 @@ async fn fetch_all_segments(
 
         handles.push(tokio::spawn(async move {
             let permit = sem.acquire_owned().await.map_err(|e| format!("semaphore: {}", e))?;
-            let seg_data = http
-                .get(&seg_url)
-                .header("User-Agent", "Mozilla/5.0")
-                .send()
-                .await
-                .map_err(|e| format!("[{}] seg {} fetch: {}", log_tag, seg_idx, e))?
-                .bytes()
-                .await
-                .map_err(|e| format!("[{}] seg {} read: {}", log_tag, seg_idx, e))?;
+            let seg_data =
+                fetch_bytes_with_retry(&http, &seg_url, &format!("{} seg {}", log_tag, seg_idx))
+                    .await
+                    .map_err(|e| format!("[{}] seg {} fetch: {}", log_tag, seg_idx, e))?;
             let bytes_this_segment = seg_data.len() as u64;
             if let Some(cb) = progress {
                 let done = counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
@@ -401,7 +428,7 @@ async fn fetch_all_segments(
             // to stay under CDN rate limits (most use 1s windows)
             tokio::time::sleep(std::time::Duration::from_millis(500)).await;
             drop(permit);
-            Ok::<(u8, Vec<u8>), String>((seg_idx, seg_data.to_vec()))
+            Ok::<(u8, Vec<u8>), String>((seg_idx, seg_data))
         }));
     }
 

crates/qbz-qobuz/src/error.rs

@@ -42,6 +42,24 @@ pub enum ApiError {
 
     #[error("Rate limited, retry after {0} seconds")]
     RateLimited(u64),
+
+    #[error("Server error (HTTP {0})")]
+    ServerError(u16),
+}
+
+impl ApiError {
+    /// True for errors worth retrying with backoff (issue #467): transport
+    /// problems (timeout/connect/reset), 5xx server errors, and 429 rate
+    /// limiting. Terminal errors — a real 404 `TrackUnavailable`, auth or
+    /// parse failures — return false and should propagate to the (bounded)
+    /// skip path instead of being retried.
+    pub fn is_transient(&self) -> bool {
+        match self {
+            ApiError::NetworkError(e) => crate::retry::reqwest_is_transient(e),
+            ApiError::RateLimited(_) | ApiError::ServerError(_) => true,
+            _ => false,
+        }
+    }
 }
 
 pub type Result<T> = std::result::Result<T, ApiError>;

crates/qbz-qobuz/src/lib.rs

@@ -38,6 +38,7 @@ pub mod endpoints;
 pub mod error;
 pub mod link_resolver;
 pub mod performers;
+pub mod retry;
 
 // Re-export main types
 pub use bundle::BundleTokens;

crates/qbz-qobuz/src/retry.rs

@@ -0,0 +1,198 @@
+//! Transient-error retry helper for the streaming fetch path.
+//!
+//! A single transient network blip (5xx, timeout, connection reset, 429)
+//! on the next track's `file/url` or a CMAF segment used to be terminal:
+//! the play failed and the frontend auto-skipped the track, walking the
+//! queue on a run of blips. Long Hi-Res tracks (Dream Theater / Yes) are
+//! the worst case because their large downloads keep the link busy for
+//! minutes, raising the odds of catching a blip. See issue #467.
+//!
+//! This module retries *transient* failures with exponential backoff while
+//! letting *terminal* failures (a real 404 "gone forever", auth errors)
+//! propagate immediately to the (now-bounded) skip path.
+
+use std::future::Future;
+use std::time::Duration;
+
+/// Number of attempts: 1 initial try + 2 retries.
+pub const DEFAULT_MAX_ATTEMPTS: u32 = 3;
+
+/// A fetch error tagged with whether it is worth retrying. Used by the CMAF
+/// CDN fetch path, whose underlying errors are otherwise stringly-typed.
+#[derive(Debug)]
+pub enum FetchError {
+    /// Worth retrying: network/timeout/connect/body error, 5xx, or 429.
+    Transient(String),
+    /// Not worth retrying: 4xx (other than 429), or a definitive failure.
+    Terminal(String),
+}
+
+impl FetchError {
+    pub fn is_transient(&self) -> bool {
+        matches!(self, FetchError::Transient(_))
+    }
+}
+
+impl std::fmt::Display for FetchError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            FetchError::Transient(s) | FetchError::Terminal(s) => write!(f, "{}", s),
+        }
+    }
+}
+
+/// True for reqwest errors worth retrying: timeout, connect, request-build,
+/// and body/decode errors are all transient transport-level problems.
+pub fn reqwest_is_transient(e: &reqwest::Error) -> bool {
+    e.is_timeout() || e.is_connect() || e.is_request() || e.is_body()
+}
+
+/// Classify a reqwest error into a `FetchError`. All reqwest transport errors
+/// are treated as transient — a definitive "gone" answer comes back as a 404
+/// *status*, not a transport error.
+pub fn classify_reqwest(e: &reqwest::Error, context: &str) -> FetchError {
+    FetchError::Transient(format!("{}: {}", context, e))
+}
+
+/// Classify a non-success HTTP status into a `FetchError`. 5xx and 429 are
+/// transient; everything else (404, 403, ...) is terminal.
+pub fn classify_status(status: reqwest::StatusCode, context: &str) -> FetchError {
+    let msg = format!("{}: HTTP {}", context, status);
+    if status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS {
+        FetchError::Transient(msg)
+    } else {
+        FetchError::Terminal(msg)
+    }
+}
+
+/// Exponential backoff with jitter for the given 1-based attempt:
+/// ~250 ms, ~500 ms, ~1 s, capped at 2 s, plus up to +25% jitter. Jitter is
+/// derived from the wall clock to avoid pulling in a `rand` dependency.
+fn backoff_delay(attempt: u32) -> Duration {
+    let exp = attempt.saturating_sub(1).min(3);
+    let base_ms = 250u64.saturating_mul(1u64 << exp).min(2000);
+    let jitter_span = base_ms / 4;
+    let jitter = if jitter_span > 0 {
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.subsec_nanos() as u64)
+            .unwrap_or(0);
+        nanos % (jitter_span + 1)
+    } else {
+        0
+    };
+    Duration::from_millis(base_ms + jitter)
+}
+
+/// Run `op` (which takes the 1-based attempt number) and retry while it
+/// returns a transient error, sleeping with exponential backoff between
+/// attempts. Terminal errors and the final attempt return immediately.
+pub async fn retry_transient<F, Fut, T, E>(
+    max_attempts: u32,
+    log_tag: &str,
+    is_transient: impl Fn(&E) -> bool,
+    mut op: F,
+) -> std::result::Result<T, E>
+where
+    F: FnMut(u32) -> Fut,
+    Fut: Future<Output = std::result::Result<T, E>>,
+    E: std::fmt::Display,
+{
+    let mut attempt = 1;
+    loop {
+        match op(attempt).await {
+            Ok(value) => return Ok(value),
+            Err(err) => {
+                if attempt >= max_attempts || !is_transient(&err) {
+                    return Err(err);
+                }
+                let delay = backoff_delay(attempt);
+                log::warn!(
+                    "[{}] transient error on attempt {}/{}: {} — retrying in {}ms",
+                    log_tag,
+                    attempt,
+                    max_attempts,
+                    err,
+                    delay.as_millis()
+                );
+                tokio::time::sleep(delay).await;
+                attempt += 1;
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn succeeds_first_try() {
+        let calls = Arc::new(AtomicU32::new(0));
+        let c = calls.clone();
+        let r: Result<u32, FetchError> = retry_transient(3, "test", FetchError::is_transient, |_| {
+            let c = c.clone();
+            async move {
+                c.fetch_add(1, Ordering::Relaxed);
+                Ok(42)
+            }
+        })
+        .await;
+        assert_eq!(r.unwrap(), 42);
+        assert_eq!(calls.load(Ordering::Relaxed), 1);
+    }
+
+    #[tokio::test]
+    async fn retries_transient_then_succeeds() {
+        let calls = Arc::new(AtomicU32::new(0));
+        let c = calls.clone();
+        let r: Result<u32, FetchError> = retry_transient(3, "test", FetchError::is_transient, |attempt| {
+            let c = c.clone();
+            async move {
+                c.fetch_add(1, Ordering::Relaxed);
+                if attempt < 3 {
+                    Err(FetchError::Transient("503".into()))
+                } else {
+                    Ok(7)
+                }
+            }
+        })
+        .await;
+        assert_eq!(r.unwrap(), 7);
+        assert_eq!(calls.load(Ordering::Relaxed), 3);
+    }
+
+    #[tokio::test]
+    async fn terminal_does_not_retry() {
+        let calls = Arc::new(AtomicU32::new(0));
+        let c = calls.clone();
+        let r: Result<u32, FetchError> = retry_transient(3, "test", FetchError::is_transient, |_| {
+            let c = c.clone();
+            async move {
+                c.fetch_add(1, Ordering::Relaxed);
+                Err(FetchError::Terminal("404".into()))
+            }
+        })
+        .await;
+        assert!(r.is_err());
+        assert_eq!(calls.load(Ordering::Relaxed), 1);
+    }
+
+    #[tokio::test]
+    async fn gives_up_after_max_attempts() {
+        let calls = Arc::new(AtomicU32::new(0));
+        let c = calls.clone();
+        let r: Result<u32, FetchError> = retry_transient(3, "test", FetchError::is_transient, |_| {
+            let c = c.clone();
+            async move {
+                c.fetch_add(1, Ordering::Relaxed);
+                Err(FetchError::Transient("timeout".into()))
+            }
+        })
+        .await;
+        assert!(r.is_err());
+        assert_eq!(calls.load(Ordering::Relaxed), 3);
+    }
+}