fix(playback): retain acquisition quality across bounded audio storage

Pair cached audio with the requested and resolved quality, reject incompatible or unverified entries, and enforce disk budgets at startup and admission. Reuse the same policy for local, gapless and external playback while assembling CMAF segments in order with bounded concurrency.

cx

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::<usize>();
-
-    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<CmafProgressCallback>,
+) -> std::result::Result<Vec<u8>, 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<u8>)
+    -> 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<u8> {
+        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(|| {