fix(playback): recover stalled streams

cx

crates/qbz-app/src/playback_driver.rs

@@ -86,6 +86,8 @@ pub struct LastTick {
     pub position: u64,
     pub duration: u64,
     pub is_playing: bool,
+    pub engine_empty_generation: u64,
+    pub source_failure_generation: u64,
 }
 
 impl LastTick {
@@ -95,6 +97,8 @@ impl LastTick {
             position: ev.position,
             duration: ev.duration,
             is_playing: ev.is_playing,
+            engine_empty_generation: ev.engine_empty_generation,
+            source_failure_generation: ev.source_failure_generation,
         }
     }
 }
@@ -223,17 +227,25 @@ pub fn plan_tick(
         }
     }
 
-    // 5. End-of-track edge (playback.rs:4489-4496): the previous tick was
+    // 5. End-of-track/recovery edge. The engine and CMAF feeder now publish
-    //    playing, this tick is not, the track id held (or went to 0), and the
+    //    durable generations because a complete playing -> stopped pulse can
-    //    previous position was within 2 s of the (current) duration. Uses the
+    //    fit between two 450 ms ticks. Keep the sampled predicate as a fallback
-    //    live `ev.duration` guard + previous `last.position` exactly as the
+    //    for non-CMAF sources, but never require it for an explicit edge.
-    //    desktop's `duration > 0 && seen_position + 2 >= duration`.
+    let engine_empty = ev.engine_empty_generation != last.engine_empty_generation
-    let track_ended = last.is_playing
+        && ev.engine_empty_track_id != 0
+        && ev.engine_empty_track_id == queue.current
+        && !ev.is_playing
+        && ev.gapless_next_track_id == 0;
+    let source_failed = ev.source_failure_generation != last.source_failure_generation
+        && ev.source_failure_track_id != 0
+        && ev.source_failure_track_id == queue.current;
+    let sampled_track_end = last.is_playing
         && !ev.is_playing
         && last.track_id != 0
         && (ev.track_id == 0 || ev.track_id == last.track_id)
         && ev.duration > 0
         && last.position + 2 >= ev.duration;
+    let track_ended = source_failed || engine_empty || sampled_track_end;
 
     // 6. QConnect report edge (playback.rs:4648-4673): report on a track/play
     //    transition OR the ~2 s periodic cadence while playing. Runs regardless
