feat(qt): harden qconnect ownership lifecycle

cx
This commit is contained in:
vicrodh committed 2026-08-30 22:22:48 -06:00
1 parent c927eb9c9c
commit 9d84f0c384
23 files changed
+6439 -728

No files matched your search

+18
View File
@@ -446,6 +446,13 @@ fn begin_test(s: &mut State) {
s.doc.test_negotiated_label = String::new();
}
fn reject_test_control(action: &str) {
log::info!("[qbz-qt] dac wizard: {action} refused while QConnect owns playback");
crate::toast_qt::info(qbz_i18n::t(
"QConnect is controlling playback. Return playback to QBZ before using the DAC test.",
));
}
/// "Play test": resolve the four curated tracks (id-hint first, then an
/// "artist title" search when the id 404s on a pulled licence — never
/// raw-id-only), stash them so the user can jump between them, and play.
@@ -522,6 +529,10 @@ async fn play_from(tracks: Vec<Track>, start: usize) {
/// the queue, so the user keeps whatever was loaded).
pub fn stop_test() {
crate::spawn(async move {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
reject_test_control("stop test");
return;
};
let _ = crate::app().core().pause();
with_state(|s| s.doc.test_playing = false);
publish();
@@ -533,6 +544,13 @@ pub fn stop_test() {
/// read-back that would sit forever on "Nothing playing".
pub fn verify_own() {
crate::spawn(async move {
// Keep the permit over the queue read and the eventual resume: a
// handoff must either drain this whole action first or make it a clean
// refusal before any delegated queue/player state is observed.
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
reject_test_control("use current queue");
return;
};
let runtime = crate::app();
let (tracks, _) = runtime.core().get_all_queue_tracks().await;
if tracks.is_empty() {
+260 -58
View File
@@ -47,6 +47,9 @@ use qbz_integrations::listenbrainz::{AdditionalInfo, ListenBrainzClient};
use qbz_integrations::NowListening;
use qbz_models::QueueTrack;
use crate::playback_qt::{
begin_owner_action, begin_owner_action_exact, OwnerActionToken,
};
use crate::settings_qt::{pref_bool, save_pref};
// ---------------------------------------------------------------------------
@@ -403,43 +406,139 @@ pub fn discord_push(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
if !discord().is_enabled() {
return;
}
let Some(owner_action) = begin_owner_action() else {
return;
};
let owner_token = owner_action.token();
drop(owner_action);
let runtime = runtime.clone();
crate::spawn(async move {
let state = runtime.core().get_queue_state().await;
let Some(track) = state.current_track else {
// Nothing playing — drop the activity.
let _ = tokio::task::spawn_blocking(|| discord().clear()).await;
// The task has not read its queue input yet, so re-admit the exact
// producer observation at the consumer. This also invalidates the
// special pre-service token if QConnect appeared before scheduling.
let Some(_owner_action) = begin_owner_action_exact(owner_token).await else {
return;
};
let pb = runtime.core().get_playback_state();
let title = match track.version.as_deref().filter(|v| !v.is_empty()) {
Some(version) => format!("{} ({version})", track.title),
None => track.title.clone(),
};
// Discord's large_image needs an http(s) URL or an asset key; local /
// Plex covers are filesystem paths Discord can't fetch, so drop them
// (the core falls back to the "cover" asset key). A sized Qobuz cover
// is rewritten to the 600 tier first — the queue can carry the 50px
// `small` from a restored session, and Discord renders whatever the
// suffix says (owner smoke 2026-08-15: pixelated presence art).
let cover_url = track
.artwork_url
.clone()
.filter(|u| u.starts_with("http://") || u.starts_with("https://"))
.map(|u| qbz_models::qobuz_cover_at_px(&u, 600).unwrap_or(u));
let meta = NowListening {
title,
artist: track.artist.clone(),
album: track.album.clone(),
is_playing: pb.is_playing,
current_time: pb.position as f64,
duration: track.duration_secs as f64,
cover_url,
};
let _ = tokio::task::spawn_blocking(move || discord().update(&meta)).await;
push_discord_current(&runtime, None, None).await;
});
}
/// Push a Discord continuation derived from an already-read playback event.
/// The exact token rejects owner -> guest -> owner cycles; track + generation
/// reject a later local edge within the same owner authority.
pub fn discord_push_observed(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
owner_token: OwnerActionToken,
expected_track_id: u64,
) {
let expected_generation = SCROBBLE_GEN.load(Ordering::SeqCst);
spawn_discord_observed(
runtime,
owner_token,
expected_track_id,
expected_generation,
);
}
/// Discord-only peer edge. It shares the playback integration generation so
/// an A -> B -> A sequence cannot revive a queued push for the first A.
pub fn discord_track_change_edge(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
owner_token: OwnerActionToken,
expected_track_id: u64,
) {
let expected_generation = SCROBBLE_GEN.fetch_add(1, Ordering::SeqCst) + 1;
spawn_discord_observed(
runtime,
owner_token,
expected_track_id,
expected_generation,
);
}
fn spawn_discord_observed(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
owner_token: OwnerActionToken,
expected_track_id: u64,
expected_generation: u64,
) {
if !discord().is_enabled() || expected_track_id == 0 {
return;
}
let runtime = runtime.clone();
crate::spawn(async move {
let Some(_owner_action) = begin_owner_action_exact(owner_token).await else {
return;
};
push_discord_current(
&runtime,
Some(expected_track_id),
Some(expected_generation),
)
.await;
});
}
async fn push_discord_current(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
expected_track_id: Option<u64>,
expected_generation: Option<u64>,
) {
if expected_generation
.is_some_and(|expected| expected != SCROBBLE_GEN.load(Ordering::SeqCst))
{
return;
}
let state = runtime.core().get_queue_state().await;
// The queue read yields. A newer track edge may have advanced the shared
// generation while this task was suspended; in particular, an old empty
// snapshot must not clear the newer track's Discord activity.
if expected_generation
.is_some_and(|expected| expected != SCROBBLE_GEN.load(Ordering::SeqCst))
{
return;
}
let Some(track) = state.current_track else {
// Nothing playing — drop the activity.
let _ = tokio::task::spawn_blocking(|| discord().clear()).await;
return;
};
if !integration_snapshot_matches(
expected_track_id,
expected_generation,
track.id,
SCROBBLE_GEN.load(Ordering::SeqCst),
) {
return;
}
let pb = runtime.core().get_playback_state();
let title = match track.version.as_deref().filter(|v| !v.is_empty()) {
Some(version) => format!("{} ({version})", track.title),
None => track.title.clone(),
};
// Discord's large_image needs an http(s) URL or an asset key; local /
// Plex covers are filesystem paths Discord can't fetch, so drop them
// (the core falls back to the "cover" asset key). A sized Qobuz cover
// is rewritten to the 600 tier first — the queue can carry the 50px
// `small` from a restored session, and Discord renders whatever the
// suffix says (owner smoke 2026-08-15: pixelated presence art).
let cover_url = track
.artwork_url
.clone()
.filter(|u| u.starts_with("http://") || u.starts_with("https://"))
.map(|u| qbz_models::qobuz_cover_at_px(&u, 600).unwrap_or(u));
let meta = NowListening {
title,
artist: track.artist.clone(),
album: track.album.clone(),
is_playing: pb.is_playing,
current_time: pb.position as f64,
duration: track.duration_secs as f64,
cover_url,
};
let _ = tokio::task::spawn_blocking(move || discord().update(&meta)).await;
}
/// Tear down the live activity + IPC connection (logout / app exit).
/// GLUE (optional): call from the logout path and the quit handler.
pub fn discord_clear() {
@@ -791,29 +890,76 @@ static SCROBBLE_GEN: AtomicU64 = AtomicU64::new(0);
/// How often the delayed-scrobble task samples the player while waiting.
const SCROBBLE_TICK: Duration = Duration::from_secs(1);
/// Track-change entry point. Fires now-playing immediately for each enabled +
/// authed service, then arms a delayed scrobble at `min(50% of duration,
/// 240s)`. No-op when no service is active (opt-in gate).
///
/// GLUE: call from the playback poll's de-duped track-change edge only — NOT
/// on resume/seek, or the scrobble re-arms.
pub fn on_track_changed(meta: ScrobbleMeta) {
// Always bump the generation so any in-flight stale timer self-cancels.
let my_gen = SCROBBLE_GEN.fetch_add(1, Ordering::SeqCst) + 1;
let cfg = scrobble_settings();
if !cfg.lastfm_active() && !cfg.listenbrainz_active() {
return;
}
/// Cancel every owner-playback integration at the authority commit boundary.
/// Exact tokens already reject work after the install; the generation wakes
/// delayed same-owner timers promptly, and clearing Discord removes the
/// owner's presence before delegated playback becomes observable.
pub fn cancel_owner_playback_tasks() {
let cancelled_generation = SCROBBLE_GEN.fetch_add(1, Ordering::SeqCst) + 1;
crate::spawn(async move {
if SCROBBLE_GEN.load(Ordering::SeqCst) != cancelled_generation {
return;
}
let _ = tokio::task::spawn_blocking(|| discord().clear()).await;
});
}
/// Shared track/generation validation for queued integration continuations.
/// `None` means the task obtained its own exact owner lease before reading the
/// queue and did not derive its input from an earlier playback snapshot.
fn integration_snapshot_matches(
expected_track_id: Option<u64>,
expected_generation: Option<u64>,
observed_track_id: u64,
observed_generation: u64,
) -> bool {
expected_track_id.is_none_or(|expected| expected == observed_track_id)
&& expected_generation.is_none_or(|expected| expected == observed_generation)
}
/// Fires now-playing immediately, then waits for the exact owner's track to
/// reach `min(50% of duration, 240s)`. No lease is held during the long wait;
/// each observation is re-admitted with the token captured before the queue
/// snapshot, and the irreversible send keeps that exact permit alive.
fn spawn_scrobble(
meta: ScrobbleMeta,
cfg: ScrobblerSettings,
owner_token: OwnerActionToken,
expected_track_id: u64,
expected_generation: u64,
) {
// Last.fm wants the time the track STARTED, captured here on the edge —
// never the time the threshold fired (a 240 s wait on a long track put
// every scrobble four minutes late).
let started_at = unix_now();
crate::spawn(async move {
let rt = crate::app();
let Some(initial_owner_action) = begin_owner_action_exact(owner_token).await else {
return;
};
let initial_event = rt.core().player().get_playback_event();
if !integration_snapshot_matches(
Some(expected_track_id),
Some(expected_generation),
initial_event.track_id,
SCROBBLE_GEN.load(Ordering::SeqCst),
) {
return;
}
// Now-playing is never queued. It may use the independent network in
// manual-offline immediate mode or an opted-in logged-out session.
if current_scrobble_policy(&cfg).await.action == ScrobbleAction::SendNow {
send_now_playing(&meta, &cfg).await;
let current_event = rt.core().player().get_playback_event();
if integration_snapshot_matches(
Some(expected_track_id),
Some(expected_generation),
current_event.track_id,
SCROBBLE_GEN.load(Ordering::SeqCst),
) {
send_now_playing(&meta, &cfg).await;
}
}
drop(initial_owner_action);
// Delayed scrobble. Unknown / too-short duration: skip (Last.fm's
// "longer than 30 seconds" rule lives in qbz_app::scrobble_timing).
@@ -830,26 +976,34 @@ pub fn on_track_changed(meta: ScrobbleMeta) {
// without a scrobble; a newer track edge self-cancels via the
// generation, as before. Seeks move the position and therefore the
// moment this fires — the same rule the daemon applies.
let rt = crate::app();
let mut ticker = tokio::time::interval(SCROBBLE_TICK);
loop {
ticker.tick().await;
if SCROBBLE_GEN.load(Ordering::SeqCst) != my_gen {
let Some(threshold_owner_action) = begin_owner_action_exact(owner_token).await else {
return;
}
};
let ev = rt.core().player().get_playback_event();
if ev.track_id == 0 {
if !integration_snapshot_matches(
Some(expected_track_id),
Some(expected_generation),
ev.track_id,
SCROBBLE_GEN.load(Ordering::SeqCst),
) {
log::debug!(
"[qbz-qt] scrobble: stopped before the threshold, dropping '{}'",
"[qbz-qt] scrobble: owner track changed before the threshold, dropping '{}'",
meta.track
);
return;
}
if ev.is_playing && ev.position >= wait {
break;
// Keep the exact permit through policy revalidation and the
// irreversible external send/queue operation.
send_scrobble(&meta, started_at).await;
drop(threshold_owner_action);
return;
}
drop(threshold_owner_action);
}
send_scrobble(&meta, started_at).await;
});
}
@@ -865,14 +1019,21 @@ pub fn on_track_changed(meta: ScrobbleMeta) {
/// following the peer's tracks while scrobbles are suppressed (the push
/// inside `refresh_now_playing_meta`, playback.rs:2236-2239, which the peer
/// track edge also reaches, :5137-5141). The poll loop's PEER edge therefore
/// calls [`discord_push`] directly and never this entry. The ephemeral-mode
/// calls [`discord_push_observed`] directly and never this entry. The ephemeral-mode
/// product law (scrobble MAY happen) is unaffected — this guard is about
/// REMOTE ownership, not ephemeral.
///
/// GLUE: call from the playback poll's DE-DUPED track-change edge
/// (`track_id != last_track_id`), never from `refresh_now_playing` — that runs
/// on every play/queue republish and would re-arm the scrobble timer.
pub fn on_track_change_edge(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
pub fn on_track_change_edge(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
owner_token: OwnerActionToken,
expected_track_id: u64,
) {
// Bump synchronously at the producer edge. If two spawned queue reads run
// out of order, the older one cannot make itself newest after the fact.
let expected_generation = SCROBBLE_GEN.fetch_add(1, Ordering::SeqCst) + 1;
let cfg = scrobble_settings();
let scrobblers_live = cfg.lastfm_active() || cfg.listenbrainz_active();
if !scrobblers_live && !discord().is_enabled() {
@@ -880,6 +1041,21 @@ pub fn on_track_change_edge(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
}
let rt = runtime.clone();
crate::spawn(async move {
let Some(_owner_action) = begin_owner_action_exact(owner_token).await else {
return;
};
let state = rt.core().get_queue_state().await;
let Some(track) = state.current_track else {
return;
};
if !integration_snapshot_matches(
Some(expected_track_id),
Some(expected_generation),
track.id,
SCROBBLE_GEN.load(Ordering::SeqCst),
) {
return;
}
if scrobblers_live {
// The guard (playback.rs:2267-2271): a remote QConnect renderer
// owns playback — skip the scrobble half ONLY. The spawn's other
@@ -892,14 +1068,24 @@ pub fn on_track_change_edge(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
None => false,
};
if !peer_active {
let state = rt.core().get_queue_state().await;
if let Some(track) = state.current_track {
on_track_changed(meta_from_queue_track(&track));
}
spawn_scrobble(
meta_from_queue_track(&track),
cfg,
owner_token,
expected_track_id,
expected_generation,
);
}
}
// UNGUARDED, 1:1 with the reference — Discord follows the peer.
discord_push(&rt);
if discord().is_enabled() {
push_discord_current(
&rt,
Some(expected_track_id),
Some(expected_generation),
)
.await;
}
});
}
@@ -1349,6 +1535,22 @@ mod tests {
assert!(!no_accumulation.queue_on_failure);
}
#[test]
fn queued_integration_requires_the_exact_track_and_generation() {
assert!(integration_snapshot_matches(Some(41), Some(7), 41, 7));
assert!(!integration_snapshot_matches(Some(41), Some(7), 42, 7));
assert!(!integration_snapshot_matches(Some(41), Some(7), 41, 8));
// A -> B -> A is still stale: the numeric track matches again, but the
// producer generation cannot be revived by the later A edge.
assert!(!integration_snapshot_matches(Some(41), Some(7), 41, 9));
}
#[test]
fn task_owned_snapshot_needs_no_prior_track_identity() {
assert!(integration_snapshot_matches(None, None, 99, 12));
}
#[test]
fn discover_preferences_rebind_between_guest_and_account_profiles() {
let guest = unique_store_dir("guest");
+28 -4
View File
@@ -20,6 +20,8 @@ use qbz_app::listen_log::{meta_from_queue_track, ListenLogger, Origin};
use qbz_models::QueueTrack;
use qbz_player::PlaybackEvent;
use crate::playback_qt::OwnerActionLease;
static LOGGER: OnceLock<RwLock<Option<Arc<ListenLogger>>>> = OnceLock::new();
/// A `track_id == 0` tick is normally the sub-second gap while the engine
@@ -64,6 +66,19 @@ pub async fn teardown() {
}
}
/// Playback authority moved from the signed-in owner to a delegated renderer.
///
/// The delegation transition calls this only after owner actions have drained,
/// so no admitted owner hook can still be writing the row. Delegated playback
/// never opens a replacement row: every hot-path hook below requires the
/// unforgeable owner lease captured with its playback snapshot.
pub async fn handoff() {
EMPTY_TICKS.store(0, std::sync::atomic::Ordering::Relaxed);
if let Some(logger) = logger() {
logger.handoff().await;
}
}
/// Orderly exit: close the row in flight as `shutdown`. Synchronous —
/// called from `main.rs` after the event loop, behind the hard-exit
/// watchdog like the session flush next to it.
@@ -76,7 +91,11 @@ pub fn shutdown_blocking() {
/// The de-duped track edge. `event` is the poll's own `PlaybackEvent`, which
/// carries the ACTUAL stream facts (bit depth / sample rate) — the queue row
/// only knows what the catalogue claimed.
pub async fn on_track_edge(track: &QueueTrack, event: &PlaybackEvent) {
pub async fn on_track_edge(
_owner_action: &OwnerActionLease,
track: &QueueTrack,
event: &PlaybackEvent,
) {
// Ephemeral rows (CD / ad-hoc folder) never reach any history — owner
// rule, same guard as `recently_qt::record_queue_track`.
if crate::local_ephemeral::is_ephemeral_id(track.id as i64) {
@@ -103,7 +122,12 @@ pub async fn on_track_edge(track: &QueueTrack, event: &PlaybackEvent) {
/// One poll tick. `position_secs` is the engine's whole-second position;
/// the accumulator only credits monotonic deltas of at most 5 s while
/// `is_playing`, so pauses and seeks add nothing.
pub async fn on_tick(track_id: u64, position_secs: u64, is_playing: bool) {
pub async fn on_tick(
_owner_action: &OwnerActionLease,
track_id: u64,
position_secs: u64,
is_playing: bool,
) {
let Some(logger) = logger() else {
return;
};
@@ -124,14 +148,14 @@ pub async fn on_tick(track_id: u64, position_secs: u64, is_playing: bool) {
/// The poll loop's natural end-of-track edge (the `track_ended` predicate),
/// BEFORE anything advances the cursor.
pub async fn on_natural_end() {
pub async fn on_natural_end(_owner_action: &OwnerActionLease) {
if let Some(logger) = logger() {
logger.ended_naturally().await;
}
}
/// A stream failure on the current track.
pub async fn on_error() {
pub async fn on_error(_owner_action: &OwnerActionLease) {
if let Some(logger) = logger() {
logger.errored().await;
}
+3
View File
@@ -969,6 +969,9 @@ async fn enqueue_rows(runtime: &Runtime, tracks: Vec<LocalTrack>, mode: &str) {
if queue.is_empty() {
return;
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
match mode {
// Reversed so a multi-track insert at the cursor keeps its order.
"next" => {
+3
View File
@@ -301,6 +301,9 @@ async fn enqueue_rows(rows: Vec<LocalTrack>, mode: &str) {
.unwrap_or_default();
let queue: Vec<QueueTrack> = rows.iter().map(local_queue_track).collect();
log::info!("[qbz-qt] local bulk {mode}: {} track(s)", queue.len());
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
match mode {
"play-next" => {
for t in queue.into_iter().rev() {
+6
View File
@@ -794,6 +794,9 @@ pub fn set_session_artwork(path: &str) {
let url = qbz_models::fs_url::file_url(&path);
crate::spawn(async move {
let runtime = crate::app();
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
if runtime.core().patch_queue_artwork(&ids, &url).await {
crate::playback_qt::refresh_now_playing(&runtime).await;
}
@@ -1071,6 +1074,9 @@ async fn play_rows(runtime: &Runtime, tracks: Vec<LocalTrack>, start: usize) {
if queue.is_empty() {
return;
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let start = start.min(queue.len() - 1);
let first = queue[start].clone();
runtime.core().set_queue(queue, Some(start)).await;
+16 -4
View File
@@ -251,10 +251,8 @@ pub fn local_queue_track(t: &LocalTrack) -> QueueTrack {
let is_offline = src == "qobuz_download";
let is_plex = src == "plex";
let is_media = src == "jellyfin" || src == "subsonic";
let artwork_url = crate::local_rows::portable_artwork_ref(
t,
crate::local_rows::ArtworkScope::Track,
);
let artwork_url =
crate::local_rows::portable_artwork_ref(t, crate::local_rows::ArtworkScope::Track);
let sample_rate_khz = if t.sample_rate >= 1000.0 {
t.sample_rate / 1000.0
} else {
@@ -395,6 +393,11 @@ pub(crate) async fn play_audible(runtime: &Runtime, track: &QueueTrack) -> bool
if crate::playback_qt::route_play_remote_track(runtime, track, "play_audible").await {
return true;
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
// The delegated queue is already audible through its stamped engine;
// a local click is a handled refusal, never a second backend start.
return true;
};
if belongs_to_the_offline_tier(track) {
return match crate::playback_qt::play_resolved_offline_aware(runtime, track.id, 0).await {
Ok(()) => true,
@@ -586,6 +589,9 @@ pub(crate) async fn play_rows(
log::warn!("[qbz-qt] local play: prepared queue has no anchor");
return;
};
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
// A source-owned play is staged: first prove that the player accepted the
// replacement, then publish its queue. A failed Plex part or moved local
@@ -810,6 +816,9 @@ pub(crate) async fn enqueue_rows(runtime: &Runtime, tracks: Vec<LocalTrack>, mod
// Same stamping seam the Qobuz enqueue paths use, so an appended local
// block carries its own origin instead of inheriting whatever is playing.
let queue = crate::playback_qt::stamped(tracks.iter().map(local_queue_track).collect(), None);
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
// Same core helpers the Qobuz rows use: "next" inserts at the cursor
// (reversed so a multi-track insert keeps its order), "later" appends to
// the block tail, anything else appends.
@@ -1131,6 +1140,9 @@ pub async fn play_single_from_source(runtime: &Runtime, track_id: u64, source: &
log::warn!("[qbz-qt] {source} play: prepared queue has no anchor");
return false;
};
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return true;
};
let played = play_audible(runtime, &first).await;
if !played {
report_staged_playback_failure(&first);
+9 -7
View File
@@ -1225,11 +1225,10 @@ pub async fn load_qobuz_offline(playlist_id: u64) -> bool {
/// Play the open local detail from `start_row_id` ("" = from the top).
///
/// D8: an offline-only playlist stamps the queue via
/// `set_queue_offline_only`, which is what keeps ANY of its tracks from being
/// pushed to Qobuz Connect. The flag is set BEFORE the queue is installed and
/// cleared for every non-offline-only playlist, or a stamp would leak into the
/// next thing the user plays.
/// D8: an offline-only playlist installs the queue together with its
/// offline-only stamp, which keeps ANY of its tracks from being pushed to
/// Qobuz Connect. Every non-offline-only replacement clears the stamp so it
/// cannot leak into the next thing the user plays.
pub async fn play(runtime: &Runtime, start_row_id: &str) {
play_in(runtime, start_row_id, false).await
}
@@ -1296,7 +1295,9 @@ async fn play_in(runtime: &Runtime, start_row_id: &str, shuffle: bool) {
.position(|t| t.id.to_string() == start_row_id)
.unwrap_or(0)
};
runtime.core().set_queue_offline_only(offline_only);
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
// The "playing from" origin is ("playlist", <local:uuid>) — the reference
// stamps it on BOTH local play paths (`local_playlist.rs:1578` play_all /
// Shuffle and `:1599` play_from_visible), with the explicit note that "the
@@ -1314,11 +1315,12 @@ async fn play_in(runtime: &Runtime, start_row_id: &str, shuffle: bool) {
// MIXED — it can hold real Qobuz tracks beside local files — so the filter
// genuinely fires here, and reading the id off the pre-filter list would
// replay one the core just dropped.
let Some(anchor) = crate::playback_qt::set_queue_stamped(
let Some(anchor) = crate::playback_qt::set_queue_stamped_with_offline_only(
runtime,
queue,
Some(start),
crate::playback_qt::PlayContext::playlist(&playlist_id),
offline_only,
)
.await
else {
+132 -50
View File
@@ -13,8 +13,8 @@
//! async work; results hop back to Qt through the bridge's `CxxQtThread`.
mod auth_qt;
mod listen_log_qt;
mod deep_link_qt;
mod listen_log_qt;
#[cfg(target_os = "linux")]
mod single_instance_qt;
// The Windows sibling. Carries its own `#![cfg(target_os = "windows")]`.
@@ -372,7 +372,9 @@ mod viz_qt;
// credential discovery + device identity, and the renderer engine over
// `runtime.core()`. Plain modules — they declare no #[cxx_qt::bridge], so
// they must NOT appear in build.rs's rust_files. Wired by the B3 facade.
mod qconnect_delegation_qt;
mod qconnect_engine_qt;
mod qconnect_lan_qt;
mod qconnect_transport_qt;
// QConnect port block B3: the facade (service singleton + controller routing)
// and the inbound event sink. Plain modules, same convention as B1/B2. The
@@ -432,17 +434,58 @@ extern "C" fn on_session_commit_data() {
/// Well under the ~5 s Windows allows before it stops waiting.
const COMMIT_DEADLINE: Duration = Duration::from_secs(3);
/// Leave one second of the Windows budget for the bounded SQLite flush.
const QCONNECT_DEADLINE: Duration = Duration::from_secs(2);
let _ = std::panic::catch_unwind(|| {
log::info!("[qbz-qt] session ending; persisting");
log::info!("[qbz-qt] session ending; withdrawing QConnect before persistence");
let (tx, rx) = channel::<()>();
let spawned = std::thread::Builder::new()
.name("qbz-session-commit".to_string())
.spawn(move || {
// A delegated queue is session-scoped and must never become
// the owner's Windows shutdown snapshot. Use the same ordered
// teardown as the ordinary event-loop exit, but reserve time
// for the synchronous SQLite flush below. Fail closed: when
// authority cannot be restored within the budget, preserve
// the previous on-disk snapshot instead of writing guest data.
let qconnect_owner_safe = match (TOKIO.get(), qconnect_qt::service()) {
(_, None) => true,
(Some(runtime), Some(service)) => runtime.block_on(async move {
match tokio::time::timeout(QCONNECT_DEADLINE, service.disconnect()).await {
Ok(Ok(())) => true,
Ok(Err(error)) => {
log::warn!(
"[qbz-qt] Windows QConnect teardown failed: {error}"
);
false
}
Err(_) => {
log::warn!(
"[qbz-qt] Windows QConnect teardown exceeded {}s",
QCONNECT_DEADLINE.as_secs()
);
false
}
}
}),
(None, Some(_)) => {
log::warn!(
"[qbz-qt] Windows QConnect teardown has no runtime; skipping persistence"
);
false
}
};
// `save_on_exit` block_on()s the capture on the tokio handle
// bound at shell entry. Safe from a plain thread; it would
// panic only from inside an async context, which this is not.
qbz_app::session_persist::save_on_exit();
if qconnect_owner_safe {
qbz_app::session_persist::save_on_exit();
} else {
log::warn!(
"[qbz-qt] Windows session persistence skipped to avoid saving delegated state"
);
}
// Listen log: the row in flight closes as `shutdown` here too.
listen_log_qt::shutdown_blocking();
let _ = tx.send(());
@@ -1033,6 +1076,26 @@ pub(crate) fn do_logout() {
deep_link_qt::set_online_session(false);
let runtime = app();
spawn(async move {
// QConnect may currently be rendering with delegated credentials and
// an ephemeral guest queue. Withdraw LAN admission and restore the
// owner authority before the owner token is removed or any session
// state can be persisted under the next account.
if let Some(service) = qconnect_qt::service() {
match service.disconnect_safely().await {
Ok(outcome) if outcome.authority_safe => {}
Ok(_) => {
integrations_qt::set_qobuz_authenticated(true);
deep_link_qt::set_online_session(true);
return;
}
Err(e) => {
log::error!("[qbz-qt] QConnect logout teardown failed: {e}");
integrations_qt::set_qobuz_authenticated(true);
deep_link_qt::set_online_session(true);
return;
}
}
}
// A connected renderer keeps playing after the session ends unless it
// is torn down explicitly (the cast service owns its own socket).
cast_qt::service().shutdown().await;
@@ -1891,6 +1954,9 @@ pub(crate) fn playlist_shuffle() {
// track every time (owner ruling 2026-08-01: every shuffle must be
// genuinely random).
if local_playlist_qt::open_id().is_some() {
let Some(_owner_action) = playback_qt::begin_owner_action() else {
return;
};
runtime.core().set_shuffle(true).await;
now_playing::set_shuffle(true);
local_playlist_qt::play_shuffled(&runtime).await;
@@ -3660,66 +3726,82 @@ fn main() {
// the watchdog `_exit(0)`s the process. Idempotent — the quit paths
// arm it earlier, at the moment quit was requested.
arm_hard_exit_watchdog("event-loop exit");
// Final full snapshot. Placed here — after the loop and AFTER the
// watchdog — rather than in each QML quit handler: window close, tray
// "Quit" and the hotkey all converge on this line, so one call covers
// paths that four scattered ones would keep missing. (The reference
// flushes from the handlers instead — `tray/mod.rs:300`.)
//
// ORDER IS LOAD-BEARING: this is a synchronous `block_on` of a SQLite
// write on the main thread with no UI left. Behind the watchdog it can
// wedge and the process still dies in 5s; in front of it, a wedge is an
// app with no window that only `kill -9` ends — the 2026-08-04 quit
// incident exactly. No-op unless `persist_session` is on.
qbz_app::session_persist::save_on_exit();
// Listen log: close the row in flight as `shutdown`. Same placement
// and the same reasoning as the session flush above (one SQLite
// write on the main thread, behind the watchdog).
listen_log_qt::shutdown_blocking();
// Same reason as logout: leaving the app must stop the renderer.
//
// BOUNDED. This is a `block_on` on the main thread after the UI is
// already gone, and it awaits a tokio `Mutex` that the cast poll task
// also takes (`cast_qt.rs:1111`, aborted only AFTER the lock is
// acquired). If anything holds that lock, the process survives with no
// window, no tray response and no way out but `kill -9` — which is
// exactly what an owner hit on 2026-08-04. A cast device that keeps
// playing for two seconds longer is a far smaller failure than an app
// that cannot be closed, so the timeout wins the tie.
// Same reason as logout: leaving the app must stop every renderer.
// QConnect goes first logically (its future withdraws LAN admission
// before touching authority), while the independent notification and
// cast cleanup run concurrently under the existing five-second hard
// exit budget. Construct every timer inside the runtime: constructing
// `tokio::time::timeout` before `block_on` has no reactor and panics.
let mut qconnect_owner_safe = true;
if let Some(rt) = TOKIO.get() {
// Construct the timer *inside* the runtime. Building
// `tokio::time::timeout` as the argument to `block_on` happens
// before the runtime is entered and panics with "no reactor
// running" during every normal quit.
let notification_withdrawn = rt.block_on(async {
tokio::time::timeout(
std::time::Duration::from_secs(1),
qbz_media_controls::withdraw_track_notification(),
)
.await
let (qconnect_stopped, notification_withdrawn, cast_stopped) = rt.block_on(async {
let qconnect_stop = async {
let Some(service) = qconnect_qt::service() else {
return true;
};
match tokio::time::timeout(
std::time::Duration::from_secs(3),
service.disconnect(),
)
.await
{
Ok(Ok(())) => true,
Ok(Err(error)) => {
log::warn!("[qbz-qt] QConnect shutdown failed: {error}");
false
}
Err(_) => false,
}
};
let notification_stop = async {
tokio::time::timeout(
std::time::Duration::from_secs(1),
qbz_media_controls::withdraw_track_notification(),
)
.await
.is_ok()
};
let cast_stop = async {
tokio::time::timeout(
std::time::Duration::from_secs(2),
cast_qt::service().shutdown(),
)
.await
.is_ok()
};
tokio::join!(qconnect_stop, notification_stop, cast_stop)
});
let notification_withdrawn = notification_withdrawn.is_ok();
qconnect_owner_safe = qconnect_stopped;
if !qconnect_stopped {
log::warn!(
"[qbz-qt] QConnect shutdown did not finish within 3s; \
skipping session persistence to avoid saving delegated state"
);
}
if !notification_withdrawn {
log::warn!(
"[qbz-qt] notification withdrawal did not finish within 1s; exiting anyway"
);
}
let stopped = rt.block_on(async {
tokio::time::timeout(
std::time::Duration::from_secs(2),
cast_qt::service().shutdown(),
)
.await
.is_ok()
});
if !stopped {
if !cast_stopped {
log::warn!(
"[qbz-qt] cast shutdown did not finish within 2s; exiting anyway \
(a cast device may keep playing until it times out)"
);
}
}
// Final full snapshot. QConnect teardown MUST precede this write: a
// delegated queue is ephemeral and may never become the owner's saved
// session. If bounded teardown failed, preserving the previous saved
// snapshot is safer than writing guest state.
if qconnect_owner_safe {
qbz_app::session_persist::save_on_exit();
}
// Listen log: close the row in flight as `shutdown`. Same placement
// and the same reasoning as the session flush above (one SQLite
// write on the main thread, behind the watchdog).
listen_log_qt::shutdown_blocking();
log::info!("[qbz-qt] shutdown complete");
}
// Explicit, INSTRUMENTED drops (2026-08-04 quit incident). Rust would run
+3
View File
@@ -561,6 +561,9 @@ fn dispatch(ev: MediaEvent) {
// sites (`src/cast_qt.rs:309`, `src/qconnect_engine_qt.rs:92`).
// `QbzCore::stop` is synchronous (`crates/qbz-core/src/core.rs:1060`).
MediaEvent::Stop => {
let Some(_transport_action) = crate::playback_qt::begin_transport_action() else {
return;
};
// Bound to a local first: `AppRuntime::core()` returns a BORROW
// (`crates/qbz-app/src/shell.rs:144`) and `app()` hands back an
// owned Arc, so the runtime must outlive the call — the same shape
+15 -1
View File
@@ -329,6 +329,9 @@ pub(crate) async fn play_all_tracks(
crate::toast_qt::error(qbz_i18n::t("This collection resolved to 0 playable tracks"));
return;
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
// Context `None` — `derive_context` infers album-or-artist (T16).
// F1: the anchor comes back from the seam, never from `tracks[0]`.
let Some(anchor) = crate::playback_qt::set_queue_stamped(runtime, tracks, Some(0), None).await
@@ -507,6 +510,9 @@ pub(crate) fn item_action(source_item_id: String, action: String) {
return;
}
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let added_castable = crate::playback_qt::batch_all_qconnect_castable(&tracks);
match mode {
@@ -647,6 +653,9 @@ fn bulk_enqueue_inner(items: Vec<MixtapeCollectionItem>, mode: BulkEnqueueMode)
if crate::playback_qt::route_enqueue_to_peer(&tracks, routed_mode).await {
return;
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let added_castable = crate::playback_qt::batch_all_qconnect_castable(&tracks);
match mode {
@@ -737,6 +746,9 @@ pub(crate) fn play_inline_track(item_source_item_id: String, track_id: String, a
crate::toast_qt::error(qbz_i18n::t("This track is no longer available"));
return;
};
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
match mode {
InlineTrackMode::Play => {
@@ -753,7 +765,9 @@ pub(crate) fn play_inline_track(item_source_item_id: String, track_id: String, a
let first_id = anchor.track_id;
// QConnect CONTROLLER mode (§7): route the play to the peer
// (after the funnel, before the local audible step).
if crate::playback_qt::route_play_remote(&runtime, first_id, "play_inline_track").await {
if crate::playback_qt::route_play_remote(&runtime, first_id, "play_inline_track")
.await
{
return;
}
if let Err(e) =
File diff suppressed because it is too large. Load diff
+12
View File
@@ -1997,6 +1997,9 @@ pub async fn enqueue_playlist_by_id(
if crate::playback_qt::route_enqueue_to_peer(&tracks, mode).await {
return Ok(());
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return Ok(());
};
let added_castable = crate::playback_qt::batch_all_qconnect_castable(&tracks);
match mode {
"next" => {
@@ -2133,6 +2136,9 @@ async fn play_queue_at(
return Err("playlist has no playable tracks".to_string());
}
let start = start.min(tracks.len() - 1);
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return Ok(());
};
// F1 (contract §5.3): the anchor is whatever survived the seam's filter at
// the remapped index. This is the playlist path, and playlists are where
// the owner says pulled tracks show up MOST — reading the id off the
@@ -2186,6 +2192,9 @@ pub async fn enqueue_all(
/// `play_album_shuffled` (playback.rs:3945-3957): mix the Vec, start at 0.
/// The MODE is still raised, so continuing past this list stays shuffled.
pub async fn play_shuffled(runtime: &Arc<AppRuntime<LoggingAdapter>>) -> Result<(), String> {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return Ok(());
};
runtime.core().set_shuffle(true).await;
crate::now_playing::set_shuffle(true);
let mut tracks = current_queue();
@@ -2228,6 +2237,9 @@ pub async fn enqueue_track(
if crate::playback_qt::route_track_to_peer(&qt, mode).await {
return Ok(());
}
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return Ok(());
};
let added_castable = crate::playback_qt::batch_all_qconnect_castable(std::slice::from_ref(&qt));
match mode {
"next" => runtime.core().add_track_next(qt).await,
+14 -3
View File
@@ -156,10 +156,21 @@ impl qbz_qconnect::QbzQConnect {
// value — Some ONLY when the operation succeeded, so a failed
// manual connect never downgrades a remembered "connected".
let (connected, record) = if service.is_running().await {
if let Err(err) = service.disconnect().await {
log::warn!("[QConnect] disconnect failed: {err}");
match service.disconnect_safely().await {
Ok(outcome) if outcome.authority_safe => {
if !outcome.owner_restored {
log::warn!(
"[QConnect] disconnected safely, but owner playback was not restored"
);
}
(false, Some(false))
}
Ok(_) => (true, None),
Err(err) => {
log::warn!("[QConnect] disconnect failed: {err}");
(true, None)
}
}
(false, Some(false))
} else {
match service.connect().await {
Ok(()) => (true, Some(true)),
File diff suppressed because it is too large. Load diff
+581 -79
View File
@@ -1,11 +1,10 @@
//! Qobuz Connect renderer engine for the Qt frontend (block B2 of the
//! 2026-08-01 QConnect Qt-port contract).
//!
//! Behavior-1:1 port of the Slint `qbz/src/qconnect_engine.rs`. Implements
//! [`qconnect_app::QconnectRendererEngine`] over the Qt `AppRuntime`'s `QbzCore`
//! + `Player`, so qbz-qt becomes a QConnect renderer that inherits the shared
//! echo/cursor/materialize/shuffle orchestration in `qconnect_app::renderer`
//! instead of re-deriving it.
//! Implements [`qconnect_app::QconnectRendererEngine`] over the Qt
//! `AppRuntime`'s `QbzCore` + `Player`, so qbz-qt becomes a QConnect renderer
//! that inherits the shared echo/cursor/materialize/shuffle orchestration in
//! `qconnect_app::renderer` instead of re-deriving it.
//!
//! The protected bit-perfect seams (`play_streaming_dynamic` / `play_data`) and
//! the HTTP feeder live here, impl-side, exactly as the Tauri `CoreBridge` impl
@@ -33,10 +32,164 @@ use std::time::Duration;
use async_trait::async_trait;
use qbz_app::shell::AppRuntime;
use qbz_core::{LoggingAdapter, QbzCore};
use qbz_core::{CoreError, LoggingAdapter, QbzCore};
use qbz_models::{Quality, QueueTrack, RepeatMode, Track};
use qbz_player::{BufferWriter, PlaybackState, Player};
use qconnect_app::QconnectRendererEngine;
use qbz_qobuz::{ApiError, DelegatedQobuzClient};
use qconnect_app::{
AuthorityActionPermit, AuthorityCell, AuthorityOrigin, AuthorityStamp, QconnectOwnerFailure,
QconnectRendererEngine,
};
const RETIRED_AUTHORITY_ERROR: &str = "qconnect renderer authority is retired";
/// Credential boundary used for every catalog and stream-resolution request.
/// A delegated context never consults the owner's `QbzCore` Qobuz client.
pub enum RendererCatalogAuthority {
Owner,
Delegated(Arc<DelegatedQobuzClient>),
}
impl std::fmt::Debug for RendererCatalogAuthority {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Owner => formatter.write_str("Owner"),
Self::Delegated(_) => formatter.write_str("Delegated(<redacted>)"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RendererAuthorityOrigin {
Owner,
Delegated,
}
/// Collapse owner API failures before they cross the QConnect engine seam.
/// Never format `error`: its nested reqwest URL or remote body can be sensitive.
fn classify_owner_api_failure(error: &ApiError) -> QconnectOwnerFailure {
match error {
ApiError::AuthenticationError(_)
| ApiError::InvalidAppId
| ApiError::InvalidAppSecret
| ApiError::BundleExtractionError(_) => QconnectOwnerFailure::Authentication,
ApiError::IneligibleUser | ApiError::Forbidden(_) | ApiError::ForbiddenCircuitOpen(_) => {
QconnectOwnerFailure::Authorization
}
ApiError::NonStreamable
| ApiError::InvalidQuality(_)
| ApiError::NoQualityAvailable
| ApiError::TrackUnavailable(_) => QconnectOwnerFailure::TrackUnavailable,
ApiError::OfflineMode => QconnectOwnerFailure::Offline,
ApiError::NetworkError(_) => QconnectOwnerFailure::Network,
ApiError::ParseError(_) | ApiError::ApiResponse(_) => QconnectOwnerFailure::InvalidResponse,
ApiError::RateLimited(_) => QconnectOwnerFailure::RateLimited,
ApiError::ServerError(_) => QconnectOwnerFailure::Server,
}
}
/// Typed owner-core classifier paired with [`classify_owner_api_failure`].
/// String-bearing variants are intentionally reduced to stable categories.
fn classify_owner_core_failure(error: &CoreError) -> QconnectOwnerFailure {
match error {
CoreError::AuthRequired | CoreError::AuthFailed(_) => QconnectOwnerFailure::Authentication,
CoreError::Api(error) => classify_owner_api_failure(error),
CoreError::Player(_) | CoreError::Playback(_) | CoreError::Audio(_) => {
QconnectOwnerFailure::Playback
}
CoreError::Queue(_) | CoreError::Internal(_) => QconnectOwnerFailure::Internal,
CoreError::NotInitialized => QconnectOwnerFailure::Unavailable,
}
}
impl RendererCatalogAuthority {
pub const fn origin(&self) -> RendererAuthorityOrigin {
match self {
Self::Owner => RendererAuthorityOrigin::Owner,
Self::Delegated(_) => RendererAuthorityOrigin::Delegated,
}
}
async fn get_track(
&self,
core: &QbzCore<LoggingAdapter>,
track_id: u64,
) -> Result<Track, String> {
match self {
Self::Owner => core
.get_track(track_id)
.await
.map_err(|err| classify_owner_core_failure(&err).to_string()),
Self::Delegated(client) => client
.get_track(track_id)
.await
.map_err(|err| format!("delegated catalog request for track {track_id}: {err}")),
}
}
async fn get_tracks_batch(
&self,
core: &QbzCore<LoggingAdapter>,
track_ids: &[u64],
) -> Result<Vec<Track>, String> {
match self {
Self::Owner => core
.get_tracks_batch(track_ids)
.await
.map_err(|err| classify_owner_core_failure(&err).to_string()),
Self::Delegated(client) => client
.get_tracks_batch(track_ids)
.await
.map_err(|err| format!("delegated catalog batch request failed: {err}")),
}
}
async fn get_stream_url(
&self,
core: &QbzCore<LoggingAdapter>,
track_id: u64,
quality: Quality,
) -> Result<qbz_models::StreamUrl, String> {
match self {
Self::Owner => core.get_stream_url(track_id, quality).await.map_err(|err| {
format!(
"resolve owner stream for track {track_id}: {}",
classify_owner_core_failure(&err)
)
}),
Self::Delegated(client) => client
.get_stream_url_with_fallback(track_id, quality)
.await
.map_err(|err| format!("resolve delegated stream for track {track_id}: {err}")),
}
}
}
fn authority_origins_match(catalog: RendererAuthorityOrigin, stamp: AuthorityOrigin) -> bool {
matches!(
(catalog, stamp),
(RendererAuthorityOrigin::Owner, AuthorityOrigin::Owner)
| (
RendererAuthorityOrigin::Delegated,
AuthorityOrigin::Delegated { .. }
)
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StreamRecoveryAction {
FullDownload,
OwnerCmaf,
FailClosed,
}
fn recovery_action(origin: RendererAuthorityOrigin, header_flood: bool) -> StreamRecoveryAction {
match (origin, header_flood) {
(RendererAuthorityOrigin::Owner, true) => StreamRecoveryAction::OwnerCmaf,
(RendererAuthorityOrigin::Delegated, true) => StreamRecoveryAction::FailClosed,
(_, false) => StreamRecoveryAction::FullDownload,
}
}
/// QConnect renderer engine backed by the Qt `AppRuntime`. Holds the shared
/// runtime and forwards every trait method through `runtime.core()`; the async
@@ -44,17 +197,102 @@ use qconnect_app::QconnectRendererEngine;
/// awaited from a runtime task).
pub struct QtRendererEngine {
runtime: Arc<AppRuntime<LoggingAdapter>>,
catalog_authority: RendererCatalogAuthority,
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
/// The progressive HTTP task currently feeding the player. Replacing a
/// track, stopping playback, or dropping this engine aborts the old task.
active_feeder: std::sync::Mutex<Option<RemoteStreamFeeder>>,
}
impl QtRendererEngine {
pub fn new(runtime: Arc<AppRuntime<LoggingAdapter>>) -> Self {
Self { runtime }
pub fn owner(
runtime: Arc<AppRuntime<LoggingAdapter>>,
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
) -> Self {
Self::with_authority(runtime, RendererCatalogAuthority::Owner, authority, stamp)
}
pub fn delegated(
runtime: Arc<AppRuntime<LoggingAdapter>>,
delegated_client: Arc<DelegatedQobuzClient>,
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
) -> Self {
Self::with_authority(
runtime,
RendererCatalogAuthority::Delegated(delegated_client),
authority,
stamp,
)
}
fn with_authority(
runtime: Arc<AppRuntime<LoggingAdapter>>,
catalog_authority: RendererCatalogAuthority,
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
) -> Self {
assert!(
authority_origins_match(catalog_authority.origin(), stamp.origin()),
"renderer catalog authority must match its runtime authority stamp"
);
Self {
runtime,
catalog_authority,
authority,
stamp,
active_feeder: std::sync::Mutex::new(None),
}
}
pub const fn authority_origin(&self) -> RendererAuthorityOrigin {
self.catalog_authority.origin()
}
fn is_current(&self) -> bool {
self.authority.is_current(self.stamp)
&& (!matches!(self.stamp.origin(), AuthorityOrigin::Owner)
|| self.authority.owner_actions_allowed())
}
fn ensure_current(&self) -> Result<(), String> {
if self.is_current() {
Ok(())
} else {
Err(RETIRED_AUTHORITY_ERROR.to_string())
}
}
fn action_permit(&self) -> Result<AuthorityActionPermit, String> {
self.authority
.try_runtime_action_permit(self.stamp)
.ok_or_else(|| RETIRED_AUTHORITY_ERROR.to_string())
}
fn core(&self) -> &Arc<QbzCore<LoggingAdapter>> {
self.runtime.core()
}
fn replace_active_feeder(&self, feeder: RemoteStreamFeeder) {
let previous = self
.active_feeder
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.replace(feeder);
drop(previous);
}
fn cancel_active_feeder(&self) {
let feeder = self
.active_feeder
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
drop(feeder);
}
/// Last-resort load for tracks the raw-URL path cannot fetch (the CDN
/// header flood defeats every reqwest attempt — see
/// `is_header_flood_error`): the CMAF path is unaffected by the h1 header
@@ -62,65 +300,111 @@ impl QtRendererEngine {
/// 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(
async fn play_via_owner_cmaf(
&self,
track_id: u64,
quality: Quality,
start_position_secs: u64,
) -> Result<(), String> {
self.core()
let _permit = self.action_permit()?;
if self.authority_origin() != RendererAuthorityOrigin::Owner {
return Err(format!(
"delegated stream for track {track_id} cannot use owner CMAF fallback"
));
}
let playback_result = 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}"))?;
.await;
self.ensure_current()?;
playback_result.map_err(|error| {
format!(
"CMAF fallback for remote track {track_id}: {}",
QconnectOwnerFailure::from_opaque_playback_error(&error)
)
})?;
self.core().sync_current_to_id(track_id).await;
self.ensure_current()?;
Ok(())
}
}
impl Drop for QtRendererEngine {
fn drop(&mut self) {
self.cancel_active_feeder();
}
}
#[async_trait]
impl QconnectRendererEngine for QtRendererEngine {
// ---- transport (sync) ----
fn resume(&self) -> Result<(), String> {
let _permit = self.action_permit()?;
self.core().resume().map_err(|err| err.to_string())
}
fn pause(&self) -> Result<(), String> {
let _permit = self.action_permit()?;
self.core().pause().map_err(|err| err.to_string())
}
fn stop(&self) -> Result<(), String> {
self.cancel_active_feeder();
let _permit = self.action_permit()?;
self.core().stop().map_err(|err| err.to_string())
}
fn seek(&self, position_secs: u64) -> Result<(), String> {
let _permit = self.action_permit()?;
self.core()
.seek(position_secs)
.map_err(|err| err.to_string())
}
fn set_volume(&self, fraction: f32) -> Result<(), String> {
let _permit = self.action_permit()?;
self.core()
.set_volume(fraction)
.map_err(|err| err.to_string())
}
fn get_playback_state(&self) -> PlaybackState {
self.core().get_playback_state()
if self.is_current() {
self.core().get_playback_state()
} else {
PlaybackState::default()
}
}
fn has_loaded_audio(&self) -> bool {
self.core().player().has_loaded_audio()
self.is_current() && self.core().player().has_loaded_audio()
}
// ---- queue / mode (async) ----
async fn set_repeat_mode(&self, mode: RepeatMode) {
self.core().set_repeat_mode(mode).await
if let Ok(_permit) = self.action_permit() {
self.core().set_repeat_mode(mode).await;
}
}
async fn set_shuffle(&self, enabled: bool) {
self.core().set_shuffle(enabled).await
if let Ok(_permit) = self.action_permit() {
self.core().set_shuffle(enabled).await;
}
}
async fn set_shuffle_flag(&self, enabled: bool) {
self.core().set_shuffle_with_order(enabled, None).await
if let Ok(_permit) = self.action_permit() {
self.core().set_shuffle_with_order(enabled, None).await;
}
}
async fn get_all_queue_tracks(&self) -> (Vec<QueueTrack>, Option<usize>) {
self.core().get_all_queue_tracks().await
if !self.is_current() {
return (Vec::new(), None);
}
let tracks = self.core().get_all_queue_tracks().await;
if self.is_current() {
tracks
} else {
(Vec::new(), None)
}
}
async fn set_queue(&self, tracks: Vec<QueueTrack>, start_index: Option<usize>) {
self.core().set_queue(tracks, start_index).await
if let Ok(_permit) = self.action_permit() {
self.core().set_queue(tracks, start_index).await;
}
}
async fn set_queue_with_order(
&self,
@@ -129,29 +413,40 @@ impl QconnectRendererEngine for QtRendererEngine {
shuffle_enabled: bool,
shuffle_order: Option<Vec<usize>>,
) {
self.core()
.set_queue_with_order(tracks, start_index, shuffle_enabled, shuffle_order)
.await
if let Ok(_permit) = self.action_permit() {
self.core()
.set_queue_with_order(tracks, start_index, shuffle_enabled, shuffle_order)
.await;
}
}
async fn clear_queue(&self, keep_current: bool) {
self.core().clear_queue(keep_current).await
if let Ok(_permit) = self.action_permit() {
self.core().clear_queue(keep_current).await;
}
}
async fn play_index(&self, index: usize) -> Option<QueueTrack> {
let _permit = self.action_permit().ok()?;
self.core().play_index(index).await
}
// ---- catalog (async) ----
async fn get_track(&self, track_id: u64) -> Result<Track, String> {
self.core()
.get_track(track_id)
.await
.map_err(|err| err.to_string())
self.ensure_current()?;
let result = self
.catalog_authority
.get_track(self.core(), track_id)
.await;
self.ensure_current()?;
result
}
async fn get_tracks_batch(&self, track_ids: &[u64]) -> Result<Vec<Track>, String> {
self.core()
.get_tracks_batch(track_ids)
.await
.map_err(|err| err.to_string())
self.ensure_current()?;
let result = self
.catalog_authority
.get_tracks_batch(self.core(), track_ids)
.await;
self.ensure_current()?;
result
}
// ---- protected audio seam (the only protected touch) ----
@@ -162,11 +457,14 @@ impl QconnectRendererEngine for QtRendererEngine {
duration_secs: u64,
start_position_secs: u64,
) -> Result<(), String> {
let stream_url = self
.core()
.get_stream_url(track_id, quality)
.await
.map_err(|err| format!("resolve stream url for remote track {track_id}: {err}"))?;
self.cancel_active_feeder();
let _permit = self.action_permit()?;
let stream_url_result = self
.catalog_authority
.get_stream_url(self.core(), track_id, quality)
.await;
self.ensure_current()?;
let stream_url = stream_url_result?;
let player = self.core().player();
let stream_result = stream_remote_track_into_player(
@@ -176,24 +474,43 @@ impl QconnectRendererEngine for QtRendererEngine {
start_position_secs,
&stream_url.url,
"QConnect",
|| self.ensure_current(),
)
.await;
self.ensure_current()?;
let Err(stream_err) = stream_result else {
return Ok(());
let stream_err = match stream_result {
Ok(feeder) => {
self.ensure_current()?;
self.replace_active_feeder(feeder);
if let Err(error) = self.ensure_current() {
self.cancel_active_feeder();
return Err(error);
}
return Ok(());
}
Err(error) => error,
};
// 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 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;
let header_flood = is_header_flood_error(&stream_err);
match recovery_action(self.authority_origin(), header_flood) {
StreamRecoveryAction::OwnerCmaf => {
log::warn!(
"[QConnect] Owner raw-URL streaming hit the CDN header limit for track {track_id}: {stream_err}. Skipping full download; last resort: CMAF."
);
return self
.play_via_owner_cmaf(track_id, quality, start_position_secs)
.await;
}
StreamRecoveryAction::FailClosed => {
log::warn!(
"[QConnect] Delegated raw-URL streaming hit the CDN header limit for track {track_id}: {stream_err}. Owner and CMAF fallback are forbidden."
);
return Err(format!(
"delegated CDN stream for track {track_id} failed: {stream_err}"
));
}
StreamRecoveryAction::FullDownload => {}
}
log::warn!(
@@ -203,17 +520,22 @@ impl QconnectRendererEngine for QtRendererEngine {
);
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}"))?;
self.ensure_current()?;
let playback_result = self.core().player().play_data(audio_data, track_id);
self.ensure_current()?;
playback_result.map_err(|err| format!("play remote track {track_id}: {err}"))?;
Ok(())
}
Err(download_err) if is_header_flood_error(&download_err) => {
Err(download_err)
if recovery_action(
self.authority_origin(),
is_header_flood_error(&download_err),
) == StreamRecoveryAction::OwnerCmaf =>
{
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)
self.play_via_owner_cmaf(track_id, quality, start_position_secs)
.await
}
Err(download_err) => Err(download_err),
@@ -221,6 +543,9 @@ impl QconnectRendererEngine for QtRendererEngine {
}
fn current_output_format(&self) -> Option<(u32, u32)> {
if !self.is_current() {
return None;
}
let player = self.core().player();
Some((player.state.get_sample_rate(), player.state.get_bit_depth()))
}
@@ -256,10 +581,8 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
}
// ---------------------------------------------------------------------------
// HTTP feeder helpers — private copies of the Slint crate's `remote_stream.rs`
// (contract §1.3; the Slint `qconnect_engine.rs` reaches the same bodies via
// `crate::remote_stream::`). Progressive feeder: probe a remote audio URL for
// size + FLAC format, open the player's progressive streaming sink
// HTTP feeder helpers. Progressive feeder: probe a remote audio URL for size
// + FLAC format, open the player's progressive streaming sink
// (`Player::play_streaming_dynamic`), then push the body to the returned
// `BufferWriter` chunk-by-chunk as it arrives. Playback starts as soon as the
// initial buffer fills — not after the whole file lands.
@@ -271,6 +594,30 @@ async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
// is untouched.
// ---------------------------------------------------------------------------
/// Owned background feeder for one progressive remote stream.
///
/// Tokio detaches a task when its bare `JoinHandle` is dropped. This wrapper
/// deliberately aborts instead: the renderer owns exactly one feeder, so an
/// old CDN request cannot keep writing after replacement or authority retire.
#[must_use = "dropping the feeder aborts the background CDN request"]
struct RemoteStreamFeeder {
task: Option<tokio::task::JoinHandle<()>>,
}
impl RemoteStreamFeeder {
fn new(task: tokio::task::JoinHandle<()>) -> Self {
Self { task: Some(task) }
}
}
impl Drop for RemoteStreamFeeder {
fn drop(&mut self) {
if let Some(task) = self.task.take() {
task.abort();
}
}
}
/// Format/size facts sniffed from a remote audio URL before streaming.
struct RemoteStreamInfo {
content_length: u64,
@@ -293,8 +640,10 @@ async fn stream_remote_track_into_player(
start_position_secs: u64,
url: &str,
log_tag: &str,
) -> Result<(), String> {
authority_check: impl FnOnce() -> Result<(), String>,
) -> Result<RemoteStreamFeeder, String> {
let stream_info = probe_remote_stream_info(url).await?;
authority_check()?;
log::info!(
"[{}/STREAMING] Track {} - {:.2} MB, {}Hz, {} ch, {}-bit, {:.1} MB/s",
log_tag,
@@ -322,7 +671,7 @@ 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();
tokio::spawn(async move {
let task = tokio::spawn(async move {
if let Err(err) =
download_and_stream_remote_track(&url, writer, track_id, content_length, &log_tag).await
{
@@ -335,7 +684,7 @@ async fn stream_remote_track_into_player(
}
});
Ok(())
Ok(RemoteStreamFeeder::new(task))
}
/// HEAD for content-length, then a small `Range: bytes=0-65535` GET to (a)
@@ -349,7 +698,7 @@ async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, String>
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|err| format!("create stream probe client: {err}"))?;
.map_err(|_| "create stream probe HTTP client failed".to_string())?;
let head_response = client
.head(url)
@@ -467,7 +816,7 @@ async fn download_and_stream_remote_track(
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(300))
.build()
.map_err(|err| format!("create remote streaming client: {err}"))?;
.map_err(|_| "create remote streaming HTTP client failed".to_string())?;
let response = client
.get(url)
@@ -555,24 +904,60 @@ 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.
/// Return a bounded, URL-free diagnostic for a reqwest failure.
///
/// The raw error and its source chain can contain the signed CDN URL (including
/// its query). Inspect the chain only to retain header-limit classification;
/// never copy an arbitrary cause into logs or a returned error.
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();
if error_chain_has_header_limit(err) {
return safe_transport_diagnostic(
"message head is too large; signed URL and query are intentionally omitted",
)
.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()
}
out
}
/// True when an error message (already chain-expanded by
/// [`describe_reqwest_error`]) shows hyper's hard-coded h1 100-header cap.
fn error_chain_has_header_limit(err: &reqwest::Error) -> bool {
use std::error::Error as _;
if is_header_flood_error(&err.to_string()) {
return true;
}
let mut source = err.source();
while let Some(cause) = source {
if is_header_flood_error(&cause.to_string()) {
return true;
}
source = cause.source();
}
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 a sanitized diagnostic (or an internal raw cause inspected before
/// logging) 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.
@@ -580,3 +965,120 @@ 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")
}
#[cfg(test)]
mod tests {
use super::*;
const SIGNED_URL_MARKER: &str =
"https://api.example.test/track/get?request_ts=42&request_sig=SIGNED-URL-SECRET";
const REMOTE_BODY_MARKER: &str = "REMOTE-BODY\nAuthorization: Bearer jwt_api-secret";
#[test]
fn owner_api_failure_categories_discard_signed_urls_and_remote_bodies() {
let errors = [
ApiError::AuthenticationError(SIGNED_URL_MARKER.to_string()),
ApiError::Forbidden(format!(" : {REMOTE_BODY_MARKER}")),
ApiError::ApiResponse(format!("{REMOTE_BODY_MARKER} {SIGNED_URL_MARKER}")),
];
for error in errors {
let diagnostic = classify_owner_api_failure(&error).to_string();
assert!(!diagnostic.contains("SIGNED-URL-SECRET"));
assert!(!diagnostic.contains("REMOTE-BODY"));
assert!(!diagnostic.contains("Bearer"));
assert!(!diagnostic.contains("jwt_api-secret"));
assert!(!diagnostic.contains('\n'));
}
}
#[test]
fn owner_core_and_opaque_playback_failures_are_payload_free() {
let core_error = CoreError::Api(ApiError::ApiResponse(format!(
"{SIGNED_URL_MARKER} {REMOTE_BODY_MARKER}"
)));
let core_diagnostic = classify_owner_core_failure(&core_error).to_string();
assert_eq!(
core_diagnostic,
QconnectOwnerFailure::InvalidResponse.to_string()
);
assert!(!core_diagnostic.contains("SIGNED-URL-SECRET"));
assert!(!core_diagnostic.contains("REMOTE-BODY"));
let opaque_error = format!("{SIGNED_URL_MARKER} {REMOTE_BODY_MARKER}");
let playback_diagnostic =
QconnectOwnerFailure::from_opaque_playback_error(&opaque_error).to_string();
assert_eq!(
playback_diagnostic,
QconnectOwnerFailure::Playback.to_string()
);
assert!(!playback_diagnostic.contains("SIGNED-URL-SECRET"));
assert!(!playback_diagnostic.contains("REMOTE-BODY"));
}
#[test]
fn delegated_header_flood_fails_closed_without_owner_cmaf() {
assert_eq!(
recovery_action(RendererAuthorityOrigin::Delegated, true),
StreamRecoveryAction::FailClosed
);
assert_ne!(
recovery_action(RendererAuthorityOrigin::Delegated, true),
StreamRecoveryAction::OwnerCmaf
);
}
#[test]
fn delegated_non_header_failure_only_reuses_the_delegated_cdn_url() {
assert_eq!(
recovery_action(RendererAuthorityOrigin::Delegated, false),
StreamRecoveryAction::FullDownload
);
}
#[test]
fn owner_retains_cmaf_recovery_for_the_known_header_limit() {
assert_eq!(
recovery_action(RendererAuthorityOrigin::Owner, true),
StreamRecoveryAction::OwnerCmaf
);
}
#[test]
fn catalog_origin_must_match_the_runtime_stamp_origin() {
assert!(authority_origins_match(
RendererAuthorityOrigin::Owner,
AuthorityOrigin::Owner
));
assert!(authority_origins_match(
RendererAuthorityOrigin::Delegated,
AuthorityOrigin::Delegated { generation: 9 }
));
assert!(!authority_origins_match(
RendererAuthorityOrigin::Owner,
AuthorityOrigin::Delegated { generation: 9 }
));
assert!(!authority_origins_match(
RendererAuthorityOrigin::Delegated,
AuthorityOrigin::Owner
));
}
#[test]
fn cdn_diagnostic_never_echoes_url_or_query() {
let raw = "request failed for https://cdn.example/audio.flac?jwt=super-secret";
let diagnostic = safe_transport_diagnostic(raw);
assert_eq!(diagnostic, "HTTP transport request failed");
assert!(!diagnostic.contains("cdn.example"));
assert!(!diagnostic.contains("super-secret"));
}
#[test]
fn sanitized_header_limit_remains_machine_classifiable() {
let raw = "https://cdn.example/audio?token=secret: message head is too large";
let diagnostic = safe_transport_diagnostic(raw);
assert!(is_header_flood_error(diagnostic));
assert!(!diagnostic.contains("cdn.example"));
assert!(!diagnostic.contains("secret"));
}
}
+289 -40
View File
@@ -29,9 +29,9 @@ use qbz_app::shell::AppRuntime;
use qbz_core::LoggingAdapter;
use qconnect_app::{
build_session_renderer_snapshot, cache_renderer_snapshot, is_peer_renderer_active,
renderer_allows_remote_volume, QconnectApp, QconnectAppEvent, QconnectEventSink,
QconnectRemoteSyncState, QconnectRendererEngine, RendererCommand, RendererReport,
RendererBufferState, RendererReportType,
renderer_allows_remote_volume, AuthorityCell, AuthorityStamp, QconnectApp, QconnectAppEvent,
QconnectEventSink, QconnectRemoteSyncState, QconnectRendererEngine, RendererBufferState,
RendererCommand, RendererReport, RendererReportType,
};
use qconnect_transport_ws::NativeWsTransport;
use serde_json::Value;
@@ -39,11 +39,18 @@ use tokio::sync::Mutex;
use uuid::Uuid;
use crate::qconnect_engine_qt::QtRendererEngine;
use crate::qconnect_lan_qt::QtLanProjectionSlot;
use crate::qconnect_transport_qt::resolve_local_identity;
/// Concrete `QconnectApp` type used by the Qt adapter.
pub type QtQconnectApp = QconnectApp<NativeWsTransport, QtQconnectEventSink>;
const CTRL_SESSION_STATE_MESSAGE_TYPE: &str = "MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE";
fn carries_lan_session_projection(message_type: &str) -> bool {
message_type == CTRL_SESSION_STATE_MESSAGE_TYPE
}
pub struct QtQconnectEventSink {
/// Renderer seam — forwards the `qconnect_app::renderer` orchestration onto
/// `runtime.core()` + the protected player.
@@ -53,6 +60,14 @@ pub struct QtQconnectEventSink {
runtime: Arc<AppRuntime<LoggingAdapter>>,
/// THE shared remote-sync accumulator (one Mutex, shared with `QconnectApp`).
sync_state: Arc<Mutex<QconnectRemoteSyncState>>,
/// Exact credential/runtime authority represented by this sink. Retired
/// runtimes may still deliver late events, so every asynchronous boundary
/// is followed by a stamp check before publishing or reporting.
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
/// Public LAN session projection. Updates are serialized with authority
/// installs so late SESSION_STATE events cannot overwrite a replacement.
projection: QtLanProjectionSlot,
/// Late-bound weak handle to the owning app, wired via `set_app` after the
/// app is built FROM this sink. Used to emit renderer reports (e.g.
/// is_active=true after SetActive(true)) and to drive the session-apply +
@@ -73,25 +88,44 @@ impl QtQconnectEventSink {
engine: QtRendererEngine,
runtime: Arc<AppRuntime<LoggingAdapter>>,
sync_state: Arc<Mutex<QconnectRemoteSyncState>>,
authority: Arc<AuthorityCell>,
stamp: AuthorityStamp,
projection: QtLanProjectionSlot,
) -> Self {
Self {
engine,
runtime,
sync_state,
authority,
stamp,
projection,
app: Arc::new(OnceLock::new()),
last_peer_active: std::sync::atomic::AtomicBool::new(false),
}
}
fn is_current(&self) -> bool {
self.authority.is_current(self.stamp)
}
/// Rebuild the DEV-modal status block (session topology / renderer roles /
/// queue) from the live sync state + app snapshot, and push it to the modal.
async fn refresh_dev_status(&self) {
if !self.is_current() {
return;
}
let Some(app) = self.app.get().and_then(Weak::upgrade) else {
return;
};
let queue = app.queue_state_snapshot().await;
if !self.is_current() {
return;
}
let status = {
let st = self.sync_state.lock().await;
if !self.is_current() {
return;
}
let session = &st.session;
let role = match (session.active_renderer_id, session.local_renderer_id) {
(Some(a), Some(l)) if a == l => "renderer (this device active)",
@@ -127,12 +161,13 @@ impl QtQconnectEventSink {
.collect::<Vec<_>>()
.join("\n")
};
let session_presence = if session.session_uuid.is_some() {
"present"
} else {
"absent"
};
format!(
"Role: {role}\nsession_uuid: {}\nactive_renderer_id: {:?} local_renderer_id: {:?}\nqueue: v{}.{} items={} autoplay={}\nrenderers:\n{}",
session
.session_uuid
.clone()
.unwrap_or_else(|| "-".to_string()),
"Role: {role}\nsession: {session_presence}\nactive_renderer_id: {:?} local_renderer_id: {:?}\nqueue: v{}.{} items={} autoplay={}\nrenderers:\n{}",
session.active_renderer_id,
session.local_renderer_id,
queue.version.major,
@@ -142,6 +177,9 @@ impl QtQconnectEventSink {
renderers,
)
};
if !self.is_current() {
return;
}
crate::qconnect_qt::dev_set_status(status);
}
@@ -150,8 +188,14 @@ impl QtQconnectEventSink {
/// renderer-list source. Maps `session.renderers` -> rows, marking the
/// local device (`is-local`, rendered as "Play here") and the active one.
async fn refresh_device_list(&self) {
if !self.is_current() {
return;
}
let (devices, active_id) = {
let st = self.sync_state.lock().await;
if !self.is_current() {
return;
}
let session = &st.session;
let devices: Vec<crate::qconnect_qt::publish::QconnectDeviceRow> = session
.renderers
@@ -170,7 +214,13 @@ impl QtQconnectEventSink {
(devices, session.active_renderer_id.unwrap_or(-1))
};
if !self.is_current() {
return;
}
crate::qconnect_qt::publish::devices(devices);
if !self.is_current() {
return;
}
crate::qconnect_qt::publish::active_renderer_id(active_id);
}
@@ -184,6 +234,9 @@ impl QtQconnectEventSink {
/// primary writer of `np_remote_volume_locked` (contract §11.3; the second
/// write site is the facade's disconnect tail).
async fn refresh_now_playing_remote_state(&self) {
if !self.is_current() {
return;
}
// Badge gate: if the transport is down (terminal teardown OR a transient
// reconnect blip), the renderer/controller badge must read NOT remote. The
// in-memory session can still name a peer as active_renderer_id long after
@@ -194,11 +247,21 @@ impl QtQconnectEventSink {
// reconnect, TransportConnected re-runs this refresh and the repopulated
// session restores the badge.
let transport_connected = match self.app.get().and_then(Weak::upgrade) {
Some(app) => app.state_handle().lock().await.transport_connected,
Some(app) => {
let state_handle = app.state_handle();
let state = state_handle.lock().await;
if !self.is_current() {
return;
}
state.transport_connected
}
None => false,
};
let (is_remote, cast_target, volume_locked) = {
let st = self.sync_state.lock().await;
if !self.is_current() {
return;
}
let session = &st.session;
let is_remote = transport_connected && is_peer_renderer_active(session);
let active = session.active_renderer_id;
@@ -222,7 +285,13 @@ impl QtQconnectEventSink {
(is_remote, cast_target, volume_locked)
};
if !self.is_current() {
return;
}
crate::now_playing::set_remote(is_remote, &cast_target);
if !self.is_current() {
return;
}
crate::now_playing::set_remote_volume_locked(volume_locked);
}
@@ -233,11 +302,20 @@ impl QtQconnectEventSink {
/// the remote controller. Reads core `current_track()` so it is authoritative
/// regardless of the order QueueUpdated / SetState arrive in.
async fn refresh_local_ui(&self) {
if !self.is_current() {
return;
}
crate::playback_qt::refresh_now_playing(&self.runtime).await;
if !self.is_current() {
return;
}
// The reference also refreshes the queue sidebar here
// (`refresh_sidebar(true)`); the Qt queue panel document is the same
// data surface, published through the existing `queue_qt::publish` path.
crate::queue_qt::publish(&self.runtime).await;
if !self.is_current() {
return;
}
self.refresh_transport_modes().await;
}
@@ -249,7 +327,13 @@ impl QtQconnectEventSink {
/// Matches Tauri, whose button reads `queue.shuffle` (not a per-renderer
/// field the cloud never populates for a peer).
async fn refresh_transport_modes(&self) {
if !self.is_current() {
return;
}
let qs = self.runtime.core().get_queue_state().await;
if !self.is_current() {
return;
}
let shuffle_on = qs.shuffle;
let repeat_mode = match qs.repeat {
qbz_models::RepeatMode::Off => 0,
@@ -257,6 +341,9 @@ impl QtQconnectEventSink {
qbz_models::RepeatMode::One => 2,
};
crate::now_playing::set_shuffle(shuffle_on);
if !self.is_current() {
return;
}
crate::now_playing::set_repeat_mode(repeat_mode);
}
@@ -268,10 +355,16 @@ impl QtQconnectEventSink {
/// Emit a StateUpdated report announcing this renderer is now active. Sent
/// after SetActive(true) is applied so the controller learns we are ready.
async fn report_active_renderer_ready(&self) {
if !self.is_current() {
return;
}
let Some(app) = self.app.get().and_then(Weak::upgrade) else {
return;
};
let queue_version = app.queue_state_snapshot().await.version;
if !self.is_current() {
return;
}
let report = RendererReport::new(
RendererReportType::RndrSrvrStateUpdated,
Uuid::new_v4().to_string(),
@@ -285,7 +378,14 @@ impl QtQconnectEventSink {
}
}),
);
if let Err(err) = app.send_renderer_report_command(report).await {
if !self.is_current() {
return;
}
let result = app.send_renderer_report_command(report).await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!("[QConnect] Failed to report active-renderer-ready: {err}");
}
}
@@ -296,6 +396,9 @@ impl QtQconnectEventSink {
/// `apply_session_management_event`; the post-lock ordering (loop mode ->
/// local-playback handoff -> projection -> freeze -> watchdog) is identical.
async fn apply_session_management_event(&self, message_type: &str, payload: &Value) {
if !self.is_current() {
return;
}
let Some(app) = self.app.get().and_then(Weak::upgrade) else {
return;
};
@@ -303,21 +406,47 @@ impl QtQconnectEventSink {
let outcome = app
.apply_session_management_event(message_type, payload, &identity)
.await;
if !self.is_current() {
return;
}
if carries_lan_session_projection(message_type) {
let session_id = {
let state = self.sync_state.lock().await;
state.session.session_uuid.clone()
};
if !self.is_current() {
return;
}
if let Some(session_id) = session_id {
self.projection
.confirm_owner_session(&self.authority, self.stamp, &session_id);
}
}
if let Some(loop_mode) = outcome.apply_loop_mode {
if let Err(err) =
qconnect_app::renderer::apply_remote_loop_mode(&self.engine, loop_mode).await
{
let result =
qconnect_app::renderer::apply_remote_loop_mode(&self.engine, loop_mode).await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!("[QConnect] Failed to apply remote loop mode: {err}");
}
}
if outcome.sync_local_playback {
self.sync_local_playback_for_renderer_ownership().await;
if !self.is_current() {
return;
}
}
if let Some(renderer_id) = outcome.remote_projection_renderer_id {
self.sync_active_renderer_projection(renderer_id).await;
if !self.is_current() {
return;
}
}
if let Some(renderer_id) = outcome.disconnected_renderer_id {
@@ -326,9 +455,15 @@ impl QtQconnectEventSink {
QconnectAppEvent::RendererDisconnected { renderer_id },
)
.await;
if !self.is_current() {
return;
}
}
if let Some((renderer_id, generation)) = outcome.watchdog_arm {
if !self.is_current() {
return;
}
app.arm_renderer_watchdog(renderer_id, generation);
}
@@ -341,13 +476,23 @@ impl QtQconnectEventSink {
// projection + poll-loop refresh resolve the real current track now.
let peer_active_now = {
let state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
is_peer_renderer_active(&state.session)
};
if !self.is_current() {
return;
}
let was_peer_active = self
.last_peer_active
.swap(peer_active_now, std::sync::atomic::Ordering::Relaxed);
if peer_active_now && !was_peer_active {
if let Err(err) = app.ask_for_active_renderer_state().await {
let result = app.ask_for_active_renderer_state().await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!(
"[QConnect] controller entry: ask_for_active_renderer_state failed: {err}"
);
@@ -359,8 +504,14 @@ impl QtQconnectEventSink {
/// the two don't double-play. Mirrors the Tauri helper, with the engine seam
/// in place of `CoreBridge`.
async fn sync_local_playback_for_renderer_ownership(&self) {
if !self.is_current() {
return;
}
let peer_renderer_active = {
let state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
is_peer_renderer_active(&state.session)
};
if !peer_renderer_active {
@@ -371,6 +522,9 @@ impl QtQconnectEventSink {
if playback_state.track_id == 0 {
return;
}
if !self.is_current() {
return;
}
log::info!(
"[QConnect] Stopping local playback because active renderer is a peer (track_id={})",
@@ -386,8 +540,14 @@ impl QtQconnectEventSink {
/// controller view + a later takeover land on the right track). Mirrors the
/// Tauri helper.
async fn sync_active_renderer_projection(&self, renderer_id: i32) {
if !self.is_current() {
return;
}
let (queue_state, renderer_state, session_loop_mode, should_align_engine) = {
let state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
let Some(active_renderer_id) = state.session.active_renderer_id else {
return;
};
@@ -414,6 +574,9 @@ impl QtQconnectEventSink {
build_session_renderer_snapshot(&queue_state, Some(&renderer_state), session_loop_mode);
{
let mut state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
cache_renderer_snapshot(&mut state, &renderer_snapshot);
}
@@ -424,10 +587,16 @@ impl QtQconnectEventSink {
let Some(current_track) = renderer_snapshot.current_track.as_ref() else {
return;
};
if !self.is_current() {
return;
}
if let Err(err) =
qconnect_app::renderer::align_queue_cursor(&self.engine, current_track.track_id).await
{
let result =
qconnect_app::renderer::align_queue_cursor(&self.engine, current_track.track_id).await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!("[QConnect] Failed to sync peer renderer cursor into engine: {err}");
}
}
@@ -436,31 +605,29 @@ impl QtQconnectEventSink {
#[async_trait]
impl QconnectEventSink for QtQconnectEventSink {
async fn on_event(&self, event: QconnectAppEvent) {
if !self.is_current() {
return;
}
match &event {
QconnectAppEvent::SessionManagementEvent {
message_type,
payload,
} => {
let payload_json =
serde_json::to_string(payload).unwrap_or_else(|_| "?".to_string());
// The server echoes renderer position/state every two seconds.
// Keep topology/session changes visible at info, but do not turn
// ordinary playback into an unbounded terminal transcript.
// ordinary playback into an unbounded terminal transcript. The
// payload is intentionally never logged: it can carry session and
// delegated-credential material.
if message_type == "MESSAGE_TYPE_SRVR_CTRL_RENDERER_STATE_UPDATED" {
log::debug!(
"[QConnect] Session management: {} payload={}",
message_type,
payload_json
);
log::debug!("[QConnect] Session management: {message_type}");
} else {
log::info!(
"[QConnect] Session management: {} payload={}",
message_type,
payload_json
);
log::info!("[QConnect] Session management: {message_type}");
}
self.apply_session_management_event(message_type, payload)
.await;
if !self.is_current() {
return;
}
}
QconnectAppEvent::RendererUpdated(renderer_state) => {
log::debug!(
@@ -470,6 +637,9 @@ impl QconnectEventSink for QtQconnectEventSink {
renderer_state.current_position_ms,
);
let mut sync_state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
cache_renderer_snapshot(&mut sync_state, renderer_state);
}
QconnectAppEvent::QueueUpdated(queue_state) => {
@@ -482,42 +652,63 @@ impl QconnectEventSink for QtQconnectEventSink {
);
{
let mut sync_state = self.sync_state.lock().await;
if !self.is_current() {
return;
}
sync_state.last_remote_queue_state = Some(queue_state.clone());
}
if let Err(err) = qconnect_app::renderer::materialize_remote_queue(
let result = qconnect_app::renderer::materialize_remote_queue(
&self.engine,
&self.sync_state,
queue_state,
)
.await
{
.await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!("[QConnect] Failed to materialize remote queue: {err}");
}
// Reflect the remote queue change in the QBZ UI (queue panel +
// now-playing card). materialize already set the core queue +
// cursor; this just pushes it to Qt.
self.refresh_local_ui().await;
if !self.is_current() {
return;
}
}
QconnectAppEvent::RendererCommandApplied { command, state } => {
// SetState is the routine playback/position command and may be
// republished. Lifecycle commands remain visible at info.
if matches!(command, RendererCommand::SetState { .. }) {
log::debug!("[QConnect] Renderer command applied: {:?}", command);
log::debug!(
"[QConnect] Renderer command applied: {}",
renderer_command_label(command)
);
} else {
log::info!("[QConnect] Renderer command applied: {:?}", command);
log::info!(
"[QConnect] Renderer command applied: {}",
renderer_command_label(command)
);
}
let became_active = matches!(command, RendererCommand::SetActive { active: true });
if let Err(err) = qconnect_app::renderer::apply_renderer_command(
let result = qconnect_app::renderer::apply_renderer_command(
&self.engine,
&self.sync_state,
command,
state,
)
.await
{
.await;
if !self.is_current() {
return;
}
if let Err(err) = result {
log::warn!("[QConnect] Failed to apply renderer command: {err}");
} else if became_active {
self.report_active_renderer_ready().await;
if !self.is_current() {
return;
}
}
// A SetState changes the current track / play-state — reflect it
// in the QBZ now-playing card + queue cursor highlight. A
@@ -526,11 +717,17 @@ impl QconnectEventSink for QtQconnectEventSink {
// full refresh would reset the now-playing card position/art).
if matches!(command, RendererCommand::SetState { .. }) {
self.refresh_local_ui().await;
if !self.is_current() {
return;
}
} else if matches!(
command,
RendererCommand::SetShuffleMode { .. } | RendererCommand::SetLoopMode { .. }
) {
self.refresh_transport_modes().await;
if !self.is_current() {
return;
}
}
}
QconnectAppEvent::RendererUnreachable { renderer_id } => {
@@ -568,16 +765,25 @@ impl QconnectEventSink for QtQconnectEventSink {
_ => {}
}
if !self.is_current() {
return;
}
// DEV diagnostics: log every event (with a relative timestamp) + refresh
// the live status block, so the QconnectDevModal reflects QC state at
// runtime without a rebuild.
crate::qconnect_qt::dev_push_event(dev_event_line(&event));
self.refresh_dev_status().await;
if !self.is_current() {
return;
}
// Controller-mode UI: rebuild the device picker + push the cast-aware
// now-playing state (is-remote / cast-target / volume-locked) from the
// live session topology after every event.
self.refresh_device_list().await;
if !self.is_current() {
return;
}
self.refresh_now_playing_remote_state().await;
}
}
@@ -601,6 +807,18 @@ fn device_icon_key(device_type: Option<i32>, friendly_name: &str) -> &'static st
}
}
fn renderer_command_label(command: &RendererCommand) -> &'static str {
match command {
RendererCommand::SetState { .. } => "set_state",
RendererCommand::SetVolume { .. } => "set_volume",
RendererCommand::SetActive { .. } => "set_active",
RendererCommand::SetMaxAudioQuality { .. } => "set_max_audio_quality",
RendererCommand::SetLoopMode { .. } => "set_loop_mode",
RendererCommand::SetShuffleMode { .. } => "set_shuffle_mode",
RendererCommand::MuteVolume { .. } => "mute_volume",
}
}
/// Format a QConnect event into a one-line DEV-log entry. Big payloads
/// (QueueUpdated / SessionManagement) are summarized; the rest use Debug.
fn dev_event_line(event: &QconnectAppEvent) -> String {
@@ -619,7 +837,9 @@ fn dev_event_line(event: &QconnectAppEvent) -> String {
"RendererUpdated playing={:?} pos={:?}ms vol={:?}",
r.playing_state, r.current_position_ms, r.volume
),
QconnectAppEvent::RendererCommandApplied { command, .. } => format!("Cmd {command:?}"),
QconnectAppEvent::RendererCommandApplied { command, .. } => {
format!("Cmd {}", renderer_command_label(command))
}
QconnectAppEvent::RendererUnreachable { renderer_id } => {
format!("RendererUnreachable #{renderer_id}")
}
@@ -636,6 +856,35 @@ fn dev_event_line(event: &QconnectAppEvent) -> String {
QconnectAppEvent::TransportConnected => "TransportConnected".to_string(),
QconnectAppEvent::TransportDisconnected => "TransportDisconnected".to_string(),
QconnectAppEvent::Diagnostic { channel, level, .. } => format!("diag {channel} [{level}]"),
other => format!("{other:?}"),
QconnectAppEvent::PendingActionStarted { .. } => "PendingActionStarted".to_string(),
QconnectAppEvent::PendingActionCompleted { .. } => "PendingActionCompleted".to_string(),
QconnectAppEvent::PendingActionTimedOut { timeout_ms, .. } => {
format!("PendingActionTimedOut after={timeout_ms}ms")
}
QconnectAppEvent::PendingActionCanceledByConcurrentRemoteEvent { .. } => {
"PendingActionCanceledByConcurrentRemoteEvent".to_string()
}
QconnectAppEvent::QueueErrorIgnoredByConcurrency { .. } => {
"QueueErrorIgnoredByConcurrency".to_string()
}
QconnectAppEvent::QueueResyncTriggered => "QueueResyncTriggered".to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn only_ctrl_session_state_events_carry_a_lan_session_projection() {
assert!(carries_lan_session_projection(
CTRL_SESSION_STATE_MESSAGE_TYPE
));
assert!(!carries_lan_session_projection(
"MESSAGE_TYPE_SRVR_CTRL_RENDERER_STATE"
));
assert!(!carries_lan_session_projection(
"MESSAGE_TYPE_SRVR_CTRL_QUEUE_STATE"
));
}
}
+550
View File
@@ -0,0 +1,550 @@
//! Thin Qt adapter for the official QConnect LAN receiver surface.
//!
//! This module projects Qt's existing QConnect identity into the LAN wire,
//! owns the listener/mDNS lifetime, and drains the bounded latest-wins
//! admission slot on one dedicated thread. Cloud validation and transactional
//! authority switching deliberately remain outside this adapter.
use std::error::Error;
use std::fmt;
use std::future::Future;
use std::sync::{Arc, Mutex as StdMutex};
use std::thread::JoinHandle;
use std::time::Duration;
use qbz_models::Quality;
use qconnect_app::{AuthorityCell, AuthorityOrigin, AuthorityStamp};
use qconnect_lan::{
ConnectInfo, DeviceType, DisplayInfo, EndpointPolicy, HandoffCandidate, LanError,
LanProjection, LanService, LanServiceConfig, MaxAudioQuality,
};
use crate::qconnect_transport_qt::default_qconnect_device_info;
const ADMISSION_WAIT: Duration = Duration::from_secs(60);
const QT_MODEL: &str = "QBZ Desktop";
const ADMISSION_THREAD_NAME: &str = "qbz-qt-qconnect-lan-admission";
/// Stamp-aware bridge between runtime authority and the synchronous LAN
/// projection read by the HTTP worker.
///
/// Every Qt authority install/clear is serialized through this short-held
/// mutex. An old runtime event can therefore never overwrite the session of a
/// replacement runtime, and a prepared candidate is not exposed until the
/// same critical section that installs its authority.
#[derive(Clone, Default)]
pub(crate) struct QtLanProjectionSlot {
inner: Arc<StdMutex<ProjectionSlotState>>,
}
#[derive(Default)]
struct ProjectionSlotState {
projection: Option<LanProjection>,
installed: Option<InstalledSessionProjection>,
}
struct InstalledSessionProjection {
stamp: AuthorityStamp,
session_id: Option<String>,
}
impl QtLanProjectionSlot {
fn lock(&self) -> std::sync::MutexGuard<'_, ProjectionSlotState> {
self.inner.lock().unwrap_or_else(|poisoned| {
log::error!("[QConnect LAN] projection slot recovered from a poisoned lock");
let guard = poisoned.into_inner();
self.inner.clear_poison();
guard
})
}
fn write_projection(state: &ProjectionSlotState, session_id: Option<String>) {
if let Some(projection) = state.projection.as_ref() {
projection.set_current_session_id(session_id);
}
}
/// Install a runtime and its already-confirmed public session atomically
/// with respect to every projection update. `session_id` is `None` for a
/// fresh owner until its first cloud SESSION_STATE arrives.
pub(crate) fn install_authority(
&self,
authority: &AuthorityCell,
stamp: AuthorityStamp,
session_id: Option<&str>,
) -> bool {
let mut state = self.lock();
// Fail closed during the tiny install boundary: the old session may no
// longer be current, while exposing the candidate before install would
// violate the handoff transaction.
Self::write_projection(&state, None);
if !authority.install(stamp) {
let current = authority.current();
let restore = state
.installed
.as_ref()
.filter(|installed| Some(installed.stamp) == current)
.and_then(|installed| installed.session_id.clone());
if current != state.installed.as_ref().map(|installed| installed.stamp) {
state.installed = None;
}
Self::write_projection(&state, restore);
return false;
}
let session_id = session_id.map(str::to_string);
state.installed = Some(InstalledSessionProjection {
stamp,
session_id: session_id.clone(),
});
Self::write_projection(&state, session_id);
true
}
/// Publish a cloud-confirmed owner session from an event belonging to the
/// exact installed runtime. Delegated session ids enter only through
/// `install_authority` after their transactional commit.
pub(crate) fn confirm_owner_session(
&self,
authority: &AuthorityCell,
stamp: AuthorityStamp,
session_id: &str,
) -> bool {
if stamp.origin() != AuthorityOrigin::Owner || session_id.is_empty() {
return false;
}
let mut state = self.lock();
if !authority.is_current(stamp)
|| state.installed.as_ref().map(|installed| installed.stamp) != Some(stamp)
{
return false;
}
let session_id = session_id.to_string();
if let Some(installed) = state.installed.as_mut() {
installed.session_id = Some(session_id.clone());
}
Self::write_projection(&state, Some(session_id));
true
}
/// Snapshot the exact installed session for listener construction. This
/// seeds GET connect-info before the HTTP worker can accept its first call.
pub(crate) fn current_session_id(
&self,
authority: &AuthorityCell,
expected: AuthorityStamp,
) -> Option<String> {
let state = self.lock();
if !authority.is_current(expected) {
return None;
}
state
.installed
.as_ref()
.filter(|installed| installed.stamp == expected)
.and_then(|installed| installed.session_id.clone())
}
pub(crate) fn attach(&self, authority: &AuthorityCell, projection: LanProjection) {
let mut state = self.lock();
let current = authority.current();
let session_id = state
.installed
.as_ref()
.filter(|installed| Some(installed.stamp) == current)
.and_then(|installed| installed.session_id.clone());
projection.set_current_session_id(session_id);
state.projection = Some(projection);
}
pub(crate) fn detach(&self) {
self.lock().projection.take();
}
pub(crate) fn clear_authority(&self, authority: &AuthorityCell) -> Option<AuthorityStamp> {
let mut state = self.lock();
Self::write_projection(&state, None);
let previous = authority.clear();
state.installed = None;
previous
}
pub(crate) fn clear_if_current(
&self,
authority: &AuthorityCell,
expected: AuthorityStamp,
) -> bool {
let mut state = self.lock();
if !authority.is_current(expected) {
return false;
}
Self::write_projection(&state, None);
if authority.clear_if_current(expected) {
state.installed = None;
true
} else {
let restore = state
.installed
.as_ref()
.filter(|installed| authority.is_current(installed.stamp))
.and_then(|installed| installed.session_id.clone());
Self::write_projection(&state, restore);
false
}
}
}
/// Map the effective local playback ceiling to the exact official LAN enum.
/// The same `local_playback_quality` source is used by stream resolution and
/// cloud renderer reports; LAN must not advertise an independent capability.
pub const fn max_audio_quality_for_quality(quality: Quality) -> MaxAudioQuality {
match quality {
Quality::Mp3 => MaxAudioQuality::MP3,
Quality::Lossless => MaxAudioQuality::UpToCd,
Quality::HiRes => MaxAudioQuality::UpToHires96,
Quality::UltraHiRes => MaxAudioQuality::UpToHires192,
}
}
#[derive(Debug)]
pub enum QtLanError {
ExistingIdentityUnavailable,
Service(LanError),
AdmissionThread(std::io::Error),
}
impl fmt::Display for QtLanError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::ExistingIdentityUnavailable => {
formatter.write_str("qconnect LAN identity is unavailable")
}
Self::Service(error) => write!(formatter, "qconnect LAN service failed: {error}"),
Self::AdmissionThread(_) => {
formatter.write_str("qconnect LAN admission thread could not start")
}
}
}
}
impl Error for QtLanError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Service(error) => Some(error),
Self::AdmissionThread(error) => Some(error),
Self::ExistingIdentityUnavailable => None,
}
}
}
impl From<LanError> for QtLanError {
fn from(error: LanError) -> Self {
Self::Service(error)
}
}
struct ExistingIdentity {
device_uuid: String,
friendly_name: String,
brand: String,
software_version: String,
}
impl ExistingIdentity {
fn resolve() -> Result<Self, QtLanError> {
let device_info = default_qconnect_device_info();
let (Some(device_uuid), Some(friendly_name), Some(brand), Some(software_version)) = (
device_info.device_uuid,
device_info.friendly_name,
device_info.brand,
device_info.software_version,
) else {
return Err(QtLanError::ExistingIdentityUnavailable);
};
if device_uuid.trim().is_empty()
|| friendly_name.trim().is_empty()
|| brand.trim().is_empty()
|| software_version.trim().is_empty()
{
return Err(QtLanError::ExistingIdentityUnavailable);
}
Ok(Self {
device_uuid,
friendly_name,
brand,
software_version,
})
}
}
fn build_projection(
identity: &ExistingIdentity,
app_id: String,
max_audio_quality: MaxAudioQuality,
current_session_id: Option<String>,
) -> LanProjection {
LanProjection::new(
DisplayInfo {
friendly_name: identity.friendly_name.clone(),
serial_number: identity.device_uuid.clone(),
brand_display_name: identity.brand.clone(),
model_display_name: QT_MODEL.to_string(),
max_audio_quality,
device_type: DeviceType::Computer,
software_version: identity.software_version.clone(),
},
ConnectInfo {
app_id,
current_session_id,
},
)
}
/// Owns the Qt LAN service and its blocking admission bridge.
///
/// The callback future is driven on `runtime_handle` while the dedicated
/// bridge thread waits for it. Consequently, candidates are admitted in the
/// exact order observed from the latest-wins slot without blocking a Tokio
/// worker or the Qt UI thread. The callback must provide its own bounded and
/// cancellation-safe transaction semantics.
pub struct QtLanRuntime {
service: Option<LanService>,
projection: LanProjection,
admission_bridge: Option<JoinHandle<()>>,
}
impl QtLanRuntime {
pub fn start<F, Fut>(
runtime_handle: tokio::runtime::Handle,
endpoint_policy: EndpointPolicy,
app_id: String,
max_audio_quality: MaxAudioQuality,
current_session_id: Option<String>,
on_handoff: F,
) -> Result<Self, QtLanError>
where
F: Fn(HandoffCandidate) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
let identity = ExistingIdentity::resolve()?;
let projection = build_projection(&identity, app_id, max_audio_quality, current_session_id);
let config =
LanServiceConfig::new(projection.clone(), endpoint_policy, identity.device_uuid);
let (mut service, inbox) = LanService::start(config)?;
let on_handoff = Arc::new(on_handoff);
let admission_bridge = match std::thread::Builder::new()
.name(ADMISSION_THREAD_NAME.to_string())
.spawn(move || loop {
match inbox.take_timeout(ADMISSION_WAIT) {
Some(candidate) => {
let callback = Arc::clone(&on_handoff);
runtime_handle.block_on(callback(candidate));
}
None if inbox.is_closed() => break,
None => {}
}
}) {
Ok(thread) => thread,
Err(error) => {
service.shutdown();
return Err(QtLanError::AdmissionThread(error));
}
};
Ok(Self {
service: Some(service),
projection,
admission_bridge: Some(admission_bridge),
})
}
pub fn projection(&self) -> LanProjection {
self.projection.clone()
}
pub fn port(&self) -> Option<u16> {
self.service.as_ref().map(LanService::port)
}
/// Idempotent blocking teardown in the normative order: withdraw
/// discovery and stop HTTP admission first, then wait for the now-woken
/// bridge to exit. Callers in async or UI contexts must run this method on
/// a blocking worker.
pub fn shutdown_blocking(&mut self) {
if let Some(mut service) = self.service.take() {
service.shutdown();
}
if let Some(bridge) = self.admission_bridge.take() {
let _ = bridge.join();
}
}
}
impl Drop for QtLanRuntime {
fn drop(&mut self) {
self.shutdown_blocking();
}
}
#[cfg(test)]
mod tests {
use super::*;
fn fixture_identity() -> ExistingIdentity {
ExistingIdentity {
device_uuid: "de0c0d70-0000-4000-8000-000000000001".to_string(),
friendly_name: "Desktop probe".to_string(),
brand: "QBZ".to_string(),
software_version: "qbz/2.1.0".to_string(),
}
}
#[test]
fn desktop_projection_uses_existing_identity_and_official_role() {
let projection = build_projection(
&fixture_identity(),
"app-id".to_string(),
MaxAudioQuality::UpToHires192,
None,
);
assert_eq!(
projection.display_info(),
DisplayInfo {
friendly_name: "Desktop probe".to_string(),
serial_number: "de0c0d70-0000-4000-8000-000000000001".to_string(),
brand_display_name: "QBZ".to_string(),
model_display_name: "QBZ Desktop".to_string(),
max_audio_quality: MaxAudioQuality::UpToHires192,
device_type: DeviceType::Computer,
software_version: "qbz/2.1.0".to_string(),
}
);
assert_eq!(
projection.connect_info(),
ConnectInfo {
app_id: "app-id".to_string(),
current_session_id: None,
}
);
}
#[test]
fn quality_mapping_matches_official_lan_families() {
assert_eq!(
max_audio_quality_for_quality(Quality::Mp3),
MaxAudioQuality::MP3
);
assert_eq!(
max_audio_quality_for_quality(Quality::Lossless),
MaxAudioQuality::UpToCd
);
assert_eq!(
max_audio_quality_for_quality(Quality::HiRes),
MaxAudioQuality::UpToHires96
);
assert_eq!(
max_audio_quality_for_quality(Quality::UltraHiRes),
MaxAudioQuality::UpToHires192
);
}
#[test]
fn projection_session_id_changes_only_when_explicitly_updated() {
let projection = build_projection(
&fixture_identity(),
"app-id".to_string(),
MaxAudioQuality::UpToCd,
None,
);
assert_eq!(projection.connect_info().current_session_id, None);
projection.set_current_session_id(Some("delegated-session".to_string()));
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("delegated-session")
);
projection.set_current_session_id(None);
assert_eq!(projection.connect_info().current_session_id, None);
}
#[test]
fn stamped_projection_publishes_only_installed_authority() {
let authority = AuthorityCell::new();
let slot = QtLanProjectionSlot::default();
let projection = build_projection(
&fixture_identity(),
"app-id".to_string(),
MaxAudioQuality::UpToCd,
None,
);
slot.attach(&authority, projection.clone());
let owner = authority.reserve(AuthorityOrigin::Owner);
assert!(slot.install_authority(&authority, owner, None));
assert!(slot.confirm_owner_session(&authority, owner, "owner-session"));
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("owner-session")
);
// Merely reserving/preparing a guest cannot expose its candidate id.
let guest = authority.reserve(AuthorityOrigin::Delegated { generation: 7 });
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("owner-session")
);
assert!(slot.install_authority(&authority, guest, Some("guest-session")));
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("guest-session")
);
// A late event from the retired owner cannot overwrite the guest.
assert!(!slot.confirm_owner_session(&authority, owner, "stale-owner-session"));
// Delegated events never replace the handoff id either; only their
// stamped install is allowed to publish it.
assert!(!slot.confirm_owner_session(&authority, guest, "delegated-event-session"));
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("guest-session")
);
}
#[test]
fn owner_session_events_update_and_stale_clear_cannot_win() {
let authority = AuthorityCell::new();
let slot = QtLanProjectionSlot::default();
let projection = build_projection(
&fixture_identity(),
"app-id".to_string(),
MaxAudioQuality::UpToCd,
None,
);
slot.attach(&authority, projection.clone());
let first_owner = authority.reserve(AuthorityOrigin::Owner);
assert!(slot.install_authority(&authority, first_owner, None));
assert!(slot.confirm_owner_session(&authority, first_owner, "owner-session-a"));
assert!(slot.confirm_owner_session(&authority, first_owner, "owner-session-b"));
assert_eq!(
slot.current_session_id(&authority, first_owner).as_deref(),
Some("owner-session-b")
);
let next_owner = authority.reserve(AuthorityOrigin::Owner);
assert!(slot.install_authority(&authority, next_owner, Some("owner-session-c")));
assert!(!slot.install_authority(&authority, first_owner, Some("stale-owner-session")));
assert!(!slot.clear_if_current(&authority, first_owner));
assert_eq!(
projection.connect_info().current_session_id.as_deref(),
Some("owner-session-c")
);
}
}
File diff suppressed because it is too large. Load diff
+68 -8
View File
@@ -17,6 +17,7 @@ use std::sync::Arc;
use qbz_app::shell::AppRuntime;
use qbz_core::LoggingAdapter;
use qbz_qobuz::ApiError;
use qconnect_transport_ws::WsTransportConfig;
use serde::{Deserialize, Serialize};
@@ -27,6 +28,61 @@ const DEFAULT_QCONNECT_SOFTWARE_PREFIX: &str = "qbz";
type Runtime = Arc<AppRuntime<LoggingAdapter>>;
#[derive(Clone, Copy)]
enum QconnectCredentialFailure {
ApiClientUnavailable,
Network,
Authentication,
Authorization,
Offline,
RateLimited,
Server,
InvalidResponse,
ServiceResponse,
Client,
MissingEndpoint,
}
impl QconnectCredentialFailure {
const fn category(self) -> &'static str {
match self {
Self::ApiClientUnavailable => "api-client-unavailable",
Self::Network => "network",
Self::Authentication => "authentication",
Self::Authorization => "authorization",
Self::Offline => "offline",
Self::RateLimited => "rate-limited",
Self::Server => "server",
Self::InvalidResponse => "invalid-response",
Self::ServiceResponse => "service-response",
Self::Client => "client",
Self::MissingEndpoint => "missing-endpoint",
}
}
}
fn classify_qconnect_credential_error(error: &ApiError) -> QconnectCredentialFailure {
match error {
ApiError::NetworkError(_) => QconnectCredentialFailure::Network,
ApiError::AuthenticationError(_)
| ApiError::InvalidAppId
| ApiError::InvalidAppSecret
| ApiError::BundleExtractionError(_) => QconnectCredentialFailure::Authentication,
ApiError::IneligibleUser | ApiError::Forbidden(_) | ApiError::ForbiddenCircuitOpen(_) => {
QconnectCredentialFailure::Authorization
}
ApiError::OfflineMode => QconnectCredentialFailure::Offline,
ApiError::RateLimited(_) => QconnectCredentialFailure::RateLimited,
ApiError::ServerError(_) => QconnectCredentialFailure::Server,
ApiError::ParseError(_) => QconnectCredentialFailure::InvalidResponse,
ApiError::ApiResponse(_) => QconnectCredentialFailure::ServiceResponse,
ApiError::NonStreamable
| ApiError::InvalidQuality(_)
| ApiError::NoQualityAvailable
| ApiError::TrackUnavailable(_) => QconnectCredentialFailure::Client,
}
}
/// Persistent QConnect device identity — delegates to the shared module so the
/// Qt install resolves the SAME uuid as Slint (same DB path/key/env override).
pub fn resolve_qconnect_device_uuid() -> String {
@@ -421,14 +477,18 @@ pub async fn resolve_transport_config(runtime: &Runtime) -> Result<WsTransportCo
endpoint_url = endpoint_url.or(discovered_endpoint);
jwt_qws = jwt_qws.or(discovered_jwt_qws);
}
Err(err) if endpoint_url.is_some() => {
Err(failure) if endpoint_url.is_some() => {
log::warn!(
"[QConnect] qws/createToken auto-discovery failed, using provided endpoint: {err}"
"[QConnect] qws/createToken auto-discovery failed, using provided endpoint \
(category={})",
failure.category()
);
}
Err(err) => {
Err(failure) => {
return Err(format!(
"QConnect endpoint_url is required (or QBZ_QCONNECT_WS_ENDPOINT). Auto-discovery via qws/createToken failed: {err}"
"QConnect endpoint_url is required (or QBZ_QCONNECT_WS_ENDPOINT). \
Auto-discovery via qws/createToken failed (category={})",
failure.category()
));
}
}
@@ -462,7 +522,7 @@ pub async fn resolve_transport_config(runtime: &Runtime) -> Result<WsTransportCo
async fn fetch_qconnect_transport_credentials(
runtime: &Runtime,
) -> Result<(Option<String>, Option<String>), String> {
) -> Result<(Option<String>, Option<String>), QconnectCredentialFailure> {
// Qt difference (same as Slint): the client is Option-wrapped (None before
// API init).
let client = runtime
@@ -471,18 +531,18 @@ async fn fetch_qconnect_transport_credentials(
.read()
.await
.clone()
.ok_or_else(|| "qws/createToken requires an initialized API client".to_string())?;
.ok_or(QconnectCredentialFailure::ApiClientUnavailable)?;
let (endpoint, _expires_at, jwt) = client
.create_qconnect_token()
.await
.map_err(|err| format!("qws/createToken failed: {err}"))?
.map_err(|error| classify_qconnect_credential_error(&error))?
.into_parts();
let endpoint_url = normalize_opt_string(endpoint);
let jwt_qws = normalize_opt_string(Some(jwt));
if endpoint_url.is_none() {
return Err("qws/createToken response missing jwt_qws.endpoint".to_string());
return Err(QconnectCredentialFailure::MissingEndpoint);
}
Ok((endpoint_url, jwt_qws))
+49 -6
View File
@@ -768,6 +768,9 @@ pub fn set_page(page: i32) {
}
pub async fn play_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_index: usize) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let Some(&upcoming_index) = current_page_indices(runtime).await.get(page_index) else {
log::warn!("[qbz-qt] queue: play_upcoming {page_index} out of range");
return;
@@ -793,6 +796,9 @@ 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 Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
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)
@@ -813,6 +819,9 @@ async fn play_extended_upcoming(
upcoming_index: usize,
expected_id: u64,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
let Some(track) = state.upcoming.get(upcoming_index).cloned() else {
return;
@@ -872,6 +881,9 @@ pub async fn play_extended(
play_extended_upcoming(runtime, phase_index, expected_id).await;
}
"history" => {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
let Some(track) = state.history.get(phase_index).cloned() else {
return;
@@ -938,6 +950,9 @@ pub async fn play_extended(
}
pub async fn remove_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_index: usize) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let Some(&upcoming_index) = current_page_indices(runtime).await.get(page_index) else {
log::warn!("[qbz-qt] queue: remove_upcoming {page_index} out of range");
return;
@@ -950,6 +965,9 @@ pub async fn remove_upcoming_flat(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
upcoming_index: usize,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
if upcoming_index >= state.upcoming.len() {
return;
@@ -973,6 +991,9 @@ pub async fn remove_upcoming_flat(
/// merely on a share this network cannot see, because an unreachable local row
/// is `Unreachable`, not dead, and never enters that predicate.
pub async fn purge_unavailable_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>>) -> usize {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return 0;
};
let doomed: Vec<usize> = {
let state = runtime.core().get_queue_state_full().await;
state
@@ -998,6 +1019,9 @@ pub async fn purge_unavailable_upcoming(runtime: &Arc<AppRuntime<LoggingAdapter>
}
pub async fn remove_all_after(runtime: &Arc<AppRuntime<LoggingAdapter>>, page_index: usize) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let Some(&upcoming_index) = current_page_indices(runtime).await.get(page_index) else {
log::warn!("[qbz-qt] queue: remove_all_after {page_index} out of range");
return;
@@ -1010,6 +1034,9 @@ pub async fn remove_all_after_flat(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
upcoming_index: usize,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
if upcoming_index >= state.upcoming.len() {
return;
@@ -1041,6 +1068,9 @@ pub async fn move_track(
from_page: usize,
to_slot: usize,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let page_rows = current_page_indices(runtime).await;
let Some(&from_q) = page_rows.get(from_page) else {
log::warn!("[qbz-qt] queue: reorder from {from_page} out of range");
@@ -1082,6 +1112,9 @@ pub async fn move_track_flat(
from: usize,
to_slot: usize,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
if from >= state.upcoming.len() || to_slot > state.upcoming.len() {
return;
@@ -1114,6 +1147,9 @@ async fn insert_queue_track_at(
track: QueueTrack,
to_slot: usize,
) -> Result<InsertOutcome, String> {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return Ok(InsertOutcome::Remote);
};
if let Some(service) = crate::qconnect_qt::service() {
if service
.insert_at_slot_on_peer_if_active(track.id, track.source.as_deref(), to_slot)
@@ -1122,7 +1158,6 @@ async fn insert_queue_track_at(
return Ok(InsertOutcome::Remote);
}
}
let added_castable =
crate::playback_qt::batch_all_qconnect_castable(std::slice::from_ref(&track));
runtime.core().add_track(track).await;
@@ -1152,6 +1187,9 @@ pub async fn drop_extended(
target_phase: &str,
to_slot: usize,
) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
match phase {
"upcoming" => {
if target_phase != "upcoming" {
@@ -1197,11 +1235,7 @@ pub async fn drop_extended(
}
if let Some(service) = crate::qconnect_qt::service() {
match service
.insert_at_slot_on_peer_outcome(
track.id,
track.source.as_deref(),
to_slot,
)
.insert_at_slot_on_peer_outcome(track.id, track.source.as_deref(), to_slot)
.await
{
crate::qconnect_qt::PeerInsertAtSlotOutcome::Inserted => {
@@ -1309,6 +1343,9 @@ pub async fn insert_dragged_track(
/// History replay (queue.rs play_history: a fresh single-track queue).
pub async fn play_history(runtime: &Arc<AppRuntime<LoggingAdapter>>, index: usize) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let state = runtime.core().get_queue_state_full().await;
let Some(track) = state.history.get(index).cloned() else {
log::warn!("[qbz-qt] queue: play_history {index} out of range");
@@ -1342,6 +1379,9 @@ pub async fn play_history(runtime: &Arc<AppRuntime<LoggingAdapter>>, index: usiz
/// come back showing "page 3 of 1" filtered by a query with nothing left to
/// match.
pub async fn clear_queue(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let playing = runtime.core().get_playback_state().is_playing;
runtime.core().clear_queue(playing).await;
{
@@ -1358,6 +1398,9 @@ pub async fn clear_queue(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
/// clears it. The core auto-clears the marker on queue mutation, and `publish`
/// reflects whatever it currently holds. `queue.rs::toggle_stop_after` 1:1.
pub async fn toggle_stop_after(runtime: &Arc<AppRuntime<LoggingAdapter>>, id: &str) {
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
return;
};
let Ok(track_id) = id.parse::<u64>() else {
return;
};
+134 -41
View File
@@ -442,15 +442,7 @@ pub(crate) fn windows_toast_identity_registered() -> bool {
let mut handle: HKEY = std::ptr::null_mut();
// SAFETY: `key` is NUL-terminated UTF-16 and outlives the call; `handle` is
// a valid out-slot, only written on success. Read-only access.
let rc = unsafe {
RegOpenKeyExW(
HKEY_CURRENT_USER,
key.as_ptr(),
0,
KEY_READ,
&mut handle,
)
};
let rc = unsafe { RegOpenKeyExW(HKEY_CURRENT_USER, key.as_ptr(), 0, KEY_READ, &mut handle) };
if rc != 0 {
return false;
}
@@ -1680,7 +1672,10 @@ fn gpu_power_choice() -> (Vec<String>, i32) {
// Index 0 is Auto; UI positions are independent of Qt's process-local
// Vulkan indices (which can contain gaps when a CPU adapter is filtered).
let index = crate::renderer_qt::resolve_saved_gpu()
.and_then(|selected| gpus.iter().position(|gpu| gpu.identity == selected.identity))
.and_then(|selected| {
gpus.iter()
.position(|gpu| gpu.identity == selected.identity)
})
.map(|position| position as i32 + 1)
.unwrap_or(0);
(opts, index)
@@ -2685,12 +2680,40 @@ pub async fn publish_snapshot() {
// ---------------------------------------------------------------------------
/// What a change requires of the live player (settings.rs `Apply`).
#[derive(Clone, Copy)]
enum Apply {
None,
Reload,
Reinit,
}
/// Admit one live audio mutation owned by the local settings surface. Some
/// callers persist before they reach the live-player seam, so the refusal is
/// explicit about that split instead of pretending to roll back half an audio
/// settings transaction.
fn begin_audio_owner_action(
action: &str,
preference_may_be_saved: bool,
) -> Option<crate::playback_qt::OwnerActionLease> {
match crate::playback_qt::begin_owner_action() {
Some(permit) => Some(permit),
None => {
log::info!("[qbz-qt] audio settings: {action} refused while QConnect owns playback");
let message = if preference_may_be_saved {
qbz_i18n::t(
"The audio setting was saved but not applied to live playback because QConnect is in control. Try again after playback returns to QBZ.",
)
} else {
qbz_i18n::t(
"QConnect is controlling playback. Return playback to QBZ and try again.",
)
};
crate::toast_qt::info(message);
None
}
}
}
async fn probe_selected_hardware_volume(
audio: &qbz_audio::settings::AudioSettings,
) -> Result<(HardwareVolumeProbe, HardwareVolumeDecision), String> {
@@ -2918,7 +2941,7 @@ async fn set_alsa_hardware_volume(
}
}
fn apply_audio(runtime: &Arc<AppRuntime<LoggingAdapter>>, apply: Apply) {
fn apply_audio_with_owner(runtime: &Arc<AppRuntime<LoggingAdapter>>, apply: Apply) {
let reinit = match apply {
Apply::None => return,
Apply::Reload => false,
@@ -2949,11 +2972,23 @@ fn apply_audio(runtime: &Arc<AppRuntime<LoggingAdapter>>, apply: Apply) {
crate::publish_settings();
}
fn apply_audio(runtime: &Arc<AppRuntime<LoggingAdapter>>, apply: Apply) {
if matches!(apply, Apply::None) {
return;
}
let Some(_owner_action) = begin_audio_owner_action("apply live audio settings", true) else {
return;
};
apply_audio_with_owner(runtime, apply);
}
/// Release the live output and wait for the audio thread to confirm that the
/// PCM/reservation is gone and any PipeWire sink QBZ suspended is awake.
/// `Player::release_device` is intentionally blocking because the ack is the
/// ordering guarantee; keep that wait off Tokio's worker threads.
async fn release_output_device(runtime: &Arc<AppRuntime<LoggingAdapter>>) -> Result<(), String> {
async fn release_output_device_with_owner(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
) -> Result<(), String> {
let player = runtime.core().player();
tokio::task::spawn_blocking(move || player.release_device())
.await
@@ -2975,7 +3010,7 @@ fn requires_alsa_direct_unity(audio: &qbz_audio::settings::AudioSettings) -> boo
/// volume seam does not reconfigure the protected device/backend; the DAC owns
/// level in this mode. A QConnect peer is deliberately exempt because its
/// remote volume is the active control.
pub(crate) async fn maybe_force_bitperfect_volume(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
async fn force_bitperfect_volume_with_owner(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
let audio = match with_audio(|store| store.get_settings()) {
Ok(audio) => audio,
Err(error) => {
@@ -3001,11 +3036,23 @@ pub(crate) async fn maybe_force_bitperfect_volume(runtime: &Arc<AppRuntime<Loggi
crate::now_playing::set_volume(1.0);
}
pub(crate) async fn maybe_force_bitperfect_volume(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
// Do not reject a delegated session merely for reading a setting whose
// active route needs no live correction.
if !requires_alsa_direct_unity(&audio_settings()) {
return;
}
let Some(_owner_action) = begin_audio_owner_action("force bit-perfect volume", true) else {
return;
};
force_bitperfect_volume_with_owner(runtime).await;
}
/// Revalidate a persisted hardware-volume preference after startup or a route
/// change. A successful probe synchronizes QBZ to the physical mixer before
/// any stream reinitialization. A failed probe disables the preference and
/// reloads the player, leaving the protected direct path locked at unity.
pub(crate) async fn reconcile_alsa_hardware_volume(
async fn reconcile_alsa_hardware_volume_with_owner(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
) -> Option<f32> {
let audio = match with_audio(|store| store.get_settings()) {
@@ -3049,7 +3096,7 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
}
set_hardware_volume_ui_status(Some(device_id), HardwareVolumeUiStatus::Unsupported);
if audio.alsa_hardware_volume {
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
}
return None;
}
@@ -3116,7 +3163,7 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
Some(selection.control),
HardwareVolumeUiStatus::Unsupported,
);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
return None;
};
let migrated = selection.source != HardwareVolumeSelectionSource::Persisted;
@@ -3137,7 +3184,7 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
Some(selection.control),
HardwareVolumeUiStatus::Unsupported,
);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
return None;
}
}
@@ -3152,7 +3199,7 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
// Legacy `alsa_hardware_volume=true` had no exact identity in
// the Player's startup snapshot. Reload the newly migrated map
// without reopening the protected PCM route.
apply_audio(runtime, Apply::Reload);
apply_audio_with_owner(runtime, Apply::Reload);
} else {
publish_snapshot().await;
}
@@ -3172,7 +3219,7 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
None,
hardware_volume_choice_status(reason),
);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
None
}
HardwareVolumeDecision::Unsupported { reason, .. } => {
@@ -3189,18 +3236,39 @@ pub(crate) async fn reconcile_alsa_hardware_volume(
None,
hardware_volume_unsupported_status(&reason),
);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
None
}
}
}
pub(crate) async fn reconcile_alsa_hardware_volume(
runtime: &Arc<AppRuntime<LoggingAdapter>>,
) -> Option<f32> {
// Reconciliation can await a mixer probe, seed the Player's live volume and
// reinitialize its output. One permit covers that whole sequence.
let _owner_action = begin_audio_owner_action("reconcile hardware volume", true)?;
reconcile_alsa_hardware_volume_with_owner(runtime).await
}
// ---------------------------------------------------------------------------
// Handlers (settings.rs handle_bool / handle_select / handle_slider /
// handle_string, including the cascades)
// ---------------------------------------------------------------------------
pub async fn settings_bool(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &str, value: bool) {
// Hardware-volume toggles can await a physical mixer write, seed the live
// Player and reinitialize its output. Admit before the first persistence or
// cascade so a rejected delegated action is a complete no-op.
let _hardware_volume_owner_action = if key == "alsa-hardware-volume" {
let Some(owner_action) = begin_audio_owner_action("change hardware volume", false) else {
return;
};
Some(owner_action)
} else {
None
};
// Cross-setting cascades (settings.rs) — force dependents off first.
let mut cascaded = false;
match key {
@@ -3526,9 +3594,11 @@ pub async fn settings_bool(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &str,
if key == "limit-quality-to-device" {
refresh_device_cap(runtime).await;
}
apply_audio(runtime, apply);
if key == "alsa-hardware-volume" {
maybe_force_bitperfect_volume(runtime).await;
apply_audio_with_owner(runtime, apply);
force_bitperfect_volume_with_owner(runtime).await;
} else {
apply_audio(runtime, apply);
}
publish_snapshot().await;
}
@@ -3537,8 +3607,8 @@ pub async fn settings_bool(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &str,
if key == "alsa-hardware-volume" {
// The failed enable path persisted `false`; push that state to
// the player/document and explain why the toggle bounced back.
apply_audio(runtime, Apply::Reinit);
maybe_force_bitperfect_volume(runtime).await;
apply_audio_with_owner(runtime, Apply::Reinit);
force_bitperfect_volume_with_owner(runtime).await;
publish_snapshot().await;
crate::toast_qt::error(qbz_i18n::t(
"This ALSA device has no compatible hardware volume control. Direct playback remains fixed at 100%.",
@@ -3563,6 +3633,12 @@ async fn select_alsa_hardware_volume_control(
log::warn!("[qbz-qt] ignored mixer selection outside ALSA Direct");
return;
}
// Selection validation can write a mixer control and seed live volume.
// Refuse before the async probe so it cannot straddle a handoff.
let Some(_owner_action) = begin_audio_owner_action("select hardware volume control", true)
else {
return;
};
let device_id = audio
.output_device
.clone()
@@ -3592,7 +3668,7 @@ async fn select_alsa_hardware_volume_control(
log::warn!("[qbz-qt] ALSA mixer revalidation failed: {error}");
let _ = with_audio(|store| store.set_alsa_hardware_volume(false));
set_hardware_volume_ui_status(Some(device_id), HardwareVolumeUiStatus::Unsupported);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
publish_snapshot().await;
return;
}
@@ -3613,7 +3689,7 @@ async fn select_alsa_hardware_volume_control(
);
let _ = with_audio(|store| store.set_alsa_hardware_volume(false));
set_hardware_volume_ui_probe(device_id, &probe, None, HardwareVolumeUiStatus::Stale);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
publish_snapshot().await;
return;
};
@@ -3621,7 +3697,7 @@ async fn select_alsa_hardware_volume_control(
log::warn!("[qbz-qt] ALSA mixer selection has no stable route identity");
let _ = with_audio(|store| store.set_alsa_hardware_volume(false));
set_hardware_volume_ui_probe(device_id, &probe, None, HardwareVolumeUiStatus::Unsupported);
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
publish_snapshot().await;
return;
};
@@ -3633,13 +3709,13 @@ async fn select_alsa_hardware_volume_control(
};
match enable_hardware_volume_selection(runtime, device_id, probe, selection).await {
Ok(apply) => {
apply_audio(runtime, apply);
maybe_force_bitperfect_volume(runtime).await;
apply_audio_with_owner(runtime, apply);
force_bitperfect_volume_with_owner(runtime).await;
}
Err(error) => {
log::error!("[qbz-qt] persist ALSA mixer selection failed: {error}");
let _ = with_audio(|store| store.set_alsa_hardware_volume(false));
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
}
}
publish_snapshot().await;
@@ -3663,6 +3739,10 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
runtime.core().player().clear_audio_cache();
}
"backend" => {
let Some(_owner_action) = begin_audio_owner_action("change audio backend", false)
else {
return;
};
// Index 0 = "Auto" (resolve-and-set, #470): PipeWire when present,
// else System. Indices >= 1 map to the concrete backends list.
let backend = if index == 0 {
@@ -3685,7 +3765,7 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
// initialized. Reinit used to drop the ALSA PCM but omitted
// the suspended-sink cleanup, while Refresh merely queued a
// release and raced its own re-enumeration.
if let Err(error) = release_output_device(runtime).await {
if let Err(error) = release_output_device_with_owner(runtime).await {
report_release_failure(&error);
}
}
@@ -3736,10 +3816,14 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
// device SELECTION did not. D7 also makes the new backend the one
// that resolves that default.
refresh_device_cap(runtime).await;
apply_audio(runtime, Apply::Reinit);
maybe_force_bitperfect_volume(runtime).await;
apply_audio_with_owner(runtime, Apply::Reinit);
force_bitperfect_volume_with_owner(runtime).await;
}
"device" => {
let Some(_owner_action) = begin_audio_owner_action("change output device", false)
else {
return;
};
let id = {
let ids = MAPS.lock().unwrap().1.clone();
ids.get(index).cloned()
@@ -3758,8 +3842,8 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
}
// #638 fix 3, trigger 4 — a different DAC has a different ceiling.
refresh_device_cap(runtime).await;
reconcile_alsa_hardware_volume(runtime).await;
apply_audio(runtime, Apply::Reinit);
reconcile_alsa_hardware_volume_with_owner(runtime).await;
apply_audio_with_owner(runtime, Apply::Reinit);
}
"dsd-mode" => {
let Some(mode) = DSD_MODE_VALUES.get(index) else {
@@ -3772,6 +3856,9 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
apply_audio(runtime, Apply::Reinit);
}
"alsa-plugin" => {
let Some(_owner_action) = begin_audio_owner_action("change ALSA plugin", false) else {
return;
};
let Some(plugin) = ALSA_PLUGIN_VALUES.get(index).copied() else {
return;
};
@@ -3779,9 +3866,9 @@ pub async fn settings_select(runtime: &Arc<AppRuntime<LoggingAdapter>>, key: &st
log::error!("[qbz-qt] persist alsa plugin failed: {e}");
return;
}
reconcile_alsa_hardware_volume(runtime).await;
apply_audio(runtime, Apply::Reinit);
maybe_force_bitperfect_volume(runtime).await;
reconcile_alsa_hardware_volume_with_owner(runtime).await;
apply_audio_with_owner(runtime, Apply::Reinit);
force_bitperfect_volume_with_owner(runtime).await;
}
"retry-behavior" => {
let Some(behavior) = RETRY_BEHAVIOR_VALUES.get(index) else {
@@ -4141,6 +4228,9 @@ pub async fn settings_string(key: &str, value: String) {
/// "Reset to defaults" — restores Audio + Playback defaults
/// (settings.rs handle_reset: store resets + apply + snapshot).
pub async fn settings_reset(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
let Some(_owner_action) = begin_audio_owner_action("reset live audio settings", false) else {
return;
};
if let Err(e) = with_audio(|s| s.reset_all().map(|_| ())) {
log::error!("[qbz-qt] audio settings reset failed: {e}");
}
@@ -4158,16 +4248,19 @@ pub async fn settings_reset(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
// so this refresh is the one that CLEARS the cap (and, through the
// before/after comparison, drops the tier-keyed cache the old cap filled).
refresh_device_cap(runtime).await;
apply_audio(runtime, Apply::Reinit);
apply_audio_with_owner(runtime, Apply::Reinit);
publish_snapshot().await;
}
/// The refresh/release button next to the output device (settings.rs:
/// frees a held ALSA-exclusive device and re-enumerates).
pub async fn refresh_devices(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
let Some(_owner_action) = begin_audio_owner_action("refresh output devices", false) else {
return;
};
// Release whatever the player holds, then re-enumerate from scratch (the
// whole point of the button is a device that was not in the last list).
if let Err(error) = release_output_device(runtime).await {
if let Err(error) = release_output_device_with_owner(runtime).await {
report_release_failure(&error);
}
invalidate_device_cache();
@@ -4178,7 +4271,7 @@ pub async fn refresh_devices(runtime: &Arc<AppRuntime<LoggingAdapter>>) {
// default". Placed after the re-enumeration so the probe reads the new
// device list, and before the publish so the row lands settled.
refresh_device_cap(runtime).await;
reconcile_alsa_hardware_volume(runtime).await;
reconcile_alsa_hardware_volume_with_owner(runtime).await;
publish_snapshot().await;
}
+11
View File
@@ -65,6 +65,17 @@ pub fn set(runtime: Arc<AppRuntime<LoggingAdapter>>, minutes: i32) {
}
let now = Instant::now();
if now >= deadline {
// A timer can be armed under owner authority and outlive a
// QConnect handoff. Re-admit the expiry itself so that stale
// local automation cannot pause the delegated renderer (and
// cannot race a transition fence). The timer still returns to
// idle: its one deadline has elapsed, only the playback side
// effect is refused.
let Some(_owner_action) = crate::playback_qt::begin_owner_action() else {
log::info!("[qbz-qt] sleep-timer: expiry ignored while QConnect owns playback");
push_state(false, String::new());
return;
};
if runtime.core().get_playback_state().is_playing {
if let Err(e) = runtime.core().pause() {
log::warn!("[qbz-qt] sleep-timer: pause on expiry failed: {e}");