From 3e414a22db1038030ae9d5123d88b9a90d635168 Mon Sep 17 00:00:00 2001 From: LargeModGames <84450916+LargeModGames@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:44:54 +0200 Subject: [PATCH 1/4] fix(network): gate Spotify transport on the owner --- CHANGELOG.md | 2 + src/core/test_helpers.rs | 22 +++ src/infra/network/mod.rs | 263 ++++++++++++++++++++++------------ src/infra/network/playback.rs | 67 ++++++--- tools/gates.count | 4 +- 5 files changed, 247 insertions(+), 111 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 43a4448d..91d99ecd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ ### Fixed +- **A Listening Party can no longer start Spotify on top of another source**: with Local Files, Subsonic, Internet Radio, YouTube or Qobuz playing, native Spotify streaming is only paused, and the party relay (a guest following the host's state, a host running a guest's command) still picked the Spotify player from its device identity. It transferred and activated the Connect device and loaded the host's track into it, or paused, skipped and seeked the paused player, so Spotify audio started on top of the track you were hearing. Every Spotify playback call now checks who owns the output first and does nothing while another source plays (switching Spotify to the spotatui device says so; switching it to another device still works), and the party ignores messages until Spotify plays again. The host publishes only a Spotify track it plays itself: while it plays another source or the cross-source queue it sends nothing instead of the suspended Spotify track, and a guest ignores a state it cannot play instead of pausing or resuming its own music to match. Party commands now run through the same path as a keypress. + - **The jump and add-to-playlist keys follow the track that plays**: with a Spotify track playing from the cross-source queue, jump to album (`a`), jump to artist (`A`) and add the playing track to a playlist (`w`/`W`) acted on the track of the *suspended* Spotify context, so they opened the album, the artist and the picker for a song you were not hearing; they now use the queued track itself, and jump to context (`o`) says the queue slot has no play context. Under Local Files, Subsonic, Internet Radio, YouTube or Qobuz those five keys (`a`, `A`, `o`, `w`, `W`) say they need a Spotify track playing instead of acting on the suspended one, and with nothing playing the three jump keys say so instead of doing nothing. Like (`F`) already followed the queued track and keeps doing so; it now shares the same resolution. - **Spotify's rate limit no longer crashes spotatui at startup**: with a cached login, startup verifies the token with one `/me` request, and that request went straight through rspotify with none of the pacing and `Retry-After` retries every other Spotify call gets. A 429 there ended the process with `Error: http error: status code 429 Too Many Requests` before the UI existed, and relaunching only added more unpaced hits on a rate limit that is shared by everyone using the same client ID. The check now goes through the same paced, retrying request path as the rest of the app, and when Spotify still answers with anything other than a rejected token (a rate limit that outlasts the retries, an outage, no network) spotatui keeps the cached token and starts instead of quitting; the native-streaming account check asks again with its own retries ([#504](https://github.com/LargeModGames/spotatui/issues/504)). diff --git a/src/core/test_helpers.rs b/src/core/test_helpers.rs index 092d02da..8c7d4719 100644 --- a/src/core/test_helpers.rs +++ b/src/core/test_helpers.rs @@ -2,6 +2,8 @@ use crate::core::app::UserInfo; use crate::core::plugin_api::PlaylistInfo; +#[cfg(feature = "streaming")] +use crate::core::plugin_api::TrackInfo; use chrono::Duration; use rspotify::model::{ idtypes::{PlaylistId, UserId}, @@ -12,6 +14,26 @@ use rspotify::model::{ }; use std::collections::HashMap; +/// A queue-slot [`TrackInfo`] with only its `uri` and `name` set. +#[cfg(feature = "streaming")] +pub fn queued_track(uri: &str, name: &str) -> TrackInfo { + TrackInfo { + uri: Some(uri.to_string()), + name: name.to_string(), + artists: vec!["Artist".to_string()], + album: "Album".to_string(), + duration_ms: 180_000, + id: None, + album_id: None, + artist_refs: Vec::new(), + is_playable: true, + is_local: false, + track_number: 1, + explicit: false, + image_url: None, + } +} + /// Domain [`UserInfo`] for tests. `display_name` mirrors `private_user`. pub fn user_info(id: &str) -> UserInfo { UserInfo { diff --git a/src/infra/network/mod.rs b/src/infra/network/mod.rs index 77b7f19f..d9caf8c7 100644 --- a/src/infra/network/mod.rs +++ b/src/infra/network/mod.rs @@ -15,7 +15,7 @@ pub mod sync; pub mod user; pub mod utils; -use crate::core::app::{App, SPOTIFY_NOT_CONNECTED_STATUS}; +use crate::core::app::{App, PlaybackOwner, SPOTIFY_NOT_CONNECTED_STATUS}; use crate::core::auth; use crate::core::config::{ClientConfig, NCSPOT_CLIENT_ID}; use crate::core::plugin_api::{ShowInfo, TrackInfo}; @@ -24,7 +24,6 @@ use anyhow::anyhow; use rspotify::model::{ album::SimplifiedAlbum, enums::{Country, RepeatState}, - idtypes::{EpisodeId, PlayableId, TrackId}, }; use rspotify::prelude::Id; // `parse_response_code` / `request_token` for the in-TUI login live on this trait. @@ -1498,25 +1497,25 @@ impl Network { _ => return, }; let _ = session; - - let (track_uri, is_playing) = match &app.current_playback_context { - Some(ctx) => { - let uri = match &ctx.item { - Some(rspotify::model::PlayableItem::Track(t)) => { - t.id.as_ref().map(|id| id.uri()).unwrap_or_default() - } - Some(rspotify::model::PlayableItem::Episode(e)) => e.id.uri(), - Some(_) | None => return, - }; - (uri, ctx.is_playing) - } - None => return, + // Publish only what a guest can follow: the same owner rule as the + // command relay, and a Spotify URI (a native `spotify:local:` track has none). + if party_yields_to_local_playback(&app) { + return; + } + let Some(snapshot) = crate::infra::media_metadata::current_playback_snapshot(&app) else { + return; + }; + let Some(track_uri) = snapshot + .item_uri + .filter(|uri| ids::playable_id(uri).is_some()) + else { + return; }; sync::SyncMessage::SyncState { track_uri, - position_ms: app.song_progress_ms as u64, - is_playing, + position_ms: snapshot.progress_ms as u64, + is_playing: snapshot.is_playing, timestamp: sync::now_ms(), } }; @@ -1665,14 +1664,15 @@ impl Network { is_playing: bool, timestamp: u64, ) { - let is_guest = { - let app = self.app.lock().await; - matches!( - &app.party_session, - Some(s) if s.role == sync::PartyRole::Guest - ) - }; - if !is_guest { + if ids::playable_id(&track_uri).is_none() { + return; + } + let mut app = self.app.lock().await; + let follows_host = matches!( + &app.party_session, + Some(s) if s.role == sync::PartyRole::Guest + ) && !party_yields_to_local_playback(&app); + if !follows_host { return; } @@ -1690,7 +1690,6 @@ impl Network { }; let (current_uri, current_is_playing, current_progress) = { - let app = self.app.lock().await; let uri = match &app.current_playback_context { Some(ctx) => match &ctx.item { Some(rspotify::model::PlayableItem::Track(t)) => { @@ -1710,25 +1709,10 @@ impl Network { (uri, playing, progress) }; - let mut switched_track = false; - // Track change takes priority - if current_uri != track_uri && !track_uri.is_empty() { - let playable: Option> = if let Ok(id) = TrackId::from_uri(&track_uri) { - let p: PlayableId<'_> = id.into(); - Some(p.into_static()) - } else if let Ok(id) = EpisodeId::from_uri(&track_uri) { - let p: PlayableId<'_> = id.into(); - Some(p.into_static()) - } else { - None - }; - if let Some(playable_id) = playable { - self - .start_playback(None, Some(vec![playable_id]), None) - .await; - switched_track = true; - } + let switched_track = current_uri != track_uri; + if switched_track { + app.start_playback_uris(vec![track_uri], None); } // Play/pause sync @@ -1736,71 +1720,61 @@ impl Network { // begin playing even when host is paused. if (switched_track && !is_playing) || (!switched_track && current_is_playing != is_playing) { if is_playing { - self.start_playback(None, None, None).await; + app.dispatch(IoEvent::StartPlayback(None, None, None)); } else { - self.pause_playback().await; + app.dispatch(IoEvent::PausePlayback); } } // Position drift correction (>3s triggers seek) let drift = current_progress.abs_diff(compensated_position); - if drift > 3000 && current_uri == track_uri { - self.seek(compensated_position as u32).await; + if drift > 3000 && !switched_track { + app.dispatch(IoEvent::Seek(compensated_position as u32)); } } async fn handle_incoming_playback_command(&mut self, action: sync::PlaybackAction) { - let is_host = { - let app = self.app.lock().await; - matches!( - &app.party_session, - Some(s) if s.role == sync::PartyRole::Host - ) - }; - if !is_host { + let mut app = self.app.lock().await; + let relays = matches!( + &app.party_session, + Some(s) if s.role == sync::PartyRole::Host + ) && !party_yields_to_local_playback(&app); + if !relays { return; } + // Plain Spotify events, not the `App` key chains: a host's Next through + // `App::next_track` would hand the sink to its own queue and lock the + // party out. match action { - sync::PlaybackAction::Play => { - self.start_playback(None, None, None).await; - } - sync::PlaybackAction::Pause => { - self.pause_playback().await; - } - sync::PlaybackAction::NextTrack => { - self.next_track().await; - } - sync::PlaybackAction::PrevTrack => { - self.previous_track().await; - } - sync::PlaybackAction::Seek { position_ms } => { - self.seek(position_ms as u32).await; - } + sync::PlaybackAction::Play => app.dispatch(IoEvent::StartPlayback(None, None, None)), + sync::PlaybackAction::Pause => app.dispatch(IoEvent::PausePlayback), + sync::PlaybackAction::NextTrack => app.dispatch(IoEvent::NextTrack), + sync::PlaybackAction::PrevTrack => app.dispatch(IoEvent::PreviousTrack), + sync::PlaybackAction::Seek { position_ms } => app.dispatch(IoEvent::Seek(position_ms as u32)), sync::PlaybackAction::PlayTrack { uri } => { - let playable: Option> = if let Ok(id) = TrackId::from_uri(&uri) { - let p: PlayableId<'_> = id.into(); - Some(p.into_static()) - } else if let Ok(id) = EpisodeId::from_uri(&uri) { - let p: PlayableId<'_> = id.into(); - Some(p.into_static()) - } else { - None - }; - if let Some(playable_id) = playable { - self - .start_playback(None, Some(vec![playable_id]), None) - .await; + if ids::playable_id(&uri).is_some() { + app.start_playback_uris(vec![uri], None); } } } - // After executing, broadcast updated state - self.sync_playback().await; + // Queued behind the command on the serial pump; the 2 s tick repeats it. + app.dispatch(IoEvent::SyncPlayback); } } +/// The party follows Spotify transport only. Coarser than the transport +/// guard on purpose: a queued Spotify track keeps librespot, but a guest must +/// not drive the host's queue slot. +fn party_yields_to_local_playback(app: &App) -> bool { + matches!( + app.playback_owner(), + PlaybackOwner::Decoded | PlaybackOwner::Queue + ) +} + #[cfg(test)] mod tests { use super::*; @@ -2113,13 +2087,7 @@ mod tests { { let mut app = app.lock().await; app.party_status = sync::PartyStatus::Hosting; - app.party_session = Some(sync::PartySession { - role: sync::PartyRole::Host, - code: "ABC123".to_string(), - guests: Vec::new(), - control_mode: sync::ControlMode::HostOnly, - host_name: "Host".to_string(), - }); + app.party_session = Some(party_session(sync::PartyRole::Host)); } network.process_party_messages().await; @@ -2129,4 +2097,113 @@ mod tests { assert_eq!(app.party_status, sync::PartyStatus::Disconnected); assert!(app.party_session.is_none()); } + + fn party_session(role: sync::PartyRole) -> sync::PartySession { + sync::PartySession { + role, + code: "ABC123".to_string(), + guests: Vec::new(), + control_mode: sync::ControlMode::HostOnly, + host_name: "Host".to_string(), + } + } + + fn app_with_a_session() -> (Arc>, std::sync::mpsc::Receiver) { + let (io_tx, io_rx) = std::sync::mpsc::channel(); + let app = App::new(io_tx, UserConfig::new(), Some(SystemTime::now())); + (Arc::new(Mutex::new(app)), io_rx) + } + + /// A party member whose Spotify client is never called: the relay dispatches. + async fn party_network(app: &Arc>, role: sync::PartyRole) -> Network { + let mut network = session_free_network(app); + network.spotify = Some(AuthCodePkceSpotify::new( + Credentials::default(), + OAuth::default(), + )); + app.lock().await.party_session = Some(party_session(role)); + network + } + + async fn relay(network: &mut Network, message: sync::SyncMessage) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + tx.send(message).unwrap(); + network.party_incoming_rx = Some(rx); + network.process_party_messages().await; + } + + const HOST_TRACK: &str = "spotify:track:0000000000000000000001"; + + fn host_state() -> sync::SyncMessage { + sync::SyncMessage::SyncState { + track_uri: HOST_TRACK.to_string(), + position_ms: 0, + is_playing: true, + timestamp: sync::now_ms(), + } + } + + fn guest_pause() -> sync::SyncMessage { + sync::SyncMessage::PlaybackCommand { + action: sync::PlaybackAction::Pause, + from: None, + } + } + + #[tokio::test] + async fn a_guest_follows_the_host_through_a_dispatched_start() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Guest).await; + + relay(&mut network, host_state()).await; + + assert!(matches!( + rx.try_recv(), + Ok(IoEvent::StartPlayback(None, Some(uris), None)) if uris == [HOST_TRACK] + )); + assert!(rx.try_recv().is_err()); + } + + #[tokio::test] + async fn a_guest_ignores_a_host_state_it_cannot_play() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Guest).await; + + let mut state = host_state(); + if let sync::SyncMessage::SyncState { track_uri, .. } = &mut state { + *track_uri = "qobuz:track:1".to_string(); + } + relay(&mut network, state).await; + + assert!(rx.try_recv().is_err()); + } + + #[tokio::test] + async fn a_host_relays_a_guest_command_then_broadcasts() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Host).await; + + relay(&mut network, guest_pause()).await; + + assert!(matches!(rx.try_recv(), Ok(IoEvent::PausePlayback))); + assert!(matches!(rx.try_recv(), Ok(IoEvent::SyncPlayback))); + assert!(rx.try_recv().is_err()); + } + + #[cfg(feature = "streaming")] + #[tokio::test] + async fn the_relay_yields_to_the_native_queue_slot() { + let (app, rx) = app_with_a_session(); + app.lock().await.queue_now = Some(crate::infra::queue::QueueNowPlaying::Spotify { + track: crate::core::test_helpers::queued_track("spotify:track:queued", "Queued"), + }); + let mut network = party_network(&app, sync::PartyRole::Guest).await; + + relay(&mut network, host_state()).await; + assert!(rx.try_recv().is_err()); + + app.lock().await.party_session.as_mut().unwrap().role = sync::PartyRole::Host; + relay(&mut network, guest_pause()).await; + assert!(rx.try_recv().is_err()); + } } diff --git a/src/infra/network/playback.rs b/src/infra/network/playback.rs index 8a2c84a0..d6bd6965 100644 --- a/src/infra/network/playback.rs +++ b/src/infra/network/playback.rs @@ -422,6 +422,17 @@ fn is_no_active_device_error(e: &anyhow::Error) -> bool { text.contains("no_active_device") || text.contains("no active device") } +/// A decoded source (or a decoded queue item) owns the sink: every Spotify +/// transport call is refused, since the paused librespot and the Web API +/// device are both the wrong player. A queued Spotify track keeps Native. +async fn decoded_source_owns_playback(network: &Network) -> bool { + let refused = network.app.lock().await.active_decoded_source(); + if refused { + log::debug!("Spotify transport refused: a decoded source owns playback"); + } + refused +} + /// Whether a native backend is positioned to claim a failed player command: /// an available player (`start_playback` activates it on `NO_ACTIVE_DEVICE`), /// or a backend/activation still materializing (the `suppressed_*` handlers @@ -1427,6 +1438,9 @@ impl PlaybackNetwork for Network { uris: Option>>, offset: Option, ) { + if decoded_source_owns_playback(self).await { + return; + } let (uris, offset) = if context_id.is_none() { match uris { Some(track_uris) => { @@ -1988,6 +2002,9 @@ impl PlaybackNetwork for Network { } async fn pause_playback(&mut self) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] { let mut app = self.app.lock().await; @@ -2029,6 +2046,9 @@ impl PlaybackNetwork for Network { } async fn next_track(&mut self) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] { let mut app = self.app.lock().await; @@ -2054,6 +2074,9 @@ impl PlaybackNetwork for Network { } async fn previous_track(&mut self) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] { let mut app = self.app.lock().await; @@ -2080,6 +2103,9 @@ impl PlaybackNetwork for Network { } async fn force_previous_track(&mut self) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] if let PlaybackBackend::Native(player) = symmetric_playback_backend(self).await { player.prev(); @@ -2123,6 +2149,9 @@ impl PlaybackNetwork for Network { } async fn seek(&mut self, position_ms: u32) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] if let PlaybackBackend::Native(player) = symmetric_playback_backend(self).await { player.seek(position_ms); @@ -2153,6 +2182,9 @@ impl PlaybackNetwork for Network { } async fn shuffle(&mut self, shuffle_state: bool) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] if let PlaybackBackend::Native(player) = symmetric_playback_backend(self).await { let _ = player.set_shuffle(shuffle_state); @@ -2199,6 +2231,9 @@ impl PlaybackNetwork for Network { } async fn repeat(&mut self, repeat_state: RepeatState) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] if let PlaybackBackend::Native(player) = symmetric_playback_backend(self).await { let _ = player.set_repeat(repeat_state); @@ -2247,6 +2282,9 @@ impl PlaybackNetwork for Network { /// On error we bail and clear everything so the UI falls back to whatever /// the API last reported. async fn change_volume(&mut self, volume: u8) { + if decoded_source_owns_playback(self).await { + return; + } #[cfg(feature = "streaming")] if let PlaybackBackend::Native(player) = symmetric_playback_backend(self).await { player.set_volume(volume); @@ -2304,12 +2342,23 @@ impl PlaybackNetwork for Network { } async fn transfert_playback_to_device(&mut self, device_id: String, persist_device_id: bool) { + #[cfg(feature = "streaming")] + let backend = transfer_playback_backend(self, &device_id).await; + // Only the hand-over to librespot touches the local sink; an external + // device stays a valid target. + #[cfg(feature = "streaming")] + if matches!(backend, PlaybackBackend::Native(_)) && decoded_source_owns_playback(self).await { + self + .show_status_message("Another source owns playback".to_string(), 4) + .await; + return; + } // A device change moves playback off the session's `from_tracks` load; // the app-owned shuffle order no longer describes what plays. #[cfg(feature = "streaming")] self.app.lock().await.clear_native_shuffle_session(); #[cfg(feature = "streaming")] - if let PlaybackBackend::Native(player) = transfer_playback_backend(self, &device_id).await { + if let PlaybackBackend::Native(player) = backend { let activation_time = Instant::now(); let native_device_id = player.device_id(); let _ = player.transfer(None); @@ -2799,21 +2848,7 @@ mod tests { #[cfg(feature = "streaming")] fn queued_track(uri: &str) -> crate::core::plugin_api::TrackInfo { - crate::core::plugin_api::TrackInfo { - uri: Some(uri.to_string()), - name: "Queued".to_string(), - artists: vec!["Artist".to_string()], - album: "Album".to_string(), - duration_ms: 180_000, - id: None, - album_id: None, - artist_refs: Vec::new(), - is_playable: true, - is_local: false, - track_number: 1, - explicit: false, - image_url: None, - } + crate::core::test_helpers::queued_track(uri, "Queued") } #[test] diff --git a/tools/gates.count b/tools/gates.count index a1f97feb..427621b4 100644 --- a/tools/gates.count +++ b/tools/gates.count @@ -13,6 +13,6 @@ synthetic_keys_in_mouse_handler = 3 # target 0 (the content-table re-entries wildcard_arms_in_action_tree = 0 # target 0, must stay 0 view_writes_outside_tui = 12 # target 0 (producers outside tui/ and core/app/ writing App::view) pub_fields_on_app = 139 # target 1 (App.view stays public for the frontend; the rest go through App methods) -direct_playback_context_reads = 92 # target 0 (readers of App::current_playback_context outside the ownership resolver and the snapshot builder, which are excluded) +direct_playback_context_reads = 91 # target 0 (readers of App::current_playback_context outside the ownership resolver and the snapshot builder, which are excluded) action_refs_in_tui_handlers = 183 # adoption: may only rise -test_attribute_total = 1838 # adoption: may only rise +test_attribute_total = 1842 # adoption: may only rise From 5f9e5414311af604759b09a3d8eea791aefb2186 Mon Sep 17 00:00:00 2001 From: LargeModGames <84450916+LargeModGames@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:58:35 +0200 Subject: [PATCH 2/4] fix(network): drop out-of-range party seeks, guard native restore --- src/infra/network/mod.rs | 51 ++++++++++++++++++++++++++++++++--- src/infra/network/playback.rs | 4 +++ tools/gates.count | 2 +- 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/src/infra/network/mod.rs b/src/infra/network/mod.rs index d9caf8c7..36dd5af7 100644 --- a/src/infra/network/mod.rs +++ b/src/infra/network/mod.rs @@ -1684,7 +1684,7 @@ impl Network { 0 }; let compensated_position = if is_playing { - position_ms + transit_ms + position_ms.saturating_add(transit_ms) } else { position_ms }; @@ -1730,7 +1730,9 @@ impl Network { let drift = current_progress.abs_diff(compensated_position); if drift > 3000 && !switched_track { - app.dispatch(IoEvent::Seek(compensated_position as u32)); + if let Ok(position_ms) = u32::try_from(compensated_position) { + app.dispatch(IoEvent::Seek(position_ms)); + } } } @@ -1752,7 +1754,11 @@ impl Network { sync::PlaybackAction::Pause => app.dispatch(IoEvent::PausePlayback), sync::PlaybackAction::NextTrack => app.dispatch(IoEvent::NextTrack), sync::PlaybackAction::PrevTrack => app.dispatch(IoEvent::PreviousTrack), - sync::PlaybackAction::Seek { position_ms } => app.dispatch(IoEvent::Seek(position_ms as u32)), + sync::PlaybackAction::Seek { position_ms } => { + if let Ok(position_ms) = u32::try_from(position_ms) { + app.dispatch(IoEvent::Seek(position_ms)); + } + } sync::PlaybackAction::PlayTrack { uri } => { if ids::playable_id(&uri).is_some() { app.start_playback_uris(vec![uri], None); @@ -2178,6 +2184,45 @@ mod tests { assert!(rx.try_recv().is_err()); } + #[tokio::test] + async fn an_oversized_party_seek_is_dropped() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Host).await; + + relay( + &mut network, + sync::SyncMessage::PlaybackCommand { + action: sync::PlaybackAction::Seek { + position_ms: u64::MAX, + }, + from: None, + }, + ) + .await; + + assert!(matches!(rx.try_recv(), Ok(IoEvent::SyncPlayback))); + assert!(rx.try_recv().is_err()); + + app.lock().await.party_session.as_mut().unwrap().role = sync::PartyRole::Guest; + let mut state = host_state(); + if let sync::SyncMessage::SyncState { + position_ms, + timestamp, + .. + } = &mut state + { + *position_ms = u64::MAX; + *timestamp = 0; + } + relay(&mut network, state).await; + + assert!(matches!( + rx.try_recv(), + Ok(IoEvent::StartPlayback(None, Some(_), None)) + )); + assert!(rx.try_recv().is_err()); + } + #[tokio::test] async fn a_host_relays_a_guest_command_then_broadcasts() { let (app, rx) = app_with_a_session(); diff --git a/src/infra/network/playback.rs b/src/infra/network/playback.rs index d6bd6965..3e56eb81 100644 --- a/src/infra/network/playback.rs +++ b/src/infra/network/playback.rs @@ -1915,6 +1915,10 @@ impl PlaybackNetwork for Network { #[cfg(feature = "streaming")] async fn restore_native_playback(&mut self, generation: u64) { + if decoded_source_owns_playback(self).await { + warn!("native restore {generation} skipped: a decoded source owns playback"); + return; + } let (player, snapshot) = { let mut app = self.app.lock().await; if app.pending_start_playback.is_some() { diff --git a/tools/gates.count b/tools/gates.count index 427621b4..f8e02af5 100644 --- a/tools/gates.count +++ b/tools/gates.count @@ -15,4 +15,4 @@ view_writes_outside_tui = 12 # target 0 (producers outside tui/ and co pub_fields_on_app = 139 # target 1 (App.view stays public for the frontend; the rest go through App methods) direct_playback_context_reads = 91 # target 0 (readers of App::current_playback_context outside the ownership resolver and the snapshot builder, which are excluded) action_refs_in_tui_handlers = 183 # adoption: may only rise -test_attribute_total = 1842 # adoption: may only rise +test_attribute_total = 1843 # adoption: may only rise From 13a609b7ee754ef66c5cabd36f5ac5b2353c437b Mon Sep 17 00:00:00 2001 From: LargeModGames <84450916+LargeModGames@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:09:59 +0200 Subject: [PATCH 3/4] fix(network): coalesce relay states, canonical track uris --- src/infra/network/mod.rs | 74 +++++++++++++++++++++++++++++++++------- tools/gates.count | 2 +- 2 files changed, 62 insertions(+), 14 deletions(-) diff --git a/src/infra/network/mod.rs b/src/infra/network/mod.rs index 36dd5af7..74cbff60 100644 --- a/src/infra/network/mod.rs +++ b/src/infra/network/mod.rs @@ -1583,6 +1583,7 @@ impl Network { } }; + let mut latest_state = None; for msg in messages { match msg { sync::SyncMessage::RoomCreated { code, .. } => { @@ -1624,16 +1625,9 @@ impl Network { }; } } - sync::SyncMessage::SyncState { - track_uri, - position_ms, - is_playing, - timestamp, - } => { - self - .handle_incoming_sync_state(track_uri, position_ms, is_playing, timestamp) - .await; - } + // Only the newest host state in a drain counts: each earlier one + // would start the track again before the pump ran the first start. + state @ sync::SyncMessage::SyncState { .. } => latest_state = Some(state), sync::SyncMessage::PlaybackCommand { action, .. } => { self.handle_incoming_playback_command(action).await; } @@ -1655,6 +1649,17 @@ impl Network { _ => {} } } + if let Some(sync::SyncMessage::SyncState { + track_uri, + position_ms, + is_playing, + timestamp, + }) = latest_state + { + self + .handle_incoming_sync_state(track_uri, position_ms, is_playing, timestamp) + .await; + } } async fn handle_incoming_sync_state( @@ -1664,9 +1669,11 @@ impl Network { is_playing: bool, timestamp: u64, ) { - if ids::playable_id(&track_uri).is_none() { + // The canonical URI: a bare id would compare unequal to the current + // track's URI and restart it on every state. + let Some(track_uri) = ids::playable_id(&track_uri).map(|id| id.uri()) else { return; - } + }; let mut app = self.app.lock().await; let follows_host = matches!( &app.party_session, @@ -1760,7 +1767,7 @@ impl Network { } } sync::PlaybackAction::PlayTrack { uri } => { - if ids::playable_id(&uri).is_some() { + if let Some(uri) = ids::playable_id(&uri).map(|id| id.uri()) { app.start_playback_uris(vec![uri], None); } } @@ -2170,6 +2177,47 @@ mod tests { assert!(rx.try_recv().is_err()); } + #[tokio::test] + async fn a_bare_track_id_from_the_host_starts_the_full_uri() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Guest).await; + + let mut state = host_state(); + if let sync::SyncMessage::SyncState { track_uri, .. } = &mut state { + *track_uri = "0000000000000000000001".to_string(); + } + relay(&mut network, state).await; + + assert!(matches!( + rx.try_recv(), + Ok(IoEvent::StartPlayback(None, Some(uris), None)) if uris == [HOST_TRACK] + )); + assert!(rx.try_recv().is_err()); + } + + #[tokio::test] + async fn two_host_states_in_one_drain_start_the_track_once() { + let (app, rx) = app_with_a_session(); + let mut network = party_network(&app, sync::PartyRole::Guest).await; + + let (tx, incoming) = tokio::sync::mpsc::unbounded_channel(); + tx.send(host_state()).unwrap(); + let mut newest = host_state(); + if let sync::SyncMessage::SyncState { track_uri, .. } = &mut newest { + *track_uri = "spotify:track:0000000000000000000002".to_string(); + } + tx.send(newest).unwrap(); + network.party_incoming_rx = Some(incoming); + network.process_party_messages().await; + + assert!(matches!( + rx.try_recv(), + Ok(IoEvent::StartPlayback(None, Some(uris), None)) + if uris == ["spotify:track:0000000000000000000002"] + )); + assert!(rx.try_recv().is_err()); + } + #[tokio::test] async fn a_guest_ignores_a_host_state_it_cannot_play() { let (app, rx) = app_with_a_session(); diff --git a/tools/gates.count b/tools/gates.count index f8e02af5..3eb22194 100644 --- a/tools/gates.count +++ b/tools/gates.count @@ -15,4 +15,4 @@ view_writes_outside_tui = 12 # target 0 (producers outside tui/ and co pub_fields_on_app = 139 # target 1 (App.view stays public for the frontend; the rest go through App methods) direct_playback_context_reads = 91 # target 0 (readers of App::current_playback_context outside the ownership resolver and the snapshot builder, which are excluded) action_refs_in_tui_handlers = 183 # adoption: may only rise -test_attribute_total = 1843 # adoption: may only rise +test_attribute_total = 1845 # adoption: may only rise From 5b6af9f6cc3f745907c1645ab9f1ca8ef7689067 Mon Sep 17 00:00:00 2001 From: LargeModGames <84450916+LargeModGames@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:56:14 +0200 Subject: [PATCH 4/4] fix(network): publish the canonical party track uri --- src/infra/network/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/infra/network/mod.rs b/src/infra/network/mod.rs index 74cbff60..d5d9384e 100644 --- a/src/infra/network/mod.rs +++ b/src/infra/network/mod.rs @@ -1507,7 +1507,7 @@ impl Network { }; let Some(track_uri) = snapshot .item_uri - .filter(|uri| ids::playable_id(uri).is_some()) + .and_then(|uri| ids::playable_id(&uri).map(|id| id.uri())) else { return; };