@@ -327,6 +339,8 @@ pub fn advance_state(
             position: prev.last.position,
             duration: prev.last.duration,
             is_playing: ev.is_playing,
+            engine_empty_generation: ev.engine_empty_generation,
+            source_failure_generation: ev.source_failure_generation,
         }
     };
     let mut gapless_requested_for = if armed {
@@ -338,7 +352,15 @@ pub fn advance_state(
     // Track-end handler resets the edge trackers + the gapless guard
     // (playback.rs:4728-4742, both the stop-after and advance branches).
     if ended {
-        last = LastTick::default();
+        // Keep the durable player generations consumed while resetting the
+        // sampled edge fields. Resetting the generations to zero would replay
+        // the same engine-empty/source-failure edge on every tick until the
+        // next track surfaced.
+        last = LastTick {
+            engine_empty_generation: ev.engine_empty_generation,
+            source_failure_generation: ev.source_failure_generation,
+            ..LastTick::default()
+        };
         gapless_requested_for = 0;
     }
 
@@ -920,6 +942,10 @@ mod tests {
             gapless_next_track_id: 0,
             bit_perfect_mode: None,
             buffer_progress: None,
+            engine_empty_generation: 0,
+            engine_empty_track_id: 0,
+            source_failure_generation: 0,
+            source_failure_track_id: 0,
         }
     }
 
@@ -948,6 +974,87 @@ mod tests {
         assert!(a.contains(&DriverAction::ReportEdge)); // play-state edge
     }
 
+    #[test]
+    fn engine_empty_generation_advances_when_playing_pulse_was_missed() {
+        // Both sampled ticks are stopped. This is the observed 2026-08-28
+        // failure: a late gapless resume and engine-empty happened inside one
+        // polling interval, so `last.is_playing` never became true.
+        let s = DriverState::after(&ev(1, false, 580, 581));
+        let mut current = ev(1, false, 581, 581);
+        current.engine_empty_generation = 1;
+        current.engine_empty_track_id = 1;
+
+        let actions = plan_tick(
+            &s,
+            &current,
+            &q(1, &[(2, true)], "off", None),
+            None,
+        );
+        assert!(actions.contains(&DriverAction::AdvanceAndPlay));
+    }
+
+    #[test]
+    fn source_failure_advances_even_when_track_never_started() {
+        let s = DriverState::after(&ev(0, false, 0, 0));
+        let mut current = ev(0, false, 0, 0);
+        current.source_failure_generation = 1;
+        current.source_failure_track_id = 1;
+
+        let actions = plan_tick(
+            &s,
+            &current,
+            &q(1, &[(2, true)], "off", None),
+            Some("CMAF segment retries exhausted"),
+        );
+        assert!(actions.contains(&DriverAction::LatchError(
+            "CMAF segment retries exhausted".to_string()
+        )));
+        assert!(actions.contains(&DriverAction::AdvanceAndPlay));
+    }
+
+    #[test]
+    fn stale_source_failure_cannot_skip_the_new_queue_track() {
+        let s = DriverState::after(&ev(2, false, 0, 581));
+        let mut current = ev(2, false, 0, 581);
+        current.source_failure_generation = 1;
+        current.source_failure_track_id = 1;
+
+        let actions = plan_tick(&s, &current, &q(2, &[(3, true)], "off", None), None);
+        assert!(!actions.contains(&DriverAction::AdvanceAndPlay));
+    }
+
+    #[test]
+    fn consumed_engine_empty_generation_does_not_advance_twice() {
+        let s = DriverState::after(&ev(1, false, 580, 581));
+        let mut current = ev(1, false, 581, 581);
+        current.engine_empty_generation = 1;
+        current.engine_empty_track_id = 1;
+        let queue = q(1, &[(2, true)], "off", None);
+        let first = plan_tick(&s, &current, &queue, None);
+        assert!(first.contains(&DriverAction::AdvanceAndPlay));
+
+        let advanced = advance_state(&s, &current, &first);
+        let second = plan_tick(&advanced, &current, &queue, None);
+        assert!(!second.contains(&DriverAction::AdvanceAndPlay));
+    }
+
+    #[test]
+    fn late_gapless_rescue_consumes_engine_empty_without_advancing() {
+        let s = DriverState::after(&ev(1, false, 580, 581));
+        let mut current = ev(1, true, 581, 581);
+        current.engine_empty_generation = 1;
+        current.engine_empty_track_id = 1;
+        current.gapless_next_track_id = 2;
+
+        let queue = q(1, &[(2, true)], "off", None);
+        let first = plan_tick(&s, &current, &queue, None);
+        assert!(!first.contains(&DriverAction::AdvanceAndPlay));
+
+        let advanced = advance_state(&s, &current, &first);
+        let second = plan_tick(&advanced, &current, &queue, None);
+        assert!(!second.contains(&DriverAction::AdvanceAndPlay));
+    }
+
     #[test]
     fn mid_track_pause_does_not_advance() {
         let s = DriverState::after(&ev(1, true, 100, 581));

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

@@ -1064,6 +1064,21 @@ pub struct PlaybackEvent {
     /// the track is fully buffered — drives the seek-bar cache overlay.
     #[serde(default)]
     pub buffer_progress: Option<f32>,
+    /// Monotonic edge raised when the audio engine consumes every queued
+    /// source. Pollers compare generations instead of trying to sample a
+    /// potentially sub-tick `playing -> stopped` transition.
+    #[serde(default)]
+    pub engine_empty_generation: u64,
+    /// Track whose engine-empty edge raised `engine_empty_generation`.
+    #[serde(default)]
+    pub engine_empty_track_id: u64,
+    /// Monotonic edge raised after the live CMAF feeder exhausts its bounded
+    /// fetch retries. This also covers failure before playback ever starts.
+    #[serde(default)]
+    pub source_failure_generation: u64,
+    /// Track whose feeder raised `source_failure_generation`.
+    #[serde(default)]
+    pub source_failure_track_id: u64,
 }
 
 /// Shared state between main thread and audio thread
@@ -1120,6 +1135,16 @@ pub struct SharedState {
     /// can detect that a queued `PlayStreaming` was superseded by a newer play
     /// and stop waiting on its initial buffer instead of blocking ~60s (#591).
     play_generation: Arc<AtomicU64>,
+    /// Monotonic engine-empty notification plus the track that raised it.
+    /// This is durable across polling intervals; a boolean pulse would have
+    /// the same sampling race as the old `was_playing` predicate.
+    engine_empty_generation: Arc<AtomicU64>,
+    engine_empty_track_id: Arc<AtomicU64>,
+    /// Monotonic terminal source-data failure notification. Kept separate
+    /// from `stream_error`, which also represents output-device failures that
+    /// must never cause the queue to skip a healthy track.
+    source_failure_generation: Arc<AtomicU64>,
+    source_failure_track_id: Arc<AtomicU64>,
     /// Handle of the live CMAF segment feeder — the background task that
     /// keeps downloading the currently-streaming track after playback has
     /// started. Stored so a superseding play intent can stop the download
@@ -1171,6 +1196,10 @@ impl SharedState {
             buffer_progress: Arc::new(AtomicU32::new(0)),
             bit_perfect_mode: Arc::new(AtomicU8::new(0)),
             play_generation: Arc::new(AtomicU64::new(0)),
+            engine_empty_generation: Arc::new(AtomicU64::new(0)),
+            engine_empty_track_id: Arc::new(AtomicU64::new(0)),
+            source_failure_generation: Arc::new(AtomicU64::new(0)),
+            source_failure_track_id: Arc::new(AtomicU64::new(0)),
             stream_feeder: Arc::new(std::sync::Mutex::new(None)),
             gapless_stream_feeder: Arc::new(std::sync::Mutex::new(None)),
         }
@@ -1246,6 +1275,26 @@ impl SharedState {
         self.current_play_generation() == gen
     }
 
+    /// Publish a durable engine-empty edge for frontend/headless pollers.
+    /// Store the identity first, then advance the generation so a reader that
+    /// observes the new generation also observes its matching track id.
+    fn record_engine_empty(&self, track_id: u64) {
+        self.engine_empty_track_id.store(track_id, Ordering::SeqCst);
+        self.engine_empty_generation.fetch_add(1, Ordering::SeqCst);
+    }
+
+    /// Publish a terminal CMAF feeder failure only while its play intent is
+    /// still current. A superseded feeder is cancellation, not a bad track.
+    fn record_source_failure(&self, track_id: u64, play_gen: u64) {
+        if !self.is_current_play(play_gen) {
+            return;
+        }
+        self.source_failure_track_id
+            .store(track_id, Ordering::SeqCst);
+        self.source_failure_generation
+            .fetch_add(1, Ordering::SeqCst);
+    }
+
     /// Cancel the in-flight segment feeder's download, if any. Called on
     /// every new play intent (and on stop) via `Player::begin_play`.
     ///
@@ -4542,6 +4591,9 @@ impl Player {
                                         && thread_state.is_playing.load(Ordering::SeqCst)
                                     {
                                         log::info!("Audio thread: track finished (engine empty)");
+                                        let ended_track_id = thread_state
+                                            .current_track_id
+                                            .load(Ordering::SeqCst);
                                         thread_state.is_playing.store(false, Ordering::SeqCst);
                                         let duration = thread_state.duration.load(Ordering::SeqCst);
                                         thread_state.position.store(duration, Ordering::SeqCst);
@@ -4557,6 +4609,11 @@ impl Player {
                                         log::info!("[gapless-trace] clear (engine-empty) pending was {:?}", gapless_pending.as_ref().map(|p| p.track_id));
                                         gapless_pending = None;
                                         gapless_request_armed = false;
+                                        // Publish only after every stopped/end
+                                        // field above is coherent. The
+                                        // generation is the release edge the
+                                        // pollers key on.
+                                        thread_state.record_engine_empty(ended_track_id);
                                     }
                                 }
                             }
@@ -4850,6 +4907,7 @@ impl Player {
                 let flac_header = cmaf_info.flac_header;
                 let n_segments = cmaf_info.n_segments;
                 let cache = self.audio_cache.clone();
+                let feeder_state = self.state.clone();
                 let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
                 let slot_writer = buffer_writer.clone();
 
@@ -4875,11 +4933,14 @@ impl Player {
                             track_id
                         ),
                         Ok(false) => {}
-                        Err(e) => log::error!(
+                        Err(e) => {
-                            "[CMAF-STREAM ERROR] Track {}: {}",
+                            feeder_state.record_source_failure(track_id, gen);
-                            track_id,
+                            log::error!(
-                            e
+                                "[CMAF-STREAM ERROR] Track {}: {} — queue recovery armed",
-                        ),
+                                track_id,
+                                e
+                            );
+                        }
                     }
                 });
                 // Register for cancel-on-supersede: the next `begin_play`
@@ -5472,17 +5533,12 @@ impl Player {
 
         for seg_idx in 1..=n_segments {
             let seg_url = url_template.replace("$SEGMENT$", &seg_idx.to_string());
-            let fetch = async {
+            let fetch_tag = format!("CMAF-STREAM track {track_id} seg {seg_idx}");
-                client
+            let fetch = qbz_qobuz::cmaf::fetch_cdn_bytes_with_retry(
-                    .get(&seg_url)
+                &client,
-                    .header("User-Agent", "Mozilla/5.0")
+                &seg_url,
-                    .send()
+                &fetch_tag,
-                    .await
+            );
-                    .map_err(|e| format!("CMAF segment {} fetch: {}", seg_idx, e))?
-                    .bytes()
-                    .await
-                    .map_err(|e| format!("CMAF segment {} read: {}", seg_idx, e))
-            };
             let seg_data = tokio::select! {
                 biased;
                 // Superseded by a newer play (or stop): drop the in-flight
@@ -6317,6 +6373,16 @@ impl Player {
             gapless_next_track_id: self.state.get_gapless_next_track_id(),
             bit_perfect_mode: self.state.get_bit_perfect_mode(),
             buffer_progress: self.state.get_buffer_progress(),
+            engine_empty_generation: self
+                .state
+                .engine_empty_generation
+                .load(Ordering::SeqCst),
+            engine_empty_track_id: self.state.engine_empty_track_id.load(Ordering::SeqCst),
+            source_failure_generation: self
+                .state
+                .source_failure_generation
+                .load(Ordering::SeqCst),
+            source_failure_track_id: self.state.source_failure_track_id.load(Ordering::SeqCst),
         }
     }
 }
@@ -6394,6 +6460,7 @@ mod tests {
     use super::compute_needs_new_stream;
     use super::external_content_type;
     use super::{BufferedMediaSource, DsdErrorReport, SharedState, StreamingConfig};
+    use std::sync::atomic::Ordering;
 
     struct FakeDsdSource {
         words: std::vec::IntoIter<i32>,
@@ -6447,6 +6514,34 @@ mod tests {
         assert_eq!(state.take_stream_error_message(), None);
     }
 
+    #[test]
+    fn engine_empty_edges_are_monotonic_and_keep_the_track_identity() {
+        let state = SharedState::new();
+        state.record_engine_empty(41);
+        assert_eq!(state.engine_empty_generation.load(Ordering::SeqCst), 1);
+        assert_eq!(state.engine_empty_track_id.load(Ordering::SeqCst), 41);
+
+        state.record_engine_empty(42);
+        assert_eq!(state.engine_empty_generation.load(Ordering::SeqCst), 2);
+        assert_eq!(state.engine_empty_track_id.load(Ordering::SeqCst), 42);
+    }
+
+    #[test]
+    fn superseded_feeder_cannot_publish_a_source_failure() {
+        let state = SharedState::new();
+        let old_generation = state.current_play_generation();
+        state.begin_play();
+        state.record_source_failure(41, old_generation);
+
+        assert_eq!(state.source_failure_generation.load(Ordering::SeqCst), 0);
+        assert_eq!(state.source_failure_track_id.load(Ordering::SeqCst), 0);
+
+        let current_generation = state.current_play_generation();
+        state.record_source_failure(42, current_generation);
+        assert_eq!(state.source_failure_generation.load(Ordering::SeqCst), 1);
+        assert_eq!(state.source_failure_track_id.load(Ordering::SeqCst), 42);
+    }
+
     #[test]
     fn superseding_play_cancels_current_and_gapless_stream_feeders() {
         let state = SharedState::new();

crates/qbz-qobuz/src/cmaf.rs

@@ -34,6 +34,7 @@
 //!   playback time. This is the security-sensitive path.
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use qbz_models::{Quality, StreamQualityInfo};
 
@@ -45,6 +46,15 @@ use crate::error::Result;
 /// past ~5 parallel requests per client IP.
 pub const CMAF_PREFETCH_CONCURRENCY: usize = 3;
 
+/// Total deadline for one CMAF CDN fetch attempt, from connect through the
+/// complete response body. Qobuz audio segments are independently retryable;
+/// leaving a single body read unbounded can strand both the live feeder and
+/// the gapless successor fetch long after the audible buffer has run dry.
+///
+/// The retry layer makes up to three attempts, so this is deliberately an
+/// attempt deadline rather than a whole-track deadline.
+pub const CMAF_FETCH_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(15);
+
 /// Progress callback shape for the download helpers. Each call reports
 /// "k of n segments complete" with the bytes received for that segment,
 /// so the caller can emit UI progress events without knowing the CMAF
@@ -144,7 +154,7 @@ 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 = fetch_bytes_with_retry(&http, &init_url, "CMAF init")
+    let init_data = fetch_cdn_bytes_with_retry(&http, &init_url, "CMAF init")
         .await
         .map_err(|e| format!("Failed to fetch init segment: {}", e))?;
 
@@ -380,16 +390,31 @@ fn build_cdn_client() -> std::result::Result<reqwest::Client, String> {
 /// 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(
+pub async fn fetch_cdn_bytes_with_retry(
     http: &reqwest::Client,
     url: &str,
     log_tag: &str,
 ) -> std::result::Result<Vec<u8>, String> {
-    use crate::retry::{
+    fetch_cdn_bytes_with_retry_config(
-        classify_reqwest, classify_status, retry_transient, FetchError, DEFAULT_MAX_ATTEMPTS,
+        http,
-    };
+        url,
+        log_tag,
+        CMAF_FETCH_ATTEMPT_TIMEOUT,
+        crate::retry::DEFAULT_MAX_ATTEMPTS,
+    )
+    .await
+}
+
+async fn fetch_cdn_bytes_with_retry_config(
+    http: &reqwest::Client,
+    url: &str,
+    log_tag: &str,
+    attempt_timeout: Duration,
+    max_attempts: u32,
+) -> std::result::Result<Vec<u8>, String> {
+    use crate::retry::{classify_reqwest, classify_status, retry_transient, FetchError};
     retry_transient(
-        DEFAULT_MAX_ATTEMPTS,
+        max_attempts,
         log_tag,
         FetchError::is_transient,
         |_attempt| async move {
@@ -404,6 +429,10 @@ async fn fetch_bytes_with_retry(
             let response = http
                 .get(url)
                 .header("User-Agent", "Mozilla/5.0")
+                // reqwest applies this from the start of connect until the
+                // response body completes. The deadline is recreated on each
+                // retry, so one stalled segment cannot monopolize a feeder.
+                .timeout(attempt_timeout)
                 .send()
                 .await
                 .map_err(|e| classify_reqwest(&e, "fetch"))?;
@@ -446,7 +475,11 @@ async fn fetch_all_segments(
 ) -> std::result::Result<Vec<Vec<u8>>, String> {
     let semaphore = Arc::new(tokio::sync::Semaphore::new(CMAF_PREFETCH_CONCURRENCY));
     let seg_indices: Vec<u8> = (1..=n_segments).collect();
-    let mut handles = Vec::with_capacity(seg_indices.len());
+    // JoinSet owns every segment task: dropping this fetch (for example when
+    // engine-empty recovery advances the queue) aborts the children too.
+    // Plain JoinHandles detach on drop and would keep the abandoned gapless
+    // download competing with the replacement track.
+    let mut fetches = tokio::task::JoinSet::new();
 
     let completed_count = Arc::new(std::sync::atomic::AtomicU32::new(0));
 
@@ -458,12 +491,15 @@ async fn fetch_all_segments(
         let progress = on_progress.clone();
         let counter = completed_count.clone();
 
-        handles.push(tokio::spawn(async move {
+        fetches.spawn(async move {
             let permit = sem.acquire_owned().await.map_err(|e| format!("semaphore: {}", e))?;
-            let seg_data =
+            let seg_data = fetch_cdn_bytes_with_retry(
-                fetch_bytes_with_retry(&http, &seg_url, &format!("{} seg {}", log_tag, seg_idx))
+                &http,
-                    .await
+                &seg_url,
-                    .map_err(|e| format!("[{}] seg {} fetch: {}", log_tag, seg_idx, e))?;
+                &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;
@@ -478,14 +514,13 @@ async fn fetch_all_segments(
             tokio::time::sleep(std::time::Duration::from_millis(500)).await;
             drop(permit);
             Ok::<(u8, Vec<u8>), String>((seg_idx, seg_data))
-        }));
+        });
     }
 
     // Collect results in arrival order, then re-sort by segment index
-    let mut segments: Vec<(u8, Vec<u8>)> = Vec::with_capacity(handles.len());
+    let mut segments: Vec<(u8, Vec<u8>)> = Vec::with_capacity(n_segments as usize);
-    for handle in handles {
+    while let Some(result) = fetches.join_next().await {
-        let (idx, data) = handle
+        let (idx, data) = result
-            .await
             .map_err(|e| format!("[{}] task panic: {}", log_tag, e))?
             .map_err(|e| format!("[{}] download failed: {}", log_tag, e))?;
         segments.push((idx, data));
@@ -548,3 +583,120 @@ pub fn decrypt_segments_into(
 fn _type_assertions() {
     let _: fn() -> Result<()> = || Ok(());
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::{Arc, Once};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    fn install_tls_provider() {
+        static INSTALL: Once = Once::new();
+        INSTALL.call_once(|| {
+            let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+        });
+    }
+
+    #[tokio::test]
+    async fn stalled_segment_body_times_out_and_retries() {
+        install_tls_provider();
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .unwrap();
+        let address = listener.local_addr().unwrap();
+        let accepts = Arc::new(AtomicU32::new(0));
+        let server_accepts = accepts.clone();
+        let server = tokio::spawn(async move {
+            loop {
+                let Ok((mut socket, _)) = listener.accept().await else {
+                    return;
+                };
+                server_accepts.fetch_add(1, Ordering::SeqCst);
+                tokio::spawn(async move {
+                    let mut request = [0u8; 1024];
+                    let _ = socket.read(&mut request).await;
+                    let _ = socket
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\n",
+                        )
+                        .await;
+                    // Headers arrive immediately; the body never does within
+                    // the attempt deadline. This reproduces the 52 s body stall
+                    // from the 2026-08-28 session without a live CDN.
+                    tokio::time::sleep(Duration::from_millis(500)).await;
+                });
+            }
+        });
+
+        let client = reqwest::Client::builder().build().unwrap();
+        let url = format!("http://{address}/segment");
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            fetch_cdn_bytes_with_retry_config(
+                &client,
+                &url,
+                "CMAF timeout test",
+                Duration::from_millis(50),
+                2,
+            ),
+        )
+        .await
+        .expect("bounded segment attempts must not hang");
+
+        let error = result.expect_err("both stalled attempts must time out");
+        assert!(
+            error.contains("read:"),
+            "stalled response must fail while reading its body: {error}"
+        );
+        assert_eq!(accepts.load(Ordering::SeqCst), 2);
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn aborting_full_fetch_cancels_its_segment_tasks() {
+        install_tls_provider();
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .unwrap();
+        let address = listener.local_addr().unwrap();
+        let (headers_tx, headers_rx) = tokio::sync::oneshot::channel();
+        let (closed_tx, closed_rx) = tokio::sync::oneshot::channel();
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let mut request = [0u8; 1024];
+            let _ = socket.read(&mut request).await;
+            socket
+                .write_all(
+                    b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\n",
+                )
+                .await
+                .unwrap();
+            let _ = headers_tx.send(());
+
+            let mut trailing_request = [0u8; 1024];
+            while socket.read(&mut trailing_request).await.unwrap_or(0) != 0 {
+                // Drain any request bytes left after the server's first read.
+            }
+            let _ = closed_tx.send(());
+        });
+
+        let client = reqwest::Client::builder().build().unwrap();
+        let url_template = format!("http://{address}/segment-$SEGMENT$");
+        let fetch = tokio::spawn(async move {
+            fetch_all_segments(&client, &url_template, 1, "CMAF abort test", None).await
+        });
+        tokio::time::timeout(Duration::from_secs(1), headers_rx)
+            .await
+            .expect("segment request must reach the body stall")
+            .unwrap();
+
+        fetch.abort();
+        let _ = fetch.await;
+        tokio::time::timeout(Duration::from_secs(1), closed_rx)
+            .await
+            .expect("aborting the owner must close its in-flight segment")
+            .unwrap();
+        server.await.unwrap();
+    }
+}

crates/qbz-qt/src/playback_qt.rs

@@ -41,6 +41,7 @@ use cxx_qt_lib::QString;
 use qbz_app::shell::AppRuntime;
 use qbz_core::LoggingAdapter;
 use qbz_models::{PageArtistTrack, Quality, QueueTrack, RepeatMode};
+use qbz_player::PlaybackEvent;
 
 /// The request tier for every play. Persisted in ui_prefs.json ("streaming_quality")
 /// and applied here from Settings > Audio (settings_qt). Default "hires_plus".
@@ -1660,6 +1661,42 @@ fn stream_error_text(msg: &str, sandboxed: bool) -> String {
     }
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum PlaybackRecoveryCause {
+    SourceFailure,
+    EngineEmpty,
+    SampledNaturalEnd,
+}
+
+/// Resolve one queue-recovery edge. Explicit player generations win over the
+/// legacy sampled predicate: a 100 ms late-gapless resume/empty pulse can fit
+/// entirely between this frontend's 1 s polls, while the generations remain
+/// observable until the next tick. Track-id checks reject a stale feeder/end
+/// edge after a manual play has already moved the queue.
+fn playback_recovery_cause(
+    event: &PlaybackEvent,
+    queue_track_id: u64,
+    seen_engine_empty_generation: u64,
+    seen_source_failure_generation: u64,
+    sampled_track_end: bool,
+) -> Option<PlaybackRecoveryCause> {
+    if event.source_failure_generation != seen_source_failure_generation
+        && event.source_failure_track_id != 0
+        && event.source_failure_track_id == queue_track_id
+    {
+        return Some(PlaybackRecoveryCause::SourceFailure);
+    }
+    if event.engine_empty_generation != seen_engine_empty_generation
+        && event.engine_empty_track_id != 0
+        && event.engine_empty_track_id == queue_track_id
+        && !event.is_playing
+        && event.gapless_next_track_id == 0
+    {
+        return Some(PlaybackRecoveryCause::EngineEmpty);
+    }
+    sampled_track_end.then_some(PlaybackRecoveryCause::SampledNaturalEnd)
+}
+
 /// Play a pre-built queue starting at `start` (ArtistView Popular Tracks:
 /// the visible list becomes the queue, anchored at the clicked track).
 ///
@@ -3102,6 +3139,15 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
         // does not re-request it on every tick while the download is in flight
         // (playback.rs:5049-5051).
         let mut gapless_requested_for: u64 = 0;
+        // Own the frontend's gapless fetch so an engine-empty recovery can
+        // abort a segment body that is still monopolizing the link. The
+        // player's incremental successor feeder has its own cancellation slot;
+        // this handle covers the outer tier-walk task.
+        let mut gapless_fetch_task: Option<tokio::task::JoinHandle<()>> = None;
+        // Durable player edges consumed by this poller. Both begin at zero;
+        // the player generations are monotonic for the process lifetime.
+        let mut seen_engine_empty_generation: u64 = 0;
+        let mut seen_source_failure_generation: u64 = 0;
         // Immediate full-cache warming has a distinct edge because it follows
         // both the local engine and the cast cursor. Unlike the gapless guard,
         // it names the NEW current track whose two successors are being warmed.
@@ -3139,6 +3185,19 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
         loop {
             ticker.tick().await;
 
+            if gapless_fetch_task
+                .as_ref()
+                .is_some_and(|task| task.is_finished())
+            {
+                if let Some(task) = gapless_fetch_task.take() {
+                    if let Err(error) = task.await {
+                        if !error.is_cancelled() {
+                            log::warn!("[qbz-qt] gapless fetch task failed: {error}");
+                        }
+                    }
+                }
+            }
+
             // --- Stream-failure toast (PARITY-DEBT #7) -------------------
             // Surface audio-stream failures as a toast, 1:1 with
             // `playback.rs:5093-5115` (#508/#534/#500). The player records a
@@ -3420,6 +3479,14 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                 // hand-off, in which case the arming guard still names the
                 // track that just ended. Clear it so the NEW current track can
                 // arm its own prefetch (playback.rs:5353).
+                if let Some(task) = gapless_fetch_task.take() {
+                    if !task.is_finished() {
+                        log::info!(
+                            "[qbz-qt] [GAPLESS] aborting stale fetch after track changed to {track_id}"
+                        );
+                        task.abort();
+                    }
+                }
                 gapless_requested_for = 0;
             }
 
@@ -3504,7 +3571,7 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                         gapless_requested_for = track_id;
                         let runtime = runtime.clone();
                         let next_id = next.id;
-                        tokio::spawn(async move {
+                        let task = tokio::spawn(async move {
                             let quality = local_playback_quality().0;
                             let streaming_only =
                                 crate::settings_qt::audio_settings().streaming_only;
@@ -3599,6 +3666,9 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                                 }
                             }
                         });
+                        if let Some(stale) = gapless_fetch_task.replace(task) {
+                            stale.abort();
+                        }
                     } else if next.id != track_id && next.is_local {
                         // Source-owned gapless. `is_local` means "does not use
                         // Qobuz's tier walk", NOT "has a filesystem path": it
@@ -3608,7 +3678,7 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                         // DSD and remote streams into `play_next`.
                         gapless_requested_for = track_id;
                         let runtime = runtime.clone();
-                        tokio::spawn(async move {
+                        let task = tokio::spawn(async move {
                             let next_id = next.id;
                             match crate::audible_qt::queue_gapless_successor(
                                 &runtime, track_id, &next,
@@ -3626,6 +3696,9 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                                 ),
                             }
                         });
+                        if let Some(stale) = gapless_fetch_task.replace(task) {
+                            stale.abort();
+                        }
                     }
                 }
             }
@@ -3676,25 +3749,74 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
                 }
             }
 
-            // --- End-of-track edge (playback.rs condition, verbatim) ------
+            // --- End-of-track / source-failure recovery -------------------
-            let track_ended = was_playing
+            // Keep the sampled predicate for source types that do not publish
+            // a stronger signal. CMAF and engine-empty recovery use durable
+            // generations, so a playing pulse shorter than this 1 s poll is
+            // still observed exactly once.
+            let sampled_track_end = was_playing
                 && !is_playing
                 && last_track_id != 0
                 && (track_id == 0 || track_id == last_track_id)
                 && duration > 0
                 && seen_position + 2 >= duration;
-            if track_ended {
+            let explicit_recovery_edge = event.engine_empty_generation
-                // Listen log: the natural end, before anything moves the
+                != seen_engine_empty_generation
-                // cursor (the next edge must find no open row).
+                || event.source_failure_generation != seen_source_failure_generation;
-                crate::listen_log_qt::on_natural_end().await;
+            let queue_track_id = if explicit_recovery_edge || sampled_track_end {
-                // Seed for InfiniteRadio, read BEFORE anything moves the
+                runtime
-                // cursor: the track that just ended is still the current one.
-                let ended_track_id = runtime
                     .core()
                     .current_track()
                     .await
                     .map(|t| t.id)
-                    .unwrap_or(0);
+                    .unwrap_or(0)
+            } else {
+                0
+            };
+            let recovery_cause = playback_recovery_cause(
+                &event,
+                queue_track_id,
+                seen_engine_empty_generation,
+                seen_source_failure_generation,
+                sampled_track_end,
+            );
+            // Consume every observed generation, including a stale edge whose
+            // track id no longer matches after a manual play. It must not wake
+            // up again if that id appears later in the session.
+            seen_engine_empty_generation = event.engine_empty_generation;
+            seen_source_failure_generation = event.source_failure_generation;
+
+            if let Some(recovery_cause) = recovery_cause {
+                match recovery_cause {
+                    PlaybackRecoveryCause::SourceFailure => {
+                        log::warn!(
+                            "[qbz-qt] recovery: source retries exhausted for track {queue_track_id}; continuing the queue"
+                        );
+                        crate::listen_log_qt::on_error().await;
+                    }
+                    PlaybackRecoveryCause::EngineEmpty => {
+                        log::info!(
+                            "[qbz-qt] recovery: engine empty for track {queue_track_id} with no gapless successor; continuing the queue"
+                        );
+                        crate::listen_log_qt::on_natural_end().await;
+                    }
+                    PlaybackRecoveryCause::SampledNaturalEnd => {
+                        crate::listen_log_qt::on_natural_end().await;
+                    }
+                }
+
+                if let Some(task) = gapless_fetch_task.take() {
+                    if !task.is_finished() {
+                        log::info!(
+                            "[qbz-qt] [GAPLESS] aborting unfinished successor fetch during recovery"
+                        );
+                        task.abort();
+                    }
+                }
+
+                // Seed for InfiniteRadio, read BEFORE anything moves the
+                // cursor: the track that just ended is still the current one.
+                let ended_track_id = queue_track_id;
                 // Stop-after-this-song: if the track that just ended carries
                 // the marker, HALT here — do not advance, do not refill. The
                 // queue stays intact and the finished track stays parked in
@@ -3799,11 +3921,13 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
 #[cfg(test)]
 mod tests {
     use super::{
-        filter_queue_with, gapless_edge_matches, prefetch_track_ids, quality_badge_from,
+        filter_queue_with, gapless_edge_matches, playback_recovery_cause, prefetch_track_ids,
-        reconcile_device_cap, should_stream_gapless_successor, stream_error_text,
+        quality_badge_from, reconcile_device_cap, should_stream_gapless_successor,
-        xorshift_shuffle_seeded, PrefetchCandidate, PrefetchTrackEdge,
+        stream_error_text, xorshift_shuffle_seeded, PlaybackRecoveryCause, PrefetchCandidate,
+        PrefetchTrackEdge,
     };
     use qbz_models::{Quality, QualityLimit, QueueTrack};
+    use qbz_player::PlaybackEvent;
     use qbz_source::QualityHint;
 
     fn remote(id: u64) -> PrefetchCandidate {
@@ -3817,6 +3941,74 @@ mod tests {
         PrefetchCandidate { id, is_local: true }
     }
 
+    fn playback_event(track_id: u64, is_playing: bool) -> PlaybackEvent {
+        PlaybackEvent {
+            is_playing,
+            position: 0,
+            duration: 0,
+            track_id,
+            volume: 1.0,
+            sample_rate: None,
+            bit_depth: None,
+            shuffle: None,
+            repeat: None,
+            normalization_gain: None,
+            gapless_ready: false,
+            gapless_next_track_id: 0,
+            bit_perfect_mode: None,
+            buffer_progress: None,
+            engine_empty_generation: 0,
+            engine_empty_track_id: 0,
+            source_failure_generation: 0,
+            source_failure_track_id: 0,
+        }
+    }
+
+    #[test]
+    fn engine_empty_edge_survives_a_missed_playing_pulse() {
+        let mut event = playback_event(41, false);
+        event.engine_empty_generation = 1;
+        event.engine_empty_track_id = 41;
+
+        assert_eq!(
+            playback_recovery_cause(&event, 41, 0, 0, false),
+            Some(PlaybackRecoveryCause::EngineEmpty)
+        );
+    }
+
+    #[test]
+    fn source_failure_recovers_a_track_that_never_started() {
+        let mut event = playback_event(0, false);
+        event.source_failure_generation = 1;
+        event.source_failure_track_id = 41;
+
+        assert_eq!(
+            playback_recovery_cause(&event, 41, 0, 0, false),
+            Some(PlaybackRecoveryCause::SourceFailure)
+        );
+    }
+
+    #[test]
+    fn stale_recovery_edge_cannot_skip_a_new_queue_track() {
+        let mut event = playback_event(42, false);
+        event.source_failure_generation = 3;
+        event.source_failure_track_id = 41;
+        event.engine_empty_generation = 2;
+        event.engine_empty_track_id = 41;
+
+        assert_eq!(playback_recovery_cause(&event, 42, 1, 2, false), None);
+    }
+
+    #[test]
+    fn late_gapless_rescue_consumes_engine_edge_without_advancing() {
+        let mut event = playback_event(41, true);
+        event.engine_empty_generation = 1;
+        event.engine_empty_track_id = 41;
+        event.gapless_next_track_id = 42;
+
+        assert_eq!(playback_recovery_cause(&event, 41, 0, 0, false), None);
+    }
+
     fn queue_track() -> QueueTrack {
         QueueTrack {
             id: 1,

crates/qbzd/src/api/playback.rs

@@ -512,6 +512,10 @@ mod tests {
             gapless_next_track_id: 0,
             bit_perfect_mode: None,
             buffer_progress: None,
+            engine_empty_generation: 0,
+            engine_empty_track_id: 0,
+            source_failure_generation: 0,
+            source_failure_track_id: 0,
         }
     }