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

fix(playback): fence transport handoffs through source installation

Browse Source 
Serialize renderer execution before acknowledgement, validate queue occurrences and merge deferred transport fields without reviving stale targets. Carry play generations, initial pause and offsets through both hosts and source installation. Complete partial ALSA PCM writes with bounded retries and cancellation, with regression gates for handoff and writer progress.

cx
This commit is contained in:
vicrodh committed 2026-09-14 13:44:13 -06:00
1 parent f539149a30
commit ff2b2a6044
23 files changed
+1197 -846
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/qbz-qt/src/qconnect_engine_qt.rs
+25 -5
View File
@@ -309,6 +309,7 @@ impl QtRendererEngine {
track_id: u64,
quality: Quality,
start_position_secs: u64,
playing: bool,
) -> Result<(), String> {
let _permit = self.action_permit()?;
if self.authority_origin() != RendererAuthorityOrigin::Owner {
@@ -318,7 +319,7 @@ impl QtRendererEngine {
}
let playback_result = self
.core()
.play_track_resolved(track_id, quality, None, None, start_position_secs)
.play_track_resolved_with_state(track_id, quality, None, None, start_position_secs, playing)
.await;
self.ensure_current()?;
playback_result.map_err(|error| {
@@ -374,6 +375,11 @@ impl QconnectRendererEngine for QtRendererEngine {
PlaybackState::default()
}
}
fn loading_state(&self) -> Option<(u64, qbz_player::player::PlaybackBufferState)> {
let event = self.playback_event();
Some((event.buffer_track_id, event.buffer_state))
}
fn has_loaded_audio(&self) -> bool {
self.is_current() && self.core().player().has_loaded_audio()
}
@@ -455,6 +461,17 @@ impl QconnectRendererEngine for QtRendererEngine {
quality: Quality,
duration_secs: u64,
start_position_secs: u64,
) -> Result<(), String> {
self.start_track_stream_with_state(track_id, quality, duration_secs, start_position_secs, true).await
}
async fn start_track_stream_with_state(
&self,
track_id: u64,
quality: Quality,
duration_secs: u64,
start_position_secs: u64,
playing: bool,
) -> Result<(), String> {
self.cancel_active_feeder();
let _permit = self.action_permit()?;
@@ -471,6 +488,7 @@ impl QconnectRendererEngine for QtRendererEngine {
track_id,
duration_secs,
start_position_secs,
playing,
&stream_url.url,
"QConnect",
|| self.ensure_current(),
@@ -498,7 +516,7 @@ impl QconnectRendererEngine for QtRendererEngine {
"[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)
.play_via_owner_cmaf(track_id, quality, start_position_secs, playing)
.await;
}
StreamRecoveryAction::FailClosed => {
@@ -520,7 +538,7 @@ impl QconnectRendererEngine for QtRendererEngine {
match download_remote_audio(&stream_url.url).await {
Ok(audio_data) => {
self.ensure_current()?;
let playback_result = self.core().player().play_data(audio_data, track_id);
let playback_result = self.core().player().play_data_at(audio_data, track_id, start_position_secs, playing);
self.ensure_current()?;
playback_result.map_err(|err| format!("play remote track {track_id}: {err}"))?;
Ok(())
@@ -534,7 +552,7 @@ impl QconnectRendererEngine for QtRendererEngine {
log::warn!(
"[QConnect] Full download hit the CDN header flood for track {track_id}: {download_err}. Last resort: CMAF."
);
self.play_via_owner_cmaf(track_id, quality, start_position_secs)
self.play_via_owner_cmaf(track_id, quality, start_position_secs, playing)
.await
}
Err(download_err) => Err(download_err),
@@ -637,6 +655,7 @@ async fn stream_remote_track_into_player(
track_id: u64,
duration_secs: u64,
start_position_secs: u64,
playing: bool,
url: &str,
log_tag: &str,
authority_check: impl FnOnce() -> Result<(), String>,
@@ -655,7 +674,7 @@ async fn stream_remote_track_into_player(
);
let writer = player
.play_streaming_dynamic(
.play_streaming_dynamic_with_state(
track_id,
stream_info.sample_rate,
stream_info.channels,
@@ -664,6 +683,7 @@ async fn stream_remote_track_into_player(
stream_info.speed_mbps,
duration_secs,
start_position_secs,
playing,
)
.map_err(|err| format!("start streaming remote track {track_id}: {err}"))?;
crates/qbz-qt/src/qconnect_event_sink_qt.rs
+31 -76
View File
@@ -623,6 +623,36 @@ impl QconnectEventSink for QtQconnectEventSink {
self.is_current().then(|| self.engine.playback_event())
}
async fn execute_renderer_command(
&self, command: &RendererCommand, state: &qconnect_app::QConnectRendererState,
) -> Result<(), String> {
if !self.is_current() {
return Err("renderer authority retired".into());
}
if remote_renderer_commands_are_fenced(&*self.sync_state.lock().await) {
return Err("renderer command fenced by local authority".into());
}
qconnect_app::renderer::apply_renderer_command(
&self.engine, &self.sync_state, command, state,
).await?;
if !self.is_current() {
return Err("renderer authority retired during execution".into());
}
if matches!(command, RendererCommand::SetActive { active: true }) && self.engine.has_loaded_audio() {
self.report_active_renderer_ready().await;
}
let (volume, muted) = local_volume_ui_projection(command, state);
if let Some(volume) = volume { crate::now_playing::set_volume(volume); }
if let Some(muted) = muted { crate::now_playing::set_muted(muted); }
if matches!(command, RendererCommand::SetState { .. }) {
self.refresh_local_ui().await;
} else if matches!(command, RendererCommand::SetShuffleMode { .. } | RendererCommand::SetLoopMode { .. }) {
self.refresh_transport_modes().await;
}
if !self.is_current() { return Err("renderer authority retired".into()); }
Ok(())
}
async fn on_event(&self, event: QconnectAppEvent) {
if !self.is_current() {
return;
@@ -727,82 +757,7 @@ impl QconnectEventSink for QtQconnectEventSink {
return;
}
}
QconnectAppEvent::RendererCommandApplied { command, state } => {
let fenced = {
let sync_state = self.sync_state.lock().await;
remote_renderer_commands_are_fenced(&sync_state)
};
if fenced {
log::info!(
"[QConnect] Ignoring stale renderer command while local queue authority settles"
);
return;
}
// 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: {}",
renderer_command_label(command)
);
} else {
log::info!(
"[QConnect] Renderer command applied: {}",
renderer_command_label(command)
);
}
let became_active = matches!(command, RendererCommand::SetActive { active: true });
let result = qconnect_app::renderer::apply_renderer_command(
&self.engine,
&self.sync_state,
command,
state,
)
.await;
if !self.is_current() {
return;
}
let applied = if let Err(err) = result {
log::warn!("[QConnect] Failed to apply renderer command: {err}");
false
} else if became_active {
self.report_active_renderer_ready().await;
if !self.is_current() {
return;
}
true
} else {
true
};
if applied {
let (volume, muted) = local_volume_ui_projection(command, state);
if let Some(volume) = volume {
crate::now_playing::set_volume(volume);
}
if let Some(muted) = muted {
crate::now_playing::set_muted(muted);
}
}
// A SetState changes the current track / play-state — reflect it
// in the QBZ now-playing card + queue cursor highlight. A
// standalone SetShuffleMode / SetLoopMode does NOT move the track,
// so only refresh the lightweight shuffle/repeat button state (a
// 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::RendererCommandApplied { .. } => {}
QconnectAppEvent::RendererUnreachable { renderer_id } => {
log::warn!("[QConnect] Renderer {renderer_id} unreachable");
crate::toast_qt::error(qbz_i18n::t("Qobuz Connect renderer unreachable"));