Fix intermittent MPRIS loss

Author: pcarrier

crates/desktop/src/mpris.rs

@@ -39,6 +39,7 @@ const ARTWORK_BUDGET: usize = 8 * 1024 * 1024;
 const PROPERTY_CALL_LIMIT: usize = 4;
 const ACTION_CALL_LIMIT: usize = 4;
 const ACTION_QUEUE_LIMIT: usize = 64;
+const MONITOR_RETRY_DELAY: Duration = Duration::from_secs(5);
 
 #[derive(Clone)]
 struct Target {
@@ -52,6 +53,11 @@ struct Target {
     position_observed_at: Instant,
 }
 
+struct DeferredTarget {
+    owner: String,
+    retry_at: Option<Instant>,
+}
+
 fn player_at(target: &Target, observed_at: Instant) -> MprisPlayer {
     let mut player = target.player.clone();
     if player.playback_status == PlaybackStatus::Playing {
@@ -77,7 +83,7 @@ pub(super) struct State {
     players: HashMap<u32, Target>,
     by_owner: HashMap<String, u32>,
     by_alias: HashMap<String, u32>,
-    deferred: HashMap<String, String>,
+    deferred: HashMap<String, DeferredTarget>,
     active: Option<u32>,
     artwork_bytes: usize,
     property_calls: Arc<Semaphore>,
@@ -178,19 +184,42 @@ impl State {
         records
     }
 
-    fn deferred_candidate(&self, limit: usize) -> Option<(String, String)> {
+    fn remove_player_for_retry(&mut self, player_id: u32, retry_at: Instant) -> Vec<MprisRecord> {
+        let Some(target) = self.players.get(&player_id) else {
+            return Vec::new();
+        };
+        let owner = target.owner.clone();
+        let aliases = target.aliases.iter().cloned().collect::<Vec<_>>();
+        let records = self.remove_player(player_id);
+        for alias in aliases {
+            self.deferred.insert(
+                alias,
+                DeferredTarget {
+                    owner: owner.clone(),
+                    retry_at: Some(retry_at),
+                },
+            );
+        }
+        records
+    }
+
+    fn deferred_candidate(&self, limit: usize, now: Instant) -> Option<(String, String)> {
+        let ready = |target: &&DeferredTarget| target.retry_at.is_none_or(|at| now >= at);
         self.deferred
             .iter()
-            .filter(|(_, owner)| self.by_owner.contains_key(owner.as_str()))
+            .filter(|(_, target)| {
+                ready(target) && self.by_owner.contains_key(target.owner.as_str())
+            })
             .min_by_key(|(alias, _)| alias.as_str())
-            .map(|(alias, owner)| (alias.clone(), owner.clone()))
+            .map(|(alias, target)| (alias.clone(), target.owner.clone()))
             .or_else(|| {
                 (self.players.len() < limit)
                     .then(|| {
                         self.deferred
                             .iter()
+                            .filter(|(_, target)| ready(target))
                             .min_by_key(|(alias, _)| alias.as_str())
-                            .map(|(alias, owner)| (alias.clone(), owner.clone()))
+                            .map(|(alias, target)| (alias.clone(), target.owner.clone()))
                     })
                     .flatten()
             })
@@ -402,7 +431,13 @@ async fn register_alias(
             return;
         }
         if state.players.len() >= player_limit() {
-            state.deferred.insert(alias, owner);
+            state.deferred.insert(
+                alias,
+                DeferredTarget {
+                    owner,
+                    retry_at: None,
+                },
+            );
             return;
         }
         state.deferred.remove(&alias);
@@ -463,7 +498,7 @@ async fn remove_alias(
             if state
                 .deferred
                 .get(alias)
-                .is_some_and(|value| value == owner)
+                .is_some_and(|value| value.owner == owner)
             {
                 state.deferred.remove(alias);
             }
@@ -476,7 +511,7 @@ async fn remove_alias(
 async fn remove_owner(state: &std::sync::Arc<Mutex<State>>, common: &Common, owner: &str) {
     let records = {
         let mut state = state.lock().await;
-        state.deferred.retain(|_, value| value != owner);
+        state.deferred.retain(|_, value| value.owner != owner);
         state
             .by_owner
             .get(owner)
@@ -493,15 +528,23 @@ async fn remove_monitored_player(
     player_id: u32,
     owner: &str,
 ) {
-    let records = {
+    let removed = {
         let mut state = state.lock().await;
         if state.by_owner.get(owner) == Some(&player_id) {
-            state.remove_player(player_id)
+            Some(state.remove_player_for_retry(player_id, Instant::now() + MONITOR_RETRY_DELAY))
         } else {
-            Vec::new()
+            None
         }
     };
+    let Some(records) = removed else {
+        return;
+    };
     send_records(common, records).await;
+    let capacity_changed = state.lock().await.capacity_changed.clone();
+    tokio::spawn(async move {
+        tokio::time::sleep(MONITOR_RETRY_DELAY).await;
+        capacity_changed.notify_one();
+    });
 }
 
 async fn retry_deferred(
@@ -510,7 +553,10 @@ async fn retry_deferred(
     common: &Common,
 ) {
     loop {
-        let candidate = state.lock().await.deferred_candidate(player_limit());
+        let candidate = state
+            .lock()
+            .await
+            .deferred_candidate(player_limit(), Instant::now());
         let Some((alias, owner)) = candidate else {
             return;
         };
@@ -1514,18 +1560,46 @@ mod tests {
         let mut state = State::default();
         state.players.insert(1, target(1, now));
         state.by_owner.insert(":1.1".into(), 1);
-        state
-            .deferred
-            .insert("org.mpris.MediaPlayer2.deferred".into(), ":1.2".into());
+        state.deferred.insert(
+            "org.mpris.MediaPlayer2.deferred".into(),
+            DeferredTarget {
+                owner: ":1.2".into(),
+                retry_at: None,
+            },
+        );
 
-        assert_eq!(state.deferred_candidate(1), None);
+        assert_eq!(state.deferred_candidate(1, now), None);
         state.remove_player(1);
         assert_eq!(
-            state.deferred_candidate(1),
+            state.deferred_candidate(1, now),
             Some(("org.mpris.MediaPlayer2.deferred".into(), ":1.2".into()))
         );
     }
 
+    #[test]
+    fn failed_monitor_retries_only_after_its_cooldown() {
+        let now = Instant::now();
+        let retry_at = now + MONITOR_RETRY_DELAY;
+        let alias = "org.mpris.MediaPlayer2.retry".to_string();
+        let mut state = State::default();
+        let mut player = target(1, now);
+        player.aliases.insert(alias.clone());
+        state.players.insert(1, player);
+        state.by_owner.insert(":1.1".into(), 1);
+        state.by_alias.insert(alias.clone(), 1);
+
+        let records = state.remove_player_for_retry(1, retry_at);
+        assert!(matches!(
+            records.first(),
+            Some(MprisRecord::Delete { player_id: 1 })
+        ));
+        assert_eq!(state.deferred_candidate(1, now), None);
+        assert_eq!(
+            state.deferred_candidate(1, retry_at),
+            Some((alias, ":1.1".into()))
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn property_calls_are_strictly_limited_to_four() {
         let limiter = Arc::new(Semaphore::new(PROPERTY_CALL_LIMIT));

crates/server/src/lib.rs

@@ -3871,6 +3871,15 @@ struct ClientState {
     drag_staging_dir: Option<std::path::PathBuf>,
 }
 
+#[cfg(target_os = "linux")]
+fn mpris_snapshot_rate_limited(client: &ClientState, enabled: bool, now: Instant) -> bool {
+    enabled
+        && client.mpris_subscribed
+        && client
+            .last_mpris_snapshot
+            .is_some_and(|last| now.duration_since(last) < Duration::from_secs(1))
+}
+
 #[cfg(target_os = "linux")]
 fn consume_mpris_action_token(client: &mut ClientState, now: Instant) -> bool {
     const BURST: f64 = 20.0;
@@ -19524,12 +19533,9 @@ async fn handle_client_registered<S: AsyncRead + AsyncWrite + Unpin + Send + 'st
                     }
                     ClientControl::MprisSubscribe { enabled } => {
                         let now = Instant::now();
-                        let rate_limited = enabled
-                            && sess.clients.get(&client_id).is_some_and(|client| {
-                                client.last_mpris_snapshot.is_some_and(|last| {
-                                    now.duration_since(last) < Duration::from_secs(1)
-                                })
-                            });
+                        let rate_limited = sess.clients.get(&client_id).is_some_and(|client| {
+                            mpris_snapshot_rate_limited(client, enabled, now)
+                        });
                         if rate_limited {
                             continue;
                         }
@@ -23773,6 +23779,24 @@ mod tests {
         ));
     }
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn mpris_snapshot_limit_never_suppresses_a_new_subscription() {
+        let mut client = test_client();
+        let now = Instant::now();
+        client.last_mpris_snapshot = Some(now);
+
+        assert!(!mpris_snapshot_rate_limited(&client, true, now));
+        client.mpris_subscribed = true;
+        assert!(mpris_snapshot_rate_limited(&client, true, now));
+        assert!(!mpris_snapshot_rate_limited(&client, false, now));
+        assert!(!mpris_snapshot_rate_limited(
+            &client,
+            true,
+            now + Duration::from_secs(1)
+        ));
+    }
+
     #[cfg(target_os = "linux")]
     fn mpris_player(position_us: i64, length_us: i64) -> blit_remote::media::MprisPlayer {
         blit_remote::media::MprisPlayer {

docs/design/media-devices-portals.md

@@ -738,7 +738,9 @@ The bridge calls `GetAll` for `org.mpris.MediaPlayer2` and
 `org.freedesktop.DBus.Properties.PropertiesChanged` and `Seeked`. A missing
 Player interface, loss of required properties, or three consecutive two-second
 D-Bus timeouts removes the player. Optional malformed properties fall back to
-their specified defaults without removing it.
+their specified defaults without removing it. A removed player whose bus name
+remains owned is reconsidered after a five-second cooldown, so a temporarily
+unresponsive application can recover without restarting.
 
 Property signals are invalidations, not trusted replacement records. The
 bridge batches them into at most one refresh per player per 50 ms and permits
@@ -977,7 +979,8 @@ Rate, 0/1 for Shuffle, and 0 None/1 Track/2 Playlist for LoopStatus. It is zero
 for other actions. `track_revision` must match for SetPosition and is zero for
 other kinds. A random nonzero nonce correlates one action result.
 
-Fresh snapshot requests are accepted at most once per connection per second.
+Fresh snapshot requests are accepted at most once per subscribed connection per
+second. A transition from unsubscribed to subscribed is always accepted.
 An excess request is ignored while the existing subscription and state stream
 continue.