Refresh default metadata snapshots on output events

Author: bhack

src/mini_eq/pipewire_backend.py

@@ -614,7 +614,14 @@ def defaults(self) -> PipeWireDefaults:
 
         return self.refresh_defaults()
 
-    def refresh_defaults(self) -> PipeWireDefaults:
+    def refresh_defaults(self, *, snapshot: bool = False) -> PipeWireDefaults:
+        snapshot_error: Exception | None = None
+        if snapshot:
+            try:
+                self._resnapshot_metadata()
+            except Exception as exc:
+                snapshot_error = exc
+
         try:
             self._cached_defaults = self._read_defaults()
             return self._cached_defaults
@@ -636,6 +643,10 @@ def refresh_defaults(self) -> PipeWireDefaults:
         except Exception:
             if self._has_cached_defaults():
                 return self._cached_defaults
+            if snapshot_error is not None:
+                raise PipeWireBackendError(f"failed to refresh PipeWire metadata snapshot: {snapshot_error}") from (
+                    snapshot_error
+                )
             raise
 
     def remember_default_metadata_change(self, key: str, value: str | None) -> bool:
@@ -660,6 +671,22 @@ def _read_defaults(self) -> PipeWireDefaults:
     def _has_cached_defaults(self) -> bool:
         return bool(self._cached_defaults.default_audio_sink or self._cached_defaults.configured_audio_sink)
 
+    def _resnapshot_metadata(self) -> None:
+        if self._metadata is None:
+            return
+
+        try:
+            # Rebinding asks PipeWire for the current metadata properties. This
+            # covers clients that missed property events while the metadata
+            # resource was pending, without polling or sleeping.
+            self._metadata.stop()
+            if not self._metadata.start():
+                raise PipeWireBackendError("failed to restart PipeWire default metadata discovery")
+            self._sync_metadata()
+            self._cached_defaults = PipeWireDefaults(None, None)
+        except Exception as exc:
+            raise PipeWireBackendError(f"failed to refresh PipeWire metadata snapshot: {exc}") from exc
+
     def move_stream_to_target(self, stream_bound_id: int, target_node_name: str) -> None:
         stream = self.output_stream_by_bound_id(stream_bound_id)
         if stream is None:

src/mini_eq/routing.py

@@ -169,14 +169,16 @@ def output_preset_target(self, *, refresh: bool = False) -> PipeWireOutputPreset
         self._output_preset_target = target
         return target
 
-    def default_output_sink_candidates(self, *, refresh: bool = False) -> tuple[str, ...]:
-        defaults = self.output_backend.refresh_defaults() if refresh else self.output_backend.defaults()
+    def default_output_sink_candidates(self, *, refresh: bool = False, snapshot: bool = False) -> tuple[str, ...]:
+        defaults = (
+            self.output_backend.refresh_defaults(snapshot=snapshot) if refresh else self.output_backend.defaults()
+        )
         return tuple(
             sink_name for sink_name in (defaults.configured_audio_sink, defaults.default_audio_sink) if sink_name
         )
 
-    def get_default_output_sink_name(self, *, refresh: bool = False) -> str:
-        candidates = self.default_output_sink_candidates(refresh=refresh)
+    def get_default_output_sink_name(self, *, refresh: bool = False, snapshot: bool = False) -> str:
+        candidates = self.default_output_sink_candidates(refresh=refresh, snapshot=snapshot)
 
         for sink_name in candidates:
             if self.is_valid_output_sink(sink_name) and self.get_sink(sink_name) is not None:
@@ -231,6 +233,7 @@ def prepare_output_analyzer(self) -> bool:
         return analyzer.prepare()
 
     def set_analyzer_enabled(self, enabled: bool) -> bool:
+        self.refresh_followed_output_sink(snapshot=True)
         analyzer = self.ensure_output_analyzer()
 
         if not enabled:
@@ -304,15 +307,15 @@ def switch_output_sink(self, sink_name: str, explicit: bool) -> None:
     def follow_system_default_output(self) -> None:
         previous_output_sink = getattr(self, "output_sink", None)
         self.follow_default_output = True
-        self.refresh_followed_output_sink()
+        self.refresh_followed_output_sink(snapshot=True)
         if getattr(self, "output_sink", None) != previous_output_sink:
             self.schedule_output_event_refresh()
 
-    def refresh_followed_output_sink(self) -> bool:
-        if not self.follow_default_output:
+    def refresh_followed_output_sink(self, *, snapshot: bool = False) -> bool:
+        if not getattr(self, "follow_default_output", False):
             return False
 
-        for default_sink in self.default_output_sink_candidates(refresh=True):
+        for default_sink in self.default_output_sink_candidates(refresh=True, snapshot=snapshot):
             if not self.is_valid_output_sink(default_sink) or self.get_sink(default_sink) is None:
                 continue
             try:
@@ -334,7 +337,7 @@ def refresh_followed_output_sink_from_event(self, sink_name: str | None) -> bool
                 self.emit_status(f"default output follow warning: {exc}")
             return True
 
