fix(qconnect): bounded CMAF fallback when the CDN header-floods a track

Akamai answers small raw-url objects with ~106 headers, over hyper's
hard-coded 100-header h1 cap, so every reqwest fetch of those tracks
failed identically (streaming probe AND full download) and the track
never played — with the cause hidden by reqwest's Display. Now:
describe_reqwest_error() walks the source chain for logs; on the
header-flood signature the engine skips the doomed reqwest re-download
and falls back to play_track_resolved (CMAF, unaffected) plus an
explicit sync_current_to_id so the queue cursor follows the audio.
Any other failure keeps the previous behavior. Live-verified on the Pi
(DSOTM track 1 plays; gapless continues).

km
This commit is contained in:
vicrodh committed 2026-07-22 09:03:28 -06:00
1 parent 46b9cd7a3b
commit 48eaebef1d
4 files changed
+212 -42

No files matched your search

+69 -16
View File
@@ -48,6 +48,28 @@ impl SlintRendererEngine {
fn core(&self) -> &Arc<QbzCore<SlintAdapter>> {
self.runtime.core()
}
/// Last-resort load for tracks the raw-URL path cannot fetch (the CDN
/// header flood defeats every reqwest attempt — see
/// `remote_stream::is_header_flood_error`): the CMAF path is unaffected by
/// the h1 header cap. `play_track_resolved` does NOT move the queue cursor
/// (nothing on the QConnect path does — the shared driver's cursor sync
/// only fires on a playing->playing track edge), so sync it explicitly or
/// the now-playing truth keeps showing the PREVIOUS track while the
/// recovered one plays.
async fn play_via_cmaf(
&self,
track_id: u64,
quality: Quality,
start_position_secs: u64,
) -> Result<(), String> {
self.core()
.play_track_resolved(track_id, quality, None, None, start_position_secs)
.await
.map_err(|err| format!("CMAF fallback for remote track {track_id}: {err}"))?;
self.core().sync_current_to_id(track_id).await;
Ok(())
}
}
#[async_trait]
@@ -138,7 +160,7 @@ impl QconnectRendererEngine for SlintRendererEngine {
.map_err(|err| format!("resolve stream url for remote track {track_id}: {err}"))?;
let player = self.core().player();
match crate::remote_stream::stream_remote_track_into_player(
let stream_result = crate::remote_stream::stream_remote_track_into_player(
&player,
track_id,
duration_secs,
@@ -146,22 +168,46 @@ impl QconnectRendererEngine for SlintRendererEngine {
&stream_url.url,
"QConnect",
)
.await
{
Ok(()) => Ok(()),
Err(stream_err) => {
log::warn!(
"[QConnect] Streaming handoff unavailable for track {}: {}. Falling back to full download.",
track_id,
stream_err
);
let audio_data = download_remote_audio(&stream_url.url).await?;
.await;
let Err(stream_err) = stream_result else {
return Ok(());
};
// Akamai small-object header flood: SMALL raw-url objects come back
// with ~106 headers, over hyper's hard-coded 100-header h1 cap, so
// EVERY reqwest fetch of this URL fails — the full download would die
// the same death. Skip it and go straight to the CMAF last resort.
if crate::remote_stream::is_header_flood_error(&stream_err) {
log::warn!(
"[QConnect] Raw-URL streaming hit the CDN header flood for track {track_id}: {stream_err}. Skipping full download; last resort: CMAF."
);
return self
.play_via_cmaf(track_id, quality, start_position_secs)
.await;
}
log::warn!(
"[QConnect] Streaming handoff unavailable for track {}: {}. Falling back to full download.",
track_id,
stream_err
);
match download_remote_audio(&stream_url.url).await {
Ok(audio_data) => {
self.core()
.player()
.play_data(audio_data, track_id)
.map_err(|err| format!("play remote track {track_id}: {err}"))?;
Ok(())
}
Err(download_err) if crate::remote_stream::is_header_flood_error(&download_err) => {
log::warn!(
"[QConnect] Full download hit the CDN header flood for track {track_id}: {download_err}. Last resort: CMAF."
);
self.play_via_cmaf(track_id, quality, start_position_secs)
.await
}
Err(download_err) => Err(download_err),
}
}
@@ -181,7 +227,12 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("download remote audio request failed: {err}"))?;
.map_err(|err| {
format!(
"download remote audio request failed: {}",
crate::remote_stream::describe_reqwest_error(&err)
)
})?;
if !response.status().is_success() {
return Err(format!(
@@ -190,9 +241,11 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
));
}
let bytes = response
.bytes()
.await
.map_err(|err| format!("read remote audio bytes failed: {err}"))?;
let bytes = response.bytes().await.map_err(|err| {
format!(
"read remote audio bytes failed: {}",
crate::remote_stream::describe_reqwest_error(&err)
)
})?;
Ok(bytes.to_vec())
}
+37 -5
View File
@@ -106,7 +106,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("probe HEAD request failed: {err}"))?;
.map_err(|err| format!("probe HEAD request failed: {}", describe_reqwest_error(&err)))?;
if !head_response.status().is_success() {
return Err(format!(
@@ -129,7 +129,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
.header("Range", "bytes=0-65535")
.send()
.await
.map_err(|err| format!("probe range request failed: {err}"))?;
.map_err(|err| format!("probe range request failed: {}", describe_reqwest_error(&err)))?;
if !range_response.status().is_success() {
return Err(format!(
@@ -141,7 +141,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
let initial_bytes = range_response
.bytes()
.await
.map_err(|err| format!("read probe bytes failed: {err}"))?;
.map_err(|err| format!("read probe bytes failed: {}", describe_reqwest_error(&err)))?;
let elapsed = start_time.elapsed();
let speed_mbps = if elapsed.as_secs_f64() > 0.0 {
@@ -214,7 +214,12 @@ pub async fn download_and_stream_remote_track(
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("start remote streaming request failed: {err}"))?;
.map_err(|err| {
format!(
"start remote streaming request failed: {}",
describe_reqwest_error(&err)
)
})?;
if !response.status().is_success() {
return Err(format!(
@@ -229,7 +234,8 @@ pub async fn download_and_stream_remote_track(
let mut last_log_time = Instant::now();
while let Some(chunk_result) = stream.next().await {
let chunk = chunk_result.map_err(|err| format!("remote streaming chunk failed: {err}"))?;
let chunk = chunk_result
.map_err(|err| format!("remote streaming chunk failed: {}", describe_reqwest_error(&err)))?;
bytes_received += chunk.len() as u64;
if let Err(err) = writer.push_chunk(&chunk) {
@@ -284,3 +290,29 @@ pub async fn download_and_stream_remote_track(
Ok(())
}
/// reqwest's `Display` hides the source chain — which is exactly where the
/// diagnosis lives (Akamai's >100-header small-object flood surfaces as hyper's
/// "message head is too large" two levels down). Walk `source()` and join the
/// chain so logs AND signature matching see the real cause.
pub fn describe_reqwest_error(err: &reqwest::Error) -> String {
use std::error::Error as _;
let mut out = err.to_string();
let mut source = err.source();
while let Some(cause) = source {
out.push_str(": ");
out.push_str(&cause.to_string());
source = cause.source();
}
out
}
/// True when an error message (already chain-expanded by
/// [`describe_reqwest_error`]) shows hyper's hard-coded h1 100-header cap.
/// Akamai answers SMALL raw-url objects with ~106 headers (the `X-AK-GRN` /
/// `X-AK-FWD-ERROR: ERR_POC_FWD_OBJ_TOO_SMALL` flood), so EVERY reqwest fetch
/// of such an URL fails this way — streaming probe and full download alike.
pub fn is_header_flood_error(message: &str) -> bool {
let haystack = message.to_ascii_lowercase();
haystack.contains("message head is too large") || haystack.contains("too many headers")
}
+69 -16
View File
@@ -97,6 +97,28 @@ impl DaemonRendererEngine {
fn core(&self) -> &Arc<QbzCore<DaemonAdapter>> {
self.runtime.core()
}
/// Last-resort load for tracks the raw-URL path cannot fetch (the CDN
/// header flood defeats every reqwest attempt — see
/// `remote_stream::is_header_flood_error`): the CMAF path is unaffected by
/// the h1 header cap. `play_track_resolved` does NOT move the queue cursor
/// (nothing on the QConnect path does — the shared driver's cursor sync
/// only fires on a playing->playing track edge), so sync it explicitly or
/// `qbzd status` / the local now-playing truth keep showing the PREVIOUS
/// track while the recovered one plays.
async fn play_via_cmaf(
&self,
track_id: u64,
quality: Quality,
start_position_secs: u64,
) -> Result<(), String> {
self.core()
.play_track_resolved(track_id, quality, None, None, start_position_secs)
.await
.map_err(|err| format!("CMAF fallback for remote track {track_id}: {err}"))?;
self.core().sync_current_to_id(track_id).await;
Ok(())
}
}
#[async_trait]
@@ -198,7 +220,7 @@ impl QconnectRendererEngine for DaemonRendererEngine {
.map_err(|err| format!("resolve stream url for remote track {track_id}: {err}"))?;
let player = self.core().player();
match super::remote_stream::stream_remote_track_into_player(
let stream_result = super::remote_stream::stream_remote_track_into_player(
&player,
track_id,
duration_secs,
@@ -206,22 +228,46 @@ impl QconnectRendererEngine for DaemonRendererEngine {
&stream_url.url,
"QConnect",
)
.await
{
Ok(()) => Ok(()),
Err(stream_err) => {
log::warn!(
"[QConnect] Streaming handoff unavailable for track {}: {}. Falling back to full download.",
track_id,
stream_err
);
let audio_data = download_remote_audio(&stream_url.url).await?;
.await;
let Err(stream_err) = stream_result else {
return Ok(());
};
// Akamai small-object header flood: SMALL raw-url objects come back
// with ~106 headers, over hyper's hard-coded 100-header h1 cap, so
// EVERY reqwest fetch of this URL fails — the full download would die
// the same death. Skip it and go straight to the CMAF last resort.
if super::remote_stream::is_header_flood_error(&stream_err) {
log::warn!(
"[QConnect] Raw-URL streaming hit the CDN header flood for track {track_id}: {stream_err}. Skipping full download; last resort: CMAF."
);
return self
.play_via_cmaf(track_id, quality, start_position_secs)
.await;
}
log::warn!(
"[QConnect] Streaming handoff unavailable for track {}: {}. Falling back to full download.",
track_id,
stream_err
);
match download_remote_audio(&stream_url.url).await {
Ok(audio_data) => {
self.core()
.player()
.play_data(audio_data, track_id)
.map_err(|err| format!("play remote track {track_id}: {err}"))?;
Ok(())
}
Err(download_err) if super::remote_stream::is_header_flood_error(&download_err) => {
log::warn!(
"[QConnect] Full download hit the CDN header flood for track {track_id}: {download_err}. Last resort: CMAF."
);
self.play_via_cmaf(track_id, quality, start_position_secs)
.await
}
Err(download_err) => Err(download_err),
}
}
@@ -241,7 +287,12 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("download remote audio request failed: {err}"))?;
.map_err(|err| {
format!(
"download remote audio request failed: {}",
super::remote_stream::describe_reqwest_error(&err)
)
})?;
if !response.status().is_success() {
return Err(format!(
@@ -250,10 +301,12 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
));
}
let bytes = response
.bytes()
.await
.map_err(|err| format!("read remote audio bytes failed: {err}"))?;
let bytes = response.bytes().await.map_err(|err| {
format!(
"read remote audio bytes failed: {}",
super::remote_stream::describe_reqwest_error(&err)
)
})?;
Ok(bytes.to_vec())
}
+37 -5
View File
@@ -108,7 +108,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("probe HEAD request failed: {err}"))?;
.map_err(|err| format!("probe HEAD request failed: {}", describe_reqwest_error(&err)))?;
if !head_response.status().is_success() {
return Err(format!(
@@ -131,7 +131,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
.header("Range", "bytes=0-65535")
.send()
.await
.map_err(|err| format!("probe range request failed: {err}"))?;
.map_err(|err| format!("probe range request failed: {}", describe_reqwest_error(&err)))?;
if !range_response.status().is_success() {
return Err(format!(
@@ -143,7 +143,7 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
let initial_bytes = range_response
.bytes()
.await
.map_err(|err| format!("read probe bytes failed: {err}"))?;
.map_err(|err| format!("read probe bytes failed: {}", describe_reqwest_error(&err)))?;
let elapsed = start_time.elapsed();
let speed_mbps = if elapsed.as_secs_f64() > 0.0 {
@@ -217,7 +217,12 @@ pub async fn download_and_stream_remote_track(
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("start remote streaming request failed: {err}"))?;
.map_err(|err| {
format!(
"start remote streaming request failed: {}",
describe_reqwest_error(&err)
)
})?;
if !response.status().is_success() {
return Err(format!(
@@ -232,7 +237,8 @@ pub async fn download_and_stream_remote_track(
let mut last_log_time = Instant::now();
while let Some(chunk_result) = stream.next().await {
let chunk = chunk_result.map_err(|err| format!("remote streaming chunk failed: {err}"))?;
let chunk = chunk_result
.map_err(|err| format!("remote streaming chunk failed: {}", describe_reqwest_error(&err)))?;
bytes_received += chunk.len() as u64;
if let Err(err) = writer.push_chunk(&chunk) {
@@ -287,3 +293,29 @@ pub async fn download_and_stream_remote_track(
Ok(())
}
/// reqwest's `Display` hides the source chain — which is exactly where the
/// diagnosis lives (Akamai's >100-header small-object flood surfaces as hyper's
/// "message head is too large" two levels down). Walk `source()` and join the
/// chain so logs AND signature matching see the real cause.
pub fn describe_reqwest_error(err: &reqwest::Error) -> String {
use std::error::Error as _;
let mut out = err.to_string();
let mut source = err.source();
while let Some(cause) = source {
out.push_str(": ");
out.push_str(&cause.to_string());
source = cause.source();
}
out
}
/// True when an error message (already chain-expanded by
/// [`describe_reqwest_error`]) shows hyper's hard-coded h1 100-header cap.
/// Akamai answers SMALL raw-url objects with ~106 headers (the `X-AK-GRN` /
/// `X-AK-FWD-ERROR: ERR_POC_FWD_OBJ_TOO_SMALL` flood), so EVERY reqwest fetch
/// of such an URL fails this way — streaming probe and full download alike.
pub fn is_header_flood_error(message: &str) -> bool {
let haystack = message.to_ascii_lowercase();
haystack.contains("message head is too large") || haystack.contains("too many headers")
}