feat(prefetch): adaptive throttle based on observed bandwidth + underrun signal

Recovered from 63ecc324 (authored on the native-port branch, never merged to
pre-release). Backend-only, no Slint. Scales the prefetch cap down on slow
links and during audio underruns, and exposes the per-segment download signal
the offline detector reuses as a positive liveness override (#467).

cc

crates/qbz-audio/src/alsa_direct.rs

@@ -12,6 +12,24 @@ use std::sync::atomic::AtomicBool;
 #[cfg(target_os = "linux")]
 use std::sync::{Arc, Mutex};
 
+/// Log a PCM recovery and record it as a network-throttle underrun signal.
+///
+/// Each call to ALSA's `pcm.recover()` that returns successfully indicates
+/// that the writer thread fell behind the kernel's playback buffer — i.e.
+/// an audio underrun. The network throttle treats this as the strongest
+/// possible "slow down" signal and immediately drops the prefetch cap to
+/// zero for `PANIC_WINDOW_SECS`, so the live stream gets the full pipe to
+/// recover.
+#[cfg(target_os = "linux")]
+fn log_pcm_recovery(suffix: &str) {
+    if suffix.is_empty() {
+        log::warn!("[ALSA Direct] Recovered from PCM error");
+    } else {
+        log::warn!("[ALSA Direct] Recovered from PCM error {}", suffix);
+    }
+    crate::network_throttle::state().record_underrun();
+}
+
 /// Direct ALSA PCM stream for hw: devices
 ///
 /// Field order is significant: Rust drops struct fields top-to-bottom, so the
@@ -223,7 +241,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -252,7 +270,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -279,7 +297,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -318,7 +336,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error (S24_3LE)");
+                            log_pcm_recovery("(S24_3LE)");
                             Ok(())
                         }
                     }
@@ -348,7 +366,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -388,7 +406,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -420,7 +438,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -456,7 +474,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }
@@ -493,7 +511,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error (S24_3LE)");
+                            log_pcm_recovery("(S24_3LE)");
                             Ok(())
                         }
                     }
@@ -523,7 +541,7 @@ impl AlsaDirectStream {
                         if let Err(recover_err) = pcm.recover(e.errno() as i32, false) {
                             Err(format!("Failed to recover from error: {}", recover_err))
                         } else {
-                            log::warn!("[ALSA Direct] Recovered from PCM error");
+                            log_pcm_recovery("");
                             Ok(())
                         }
                     }

crates/qbz-audio/src/lib.rs

@@ -45,6 +45,7 @@ pub mod dynamic_amplify;
 pub mod loudness;
 pub mod loudness_analyzer;
 pub mod loudness_cache;
+pub mod network_throttle;
 pub mod output_sinks;
 pub mod settings;
 pub mod visualizer;

crates/qbz-audio/src/network_throttle.rs