-        return self.refresh_followed_output_sink()
+        return self.refresh_followed_output_sink(snapshot=True)
 
     def schedule_output_event_refresh(self) -> None:
         if not getattr(self, "accept_output_events", False):
@@ -424,7 +427,7 @@ def on_output_event_idle(self) -> bool:
         if pending_followed_output_sink is not None:
             self.refresh_followed_output_sink_from_event(pending_followed_output_sink)
         else:
-            self.refresh_followed_output_sink()
+            self.refresh_followed_output_sink(snapshot=True)
         self.refresh_output_route_param_monitor()
 
         if self.outputs_changed_callback is not None:
@@ -451,7 +454,7 @@ def start_output_event_monitoring(self) -> None:
             )
 
         self.invalidate_output_preset_target()
-        self.refresh_followed_output_sink()
+        self.refresh_followed_output_sink(snapshot=True)
         self.refresh_output_route_param_monitor()
 
         if self.outputs_changed_callback is not None:

tests/test_mini_eq_pipewire_backend.py

@@ -1439,6 +1439,34 @@ def raise_decode_error():
     assert syncs == [True]
 
 
+def test_refresh_defaults_can_resnapshot_metadata_before_read(monkeypatch) -> None:
+    backend = pw_backend.PipeWireBackend()
+    calls: list[str] = []
+
+    class FakeMetadata:
+        def stop(self) -> None:
+            calls.append("stop")
+
+        def start(self) -> bool:
+            calls.append("start")
+            return True
+
+    backend._metadata = FakeMetadata()
+    backend._cached_defaults = pw_backend.PipeWireDefaults("stale.default", "stale.configured")
+    monkeypatch.setattr(backend, "_sync_metadata", lambda: calls.append("sync"))
+    monkeypatch.setattr(
+        backend,
+        "_read_defaults",
+        lambda: calls.append("read") or pw_backend.PipeWireDefaults("fresh.default", "fresh.configured"),
+    )
+
+    assert backend.refresh_defaults(snapshot=True) == pw_backend.PipeWireDefaults(
+        "fresh.default",
+        "fresh.configured",
+    )
+    assert calls == ["stop", "start", "sync", "read"]
+
+
 def test_remember_default_metadata_change_updates_cache() -> None:
     backend = pw_backend.PipeWireBackend()
 

tests/test_mini_eq_routing.py

@@ -53,12 +53,14 @@ def __init__(
         self.cached_defaults = cached_defaults
         self.refreshed_defaults = refreshed_defaults
         self.refresh_count = 0
+        self.snapshot_values: list[bool] = []
 
     def defaults(self) -> pw_backend.PipeWireDefaults:
         return self.cached_defaults
 
-    def refresh_defaults(self) -> pw_backend.PipeWireDefaults:
+    def refresh_defaults(self, *, snapshot: bool = False) -> pw_backend.PipeWireDefaults:
         self.refresh_count += 1
+        self.snapshot_values.append(snapshot)
         return self.refreshed_defaults
 
 
@@ -211,9 +213,35 @@ def test_refresh_followed_output_sink_refreshes_metadata_and_skips_virtual_defau
 
     assert routing.SystemWideEqController.refresh_followed_output_sink(controller) is True
     assert backend.refresh_count == 1
+    assert backend.snapshot_values == [False]
     assert calls == [("speakers", False)]
 
 
+def test_output_event_idle_refreshes_followed_sink_from_metadata_snapshot() -> None:
+    controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController)
+    backend = FakeDefaultOutputBackend(
+        [make_node(1, "hdmi")],
+        cached_defaults=pw_backend.PipeWireDefaults("speakers", None),
+        refreshed_defaults=pw_backend.PipeWireDefaults("hdmi", None),
+    )
+    controller.accept_output_events = True
+    controller.output_event_source_id = 123
+    controller.pending_followed_output_sink = None
+    controller.follow_default_output = True
+    controller.output_backend = backend
+    controller._output_preset_target_sink = "speakers"
+    controller._output_preset_target = pw_routes.PipeWireOutputPresetTarget("speakers", None, ("speakers",))
+    calls: list[object] = []
+    controller.switch_output_sink = lambda sink_name, explicit: calls.append(("switch", sink_name, explicit))
+    controller.refresh_output_route_param_monitor = lambda: calls.append("route-monitor")
+    controller.outputs_changed_callback = lambda: calls.append("outputs")
+
+    assert routing.SystemWideEqController.on_output_event_idle(controller) is False
+
+    assert backend.snapshot_values == [True]
+    assert calls == [("switch", "hdmi", False), "route-monitor", "outputs"]
+
+
 def test_output_metadata_change_schedules_one_refresh(monkeypatch: pytest.MonkeyPatch) -> None:
     controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController)
     controller.accept_output_events = True
