Fix connected devices

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<Arc<std::fs::File>>,
     conn: Arc<std::sync::Mutex<Connection>>,
     library: Arc<Library>,
     event_tx: Arc<std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<AppEvent>>>>,
@@ -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<std::fs::File> {
+    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<Library>) -> Result<Arc<Self>> {
         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<crate::federation::TransportStats>,
 ) {
     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<Vec<u8>> {
 #[path = "devices/tests.rs"]
 mod tests;
 
+#[cfg(test)]
+mod interop_tests;
+
 fn playback_clock() -> u64 {
     static START: std::sync::OnceLock<std::time::Instant> = std::sync::OnceLock::new();
     START

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");
+}

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(