From 966cf4b437f7529208e9525b3d5fde13fcdd8d14 Mon Sep 17 00:00:00 2001 From: AB Date: Thu, 10 Sep 2026 18:05:00 +0300 Subject: [PATCH] Fix connected devices --- ARCHITECTURE.md | 7 + README.md | 24 ++++ scripts/test_device_interop.py | 65 +++++++++ src/devices.rs | 68 ++++++++- src/devices/interop_tests.rs | 133 +++++++++++++++++ src/devices/tests.rs | 251 +++++++++++++++++++++++++++++++++ 6 files changed, 542 insertions(+), 6 deletions(-) create mode 100644 scripts/test_device_interop.py create mode 100644 src/devices/interop_tests.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 94a672e..66c1019 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -190,6 +190,13 @@ must propagate and converge, not an ephemeral server-side session flag. ## Playback across devices +A persisted device identity has exactly one live coordinator. The TUI holds +an OS file lock beside its device-sync database for the entire runtime; two +installations using the same user data directory cannot open it concurrently. +Network polling is independent per trusted peer, with one in-flight exchange +per peer and a bounded deadline. A stalled peer must not delay the next poll +of a healthy output. Poll tasks are cancelled when federation shuts down. + Playback has one logical state but remains physically local to the device producing audio. diff --git a/README.md b/README.md index 5b424e2..c47641a 100644 --- a/README.md +++ b/README.md @@ -203,6 +203,15 @@ 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. +Only one TUI process may use a device identity. Installed and locally built +binaries use the same user data directory: close the installed player before +starting a development build. An OS file lock prevents duplicate coordinators +and is released automatically on exit or a crash. Older builds do not take +this lock, so close those explicitly when upgrading. + +Connected-device polling runs independently per peer, so an offline or stalled +device does not postpone updates to live players. + The existing device menu handles manual transfers. Advanced policy can be set in `settings.toml` without changing the UI: @@ -220,6 +229,21 @@ 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. +Run the local protocol checks from the TUI repository: + +```bash +cargo test localhost_devices_exchange_state_and_handoff +python scripts/test_device_interop.py ../furumusic +``` + +The first test uses real iroh streams, isolated SQLite databases, ownership +handoff, and a stalled peer. The second builds both player test binaries and +exchanges their actual JSON messages over `127.0.0.1`: queue metadata, +bidirectional handoff, pause/seek and duplicate command fencing. It exercises +the TUI command adapter and web player hub, but does not start a browser or +PostgreSQL and does not cover web database migrations. Both tests use temporary +identities and data; no running player or user library is used. + ## License Furumi is released under the diff --git a/scripts/test_device_interop.py b/scripts/test_device_interop.py new file mode 100644 index 0000000..1c8ec31 --- /dev/null +++ b/scripts/test_device_interop.py @@ -0,0 +1,65 @@ +"""Build and run the real TUI and web protocol adapters against each other. + +Usage: python scripts/test_device_interop.py [path/to/furumusic] +No user databases, accounts, audio outputs, or external servers are used. +""" +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile + + +def test_binary(repo): + result = subprocess.run( + ["cargo", "test", "--locked", "--no-run", "--message-format=json"], + cwd=repo, text=True, encoding="utf-8", stdout=subprocess.PIPE, + ) + result.check_returncode() + binaries = [] + for line in result.stdout.splitlines(): + event = json.loads(line) + if event.get("reason") == "compiler-artifact" and event.get("profile", {}).get("test") and event.get("executable"): + binaries.append(event["executable"]) + if len(binaries) != 1: + raise RuntimeError(f"Expected one player test binary in {repo}, found {binaries}") + return binaries[0] + + +def main(): + tui = Path(__file__).resolve().parents[1] + web = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else tui.parent / "furumusic" + tui_bin, web_bin = test_binary(tui), test_binary(web) + with tempfile.TemporaryDirectory(prefix="furumi-interop-") as directory: + env = dict(os.environ, FURUMI_INTEROP_DIR=directory) + processes = [] + try: + for binary, test in [(web_bin, "federation::devices::interop_tests::localhost_tui_peer"), + (tui_bin, "devices::interop_tests::localhost_web_peer")]: + listing = subprocess.check_output([binary, "--list"], text=True, encoding="utf-8") + if f"{test}: test" not in listing: + raise RuntimeError(f"Required test {test} is missing from {binary}") + processes.append(subprocess.Popen( + [binary, "--ignored", "--exact", test, "--nocapture"], env=env, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, encoding="utf-8", + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + )) + codes = [] + for process in processes: + output, _ = process.communicate(timeout=45) + print(output, end="") + codes.append(process.returncode) + if any(codes): + raise RuntimeError(f"Cross-player test failed: exit codes {codes}") + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.wait() + print("PASS: TUI <-> WEB state, bidirectional handoff, pause/seek, duplicate fencing") + + +if __name__ == "__main__": + main() diff --git a/src/devices.rs b/src/devices.rs index 980d9aa..60259c1 100644 --- a/src/devices.rs +++ b/src/devices.rs @@ -82,6 +82,10 @@ pub struct DeviceSyncStatus { #[derive(Clone)] pub struct DeviceSync { + // A device identity has exactly one coordinator, including across binaries + // launched from different installation directories. The OS releases this + // lock on crashes; the file itself is deliberately never removed. + _identity_lock: Option>, conn: Arc>, library: Arc, event_tx: Arc>>>, @@ -599,16 +603,33 @@ fn default_db_path() -> PathBuf { .unwrap_or_else(|| PathBuf::from("devices").join("sync.sqlite3")) } +fn acquire_identity_lock(database: &std::path::Path) -> Result { + let path = database.with_extension("lock"); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .with_context(|| format!("opening device identity lock {}", path.display()))?; + file.try_lock().with_context(|| format!( + "Cannot acquire device identity lock {}. Another Furumi instance may already be using this device. Close it before launching another binary; two players must not share one device identity.", path.display() + ))?; + Ok(file) +} + impl DeviceSync { pub fn new(library: Arc) -> Result> { let path = default_db_path(); if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } + let identity_lock = acquire_identity_lock(&path)?; let conn = Connection::open(&path).with_context(|| format!("opening {}", path.display()))?; init_schema(&conn)?; let sync = Arc::new(Self { + _identity_lock: Some(Arc::new(identity_lock)), conn: Arc::new(std::sync::Mutex::new(conn)), library, event_tx: Arc::new(std::sync::Mutex::new(None)), @@ -2829,13 +2850,45 @@ pub async fn sync_loop( transport_stats: Arc, ) { let mut interval = tokio::time::interval(SYNC_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Independent polls: an offline device must not hold up live outputs. + // JoinSet aborts outstanding IO when federation stops or restarts. + let mut polls = tokio::task::JoinSet::new(); + let mut active = std::collections::HashMap::new(); loop { - interval.tick().await; - if let Err(err) = sync - .sync_once(Arc::clone(&service), Arc::clone(&transport_stats)) - .await - { - tracing::debug!("personal sync tick failed: {err:#}"); + tokio::select! { + _ = interval.tick() => { + match sync.active_remote_devices() { + Ok(devices) => for device in devices { + if device.endpoint_ticket.trim().is_empty() + || active.values().any(|id| id == &device.device_id) { continue; } + let device_id = device.device_id.clone(); + let sync = Arc::clone(&sync); + let service = Arc::clone(&service); + let stats = Arc::clone(&transport_stats); + let handle = polls.spawn(async move { + tokio::time::timeout(Duration::from_secs(30), sync.sync_device(service, &device, stats)) + .await.context("device sync exchange timed out")? + }); + active.insert(handle.id(), device_id); + }, + Err(err) => { let _ = sync.set_last_error(Some(format!("{err:#}"))); } + } + if let Err(err) = sync.gc_tombstones() { + tracing::debug!("personal sync cleanup failed: {err:#}"); + } + } + Some(completed) = polls.join_next_with_id(), if !polls.is_empty() => { + let (task, result) = match completed { + Ok((task, result)) => (task, result), + Err(error) => (error.id(), Err(anyhow::Error::from(error))), + }; + if let Some(device_id) = active.remove(&task) + && let Err(err) = result { + tracing::debug!(device = %device_id, "device sync failed: {err:#}"); + let _ = sync.set_last_error(Some(format!("{}: {err:#}", short_id(&device_id)))); + } + } } if let Some(tx) = lock(&sync.event_tx).as_ref() { let _ = tx.send(AppEvent::DeviceSyncStatus(sync.status())); @@ -3758,6 +3811,9 @@ fn base64url_decode(value: &str) -> Result> { #[path = "devices/tests.rs"] mod tests; +#[cfg(test)] +mod interop_tests; + fn playback_clock() -> u64 { static START: std::sync::OnceLock = std::sync::OnceLock::new(); START diff --git a/src/devices/interop_tests.rs b/src/devices/interop_tests.rs new file mode 100644 index 0000000..4c7c802 --- /dev/null +++ b/src/devices/interop_tests.rs @@ -0,0 +1,133 @@ +//! Cross-binary wire test. Run with scripts/test_device_interop.py; the other +//! endpoint is compiled from furumusic's real protocol types and player hub. +use super::*; +use tokio::io::AsyncWriteExt; + +#[tokio::test] +#[ignore = "run scripts/test_device_interop.py to start both player test binaries"] +async fn localhost_web_peer() { + tokio::time::timeout(Duration::from_secs(30), async { + let dir = PathBuf::from(std::env::var("FURUMI_INTEROP_DIR").expect("interop runner")); + let address = loop { + if let Ok(address) = std::fs::read_to_string(dir.join("web-address")) { + break address; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + let mut stream = tokio::net::TcpStream::connect(address.trim()) + .await + .unwrap(); + let conn = Connection::open_in_memory().unwrap(); + init_schema(&conn).unwrap(); + let sync = DeviceSync { + _identity_lock: None, + conn: Arc::new(std::sync::Mutex::new(conn)), + library: Arc::new(Library::open(&dir.join("interop-library.sqlite3")).unwrap()), + event_tx: Default::default(), + playback: Default::default(), + }; + let id = sync.ensure_identity().unwrap(); + let (tx, mut events) = tokio::sync::mpsc::unbounded_channel(); + sync.set_event_tx(tx); + sync.claim_playback(&id.device_id).unwrap(); + sync.playback_tick(true, true).unwrap(); + let mut state: PlaybackStateWire = serde_json::from_value(serde_json::json!({ + "queue": [{ "id": 7, "title": "Interop track", "duration_seconds": 123.5, + "release_id": 1, "release_title": "Interop album", "artist_names": ["Interop artist"], + "content_id": "b3:0000000000000000000000000000000000000000000000000000000000000000" }], + "queue_pos": 0, "playing": true, "paused": false, + "position_secs": 42.5, "volume": 73, "shuffle": true, "repeat": "all" + })) + .unwrap(); + for phase in 0..3 { + sync.publish_playback(PlaybackSnapshot { + device_id: id.device_id.clone(), + device_name: "TUI interop".into(), + active: true, + updated_at_ms: now_ms(), + state: state.clone(), + coordination: None, + }); + let hello = WireMessage::Hello { + group_id: id.group_id.clone(), + profile: sync.own_profile("").unwrap(), + devices: vec![], + vector: BTreeMap::new(), + ops: vec![], + snapshot: SyncSnapshot::default(), + playback: sync.local_playback_snapshot(), + }; + let mut bytes = serde_json::to_vec(&hello).unwrap(); + bytes.push(b'\n'); + stream.write_all(&bytes).await.unwrap(); + let response: WireMessage = + serde_json::from_slice(&read_line(&mut stream).await.unwrap()).unwrap(); + let WireMessage::SyncResponse { + accepted: true, + playback: Some(snapshot), + ops, + .. + } = response + else { + panic!("expected web response") + }; + let web_id = snapshot.device_id.clone(); + sync.apply_playback_snapshot(&web_id, snapshot).unwrap(); + assert_eq!(ops.len(), 1); + let op = &ops[0]; + // Production command adapter including durable fencing/deduplication. + sync.apply_op(op).unwrap(); + sync.apply_op(op).unwrap(); + let mut commands = Vec::new(); + while let Ok(event) = events.try_recv() { + if let AppEvent::PlaybackCommand { + command, + authority, + origin, + } = event + { + assert!(sync.playback_command_is_current(&origin, &authority)); + commands.push(command); + } + } + assert_eq!(commands.len(), 1, "one event even after duplicate delivery"); + match commands.pop().unwrap() { + PlaybackCommand::ActiveChanged { + active_device_id, + state: next, + .. + } => { + assert_eq!( + active_device_id, + if phase == 0 { + web_id.clone() + } else { + id.device_id.clone() + } + ); + assert_eq!(next.position_secs, 42.5); + assert_eq!(next.queue, state.queue); + state = next; + } + PlaybackCommand::SetState { state: next, seek } => { + assert_eq!(phase, 2); + assert!(seek); + assert!(next.paused); + assert_eq!(next.position_secs, 87.0); + state = next; + } + } + assert_eq!( + sync.playback_tick(true, false).unwrap(), + Some(if phase == 0 { + web_id + } else { + id.device_id.clone() + }) + ); + } + stream.write_all(b"ok\n").await.unwrap(); + }) + .await + .expect("web/TUI exchange timed out"); +} diff --git a/src/devices/tests.rs b/src/devices/tests.rs index 27e6aaf..a47a0cc 100644 --- a/src/devices/tests.rs +++ b/src/devices/tests.rs @@ -1,5 +1,223 @@ use super::*; +/// Exercises the production stream handlers and SQLite adapter, not just the +/// ownership reducer. Each peer has a fresh identity and an isolated library. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn localhost_devices_exchange_state_and_handoff() { + tokio::time::timeout(Duration::from_secs(30), async { + let dirs = [tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()]; + let network = NetworkId::from_name(&format!("device-test-{}", random_hex(16))); + let mut peers = Vec::new(); + let mut servers = Vec::new(); + let mut receivers = Vec::new(); + for dir in &dirs { + std::fs::create_dir_all(dir.path().join("network")).unwrap(); + let config = music_dht::MusicDhtConfig::builder() + .data_dir(dir.path().join("network")) + .network_id(network) + .stream_protocol(SYNC_ALPN) + .build() + .unwrap(); + let (service, events) = MusicDhtService::start(config).await.unwrap(); + let service = Arc::new(service); + let conn = Connection::open_in_memory().unwrap(); + init_schema(&conn).unwrap(); + let sync = Arc::new(DeviceSync { + _identity_lock: Some(Arc::new( + acquire_identity_lock(&dir.path().join("sync.sqlite3")).unwrap(), + )), + conn: Arc::new(std::sync::Mutex::new(conn)), + library: Arc::new(Library::open(&dir.path().join("library.sqlite3")).unwrap()), + event_tx: Default::default(), + playback: Default::default(), + }); + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + sync.set_event_tx(tx); + let stats = Arc::new(crate::federation::TransportStats::default()); + servers.push(tokio::spawn(serve_peers( + service.stream_acceptor(SYNC_ALPN).unwrap(), + sync.clone(), + service.clone(), + stats.clone(), + ))); + receivers.push(rx); + peers.push((sync, service, stats, events)); + } + let (a, service_a, stats_a, _) = &peers[0]; + let (b, service_b, stats_b, _) = &peers[1]; + b.ensure_identity().unwrap(); + b.set_group_id(&a.ensure_identity().unwrap().group_id) + .unwrap(); + let profile_a = a + .own_profile(&service_a.ticket().await.unwrap().to_string()) + .unwrap(); + let profile_b = b + .own_profile(&service_b.ticket().await.unwrap().to_string()) + .unwrap(); + a.apply_device_profile(&profile_b, true).unwrap(); + b.apply_device_profile(&profile_a, true).unwrap(); + a.claim_playback(&profile_a.device_id).unwrap(); + a.playback_tick(true, true).unwrap(); + let mut state = empty_playback_state(); + state.playing = true; + state.position_secs = 42.5; + a.publish_playback(PlaybackSnapshot { + device_id: profile_a.device_id.clone(), + device_name: "TUI".into(), + active: true, + updated_at_ms: now_ms(), + state: state.clone(), + coordination: None, + }); + a.sync_device( + service_a.clone(), + &a.active_remote_devices().unwrap()[0], + stats_a.clone(), + ) + .await + .unwrap(); + assert_eq!( + b.with_playback_engine(|e| e.owner().map(str::to_owned)) + .unwrap(), + Some(profile_a.device_id.clone()) + ); + assert_eq!(lock(&b.playback).remote[&profile_a.device_id].state, state); + assert!( + b.status() + .devices + .iter() + .any(|d| d.device_id == profile_a.device_id && d.last_seen_ms.is_some()) + ); + + a.record_playback_command( + &profile_b.device_id, + PlaybackCommand::ActiveChanged { + active_device_id: profile_b.device_id.clone(), + active_device_name: "Second player".into(), + state: state.clone(), + }, + ) + .unwrap(); + a.sync_device( + service_a.clone(), + &a.active_remote_devices().unwrap()[0], + stats_a.clone(), + ) + .await + .unwrap(); + let mut transferred = false; + while let Ok(event) = receivers[1].try_recv() { + if let AppEvent::PlaybackCommand { + command: + PlaybackCommand::ActiveChanged { + state: received, .. + }, + authority, + origin, + } = event + { + assert_eq!(received, state); + assert!(b.playback_command_is_current(&origin, &authority)); + transferred = true; + } + } + assert!(transferred, "handoff must reach the player's event loop"); + assert_eq!( + b.playback_tick(true, true).unwrap(), + Some(profile_b.device_id.clone()) + ); + b.publish_playback(PlaybackSnapshot { + device_id: profile_b.device_id.clone(), + device_name: "Second player".into(), + active: true, + updated_at_ms: now_ms(), + state, + coordination: None, + }); + b.sync_device( + service_b.clone(), + &b.active_remote_devices().unwrap()[0], + stats_b.clone(), + ) + .await + .unwrap(); + assert_eq!( + a.playback_tick(true, false).unwrap(), + Some(profile_b.device_id) + ); + // A paired peer that accepts a connection but never answers must not + // serialize or stop subsequent polls to the responsive peer. + let silent_dir = tempfile::tempdir().unwrap(); + let (silent, _events) = MusicDhtService::start( + music_dht::MusicDhtConfig::builder() + .data_dir(silent_dir.path()) + .network_id(network) + .stream_protocol(SYNC_ALPN) + .build() + .unwrap(), + ) + .await + .unwrap(); + let _silent_acceptor = silent.stream_acceptor(SYNC_ALPN).unwrap(); + let mut silent_profile = profile_a.clone(); + silent_profile.device_id = "silent-peer".into(); + silent_profile.endpoint_id = silent.endpoint_id().to_string(); + silent_profile.endpoint_ticket = silent.ticket().await.unwrap().to_string(); + a.apply_device_profile(&silent_profile, true).unwrap(); + lock(&a.conn) + .execute( + "UPDATE sync_devices SET last_seen_ms = ?1 WHERE device_id = 'silent-peer'", + [now_ms() + 1_000], + ) + .unwrap(); + assert_eq!( + a.active_remote_devices().unwrap()[0].device_id, + "silent-peer" + ); + a.claim_playback(&profile_a.device_id).unwrap(); + let poller = tokio::spawn(sync_loop(a.clone(), service_a.clone(), stats_a.clone())); + for position in [99.0, 100.0] { + a.playback_tick(true, true).unwrap(); + let mut next = empty_playback_state(); + next.playing = true; + next.position_secs = position; + a.publish_playback(PlaybackSnapshot { + device_id: profile_a.device_id.clone(), + device_name: "TUI".into(), + active: true, + updated_at_ms: now_ms(), + state: next, + coordination: None, + }); + tokio::time::timeout(Duration::from_secs(6), async { + loop { + if lock(&b.playback) + .remote + .get(&profile_a.device_id) + .is_some_and(|snapshot| snapshot.state.position_secs == position) + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("a stalled peer must not delay live playback polls"); + } + poller.abort(); + let _ = poller.await; + silent.shutdown().await.unwrap(); + for server in servers { + server.abort(); + } + for (_, service, _, _) in &peers { + service.shutdown().await.unwrap(); + } + }) + .await + .expect("localhost sync must complete within 30 seconds"); +} + fn empty_playback_state() -> PlaybackStateWire { PlaybackStateWire { queue: vec![], @@ -119,6 +337,7 @@ fn test_sync() -> DeviceSync { unique )); let sync = DeviceSync { + _identity_lock: None, conn: Arc::new(std::sync::Mutex::new(conn)), library: Arc::new(Library::open(&library_path).unwrap()), event_tx: Arc::new(std::sync::Mutex::new(None)), @@ -128,6 +347,38 @@ fn test_sync() -> DeviceSync { sync } +#[test] +fn device_identity_has_one_coordinator_across_installation_paths() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("sync.sqlite3"); + let first = acquire_identity_lock(&path).unwrap(); + assert!(acquire_identity_lock(&path).is_err()); + let child = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "devices::tests::identity_lock_child_process", + "--nocapture", + ]) + .env("FURUMI_IDENTITY_LOCK_TEST_PATH", &path) + .output() + .unwrap(); + assert!( + child.status.success(), + "{}", + String::from_utf8_lossy(&child.stderr) + ); + drop(first); + // Reopening a leftover lock file after shutdown must succeed. + assert!(acquire_identity_lock(&path).is_ok()); +} + +#[test] +fn identity_lock_child_process() { + if let Some(path) = std::env::var_os("FURUMI_IDENTITY_LOCK_TEST_PATH") { + assert!(acquire_identity_lock(std::path::Path::new(&path)).is_err()); + } +} + fn device_revoked(sync: &DeviceSync, device_id: &str) -> bool { let conn = lock(&sync.conn); conn.query_row(