@@ -0,0 +1,244 @@
+//! Adaptive prefetch throttling based on observed network conditions.
+//!
+//! Default QBZ prefetches up to 5 tracks concurrently (≈30 simultaneous
+//! HTTP/2 streams once the CMAF init + segment fan-out is counted). That
+//! aggressive concurrency is exactly what makes Hi-Res playback smooth on
+//! a fast connection, and we don't want to dial it down for everyone just
+//! because some users have a slow link.
+//!
+//! Instead, this module watches two real-time signals — measured per-segment
+//! bandwidth and audio underruns — and computes a *dynamic* cap that the
+//! prefetch dispatcher consults. When conditions are good the cap is the
+//! full memory-profile default (5 on Normal-class hosts, 1 on LowMemory).
+//! When the live stream starts losing the race against the consumer, the
+//! cap collapses to 0 and the prefetch fan-out gets out of the way.
+//!
+//! Recovery follows TCP-style slow-start logic: after `PANIC_WINDOW_SECS`
+//! of no fresh underruns the cap walks back up one level at a time,
+//! re-validated by each new bandwidth sample.
+
+use std::sync::OnceLock;
+use std::sync::RwLock;
+use std::time::Instant;
+
+/// EMA smoothing factor for observed bandwidth. Higher = more reactive to
+/// the latest sample, lower = smoother but laggier. 0.4 is a compromise:
+/// a single bad sample dents the estimate but doesn't dominate it.
+const BANDWIDTH_EMA_ALPHA: f64 = 0.4;
+
+/// How long we stay in panic mode after an audio underrun. The user just
+/// experienced a glitch; we want the stream to recover with the entire
+/// pipe to itself for a meaningful window before letting prefetch back in.
+const PANIC_WINDOW_SECS: u64 = 30;
+
+/// Bandwidth-to-playback ratio thresholds for the throttle levels.
+///
+/// - At or below `SURVIVING_RATIO`: no prefetch at all. The live stream
+///   barely has bandwidth to keep itself fed.
+/// - At or below `CAUTIOUS_RATIO`: only one prefetch track in flight.
+/// - At or below `RELAXED_RATIO`: two prefetch tracks.
+/// - Above `RELAXED_RATIO`: full memory-profile default.
+const SURVIVING_RATIO: f64 = 1.5;
+const CAUTIOUS_RATIO: f64 = 2.5;
+const RELAXED_RATIO: f64 = 4.0;
+
+/// Approximate sustained bandwidth required for live playback by quality
+/// tier, in MB/s. These numbers are for CMAF/FLAC compressed streams;
+/// they're inputs to the ratio comparison above, not hard limits.
+pub fn playback_mbps_for_quality(quality_tag: PlaybackQualityTag) -> f64 {
+    match quality_tag {
+        PlaybackQualityTag::UltraHiRes => 2.5, // 24-bit / 192 kHz FLAC
+        PlaybackQualityTag::HiRes => 1.4,      // 24-bit / 96 kHz FLAC
+        PlaybackQualityTag::Lossless => 0.5,   // 16-bit / 44.1 kHz FLAC
+        PlaybackQualityTag::Lossy => 0.04,     // 320 kbps MP3
+    }
+}
+
+/// Small enum that mirrors `qbz-models::Quality` without taking a runtime
+/// dependency on it. The callsite translates its own quality enum into one
+/// of these four buckets before consulting the throttle.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PlaybackQualityTag {
+    UltraHiRes,
+    HiRes,
+    Lossless,
+    Lossy,
+}
+
+#[derive(Debug, Default)]
+struct ThrottleInner {
+    /// Exponential moving average of observed segment-download bandwidth
+    /// in MB/s. `None` until the first sample arrives.
+    bandwidth_ema_mbps: Option<f64>,
+    /// Timestamp of the most recent audio underrun. `None` means we have
+    /// not observed one this session.
+    last_underrun: Option<Instant>,
+}
+
+pub struct ThrottleState {
+    inner: RwLock<ThrottleInner>,
+}
+
+static GLOBAL: OnceLock<ThrottleState> = OnceLock::new();
+
+/// Singleton accessor. Lazily initialized on first call.
+pub fn state() -> &'static ThrottleState {
+    GLOBAL.get_or_init(|| ThrottleState {
+        inner: RwLock::new(ThrottleInner::default()),
+    })
+}
+
+impl ThrottleState {
+    /// Feed a fresh per-segment bandwidth measurement (MB/s). Called from
+    /// the CMAF streaming loop every few segments.
+    pub fn record_segment_bandwidth(&self, mbps: f64) {
+        if !mbps.is_finite() || mbps <= 0.0 {
+            return;
+        }
+        if let Ok(mut inner) = self.inner.write() {
+            inner.bandwidth_ema_mbps = Some(match inner.bandwidth_ema_mbps {
+                Some(prev) => prev * (1.0 - BANDWIDTH_EMA_ALPHA) + mbps * BANDWIDTH_EMA_ALPHA,
+                None => mbps,
+            });
+        }
+    }
+
+    /// Signal that an audio buffer underrun just happened. Forces the
+    /// throttle into panic mode for `PANIC_WINDOW_SECS`.
+    pub fn record_underrun(&self) {
+        if let Ok(mut inner) = self.inner.write() {
+            inner.last_underrun = Some(Instant::now());
+        }
+    }
+
+    /// Current EMA bandwidth in MB/s, or `None` if no samples yet.
+    pub fn current_bandwidth_mbps(&self) -> Option<f64> {
+        self.inner.read().ok().and_then(|i| i.bandwidth_ema_mbps)
+    }
+
+    /// True when an underrun was recorded within `PANIC_WINDOW_SECS`.
+    pub fn in_panic_mode(&self) -> bool {
+        let inner = match self.inner.read() {
+            Ok(g) => g,
+            Err(_) => return false,
+        };
+        match inner.last_underrun {
+            Some(t) => t.elapsed().as_secs() < PANIC_WINDOW_SECS,
+            None => false,
+        }
+    }
+
+    /// Decide the prefetch cap given the current track's playback rate and
+    /// the memory-profile default. The cap is always clamped to `[0, default_cap]`
+    /// — we never *raise* prefetch above the memory profile, only restrict.
+    pub fn current_prefetch_cap(&self, playback_mbps: f64, default_cap: usize) -> usize {
+        if self.in_panic_mode() {
+            return 0;
+        }
+        let bw = match self.current_bandwidth_mbps() {
+            Some(v) => v,
+            // No samples yet — trust the memory profile default. The first
+            // segment of the first track will land within a few seconds and
+            // give us real numbers.
+            None => return default_cap,
+        };
+        let ratio = if playback_mbps > 0.0 {
+            bw / playback_mbps
+        } else {
+            f64::INFINITY
+        };
+        if ratio <= SURVIVING_RATIO {
+            0
+        } else if ratio <= CAUTIOUS_RATIO {
+            1.min(default_cap)
+        } else if ratio <= RELAXED_RATIO {
+            2.min(default_cap)
+        } else {
+            default_cap
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn fresh_state_returns_default_cap() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 5);
+    }
+
+    #[test]
+    fn panic_mode_zeros_cap() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        s.record_underrun();
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 0);
+    }
+
+    #[test]
+    fn surviving_ratio_zeros_prefetch() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        // bw = 3.0, playback = 2.5 → ratio = 1.2 (< 1.5)
+        s.record_segment_bandwidth(3.0);
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 0);
+    }
+
+    #[test]
+    fn cautious_ratio_allows_one() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        // bw = 5.0, playback = 2.5 → ratio = 2.0 (between 1.5 and 2.5)
+        s.record_segment_bandwidth(5.0);
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 1);
+    }
+
+    #[test]
+    fn relaxed_ratio_allows_two() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        // bw = 8.0, playback = 2.5 → ratio = 3.2 (between 2.5 and 4.0)
+        s.record_segment_bandwidth(8.0);
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 2);
+    }
+
+    #[test]
+    fn abundant_bandwidth_unlocks_default() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        // bw = 20.0, playback = 2.5 → ratio = 8.0 (well above 4.0)
+        s.record_segment_bandwidth(20.0);
+        assert_eq!(s.current_prefetch_cap(2.5, 5), 5);
+    }
+
+    #[test]
+    fn cap_never_exceeds_default() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        s.record_segment_bandwidth(20.0);
+        // Memory profile says 1 — never raise above.
+        assert_eq!(s.current_prefetch_cap(2.5, 1), 1);
+    }
+
+    #[test]
+    fn ema_smooths_spikes() {
+        let s = ThrottleState {
+            inner: RwLock::new(ThrottleInner::default()),
+        };
+        s.record_segment_bandwidth(10.0);
+        s.record_segment_bandwidth(1.0);
+        // EMA: 10 * 0.6 + 1 * 0.4 = 6.4
+        let bw = s.current_bandwidth_mbps().unwrap();
+        assert!((bw - 6.4).abs() < 0.01);
+    }
+}

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

