From 532d6ee1e9a14b73e5c2d939cf6f973ed4152dbd Mon Sep 17 00:00:00 2001 From: AB Date: Thu, 10 Sep 2026 17:07:41 +0300 Subject: [PATCH] Integrate shared playback coordination and release 0.3.1 --- Cargo.lock | 10 +- Cargo.toml | 6 +- README.md | 25 +++++ src/app/event.rs | 6 +- src/app/mod.rs | 235 +++++++++++++++++------------------------ src/app/state.rs | 8 +- src/config/settings.rs | 3 + src/devices.rs | 208 +++++++++++++++++++++++++++++------- src/devices/tests.rs | 141 +++++++++++++++++++++++-- src/jam.rs | 1 + src/player/mod.rs | 34 +++++- 11 files changed, 473 insertions(+), 204 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9bea237..dbd1e0a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1698,9 +1698,9 @@ dependencies = [ [[package]] name = "furumi-library" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6f15ab98c65d89ea18e55ed8852abd2f029f63876d357270c4647d73601d0a2" +checksum = "f78466b6345cb90d7bb37db5d9dd67476ab6c2c3b2202876d24f448c314a9a08" dependencies = [ "anyhow", "blake3", @@ -1715,7 +1715,7 @@ dependencies = [ [[package]] name = "furumi_tui" -version = "0.3.0" +version = "0.3.1" dependencies = [ "anyhow", "base64", @@ -3239,9 +3239,9 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.4.1" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7bfe5cee89fa891b00e84738f947c924c6845fd84fedb4697cb073e3fe63bb81" +checksum = "4c3f75d49d3a742a6777a1c4cc53f397399dfc9ddb70034c57aeb44d75d2ee4f" dependencies = [ "async-trait", "blake3", diff --git a/Cargo.toml b/Cargo.toml index 5439ef9..e5e5668 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumi_tui" -version = "0.3.0" +version = "0.3.1" edition = "2024" rust-version = "1.97" description = "A federated P2P player for personal music libraries" @@ -27,12 +27,12 @@ crokey = "1.4.0" crossterm = { version = "0.29.0", features = ["event-stream"] } directories = "6.0.0" futures-util = "0.3.32" -furumi-library = "0.1.0" +furumi-library = "0.2.0" image = { version = "0.25.10", default-features = false, features = ["jpeg", "png", "webp", "gif", "bmp"] } lofty = "0.22" # P2P federation: library index in a shared DHT + audio streaming between # peers (same protocol as furumi-fd). -music-dht = "0.4.1" +music-dht = "0.5.0" ratatui = "0.30.1" reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls", "stream", "blocking"] } rhai = { version = "1", features = ["sync"] } diff --git a/README.md b/README.md index 99098f9..5b424e2 100644 --- a/README.md +++ b/README.md @@ -195,6 +195,31 @@ cargo check --all-targets cargo test --all-targets ``` +## Playback coordination + +Connected Devices uses the shared `music_dht::playback` engine from frid. +Newly started players take an idle/paused output after discovery and become +controllers when another device is playing. Missing output reports trigger +automatic failover; concurrent claims converge to one owner. The web gateway +uses a passive server profile and reports actual browser activity. + +The existing device menu handles manual transfers. Advanced policy can be set +in `settings.toml` without changing the UI: + +```toml +[playback] +claim_on_startup = true +automatic_failover = true +take_paused_on_startup = true +discovery_ms = 3000 +owner_timeout_ms = 120000 +``` + +All clients must support the new coordination envelope for eventual single +output ownership. frid's `PLAYBACK_PROTOCOL.md` describes the protocol, adapter +contract and rollout. Local Cargo patches are only for development; publish +frid and bump the client dependency versions before releasing these changes. + ## License Furumi is released under the diff --git a/src/app/event.rs b/src/app/event.rs index 88a0c84..feeee90 100644 --- a/src/app/event.rs +++ b/src/app/event.rs @@ -185,7 +185,11 @@ pub enum AppEvent { /// Trusted device playback state, delivered by personal-device sync. DevicePlayback(crate::devices::PlaybackSnapshot), /// Playback command addressed to this device. - PlaybackCommand(crate::devices::PlaybackCommand), + PlaybackCommand { + command: crate::devices::PlaybackCommand, + authority: music_dht::playback::CommandStamp, + origin: String, + }, /// Current lifecycle/status of the federation Jam. JamStatus(crate::jam::JamStatus), /// Host playback snapshot received by a Jam participant. diff --git a/src/app/mod.rs b/src/app/mod.rs index 371f752..42775e2 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -32,7 +32,6 @@ use update::{Effect, update}; const TICK_INTERVAL: Duration = Duration::from_millis(250); const VISUALIZER_TICK_INTERVAL: Duration = Duration::from_millis(50); -const ACTIVE_IDLE_LEASE_MS: i64 = 5 * 60 * 1000; /// Handles shared by background tasks; AppState stays pure UI data. pub struct Runtime { @@ -297,6 +296,8 @@ pub async fn run( } let devices = crate::devices::DeviceSync::new(Arc::clone(&library))?; + devices.configure_playback(settings.playback.clone())?; + state.device_playback.config = settings.playback.clone(); devices.set_event_tx(event_tx.clone()); let jam = crate::jam::JamManager::new(event_tx.clone()); let similarity = crate::similarity::Manager::new( @@ -321,7 +322,6 @@ pub async fn run( state.device_playback.self_device_name = device_name.clone(); state.device_playback.active_device_id = Some(device_id); state.device_playback.active_device_name = Some(device_name); - state.device_playback.startup_takeover_pending = true; } let player_events = event_tx.clone(); let mut runtime = Runtime { @@ -416,6 +416,7 @@ pub async fn run( }, Some(app_event) = event_rx.recv() => handle_app_event(&mut state, &mut runtime, app_event), _ = tick.tick() => { + reconcile_personal_playback(&mut state, &mut runtime); state.advance_spinner(); expire_quit_confirmation(&mut state); sync_player_shared(&mut state, &runtime); @@ -626,6 +627,7 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime state.device_playback.active_device_name = Some(device_name.clone()); } let snapshot = crate::devices::PlaybackSnapshot { + coordination: None, device_id, device_name, active, @@ -639,6 +641,7 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime runtime .jam .publish_host_playback(crate::devices::PlaybackSnapshot { + coordination: None, device_id: state.device_playback.self_device_id.clone(), device_name: state.device_playback.self_device_name.clone(), active: true, @@ -658,36 +661,6 @@ fn update_local_idle_since(state: &mut AppState) { } } -fn active_snapshot_idle_since(snapshot: &crate::devices::PlaybackSnapshot) -> Option { - if snapshot.state.playing && !snapshot.state.paused { - None - } else { - snapshot - .state - .idle_since_ms - .or(Some(snapshot.updated_at_ms)) - } -} - -fn active_idle_lease_expired(snapshot: &crate::devices::PlaybackSnapshot, now: i64) -> bool { - active_snapshot_idle_since(snapshot) - .is_some_and(|idle_since| now.saturating_sub(idle_since) >= ACTIVE_IDLE_LEASE_MS) -} - -fn local_active_lease_protected(state: &mut AppState, now: i64) -> bool { - if !state.device_playback.is_audio_owner() || !state.player.playing { - return false; - } - if !state.player.paused { - return true; - } - update_local_idle_since(state); - state - .device_playback - .local_idle_since_ms - .is_some_and(|idle_since| now.saturating_sub(idle_since) < ACTIVE_IDLE_LEASE_MS) -} - fn extrapolate_control_position(state: &mut AppState) { let Some(snapshot) = state.device_playback.last_remote_snapshot.as_ref() else { return; @@ -719,6 +692,7 @@ pub(crate) fn become_control_device( runtime: &Runtime, snapshot: crate::devices::PlaybackSnapshot, ) { + runtime.player.set_playback_allowed(false); if state.device_playback.is_audio_owner() { runtime.player.stop(); publish_inactive_playback_snapshot(state, runtime); @@ -737,6 +711,7 @@ pub(crate) fn become_control_device( } pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, start_audio: bool) { + runtime.player.set_playback_allowed(true); let was_control = state.device_playback.role == state::DevicePlaybackRole::Control; state.device_playback.role = state::DevicePlaybackRole::Active; state.device_playback.jam_host = false; @@ -761,6 +736,13 @@ pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, } pub(crate) fn transfer_active_to_this_device(state: &mut AppState, runtime: &mut Runtime) { + if let Err(error) = runtime + .devices + .claim_playback(&state.device_playback.self_device_id) + { + state.status_message = Some(format!("device handoff: {error}")); + return; + } if state.device_playback.is_audio_owner() { publish_playback_snapshot(state, runtime); request_urgent_device_sync(runtime); @@ -800,6 +782,10 @@ pub(crate) fn transfer_active_to_remote_device( transfer_active_to_this_device(state, runtime); return; } + if let Err(error) = runtime.devices.claim_playback(&target_device_id) { + state.status_message = Some(format!("device handoff: {error}")); + return; + } extrapolate_control_position(state); let previous_active_id = state.device_playback.active_device_id.clone(); if state.player.current.is_none() && !state.player.queue.is_empty() { @@ -824,6 +810,7 @@ pub(crate) fn transfer_active_to_remote_device( record_playback_command_async(runtime, previous, command.clone(), "device handoff"); } let snapshot = crate::devices::PlaybackSnapshot { + coordination: None, device_id: target_device_id.clone(), device_name: target_device_name.clone(), active: true, @@ -2877,6 +2864,7 @@ fn refresh_artists(state: &mut AppState, runtime: &Runtime) { fn save_app_settings(state: &AppState) { let settings = crate::config::settings::AppSettings { + playback: state.device_playback.config.clone(), volume: state.player.volume, library: state.global.filters, music_dir: state.music_dir.clone(), @@ -2983,81 +2971,67 @@ fn apply_queue_refresh( } } +fn reconcile_personal_playback(state: &mut AppState, runtime: &mut Runtime) { + if state.device_playback.role == state::DevicePlaybackRole::Jam { + runtime + .player + .set_playback_allowed(state.device_playback.is_audio_owner()); + let _ = runtime.devices.playback_tick(false, false); + return; + } + let playing = + state.device_playback.is_audio_owner() && state.player.playing && !state.player.paused; + let owner = match runtime.devices.playback_tick(true, playing) { + Ok(owner) => owner, + Err(error) => { + runtime.player.set_playback_allowed(false); + runtime.player.stop(); + state.status_message = Some(format!("playback coordination: {error}")); + return; + } + }; + let Some(owner) = owner else { + runtime.player.set_playback_allowed(false); + runtime.player.stop(); + runtime.player_start_pending = false; + state.device_playback.role = state::DevicePlaybackRole::Control; + state.device_playback.active_device_id = None; + return; + }; + if owner == state.device_playback.self_device_id { + runtime.player.set_playback_allowed(true); + if !state.device_playback.is_audio_owner() { + become_active_device(state, runtime, true); + } + } else if let Some(snapshot) = state.device_playback.remote.get(&owner).cloned() { + if state.device_playback.last_remote_snapshot.as_ref() != Some(&snapshot) + || state.device_playback.is_audio_owner() + { + become_control_device(state, runtime, snapshot); + } + } else { + runtime.player.set_playback_allowed(false); + runtime.player.stop(); + runtime.player_start_pending = false; + state.device_playback.role = state::DevicePlaybackRole::Control; + state.device_playback.active_device_id = Some(owner); + } + publish_playback_snapshot_with_active(state, runtime, state.device_playback.is_audio_owner()); +} + fn handle_device_playback_snapshot( state: &mut AppState, runtime: &mut Runtime, snapshot: crate::devices::PlaybackSnapshot, ) { - // Personal-device reconciliation must never change Jam ownership. Jam - // has its own authority and lifecycle even when the same TUI also belongs - // to a trusted-device group. - if state.device_playback.role == state::DevicePlaybackRole::Jam { - return; - } if snapshot.device_id == state.device_playback.self_device_id { return; } state .device_playback .remote - .insert(snapshot.device_id.clone(), snapshot.clone()); - let now = unix_time_ms(); - state.device_playback.online_devices = state - .device_playback - .remote - .values() - .filter(|snapshot| { - now.saturating_sub(snapshot.updated_at_ms) <= state::DEVICE_ONLINE_TTL_MS - }) - .count() - + 1; - - if !snapshot.active { - return; - } - // Starting a player is an explicit claim of the active role. Import the - // current queue/position from the previously active peer, then announce a - // normal handoff so that the old owner becomes a control device. This is - // intentionally one-shot: subsequent snapshots use the regular lease and - // explicit-transfer rules. - if state.device_playback.startup_takeover_pending { - state.device_playback.startup_takeover_pending = false; - become_control_device(state, runtime, snapshot); - transfer_active_to_this_device(state, runtime); - state.status_message = Some("playback moved to this newly started player".to_string()); - return; - } - if local_active_lease_protected(state, now) { - tracing::debug!( - remote = %snapshot.device_id, - "ignored remote active snapshot while local active playback is protected" - ); - return; - } - let lease_expired = active_idle_lease_expired(&snapshot, now); - let already_controls_this_device = state.device_playback.is_personal_control() - && state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str()); - if !lease_expired || already_controls_this_device { - let was_active = state.device_playback.is_audio_owner(); - let was_paused = state.player.playing && state.player.paused; - become_control_device(state, runtime, snapshot.clone()); - if was_active && was_paused { - state.popup = Some(state::Popup::FedText { - title: "Active device moved".to_string(), - text: format!("Playback is now controlled by {}.", snapshot.device_name), - }); - } - return; - } - - if state.device_playback.is_personal_control() { - return; - } - become_active_device(state, runtime, false); - state.status_message = Some(format!( - "active playback moved here; {} was idle for 5m", - snapshot.device_name - )); + .insert(snapshot.device_id.clone(), snapshot); + reconcile_personal_playback(state, runtime); } fn handle_playback_command( @@ -3112,10 +3086,19 @@ fn handle_playback_command( state: wire, } => { if active_device_id == state.device_playback.self_device_id { + handle_playback_command( + state, + runtime, + crate::devices::PlaybackCommand::SetState { + state: wire, + seek: true, + }, + ); return; } let was_active = state.device_playback.is_audio_owner(); let snapshot = crate::devices::PlaybackSnapshot { + coordination: None, device_id: active_device_id, device_name: active_device_name, active: true, @@ -3237,47 +3220,6 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent }) .unwrap_or(1) .max(1); - let active_revoked = state.device_playback.is_personal_control() - && state - .device_playback - .active_device_id - .as_ref() - .is_some_and(|active| { - state - .federation - .devices - .as_ref() - .and_then(|status| { - status - .devices - .iter() - .find(|device| device.device_id == *active) - }) - .is_some_and(|device| device.revoked) - }); - let active_missing = state.device_playback.is_personal_control() - && state - .device_playback - .active_device_id - .as_ref() - .is_some_and(|active| { - active != &state.device_playback.self_device_id - && !state.federation.devices.as_ref().is_some_and(|status| { - status - .devices - .iter() - .any(|device| device.device_id == *active) - }) - }); - if active_revoked || active_missing { - runtime.player.stop(); - state.player.playing = false; - state.player.current = None; - state.player.paused = false; - state.player.position_secs = 0.0; - become_active_device(state, runtime, false); - state.status_message = Some("active playback moved to this device".into()); - } clamp_settings_cursor(state); } AppEvent::FedSyncFinished(message) => { @@ -3318,18 +3260,28 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent AppEvent::DevicePlayback(snapshot) => { handle_device_playback_snapshot(state, runtime, snapshot); } - AppEvent::PlaybackCommand(_) + AppEvent::PlaybackCommand { .. } if state.device_playback.role == state::DevicePlaybackRole::Jam => { tracing::debug!("ignored personal-device playback command while Jam is active"); } - AppEvent::PlaybackCommand(command) => { - handle_playback_command(state, runtime, command); + AppEvent::PlaybackCommand { + command, + authority, + origin, + } => { + if runtime + .devices + .playback_command_is_current(&origin, &authority) + { + handle_playback_command(state, runtime, command); + } } AppEvent::JamStatus(status) => { state.jam = status.clone(); match status.role { crate::jam::JamRole::Host => { + runtime.player.set_playback_allowed(true); state.device_playback.role = state::DevicePlaybackRole::Jam; state.device_playback.jam_host = true; publish_playback_snapshot(state, runtime); @@ -3848,6 +3800,9 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent } AppEvent::Player(event) if state.device_playback.is_control() => { runtime.player_start_pending = false; + // A background decoder may finish after ownership was handed off. + // Ignoring Started alone would leave its obsolete audio playing. + runtime.player.stop(); tracing::debug!( ?event, "ignored local player event while controlling remote playback" diff --git a/src/app/state.rs b/src/app/state.rs index b665f1e..82230b3 100644 --- a/src/app/state.rs +++ b/src/app/state.rs @@ -1541,6 +1541,7 @@ impl DevicePlaybackRole { #[derive(Debug, Default)] pub struct DevicePlaybackState { + pub config: music_dht::playback::Config, pub role: DevicePlaybackRole, pub self_device_id: String, pub self_device_name: String, @@ -1551,9 +1552,6 @@ pub struct DevicePlaybackState { pub remote: BTreeMap, pub last_remote_snapshot: Option, pub jam_host: bool, - /// A freshly started TUI owns playback by protocol. The first active - /// snapshot discovered during startup is imported and handed off here. - pub startup_takeover_pending: bool, } impl DevicePlaybackState { @@ -1562,10 +1560,6 @@ impl DevicePlaybackState { || (self.role == DevicePlaybackRole::Jam && !self.jam_host) } - pub fn is_personal_control(&self) -> bool { - self.role == DevicePlaybackRole::Control - } - pub fn is_audio_owner(&self) -> bool { self.role == DevicePlaybackRole::Active || (self.role == DevicePlaybackRole::Jam && self.jam_host) diff --git a/src/config/settings.rs b/src/config/settings.rs index 380055c..413a85b 100644 --- a/src/config/settings.rs +++ b/src/config/settings.rs @@ -50,6 +50,8 @@ impl Default for SimilaritySettings { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AppSettings { + #[serde(default)] + pub playback: music_dht::playback::Config, #[serde(default = "default_volume")] pub volume: u8, #[serde(default)] @@ -64,6 +66,7 @@ pub struct AppSettings { impl Default for AppSettings { fn default() -> Self { Self { + playback: music_dht::playback::Config::default(), volume: default_volume(), library: LibraryFilters::default(), music_dir: default_music_dir(), diff --git a/src/devices.rs b/src/devices.rs index 19afb8f..980d9aa 100644 --- a/src/devices.rs +++ b/src/devices.rs @@ -5,6 +5,7 @@ //! materialized tables, so offline clients can merge likes, playlists and //! membership changes deterministically. +use music_dht::playback::{Checkpoint, CommandStamp, Config as PlaybackConfig, Engine}; use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; use std::str::FromStr; @@ -24,7 +25,7 @@ use crate::library::models::{ArtistRef, TrackItem}; pub const SYNC_ALPN: &[u8] = b"furumi/sync/2"; const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION"); -pub const PROTOCOL_VERSION: u16 = 2; +pub const PROTOCOL_VERSION: u16 = 3; const INVITE_TTL_MS: i64 = 10 * 60 * 1000; const PAIRING_WAIT_MS: i64 = 5 * 60 * 1000; const PAIRING_RETRY_DELAY: Duration = Duration::from_secs(1); @@ -245,32 +246,13 @@ pub struct PlaybackStateWire { pub repeat: PlaybackRepeat, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct PlaybackSnapshot { - pub device_id: String, - pub device_name: String, - pub active: bool, - pub updated_at_ms: i64, - pub state: PlaybackStateWire, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum PlaybackCommand { - SetState { - state: PlaybackStateWire, - #[serde(default)] - seek: bool, - }, - ActiveChanged { - active_device_id: String, - active_device_name: String, - state: PlaybackStateWire, - }, -} +pub type PlaybackSnapshot = music_dht::playback::Snapshot; +pub type PlaybackCommand = music_dht::playback::Command; #[derive(Debug, Clone, Default)] struct PlaybackShared { + engine: Option, + engine_group: String, local: Option, remote: BTreeMap, } @@ -369,6 +351,8 @@ pub enum SyncOpPayload { PlaybackCommand { target_device_id: String, command: PlaybackCommand, + #[serde(default)] + authority: Option, }, ListenRecorded { event: ListenEvent, @@ -648,10 +632,92 @@ impl DeviceSync { format!("{}-{}", now_ms(), random_hex(12)) } + pub fn configure_playback(&self, config: PlaybackConfig) -> Result<()> { + set_meta( + &lock(&self.conn), + "playback_config_v1", + &serde_json::to_string(&config)?, + )?; + lock(&self.playback).engine = None; + Ok(()) + } + + fn with_playback_engine(&self, f: impl FnOnce(&mut Engine) -> R) -> Result { + let identity = self.ensure_identity()?; + let mut shared = lock(&self.playback); + if shared.engine.is_none() || shared.engine_group != identity.group_id { + shared.engine_group = identity.group_id.clone(); + let conn = lock(&self.conn); + let durable = get_meta(&conn, "playback_coordination_v1")? + .map(|json| serde_json::from_str::(&json)) + .transpose()? + .filter(|checkpoint| checkpoint.scope == identity.group_id) + .map(|checkpoint| checkpoint.state) + .unwrap_or_default(); + let config = get_meta(&conn, "playback_config_v1")? + .map(|json| serde_json::from_str::(&json)) + .transpose()? + .unwrap_or_default(); + shared.engine = Some(Engine::new( + identity.device_id, + config, + durable, + playback_clock(), + )); + } + let engine = shared.engine.as_mut().expect("initialized playback engine"); + let previous = engine.clone(); + let result = f(engine); + if previous.durable() != engine.durable() { + let saved = set_meta( + &lock(&self.conn), + "playback_coordination_v1", + &serde_json::to_string(&Checkpoint { + scope: identity.group_id, + state: engine.durable().clone(), + })?, + ); + if let Err(error) = saved { + *engine = previous; + return Err(error); + } + } + Ok(result) + } + + /// Evaluate shared ownership independently of the UI's online-device list. + pub fn playback_tick(&self, available: bool, playing: bool) -> Result> { + self.with_playback_engine(|engine| { + engine.set_output(available, playing); + engine.heartbeat(playback_clock()); + engine.tick(playback_clock()); + engine.owner().map(str::to_string) + }) + } + + pub fn playback_command_is_current(&self, origin: &str, stamp: &CommandStamp) -> bool { + self.with_playback_engine(|engine| engine.command_is_current(origin, stamp)) + .unwrap_or(false) + } + + pub fn claim_playback(&self, target: &str) -> Result<()> { + self.with_playback_engine(|engine| engine.transfer(target, playback_clock()))? + .then_some(()) + .context("cannot allocate playback ownership term") + } + pub fn publish_playback(&self, mut snapshot: PlaybackSnapshot) { if snapshot.updated_at_ms <= 0 { snapshot.updated_at_ms = now_ms(); } + let Ok(coordination) = self.with_playback_engine(|engine| engine.announcement()) else { + return; + }; + snapshot.active = coordination + .claim + .as_ref() + .is_some_and(|claim| claim.owner == snapshot.device_id); + snapshot.coordination = Some(coordination); lock(&self.playback).local = Some(snapshot); } @@ -663,9 +729,22 @@ impl DeviceSync { if target_device_id.trim().is_empty() { return Ok(()); } + let authority = self + .with_playback_engine(|engine| { + if let PlaybackCommand::ActiveChanged { + active_device_id, .. + } = &command + && engine.owner() != Some(active_device_id.as_str()) + { + engine.transfer(active_device_id, playback_clock()); + } + engine.stamp() + })? + .context("no playback owner; wait for discovery or select an output")?; self.record_local_op(SyncOpPayload::PlaybackCommand { target_device_id: target_device_id.to_string(), command, + authority: Some(authority), }) } @@ -845,7 +924,7 @@ impl DeviceSync { if let Some(profile) = profile { self.apply_device_profile(&profile, true)?; if let Some(playback) = playback { - self.apply_playback_snapshot(playback)?; + self.apply_playback_snapshot(&profile.device_id, playback)?; } } self.apply_device_profiles(&devices)?; @@ -1305,7 +1384,7 @@ impl DeviceSync { } => { self.apply_device_profiles(&devices)?; if let Some(playback) = playback { - self.apply_playback_snapshot(playback)?; + self.apply_playback_snapshot(&device.device_id, playback)?; } self.apply_snapshot(snapshot)?; self.apply_ops(ops)?; @@ -1636,8 +1715,15 @@ impl DeviceSync { SyncOpPayload::PlaybackCommand { target_device_id, command, + authority, } => { - self.apply_playback_command(target_device_id, command, &op.op_id)?; + self.apply_playback_command( + target_device_id, + command, + authority.as_ref(), + &op.origin_device_id, + &op.op_id, + )?; false } SyncOpPayload::ListenRecorded { event } => self @@ -1651,10 +1737,32 @@ impl DeviceSync { &self, target_device_id: &str, command: &PlaybackCommand, + authority: Option<&CommandStamp>, + origin: &str, op_id: &str, ) -> Result<()> { let identity = self.ensure_identity()?; - if target_device_id != identity.device_id { + // Legacy commands still replicate with the library log, but cannot + // override the versioned personal-playback protocol. + let Some(authority) = authority else { + return Ok(()); + }; + let handoff = matches!(command, PlaybackCommand::ActiveChanged { .. }); + if !handoff && target_device_id != identity.device_id { + return Ok(()); + } + if let PlaybackCommand::ActiveChanged { + active_device_id, .. + } = command + && active_device_id != &authority.claim.owner + { + return Ok(()); + } + let accepted = self.with_playback_engine(|engine| { + engine.accept_command(origin, authority, handoff, playback_clock()) + && (handoff || engine.is_owner()) + })?; + if !accepted || target_device_id != identity.device_id { return Ok(()); } let inserted = { @@ -1669,7 +1777,11 @@ impl DeviceSync { return Ok(()); } if let Some(tx) = lock(&self.event_tx).as_ref() { - let _ = tx.send(AppEvent::PlaybackCommand(command.clone())); + let _ = tx.send(AppEvent::PlaybackCommand { + command: command.clone(), + authority: authority.clone(), + origin: origin.to_string(), + }); } Ok(()) } @@ -2595,19 +2707,29 @@ impl DeviceSync { lock(&self.playback).local.clone() } - fn apply_playback_snapshot(&self, snapshot: PlaybackSnapshot) -> Result<()> { + fn apply_playback_snapshot(&self, sender: &str, snapshot: PlaybackSnapshot) -> Result<()> { let identity = self.ensure_identity()?; - if snapshot.device_id == identity.device_id { + if snapshot.device_id != sender || snapshot.device_id == identity.device_id { + return Ok(()); + } + let Some(coordination) = &snapshot.coordination else { + return Ok(()); + }; + if !self.with_playback_engine(|engine| { + engine.observe(&snapshot.device_id, coordination, playback_clock()) + })? { return Ok(()); } let should_send = { - let mut playback = lock(&self.playback); - let changed = playback - .remote - .get(&snapshot.device_id) - .is_none_or(|current| snapshot.updated_at_ms > current.updated_at_ms); + let mut shared = lock(&self.playback); + let changed = shared.remote.get(&snapshot.device_id).is_none_or(|old| { + old.coordination.as_ref().is_none_or(|c| { + coordination.claim > c.claim + || (coordination.claim == c.claim && coordination.heartbeat > c.heartbeat) + }) + }); if changed { - playback + shared .remote .insert(snapshot.device_id.clone(), snapshot.clone()); } @@ -2936,7 +3058,7 @@ async fn handle_pair_request( } sync.apply_device_profile(&profile, true)?; if let Some(playback) = playback { - sync.apply_playback_snapshot(playback)?; + sync.apply_playback_snapshot(&profile.device_id, playback)?; } sync.apply_snapshot(snapshot)?; sync.apply_ops(ops)?; @@ -3026,7 +3148,7 @@ async fn handle_hello( sync.apply_device_profile(&profile, false)?; sync.apply_device_profiles(&devices)?; if let Some(playback) = playback { - sync.apply_playback_snapshot(playback)?; + sync.apply_playback_snapshot(&profile.device_id, playback)?; } sync.apply_snapshot(snapshot)?; sync.apply_ops(ops)?; @@ -3635,3 +3757,11 @@ fn base64url_decode(value: &str) -> Result> { #[cfg(test)] #[path = "devices/tests.rs"] mod tests; + +fn playback_clock() -> u64 { + static START: std::sync::OnceLock = std::sync::OnceLock::new(); + START + .get_or_init(std::time::Instant::now) + .elapsed() + .as_millis() as u64 +} diff --git a/src/devices/tests.rs b/src/devices/tests.rs index be77b8c..27e6aaf 100644 --- a/src/devices/tests.rs +++ b/src/devices/tests.rs @@ -1,5 +1,111 @@ use super::*; +fn empty_playback_state() -> PlaybackStateWire { + PlaybackStateWire { + queue: vec![], + queue_pos: 0, + playing: false, + paused: false, + idle_since_ms: None, + position_secs: 0.0, + volume: 80, + shuffle: false, + repeat: PlaybackRepeat::Off, + } +} + +#[test] +fn coordination_ignores_legacy_commands_and_wrong_snapshot_sender() { + let sync = test_sync(); + let identity = sync.ensure_identity().unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + sync.set_event_tx(tx); + let command = PlaybackCommand::SetState { + state: empty_playback_state(), + seek: false, + }; + sync.apply_playback_command(&identity.device_id, &command, None, "remote", "legacy") + .unwrap(); + assert!(rx.try_recv().is_err()); + let mut remote = Engine::new( + "remote".into(), + PlaybackConfig::default(), + Default::default(), + 0, + ); + remote.transfer("remote", 0); + remote.set_output(true, true); + remote.heartbeat(1); + let snapshot = PlaybackSnapshot { + device_id: "remote".into(), + device_name: "Remote".into(), + active: true, + updated_at_ms: now_ms(), + state: empty_playback_state(), + coordination: Some(remote.announcement()), + }; + sync.apply_playback_snapshot("different-sender", snapshot.clone()) + .unwrap(); + assert!( + sync.with_playback_engine(|engine| engine.owner().is_none()) + .unwrap() + ); + sync.apply_playback_snapshot("remote", snapshot).unwrap(); + assert_eq!( + sync.playback_tick(true, false).unwrap().as_deref(), + Some("remote") + ); + sync.publish_playback(PlaybackSnapshot { + device_id: identity.device_id, + device_name: identity.name, + active: false, + updated_at_ms: now_ms(), + state: empty_playback_state(), + coordination: None, + }); + let gossip = sync + .local_playback_snapshot() + .unwrap() + .coordination + .unwrap(); + assert_eq!(gossip.claim.unwrap().owner, "remote"); +} + +#[test] +fn checkpoint_is_not_reused_after_changing_trusted_group() { + let sync = test_sync(); + sync.claim_playback("old-group-owner").unwrap(); + sync.set_group_id("new-test-group").unwrap(); + assert!( + sync.with_playback_engine(|engine| engine.owner().is_none()) + .unwrap() + ); +} + +#[test] +fn queued_command_loses_its_fence_when_another_owner_wins() { + let sync = test_sync(); + let identity = sync.ensure_identity().unwrap(); + sync.claim_playback(&identity.device_id).unwrap(); + let stamp = sync + .with_playback_engine(|engine| engine.stamp().unwrap()) + .unwrap(); + sync.apply_playback_command( + &identity.device_id, + &PlaybackCommand::SetState { + state: empty_playback_state(), + seek: false, + }, + Some(&stamp), + &identity.device_id, + "queued", + ) + .unwrap(); + assert!(sync.playback_command_is_current(&identity.device_id, &stamp)); + sync.claim_playback("new-owner").unwrap(); + assert!(!sync.playback_command_is_current(&identity.device_id, &stamp)); +} + static NEXT_TEST_DB: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); fn test_sync() -> DeviceSync { @@ -212,19 +318,41 @@ fn playback_command_is_targeted_and_deduplicated() { seek: false, }; - sync.apply_playback_command("dev_other", &command, "op_other") + sync.claim_playback(&identity.device_id).unwrap(); + let stamp = sync + .with_playback_engine(|engine| engine.stamp().unwrap()) .unwrap(); + sync.apply_playback_command( + "dev_other", + &command, + Some(&stamp), + &identity.device_id, + "op_other", + ) + .unwrap(); assert!(rx.try_recv().is_err()); - sync.apply_playback_command(&identity.device_id, &command, "op_1") - .unwrap(); + sync.apply_playback_command( + &identity.device_id, + &command, + Some(&stamp), + &identity.device_id, + "op_1", + ) + .unwrap(); assert!(matches!( rx.try_recv().unwrap(), - crate::app::event::AppEvent::PlaybackCommand(_) + crate::app::event::AppEvent::PlaybackCommand { .. } )); - sync.apply_playback_command(&identity.device_id, &command, "op_1") - .unwrap(); + sync.apply_playback_command( + &identity.device_id, + &command, + Some(&stamp), + &identity.device_id, + "op_1", + ) + .unwrap(); assert!(rx.try_recv().is_err()); } @@ -245,6 +373,7 @@ fn playback_commands_are_caught_up_only_by_their_target_while_fresh() { }, seek: false, }; + sync.claim_playback("dev_target").unwrap(); sync.record_playback_command("dev_target", command.clone()) .unwrap(); sync.record_playback_command("dev_other", command).unwrap(); diff --git a/src/jam.rs b/src/jam.rs index 248c586..b6bff7a 100644 --- a/src/jam.rs +++ b/src/jam.rs @@ -668,6 +668,7 @@ mod tests { repeat: crate::devices::PlaybackRepeat::Off, }; host.publish_host_playback(PlaybackSnapshot { + coordination: None, device_id: "dev_host".into(), device_name: "Host".into(), active: true, diff --git a/src/player/mod.rs b/src/player/mod.rs index c707551..f25d167 100644 --- a/src/player/mod.rs +++ b/src/player/mod.rs @@ -90,9 +90,18 @@ impl Shared { pub struct Controller { tx: Sender, pub shared: Arc, + playback_allowed: Arc, } impl Controller { + /// Revoke the output at the audio thread boundary, including work queued + /// by background decoders before an ownership transfer. + pub fn set_playback_allowed(&self, allowed: bool) { + if self.playback_allowed.swap(allowed, Ordering::AcqRel) && !allowed { + self.stop(); + } + } + pub fn play(&self, reader: TrackReader, byte_len: Option, volume: f32) { let _ = self.tx.send(Command::Play { reader, @@ -142,11 +151,17 @@ pub fn spawn(on_event: impl Fn(PlayerEvent) + Send + 'static) -> Controller { let (tx, rx) = std::sync::mpsc::channel(); let shared = Arc::new(Shared::default()); let thread_shared = Arc::clone(&shared); + let playback_allowed = Arc::new(AtomicBool::new(true)); + let thread_allowed = Arc::clone(&playback_allowed); std::thread::Builder::new() .name("audio".to_string()) - .spawn(move || run(rx, thread_shared, on_event)) + .spawn(move || run(rx, thread_shared, thread_allowed, on_event)) .expect("spawning the audio thread cannot fail"); - Controller { tx, shared } + Controller { + tx, + shared, + playback_allowed, + } } struct Output { @@ -155,7 +170,12 @@ struct Output { player: Player, } -fn run(rx: Receiver, shared: Arc, on_event: impl Fn(PlayerEvent)) { +fn run( + rx: Receiver, + shared: Arc, + playback_allowed: Arc, + on_event: impl Fn(PlayerEvent), +) { let mut output: Option = None; let mut track_loaded = false; let mut last_len = 0usize; @@ -163,6 +183,14 @@ fn run(rx: Receiver, shared: Arc, on_event: impl Fn(PlayerEvent loop { match rx.recv_timeout(Duration::from_millis(100)) { Ok(command) => { + if !playback_allowed.load(Ordering::Acquire) + && matches!( + command, + Command::Play { .. } | Command::Enqueue { .. } | Command::Resume + ) + { + continue; + } // Commands change the source queue legitimately; resync the // length so the next tick doesn't read it as a track ending. handle(command, &shared, &mut output, &mut track_loaded, &on_event);