Ensure recorder shuts down when its startup future is canceled out from under it (#72866)

homeassistant/components/recorder/core.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Callable, Iterable
+from concurrent.futures import CancelledError
 import contextlib
 from datetime import datetime, timedelta
 import logging
@@ -518,9 +519,16 @@ class Recorder(threading.Thread):
 
     def _wait_startup_or_shutdown(self) -> object | None:
         """Wait for startup or shutdown before starting."""
-        return asyncio.run_coroutine_threadsafe(
+        try:
-            self._async_wait_for_started(), self.hass.loop
+            return asyncio.run_coroutine_threadsafe(
-        ).result()
+                self._async_wait_for_started(), self.hass.loop
+            ).result()
+        except CancelledError as ex:
+            _LOGGER.warning(
+                "Recorder startup was externally canceled before it could complete: %s",
+                ex,
+            )
+            return SHUTDOWN_TASK
 
     def run(self) -> None:
         """Start processing events to save."""

tests/components/recorder/test_init.py

@@ -118,6 +118,26 @@ async def test_shutdown_before_startup_finishes(
     assert run_info.end is not None
 
 
+async def test_canceled_before_startup_finishes(
+    hass: HomeAssistant,
+    async_setup_recorder_instance: SetupRecorderInstanceT,
+    caplog: pytest.LogCaptureFixture,
+):
+    """Test recorder shuts down when its startup future is canceled out from under it."""
+    hass.state = CoreState.not_running
+    await async_setup_recorder_instance(hass)
+    instance = get_instance(hass)
+    await instance.async_db_ready
+    instance._hass_started.cancel()
+    with patch.object(instance, "engine"):
+        await hass.async_block_till_done()
+        await hass.async_add_executor_job(instance.join)
+    assert (
+        "Recorder startup was externally canceled before it could complete"
+        in caplog.text
+    )
+
+
 async def test_shutdown_closes_connections(hass, recorder_mock):
     """Test shutdown closes connections."""