Fix startup auto-route race

src/mini_eq/app.py

@@ -23,15 +23,28 @@
     set_background_status,
 )
 from .cli import parse_args
+from .core import AudioBackendError
 from .dbus_control import MiniEqDbusControl, call_present_window
 from .desktop_integration import APP_ICON_NAME, APP_ID, install_app_icon, install_desktop_integration
 from .glib_utils import destroy_glib_source
 from .instance import MiniEqAlreadyRunningError, MiniEqInstanceGuard
+from .pipewire_backend import PipeWireBackendError
 from .routing import SystemWideEqController
 from .window import MiniEqWindow
 from .window_presets import imported_apo_curve_label
 
 STARTUP_NOTIFICATION_ENV_KEYS = ("XDG_ACTIVATION_TOKEN", "DESKTOP_STARTUP_ID")
+STARTUP_AUTO_ROUTE_RETRY_INTERVAL_SECONDS = 1
+STARTUP_AUTO_ROUTE_RETRY_TIMEOUT_US = 30_000_000
+STARTUP_AUTO_ROUTE_RETRYABLE_PIPEWIRE_PREFIXES = (
+    "failed to connect to PipeWire",
+    "failed to start PipeWire registry discovery",
+    "failed to start PipeWire default metadata discovery",
+    "PipeWire registry sync failed",
+    "PipeWire metadata sync failed",
+    "PipeWire core sync failed",
+    "PipeWire initialization did not report:",
+)
 
 
 class MiniEqApplication(Adw.Application):
@@ -45,6 +58,9 @@ def __init__(self, args: Namespace, startup_notification_id: str | None = None)
         self.window_present_source_id = 0
         self.window_starting = False
         self.window_start_hold = False
