[core] Generalize the resolver's async thread dispatch into run_async (#18088)

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)

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

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

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: