Furumi/furumi-tui
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

Integrate shared playback coordination and release 0.3.1

Browse Source
This commit is contained in:
ab
2026-09-10 17:07:41 +03:00
parent 5a193d839f
commit 532d6ee1e9
11 changed files with 473 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+5 -5
View File
@@ -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",
Cargo.toml
+3 -3
View File
@@ -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"] }
README.md
+25
View File
@@ -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
src/app/event.rs
+5 -1
View File
@@ -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.
src/app/mod.rs
+95 -140
View File
@@ -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<i64> {
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"
src/app/state.rs
+1 -7
View File
@@ -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<String, crate::devices::PlaybackSnapshot>,
pub last_remote_snapshot: Option<crate::devices::PlaybackSnapshot>,
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)
src/config/settings.rs
+3
View File
@@ -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(),
src/devices.rs
+169 -39
View File
@@ -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<PlaybackStateWire>;
pub type PlaybackCommand = music_dht::playback::Command<PlaybackStateWire>;
#[derive(Debug, Clone, Default)]
struct PlaybackShared {
engine: Option<Engine>,
engine_group: String,
local: Option<PlaybackSnapshot>,
remote: BTreeMap<String, PlaybackSnapshot>,
}
@@ -369,6 +351,8 @@ pub enum SyncOpPayload {
PlaybackCommand {
target_device_id: String,
command: PlaybackCommand,
#[serde(default)]
authority: Option<CommandStamp>,
},
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<R>(&self, f: impl FnOnce(&mut Engine) -> R) -> Result<R> {
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::<Checkpoint>(&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::<PlaybackConfig>(&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<Option<String>> {
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<Vec<u8>> {
#[cfg(test)]
#[path = "devices/tests.rs"]
mod tests;
fn playback_clock() -> u64 {
static START: std::sync::OnceLock<std::time::Instant> = std::sync::OnceLock::new();
START
.get_or_init(std::time::Instant::now)
.elapsed()
.as_millis() as u64
}
src/devices/tests.rs
+135 -6
View File
@@ -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();
src/jam.rs
+1
View File
@@ -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,
src/player/mod.rs
+31 -3
View File
@@ -90,9 +90,18 @@ impl Shared {
pub struct Controller {
tx: Sender<Command>,
pub shared: Arc<Shared>,
playback_allowed: Arc<AtomicBool>,
}
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<u64>, 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<Command>, shared: Arc<Shared>, on_event: impl Fn(PlayerEvent)) {
fn run(
rx: Receiver<Command>,
shared: Arc<Shared>,
playback_allowed: Arc<AtomicBool>,
on_event: impl Fn(PlayerEvent),
) {
let mut output: Option<Output> = None;
let mut track_loaded = false;
let mut last_len = 0usize;
@@ -163,6 +183,14 @@ fn run(rx: Receiver<Command>, shared: Arc<Shared>, 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);