fix(qconnect): enforce shared lifecycle authority

cx
This commit is contained in:
vicrodh committed 2026-08-30 22:22:08 -06:00
1 parent 74a41150e9
commit b4404f45ba
22 files changed
+3233 -406

No files matched your search

+1 -1
View File
@@ -41,4 +41,4 @@ pub use player::{
InMemorySource, PlaybackBufferState, PlaybackEvent, PlaybackState,
Player, SharedState, StreamingConfig,
};
pub use queue::QueueManager;
pub use queue::{QueueAuthoritySnapshot, QueueManager};
+13 -11
View File
@@ -5343,9 +5343,9 @@ impl Player {
let stream_url = client
.get_stream_url_with_fallback(track_id, quality)
.await
.map_err(|e| {
log::error!("Player: Failed to get stream URL: {}", e);
format!("Failed to get stream URL: {}", e)
.map_err(|_| {
log::error!("Player: Failed to get stream URL (details omitted)");
"Failed to get stream URL".to_string()
})?;
if !self.is_current_play(gen) {
@@ -5355,11 +5355,7 @@ impl Player {
return Ok(());
}
log::info!(
"Player: Got stream URL: {} (format: {})",
stream_url.url,
stream_url.mime_type
);
log::info!("Player: Got stream URL (format: {})", stream_url.mime_type);
// Download the audio data
log::info!("Player: Starting audio caching...");
@@ -6695,7 +6691,7 @@ impl Player {
.timeout(Duration::from_secs(60))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|e| format!("Failed to create HTTP client: {}", e))?;
.map_err(|_| "Failed to create HTTP client".to_string())?;
log::info!("Caching audio from URL...");
@@ -6704,7 +6700,10 @@ impl Player {
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|e| format!("Failed to fetch audio: {}", e))?;
.map_err(|e| format!(
"Failed to fetch audio: {}",
crate::remote_stream::describe_reqwest_error(&e)
))?;
if !response.status().is_success() {
return Err(format!("HTTP error: {}", response.status()));
@@ -6715,7 +6714,10 @@ impl Player {
let bytes = response
.bytes()
.await
.map_err(|e| format!("Failed to read audio bytes: {}", e))?;
.map_err(|e| format!(
"Failed to read audio bytes: {}",
crate::remote_stream::describe_reqwest_error(&e)
))?;
log::info!("Cached {} bytes", bytes.len());
Ok(bytes.to_vec())
+121 -8
View File
@@ -21,6 +21,7 @@ enum QueueMoveDirection {
const MAX_HISTORY_LEN: usize = 50;
/// Internal queue state - all in one struct to avoid deadlocks
#[derive(Clone)]
struct InternalState {
/// All tracks in the queue (original order)
tracks: Vec<QueueTrack>,
@@ -45,6 +46,16 @@ struct InternalState {
manual_next_count: usize,
}
/// An exact, in-process snapshot of queue playback authority.
///
/// The payload is deliberately opaque: callers can preserve and later restore
/// the queue, but cannot manufacture a partial or internally inconsistent
/// state. It is not a persistence format and is intended for transactional
/// authority handoffs such as QConnect delegation.
#[must_use = "an authority snapshot must be retained until it is restored or deliberately discarded"]
#[derive(Clone)]
pub struct QueueAuthoritySnapshot(InternalState);
/// Queue manager for handling playback queue
pub struct QueueManager {
state: Mutex<InternalState>,
@@ -370,7 +381,10 @@ impl QueueManager {
// History stores indices into `tracks`. Remap by track id
// so entries for removed rows drop and any entry that still
// refers to the kept track points at index 0.
Self::remap_history_by_track_id_internal(&mut state, std::slice::from_ref(&kept));
Self::remap_history_by_track_id_internal(
&mut state,
std::slice::from_ref(&kept),
);
state.tracks = vec![kept];
state.current_index = Some(0);
} else {
@@ -1058,9 +1072,7 @@ impl QueueManager {
// shrinks the block; exhausting or wrapping the queue dissolves it.
match next_idx {
Some(0) | None => state.manual_next_count = 0,
Some(_) => {
state.manual_next_count = state.manual_next_count.saturating_sub(1)
}
Some(_) => state.manual_next_count = state.manual_next_count.saturating_sub(1),
}
next_idx.and_then(|idx| state.tracks.get(idx).cloned())
}
@@ -1487,6 +1499,27 @@ impl QueueManager {
(state.tracks.clone(), state.current_index)
}
/// Capture the complete queue authority state under one lock.
///
/// Unlike the frontend and persistence projections, this preserves every
/// playback-order detail without caps or normalization: canonical tracks,
/// current cursor, shuffle order and cursor, repeat mode, chronological
/// history, stop-after marker, and the manual-next block size.
pub fn capture_authority_snapshot(&self) -> QueueAuthoritySnapshot {
let state = self.state.lock().unwrap();
QueueAuthoritySnapshot(state.clone())
}
/// Atomically replace the complete queue authority state with `snapshot`.
///
/// Consuming the opaque snapshot prevents callers from restoring only a
/// subset of the coupled queue fields or mutating them between capture and
/// restore.
pub fn restore_authority_snapshot(&self, snapshot: QueueAuthoritySnapshot) {
let mut state = self.state.lock().unwrap();
*state = snapshot.0;
}
/// Capture the queue rows, cursor, and chronological play-history indices
/// under one lock for durable session persistence.
///
@@ -1777,6 +1810,89 @@ mod tests {
}
}
#[test]
fn authority_snapshot_round_trips_complete_shuffled_state() {
let queue = QueueManager::new();
let mut tracks: Vec<_> = (101..=105).map(create_test_track).collect();
tracks[2].version = Some("Authority snapshot edition".to_string());
tracks[2].bit_depth = Some(24);
tracks[2].sample_rate = Some(192.0);
tracks[2].isrc = Some("TEST00000103".to_string());
let shuffle_order = vec![4, 2, 0, 3, 1];
queue.set_queue_with_order(tracks, Some(2), true, Some(shuffle_order.clone()));
queue.play_index(3);
queue.set_repeat(RepeatMode::All);
queue.set_stop_after(105);
let expected_tracks = {
let state = queue.state.lock().unwrap();
format!("{:?}", state.tracks)
};
let snapshot = queue.capture_authority_snapshot();
queue.clear(false);
queue.add_track(create_test_track(999));
queue.play_index(0);
queue.set_repeat(RepeatMode::One);
queue.restore_authority_snapshot(snapshot);
let state = queue.state.lock().unwrap();
assert_eq!(format!("{:?}", state.tracks), expected_tracks);
assert_eq!(state.current_index, Some(3));
assert!(state.shuffle);
assert_eq!(state.shuffle_order, shuffle_order);
assert_eq!(state.shuffle_position, 3);
assert_eq!(state.repeat, RepeatMode::All);
assert_eq!(state.history, VecDeque::from([2]));
assert_eq!(state.stop_after_track_id, Some(105));
assert_eq!(state.manual_next_count, 0);
}
#[test]
fn authority_snapshot_round_trips_manual_next_block() {
let queue = QueueManager::new();
queue.set_queue(
vec![
create_test_track(1),
create_test_track(2),
create_test_track(3),
],
Some(0),
);
queue.add_track_next(create_test_track(10));
queue.add_track_later(create_test_track(11));
queue.set_repeat(RepeatMode::One);
queue.set_stop_after(3);
let (expected_shuffle_order, expected_shuffle_position) = {
let state = queue.state.lock().unwrap();
(state.shuffle_order.clone(), state.shuffle_position)
};
let snapshot = queue.capture_authority_snapshot();
queue.clear(false);
queue.restore_authority_snapshot(snapshot);
let state = queue.state.lock().unwrap();
assert_eq!(
state
.tracks
.iter()
.map(|track| track.id)
.collect::<Vec<_>>(),
vec![1, 10, 11, 2, 3]
);
assert_eq!(state.current_index, Some(0));
assert!(!state.shuffle);
assert_eq!(state.shuffle_order, expected_shuffle_order);
assert_eq!(state.shuffle_position, expected_shuffle_position);
assert_eq!(state.repeat, RepeatMode::One);
assert!(state.history.is_empty());
assert_eq!(state.stop_after_track_id, Some(3));
assert_eq!(state.manual_next_count, 2);
}
#[test]
fn repeat_all_wraps_a_local_album_for_renderer_advance() {
let queue = QueueManager::new();
@@ -3140,10 +3256,7 @@ mod tests {
fn test_remove_upcoming_after_linear() {
let queue = QueueManager::new();
// 101 playing, upcoming = [102, 103, 104, 105].
queue.set_queue(
(101..=105).map(create_test_track).collect(),
Some(0),
);
queue.set_queue((101..=105).map(create_test_track).collect(), Some(0));
// Keep upcoming positions 0..=1 (102, 103); drop 2, 3 (104, 105).
let removed = queue.remove_upcoming_after(1);
+54 -9
View File
@@ -459,20 +459,53 @@ fn parsed_request_headers(
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
/// chain so logs AND signature matching see the real cause.
/// Return a bounded, URL-free diagnostic for a reqwest failure.
///
/// The raw error and its source chain can contain a signed CDN URL. Inspect the
/// chain only to preserve header-limit classification; never copy an arbitrary
/// cause into logs or a returned error.
pub fn describe_reqwest_error(err: &reqwest::Error) -> String {
if error_chain_has_header_limit(err) {
return safe_transport_diagnostic("message head is too large").to_string();
}
if err.is_timeout() {
"HTTP transport timed out".to_string()
} else if err.is_connect() {
"HTTP transport connection failed".to_string()
} else if err.is_body() {
"HTTP response body failed".to_string()
} else if err.is_decode() {
"HTTP response decode failed".to_string()
} else if err.is_status() {
"HTTP status rejected".to_string()
} else {
"HTTP transport request failed".to_string()
}
}
fn error_chain_has_header_limit(err: &reqwest::Error) -> bool {
use std::error::Error as _;
let mut out = err.to_string();
if is_header_flood_error(&err.to_string()) {
return true;
}
let mut source = err.source();
while let Some(cause) = source {
out.push_str(": ");
out.push_str(&cause.to_string());
if is_header_flood_error(&cause.to_string()) {
return true;
}
source = cause.source();
}
out
false
}
fn safe_transport_diagnostic(message: &str) -> &'static str {
if is_header_flood_error(message) {
"HTTP response header limit exceeded (message head is too large)"
} else {
"HTTP transport request failed"
}
}
/// True when an error message (already chain-expanded by
@@ -487,7 +520,19 @@ pub fn is_header_flood_error(message: &str) -> bool {
#[cfg(test)]
mod tests {
use super::{content_range_total, parsed_request_headers, probe_content_length};
use super::{
content_range_total, parsed_request_headers, probe_content_length,
safe_transport_diagnostic,
};
#[test]
fn transport_diagnostic_never_echoes_signed_url() {
let marker = "https://cdn.example/audio.flac?jwt=secret&request_sig=signed";
let diagnostic = safe_transport_diagnostic(marker);
assert_eq!(diagnostic, "HTTP transport request failed");
assert!(!diagnostic.contains(marker));
assert!(!diagnostic.contains("secret"));
}
#[test]
fn source_request_headers_are_parsed_for_the_http_request() {