@@ -296,7 +324,7 @@ def test_output_event_idle_uses_pending_followed_sink_before_metadata_refresh()
     controller._output_preset_target = pw_routes.PipeWireOutputPresetTarget("speakers", None, ("speakers",))
     calls: list[object] = []
     controller.switch_output_sink = lambda sink_name, explicit: calls.append(("switch", sink_name, explicit))
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.refresh_output_route_param_monitor = lambda: calls.append("route-monitor")
     controller.outputs_changed_callback = lambda: calls.append("outputs")
 
@@ -340,7 +368,7 @@ def test_output_event_idle_invalidates_output_preset_target_cache() -> None:
     controller._output_preset_target_sink = "speakers"
     controller._output_preset_target = pw_routes.PipeWireOutputPresetTarget("speakers", None, ("speakers",))
     calls: list[str] = []
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.outputs_changed_callback = lambda: calls.append("outputs")
 
     assert routing.SystemWideEqController.on_output_event_idle(controller) is False
@@ -431,7 +459,7 @@ def disconnect_device_handler(self, handler_id: int) -> None:
     controller.output_route_param_handler_id = 0
     controller.output_route_param_device_id = 0
     controller.follow_default_output = False
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.outputs_changed_callback = lambda: calls.append("outputs")
 
     monkeypatch.setattr(routing.GLib, "idle_add", lambda callback: calls.append(("idle", callback)) or 321)
@@ -492,7 +520,7 @@ def test_follow_system_default_output_enables_follow_mode_and_refreshes() -> Non
     controller.follow_default_output = False
     calls: list[str] = []
 
-    def fake_refresh() -> bool:
+    def fake_refresh(**_kwargs) -> bool:
         calls.append("refresh")
         return True
 
@@ -512,7 +540,7 @@ def test_follow_system_default_output_schedules_refresh_when_output_changes(monk
     controller.output_event_source_id = 0
     scheduled_callbacks: list[object] = []
 
-    def fake_refresh() -> bool:
+    def fake_refresh(**_kwargs) -> bool:
         controller.output_sink = "hdmi"
         return True
 
@@ -956,7 +984,7 @@ def test_start_prepares_analyzer_before_filter_chain_engine() -> None:
     controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController)
     calls: list[str] = []
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.prepare_output_analyzer = lambda: calls.append("prepare") or True
     controller.start_engine = lambda *, on_ready=None, on_error=None: (calls.append("engine"), on_ready and on_ready())
     controller.start_output_event_monitoring = lambda: calls.append("monitor")
@@ -1030,7 +1058,7 @@ def close(self) -> None:
     controller.stream_router = FakeStreamRouter()
     controller.output_analyzer = None
     controller.output_backend = FakeBackend()
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.stop_output_event_monitoring = lambda: calls.append("stop-monitor")
     controller.engine_module = object()
     controller.filter_node_id = 42
@@ -1059,7 +1087,7 @@ def test_route_system_audio_does_not_enable_during_shutdown() -> None:
     controller.routed = False
     calls: list[str] = []
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.ensure_stream_router = lambda: calls.append("router")
 
     routing.SystemWideEqController.route_system_audio(controller, True)
@@ -1077,7 +1105,7 @@ def test_route_system_audio_requires_ready_engine_before_enabling() -> None:
     controller.eq_enabled = True
     calls: list[str] = []
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.ensure_stream_router = lambda: calls.append("router")
 
     with pytest.raises(RuntimeError, match="filter-chain PipeWire EQ is not ready"):
@@ -1105,7 +1133,7 @@ def set_eq_enabled(enabled: bool) -> None:
         calls.append(f"eq:{enabled}")
         controller.eq_enabled = enabled
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.set_eq_enabled = set_eq_enabled
     controller.ensure_stream_router = lambda: calls.append("router") or FakeRouter()
     controller.apply_state_to_engine = lambda: calls.append("apply")
@@ -1139,7 +1167,7 @@ class FakeRouter:
         def enable(self) -> None:
             calls.append("enable")
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.ensure_stream_router = lambda: calls.append("router") or FakeRouter()
     controller.apply_state_to_engine = lambda: calls.append("apply")
     controller.emit_status = lambda message: calls.append(f"status:{message}")
@@ -1174,7 +1202,7 @@ def set_eq_enabled(enabled: bool) -> None:
         calls.append(f"eq:{enabled}")
         controller.eq_enabled = enabled
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.set_eq_enabled = set_eq_enabled
     controller.ensure_stream_router = lambda: calls.append("router") or FailingRouter()
 
@@ -1201,7 +1229,7 @@ def set_eq_enabled(enabled: bool) -> None:
         if enabled:
             raise RuntimeError("control update failed")
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.set_eq_enabled = set_eq_enabled
     controller.ensure_stream_router = lambda: calls.append("router")
 
@@ -1225,7 +1253,7 @@ class FakeRouter:
         def disable(self, announce: bool = True) -> None:
             calls.append(f"disable:{announce}")
 
-    controller.refresh_followed_output_sink = lambda: calls.append("refresh")
+    controller.refresh_followed_output_sink = lambda **_kwargs: calls.append("refresh")
     controller.ensure_stream_router = lambda: calls.append("router") or FakeRouter()
     controller.emit_status = lambda message: calls.append(f"status:{message}")