From 54124979b0c8c87a60a2cb90031ff5abb56c581f Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Mon, 6 Jul 2026 23:26:57 +0800 Subject: [PATCH 1/4] feat: add event loop diagnostics --- astrbot/core/core_lifecycle.py | 10 +- astrbot/core/utils/event_loop_diagnostics.py | 221 +++++++++++++++++++ tests/unit/test_event_loop_diagnostics.py | 109 +++++++++ 3 files changed, 339 insertions(+), 1 deletion(-) create mode 100644 astrbot/core/utils/event_loop_diagnostics.py create mode 100644 tests/unit/test_event_loop_diagnostics.py diff --git a/astrbot/core/core_lifecycle.py b/astrbot/core/core_lifecycle.py index c325a2ea38..215b817544 100644 --- a/astrbot/core/core_lifecycle.py +++ b/astrbot/core/core_lifecycle.py @@ -35,6 +35,9 @@ from astrbot.core.subagent_orchestrator import SubAgentOrchestrator from astrbot.core.umop_config_router import UmopConfigRouter from astrbot.core.updator import AstrBotUpdator +from astrbot.core.utils.event_loop_diagnostics import ( + create_event_loop_diagnostic_tasks, +) from astrbot.core.utils.llm_metadata import update_llm_metadata from astrbot.core.utils.migra_helper import migra from astrbot.core.utils.temp_dir_cleaner import TempDirCleaner @@ -296,13 +299,18 @@ def _load(self) -> None: self.temp_dir_cleaner.run(), name="temp_dir_cleaner", ) + diagnostic_tasks = create_event_loop_diagnostic_tasks() # 把插件中注册的所有协程函数注册到事件总线中并执行 extra_tasks = [] for task in self.star_context._register_tasks: extra_tasks.append(asyncio.create_task(task, name=task.__name__)) # type: ignore - tasks_ = [event_bus_task, *(extra_tasks if extra_tasks else [])] + tasks_ = [ + event_bus_task, + *diagnostic_tasks, + *(extra_tasks if extra_tasks else []), + ] if cron_task: tasks_.append(cron_task) if temp_dir_cleaner_task: diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py new file mode 100644 index 0000000000..487fee0613 --- /dev/null +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -0,0 +1,221 @@ +import asyncio +import faulthandler +import os +import sys +from dataclasses import dataclass +from typing import TextIO + +from astrbot import logger + +LAG_MONITOR_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_LAG_MONITOR" +LAG_MONITOR_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_LAG_INTERVAL" +LAG_MONITOR_THRESHOLD_ENV = "ASTRBOT_EVENT_LOOP_LAG_THRESHOLD" +WATCHDOG_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG" +WATCHDOG_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_INTERVAL" +WATCHDOG_TIMEOUT_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_TIMEOUT" + +DEFAULT_LAG_MONITOR_INTERVAL = 1.0 +DEFAULT_LAG_MONITOR_THRESHOLD = 5.0 +DEFAULT_WATCHDOG_INTERVAL = 1.0 +DEFAULT_WATCHDOG_TIMEOUT = 15.0 + + +@dataclass(frozen=True) +class EventLoopDiagnosticSettings: + """Settings for event loop lag and blockage diagnostics. + + Args: + lag_monitor_enabled: Whether to log event loop scheduling lag. + lag_monitor_interval: Seconds between lag monitor wakeups. + lag_monitor_threshold: Minimum lag seconds before logging a warning. + watchdog_enabled: Whether to arm the faulthandler watchdog. + watchdog_interval: Seconds between faulthandler watchdog refreshes. + watchdog_timeout: Seconds without event loop progress before dumping stacks. + """ + + lag_monitor_enabled: bool + lag_monitor_interval: float + lag_monitor_threshold: float + watchdog_enabled: bool + watchdog_interval: float + watchdog_timeout: float + + +def _env_flag(name: str, default: bool) -> bool: + """Read a boolean flag from the environment. + + Args: + name: Environment variable name. + default: Value to use when the variable is unset or empty. + + Returns: + Parsed boolean value. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_float(name: str, default: float, minimum: float) -> float: + """Read a bounded float from the environment. + + Args: + name: Environment variable name. + default: Value to use when parsing fails or the value is too small. + minimum: Smallest accepted value. + + Returns: + Parsed float value or the default. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + try: + parsed = float(value) + except ValueError: + logger.warning( + "Invalid %s=%r, fallback to %.3fs.", + name, + value, + default, + ) + return default + if parsed < minimum: + logger.warning( + "Invalid %s=%r, expected at least %.3fs; fallback to %.3fs.", + name, + value, + minimum, + default, + ) + return default + return parsed + + +def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: + """Load event loop diagnostic settings from environment variables. + + Returns: + Event loop diagnostic settings. + """ + return EventLoopDiagnosticSettings( + lag_monitor_enabled=_env_flag(LAG_MONITOR_ENABLED_ENV, True), + lag_monitor_interval=_env_float( + LAG_MONITOR_INTERVAL_ENV, + DEFAULT_LAG_MONITOR_INTERVAL, + 0.1, + ), + lag_monitor_threshold=_env_float( + LAG_MONITOR_THRESHOLD_ENV, + DEFAULT_LAG_MONITOR_THRESHOLD, + 0.1, + ), + watchdog_enabled=_env_flag(WATCHDOG_ENABLED_ENV, False), + watchdog_interval=_env_float( + WATCHDOG_INTERVAL_ENV, + DEFAULT_WATCHDOG_INTERVAL, + 0.1, + ), + watchdog_timeout=_env_float( + WATCHDOG_TIMEOUT_ENV, + DEFAULT_WATCHDOG_TIMEOUT, + 1.0, + ), + ) + + +async def monitor_event_loop_lag( + *, + interval: float = DEFAULT_LAG_MONITOR_INTERVAL, + warn_after: float = DEFAULT_LAG_MONITOR_THRESHOLD, +) -> None: + """Log a warning when the event loop wakes significantly later than expected. + + Args: + interval: Seconds between monitor wakeups. + warn_after: Minimum lag seconds before logging a warning. + """ + loop = asyncio.get_running_loop() + expected = loop.time() + interval + while True: + await asyncio.sleep(interval) + now = loop.time() + lag = now - expected + if lag > warn_after: + logger.warning( + "Event loop lag detected: %.3fs (threshold %.3fs).", + lag, + warn_after, + ) + expected = now + interval + + +async def faulthandler_event_loop_watchdog( + *, + timeout: float = DEFAULT_WATCHDOG_TIMEOUT, + interval: float = DEFAULT_WATCHDOG_INTERVAL, + dump_file: TextIO | None = None, +) -> None: + """Dump all thread stacks if the event loop is blocked for too long. + + Args: + timeout: Seconds without watchdog refresh before faulthandler dumps stacks. + interval: Seconds between watchdog refreshes while the event loop is healthy. + dump_file: File object that receives faulthandler output. + """ + output = dump_file or sys.stderr + if not faulthandler.is_enabled(): + faulthandler.enable(file=output) + + try: + while True: + faulthandler.cancel_dump_traceback_later() + faulthandler.dump_traceback_later( + timeout, + repeat=False, + file=output, + ) + await asyncio.sleep(interval) + finally: + faulthandler.cancel_dump_traceback_later() + + +def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]: + """Create enabled event loop diagnostic tasks for the current loop. + + Returns: + A list of created asyncio tasks. + """ + settings = load_event_loop_diagnostic_settings() + tasks: list[asyncio.Task] = [] + + if settings.lag_monitor_enabled: + tasks.append( + asyncio.create_task( + monitor_event_loop_lag( + interval=settings.lag_monitor_interval, + warn_after=settings.lag_monitor_threshold, + ), + name="event_loop_lag_monitor", + ) + ) + + if settings.watchdog_enabled: + logger.warning( + "Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. " + "If the loop is blocked, Python thread stacks will be written to stderr.", + settings.watchdog_timeout, + settings.watchdog_interval, + ) + tasks.append( + asyncio.create_task( + faulthandler_event_loop_watchdog( + timeout=settings.watchdog_timeout, + interval=settings.watchdog_interval, + ), + name="event_loop_faulthandler_watchdog", + ) + ) + + return tasks diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py new file mode 100644 index 0000000000..a1e4d71440 --- /dev/null +++ b/tests/unit/test_event_loop_diagnostics.py @@ -0,0 +1,109 @@ +import asyncio + +import pytest + +from astrbot.core.utils import event_loop_diagnostics as diagnostics + + +def _clear_diagnostic_env(monkeypatch): + """Clear event loop diagnostic environment variables. + + Args: + monkeypatch: Pytest monkeypatch fixture. + """ + for name in ( + diagnostics.LAG_MONITOR_ENABLED_ENV, + diagnostics.LAG_MONITOR_INTERVAL_ENV, + diagnostics.LAG_MONITOR_THRESHOLD_ENV, + diagnostics.WATCHDOG_ENABLED_ENV, + diagnostics.WATCHDOG_INTERVAL_ENV, + diagnostics.WATCHDOG_TIMEOUT_ENV, + ): + monkeypatch.delenv(name, raising=False) + + +def test_load_event_loop_diagnostic_settings_defaults(monkeypatch): + """Default settings enable low-overhead lag monitoring only.""" + _clear_diagnostic_env(monkeypatch) + + settings = diagnostics.load_event_loop_diagnostic_settings() + + assert settings.lag_monitor_enabled is True + assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL + assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD + assert settings.watchdog_enabled is False + + +def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): + """Environment variables override event loop diagnostic settings.""" + _clear_diagnostic_env(monkeypatch) + monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "0") + monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "yes") + monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "30") + + settings = diagnostics.load_event_loop_diagnostic_settings() + + assert settings.lag_monitor_enabled is False + assert settings.watchdog_enabled is True + assert settings.watchdog_timeout == 30 + + +@pytest.mark.asyncio +async def test_create_event_loop_diagnostic_tasks_respects_env(monkeypatch): + """Disabled lag monitor and watchdog should create no tasks.""" + _clear_diagnostic_env(monkeypatch) + monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "false") + monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "false") + + tasks = diagnostics.create_event_loop_diagnostic_tasks() + + assert tasks == [] + + +@pytest.mark.asyncio +async def test_create_event_loop_diagnostic_tasks_default_lag_monitor(monkeypatch): + """Default diagnostics should create only the lag monitor task.""" + _clear_diagnostic_env(monkeypatch) + + tasks = diagnostics.create_event_loop_diagnostic_tasks() + + try: + assert [task.get_name() for task in tasks] == ["event_loop_lag_monitor"] + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch): + """The faulthandler watchdog should cancel its pending dump on shutdown.""" + calls = [] + + class FakeFaultHandler: + def is_enabled(self): + calls.append("is_enabled") + return False + + def enable(self, file): + calls.append(("enable", file)) + + def cancel_dump_traceback_later(self): + calls.append("cancel") + + def dump_traceback_later(self, timeout, repeat, file): + calls.append(("dump", timeout, repeat, file)) + + fake_faulthandler = FakeFaultHandler() + monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler) + + task = asyncio.create_task( + diagnostics.faulthandler_event_loop_watchdog(timeout=10, interval=1) + ) + await asyncio.sleep(0) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert "is_enabled" in calls + assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) + assert calls[-1] == "cancel" From f5b08eb135293e42cf929074d75d917d1d206c07 Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Mon, 6 Jul 2026 23:59:43 +0800 Subject: [PATCH 2/4] fix: write event loop watchdog dumps to rotating log --- astrbot/core/utils/event_loop_diagnostics.py | 138 +++++++++++++++++-- tests/unit/test_event_loop_diagnostics.py | 54 ++++++-- 2 files changed, 172 insertions(+), 20 deletions(-) diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py index 487fee0613..d11df16f8b 100644 --- a/astrbot/core/utils/event_loop_diagnostics.py +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -1,11 +1,12 @@ import asyncio import faulthandler import os -import sys from dataclasses import dataclass +from pathlib import Path from typing import TextIO from astrbot import logger +from astrbot.core.utils.astrbot_path import get_astrbot_data_path LAG_MONITOR_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_LAG_MONITOR" LAG_MONITOR_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_LAG_INTERVAL" @@ -13,11 +14,15 @@ WATCHDOG_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG" WATCHDOG_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_INTERVAL" WATCHDOG_TIMEOUT_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_TIMEOUT" +WATCHDOG_LOG_PATH_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_PATH" +WATCHDOG_LOG_MAX_BYTES_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_MAX_BYTES" DEFAULT_LAG_MONITOR_INTERVAL = 1.0 DEFAULT_LAG_MONITOR_THRESHOLD = 5.0 DEFAULT_WATCHDOG_INTERVAL = 1.0 DEFAULT_WATCHDOG_TIMEOUT = 15.0 +DEFAULT_WATCHDOG_LOG_RELATIVE_PATH = Path("logs") / "event_loop_watchdog.log" +DEFAULT_WATCHDOG_LOG_MAX_BYTES = 1024 * 1024 @dataclass(frozen=True) @@ -31,6 +36,8 @@ class EventLoopDiagnosticSettings: watchdog_enabled: Whether to arm the faulthandler watchdog. watchdog_interval: Seconds between faulthandler watchdog refreshes. watchdog_timeout: Seconds without event loop progress before dumping stacks. + watchdog_log_path: File that receives faulthandler watchdog output. + watchdog_log_max_bytes: Maximum watchdog log bytes before rotation. """ lag_monitor_enabled: bool @@ -39,6 +46,8 @@ class EventLoopDiagnosticSettings: watchdog_enabled: bool watchdog_interval: float watchdog_timeout: float + watchdog_log_path: Path + watchdog_log_max_bytes: int def _env_flag(name: str, default: bool) -> bool: @@ -93,6 +102,59 @@ def _env_float(name: str, default: float, minimum: float) -> float: return parsed +def _env_int(name: str, default: int, minimum: int) -> int: + """Read a bounded integer from the environment. + + Args: + name: Environment variable name. + default: Value to use when parsing fails or the value is too small. + minimum: Smallest accepted value. + + Returns: + Parsed integer value or the default. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + try: + parsed = int(value) + except ValueError: + logger.warning( + "Invalid %s=%r, fallback to %d.", + name, + value, + default, + ) + return default + if parsed < minimum: + logger.warning( + "Invalid %s=%r, expected at least %d; fallback to %d.", + name, + value, + minimum, + default, + ) + return default + return parsed + + +def _watchdog_log_path() -> Path: + """Resolve the watchdog stack dump log path. + + Returns: + Absolute path for watchdog stack dump output. + """ + configured = os.environ.get(WATCHDOG_LOG_PATH_ENV, "").strip() + path = ( + Path(configured).expanduser() + if configured + else DEFAULT_WATCHDOG_LOG_RELATIVE_PATH + ) + if not path.is_absolute(): + path = Path(get_astrbot_data_path()) / path + return path + + def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: """Load event loop diagnostic settings from environment variables. @@ -122,6 +184,12 @@ def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: DEFAULT_WATCHDOG_TIMEOUT, 1.0, ), + watchdog_log_path=_watchdog_log_path(), + watchdog_log_max_bytes=_env_int( + WATCHDOG_LOG_MAX_BYTES_ENV, + DEFAULT_WATCHDOG_LOG_MAX_BYTES, + 1024, + ), ) @@ -151,11 +219,46 @@ async def monitor_event_loop_lag( expected = now + interval +def _rotate_watchdog_log_file(log_path: Path, max_bytes: int) -> None: + """Rotate the watchdog log when it reaches the configured size limit. + + Args: + log_path: Current watchdog log path. + max_bytes: Maximum current log size before rotation. + """ + try: + if not log_path.exists() or log_path.stat().st_size < max_bytes: + return + rotated_path = log_path.with_name(f"{log_path.name}.1") + if rotated_path.exists(): + rotated_path.unlink() + log_path.replace(rotated_path) + except OSError as e: + logger.warning("Failed to rotate event loop watchdog log %s: %s", log_path, e) + + +def _open_watchdog_log_file(log_path: Path, max_bytes: int) -> TextIO: + """Open the watchdog log file after applying size-based rotation. + + Args: + log_path: Current watchdog log path. + max_bytes: Maximum current log size before rotation. + + Returns: + Writable text file object for faulthandler output. + """ + log_path.parent.mkdir(parents=True, exist_ok=True) + _rotate_watchdog_log_file(log_path, max_bytes) + return log_path.open("a", encoding="utf-8") + + async def faulthandler_event_loop_watchdog( *, timeout: float = DEFAULT_WATCHDOG_TIMEOUT, interval: float = DEFAULT_WATCHDOG_INTERVAL, dump_file: TextIO | None = None, + dump_path: Path | None = None, + max_bytes: int = DEFAULT_WATCHDOG_LOG_MAX_BYTES, ) -> None: """Dump all thread stacks if the event loop is blocked for too long. @@ -163,20 +266,26 @@ async def faulthandler_event_loop_watchdog( timeout: Seconds without watchdog refresh before faulthandler dumps stacks. interval: Seconds between watchdog refreshes while the event loop is healthy. dump_file: File object that receives faulthandler output. + dump_path: Path that receives faulthandler output when dump_file is unset. + max_bytes: Maximum current log size before rotation. """ - output = dump_file or sys.stderr - if not faulthandler.is_enabled(): - faulthandler.enable(file=output) - + log_path = dump_path or _watchdog_log_path() try: while True: faulthandler.cancel_dump_traceback_later() - faulthandler.dump_traceback_later( - timeout, - repeat=False, - file=output, - ) - await asyncio.sleep(interval) + output = dump_file or _open_watchdog_log_file(log_path, max_bytes) + should_close = dump_file is None + try: + faulthandler.dump_traceback_later( + timeout, + repeat=False, + file=output, + ) + await asyncio.sleep(interval) + finally: + faulthandler.cancel_dump_traceback_later() + if should_close: + output.close() finally: faulthandler.cancel_dump_traceback_later() @@ -204,15 +313,20 @@ def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]: if settings.watchdog_enabled: logger.warning( "Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. " - "If the loop is blocked, Python thread stacks will be written to stderr.", + "If the loop is blocked, Python thread stacks will be written to %s " + "(rotates at %d bytes).", settings.watchdog_timeout, settings.watchdog_interval, + settings.watchdog_log_path, + settings.watchdog_log_max_bytes, ) tasks.append( asyncio.create_task( faulthandler_event_loop_watchdog( timeout=settings.watchdog_timeout, interval=settings.watchdog_interval, + dump_path=settings.watchdog_log_path, + max_bytes=settings.watchdog_log_max_bytes, ), name="event_loop_faulthandler_watchdog", ) diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py index a1e4d71440..20af5d425d 100644 --- a/tests/unit/test_event_loop_diagnostics.py +++ b/tests/unit/test_event_loop_diagnostics.py @@ -18,6 +18,8 @@ def _clear_diagnostic_env(monkeypatch): diagnostics.WATCHDOG_ENABLED_ENV, diagnostics.WATCHDOG_INTERVAL_ENV, diagnostics.WATCHDOG_TIMEOUT_ENV, + diagnostics.WATCHDOG_LOG_PATH_ENV, + diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, ): monkeypatch.delenv(name, raising=False) @@ -32,6 +34,7 @@ def test_load_event_loop_diagnostic_settings_defaults(monkeypatch): assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD assert settings.watchdog_enabled is False + assert settings.watchdog_log_max_bytes == diagnostics.DEFAULT_WATCHDOG_LOG_MAX_BYTES def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): @@ -40,12 +43,16 @@ def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "0") monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "yes") monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "30") + monkeypatch.setenv(diagnostics.WATCHDOG_LOG_PATH_ENV, "/tmp/astrbot-watchdog.log") + monkeypatch.setenv(diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, "2048") settings = diagnostics.load_event_loop_diagnostic_settings() assert settings.lag_monitor_enabled is False assert settings.watchdog_enabled is True assert settings.watchdog_timeout == 30 + assert settings.watchdog_log_path.as_posix() == "/tmp/astrbot-watchdog.log" + assert settings.watchdog_log_max_bytes == 2048 @pytest.mark.asyncio @@ -81,13 +88,6 @@ async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch): calls = [] class FakeFaultHandler: - def is_enabled(self): - calls.append("is_enabled") - return False - - def enable(self, file): - calls.append(("enable", file)) - def cancel_dump_traceback_later(self): calls.append("cancel") @@ -104,6 +104,44 @@ def dump_traceback_later(self, timeout, repeat, file): task.cancel() await asyncio.gather(task, return_exceptions=True) - assert "is_enabled" in calls assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) assert calls[-1] == "cancel" + + +@pytest.mark.asyncio +async def test_faulthandler_watchdog_writes_rotating_log(tmp_path, monkeypatch): + """The faulthandler watchdog should write to and rotate its log file.""" + log_path = tmp_path / "logs" / "event_loop_watchdog.log" + log_path.parent.mkdir() + log_path.write_text("x" * 8, encoding="utf-8") + calls = [] + + class FakeFaultHandler: + def cancel_dump_traceback_later(self): + calls.append("cancel") + + def dump_traceback_later(self, timeout, repeat, file): + calls.append(("dump", timeout, repeat, file.name)) + file.write("watchdog dump\n") + file.flush() + + fake_faulthandler = FakeFaultHandler() + monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler) + + task = asyncio.create_task( + diagnostics.faulthandler_event_loop_watchdog( + timeout=10, + interval=1, + dump_path=log_path, + max_bytes=4, + ) + ) + await asyncio.sleep(0) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert log_path.read_text(encoding="utf-8") == "watchdog dump\n" + assert log_path.with_name("event_loop_watchdog.log.1").read_text( + encoding="utf-8" + ) == "x" * 8 + assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) From f683adc74cfebd9d30196fbbfe4a666275fe35d1 Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Tue, 7 Jul 2026 00:03:39 +0800 Subject: [PATCH 3/4] chore: use fixed event loop diagnostic defaults --- astrbot/core/utils/event_loop_diagnostics.py | 154 ++----------------- tests/unit/test_event_loop_diagnostics.py | 70 ++------- 2 files changed, 26 insertions(+), 198 deletions(-) diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py index d11df16f8b..47f51c9718 100644 --- a/astrbot/core/utils/event_loop_diagnostics.py +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -1,6 +1,5 @@ import asyncio import faulthandler -import os from dataclasses import dataclass from pathlib import Path from typing import TextIO @@ -8,19 +7,12 @@ from astrbot import logger from astrbot.core.utils.astrbot_path import get_astrbot_data_path -LAG_MONITOR_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_LAG_MONITOR" -LAG_MONITOR_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_LAG_INTERVAL" -LAG_MONITOR_THRESHOLD_ENV = "ASTRBOT_EVENT_LOOP_LAG_THRESHOLD" -WATCHDOG_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG" -WATCHDOG_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_INTERVAL" -WATCHDOG_TIMEOUT_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_TIMEOUT" -WATCHDOG_LOG_PATH_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_PATH" -WATCHDOG_LOG_MAX_BYTES_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_MAX_BYTES" - -DEFAULT_LAG_MONITOR_INTERVAL = 1.0 -DEFAULT_LAG_MONITOR_THRESHOLD = 5.0 -DEFAULT_WATCHDOG_INTERVAL = 1.0 -DEFAULT_WATCHDOG_TIMEOUT = 15.0 +DEFAULT_LAG_MONITOR_ENABLED = True +DEFAULT_LAG_MONITOR_INTERVAL = 5.0 +DEFAULT_LAG_MONITOR_THRESHOLD = 15.0 +DEFAULT_WATCHDOG_ENABLED = True +DEFAULT_WATCHDOG_INTERVAL = 5.0 +DEFAULT_WATCHDOG_TIMEOUT = 30.0 DEFAULT_WATCHDOG_LOG_RELATIVE_PATH = Path("logs") / "event_loop_watchdog.log" DEFAULT_WATCHDOG_LOG_MAX_BYTES = 1024 * 1024 @@ -50,146 +42,30 @@ class EventLoopDiagnosticSettings: watchdog_log_max_bytes: int -def _env_flag(name: str, default: bool) -> bool: - """Read a boolean flag from the environment. - - Args: - name: Environment variable name. - default: Value to use when the variable is unset or empty. - - Returns: - Parsed boolean value. - """ - value = os.environ.get(name) - if value is None or value.strip() == "": - return default - return value.strip().lower() in {"1", "true", "yes", "on"} - - -def _env_float(name: str, default: float, minimum: float) -> float: - """Read a bounded float from the environment. - - Args: - name: Environment variable name. - default: Value to use when parsing fails or the value is too small. - minimum: Smallest accepted value. - - Returns: - Parsed float value or the default. - """ - value = os.environ.get(name) - if value is None or value.strip() == "": - return default - try: - parsed = float(value) - except ValueError: - logger.warning( - "Invalid %s=%r, fallback to %.3fs.", - name, - value, - default, - ) - return default - if parsed < minimum: - logger.warning( - "Invalid %s=%r, expected at least %.3fs; fallback to %.3fs.", - name, - value, - minimum, - default, - ) - return default - return parsed - - -def _env_int(name: str, default: int, minimum: int) -> int: - """Read a bounded integer from the environment. - - Args: - name: Environment variable name. - default: Value to use when parsing fails or the value is too small. - minimum: Smallest accepted value. - - Returns: - Parsed integer value or the default. - """ - value = os.environ.get(name) - if value is None or value.strip() == "": - return default - try: - parsed = int(value) - except ValueError: - logger.warning( - "Invalid %s=%r, fallback to %d.", - name, - value, - default, - ) - return default - if parsed < minimum: - logger.warning( - "Invalid %s=%r, expected at least %d; fallback to %d.", - name, - value, - minimum, - default, - ) - return default - return parsed - - def _watchdog_log_path() -> Path: """Resolve the watchdog stack dump log path. Returns: Absolute path for watchdog stack dump output. """ - configured = os.environ.get(WATCHDOG_LOG_PATH_ENV, "").strip() - path = ( - Path(configured).expanduser() - if configured - else DEFAULT_WATCHDOG_LOG_RELATIVE_PATH - ) - if not path.is_absolute(): - path = Path(get_astrbot_data_path()) / path - return path + return Path(get_astrbot_data_path()) / DEFAULT_WATCHDOG_LOG_RELATIVE_PATH def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: - """Load event loop diagnostic settings from environment variables. + """Load fixed event loop diagnostic settings. Returns: Event loop diagnostic settings. """ return EventLoopDiagnosticSettings( - lag_monitor_enabled=_env_flag(LAG_MONITOR_ENABLED_ENV, True), - lag_monitor_interval=_env_float( - LAG_MONITOR_INTERVAL_ENV, - DEFAULT_LAG_MONITOR_INTERVAL, - 0.1, - ), - lag_monitor_threshold=_env_float( - LAG_MONITOR_THRESHOLD_ENV, - DEFAULT_LAG_MONITOR_THRESHOLD, - 0.1, - ), - watchdog_enabled=_env_flag(WATCHDOG_ENABLED_ENV, False), - watchdog_interval=_env_float( - WATCHDOG_INTERVAL_ENV, - DEFAULT_WATCHDOG_INTERVAL, - 0.1, - ), - watchdog_timeout=_env_float( - WATCHDOG_TIMEOUT_ENV, - DEFAULT_WATCHDOG_TIMEOUT, - 1.0, - ), + lag_monitor_enabled=DEFAULT_LAG_MONITOR_ENABLED, + lag_monitor_interval=DEFAULT_LAG_MONITOR_INTERVAL, + lag_monitor_threshold=DEFAULT_LAG_MONITOR_THRESHOLD, + watchdog_enabled=DEFAULT_WATCHDOG_ENABLED, + watchdog_interval=DEFAULT_WATCHDOG_INTERVAL, + watchdog_timeout=DEFAULT_WATCHDOG_TIMEOUT, watchdog_log_path=_watchdog_log_path(), - watchdog_log_max_bytes=_env_int( - WATCHDOG_LOG_MAX_BYTES_ENV, - DEFAULT_WATCHDOG_LOG_MAX_BYTES, - 1024, - ), + watchdog_log_max_bytes=DEFAULT_WATCHDOG_LOG_MAX_BYTES, ) diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py index 20af5d425d..73275c5e98 100644 --- a/tests/unit/test_event_loop_diagnostics.py +++ b/tests/unit/test_event_loop_diagnostics.py @@ -5,77 +5,29 @@ from astrbot.core.utils import event_loop_diagnostics as diagnostics -def _clear_diagnostic_env(monkeypatch): - """Clear event loop diagnostic environment variables. - - Args: - monkeypatch: Pytest monkeypatch fixture. - """ - for name in ( - diagnostics.LAG_MONITOR_ENABLED_ENV, - diagnostics.LAG_MONITOR_INTERVAL_ENV, - diagnostics.LAG_MONITOR_THRESHOLD_ENV, - diagnostics.WATCHDOG_ENABLED_ENV, - diagnostics.WATCHDOG_INTERVAL_ENV, - diagnostics.WATCHDOG_TIMEOUT_ENV, - diagnostics.WATCHDOG_LOG_PATH_ENV, - diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, - ): - monkeypatch.delenv(name, raising=False) - - -def test_load_event_loop_diagnostic_settings_defaults(monkeypatch): - """Default settings enable low-overhead lag monitoring only.""" - _clear_diagnostic_env(monkeypatch) - +def test_load_event_loop_diagnostic_settings_defaults(): + """Default settings enable lag monitoring and stack dump watchdog.""" settings = diagnostics.load_event_loop_diagnostic_settings() assert settings.lag_monitor_enabled is True assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD - assert settings.watchdog_enabled is False - assert settings.watchdog_log_max_bytes == diagnostics.DEFAULT_WATCHDOG_LOG_MAX_BYTES - - -def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): - """Environment variables override event loop diagnostic settings.""" - _clear_diagnostic_env(monkeypatch) - monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "0") - monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "yes") - monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "30") - monkeypatch.setenv(diagnostics.WATCHDOG_LOG_PATH_ENV, "/tmp/astrbot-watchdog.log") - monkeypatch.setenv(diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, "2048") - - settings = diagnostics.load_event_loop_diagnostic_settings() - - assert settings.lag_monitor_enabled is False assert settings.watchdog_enabled is True - assert settings.watchdog_timeout == 30 - assert settings.watchdog_log_path.as_posix() == "/tmp/astrbot-watchdog.log" - assert settings.watchdog_log_max_bytes == 2048 - - -@pytest.mark.asyncio -async def test_create_event_loop_diagnostic_tasks_respects_env(monkeypatch): - """Disabled lag monitor and watchdog should create no tasks.""" - _clear_diagnostic_env(monkeypatch) - monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "false") - monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "false") - - tasks = diagnostics.create_event_loop_diagnostic_tasks() - - assert tasks == [] + assert settings.watchdog_interval == diagnostics.DEFAULT_WATCHDOG_INTERVAL + assert settings.watchdog_timeout == diagnostics.DEFAULT_WATCHDOG_TIMEOUT + assert settings.watchdog_log_max_bytes == diagnostics.DEFAULT_WATCHDOG_LOG_MAX_BYTES @pytest.mark.asyncio -async def test_create_event_loop_diagnostic_tasks_default_lag_monitor(monkeypatch): - """Default diagnostics should create only the lag monitor task.""" - _clear_diagnostic_env(monkeypatch) - +async def test_create_event_loop_diagnostic_tasks_defaults(): + """Default diagnostics should create both event loop diagnostic tasks.""" tasks = diagnostics.create_event_loop_diagnostic_tasks() try: - assert [task.get_name() for task in tasks] == ["event_loop_lag_monitor"] + assert [task.get_name() for task in tasks] == [ + "event_loop_lag_monitor", + "event_loop_faulthandler_watchdog", + ] finally: for task in tasks: task.cancel() From a18054c8b49227b782a41cc83219d248fdb67720 Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Tue, 7 Jul 2026 00:12:05 +0800 Subject: [PATCH 4/4] fix: keep event loop watchdog alive after dump failures --- astrbot/core/utils/event_loop_diagnostics.py | 11 ++++-- tests/unit/test_event_loop_diagnostics.py | 35 ++++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py index 47f51c9718..50eabea5df 100644 --- a/astrbot/core/utils/event_loop_diagnostics.py +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -149,18 +149,23 @@ async def faulthandler_event_loop_watchdog( try: while True: faulthandler.cancel_dump_traceback_later() - output = dump_file or _open_watchdog_log_file(log_path, max_bytes) - should_close = dump_file is None + output: TextIO | None = None + should_close = False try: + output = dump_file or _open_watchdog_log_file(log_path, max_bytes) + should_close = dump_file is None faulthandler.dump_traceback_later( timeout, repeat=False, file=output, ) await asyncio.sleep(interval) + except Exception as e: + logger.warning("Event loop faulthandler watchdog failed: %s", e) + await asyncio.sleep(interval) finally: faulthandler.cancel_dump_traceback_later() - if should_close: + if should_close and output is not None: output.close() finally: faulthandler.cancel_dump_traceback_later() diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py index 73275c5e98..0737478f9e 100644 --- a/tests/unit/test_event_loop_diagnostics.py +++ b/tests/unit/test_event_loop_diagnostics.py @@ -97,3 +97,38 @@ def dump_traceback_later(self, timeout, repeat, file): encoding="utf-8" ) == "x" * 8 assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) + + +@pytest.mark.asyncio +async def test_faulthandler_watchdog_survives_dump_failure(tmp_path, monkeypatch): + """The watchdog should keep running after faulthandler arm failures.""" + log_path = tmp_path / "event_loop_watchdog.log" + armed_again = asyncio.Event() + calls = [] + + class FakeFaultHandler: + def cancel_dump_traceback_later(self): + calls.append("cancel") + + def dump_traceback_later(self, timeout, repeat, file): + calls.append(("dump", timeout, repeat, file.name)) + if len([call for call in calls if isinstance(call, tuple)]) == 1: + raise RuntimeError("boom") + armed_again.set() + + fake_faulthandler = FakeFaultHandler() + monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler) + + task = asyncio.create_task( + diagnostics.faulthandler_event_loop_watchdog( + timeout=10, + interval=0.01, + dump_path=log_path, + ) + ) + await asyncio.wait_for(armed_again.wait(), timeout=1) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + dump_calls = [call for call in calls if isinstance(call, tuple)] + assert len(dump_calls) >= 2