@@ -282,6 +282,15 @@ impl BufferedMediaSource {
     ///
     /// Used to store in cache after streaming playback completes.
     /// Returns None if download is not complete or failed.
+    ///
+    /// IMPORTANT: This clones the buffer rather than moving it. Earlier
+    /// attempts to use `mem::take` here regressed playback — the
+    /// `Source` impl is still actively reading from `state.data` when
+    /// the promotion path calls this, and zeroing the buffer out from
+    /// under the reader caused immediate EOF (tracks ending 10s into a
+    /// 104s file). Cloning is the safe choice; the audible hiccup at
+    /// promotion that the move attempted to fix needs a different
+    /// approach (shared `Arc<Vec<u8>>` ownership, or off-thread copy).
     pub fn take_complete_data(&self) -> Option<Vec<u8>> {
         let (lock, _) = &*self.state;
         if let Ok(state) = lock.lock() {

src-tauri/src/commands_v2/helpers.rs

@@ -960,17 +960,22 @@ pub async fn v2_cmaf_stream(
         // Progress logging every 5 segments or on last segment
         if seg_idx % 5 == 0 || seg_idx == n_segments {
             let elapsed = start.elapsed().as_secs_f64();
+            let mbps = if elapsed > 0.0 {
+                total_written as f64 / (1024.0 * 1024.0) / elapsed
+            } else {
+                0.0
+            };
             log::info!(
                 "[V2/CMAF-STREAM] Segment {}/{} ({:.1} MB, {:.1} MB/s)",
                 seg_idx,
                 n_segments - 1,
                 total_written as f64 / (1024.0 * 1024.0),
-                if elapsed > 0.0 {
-                    total_written as f64 / (1024.0 * 1024.0) / elapsed
-                } else {
-                    0.0
-                }
+                mbps
             );
+            // Feed the network-throttle state with the latest observed
+            // bandwidth so the prefetch scheduler can scale down on slow
+            // connections and stay out of the live stream's way.
+            qbz_audio::network_throttle::state().record_segment_bandwidth(mbps);
         }
     }
 

src-tauri/src/commands_v2/playback.rs

@@ -26,14 +26,53 @@ use super::cached_audio_incompatible_with_hw;
 
 // ==================== Prefetch (V2) ====================
 
-/// Number of Qobuz tracks to prefetch — resolved per-host from the
-/// detected memory profile. Normal hosts get 5 tracks (~2 minutes
-/// of HiRes cache ahead); LowMemory hosts (issue #331) get 1 to keep
-/// the in-memory footprint manageable on devices like the Pi 3B.
-fn v2_prefetch_count() -> usize {
+/// Memory-profile default number of Qobuz tracks to prefetch. Normal
+/// hosts get 5 (~2 minutes of HiRes cache ahead); LowMemory hosts
+/// (issue #331) get 1 to keep the in-memory footprint manageable on
+/// devices like the Pi 3B. This is the upper bound the adaptive
+/// throttle is allowed to walk back up to — it never raises above this.
+fn v2_prefetch_count_default() -> usize {
     qbz_core::system_capabilities::memory_profile().prefetch_count
 }
 
+/// Translate the active playback `Quality` into the throttle's coarse
+/// quality bucket, used to estimate the live stream's sustained
+/// bandwidth requirement (MB/s). Inputs to `current_prefetch_cap()`.
+fn quality_to_throttle_tag(q: qbz_models::Quality) -> qbz_audio::network_throttle::PlaybackQualityTag {
+    use qbz_audio::network_throttle::PlaybackQualityTag;
+    use qbz_models::Quality;
+    match q {
+        Quality::UltraHiRes => PlaybackQualityTag::UltraHiRes,
+        Quality::HiRes => PlaybackQualityTag::HiRes,
+        Quality::Lossless => PlaybackQualityTag::Lossless,
+        Quality::Mp3 => PlaybackQualityTag::Lossy,
+    }
+}
+
+/// Adaptive prefetch cap. Starts at the memory-profile default and gets
+/// scaled down by the network-throttle state based on observed
+/// bandwidth and recent audio underruns. The throttle never raises the
+/// cap above the memory-profile default, so a user with a fat
+/// connection still gets the full 5-track fan-out the firmware intends.
+fn v2_prefetch_count(quality: qbz_models::Quality) -> usize {
+    let default_cap = v2_prefetch_count_default();
+    let tag = quality_to_throttle_tag(quality);
+    let playback_mbps = qbz_audio::network_throttle::playback_mbps_for_quality(tag);
+    let throttled_cap = qbz_audio::network_throttle::state()
+        .current_prefetch_cap(playback_mbps, default_cap);
+    if throttled_cap < default_cap {
+        log::debug!(
+            "[V2/PREFETCH] Adaptive cap: {} (down from default {}). bw={:?} MB/s, panic={}, playback_need={:.2} MB/s",
+            throttled_cap,
+            default_cap,
+            qbz_audio::network_throttle::state().current_bandwidth_mbps(),
+            qbz_audio::network_throttle::state().in_panic_mode(),
+            playback_mbps,
+        );
+    }
+    throttled_cap
+}
+
 /// How far ahead to look for tracks to prefetch (to handle mixed playlists
 /// with local/offline tracks interspersed with Qobuz tracks)
 const V2_PREFETCH_LOOKAHEAD: usize = 15;
@@ -98,7 +137,7 @@ fn spawn_v2_prefetch_with_hw_check(
 
     let mut qobuz_prefetched = 0;
 
-    let prefetch_cap = v2_prefetch_count();
+    let prefetch_cap = v2_prefetch_count(quality);
     for track in upcoming_tracks {
         // Stop once we've prefetched enough Qobuz tracks
         if qobuz_prefetched >= prefetch_cap {