diff --git a/esphome/async_thread.py b/esphome/async_thread.py index 3972d735f5..3296d65af6 100644 --- a/esphome/async_thread.py +++ b/esphome/async_thread.py @@ -11,43 +11,136 @@ from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable +from itertools import count +import logging import threading +from typing import cast + +_LOGGER = logging.getLogger(__name__) + +# How long the orphan watcher waits for an abandoned coroutine before giving +# up, so a hung operation does not park a watcher thread forever. +ORPHAN_WAIT_TIMEOUT = 300.0 + + +_runner_ids = count(1) + + +class AsyncDispatchTimeout(TimeoutError): + """The caller stopped waiting; the coroutine was abandoned. + + A subclass so callers can tell the dispatcher's own expiry apart from a + ``TimeoutError`` raised inside the coroutine, while existing + ``except TimeoutError`` handlers keep working. + """ class AsyncThreadRunner[T](threading.Thread): """Run an async coroutine in a daemon thread and expose its result. - The runner catches all exceptions from the coroutine and stores them in - ``exception`` so ``event`` is always set — this prevents callers waiting - on ``event`` from hanging forever when the coroutine crashes. - - Typical usage:: - - runner = AsyncThreadRunner(lambda: my_coro(arg)) - runner.start() - if not runner.event.wait(timeout=5.0): - ... # timed out - if runner.exception is not None: - raise runner.exception - result = runner.result + ``event`` is always set, even when the coroutine crashes, so waiters + never hang; ``completed`` distinguishes a delivered result (even a + legitimate ``None``) from a coroutine that never finished. Prefer + :func:`run_async`; use this class directly only when a failure should + degrade to a default value instead of raising. """ def __init__(self, coro_factory: Callable[[], Awaitable[T]]) -> None: - super().__init__(daemon=True) + super().__init__(daemon=True, name=f"async-thread-runner-{next(_runner_ids)}") self._coro_factory = coro_factory self.result: T | None = None self.exception: BaseException | None = None + self.completed = False self.event = threading.Event() async def _runner(self) -> None: try: self.result = await self._coro_factory() - except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except - # Capture all exceptions so ``event`` is always set — otherwise a - # crash would hang the waiter forever. + # Distinguishes a delivered result from "never ran", since None + # is a valid result value. + self.completed = True + except BaseException as exc: # noqa: BLE001 # pylint: disable=broad-except + # Capture everything, including BaseException — otherwise a + # cancellation or SystemExit would leave ``exception`` unset and + # waiters would mistake the empty ``result`` for success. self.exception = exc finally: self.event.set() def run(self) -> None: - asyncio.run(self._runner()) + try: + asyncio.run(self._runner()) + except BaseException as exc: # noqa: BLE001 # pylint: disable=broad-except + # asyncio.run itself can fail before _runner executes (e.g. loop + # creation under fd exhaustion); record it so waiters never hang. + # A failure during loop cleanup after the coroutine completed + # must not clobber the delivered result, hence the guard. + if self.exception is None and not self.completed: + self.exception = exc + else: + _LOGGER.debug( + "Event loop teardown failed after outcome recorded", + exc_info=True, + ) + finally: + self.event.set() + + +def run_async[T]( + coro_factory: Callable[[], Awaitable[T]], + timeout: float | None = None, + on_orphan: Callable[[T], None] | None = None, +) -> T: + """Run a coroutine in a daemon-thread event loop and return its result. + + Raises :class:`AsyncDispatchTimeout` if the coroutine does not finish + within ``timeout`` seconds; the thread is abandoned and exits with the + interpreter. If the abandoned coroutine later produces a result, + ``on_orphan`` (if given) is called with it so resources such as a + connected socket can be released; delivery is best effort and bounded + by ``ORPHAN_WAIT_TIMEOUT``. + """ + runner: AsyncThreadRunner[T] = AsyncThreadRunner(coro_factory) + runner.start() + if not runner.event.wait(timeout): + + def _cleanup() -> None: + if not runner.event.wait(ORPHAN_WAIT_TIMEOUT): + # The one state where a resource can genuinely leak; leave + # a trace so a recurring hang is attributable. + _LOGGER.info( + "Orphan watcher gave up after %.0fs; a late result may leak", + ORPHAN_WAIT_TIMEOUT, + ) + return + if not runner.completed: + # The only place an abandoned thread's real error surfaces; + # without it a late failure hides behind the TimeoutError. + # INFO, not DEBUG: it fires at most once per abandoned + # operation and the cause may not reproduce on a rerun. + _LOGGER.info( + "Abandoned async operation failed", + exc_info=runner.exception, + ) + return + if (result := runner.result) is None: + return + if on_orphan is None: + _LOGGER.debug("Discarding late result; no on_orphan handler") + return + try: + on_orphan(result) + except Exception: # pylint: disable=broad-except + # INFO, not DEBUG: a failed release means a real leak, and + # it fires at most once per abandoned operation. + _LOGGER.info("Error releasing orphaned result", exc_info=True) + + threading.Thread( + target=_cleanup, daemon=True, name="async-orphan-cleanup" + ).start() + raise AsyncDispatchTimeout("Timed out waiting for async operation") + if (exc := runner.exception) is not None: + raise exc + if not runner.completed: + raise RuntimeError("Async operation finished without a result or an exception") + return cast("T", runner.result) diff --git a/esphome/resolver.py b/esphome/resolver.py index f80a910afe..68bf37eecd 100644 --- a/esphome/resolver.py +++ b/esphome/resolver.py @@ -8,7 +8,7 @@ import os from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError import aioesphomeapi.host_resolver as hr -from esphome.async_thread import AsyncThreadRunner +from esphome.async_thread import AsyncDispatchTimeout, run_async from esphome.core import EsphomeError _LOGGER = logging.getLogger(__name__) @@ -31,9 +31,9 @@ class AsyncResolver: This resolver uses aioesphomeapi's async_resolve_host to handle DNS resolution, including proper .local domain fallback. Running in a thread - (via :class:`AsyncThreadRunner`) allows us to get the result immediately - without waiting for ``asyncio.run()`` to complete its cleanup cycle, which - can take significant time. + (via :func:`run_async`) allows us to get the result immediately without + waiting for ``asyncio.run()`` to complete its cleanup cycle, which can + take significant time. """ def __init__(self, hosts: list[str], port: int) -> None: @@ -48,21 +48,13 @@ class AsyncResolver: ) def resolve(self) -> list[hr.AddrInfo]: - """Start the thread and wait for the result.""" - runner: AsyncThreadRunner[list[hr.AddrInfo]] = AsyncThreadRunner(self._resolve) - runner.start() - - if not runner.event.wait( - timeout=RESOLVE_TIMEOUT + 1.0 - ): # Give it 1 second more than the resolver timeout - raise EsphomeError("Timeout resolving IP address") - - if exc := runner.exception: - if isinstance(exc, ResolveTimeoutAPIError): - raise EsphomeError(f"Timeout resolving IP address: {exc}") from exc - if isinstance(exc, ResolveAPIError): - raise EsphomeError(f"Error resolving IP address: {exc}") from exc - raise exc - - assert runner.result is not None # guaranteed when event set and no exception - return runner.result + """Resolve and wait for the result.""" + try: + # Give it 1 second more than the resolver timeout + return run_async(self._resolve, timeout=RESOLVE_TIMEOUT + 1.0) + except ResolveTimeoutAPIError as exc: + raise EsphomeError(f"Timeout resolving IP address: {exc}") from exc + except ResolveAPIError as exc: + raise EsphomeError(f"Error resolving IP address: {exc}") from exc + except AsyncDispatchTimeout as exc: + raise EsphomeError("Timeout resolving IP address") from exc diff --git a/tests/unit_tests/test_async_thread.py b/tests/unit_tests/test_async_thread.py new file mode 100644 index 0000000000..a64be2f7c7 --- /dev/null +++ b/tests/unit_tests/test_async_thread.py @@ -0,0 +1,316 @@ +"""Tests for the async thread helpers.""" + +from __future__ import annotations + +import asyncio +import threading +from typing import Any +from unittest.mock import patch + +import pytest + +from esphome.async_thread import AsyncDispatchTimeout, AsyncThreadRunner, run_async + + +def _cleanup_threads() -> set[threading.Thread]: + """Return the currently live orphan-cleanup threads.""" + return {t for t in threading.enumerate() if t.name == "async-orphan-cleanup"} + + +def _join_new_cleanup_threads(before: set[threading.Thread]) -> None: + """Wait for cleanup threads spawned since ``before`` to finish.""" + for thread in _cleanup_threads() - before: + thread.join(5) + assert not thread.is_alive() + + +def test_run_async_returns_result() -> None: + """The coroutine's result is returned to the sync caller.""" + + async def coro() -> int: + await asyncio.sleep(0) + return 42 + + assert run_async(coro) == 42 + + +def test_run_async_propagates_exception() -> None: + """Exceptions raised by the coroutine surface in the caller.""" + + async def coro() -> None: + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + run_async(coro) + + +def test_run_async_propagates_base_exception() -> None: + """A BaseException from the coroutine surfaces instead of a None result.""" + + class Boom(BaseException): + pass + + async def coro() -> None: + raise Boom + + with pytest.raises(Boom): + run_async(coro) + + +def test_run_async_timeout() -> None: + """A coroutine that does not finish in time raises TimeoutError.""" + release = threading.Event() + + async def coro() -> None: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + + before = _cleanup_threads() + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.05) + + # Unblock the abandoned runner so its cleanup thread exits promptly. + release.set() + _join_new_cleanup_threads(before) + + +def test_run_async_surfaces_loop_startup_failure() -> None: + """A failure before the coroutine runs raises instead of hanging.""" + + def failing_run(main: Any) -> None: + # Close the never-awaited coroutine so the test does not leave a + # RuntimeWarning attributed to whatever module GC runs in later. + main.close() + raise OSError("no fds for the event loop") + + with ( + patch("esphome.async_thread.asyncio.run", side_effect=failing_run), + pytest.raises(OSError, match="no fds"), + ): + run_async(lambda: asyncio.sleep(0), timeout=5) + + +def test_run_preserves_result_when_cleanup_fails( + caplog: pytest.LogCaptureFixture, +) -> None: + """A loop-cleanup failure after success is logged, not raised.""" + + async def coro() -> str: + return "ok" + + runner: AsyncThreadRunner[str] = AsyncThreadRunner(coro) + + def fake_run(main: Any) -> None: + main.close() + # Emulate _runner delivering the result before cleanup raised. A + # None result must count as delivered too, hence the completed flag. + runner.result = "ok" + runner.completed = True + raise KeyboardInterrupt + + with ( + caplog.at_level("DEBUG", logger="esphome.async_thread"), + patch("esphome.async_thread.asyncio.run", side_effect=fake_run), + ): + runner.run() + + assert runner.event.is_set() + assert runner.exception is None + assert runner.result == "ok" + assert "teardown failed after outcome recorded" in caplog.text + + +def test_run_async_none_result_is_success() -> None: + """A coroutine legitimately returning None is not treated as a failure.""" + + async def coro() -> None: + return None + + assert run_async(coro) is None + + +def test_run_async_on_orphan_skips_none_result() -> None: + """A late None result completes cleanly without invoking on_orphan.""" + orphaned: list[Any] = [] + finished = threading.Event() + release = threading.Event() + + async def coro() -> None: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + finished.set() + + before = _cleanup_threads() + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01, on_orphan=orphaned.append) + + release.set() + assert finished.wait(5) + _join_new_cleanup_threads(before) + assert not orphaned + + +def test_late_failure_without_on_orphan_is_logged( + caplog: pytest.LogCaptureFixture, +) -> None: + """An abandoned thread's real error leaves a visible trace.""" + release = threading.Event() + + async def coro() -> str: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + raise ValueError("the real cause") + + before = _cleanup_threads() + with caplog.at_level("DEBUG", logger="esphome.async_thread"): + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01) + + release.set() + _join_new_cleanup_threads(before) + assert "Abandoned async operation failed" in caplog.text + assert "the real cause" in caplog.text + + +def test_run_async_on_orphan_failure_is_contained( + caplog: pytest.LogCaptureFixture, +) -> None: + """An on_orphan callback that raises is logged, not propagated.""" + released = threading.Event() + release = threading.Event() + + def on_orphan(result: str) -> None: + released.set() + raise OSError("close failed") + + async def coro() -> str: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + return "late result" + + before = _cleanup_threads() + with caplog.at_level("DEBUG", logger="esphome.async_thread"): + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01, on_orphan=on_orphan) + + release.set() + assert released.wait(5) + _join_new_cleanup_threads(before) + assert "Error releasing orphaned result" in caplog.text + + +def test_run_async_on_orphan_releases_late_result() -> None: + """A result produced after the timeout is handed to on_orphan.""" + orphaned: list[Any] = [] + delivered = threading.Event() + release = threading.Event() + + def on_orphan(result: str) -> None: + orphaned.append(result) + delivered.set() + + async def coro() -> str: + # Block until the test has observed the timeout, so the result is + # guaranteed to arrive late no matter how slowly the runner is + # scheduled. + await asyncio.get_running_loop().run_in_executor(None, release.wait) + return "late result" + + before = _cleanup_threads() + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01, on_orphan=on_orphan) + + release.set() + assert delivered.wait(5) + _join_new_cleanup_threads(before) + assert orphaned == ["late result"] + + +def test_run_async_on_orphan_skips_late_failure() -> None: + """A late failure after the timeout is not handed to on_orphan.""" + orphaned: list[Any] = [] + failed = threading.Event() + release = threading.Event() + + async def coro() -> str: + # Block until the test has observed the timeout, so the failure is + # guaranteed to arrive late. + await asyncio.get_running_loop().run_in_executor(None, release.wait) + failed.set() + raise ValueError("late failure") + + before = _cleanup_threads() + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01, on_orphan=orphaned.append) + + release.set() + assert failed.wait(5) + _join_new_cleanup_threads(before) + assert not orphaned + + +def test_run_async_detects_missing_outcome() -> None: + """A run that records neither result nor exception raises loudly.""" + + def fake_run(main: Any) -> None: + # Simulate a loop that silently dropped the coroutine. + main.close() + + with ( + patch("esphome.async_thread.asyncio.run", side_effect=fake_run), + pytest.raises(RuntimeError, match="without a result"), + ): + run_async(lambda: asyncio.sleep(0), timeout=5) + + +def test_run_async_raises_distinguishable_timeout() -> None: + """The dispatcher's own expiry is a distinct TimeoutError subclass.""" + release = threading.Event() + + async def coro() -> None: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + + before = _cleanup_threads() + with pytest.raises(AsyncDispatchTimeout): + run_async(coro, timeout=0.01) + release.set() + _join_new_cleanup_threads(before) + + +def test_orphan_watcher_gives_up_on_a_hung_coroutine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The watcher exits after its bound instead of parking forever.""" + from esphome import async_thread + + monkeypatch.setattr(async_thread, "ORPHAN_WAIT_TIMEOUT", 0.01) + release = threading.Event() + orphaned: list[Any] = [] + + async def coro() -> str: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + return "too late" + + before = _cleanup_threads() + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01, on_orphan=orphaned.append) + + _join_new_cleanup_threads(before) + assert not orphaned + release.set() + + +def test_late_real_result_without_handler_is_logged( + caplog: pytest.LogCaptureFixture, +) -> None: + """A genuinely dropped late result leaves the discard trace.""" + release = threading.Event() + + async def coro() -> str: + await asyncio.get_running_loop().run_in_executor(None, release.wait) + return "dropped" + + before = _cleanup_threads() + with caplog.at_level("DEBUG", logger="esphome.async_thread"): + with pytest.raises(TimeoutError): + run_async(coro, timeout=0.01) + + release.set() + _join_new_cleanup_threads(before) + assert "Discarding late result" in caplog.text diff --git a/tests/unit_tests/test_resolver.py b/tests/unit_tests/test_resolver.py index 7862c268ca..16294a3813 100644 --- a/tests/unit_tests/test_resolver.py +++ b/tests/unit_tests/test_resolver.py @@ -4,12 +4,13 @@ from __future__ import annotations import re import socket -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, patch from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr import pytest +from esphome.async_thread import AsyncDispatchTimeout from esphome.core import EsphomeError from esphome.resolver import RESOLVE_TIMEOUT, AsyncResolver @@ -116,20 +117,17 @@ def test_async_resolver_generic_exception() -> None: def test_async_resolver_thread_timeout() -> None: """Test timeout when the runner thread doesn't complete in time.""" - # Patch AsyncThreadRunner inside esphome.resolver so we never actually - # start a thread and can control the wait return value directly. - fake_runner = MagicMock() - fake_runner.start = MagicMock() - fake_runner.event.wait.return_value = False # simulate timeout - + # Patch run_async inside esphome.resolver so we never actually start a + # thread and can simulate the wait timing out. with ( - patch("esphome.resolver.AsyncThreadRunner", return_value=fake_runner), - patch("esphome.resolver.hr.async_resolve_host"), + patch( + "esphome.resolver.run_async", side_effect=AsyncDispatchTimeout + ) as mock_run, pytest.raises(EsphomeError, match=re.escape("Timeout resolving IP address")), ): AsyncResolver(["test.local"], 6053).resolve() - fake_runner.start.assert_called_once() + mock_run.assert_called_once_with(ANY, timeout=RESOLVE_TIMEOUT + 1.0) def test_async_resolver_ip_addresses(mock_addr_info_ipv4: AddrInfo) -> None: