diff --git a/gateway/run.py b/gateway/run.py index 6082a94053..4107f53b03 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -12181,10 +12181,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew enabled_platform_count = 0 startup_nonretryable_errors: list[str] = [] startup_retryable_errors: list[str] = [] - - # Initialize and connect each configured platform _multiplex_on = bool(getattr(self.config, "multiplex_profiles", False)) _multiplex_skipped_platforms: list[Platform] = [] + # Initialize and connect each configured platform. + # + # Parallel startup connect (#83791): the original code ran a serial for-loop, + # so every platform's connect() (with its own timeout) had to finish before + # the next began. A single slow/failing platform (e.g. Telegram behind a dead + # proxy) therefore delayed every other platform's connect by a full timeout + # window, cascading one platform's failure onto WeChat/QQ/etc. We now launch + # all platform connects concurrently and let each resolve on its own timeline; + # per-platform timeouts and error handling are unchanged. + # The serial pre-filter (cheap checks, adapter creation, handler wiring) stays + # sequential -- only the (slow) connect() calls run in parallel. + _pending_connects = [] # (platform, platform_config, adapter) for platform, platform_config in self.config.platforms.items(): if await self._abort_startup_if_shutdown_requested(): return True @@ -12196,7 +12206,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # empty token fails immediately and queues an infinite reconnect # loop that can never heal (#64674). Secondary profiles still # start their own adapters under _profile_runtime_scope with the - # real token — skip the empty primary instead of failing loudly. + # real token -- skip the empty primary instead of failing loudly. if _multiplex_on and not _platform_has_bot_credential(platform, platform_config): logger.info( "Skipping %s on default profile: no bot credential in this " @@ -12207,7 +12217,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _multiplex_skipped_platforms.append(platform) continue enabled_platform_count += 1 - + adapter = self._create_adapter(platform, platform_config) if not adapter: # Distinguish between missing builtin deps and missing plugin @@ -12215,14 +12225,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _builtin_names = {m.value for m in Platform.__members__.values()} if _pval not in _builtin_names: logger.warning( - "No adapter for '%s' — is the plugin installed? " + "No adapter for '%s' -- is the plugin installed? " "(platform is enabled in config.yaml but no plugin registered it)", _pval, ) else: logger.warning("No adapter available for %s", _pval) continue - + # Set up message + fatal error handlers. Under multiplexing the # default profile needs the same whole-handler runtime scope as a # secondary profile: authorization and prompt rendering both run @@ -12238,130 +12248,124 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform)) adapter.set_platform_event_handler(self._primary_platform_event_handler()) adapter._busy_text_mode = self._busy_text_mode - - # Try to connect - logger.info("Connecting to %s...", platform.value) + _pending_connects.append((platform, platform_config, adapter)) + + if await self._abort_startup_if_shutdown_requested(): + return True + + async def _connect_one_startup(p, p_cfg, adp): + """Connect a single platform; never let one block the others (#83791).""" + if await self._abort_startup_if_shutdown_requested(adp, p): + return (p, adp, p_cfg, "aborted", None) + logger.info("Connecting to %s...", p.value) self._update_platform_runtime_status( - platform.value, - platform_state="connecting", - error_code=None, - error_message=None, + p.value, platform_state="connecting", error_code=None, error_message=None, ) try: - success = await self._connect_initial_adapter_with_timeout( - adapter, platform - ) - if await self._abort_startup_if_shutdown_requested(adapter, platform): - return True - if success: - self.adapters[platform] = adapter - self._sync_voice_mode_state_to_adapter(adapter) - # Wire voice input callback at connect time so voice - # transcription is forwarded without requiring /voice join. - if hasattr(adapter, "_voice_input_callback"): - adapter._voice_input_callback = self._handle_voice_channel_input - connected_count += 1 - self._update_platform_runtime_status( - platform.value, - platform_state="connected", - error_code=None, - error_message=None, - needs_attention=False, - retrying_since=None, - ) - logger.info("✓ %s connected", platform.value) - else: - logger.warning("✗ %s failed to connect", platform.value) - # Defensive cleanup: a failed connect() may have - # allocated resources (aiohttp.ClientSession, poll - # tasks, bridge subprocesses) before giving up. - # Without this call, those resources are orphaned - # and Python logs "Unclosed client session" at - # process exit. Adapter disconnect() implementations - # are expected to be idempotent and tolerate - # partial-init state. - await self._safe_adapter_disconnect(adapter, platform) - if adapter.has_fatal_error: - self._update_platform_runtime_status( - platform.value, - platform_state="retrying" if adapter.fatal_error_retryable else "fatal", - error_code=adapter.fatal_error_code, - error_message=adapter.fatal_error_message, - ) - target = ( - startup_retryable_errors - if adapter.fatal_error_retryable - else startup_nonretryable_errors - ) - target.append( - f"{platform.value}: {adapter.fatal_error_message}" - ) - # Queue for reconnection if the error is retryable - if adapter.fatal_error_retryable: - self._failed_platforms[platform] = { - "config": platform_config, - "attempts": 1, - "next_retry": time.monotonic() + 30, - "queued_at": time.monotonic(), - "credential_claim": self._adapter_credential_claim( - platform, adapter - ), - "listener_claim": self._adapter_listener_claim( - platform, adapter - ), - } - else: - self._update_platform_runtime_status( - platform.value, - platform_state="retrying", - error_code=None, - error_message="failed to connect", - ) - startup_retryable_errors.append( - f"{platform.value}: failed to connect" - ) - # No fatal error info means likely a transient issue — queue for retry - self._failed_platforms[platform] = { - "config": platform_config, - "attempts": 1, - "next_retry": time.monotonic() + 30, - "queued_at": time.monotonic(), - "credential_claim": self._adapter_credential_claim( - platform, adapter - ), - "listener_claim": self._adapter_listener_claim( - platform, adapter - ), - } - except Exception as e: - logger.error("✗ %s error: %s", platform.value, e) - # Same defensive cleanup path for exceptions — an adapter - # that raised mid-connect may still have a live - # aiohttp.ClientSession or child subprocess. + ok = await self._connect_initial_adapter_with_timeout(adp, p) + except Exception as _exc: # noqa: BLE001 - surfaced below as a retryable error + return (p, adp, p_cfg, "exception", _exc) + return (p, adp, p_cfg, "ok" if ok else "failed", None) + + if _pending_connects: + _raw = await asyncio.gather( + *(_connect_one_startup(p, c, a) for (p, c, a) in _pending_connects), + return_exceptions=True, + ) + else: + _raw = [] + + # Aggregate results single-threaded so shared state (self.adapters, + # self._failed_platforms, the error lists, connected_count) is mutated + # exactly as the original serial loop did -- only the connect() wall-clock + # overlap changed. + for _item in _raw: + if isinstance(_item, Exception): + # Unexpected escape from _connect_one_startup (shouldn't happen); + # log and skip rather than aborting the whole startup. + logger.error("Unexpected startup connect error: %s", _item) + continue + platform, adapter, platform_config, outcome, exc = _item + if outcome == "aborted": + continue + if outcome == "exception": + logger.error("\u2717 %s error: %s", platform.value, exc) + # Same defensive cleanup path for exceptions -- an adapter that + # raised mid-connect may still have a live aiohttp.ClientSession or + # child subprocess. await self._safe_adapter_disconnect(adapter, platform) self._update_platform_runtime_status( - platform.value, - platform_state="retrying", - error_code=None, - error_message=str(e), + platform.value, platform_state="retrying", error_code=None, error_message=str(exc), ) - startup_retryable_errors.append(f"{platform.value}: {e}") - # Unexpected exceptions are typically transient — queue for retry + startup_retryable_errors.append(f"{platform.value}: {exc}") + # Unexpected exceptions are typically transient -- queue for retry self._failed_platforms[platform] = { "config": platform_config, "attempts": 1, "next_retry": time.monotonic() + 30, "queued_at": time.monotonic(), - "credential_claim": self._adapter_credential_claim( - platform, adapter - ), - "listener_claim": self._adapter_listener_claim( - platform, adapter - ), + "credential_claim": self._adapter_credential_claim(platform, adapter), + "listener_claim": self._adapter_listener_claim(platform, adapter), } - if await self._abort_startup_if_shutdown_requested(): - return True + continue + if outcome == "ok": + self.adapters[platform] = adapter + self._sync_voice_mode_state_to_adapter(adapter) + # Wire voice input callback at connect time so voice + # transcription is forwarded without requiring /voice join. + if hasattr(adapter, "_voice_input_callback"): + adapter._voice_input_callback = self._handle_voice_channel_input + connected_count += 1 + self._update_platform_runtime_status( + platform.value, platform_state="connected", error_code=None, error_message=None, + ) + logger.info("\u2713 %s connected", platform.value) + else: # outcome == "failed" + logger.warning("\u2717 %s failed to connect", platform.value) + # Defensive cleanup: a failed connect() may have allocated resources + # (aiohttp.ClientSession, poll tasks, bridge subprocesses) before + # giving up. Without this call, those resources are orphaned and + # Python logs "Unclosed client session" at process exit. + await self._safe_adapter_disconnect(adapter, platform) + if adapter.has_fatal_error: + self._update_platform_runtime_status( + platform.value, + platform_state="retrying" if adapter.fatal_error_retryable else "fatal", + error_code=adapter.fatal_error_code, + error_message=adapter.fatal_error_message, + ) + target = ( + startup_retryable_errors + if adapter.fatal_error_retryable + else startup_nonretryable_errors + ) + target.append(f"{platform.value}: {adapter.fatal_error_message}") + # Queue for reconnection if the error is retryable + if adapter.fatal_error_retryable: + self._failed_platforms[platform] = { + "config": platform_config, + "attempts": 1, + "next_retry": time.monotonic() + 30, + "credential_claim": self._adapter_credential_claim(platform, adapter), + "listener_claim": self._adapter_listener_claim(platform, adapter), + } + else: + self._update_platform_runtime_status( + platform.value, platform_state="retrying", error_code=None, error_message="failed to connect", + ) + startup_retryable_errors.append(f"{platform.value}: failed to connect") + # No fatal error info means likely a transient issue -- queue for retry + self._failed_platforms[platform] = { + "config": platform_config, + "attempts": 1, + "next_retry": time.monotonic() + 30, + "queued_at": time.monotonic(), + "credential_claim": self._adapter_credential_claim(platform, adapter), + "listener_claim": self._adapter_listener_claim(platform, adapter), + } + if await self._abort_startup_if_shutdown_requested(): + return True # Multi-profile multiplexing: bring up adapters for every OTHER profile # this gateway serves. Each profile's adapters connect under that # profile's home + credential scope and stamp their inbound events with diff --git a/tests/gateway/test_startup_connect_parallel.py b/tests/gateway/test_startup_connect_parallel.py new file mode 100644 index 0000000000..0dd4ac08f9 --- /dev/null +++ b/tests/gateway/test_startup_connect_parallel.py @@ -0,0 +1,149 @@ +"""Regression tests for parallel platform connect at gateway startup (#83791). + +The old ``GatewayRunner.start()`` loop awaited each platform's connect() +(including its own timeout) in turn. A single slow/failing platform (e.g. +Telegram behind a dead proxy) therefore delayed every later platform's +connect by a full timeout window, cascading one platform's failure onto +WeChat/QQ/etc. These tests prove the connects now run concurrently. +""" + +import asyncio +import time + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter +from gateway.run import GatewayRunner + + +class _TimingAdapter(BasePlatformAdapter): + """Adapter whose ``connect()`` records start/end wall time and sleeps. + + Used to prove the startup connect loop launches every platform's + connect() concurrently rather than serially. + """ + + _connect_timings: dict = {} + + def __init__(self, platform: Platform, sleep: float): + super().__init__(PlatformConfig(enabled=True, token="***"), platform) + self._sleep = sleep + + async def connect(self, *, is_reconnect: bool = False) -> bool: + start = time.monotonic() + _TimingAdapter._connect_timings[self.platform.value] = (start, None) + await asyncio.sleep(self._sleep) + _start, _ = _TimingAdapter._connect_timings[self.platform.value] + _TimingAdapter._connect_timings[self.platform.value] = (_start, time.monotonic()) + return True + + async def disconnect(self) -> None: + self._mark_disconnected() + + async def send(self, chat_id, content, reply_to=None, metadata=None): + raise NotImplementedError + + async def get_chat_info(self, chat_id): + return {"id": chat_id} + + +@pytest.mark.asyncio +async def test_startup_connects_platforms_concurrently(monkeypatch, tmp_path): + """A slow platform must not block a later platform at startup (#83791). + + "slow" (Telegram) is listed first so a serial loop would fully block + "fast" (Discord). We prove the connect calls overlap: the slow platform's + connect starts before the fast platform's connect finishes. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + _TimingAdapter._connect_timings = {} + + config = GatewayConfig( + platforms={ + Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"), + Platform.DISCORD: PlatformConfig(enabled=True, token="***"), + }, + sessions_dir=tmp_path / "sessions", + ) + runner = GatewayRunner(config) + + def _make_adapter(platform, platform_config): + sleep = 0.3 if platform is Platform.TELEGRAM else 0.0 + return _TimingAdapter(platform, sleep) + + monkeypatch.setattr(runner, "_create_adapter", _make_adapter) + # Keep the rest of startup lightweight / non-fatal. + monkeypatch.setattr(runner, "_start_secondary_profile_adapters", lambda: 0) + + await runner.start() + + timings = _TimingAdapter._connect_timings + assert timings, "no connect() timing was recorded" + slow_start, slow_end = timings[Platform.TELEGRAM.value] + _fast_start, fast_end = timings[Platform.DISCORD.value] + + # Overlap: slow platform began connecting before the fast one finished. + assert slow_start < fast_end, ( + f"connects did not overlap (serial loop?): slow_start={slow_start}, " + f"fast_end={fast_end}" + ) + # Sanity: the slow connect actually ran for ~its sleep duration. + assert (slow_end - slow_start) >= 0.25 + # Both platforms should be registered once startup settles. + assert Platform.TELEGRAM in runner.adapters + assert Platform.DISCORD in runner.adapters + + +@pytest.mark.asyncio +async def test_startup_one_failing_platform_does_not_block_others(monkeypatch, tmp_path): + """A failing/slow platform must not prevent others from connecting (#83791). + + Mirrors the reported Windows symptom: Telegram (dead proxy) must not keep + WeChat/QQ offline. Here Telegram fails (returns False after a sleep) while + Discord connects successfully and is registered. + """ + + class _FailingSlowAdapter(BasePlatformAdapter): + def __init__(self): + super().__init__(PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM) + + async def connect(self, *, is_reconnect: bool = False) -> bool: + await asyncio.sleep(0.3) + self._set_fatal_error("telegram_proxy_dead", "proxy unreachable", retryable=True) + return False + + async def disconnect(self) -> None: + self._mark_disconnected() + + async def send(self, chat_id, content, reply_to=None, metadata=None): + raise NotImplementedError + + async def get_chat_info(self, chat_id): + return {"id": chat_id} + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + config = GatewayConfig( + platforms={ + Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"), + Platform.DISCORD: PlatformConfig(enabled=True, token="***"), + }, + sessions_dir=tmp_path / "sessions", + ) + runner = GatewayRunner(config) + + def _make_adapter(platform, platform_config): + if platform is Platform.TELEGRAM: + return _FailingSlowAdapter() + return _TimingAdapter(platform, 0.0) + + monkeypatch.setattr(runner, "_create_adapter", _make_adapter) + monkeypatch.setattr(runner, "_start_secondary_profile_adapters", lambda: 0) + + await runner.start() + + # The healthy platform connected and is registered despite Telegram failing. + assert Platform.DISCORD in runner.adapters + # The failed platform is queued for retry, not silently dropped. + assert Platform.TELEGRAM in runner._failed_platforms