Fix stop listener memory leak in DataUpdateCoordinator on retry (#49186)

* Fix stop listener leak in DataUpdateCoordinator

When an integration retries setup it will add a new stop listener

* Skip scheduled refreshes when hass is stopping

* Update homeassistant/helpers/update_coordinator.py

* ensure manual refresh after stop

homeassistant/helpers/update_coordinator.py

@@ -12,7 +12,6 @@ import aiohttp
 import requests
 
 from homeassistant import config_entries
-from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import CALLBACK_TYPE, Event, HassJob, HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 from homeassistant.helpers import entity, event
@@ -74,10 +73,6 @@ class DataUpdateCoordinator(Generic[T]):
 
         self._debounced_refresh = request_refresh_debouncer
 
-        self.hass.bus.async_listen_once(
-            EVENT_HOMEASSISTANT_STOP, self._async_stop_refresh
-        )
-
     @callback
     def async_add_listener(self, update_callback: CALLBACK_TYPE) -> Callable[[], None]:
         """Listen for data updates."""
@@ -128,7 +123,7 @@ class DataUpdateCoordinator(Generic[T]):
     async def _handle_refresh_interval(self, _now: datetime) -> None:
         """Handle a refresh interval occurrence."""
         self._unsub_refresh = None
-        await self.async_refresh()
+        await self._async_refresh(log_failures=True, scheduled=True)
 
     async def async_request_refresh(self) -> None:
         """Request a refresh.
@@ -162,7 +157,10 @@ class DataUpdateCoordinator(Generic[T]):
         await self._async_refresh(log_failures=True)
 
     async def _async_refresh(
-        self, log_failures: bool = True, raise_on_auth_failed: bool = False
+        self,
+        log_failures: bool = True,
+        raise_on_auth_failed: bool = False,
+        scheduled: bool = False,
     ) -> None:
         """Refresh data."""
         if self._unsub_refresh:
@@ -170,6 +168,10 @@ class DataUpdateCoordinator(Generic[T]):
             self._unsub_refresh = None
 
         self._debounced_refresh.async_cancel()
+
+        if scheduled and self.hass.is_stopping:
+            return
+
         start = monotonic()
         auth_failed = False
 
@@ -249,7 +251,7 @@ class DataUpdateCoordinator(Generic[T]):
                 self.name,
                 monotonic() - start,
             )
-            if not auth_failed and self._listeners:
+            if not auth_failed and self._listeners and not self.hass.is_stopping:
                 self._schedule_refresh()
 
         for update_callback in self._listeners:

tests/helpers/test_update_coordinator.py

@@ -335,6 +335,15 @@ async def test_stop_refresh_on_ha_stop(hass, crd):
     await hass.async_block_till_done()
     assert crd.data == 1
 
+    # Ensure we can still manually refresh after stop
+    await crd.async_refresh()
+    assert crd.data == 2
+
+    # ...and that the manual refresh doesn't setup another scheduled refresh
+    async_fire_time_changed(hass, utcnow() + update_interval)
+    await hass.async_block_till_done()
+    assert crd.data == 2
+
 
 @pytest.mark.parametrize(
     "err_msg",