+        self.window_start_retry_source_id = 0
+        self.window_start_retry_deadline_us = 0
+        self.window_start_last_error: Exception | None = None
         self.pending_present_when_ready = False
         self.pending_startup_notification_id = (
             None if bool(getattr(args, "background", False)) else startup_notification_id
@@ -107,6 +123,77 @@ def ensure_window(self, *, present: bool, startup_id: str | None = None) -> None
             self.pending_present_when_ready = self.pending_present_when_ready or present
             return
 
+        self.begin_window_start(present)
+        self.start_window_controller()
+
+    def begin_window_start(self, present: bool) -> None:
+        self.window_starting = True
+        self.window_start_hold = True
+        self.pending_present_when_ready = present
+        if self.should_retry_startup_auto_route():
+            self.window_start_retry_deadline_us = GLib.get_monotonic_time() + STARTUP_AUTO_ROUTE_RETRY_TIMEOUT_US
+        else:
+            self.window_start_retry_deadline_us = 0
+        self.hold()
+
+    def release_window_start_hold(self) -> None:
+        if not self.window_start_hold:
+            return
+
+        self.window_start_hold = False
+        self.release()
+
+    def should_retry_startup_auto_route(self) -> bool:
+        return bool(getattr(self.args, "auto_route", False))
+
+    def is_startup_auto_route_retryable_error(self, exc: Exception) -> bool:
+        message = str(exc)
+        if isinstance(exc, AudioBackendError):
+            return message.startswith("output sink not found:")
+
+        if isinstance(exc, PipeWireBackendError):
+            return message.startswith(STARTUP_AUTO_ROUTE_RETRYABLE_PIPEWIRE_PREFIXES)
+
+        return False
+
+    def retry_startup_auto_route_after_error(self, exc: Exception) -> bool:
+        if not self.should_retry_startup_auto_route() or not self.is_startup_auto_route_retryable_error(exc):
+            return False
+
+        deadline_us = getattr(self, "window_start_retry_deadline_us", 0)
+        if deadline_us <= 0 or GLib.get_monotonic_time() >= deadline_us:
+            return False
+
+        self.window_start_last_error = exc
+        if self.window_start_retry_source_id == 0:
+            self.window_start_retry_source_id = GLib.timeout_add_seconds(
+                STARTUP_AUTO_ROUTE_RETRY_INTERVAL_SECONDS,
+                self.on_window_start_retry_timeout,
+            )
+        return True
+
+    def on_window_start_retry_timeout(self) -> bool:
+        self.window_start_retry_source_id = 0
+        if not self.window_starting or self.window is not None:
+            return False
+
+        self.start_window_controller()
+        return False
+
+    def fail_window_start(self, exc: Exception) -> None:
+        self.window_starting = False
+        self.pending_present_when_ready = False
+        print(str(exc), file=sys.stderr)
+        self.release_window_start_hold()
+        self.quit()
+
+    def raise_window_start_error(self, exc: Exception) -> None:
+        self.window_starting = False
+        self.pending_present_when_ready = False
+        self.release_window_start_hold()
+        raise SystemExit(str(exc)) from exc
+
+    def start_window_controller(self) -> None:
         controller: SystemWideEqController | None = None
         initial_curve_label: str | None = None
 
@@ -118,20 +205,15 @@ def ensure_window(self, *, present: bool, startup_id: str | None = None) -> None
         except Exception as exc:
             if controller is not None:
                 controller.shutdown()
-            raise SystemExit(str(exc)) from exc
-
-        self.controller = controller
-        self.window_starting = True
-        self.window_start_hold = True
-        self.pending_present_when_ready = present
-        self.hold()
-
-        def release_start_hold() -> None:
-            if not self.window_start_hold:
+            if self.retry_startup_auto_route_after_error(exc):
                 return
+            if self.should_retry_startup_auto_route() and self.is_startup_auto_route_retryable_error(exc):
+                self.fail_window_start(exc)
+                return
+            self.raise_window_start_error(exc)
+            return
 
-            self.window_start_hold = False
-            self.release()
+        self.controller = controller
 
         def on_ready() -> None:
             if self.controller is not controller:
@@ -152,19 +234,17 @@ def on_ready() -> None:
                     self.update_background_status()
                     self.emit_control_state_changed()
             finally:
-                release_start_hold()
+                self.release_window_start_hold()
 
         def on_error(exc: Exception) -> None:
-            self.window_starting = False
-            self.pending_present_when_ready = False
             try:
                 controller.shutdown()
             finally:
                 if self.controller is controller:
                     self.controller = None
-            print(str(exc), file=sys.stderr)
-            release_start_hold()
-            self.quit()
+            if self.retry_startup_auto_route_after_error(exc):
+                return
+            self.fail_window_start(exc)
 
         controller.start(on_ready=on_ready, on_error=on_error)
 
@@ -248,6 +328,9 @@ def do_shutdown(self) -> None:
         for source_id in self.signal_source_ids:
             destroy_glib_source(source_id)
         self.signal_source_ids = []
+        if self.window_start_retry_source_id > 0:
+            destroy_glib_source(self.window_start_retry_source_id)
+        self.window_start_retry_source_id = 0
         if self.window_present_source_id > 0:
             destroy_glib_source(self.window_present_source_id)
         self.window_present_source_id = 0

src/mini_eq/routing.py

@@ -59,56 +59,60 @@
 class SystemWideEqController:
     def __init__(self, output_sink: str | None) -> None:
         self.output_backend = PipeWireBackend()
-        self.output_backend.connect()
-        self.virtual_sink_name = self.pick_virtual_sink_name()
-        self.original_default_sink = self.resolve_default_output_sink_name()
-        self.follow_default_output = output_sink is None
-        self.output_sink = output_sink or self.original_default_sink
-        self._output_preset_target_sink: str | None = None
-        self._output_preset_target: PipeWireOutputPresetTarget | None = None
-        self.filter_output_name = f"{self.virtual_sink_name}{FILTER_OUTPUT_SUFFIX}"
-        self.engine_module = None
-        self.engine_start_watch = None
-        self.engine_start_pending = False
-        self.filter_node_id: int | None = None
-        self.output_event_source_id = 0
-        self.pending_followed_output_sink: str | None = None
-        self.pending_current_output_sink_refresh = False
-        self.output_object_added_handler_id = 0
-        self.output_object_removed_handler_id = 0
-        self.output_metadata_changed_handler_id = 0
-        self.output_route_param_handler_id = 0
-        self.output_route_param_device_id = 0
-        self.filter_node_state_handler_id = 0
-        self.filter_control_param_handler_id = 0
-        self.filter_control_reapply_source_id = 0
-        self.filter_control_reapply_source_is_verification = False
-        self.applying_filter_control_verification = False
-        self.ignored_filter_control_param_events = 0
-        self.ignored_filter_control_param_events_to_verify = 0
-        self.ignored_filter_control_param_events_deadline_us = 0
-        self.accept_output_events = False
-        self.routed = False
-        self.running = False
-        self.shutting_down = False
-        self.status_callback: Callable[[str], None] | None = None
-        self.outputs_changed_callback: Callable[[], None] | None = None
-        self.analyzer_levels_callback: Callable[[list[float]], None] | None = None
-        self.analyzer_loudness_callback: Callable[[AnalyzerLoudnessSnapshot | None], None] | None = None
-        self.eq_enabled = True
-        self.eq_mode = next(iter(EQ_MODES.values()))
-        self.preamp_db = 0.0
-        self.default_bands: list[EqBand] = self.build_default_bands()
-        self.bands: list[EqBand] = [replace(band) for band in self.default_bands]
-        self.stream_router: PipeWireStreamRouter | None = None
-        self.output_analyzer: OutputSpectrumAnalyzer | None = None
-        self.analyzer_response_speed = ANALYZER_RESPONSE_DEFAULT
-
-        if not self.is_valid_output_sink(self.output_sink):
-            raise AudioBackendError("output sink cannot be a Mini EQ virtual sink")
-
-        if not self.output_sink or self.get_sink(self.output_sink) is None:
-            raise AudioBackendError(f"output sink not found: {self.output_sink}")
+        try:
+            self.output_backend.connect()
+            self.virtual_sink_name = self.pick_virtual_sink_name()
+            self.original_default_sink = self.resolve_default_output_sink_name()
+            self.follow_default_output = output_sink is None
+            self.output_sink = output_sink or self.original_default_sink
+            self._output_preset_target_sink: str | None = None
+            self._output_preset_target: PipeWireOutputPresetTarget | None = None
+            self.filter_output_name = f"{self.virtual_sink_name}{FILTER_OUTPUT_SUFFIX}"
+            self.engine_module = None
+            self.engine_start_watch = None
+            self.engine_start_pending = False
+            self.filter_node_id: int | None = None
+            self.output_event_source_id = 0
+            self.pending_followed_output_sink: str | None = None
+            self.pending_current_output_sink_refresh = False
+            self.output_object_added_handler_id = 0
+            self.output_object_removed_handler_id = 0
+            self.output_metadata_changed_handler_id = 0
+            self.output_route_param_handler_id = 0
+            self.output_route_param_device_id = 0
+            self.filter_node_state_handler_id = 0
+            self.filter_control_param_handler_id = 0
+            self.filter_control_reapply_source_id = 0
+            self.filter_control_reapply_source_is_verification = False
+            self.applying_filter_control_verification = False
+            self.ignored_filter_control_param_events = 0
+            self.ignored_filter_control_param_events_to_verify = 0
+            self.ignored_filter_control_param_events_deadline_us = 0
+            self.accept_output_events = False
+            self.routed = False
+            self.running = False
+            self.shutting_down = False
+            self.status_callback: Callable[[str], None] | None = None
+            self.outputs_changed_callback: Callable[[], None] | None = None
+            self.analyzer_levels_callback: Callable[[list[float]], None] | None = None
+            self.analyzer_loudness_callback: Callable[[AnalyzerLoudnessSnapshot | None], None] | None = None
+            self.eq_enabled = True
+            self.eq_mode = next(iter(EQ_MODES.values()))
+            self.preamp_db = 0.0
+            self.default_bands: list[EqBand] = self.build_default_bands()
+            self.bands: list[EqBand] = [replace(band) for band in self.default_bands]
+            self.stream_router: PipeWireStreamRouter | None = None
+            self.output_analyzer: OutputSpectrumAnalyzer | None = None
+            self.analyzer_response_speed = ANALYZER_RESPONSE_DEFAULT
+
+            if not self.is_valid_output_sink(self.output_sink):
+                raise AudioBackendError("output sink cannot be a Mini EQ virtual sink")
+
+            if not self.output_sink or self.get_sink(self.output_sink) is None:
+                raise AudioBackendError(f"output sink not found: {self.output_sink}")
+        except Exception:
+            self.output_backend.close()
+            raise
 
     def emit_status(self, message: str) -> None:
         if getattr(self, "shutting_down", False):

src/mini_eq/window.py

@@ -32,12 +32,13 @@
     EQ_MODES,
     MODE_ORDER,
     SAMPLE_RATE,
+    AudioBackendError,
     ensure_preset_storage_dir,
     estimate_response_peak_db,
 )
 from .glib_utils import destroy_glib_source
 from .gtk_utils import create_dropdown_from_strings
