diff --git a/crates/Cargo.lock b/crates/Cargo.lock index db7bb6656..79f38a906 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -4006,6 +4006,7 @@ dependencies = [ "qbz-models", "rusqlite", "serde", + "serde_json", "tempfile", "windows 0.61.3", ] diff --git a/crates/qbz-cache/Cargo.toml b/crates/qbz-cache/Cargo.toml index cf80d51d7..2cf9dbd59 100644 --- a/crates/qbz-cache/Cargo.toml +++ b/crates/qbz-cache/Cargo.toml @@ -12,6 +12,7 @@ log = "0.4" # Serialization (for stats) serde = { version = "1", features = ["derive"] } +serde_json = { workspace = true } # File system utilities dirs = "6" diff --git a/crates/qbz-cache/src/audio_cache.rs b/crates/qbz-cache/src/audio_cache.rs index 11e06d7f5..87cc2898d 100644 --- a/crates/qbz-cache/src/audio_cache.rs +++ b/crates/qbz-cache/src/audio_cache.rs @@ -16,10 +16,12 @@ pub struct CachedTrack { pub track_id: u64, pub data: Vec, pub size_bytes: usize, + pub quality: Option, } #[derive(Clone)] struct SharedTrack { + quality: Option, track_id: u64, data: Arc>, size_bytes: usize, @@ -154,6 +156,7 @@ impl AudioCache { track_id: t.track_id, data: (*t.data).clone(), size_bytes: t.size_bytes, + quality: t.quality, }) } @@ -162,6 +165,12 @@ impl AudioCache { self.state.lock().unwrap().tracks.contains_key(&track_id) } + /// Inspect one immutable entry without copying a track-sized allocation. + pub fn peek_with_quality(&self, track_id: u64) -> Option<(Arc>, Option)> { + self.state.lock().unwrap().tracks.get(&track_id) + .map(|track| (track.data.clone(), track.quality)) + } + /// Check if a track is currently being fetched pub fn is_fetching(&self, track_id: u64) -> bool { self.state.lock().unwrap().fetching.contains(&track_id) @@ -297,12 +306,24 @@ impl AudioCache { /// Admit a sealed streaming buffer without allocating a second track-sized copy. /// Reservation, eviction and insertion are atomic with respect to other writers. pub fn insert_shared(&self, track_id: u64, data: Arc>) -> CacheAdmission { - self.admit(track_id, data, true) + self.insert_shared_with_quality(track_id, data, None) + } + + pub fn insert_with_quality(&self, track_id: u64, data: Vec, quality: Option) -> CacheAdmission { + self.insert_shared_with_quality(track_id, Arc::new(data), quality) + } + + pub fn insert_shared_with_quality(&self, track_id: u64, data: Arc>, quality: Option) -> CacheAdmission { + self.admit(track_id, data, true, quality) } /// L2 reads may warm L1, but an oversized hit must not rewrite the same file. pub fn promote_from_disk(&self, track_id: u64, data: Vec) -> CacheAdmission { - self.admit(track_id, Arc::new(data), false) + self.promote_from_disk_with_quality(track_id, data, None) + } + + pub fn promote_from_disk_with_quality(&self, track_id: u64, data: Vec, quality: Option) -> CacheAdmission { + self.admit(track_id, Arc::new(data), false, quality) } /// Decide storage before a download allocates its payload. Dynamic growth @@ -349,7 +370,7 @@ impl AudioCache { } } - fn admit(&self, track_id: u64, data: Arc>, spill_rejected: bool) -> CacheAdmission { + fn admit(&self, track_id: u64, data: Arc>, spill_rejected: bool, quality: Option) -> CacheAdmission { let size = data.len(); let mut retired = Vec::new(); let admitted = { @@ -369,6 +390,7 @@ impl AudioCache { state.tracks.insert( track_id, SharedTrack { + quality, track_id, data: data.clone(), size_bytes: size, @@ -380,7 +402,7 @@ impl AudioCache { }; if let Some(disk) = &self.playback_cache { for track in retired { - disk.insert(track.track_id, &track.data); + disk.insert_with_quality(track.track_id, &track.data, track.quality); } } if admitted { @@ -389,7 +411,7 @@ impl AudioCache { && self .playback_cache .as_ref() - .is_some_and(|disk| disk.insert(track_id, &data)) + .is_some_and(|disk| disk.insert_with_quality(track_id, &data, quality)) { CacheAdmission::Disk } else { diff --git a/crates/qbz-cache/src/lib.rs b/crates/qbz-cache/src/lib.rs index 8c97c324c..52def6c40 100644 --- a/crates/qbz-cache/src/lib.rs +++ b/crates/qbz-cache/src/lib.rs @@ -56,6 +56,8 @@ //! ``` mod audio_cache; +mod quality; +pub use quality::CacheQuality; mod memory; mod image_cache; mod playback_cache; diff --git a/crates/qbz-cache/src/playback_cache.rs b/crates/qbz-cache/src/playback_cache.rs index 990910b0c..acc2080da 100644 --- a/crates/qbz-cache/src/playback_cache.rs +++ b/crates/qbz-cache/src/playback_cache.rs @@ -10,6 +10,38 @@ use std::path::PathBuf; use std::sync::Mutex; use std::time::SystemTime; +/// The sidecar is valid only for the exact atomically published audio file. +/// Readers compare against their open handle, so a concurrent replacement +/// cannot lend the old file the new acquisition ceiling (or vice versa). +#[derive(serde::Serialize, serde::Deserialize, PartialEq, Eq)] +struct FileIdentity { + size: u64, + modified_ns: u128, + #[cfg(unix)] + device: u64, + #[cfg(unix)] + inode: u64, +} +impl FileIdentity { + fn read(file: &fs::File) -> Option { + let meta = file.metadata().ok()?; + #[cfg(unix)] + use std::os::unix::fs::MetadataExt; + Some(Self { + size: meta.len(), + modified_ns: meta.modified().ok()?.duration_since(SystemTime::UNIX_EPOCH).ok()?.as_nanos(), + #[cfg(unix)] device: meta.dev(), + #[cfg(unix)] inode: meta.ino(), + }) + } +} +#[derive(serde::Serialize, serde::Deserialize)] +struct QualityRecord { + version: u8, + file: FileIdentity, + quality: crate::CacheQuality, +} + /// Entry metadata for tracking cache usage #[derive(Debug, Clone)] struct CacheEntry { @@ -72,6 +104,9 @@ impl PlaybackCache { // Scan existing files to rebuild state cache.rebuild_state(); + if !cache.evict_if_needed(0) { + log::warn!("Playback cache remains above its budget; further admission is blocked"); + } log::info!( "Playback cache initialized at {:?} (max {} MB)", @@ -145,6 +180,37 @@ impl PlaybackCache { Some(file) } + pub fn open_with_quality(&self, track_id: u64) -> Option<(fs::File, Option)> { + let file = self.open(track_id)?; + let quality = (|| { + let identity = FileIdentity::read(&file)?; + let mut bytes = Vec::new(); + fs::File::open(self.quality_path(track_id)).ok()?.take(4096).read_to_end(&mut bytes).ok()?; + let record: QualityRecord = serde_json::from_slice(&bytes).ok()?; + (record.version == 1 && record.file == identity).then_some(record.quality) + })(); + Some((file, quality)) + } + + fn quality_path(&self, track_id: u64) -> PathBuf { + self.cache_dir.join(format!("{track_id}.quality.json")) + } + + fn publish_quality(&self, track_id: u64, quality: Option) { + let path = self.quality_path(track_id); + let published = (|| -> Option<()> { + let quality = quality?; + let file = fs::File::open(self.track_path(track_id)).ok()?; + let record = QualityRecord { version: 1, file: FileIdentity::read(&file)?, quality }; + let bytes = serde_json::to_vec(&record).ok()?; + let mut pending = tempfile::NamedTempFile::new_in(&self.cache_dir).ok()?; + pending.write_all(&bytes).ok()?; + pending.persist(&path).ok()?; + Some(()) + })(); + if published.is_none() { let _ = fs::remove_file(path); } + } + /// Temporary playback spools live on the cache disk, never the system /tmp /// (commonly tmpfs on streamers). They disappear with their last file handle. pub fn create_spool(&self) -> std::io::Result { @@ -202,6 +268,10 @@ impl PlaybackCache { /// Insert a track into the cache (called when evicting from memory cache) pub fn insert(&self, track_id: u64, data: &[u8]) -> bool { + self.insert_with_quality(track_id, data, None) + } + + pub fn insert_with_quality(&self, track_id: u64, data: &[u8], quality: Option) -> bool { let _write = self.writes.lock().unwrap(); let size = data.len() as u64; @@ -217,7 +287,7 @@ impl PlaybackCache { } // Evict old entries if needed - self.evict_if_needed(size); + if !self.evict_if_needed(size) { return false; } let path = self.track_path(track_id); let pending = path.with_extension(format!("audio.{}.tmp", std::process::id())); @@ -234,6 +304,7 @@ impl PlaybackCache { let _ = fs::remove_file(&pending); return false; } + self.publish_quality(track_id, quality); let mut state = self.state.lock().unwrap(); // Remove old entry if exists @@ -286,6 +357,13 @@ impl PlaybackCache { /// rejection and pre-write eviction; the entry records the actual bytes /// written. pub fn insert_from(&self, track_id: u64, size_hint: u64, fill: F) -> bool + where + F: FnOnce(&mut fs::File) -> std::io::Result, + { + self.insert_from_with_quality(track_id, size_hint, None, fill) + } + + pub fn insert_from_with_quality(&self, track_id: u64, size_hint: u64, quality: Option, fill: F) -> bool where F: FnOnce(&mut fs::File) -> std::io::Result, { @@ -307,12 +385,15 @@ impl PlaybackCache { )); } drop(file); - self.evict_if_needed(written); + if !self.evict_if_needed(written) { + return Err(std::io::Error::other("playback cache budget could not be reclaimed")); + } fs::rename(&pending, self.track_path(track_id))?; Ok(written) })(); match result { Ok(size) => { + self.publish_quality(track_id, quality); let mut state = self.state.lock().unwrap(); if let Some(old) = state.entries.remove(&track_id) { state.current_size = state.current_size.saturating_sub(old.size_bytes); @@ -338,37 +419,32 @@ impl PlaybackCache { } /// Evict oldest entries to make room for new data - fn evict_if_needed(&self, needed_bytes: u64) { + fn evict_if_needed(&self, needed_bytes: u64) -> bool { let mut state = self.state.lock().unwrap(); - - while state.current_size + needed_bytes > self.max_size_bytes && !state.entries.is_empty() { - // Find oldest entry - let oldest_id = state - .entries - .iter() - .min_by_key(|(_, e)| e.last_accessed) - .map(|(id, _)| *id); - - if let Some(track_id) = oldest_id { - if let Some(entry) = state.entries.remove(&track_id) { - state.current_size = state.current_size.saturating_sub(entry.size_bytes); - - // Delete file - let path = self.cache_dir.join(format!("{}.audio", track_id)); - if let Err(e) = fs::remove_file(&path) { - log::debug!("Failed to delete playback cache file: {}", e); - } else { - log::debug!( - "Evicted track {} from playback cache ({} KB)", - track_id, - entry.size_bytes / 1024 - ); - } - } - } else { + let mut candidates: Vec<_> = state.entries.values() + .map(|entry| (entry.track_id, entry.last_accessed)).collect(); + candidates.sort_by_key(|(_, accessed)| *accessed); + for (track_id, _) in candidates { + if state.current_size <= self.max_size_bytes.saturating_sub(needed_bytes) { break; } + match fs::remove_file(self.track_path(track_id)) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + // Keep accounting for files we failed to remove. Try the + // remaining entries once, never loop on the failed oldest. + log::warn!("Cannot evict playback cache track {track_id}: {error}"); + continue; + } + } + let _ = fs::remove_file(self.quality_path(track_id)); + if let Some(entry) = state.entries.remove(&track_id) { + state.current_size = state.current_size.saturating_sub(entry.size_bytes); + } } + needed_bytes <= self.max_size_bytes + && state.current_size <= self.max_size_bytes - needed_bytes } /// Clear the entire cache @@ -377,6 +453,7 @@ impl PlaybackCache { let mut state = self.state.lock().unwrap(); for track_id in state.entries.keys() { + let _ = fs::remove_file(self.quality_path(*track_id)); let path = self.cache_dir.join(format!("{}.audio", track_id)); let _ = fs::remove_file(&path); } @@ -415,6 +492,78 @@ pub struct PlaybackCacheStats { mod disk_reader_tests { use super::*; + #[test] + fn failed_eviction_preserves_accounting_and_rejects_admission() { + let temp = tempfile::tempdir().unwrap(); + let cache = PlaybackCache::with_path(temp.path().into(), 4).unwrap(); + assert!(cache.insert(1, b"full")); + fs::remove_file(cache.track_path(1)).unwrap(); + // A directory cannot be unlinked as a cache file, even as root. + fs::create_dir(cache.track_path(1)).unwrap(); + assert!(!cache.insert(2, b"new")); + assert_eq!(cache.stats().current_size_bytes, 4); + assert!(!cache.contains(2)); + fs::remove_dir(cache.track_path(1)).unwrap(); + assert!(cache.insert(2, b"new")); + assert_eq!(cache.stats().current_size_bytes, 3); + } + + #[test] + fn acquisition_survives_reopen_and_cannot_attach_to_replaced_bytes() { + use qbz_models::Quality; + let temp = tempfile::tempdir().unwrap(); + let cache = PlaybackCache::with_path(temp.path().into(), 1024).unwrap(); + let quality = crate::CacheQuality::from_resolved(Quality::UltraHiRes, 27); + assert!(cache.insert_with_quality(7, b"first", quality)); + let record = fs::read(cache.quality_path(7)).unwrap(); + assert_eq!(cache.open_with_quality(7).unwrap().1, quality); + drop(cache); + let cache = PlaybackCache::with_path(temp.path().into(), 1024).unwrap(); + assert_eq!(cache.open_with_quality(7).unwrap().1, quality); + assert!(cache.insert(7, b"other")); // same length, new inode + fs::write(cache.quality_path(7), record).unwrap(); + assert_eq!(cache.open_with_quality(7).unwrap().1, None); + fs::write(cache.quality_path(7), b"invalid metadata").unwrap(); + assert_eq!(cache.open_with_quality(7).unwrap().1, None); + let mut file = cache.open(7).unwrap(); + let mut bytes = Vec::new(); file.read_to_end(&mut bytes).unwrap(); + assert_eq!(bytes, b"other"); + } + + #[test] + fn l1_spill_keeps_acquisition_paired_with_its_bytes() { + let temp = tempfile::tempdir().unwrap(); + let disk = std::sync::Arc::new(PlaybackCache::with_path(temp.path().into(), 1024).unwrap()); + let memory = crate::AudioCache::with_playback_cache(5, disk.clone()); + let quality = crate::CacheQuality::from_resolved(qbz_models::Quality::UltraHiRes, 27); + memory.insert_with_quality(1, b"first".to_vec(), quality); + assert_eq!(memory.get(1).unwrap().quality, quality); + memory.insert(2, b"other".to_vec()); + assert!(memory.get(1).is_none()); + assert_eq!(disk.open_with_quality(1).unwrap().1, quality); + assert_eq!(disk.get(1).unwrap(), b"first"); + } + + #[test] + fn startup_enforces_reduced_budget_and_preserves_unowned_files() { + let temp = tempfile::tempdir().unwrap(); + let cache = PlaybackCache::with_path(temp.path().into(), 1024).unwrap(); + assert!(cache.insert(1, b"older")); + assert!(cache.insert(2, b"newer")); + filetime::set_file_atime(&temp.path().join("1.audio"), filetime::FileTime::from_unix_time(1, 0)).unwrap(); + filetime::set_file_atime(&temp.path().join("2.audio"), filetime::FileTime::from_unix_time(2, 0)).unwrap(); + fs::write(temp.path().join("offline.flac"), b"keep").unwrap(); + drop(cache); + let cache = PlaybackCache::with_path(temp.path().into(), 5).unwrap(); + assert!(!cache.contains(1)); + assert_eq!(cache.get(2).unwrap(), b"newer"); + assert_eq!(fs::read(temp.path().join("offline.flac")).unwrap(), b"keep"); + drop(cache); + let cache = PlaybackCache::with_path(temp.path().into(), 0).unwrap(); + assert!(!cache.contains(2)); + assert!(temp.path().join("offline.flac").exists()); + } + #[test] fn open_reader_survives_replacement_and_eviction() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/qbz-cache/src/quality.rs b/crates/qbz-cache/src/quality.rs new file mode 100644 index 000000000..e348f2bc0 --- /dev/null +++ b/crates/qbz-cache/src/quality.rs @@ -0,0 +1,39 @@ +use qbz_models::Quality; +use serde::{Deserialize, Serialize}; + +/// Acquisition limits, distinct from the PCM format of the available master. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct CacheQuality { + pub requested: Quality, + pub resolved: Quality, +} + +impl CacheQuality { + pub fn from_resolved(requested: Quality, format_id: u32) -> Option { + Some(Self { + requested, + resolved: Quality::from_id(format_id)?, + }) + } + + pub fn satisfies(self, requested: Quality) -> bool { + self.requested >= requested && self.resolved >= requested + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn acquisition_ceiling_distinguishes_best_master_from_lower_tier_copy() { + let best = CacheQuality::from_resolved(Quality::UltraHiRes, 27).unwrap(); + assert!(best.satisfies(Quality::UltraHiRes)); + assert!(best.satisfies(Quality::HiRes)); + let limited = CacheQuality::from_resolved(Quality::HiRes, 7).unwrap(); + assert!(!limited.satisfies(Quality::UltraHiRes)); + let fallback = CacheQuality::from_resolved(Quality::UltraHiRes, 7).unwrap(); + assert!(!fallback.satisfies(Quality::UltraHiRes)); + assert!(CacheQuality::from_resolved(Quality::UltraHiRes, 999).is_none()); + } +} diff --git a/crates/qbz-player/src/player/disk_playback.rs b/crates/qbz-player/src/player/disk_playback.rs index c8ab90156..22c9aefe0 100644 --- a/crates/qbz-player/src/player/disk_playback.rs +++ b/crates/qbz-player/src/player/disk_playback.rs @@ -51,16 +51,17 @@ impl Player { cache: Arc, writer: BufferWriter, track_id: u64, + quality: Option, ) -> Result<(), String> { tokio::task::spawn_blocking(move || -> Result<(), String> { let size = writer.buffer_size(); let admission = if let Some(data) = writer.complete_data_shared() { - cache.insert_shared(track_id, data) + cache.insert_shared_with_quality(track_id, data, quality) } else { // The spool is copied in bounded chunks; never materialize an // oversized file just to pass it to the L2 writer. let saved = cache.get_playback_cache().is_some_and(|disk| { - disk.insert_from(track_id, size as u64, |file| writer.write_buffered_to(file)) + disk.insert_from_with_quality(track_id, size as u64, quality, |file| writer.write_buffered_to(file)) }); if saved { qbz_cache::CacheAdmission::Disk @@ -75,25 +76,39 @@ impl Player { .map_err(|error| format!("Playback cache task failed: {error}"))? } + pub(super) fn cached_data_for_quality(&self, track_id: u64, quality: Quality) + -> Option<(Vec, Option)> { + if let Some(cached) = self.audio_cache.get(track_id) { + if !cached_quality_below_requested(&cached.data, quality, cached.quality) { + return Some((cached.data, cached.quality)); + } + } + let (mut file, acquisition) = self.audio_cache.get_playback_cache()?.open_with_quality(track_id)?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).ok()?; + if cached_quality_below_requested(&bytes, quality, acquisition) { return None; } + self.audio_cache.promote_from_disk_with_quality(track_id, bytes.clone(), acquisition); + Some((bytes, acquisition)) + } + pub(super) fn cached_disk_source( &self, track_id: u64, quality: Quality, ) -> Result, AudioMetadata, u64)>, String> { - let Some(file) = self + let Some((file, acquisition)) = self .audio_cache .get_playback_cache() - .and_then(|cache| cache.open(track_id)) + .and_then(|cache| cache.open_with_quality(track_id)) else { return Ok(None); }; let source = Arc::new(BufferedMediaSource::from_file(file).map_err(|e| e.to_string())?); - let (meta, duration) = source_metadata(&source)?; - let below = match quality { - Quality::UltraHiRes => meta.bit_depth.unwrap_or(16) < 24 || meta.sample_rate <= 96_000, - Quality::HiRes => meta.bit_depth.unwrap_or(16) < 24, - _ => false, + let (meta, duration) = match source_metadata(&source) { + Ok(metadata) => metadata, + Err(_) => return Ok(None), }; + let below = cached_metadata_below_requested(&meta, quality, acquisition); if below { log::info!("[CACHE] Disk track {track_id} below requested {quality:?}; re-fetching"); return Ok(None); diff --git a/crates/qbz-player/src/player/mod.rs b/crates/qbz-player/src/player/mod.rs index a55c0b3ef..d1632f313 100644 --- a/crates/qbz-player/src/player/mod.rs +++ b/crates/qbz-player/src/player/mod.rs @@ -412,22 +412,26 @@ fn audio_metadata_from_codec_params( } /// True when a cached FLAC is a lower quality than `requested`, so the -/// cache entry should be bypassed and the track re-fetched. Ported from -/// the Tauri `cached_quality_below_requested` helper: an unparseable -/// buffer is assumed compatible. -fn cached_quality_below_requested(data: &[u8], requested: Quality) -> bool { - let meta = match extract_audio_metadata_full(data) { - Ok(m) => m, - Err(_) => return false, - }; - let sample_rate = meta.sample_rate; - let bit_depth = meta.bit_depth.unwrap_or(16); +/// cache entry should be bypassed and the track re-fetched. Unparseable +/// bytes cannot prove compatibility with the requested quality. +fn cached_quality_below_requested(data: &[u8], requested: Quality, acquisition: Option) -> bool { + let Ok(meta) = extract_audio_metadata_full(data) else { return true; }; + cached_metadata_below_requested(&meta, requested, acquisition) +} + +fn cached_metadata_below_requested(meta: &AudioMetadata, requested: Quality, acquisition: Option) -> bool { + use symphonia::core::codecs::{CODEC_TYPE_MP3, CODEC_TYPE_AAC, CODEC_TYPE_OPUS, CODEC_TYPE_VORBIS}; + if requested != Quality::Mp3 && matches!(meta.codec, CODEC_TYPE_MP3 | CODEC_TYPE_AAC | CODEC_TYPE_OPUS | CODEC_TYPE_VORBIS) { + return true; + } + if let Some(acquisition) = acquisition { + return !acquisition.satisfies(requested); + } + // Legacy entries have no acquisition ceiling. Their actual format can + // prove sufficient quality, but 24/96 alone cannot prove a 192k request. match requested { - // Hi-Res+: expect 24-bit AND > 96 kHz. - Quality::UltraHiRes => bit_depth < 24 || sample_rate <= 96000, - // Hi-Res: expect 24-bit. - Quality::HiRes => bit_depth < 24, - // Lossless / Mp3: any FLAC satisfies the request. + Quality::UltraHiRes => meta.bit_depth.unwrap_or(16) < 24 || meta.sample_rate <= 96_000, + Quality::HiRes => meta.bit_depth.unwrap_or(16) < 24, _ => false, } } @@ -5500,7 +5504,7 @@ impl Player { // Cache hit: replay instantly from L1/L2 unless the cached copy is // a lower quality than now requested. if let Some(cached) = self.audio_cache.get(track_id) { - if cached_quality_below_requested(&cached.data, quality) { + if cached_quality_below_requested(&cached.data, quality, cached.quality) { log::info!( "[CACHE] Track {} cached below requested {:?} — re-fetching", track_id, @@ -5631,6 +5635,7 @@ impl Player { let content_key = cmaf_info.content_key; let flac_header = cmaf_info.flac_header; let n_segments = cmaf_info.n_segments; + let cache_quality = qbz_cache::CacheQuality::from_resolved(quality, cmaf_info.format_id); let cache = self.audio_cache.clone(); let feeder_state = self.state.clone(); let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); @@ -5645,6 +5650,7 @@ impl Player { buffer_writer, track_id, cache, + cache_quality, skip_cache, total_flac_size, cancel_rx, @@ -5713,7 +5719,7 @@ impl Player { let (source, writer) = self.download_buffered(&stream_url.url).await?; if !self.is_current_play(gen) { return Ok(()); } if !skip_cache { - Self::persist_stream(self.audio_cache.clone(), writer, track_id).await?; + Self::persist_stream(self.audio_cache.clone(), writer, track_id, qbz_cache::CacheQuality::from_resolved(quality, stream_url.format_id)).await?; } if !self.is_current_play(gen) { return Ok(()); } let (meta, duration) = disk_playback::source_metadata(&source)?; @@ -5740,7 +5746,7 @@ impl Player { } if let Some(cached) = self.audio_cache.get(track_id) { - if !cached_quality_below_requested(&cached.data, quality) { + if !cached_quality_below_requested(&cached.data, quality, cached.quality) { return self.play_next(cached.data, track_id); } } @@ -5752,7 +5758,7 @@ impl Player { let (source, writer) = self.download_buffered(&url.url).await?; if !self.is_current_play(gen) { return Err("gapless legacy fetch superseded".into()); } let skip_cache = self.audio_settings.lock().map(|s| s.streaming_only).unwrap_or(false); - if !skip_cache { Self::persist_stream(self.audio_cache.clone(), writer, track_id).await?; } + if !skip_cache { Self::persist_stream(self.audio_cache.clone(), writer, track_id, qbz_cache::CacheQuality::from_resolved(quality, url.format_id)).await?; } if !self.is_current_play(gen) { return Err("gapless legacy fetch superseded".into()); } let (meta, duration_secs) = disk_playback::source_metadata(&source)?; return self.tx.send(AudioCommand::PlayNextStreaming { @@ -5816,6 +5822,7 @@ impl Player { let content_key = cmaf_info.content_key; let flac_header = cmaf_info.flac_header; let n_segments = cmaf_info.n_segments; + let cache_quality = qbz_cache::CacheQuality::from_resolved(quality, cmaf_info.format_id); let cache = self.audio_cache.clone(); let skip_cache = self.audio_settings.lock().map(|settings| settings.streaming_only).unwrap_or(false); tokio::spawn(async move { @@ -5827,6 +5834,7 @@ impl Player { writer, track_id, cache, + cache_quality, skip_cache, total_flac_size, cancel_rx, @@ -5917,7 +5925,8 @@ impl Player { // Already cached, or another prefetch for this id is already // running — nothing to do. - if self.audio_cache.contains(track_id) { + if self.audio_cache.peek_with_quality(track_id).is_some_and(|(data, acquisition)| + !cached_quality_below_requested(&data, quality, acquisition)) { log::debug!("[PREFETCH] Track {track_id} already cached"); return Ok(()); } @@ -5939,9 +5948,7 @@ impl Player { return Ok(()); } - if self.audio_cache.get_playback_cache().is_some_and(|cache| cache.contains(track_id)) { - return Ok(()); - } + if self.cached_disk_source(track_id, quality)?.is_some() { return Ok(()); } let _fetching = disk_playback::PrefetchGuard::new(self.audio_cache.clone(), track_id); log::info!("[PREFETCH] Prefetching track {track_id} at {quality:?}"); @@ -5952,7 +5959,8 @@ impl Player { let (_source, writer) = self.streaming_buffer(StreamingConfig::fast_start(), Some(size))?; let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); Self::cmaf_stream_segments(&info.url_template, info.n_segments, info.content_key, - info.flac_header, writer, track_id, self.audio_cache.clone(), false, size, cancel_rx).await?; + info.flac_header, writer, track_id, self.audio_cache.clone(), + qbz_cache::CacheQuality::from_resolved(quality, info.format_id), false, size, cancel_rx).await?; Ok::<(), String>(()) }.await; let result = match cmaf_result { @@ -5962,7 +5970,7 @@ impl Player { async { let url = client.get_stream_url_with_fallback(track_id, quality).await.map_err(|e| e.to_string())?; let (_source, writer) = self.download_buffered(&url.url).await?; - Self::persist_stream(self.audio_cache.clone(), writer, track_id).await + Self::persist_stream(self.audio_cache.clone(), writer, track_id, qbz_cache::CacheQuality::from_resolved(quality, url.format_id)).await }.await } }; @@ -6123,41 +6131,22 @@ impl Player { track_id: u64, quality: Quality, ) -> Option> { - // L1: in-memory cache. - if let Some(cached) = self.audio_cache.get(track_id) { - log::info!( - "[GAPLESS] Track {track_id} from MEMORY cache ({} bytes)", - cached.size_bytes - ); - return Some(cached.data); - } - - // L2: on-disk plain-FLAC playback cache. Warm L1 on the way out. - if let Some(playback_cache) = self.audio_cache.get_playback_cache() { - if let Some(audio_data) = playback_cache.get(track_id) { - log::info!( - "[GAPLESS] Track {track_id} from DISK cache ({} bytes)", - audio_data.len() - ); - self.audio_cache.promote_from_disk(track_id, audio_data.clone()); - return Some(audio_data); - } - } + if let Some((bytes, _)) = self.cached_data_for_quality(track_id, quality) { return Some(bytes); } // CMAF full download (Akamai CDN), legacy full download as // fallback. Warm L1 so a re-gapless / replay skips the network. // The catalog row rides along with the download (once per track). let (downloaded, _) = tokio::join!( - qbz_qobuz::cmaf::download_full(client, track_id, quality), + qbz_qobuz::cmaf::download_full_with_quality(client, track_id, quality), self.ensure_loudness_row(client, track_id) ); let downloaded = match downloaded { - Ok(data) => Some(data), + Ok((data, resolved)) => Some((data, qbz_cache::CacheQuality::from_resolved(quality, resolved.format_id))), Err(e) => { log::warn!("[GAPLESS] CMAF failed for track {track_id}: {e}, trying legacy"); match client.get_stream_url_with_fallback(track_id, quality).await { Ok(stream_url) => match self.download_audio(&stream_url.url).await { - Ok(data) => Some(data), + Ok(data) => Some((data, qbz_cache::CacheQuality::from_resolved(quality, stream_url.format_id))), Err(e) => { log::warn!("[GAPLESS] Legacy download failed for {track_id}: {e}"); None @@ -6171,16 +6160,16 @@ impl Player { } }; - if let Some(ref data) = downloaded { + if let Some((ref data, acquisition)) = downloaded { log::info!( "[GAPLESS] Track {track_id} downloaded for gapless ({} bytes)", data.len() ); - self.audio_cache.insert(track_id, data.clone()); + self.audio_cache.insert_with_quality(track_id, data.clone(), acquisition); } else { log::info!("[GAPLESS] Track {track_id} not available, gapless not possible"); } - downloaded + downloaded.map(|(data, _)| data) } /// Resolve a fully-materialized audio asset for an EXTERNAL renderer @@ -6228,6 +6217,7 @@ impl Player { let content_key = cmaf_info.content_key; let flac_header = cmaf_info.flac_header; let n_segments = cmaf_info.n_segments; + let cache_quality = qbz_cache::CacheQuality::from_resolved(quality, cmaf_info.format_id); let cache = self.audio_cache.clone(); tokio::spawn(async move { match Self::cmaf_stream_segments( @@ -6238,6 +6228,7 @@ impl Player { writer, track_id, cache, + cache_quality, false, total_bytes, cancel_rx, @@ -6293,37 +6284,16 @@ impl Player { track_id: u64, quality: Quality, ) -> Option { - // L1: in-memory cache (warmed by the gapless prefetch / a prior play). - if let Some(cached) = self.audio_cache.get(track_id) { - log::info!( - "[CAST-FETCH] Track {track_id} from MEMORY cache ({} bytes)", - cached.size_bytes - ); + if let Some((bytes, acquisition)) = self.cached_data_for_quality(track_id, quality) { + let meta = extract_audio_metadata_full(&bytes).ok()?; return Some(ExternalStreamAsset { - bytes: cached.data, - content_type: "audio/flac".to_string(), - quality: StreamQualityInfo::from_raw(0, None, None), - duration_secs: None, - origin: AssetOrigin::Cache, + bytes, + content_type: external_content_type("", acquisition.map(|q| q.resolved.id()).unwrap_or(6)), + quality: StreamQualityInfo::from_raw(acquisition.map(|q| q.resolved.id()).unwrap_or(0), + Some(meta.sample_rate as f64), meta.bit_depth), + duration_secs: None, origin: AssetOrigin::Cache, }); } - // L2: on-disk plain-FLAC playback cache; warm L1 on the way out. - if let Some(playback_cache) = self.audio_cache.get_playback_cache() { - if let Some(audio_data) = playback_cache.get(track_id) { - log::info!( - "[CAST-FETCH] Track {track_id} from DISK cache ({} bytes)", - audio_data.len() - ); - self.audio_cache.promote_from_disk(track_id, audio_data.clone()); - return Some(ExternalStreamAsset { - bytes: audio_data, - content_type: "audio/flac".to_string(), - quality: StreamQualityInfo::from_raw(0, None, None), - duration_secs: None, - origin: AssetOrigin::Cache, - }); - } - } // Cold: CMAF full download (Akamai CDN) -> decrypted FLAC. match qbz_qobuz::cmaf::download_full_with_quality(client, track_id, quality).await { @@ -6336,7 +6306,7 @@ impl Player { q.bit_depth ); // Warm L1 so a subsequent local replay skips the network. - self.audio_cache.insert(track_id, bytes.clone()); + self.audio_cache.insert_with_quality(track_id, bytes.clone(), qbz_cache::CacheQuality::from_resolved(quality, q.format_id)); return Some(ExternalStreamAsset { bytes, content_type: "audio/flac".to_string(), @@ -6369,7 +6339,7 @@ impl Player { q.format_id, content_type ); - self.audio_cache.insert(track_id, bytes.clone()); + self.audio_cache.insert_with_quality(track_id, bytes.clone(), qbz_cache::CacheQuality::from_resolved(quality, q.format_id)); Some(ExternalStreamAsset { bytes, content_type, @@ -6412,6 +6382,7 @@ impl Player { writer: BufferWriter, track_id: u64, cache: Arc, + cache_quality: Option, skip_cache: bool, _expected_total_bytes: u64, mut cancel_rx: tokio::sync::watch::Receiver, @@ -6551,7 +6522,7 @@ impl Player { ); if !skip_cache { - Self::persist_stream(cache, writer.clone(), track_id).await?; + Self::persist_stream(cache, writer.clone(), track_id, cache_quality).await?; } Ok(true) @@ -7425,6 +7396,8 @@ pub fn external_content_type(mime: &str, format_id: u32) -> String { #[cfg(test)] mod tests { + use super::{AudioMetadata, cached_metadata_below_requested, cached_quality_below_requested}; + use qbz_models::Quality; use super::compute_needs_new_stream; use super::{wedge_recovery, WedgeRecovery, WEDGE_REBUILD_BACKOFF}; use super::external_content_type; @@ -7440,6 +7413,19 @@ mod tests { }; use std::sync::atomic::Ordering; + #[test] + fn best_96khz_master_is_reused_only_with_matching_acquisition() { + let meta = AudioMetadata { sample_rate: 96_000, channels: 2, bit_depth: Some(24), + codec: symphonia::core::codecs::CODEC_TYPE_FLAC }; + let highest = qbz_cache::CacheQuality::from_resolved(Quality::UltraHiRes, 27); + let limited = qbz_cache::CacheQuality::from_resolved(Quality::HiRes, 7); + assert!(!cached_metadata_below_requested(&meta, Quality::UltraHiRes, highest)); + assert!(cached_metadata_below_requested(&meta, Quality::UltraHiRes, limited)); + assert!(cached_metadata_below_requested(&meta, Quality::UltraHiRes, None)); + assert!(!cached_metadata_below_requested(&meta, Quality::HiRes, limited)); + assert!(cached_quality_below_requested(b"corrupt", Quality::UltraHiRes, highest)); + } + #[test] fn duplicate_resume_preserves_live_clock_and_real_pause_can_resume() { let state = SharedState::new(); diff --git a/crates/qbz-qobuz/src/cmaf.rs b/crates/qbz-qobuz/src/cmaf.rs index 2512c90ac..2ae4250f9 100644 --- a/crates/qbz-qobuz/src/cmaf.rs +++ b/crates/qbz-qobuz/src/cmaf.rs @@ -243,27 +243,11 @@ pub async fn download_full_with_quality_progress( let setup = setup_streaming(client, track_id, quality).await?; let http = build_cdn_client()?; - let total_size: usize = setup.flac_header.len() - + setup.segment_table.iter().map(|s| s.byte_len as usize).sum::(); - - let segments = fetch_all_segments( - &http, - &setup.url_template, - setup.n_segments, - "CMAF-FULL", - on_progress, - ) - .await?; - - let mut output = Vec::with_capacity(total_size); - output.extend_from_slice(&setup.flac_header); - decrypt_segments_into(&segments, &setup.content_key, &mut output)?; - + let output = assemble_download(&http, &setup, on_progress).await?; log::info!( - "[CMAF-FULL] Track {} complete: {:.2} MB FLAC, expected {:.2} MB", + "[CMAF-FULL] Track {} complete: {:.2} MB FLAC (segment lengths verified)", track_id, output.len() as f64 / (1024.0 * 1024.0), - total_size as f64 / (1024.0 * 1024.0), ); // `from_raw` normalizes the rate unit (kHz vs Hz) defensively. @@ -275,6 +259,69 @@ pub async fn download_full_with_quality_progress( Ok((output, quality_info)) } +/// Hold at most three encrypted segments while assembling in source order. +/// Futures belong to this stream: cancellation/error drops outstanding HTTP +/// work, and a slow leading segment cannot accumulate the rest of the track. +async fn assemble_download( + http: &reqwest::Client, setup: &CmafStreamingInfo, + on_progress: Option, +) -> std::result::Result, String> { + use futures_util::{stream, StreamExt}; + if setup.segment_table.len() != usize::from(setup.n_segments) { + return Err("CMAF segment count disagrees with init table".into()); + } + let expected_size = setup.segment_table.iter().try_fold(setup.flac_header.len(), |total, entry| { + total.checked_add(entry.byte_len as usize).ok_or("CMAF output size overflow") + })?; + let mut output = Vec::new(); + output.try_reserve_exact(expected_size).map_err(|_| "CMAF output allocation failed")?; + output.extend_from_slice(&setup.flac_header); + let mut segments = stream::iter(1..=setup.n_segments).map(|index| async move { + let url = setup.url_template.replace("$SEGMENT$", &index.to_string()); + let data = fetch_cdn_bytes_with_retry(http, &url, "CMAF-FULL").await?; + // Preserve the existing CDN pacing per occupied request slot. + tokio::time::sleep(Duration::from_millis(500)).await; + Ok::<_, String>((index, data)) + }).buffered(CMAF_PREFETCH_CONCURRENCY); + while let Some(segment) = segments.next().await { + let (index, data) = segment?; + append_verified_segment(&data, &setup.content_key, + setup.segment_table[usize::from(index) - 1].byte_len as usize, &mut output)?; + if let Some(progress) = &on_progress { + progress(CmafProgressUpdate { segments_completed: u32::from(index), + n_segments: u32::from(setup.n_segments), bytes_this_segment: data.len() as u64 }); + } + } + if output.len() != expected_size { return Err("CMAF assembled size mismatch".into()); } + Ok(output) +} + +fn append_verified_segment(data: &[u8], key: &[u8; 16], expected: usize, output: &mut Vec) + -> std::result::Result<(), String> { + let crypto = qbz_cmaf::parse_segment_crypto(data).map_err(|e| e.to_string())?; + let size = crypto.mdat_end.checked_sub(crypto.data_offset).ok_or("CMAF invalid media bounds")?; + if crypto.mdat_end > data.len() || size != expected { + return Err("CMAF segment size disagrees with init table".into()); + } + let mut cursor = crypto.data_offset; + for entry in &crypto.entries { + cursor = cursor.checked_add(entry.size as usize).filter(|end| *end <= crypto.mdat_end) + .ok_or("CMAF frame exceeds media bounds")?; + } + // All bounds have been checked before the first mutation. Append and + // decrypt directly in the final allocation; no second decoded segment. + cursor = crypto.data_offset; + for entry in &crypto.entries { + let end = cursor + entry.size as usize; + let start = output.len(); + output.extend_from_slice(&data[cursor..end]); + if entry.flags != 0 { qbz_cmaf::decrypt_frame(key, &entry.iv, &mut output[start..]); } + cursor = end; + } + output.extend_from_slice(&data[cursor..crypto.mdat_end]); + Ok(()) +} + /// Download a track's complete CMAF stream and return it as a raw (still /// encrypted) bundle suitable for offline storage. /// @@ -591,6 +638,88 @@ mod tests { use std::sync::{Arc, Once}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + fn segment_fixture(bytes: &[u8], frame_size: u32) -> Vec { + let mut data = Vec::new(); + data.extend_from_slice(&44u32.to_be_bytes()); + data.extend_from_slice(b"uuid"); + data.extend_from_slice(&[0x3b,0x42,0x12,0x92,0x56,0xf3,0x5f,0x75,0x92,0x36,0x63,0xb6,0x9a,0x1f,0x52,0xb2]); + data.extend_from_slice(&0u32.to_be_bytes()); + data.extend_from_slice(&52u32.to_be_bytes()); + data.extend_from_slice(&[0, 0, 0, 1]); // no IV, one clear frame + data.extend_from_slice(&frame_size.to_be_bytes()); + data.extend_from_slice(&[0; 4]); // skip, flags + data.extend_from_slice(&(8u32 + bytes.len() as u32).to_be_bytes()); + data.extend_from_slice(b"mdat"); + data.extend_from_slice(bytes); + data + } + + #[test] + fn incremental_assembly_preserves_frames_and_clear_tail() { + let segment = segment_fixture(b"frame-tail", 5); + let mut output = b"fLaC".to_vec(); + append_verified_segment(&segment, &[0; 16], 10, &mut output).unwrap(); + assert_eq!(output, b"fLaCframe-tail"); + for (data, expected) in [(segment.clone(), 9), (segment_fixture(b"frame", 6), 5), + (segment[..segment.len()-1].to_vec(), 10)] { + let mut output = b"fLaC".to_vec(); + assert!(append_verified_segment(&data, &[0; 16], expected, &mut output).is_err()); + assert_eq!(output, b"fLaC"); + } + } + + #[tokio::test] + async fn incremental_download_bounds_out_of_order_requests_and_preserves_order() { + install_tls_provider(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let requested = Arc::new(AtomicU32::new(0)); + let observed = requested.clone(); + let release_first = Arc::new(tokio::sync::Notify::new()); + let release = release_first.clone(); + let server = tokio::spawn(async move { + let mut requests = tokio::task::JoinSet::new(); + for _ in 0..6 { + let (mut socket, _) = listener.accept().await.unwrap(); + let release = release.clone(); + let observed = observed.clone(); + requests.spawn(async move { + let mut request = [0; 1024]; + let len = socket.read(&mut request).await.unwrap(); + let request = std::str::from_utf8(&request[..len]).unwrap(); + let index: u8 = request.split_whitespace().nth(1).unwrap()[1..].parse().unwrap(); + observed.fetch_add(1, Ordering::SeqCst); + if index == 1 { release.notified().await; } + let body = segment_fixture(&[index], 1); + let header = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()); + socket.write_all(header.as_bytes()).await.unwrap(); + socket.write_all(&body).await.unwrap(); + }); + } + while let Some(result) = requests.join_next().await { result.unwrap(); } + }); + let download = tokio::spawn(async move { + let setup = CmafStreamingInfo { + url_template: format!("http://{address}/$SEGMENT$"), n_segments: 6, + content_key: [0; 16], flac_header: b"fLaC".to_vec(), + segment_table: vec![qbz_cmaf::SegmentTableEntry { byte_len: 1, sample_count: 1 }; 6], + format_id: 27, sampling_rate: Some(96_000), bit_depth: Some(24), init_fetch_ms: 0, + }; + assemble_download(&build_cdn_client().unwrap(), &setup, None).await.unwrap() + }); + tokio::time::timeout(Duration::from_secs(2), async { + while requested.load(Ordering::SeqCst) < 3 { tokio::task::yield_now().await; } + }).await.unwrap(); + // The following two requests have completed, but the leading request + // still owns its slot: no unbounded task/segment accumulation. + tokio::time::sleep(Duration::from_millis(600)).await; + assert_eq!(requested.load(Ordering::SeqCst), 3); + release_first.notify_one(); + let bytes = tokio::time::timeout(Duration::from_secs(4), download).await.unwrap().unwrap(); + assert_eq!(bytes, b"fLaC\x01\x02\x03\x04\x05\x06"); + server.await.unwrap(); + } + fn install_tls_provider() { static INSTALL: Once = Once::new(); INSTALL.call_once(|| {