Joren/qbz
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(player): preserve audible state on source failures

Browse Source 
Preflight fallible queue transitions and publish now-playing state only after the requested source is audible, leaving the prior stream intact when a file or server is unavailable.

Carry source-owned request headers through remote probes, streaming and fallback downloads without exposing sensitive values.

cx
This commit is contained in:
vicrodh committed 2026-08-25 18:26:35 -06:00
1 parent d244213a64
commit ecae778d58
7 files changed
+526 -150
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/qbz-core/src/core.rs
+6
View File
@@ -866,6 +866,12 @@ impl<A: FrontendAdapter + Send + Sync + 'static> QbzCore<A> {
queue.peek_upcoming(count)
}
/// Inspect the previous queue row without changing the cursor/history.
pub async fn peek_previous_track(&self) -> Option<QueueTrack> {
let queue = self.queue.read().await;
queue.peek_previous()
}
/// The current queue track, if any (source-aware playback routing).
pub async fn current_track(&self) -> Option<QueueTrack> {
let queue = self.queue.read().await;
crates/qbz-player/src/queue.rs
+48
View File
@@ -777,6 +777,39 @@ impl QueueManager {
result
}
/// Inspect the track [`Self::previous`] would select without consuming
/// history or changing either queue cursor. Frontends use this before a
/// fallible physical-file handoff: publishing the previous row first can
/// leave metadata claiming a moved file while the old stream is audible.
pub fn peek_previous(&self) -> Option<QueueTrack> {
let state = self.state.lock().unwrap();
if state.tracks.is_empty() {
return None;
}
if let Some(&prev_idx) = state.history.back() {
return state.tracks.get(prev_idx).cloned();
}
let prev_idx = if state.shuffle {
if state.shuffle_position > 0 {
state.shuffle_order.get(state.shuffle_position - 1).copied()
} else if state.repeat == RepeatMode::All {
state.shuffle_order.last().copied()
} else {
state.shuffle_order.first().copied()
}
} else {
let curr_idx = state.current_index.unwrap_or(0);
if curr_idx > 0 {
Some(curr_idx - 1)
} else if state.repeat == RepeatMode::All {
Some(state.tracks.len().saturating_sub(1))
} else {
Some(0)
}
};
prev_idx.and_then(|idx| state.tracks.get(idx).cloned())
}
/// Advance to next track and return it
pub fn next(&self) -> Option<QueueTrack> {
let mut state = self.state.lock().unwrap();
@@ -1816,6 +1849,21 @@ mod tests {
assert_eq!(track.id, 4);
}
#[test]
fn peek_previous_matches_previous_without_consuming_history() {
let queue = QueueManager::new();
for i in 1..=4 {
queue.add_track(create_test_track(i));
}
queue.play_index(0);
assert_eq!(queue.next().expect("next").id, 2);
assert_eq!(queue.next().expect("next").id, 3);
assert_eq!(queue.peek_previous().expect("peek").id, 2);
assert_eq!(queue.peek_previous().expect("second peek").id, 2);
assert_eq!(queue.previous().expect("previous").id, 2);
}
#[test]
fn test_play_upcoming_at_with_shuffle_follows_shuffle_order() {
let queue = QueueManager::new();
crates/qbz-player/src/remote_stream.rs
+123 -17
View File
@@ -63,7 +63,35 @@ pub async fn stream_remote_track_into_player(
url: &str,
log_tag: &str,
) -> Result<(), String> {
let stream_info = probe_remote_stream_info(url).await?;
stream_remote_track_into_player_with_headers(
player,
track_id,
duration_secs,
start_position_secs,
url,
log_tag,
&[],
)
.await
}
/// Variant of [`stream_remote_track_into_player`] that preserves a source's
/// HTTP request contract while keeping this transport source-agnostic.
///
/// Plex media parts are the motivating case: their token is in the URL, but
/// some servers still reject the Range/body request unless the same
/// `X-Plex-*` identity used for metadata is present. Header values are never
/// included in errors or logs.
pub async fn stream_remote_track_into_player_with_headers(
player: &Player,
track_id: u64,
duration_secs: u64,
start_position_secs: u64,
url: &str,
log_tag: &str,
request_headers: &[(String, String)],
) -> Result<(), String> {
let stream_info = probe_remote_stream_info_with_headers(url, request_headers).await?;
log::info!(
"[{}/STREAMING] Track {} - {:.2} MB, {}Hz, {} ch, {}-bit {}, {:.1} MB/s",
log_tag,
@@ -92,9 +120,17 @@ pub async fn stream_remote_track_into_player(
let url = url.to_string();
let content_length = stream_info.content_length;
let log_tag = log_tag.to_string();
let request_headers = request_headers.to_vec();
tokio::spawn(async move {
if let Err(err) =
download_and_stream_remote_track(&url, writer, track_id, content_length, &log_tag).await
if let Err(err) = download_and_stream_remote_track_with_headers(
&url,
writer,
track_id,
content_length,
&log_tag,
&request_headers,
)
.await
{
log::error!(
"[{}/STREAMING] Track {} failed while streaming: {}",
@@ -118,6 +154,13 @@ pub async fn stream_remote_track_into_player(
/// than downloading the track once as a "probe" and then a second time into
/// the player.
pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, String> {
probe_remote_stream_info_with_headers(url, &[]).await
}
pub async fn probe_remote_stream_info_with_headers(
url: &str,
request_headers: &[(String, String)],
) -> Result<RemoteStreamInfo, String> {
use futures_util::StreamExt;
use reqwest::header::{ACCEPT_ENCODING, CONTENT_LENGTH, CONTENT_RANGE};
use reqwest::StatusCode;
@@ -130,14 +173,20 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
.map_err(|err| format!("create stream probe client: {err}"))?;
let start_time = Instant::now();
let range_response = client
let request = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.header(ACCEPT_ENCODING, "identity")
.header("Range", "bytes=0-65535")
.header("Range", "bytes=0-65535");
let range_response = apply_request_headers(request, request_headers)?
.send()
.await
.map_err(|err| format!("probe range request failed: {}", describe_reqwest_error(&err)))?;
.map_err(|err| {
format!(
"probe range request failed: {}",
describe_reqwest_error(&err)
)
})?;
if !range_response.status().is_success() {
return Err(format!(
@@ -162,7 +211,9 @@ pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, Str
let mut initial_bytes = Vec::with_capacity(PROBE_BYTES);
let mut stream = range_response.bytes_stream();
while initial_bytes.len() < PROBE_BYTES {
let Some(chunk) = stream.next().await else { break };
let Some(chunk) = stream.next().await else {
break;
};
let chunk = chunk
.map_err(|err| format!("read probe bytes failed: {}", describe_reqwest_error(&err)))?;
let take = (PROBE_BYTES - initial_bytes.len()).min(chunk.len());
@@ -251,6 +302,25 @@ pub async fn download_and_stream_remote_track(
track_id: u64,
content_length: u64,
log_tag: &str,
) -> Result<(), String> {
download_and_stream_remote_track_with_headers(
url,
writer,
track_id,
content_length,
log_tag,
&[],
)
.await
}
pub async fn download_and_stream_remote_track_with_headers(
url: &str,
writer: BufferWriter,
track_id: u64,
content_length: u64,
log_tag: &str,
request_headers: &[(String, String)],
) -> Result<(), String> {
use futures_util::StreamExt;
use reqwest::header::ACCEPT_ENCODING;
@@ -281,10 +351,11 @@ pub async fn download_and_stream_remote_track(
.build()
.map_err(|err| format!("create remote streaming client: {err}"))?;
let response = client
let request = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.header(ACCEPT_ENCODING, "identity")
.header(ACCEPT_ENCODING, "identity");
let response = apply_request_headers(request, request_headers)?
.send()
.await
.map_err(|err| {
@@ -307,8 +378,12 @@ 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: {}", describe_reqwest_error(&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) {
@@ -364,6 +439,29 @@ pub async fn download_and_stream_remote_track(
Ok(())
}
fn apply_request_headers(
request: reqwest::RequestBuilder,
headers: &[(String, String)],
) -> Result<reqwest::RequestBuilder, String> {
Ok(request.headers(parsed_request_headers(headers)?))
}
fn parsed_request_headers(
headers: &[(String, String)],
) -> Result<reqwest::header::HeaderMap, String> {
use reqwest::header::{HeaderName, HeaderValue};
let mut parsed = reqwest::header::HeaderMap::new();
for (name, value) in headers {
let header_name = HeaderName::from_bytes(name.as_bytes())
.map_err(|_| format!("invalid remote request header name {name:?}"))?;
let header_value = HeaderValue::from_str(value)
.map_err(|_| format!("invalid value for remote request header {name:?}"))?;
parsed.insert(header_name, header_value);
}
Ok(parsed)
}
/// 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
@@ -392,7 +490,19 @@ pub fn is_header_flood_error(message: &str) -> bool {
#[cfg(test)]
mod tests {
use super::{content_range_total, probe_content_length};
use super::{content_range_total, parsed_request_headers, probe_content_length};
#[test]
fn source_request_headers_are_parsed_for_the_http_request() {
let headers =
parsed_request_headers(&[("X-Plex-Product".to_string(), "QBZ".to_string())]).unwrap();
assert_eq!(
headers
.get("X-Plex-Product")
.and_then(|value| value.to_str().ok()),
Some("QBZ")
);
}
#[test]
fn content_range_uses_the_whole_object_length() {
@@ -406,11 +516,7 @@ mod tests {
#[test]
fn partial_content_length_is_never_mistaken_for_the_track_size() {
assert_eq!(
probe_content_length(
true,
Some("bytes 0-65535/44790678"),
Some("65536")
),
probe_content_length(true, Some("bytes 0-65535/44790678"), Some("65536")),
Some(44_790_678)
);
assert_eq!(probe_content_length(true, None, Some("65536")), None);
crates/qbz-qt/src/audible_qt.rs
+65 -19
View File
@@ -129,7 +129,10 @@ pub(crate) async fn play_ticket(runtime: &Runtime, ticket: PlaybackTicket) -> bo
PlaybackTicket::Bytes { bytes, play_id } => {
let len = bytes.len();
qbz_audio::register_seek_waveform_key(play_id, qbz_audio::seek_waveform_content_key(&bytes));
qbz_audio::register_seek_waveform_key(
play_id,
qbz_audio::seek_waveform_content_key(&bytes),
);
if let Err(e) = runtime.core().player().play_data(bytes, play_id) {
log::error!("[qbz-qt] audible: play_data {play_id} ({len} B) failed: {e}");
return false;
@@ -148,7 +151,19 @@ pub(crate) async fn play_ticket(runtime: &Runtime, ticket: PlaybackTicket) -> bo
duration_secs,
start_secs,
log_tag,
} => play_stream(runtime, &url, play_id, duration_secs, start_secs, log_tag).await,
request_headers,
} => {
play_stream(
runtime,
&url,
play_id,
duration_secs,
start_secs,
log_tag,
&request_headers,
)
.await
}
PlaybackTicket::SeekLoaded { play_id, secs } => {
let _ = runtime.core().player().seek(secs as u64);
@@ -208,9 +223,14 @@ pub(crate) async fn queue_gapless_successor(
| PlaybackTicket::CdTrack { .. }
| PlaybackTicket::Catalog { .. } => return Ok(false),
PlaybackTicket::Bytes { bytes, .. } => Some(bytes),
PlaybackTicket::Stream { url, log_tag, .. } => {
PlaybackTicket::Stream {
url,
log_tag,
request_headers,
..
} => {
let started = std::time::Instant::now();
let bytes = fetch_body(&url).await?;
let bytes = fetch_body(&url, &request_headers).await?;
log::info!(
"[qbz-qt][GAPLESS] {log_tag} successor {next_id}: {} bytes in {:?}",
bytes.len(),
@@ -241,9 +261,7 @@ pub(crate) async fn queue_gapless_successor(
if player.state.current_track_id() != predecessor_id
|| player.state.get_gapless_next_track_id() != 0
{
log::debug!(
"[qbz-qt][GAPLESS] successor {next_id} arrived after its predecessor moved"
);
log::debug!("[qbz-qt][GAPLESS] successor {next_id} arrived after its predecessor moved");
return Ok(false);
}
qbz_audio::register_seek_waveform_key(next_id, qbz_audio::seek_waveform_content_key(&bytes));
@@ -252,12 +270,7 @@ pub(crate) async fn queue_gapless_successor(
/// Read a file and hand the bytes to the player, with the CUE fast path in
/// front of it.
async fn play_file(
runtime: &Runtime,
path: PathBuf,
play_id: u64,
seek_secs: Option<f64>,
) -> bool {
async fn play_file(runtime: &Runtime, path: PathBuf, play_id: u64, seek_secs: Option<f64>) -> bool {
// CUE fast path — see the module header for why this test changed shape.
if let Some(start) = seek_secs.filter(|s| *s > 0.0) {
if container_is_loaded(runtime, &path) {
@@ -336,6 +349,7 @@ async fn play_stream(
duration_secs: u64,
start_secs: u64,
log_tag: &'static str,
request_headers: &[(String, String)],
) -> bool {
qbz_audio::register_seek_waveform_key(play_id, format!("remote:{log_tag}:{play_id}"));
let t0 = std::time::Instant::now();
@@ -346,13 +360,14 @@ async fn play_stream(
// big file" and "slow because it left the building".
let origin = url.split('?').next().unwrap_or("").to_string();
match qbz_player::remote_stream::stream_remote_track_into_player(
match qbz_player::remote_stream::stream_remote_track_into_player_with_headers(
&runtime.core().player(),
play_id,
duration_secs,
start_secs,
url,
log_tag,
request_headers,
)
.await
{
@@ -374,7 +389,7 @@ async fn play_stream(
),
}
let bytes = match fetch_body(url).await {
let bytes = match fetch_body(url, request_headers).await {
Ok(b) => b,
Err(e) => {
log::error!("[qbz-qt] {log_tag} play {play_id}: download failed ({e}) — {origin}");
@@ -406,12 +421,28 @@ async fn play_stream(
/// Navidrome 0.63.2, 2026-08-20). Handing that to `play_data` would push a JSON
/// blob into the decoder. So the body is also checked for a plausible audio
/// length before it is allowed anywhere near the player.
async fn fetch_body(url: &str) -> Result<Vec<u8>, String> {
let resp = reqwest::get(url)
async fn fetch_body(url: &str, request_headers: &[(String, String)]) -> Result<Vec<u8>, String> {
use reqwest::header::{HeaderName, HeaderValue};
let client = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(10))
.timeout(std::time::Duration::from_secs(300))
.build()
.map_err(|e| format!("create download client failed: {e}"))?;
let mut request = client.get(url);
for (name, value) in request_headers {
let name = HeaderName::from_bytes(name.as_bytes())
.map_err(|_| format!("invalid remote request header name {name:?}"))?;
let value = HeaderValue::from_str(value)
.map_err(|_| format!("invalid value for remote request header {name:?}"))?;
request = request.header(name, value);
}
let resp = request
.send()
.await
.map_err(|e| format!("request failed: {e}"))?
.map_err(|e| format!("request failed: {}", safe_reqwest_error(&e)))?
.error_for_status()
.map_err(|e| format!("status error: {e}"))?;
.map_err(|e| format!("status error: {}", safe_reqwest_error(&e)))?;
let ctype = resp
.headers()
.get(reqwest::header::CONTENT_TYPE)
@@ -435,6 +466,21 @@ async fn fetch_body(url: &str) -> Result<Vec<u8>, String> {
Ok(bytes)
}
/// Do not let a token-bearing media URL leak through reqwest's Display impl.
fn safe_reqwest_error(error: &reqwest::Error) -> String {
if error.is_timeout() {
"request timed out".into()
} else if let Some(status) = error.status() {
format!("server returned HTTP {}", status.as_u16())
} else if error.is_connect() {
"connection failed".into()
} else if error.is_body() {
"response body failed".into()
} else {
"request failed".into()
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/qbz-qt/src/playback_qt.rs
+216 -97
View File
@@ -52,9 +52,9 @@
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use cxx_qt_lib::QString;
use qbz_app::shell::AppRuntime;
use qbz_core::LoggingAdapter;
use cxx_qt_lib::QString;
use qbz_models::{PageArtistTrack, Quality, QueueTrack, RepeatMode};
/// The request tier for every play. Persisted in ui_prefs.json ("streaming_quality")
@@ -97,12 +97,18 @@ pub(crate) fn current_quality() -> Quality {
/// probes hardware. That is what lets the play funnel own the decision
/// instead of taking it as a parameter.
pub(crate) fn local_playback_quality() -> (Quality, qbz_models::QualityLimit) {
reconcile_device_cap(current_quality(), qbz_app::device_cap::cap().map(|(t, _)| t))
reconcile_device_cap(
current_quality(),
qbz_app::device_cap::cap().map(|(t, _)| t),
)
}
/// The decision itself, lifted out of the two lock reads so it is testable.
/// Port of the match in `crates/qbz/src/playback.rs local_playback_quality`.
fn reconcile_device_cap(pref: Quality, cap: Option<Quality>) -> (Quality, qbz_models::QualityLimit) {
fn reconcile_device_cap(
pref: Quality,
cap: Option<Quality>,
) -> (Quality, qbz_models::QualityLimit) {
use qbz_models::QualityLimit;
match cap {
Some(cap) if cap < pref => (cap, QualityLimit::LocalDeviceCap),
@@ -345,7 +351,10 @@ impl PlayContext {
if kind.is_empty() || id.is_empty() {
return None;
}
Some(Self { kind: kind.to_string(), id: id.to_string() })
Some(Self {
kind: kind.to_string(),
id: id.to_string(),
})
}
pub fn album(id: &str) -> Option<Self> {
@@ -495,7 +504,8 @@ pub(crate) fn queue_track_unavailable(track: &QueueTrack) -> bool {
// streamable — which is the owner's test case exactly, since `/album/get`
// is honest but plenty of rows only fail at the stream-url step. The cache
// exemption applies to both clauses: downloaded bytes play regardless.
let dead = !track.streamable || crate::track_replace_qt::session_unavailable::contains(track.id);
let dead =
!track.streamable || crate::track_replace_qt::session_unavailable::contains(track.id);
dead && !crate::offline_qt::is_cached_id(track.id)
}
@@ -646,26 +656,33 @@ pub(crate) struct QueueStart {
pub track_id: u64,
}
/// The ONLY `core().set_queue` call in this module — every play path goes
/// through it, so the origin can never be dropped on the floor, and neither can
/// the blacklist (PARITY-DEBT #1) nor a track Qobuz pulled (contract §5.3 D5).
/// A replacement queue after blacklist/availability filtering and context
/// stamping, but before it is visible to the core or any UI surface.
///
/// Returns the surviving anchor; see [`QueueStart`] for why no caller may
/// compute it itself any more.
pub(crate) async fn set_queue_stamped(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
/// Source-owned playback can fail before the player accepts new audio (dead
/// Plex part, unavailable NAS, server outage). Preparing first lets those
/// callers prove the anchor is audible without making Queue, NPB and MPRIS
/// announce it over the track that is still actually playing.
pub(crate) struct PreparedQueue {
tracks: Vec<QueueTrack>,
start: Option<usize>,
anchor: Option<QueueStart>,
unavailable_dropped: usize,
}
impl PreparedQueue {
pub(crate) fn anchor_track(&self) -> Option<&QueueTrack> {
self.start.and_then(|index| self.tracks.get(index))
}
}
pub(crate) fn prepare_queue_stamped(
tracks: Vec<QueueTrack>,
start: Option<usize>,
context: Option<PlayContext>,
) -> Option<QueueStart> {
) -> Option<PreparedQueue> {
let asked = tracks.len();
// Filter BEFORE stamping: a dropped row needs no origin, and the start
// index has to be remapped against the list the core actually receives.
let (mut tracks, start, dropped) = filter_unplayable_queue(tracks, start);
// Everything was filtered out of a non-empty build: do NOT publish an empty
// queue over whatever is playing (§5.3, last bullet). Say why and leave the
// core alone — wiping a working queue because the new one was all-dead is a
// second failure stacked on the first.
if tracks.is_empty() && asked > 0 {
log::info!(
"[qbz-qt] set_queue: all {asked} track(s) filtered ({} blacklisted, {} unavailable) \
@@ -678,27 +695,54 @@ pub(crate) async fn set_queue_stamped(
}
stamp_context(&mut tracks, context);
warn_dead_context(&tracks, "set_queue");
// Read the anchor off the list the CORE receives, not off the caller's —
// that difference IS F1. Computed before the move into `set_queue`.
let anchor = start.and_then(|index| {
tracks.get(index).map(|track| QueueStart {
index,
track_id: track.id,
})
});
Some(PreparedQueue {
tracks,
start,
anchor,
unavailable_dropped: dropped.unavailable,
})
}
pub(crate) async fn commit_prepared_queue(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
prepared: PreparedQueue,
) -> Option<QueueStart> {
let PreparedQueue {
tracks,
start,
anchor,
unavailable_dropped,
} = prepared;
runtime.core().set_queue(tracks, start).await;
toast_unavailable_dropped(dropped.unavailable);
// QConnect (contract §6.3, Slint playback.rs:3697-3703): when WE are the
// active renderer, push the new queue to the session immediately so a
// controller's UI reflects the freshly-built queue — covers infinite-play
// refills + album/radio plays. Self-gates to a no-op when not connected
// or when a peer owns playback.
toast_unavailable_dropped(unavailable_dropped);
if let Some(svc) = crate::qconnect_qt::service() {
svc.sync_local_queue_if_changed().await;
}
anchor
}
/// The ONLY `core().set_queue` call in this module — every play path goes
/// through it, so the origin can never be dropped on the floor, and neither can
/// the blacklist (PARITY-DEBT #1) nor a track Qobuz pulled (contract §5.3 D5).
///
/// Returns the surviving anchor; see [`QueueStart`] for why no caller may
/// compute it itself any more.
pub(crate) async fn set_queue_stamped(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
tracks: Vec<QueueTrack>,
start: Option<usize>,
context: Option<PlayContext>,
) -> Option<QueueStart> {
let prepared = prepare_queue_stamped(tracks, start, context)?;
commit_prepared_queue(runtime, prepared).await
}
/// Same guarantee for the ADD paths (play-next / add-to-queue): appended tracks
/// carry their own origin, so the glyph stays right after the queue advances
/// into them.
@@ -765,7 +809,10 @@ pub(crate) fn batch_all_qconnect_castable(tracks: &[QueueTrack]) -> bool {
tracks.iter().all(|qt| {
qt.id > 0
&& !matches!(
qt.source.as_deref().map(|s| s.to_ascii_lowercase()).as_deref(),
qt.source
.as_deref()
.map(|s| s.to_ascii_lowercase())
.as_deref(),
Some("local") | Some("plex")
)
})
@@ -800,10 +847,8 @@ pub(crate) async fn route_enqueue_to_peer(tracks: &[QueueTrack], mode: &str) ->
let Some(svc) = crate::qconnect_qt::service() else {
return false;
};
let routed: Vec<(u64, Option<String>)> = tracks
.iter()
.map(|qt| (qt.id, qt.source.clone()))
.collect();
let routed: Vec<(u64, Option<String>)> =
tracks.iter().map(|qt| (qt.id, qt.source.clone())).collect();
match mode {
"next" => svc.play_next_batch_on_peer_if_active(&routed).await,
"later" => svc.play_later_batch_on_peer_if_active(&routed).await,
@@ -818,9 +863,18 @@ pub(crate) async fn route_track_to_peer(track: &QueueTrack, mode: &str) -> bool
return false;
};
match mode {
"next" => svc.play_next_on_peer_if_active(track.id, track.source.as_deref()).await,
"later" => svc.play_later_on_peer_if_active(track.id, track.source.as_deref()).await,
_ => svc.add_to_queue_on_peer_if_active(track.id, track.source.as_deref()).await,
"next" => {
svc.play_next_on_peer_if_active(track.id, track.source.as_deref())
.await
}
"later" => {
svc.play_later_on_peer_if_active(track.id, track.source.as_deref())
.await
}
_ => {
svc.add_to_queue_on_peer_if_active(track.id, track.source.as_deref())
.await
}
}
}
@@ -896,9 +950,7 @@ pub(crate) async fn play_resolved_offline_aware(
// `is_terminal_unavailable` routes it into the SAME bounded
// skip walk a pulled Qobuz track takes — to the user the two
// failures are the same one.
Ok(false) => Err(format!(
"local file no longer available (track {track_id})"
)),
Ok(false) => Err(format!("local file no longer available (track {track_id})")),
Err(e) => Err(e.to_string()),
};
}
@@ -1010,9 +1062,9 @@ pub(crate) async fn fetch_album_queue(
return Ok(tracks);
}
Ok(_) => {}
Err(e) => log::debug!(
"[qbz-qt] play_album {album_id}: offline fallback unavailable: {e}"
),
Err(e) => {
log::debug!("[qbz-qt] play_album {album_id}: offline fallback unavailable: {e}")
}
}
return Err(format!("get_album {album_id} failed: {catalog_error}"));
}
@@ -1142,7 +1194,9 @@ async fn offline_album_queue(album_id: &str) -> Result<Vec<QueueTrack>, String>
let sample_rate = cached
.sample_rate
.or_else(|| local.map(|row| row.sample_rate).filter(|rate| *rate > 0.0));
let bit_depth = cached.bit_depth.or_else(|| local.and_then(|row| row.bit_depth));
let bit_depth = cached
.bit_depth
.or_else(|| local.and_then(|row| row.bit_depth));
QueueTrack {
id: cached.track_id,
title: cached.title,
@@ -1192,7 +1246,9 @@ fn artist_top_queue_tracks(
// full variant (best()) — the thumbnail down-tier was reverted
// after the 2026-08-15 owner smoke; register the full set so the
// immersive large-art feed can re-resolve (D2).
if let (Some(id), Some(img)) = (album_id.as_deref(), album.and_then(|a| a.image.as_ref())) {
if let (Some(id), Some(img)) =
(album_id.as_deref(), album.and_then(|a| a.image.as_ref()))
{
crate::artwork_qt::note_np_variants(id, img);
}
QueueTrack {
@@ -1241,7 +1297,10 @@ fn artist_top_queue_tracks(
/// buckets album/epSingle/ep/single in page order, deduped), concatenating
/// each album's tracks and skipping albums that fail (a bulk play must not
/// abort on one unavailable album).
pub async fn play_artist(runtime: &Arc<AppRuntime<LoggingAdapter>>, artist_id: &str) -> Result<(), String> {
pub async fn play_artist(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
artist_id: &str,
) -> Result<(), String> {
const STUDIO_TYPES: &[&str] = &["album", "epSingle", "ep", "single"];
let id: u64 = artist_id
.parse()
@@ -1254,8 +1313,11 @@ pub async fn play_artist(runtime: &Arc<AppRuntime<LoggingAdapter>>, artist_id: &
let artist_name = page.name.display.clone();
// 1) Popular tracks — the primary behavior.
let top: Vec<QueueTrack> =
artist_top_queue_tracks(page.top_tracks.as_deref().unwrap_or(&[]), artist_id, &artist_name);
let top: Vec<QueueTrack> = artist_top_queue_tracks(
page.top_tracks.as_deref().unwrap_or(&[]),
artist_id,
&artist_name,
);
if !top.is_empty() {
// F1: the anchor comes back from the seam, never from `top` — the
// filter may have dropped `top[0]`.
@@ -1297,7 +1359,9 @@ pub async fn play_artist(runtime: &Arc<AppRuntime<LoggingAdapter>>, artist_id: &
}
}
if album_ids.is_empty() {
return Err(format!("artist {artist_id} has no top tracks and no studio releases"));
return Err(format!(
"artist {artist_id} has no top tracks and no studio releases"
));
}
let mut queue: Vec<QueueTrack> = Vec::new();
for aid in &album_ids {
@@ -1307,9 +1371,14 @@ pub async fn play_artist(runtime: &Arc<AppRuntime<LoggingAdapter>>, artist_id: &
}
}
if queue.is_empty() {
return Err(format!("artist {artist_id} studio discography produced no playable tracks"));
return Err(format!(
"artist {artist_id} studio discography produced no playable tracks"
));
}
log::info!("[qbz-qt] artist-play {artist_id}: discography fallback, {} tracks", queue.len());
log::info!(
"[qbz-qt] artist-play {artist_id}: discography fallback, {} tracks",
queue.len()
);
// The discography queue arrives pre-stamped per ALBUM (fetch_album_queue);
// the artist origin is what the user launched, so it wins here — the
// explicit context overrides the per-album stamp for the whole queue.
@@ -1378,7 +1447,10 @@ pub async fn enqueue_album(
log::info!("[qbz-qt] enqueue_album {album_id}: every track was filtered");
return Ok(());
}
log::info!("[qbz-qt] enqueue_album {album_id} ({mode}): {} tracks", tracks.len());
log::info!(
"[qbz-qt] enqueue_album {album_id} ({mode}): {} tracks",
tracks.len()
);
// QConnect CONTROLLER mode (contract §7): route the add to the peer's
// queue — early-returns when handled, so the local insert + sync tail
// below only run in local/renderer mode. The mode is normalized to what
@@ -1657,9 +1729,7 @@ async fn try_infinite_refill(
Ok(track) => match track.performer.as_ref().map(|p| p.id) {
Some(id) => id,
None => {
log::warn!(
"[qbz-qt] infinite radio: seed track {seed_track_id} has no performer"
);
log::warn!("[qbz-qt] infinite radio: seed track {seed_track_id} has no performer");
return false;
}
},
@@ -1670,8 +1740,10 @@ async fn try_infinite_refill(
};
match runtime.core().create_smart_artist_radio(artist_id).await {
Ok(tracks) if !tracks.is_empty() => {
let queue: Vec<QueueTrack> =
tracks.iter().map(crate::foryou_qt::to_queue_track).collect();
let queue: Vec<QueueTrack> = tracks
.iter()
.map(crate::foryou_qt::to_queue_track)
.collect();
match play_track_list(runtime, queue, 0, false).await {
Ok(()) => {
log::info!(
@@ -1906,8 +1978,15 @@ fn feed_queue_track(track_id: u64) -> Result<QueueTrack, String> {
}
let duration_secs = {
let mut parts = item.duration.split(':');
parts.next().and_then(|m| m.parse::<u64>().ok()).unwrap_or(0) * 60
+ parts.next().and_then(|s| s.parse::<u64>().ok()).unwrap_or(0)
parts
.next()
.and_then(|m| m.parse::<u64>().ok())
.unwrap_or(0)
* 60
+ parts
.next()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0)
};
Ok(QueueTrack {
id: track_id,
@@ -2198,12 +2277,8 @@ pub async fn toggle_play(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
// release the latch. The poll loop's track-change edge clears
// it too — whichever runs first wins, and both are idempotent.
// Leaving it set on failure is precisely the Tauri scar.
PENDING_PLAY_ID.compare_exchange(
track.id,
0,
Ordering::Relaxed,
Ordering::Relaxed,
)
PENDING_PLAY_ID
.compare_exchange(track.id, 0, Ordering::Relaxed, Ordering::Relaxed)
.ok();
}
None => {
@@ -2246,6 +2321,17 @@ pub async fn next(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
}
}
}
// Prove a local file still exists before advancing the cursor. The player
// keeps the old stream alive until a replacement starts, so advancing
// first would make queue, NPB and MPRIS lie about what is audible.
if let Some(candidate) = runtime.core().peek_upcoming(1).await.into_iter().next() {
if crate::local_playback::preflight_queue_track(&candidate)
.await
.is_err()
{
return;
}
}
if let Some(track) = runtime.core().next_track().await {
play_queue_track(runtime, track.id, 0).await;
}
@@ -2263,12 +2349,26 @@ pub async fn previous(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
}
}
}
// `previous_track` consumes history and publishes QueueUpdated. Validate
// a physical local target first so a missing/moved file cannot leave the
// cursor, NPB and MPRIS one row behind the audio the user still hears.
if let Some(candidate) = runtime.core().peek_previous_track().await {
if crate::local_playback::preflight_queue_track(&candidate)
.await
.is_err()
{
return;
}
}
if let Some(track) = runtime.core().previous_track().await {
play_queue_track(runtime, track.id, 0).await;
}
}
pub(crate) async fn play_queue_track_public(runtime: &Arc<AppRuntime<LoggingAdapter>>, track_id: u64) {
pub(crate) async fn play_queue_track_public(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
track_id: u64,
) {
play_queue_track(runtime, track_id, 0).await;
}
@@ -2324,10 +2424,7 @@ async fn play_queue_track(
}
crate::local_playback::LocalPlay::NotLocal => {}
}
if let Err(e) =
play_resolved_offline_aware(runtime, track_id, start_position_secs)
.await
{
if let Err(e) = play_resolved_offline_aware(runtime, track_id, start_position_secs).await {
log::error!("[qbz-qt] playback: play_track {track_id} failed: {e}");
// The bar must not keep showing the PREVIOUS track while the cursor has
// already moved. The old early `return` skipped both refreshes, which is
@@ -2439,7 +2536,10 @@ fn auto_skip_unavailable<'a>(
}
pub async fn seek_frac(runtime: &Arc<AppRuntime<LoggingAdapter>>, frac: f32) {
match crate::cast_qt::service().seek_fraction_if_cast(frac as f64).await {
match crate::cast_qt::service()
.seek_fraction_if_cast(frac as f64)
.await
{
Ok(true) => return,
Ok(false) => {}
Err(e) => {
@@ -2454,8 +2554,7 @@ pub async fn seek_frac(runtime: &Arc<AppRuntime<LoggingAdapter>>, frac: f32) {
if let Some(svc) = crate::qconnect_qt::service() {
let fraction = frac.clamp(0.0, 1.0);
let duration_secs = runtime.core().get_playback_state().duration;
let position_ms =
(fraction as f64 * duration_secs as f64 * 1000.0).round() as i64;
let position_ms = (fraction as f64 * duration_secs as f64 * 1000.0).round() as i64;
match svc.set_position_if_remote(position_ms).await {
Ok(true) => return,
Ok(false) => {}
@@ -2523,7 +2622,10 @@ pub async fn seek_frac(runtime: &Arc<AppRuntime<LoggingAdapter>>, frac: f32) {
}
pub async fn set_volume(runtime: &Arc<AppRuntime<LoggingAdapter>>, volume: f32) {
if matches!(crate::cast_qt::service().set_volume_if_cast(volume).await, Ok(true)) {
if matches!(
crate::cast_qt::service().set_volume_if_cast(volume).await,
Ok(true)
) {
return;
}
// QConnect CONTROLLER mode (Slint main.rs:14378-14389): set the REMOTE
@@ -2543,7 +2645,10 @@ pub async fn set_volume(runtime: &Arc<AppRuntime<LoggingAdapter>>, volume: f32)
}
}
let _ = runtime.core().set_volume(volume.clamp(0.0, 1.0));
MUTED.store(volume <= 0.0 && PREMUTE_VOLUME.load(Ordering::Relaxed) != 0, Ordering::Relaxed);
MUTED.store(
volume <= 0.0 && PREMUTE_VOLUME.load(Ordering::Relaxed) != 0,
Ordering::Relaxed,
);
}
pub async fn toggle_mute(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
@@ -2684,7 +2789,8 @@ pub(crate) async fn refresh_now_playing(runtime: &Arc<AppRuntime<LoggingAdapter>
String::new()
};
crate::player_bridge::ui(move |mut b| {
b.as_mut().set_np_track_id(QString::from(track_id_str.as_str()));
b.as_mut()
.set_np_track_id(QString::from(track_id_str.as_str()));
b.as_mut()
.set_np_local_track_id(QString::from(local_track_id_str.as_str()));
});
@@ -2750,14 +2856,11 @@ pub(crate) async fn refresh_now_playing(runtime: &Arc<AppRuntime<LoggingAdapter>
_ => ("album".to_string(), album_id.clone()),
};
crate::now_playing::set_track(crate::now_playing::TrackMeta {
title,
title: title.clone(),
artist: track.artist.clone(),
album: album_display,
album: album_display.clone(),
album_id,
artist_id: track
.artist_id
.map(|id| id.to_string())
.unwrap_or_default(),
artist_id: track.artist_id.map(|id| id.to_string()).unwrap_or_default(),
track_id: track.id,
source: track.source.clone().unwrap_or_default(),
context_kind,
@@ -2788,7 +2891,7 @@ pub(crate) async fn refresh_now_playing(runtime: &Arc<AppRuntime<LoggingAdapter>
crate::now_playing::set_catalog_quality(track.bit_depth, track.sample_rate, governed);
// Artwork through the same cache pipeline as Home (attach + background
// download + republish — single url here).
crate::artwork_qt::attach_now_playing(&track.artwork_url.clone().unwrap_or_default());
crate::artwork_qt::attach_now_playing(&track, &title, &album_display);
// Ambient background triad (phase 14): recompute from the new track's
// cover (no-op until the cover lands on disk — the previous palette
// stays, like the Slint's default-until-resolved).
@@ -2853,8 +2956,7 @@ fn quality_badge_from(
let sample_rate = track
.sample_rate
.or_else(|| source_hint.and_then(|hint| hint.sample_rate_khz));
let is_mp3 = source_hint
.is_some_and(|hint| hint.format.eq_ignore_ascii_case("mp3"));
let is_mp3 = source_hint.is_some_and(|hint| hint.format.eq_ignore_ascii_case("mp3"));
if is_mp3 {
let label = sample_rate
.filter(|rate| *rate > 0.0)
@@ -3043,8 +3145,8 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
// Extrapolate position while playing; clamp to the track length.
let mut position_ms = remote.position_ms;
if remote.playing && remote.updated_at_ms > 0 {
position_ms = position_ms
.saturating_add(now_ms().saturating_sub(remote.updated_at_ms));
position_ms =
position_ms.saturating_add(now_ms().saturating_sub(remote.updated_at_ms));
}
if duration_ms > 0 && position_ms > duration_ms {
position_ms = duration_ms;
@@ -3176,7 +3278,10 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
// has downloaded; a fully-available track (None) seeks freely.
// `cache` above is the DECORATIVE overlay of the same fill — this
// is the value the seek bars must enforce.
let seekable_max = event.buffer_progress.map(|p| p.clamp(0.0, 1.0)).unwrap_or(1.0);
let seekable_max = event
.buffer_progress
.map(|p| p.clamp(0.0, 1.0))
.unwrap_or(1.0);
// This observation is cheap and happens each tick; the work does
// not. Only a NEW non-zero engine track — including a gapless
@@ -3305,8 +3410,7 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
let quality = local_playback_quality().0;
let streaming_only =
crate::settings_qt::audio_settings().streaming_only;
let player_cached =
runtime.core().player().is_track_cached(next_id);
let player_cached = runtime.core().player().is_track_cached(next_id);
let offline_cached = crate::offline_qt::is_cached_id(next_id);
if streaming_only && !player_cached && !offline_cached {
@@ -3403,9 +3507,7 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
tokio::spawn(async move {
let next_id = next.id;
match crate::audible_qt::queue_gapless_successor(
&runtime,
track_id,
&next,
&runtime, track_id, &next,
)
.await
{
@@ -3450,7 +3552,12 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
};
let position_ms = (position as i64) * 1000;
let duration_ms = (duration as i64) * 1000;
svc.report_playback_state(playing_state, position_ms, duration_ms, track_id)
svc.report_playback_state(
playing_state,
position_ms,
duration_ms,
track_id,
)
.await;
// On a track change, also reconcile the session queue:
// if the user started a new album/playlist on QBZ,
@@ -3489,8 +3596,7 @@ pub fn start_poll_loop(runtime: Arc<AppRuntime<LoggingAdapter>>) {
// marker), and this runs AHEAD of repeat/shuffle exactly as
// the reference does (playback.rs:5749-5772) — behind them the
// marker would lose to a repeat-one and never fire.
if ended_track_id != 0
&& runtime.core().consume_stop_after_if(ended_track_id).await
if ended_track_id != 0 && runtime.core().consume_stop_after_if(ended_track_id).await
{
if let Err(e) = runtime.core().pause() {
log::warn!("[qbz-qt] stop-after: pause failed: {e}");
@@ -3845,7 +3951,9 @@ mod tests {
// Skip the definition, doc comments and ordinary comments —
// this audits CALLS, not prose about them.
let t = line.trim_start();
if t.starts_with("//") || t.starts_with("fn ") || t.starts_with("pub(crate) fn ")
if t.starts_with("//")
|| t.starts_with("fn ")
|| t.starts_with("pub(crate) fn ")
{
continue;
}
@@ -4005,8 +4113,7 @@ mod tests {
/// Everything blocked: an empty queue carries NO start index.
#[test]
fn everything_blocked_yields_no_start() {
let (kept, start, _) =
filter_queue_with(vec![1, 3, 5], Some(1), |n| n % 2 == 1, |_| false);
let (kept, start, _) = filter_queue_with(vec![1, 3, 5], Some(1), |n| n % 2 == 1, |_| false);
assert!(kept.is_empty());
assert_eq!(start, None);
}
@@ -4061,4 +4168,16 @@ mod tests {
assert_eq!(dropped.blacklisted, 1);
assert_eq!(dropped.unavailable, 0);
}
#[test]
fn song_card_near_fit_keeps_the_final_glyphs_instead_of_an_ellipsis() {
// TextMetrics.width is a painted bounding box, while the line layout
// consumes advanceWidth. Budgeting the former exactly turned
// "Metallica" into "Metalli…" in a runway with visible spare space.
let qml = include_str!("../qml/shell/SongCard.qml");
assert!(qml.contains("artistMetrics.advanceWidth"));
assert!(qml.contains("albumMetrics.advanceWidth"));
assert!(qml.contains("a - qa <= near"));
assert!(qml.contains("b - qb <= near"));
}
}
crates/qbz-qt/src/queue_qt.rs
+42 -17
View File
@@ -21,9 +21,9 @@
use std::collections::HashSet;
use std::sync::Mutex;
use cxx_qt_lib::QString;
use qbz_app::shell::AppRuntime;
use qbz_core::LoggingAdapter;
use cxx_qt_lib::QString;
use qbz_models::QueueTrack;
use serde::Serialize;
use std::sync::Arc;
@@ -418,8 +418,7 @@ static LAST_MINI_QUEUE_JSON: Mutex<String> = Mutex::new(String::new());
/// restored queue may contain thousands of tracks; the sidebar document stays
/// paged at 40 and should not pay to serialize the full sequence in the normal
/// shell path.
static EXTENDED_OPEN: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
static EXTENDED_OPEN: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
static LAST_EXTENDED_JSON: Mutex<String> = Mutex::new(String::new());
fn extended_rows(
@@ -669,7 +668,10 @@ pub async fn publish(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
.iter()
.map(|t| row_from(t, &favorites))
.collect();
let current = state.current_track.as_ref().map(|t| row_from(t, &favorites));
let current = state
.current_track
.as_ref()
.map(|t| row_from(t, &favorites));
// One core read for the marker (the reference reads it in the same pass,
// `queue.rs:396`), so a row can compare its id without a second round trip.
@@ -768,6 +770,16 @@ pub async fn play_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_index
log::warn!("[qbz-qt] queue: play_upcoming {page_index} out of range");
return;
};
let state = runtime.core().get_queue_state_full().await;
let Some(candidate) = state.upcoming.get(upcoming_index) else {
return;
};
if crate::local_playback::preflight_queue_track(candidate)
.await
.is_err()
{
return;
}
if let Some(track) = runtime.core().play_upcoming_at(upcoming_index).await {
crate::playback_qt::play_queue_track_public(runtime, track.id).await;
}
@@ -779,6 +791,15 @@ pub async fn play_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_index
/// so it resolves the right track even when the desktop panel was left
/// filtered or paged (state.slint:4896-4900 `play-coverflow-upcoming`).
pub async fn play_upcoming_flat(runtime: &Arc<AppRuntime<LoggingAdapter>>, upcoming_index: usize) {
let state = runtime.core().get_queue_state_full().await;
if let Some(candidate) = state.upcoming.get(upcoming_index) {
if crate::local_playback::preflight_queue_track(candidate)
.await
.is_err()
{
return;
}
}
match runtime.core().play_upcoming_at(upcoming_index).await {
Some(track) => crate::playback_qt::play_queue_track_public(runtime, track.id).await,
None => log::warn!("[qbz-qt] queue: play_upcoming_flat {upcoming_index} out of range"),
@@ -846,14 +867,21 @@ pub async fn play_extended(
if crate::qconnect_qt::publish::is_connected()
&& !crate::qconnect_qt::is_qconnect_queue_track(&track)
{
log::info!("[qbz-qt] queue: skipped QConnect-incompatible history row {}", track.id);
log::info!(
"[qbz-qt] queue: skipped QConnect-incompatible history row {}",
track.id
);
return;
}
match insert_queue_track_at(runtime, track.clone(), 0).await {
Ok(InsertOutcome::Remote) => {
if let Some(service) = crate::qconnect_qt::service() {
if let Err(error) = service.play_remote_renderer_track_if_active(track.id).await {
log::warn!("[qbz-qt] queue: requeued remote history play failed: {error}");
if let Err(error) =
service.play_remote_renderer_track_if_active(track.id).await
{
log::warn!(
"[qbz-qt] queue: requeued remote history play failed: {error}"
);
}
}
}
@@ -1099,7 +1127,10 @@ pub async fn drop_extended(
if crate::qconnect_qt::publish::is_connected()
&& !crate::qconnect_qt::is_qconnect_queue_track(&track)
{
log::info!("[qbz-qt] queue: skipped QConnect-incompatible history drag {}", track.id);
log::info!(
"[qbz-qt] queue: skipped QConnect-incompatible history drag {}",
track.id
);
return;
}
if let Err(error) = insert_queue_track_at(runtime, track, to_slot).await {
@@ -1166,8 +1197,7 @@ pub async fn insert_dragged_track(
vec![crate::playback_qt::queue_track_for(runtime, track_id).await?],
None,
)
.pop()
else {
.pop() else {
log::info!("[qbz-qt] queue: dragged track {track_id} was filtered out");
return Ok(());
};
@@ -1365,11 +1395,7 @@ pub async fn add_to_playlist(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_ind
/// documents the same gap), so the mutation is refused here AND the row
/// publishes `isLocal` so the panel can drop the control instead of showing
/// one that cannot work.
pub async fn toggle_favorite(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
kind: &str,
id: &str,
) {
pub async fn toggle_favorite(runtime: &Arc<AppRuntime<LoggingAdapter>>, kind: &str, id: &str) {
if kind == "track" {
let local = canonical_u64(id)
.map(|tid| LOCAL_ROW_IDS.lock().unwrap().contains(&tid))
@@ -1543,8 +1569,7 @@ mod tests {
stop_after_track_id: None,
manual_next_count: 0,
};
let (rows, current_index) =
extended_rows(&state, &HashSet::new(), "needle");
let (rows, current_index) = extended_rows(&state, &HashSet::new(), "needle");
assert_eq!(current_index, -1);
assert_eq!(rows.len(), 2);
crates/qbz-source/src/playback.rs
+26
View File
@@ -73,6 +73,14 @@ pub enum PlaybackTicket {
/// Log prefix the feeder stamps its lines with (`"PLEX"`). Static
/// because it names the SOURCE, never the item.
log_tag: &'static str,
/// Source-owned HTTP request headers required by the media endpoint.
///
/// These are transport facts, not audio policy. Plex, for example,
/// accepts the token in the URL but can reject a part request without
/// its `X-Plex-*` client identity. Carrying the headers on the ticket
/// keeps the shared player transport source-agnostic while preserving
/// the request contract the source used to fetch the same bytes.
request_headers: Vec<(String, String)>,
},
/// Let the core resolve + stream it: `core().play_track_resolved(track_id, …)`.
Catalog { track_id: u64 },
@@ -142,6 +150,7 @@ impl std::fmt::Debug for PlaybackTicket {
duration_secs,
start_secs,
log_tag,
request_headers,
} => f
.debug_struct("Stream")
.field("url", &redact_url(url))
@@ -149,6 +158,15 @@ impl std::fmt::Debug for PlaybackTicket {
.field("duration_secs", duration_secs)
.field("start_secs", start_secs)
.field("log_tag", log_tag)
// Header VALUES may become sensitive for a future source.
// Names are enough to diagnose a missing request profile.
.field(
"request_headers",
&request_headers
.iter()
.map(|(name, _)| name.as_str())
.collect::<Vec<_>>(),
)
.finish(),
PlaybackTicket::Catalog { track_id } => f
.debug_struct("Catalog")
@@ -215,9 +233,15 @@ mod tests {
duration_secs: 188,
start_secs: 0,
log_tag: "PLEX",
request_headers: vec![("X-Plex-Product".into(), "secret-value".into())],
};
let s = format!("{t:?}");
assert!(!s.contains("sekrit"), "the token leaked into Debug: {s}");
assert!(s.contains("X-Plex-Product"), "{s}");
assert!(
!s.contains("secret-value"),
"header value leaked into Debug: {s}"
);
assert!(s.contains("/library/parts/1/f.flac?<redacted>"), "{s}");
// The origin SURVIVES: telling a LAN address from a plex.tv relay is
// the question those perf log lines exist to answer.
@@ -232,6 +256,7 @@ mod tests {
duration_secs: 0,
start_secs: 0,
log_tag: "PLEX",
request_headers: Vec::new(),
};
assert!(format!("{t:?}").contains("http://nas.local/music/a.flac"));
}
@@ -244,6 +269,7 @@ mod tests {
duration_secs: 0,
start_secs: 0,
log_tag: "PLEX",
request_headers: Vec::new(),
};
assert_eq!(t.play_id(), Some(42));
// `Catalog` is the one variant with no player-side id — the core mints