-from .pipewire_backend import PipeWireNode, node_sample_rate, parse_positive_int
+from .pipewire_backend import PipeWireBackendError, PipeWireNode, node_sample_rate, parse_positive_int
 from .routing import SystemWideEqController
 from .settings import load_monitor_enabled
 from .window_analyzer import MiniEqWindowAnalyzerMixin
@@ -58,6 +59,8 @@
 DEFAULT_WINDOW_HEIGHT = 720
 DEFAULT_WINDOW_SCREEN_MARGIN = 32
 ROUTING_CLOSE_SETTLE_MS = 300
+STARTUP_AUTO_ROUTE_RETRY_INTERVAL_SECONDS = 1
+STARTUP_AUTO_ROUTE_RETRY_TIMEOUT_US = 30_000_000
 TOAST_IGNORED_PREFIXES = (
     "filter-chain PipeWire EQ ready:",
     "filter-chain PipeWire EQ stopped",
@@ -185,6 +188,7 @@ def __init__(
         self.auto_route_on_startup = auto_route
         self.startup_ready_source_id = 0
         self.startup_auto_route_source_id = 0
+        self.startup_auto_route_deadline_us = 0
         self.startup_ready = False
         self.present_when_ready = True
         self.responsive_layout_source_id = 0
@@ -369,18 +373,52 @@ def on_startup_ready_idle(self) -> bool:
         self.notify_control_state_changed()
         return False
 
-    def schedule_startup_auto_route(self) -> None:
+    def schedule_startup_auto_route(self, *, retry: bool = False) -> None:
         if self.startup_auto_route_source_id != 0:
             return
 
+        if retry:
+            self.startup_auto_route_source_id = GLib.timeout_add_seconds(
+                STARTUP_AUTO_ROUTE_RETRY_INTERVAL_SECONDS,
+                self.on_startup_auto_route_idle,
+            )
+            return
+
         self.startup_auto_route_source_id = GLib.idle_add(self.on_startup_auto_route_idle)
 
+    def is_startup_auto_route_retryable_error(self, exc: Exception) -> bool:
+        if isinstance(exc, PipeWireBackendError | AudioBackendError):
+            return True
+
+        return str(exc) == "filter-chain PipeWire EQ is not ready"
+
+    def schedule_startup_auto_route_retry_after_error(self, exc: Exception) -> bool:
+        if (
+            self.ui_shutting_down
+            or not self.auto_route_on_startup
+            or not self.is_startup_auto_route_retryable_error(exc)
+        ):
+            return False
+
+        deadline_us = getattr(self, "startup_auto_route_deadline_us", 0)
+        if deadline_us <= 0:
+            deadline_us = GLib.get_monotonic_time() + STARTUP_AUTO_ROUTE_RETRY_TIMEOUT_US
+            self.startup_auto_route_deadline_us = deadline_us
+        if GLib.get_monotonic_time() >= deadline_us:
+            return False
+
+        self.schedule_startup_auto_route(retry=True)
+        return True
+
     def apply_startup_auto_route(self) -> None:
         eq_was_enabled = self.controller.eq_enabled
         try:
             self.controller.route_system_audio(True)
         except Exception as exc:
-            self.set_status(str(exc))
+            if not self.schedule_startup_auto_route_retry_after_error(exc):
+                self.set_status(str(exc))
+        else:
+            self.startup_auto_route_deadline_us = 0
         self.refresh_after_route_state_changed(eq_was_enabled=eq_was_enabled)
 
     def on_startup_auto_route_idle(self) -> bool:

tests/test_check_live_ui_runtime.py

@@ -57,3 +57,18 @@ def test_live_ui_runtime_rejects_inactive_processing_path(monkeypatch) -> None:
     )
 
     assert check_live_ui_runtime.processing_path_has_active_links() is False
+
+
+def test_live_ui_runtime_can_skip_static_sink_config(tmp_path) -> None:
+    check_live_ui_runtime.write_pipewire_config(tmp_path, include_static_sinks=False)
+
+    assert not (tmp_path / "pipewire" / "pipewire.conf.d" / "10-mini-eq-live-ui-null-sinks.conf").exists()
+
+
+def test_live_ui_runtime_dynamic_sink_properties_create_null_audio_sink() -> None:
+    properties = check_live_ui_runtime.dynamic_sink_properties("ci_null_sink", "CI Null Sink")
+
+    assert "factory.name = support.null-audio-sink" in properties
+    assert 'node.name = "ci_null_sink"' in properties
+    assert 'node.description = "CI Null Sink"' in properties
+    assert 'media.class = "Audio/Sink"' in properties