fix(gateway): connect messaging platforms in parallel at startup (#83791)
GatewayRunner.start() previously awaited each platform's connect() (with its own timeout) in a serial for-loop. A single slow/failing platform (e.g. Telegram behind a dead proxy) delayed every later platform's connect by a full timeout window, cascading one platform's failure onto WeChat/QQ/etc. Now the slow connect() calls run concurrently via asyncio.gather while the serial pre-filter (checks, adapter creation, handler wiring) and the single-threaded result aggregation (shared-state mutation, error handling) are unchanged. A failing platform no longer blocks the others. Adds regression tests proving connect() calls overlap and that one failing platform leaves the others connected.
This commit is contained in:
EvanProgramming
+121
-117
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user