Remove async_late_forward_entry_setups and instead implicitly hold the lock (#119088)

* Refactor config entry forwards to implictly obtain the lock instead of explictly

This is a bit of a tradeoff to not need async_late_forward_entry_setups

The downside is we can no longer detect non-awaited plastform setups
as we will always implicitly obtain the lock instead of explictly.

Note, this PR is incomplete and is only for discussion purposes
at this point

* preen

* cover

* cover

* restore check for non-awaited platform setup

* cleanup

* fix missing word

* make non-awaited test safer

homeassistant/config_entries.py

@@ -1170,18 +1170,13 @@ class FlowCancelledError(Exception):
     """Error to indicate that a flow has been cancelled."""
 
 
-def _report_non_locked_platform_forwards(entry: ConfigEntry) -> None:
-    """Report non awaited and non-locked platform forwards."""
+def _report_non_awaited_platform_forwards(entry: ConfigEntry, what: str) -> None:
+    """Report non awaited platform forwards."""
     report(
-        f"calls async_forward_entry_setup after the entry for "
-        f"integration, {entry.domain} with title: {entry.title} "
-        f"and entry_id: {entry.entry_id}, has been set up, "
-        "without holding the setup lock that prevents the config "
-        "entry from being set up multiple times. "
-        "Instead await hass.config_entries.async_forward_entry_setup "
-        "during setup of the config entry or call "
-        "hass.config_entries.async_late_forward_entry_setups "
-        "in a tracked task. "
+        f"calls {what} for integration {entry.domain} with "
+        f"title: {entry.title} and entry_id: {entry.entry_id}, "
+        f"during setup without awaiting {what}, which can cause "
+        "the setup lock to be released before the setup is done. "
         "This will stop working in Home Assistant 2025.1",
         error_if_integration=False,
         error_if_core=False,
@@ -2041,9 +2036,6 @@ class ConfigEntries:
         before the entry is set up. This ensures that the config entry cannot
         be unloaded before all platforms are loaded.
 
-        If platforms must be loaded late (after the config entry is setup),
-        use async_late_forward_entry_setup instead.
-
         This method is more efficient than async_forward_entry_setup as
         it can load multiple platforms at once and does not require a separate
         import executor job for each platform.
@@ -2052,14 +2044,32 @@ class ConfigEntries:
         if not integration.platforms_are_loaded(platforms):
             with async_pause_setup(self.hass, SetupPhases.WAIT_IMPORT_PLATFORMS):
                 await integration.async_get_platforms(platforms)
-        if non_locked_platform_forwards := not entry.setup_lock.locked():
-            _report_non_locked_platform_forwards(entry)
+
+        if not entry.setup_lock.locked():
+            async with entry.setup_lock:
+                if entry.state is not ConfigEntryState.LOADED:
+                    raise OperationNotAllowed(
+                        f"The config entry {entry.title} ({entry.domain}) with entry_id"
+                        f" {entry.entry_id} cannot forward setup for {platforms} because it"
+                        f" is not loaded in the {entry.state} state"
+                    )
+                await self._async_forward_entry_setups_locked(entry, platforms)
+        else:
+            await self._async_forward_entry_setups_locked(entry, platforms)
+            # If the lock was held when we stated, and it was released during
+            # the platform setup, it means they did not await the setup call.
+            if not entry.setup_lock.locked():
+                _report_non_awaited_platform_forwards(
+                    entry, "async_forward_entry_setups"
+                )
+
+    async def _async_forward_entry_setups_locked(
+        self, entry: ConfigEntry, platforms: Iterable[Platform | str]
+    ) -> None:
         await asyncio.gather(
             *(
                 create_eager_task(
-                    self._async_forward_entry_setup(
-                        entry, platform, False, non_locked_platform_forwards
-                    ),
+                    self._async_forward_entry_setup(entry, platform, False),
                     name=(
                         f"config entry forward setup {entry.title} "
                         f"{entry.domain} {entry.entry_id} {platform}"
@@ -2070,25 +2080,6 @@ class ConfigEntries:
             )
         )
 
-    async def async_late_forward_entry_setups(
-        self, entry: ConfigEntry, platforms: Iterable[Platform | str]
-    ) -> None:
-        """Forward the setup of an entry to platforms after setup.
-
-        If platforms must be loaded late (after the config entry is setup),
-        use this method instead of async_forward_entry_setups as it holds
-        the setup lock until the platforms are loaded to ensure that the
-        config entry cannot be unloaded while platforms are loaded.
-        """
-        async with entry.setup_lock:
-            if entry.state is not ConfigEntryState.LOADED:
-                raise OperationNotAllowed(
-                    f"The config entry {entry.title} ({entry.domain}) with entry_id"
-                    f" {entry.entry_id} cannot forward setup for {platforms} "
-                    f"because it is not loaded in the {entry.state} state"
-                )
-            await self.async_forward_entry_setups(entry, platforms)
-
     async def async_forward_entry_setup(
         self, entry: ConfigEntry, domain: Platform | str
     ) -> bool:
@@ -2103,32 +2094,37 @@ class ConfigEntries:
         Instead, await async_forward_entry_setups as it can load
         multiple platforms at once and is more efficient since it
         does not require a separate import executor job for each platform.
-
-        If platforms must be loaded late (after the config entry is setup),
-        use async_late_forward_entry_setup instead.
         """
-        if non_locked_platform_forwards := not entry.setup_lock.locked():
-            _report_non_locked_platform_forwards(entry)
-        else:
-            report(
-                "calls async_forward_entry_setup for "
-                f"integration, {entry.domain} with title: {entry.title} "
-                f"and entry_id: {entry.entry_id}, which is deprecated and "
-                "will stop working in Home Assistant 2025.6, "
-                "await async_forward_entry_setups instead",
-                error_if_core=False,
-                error_if_integration=False,
-            )
-        return await self._async_forward_entry_setup(
-            entry, domain, True, non_locked_platform_forwards
+        report(
+            "calls async_forward_entry_setup for "
+            f"integration, {entry.domain} with title: {entry.title} "
+            f"and entry_id: {entry.entry_id}, which is deprecated and "
+            "will stop working in Home Assistant 2025.6, "
+            "await async_forward_entry_setups instead",
+            error_if_core=False,
+            error_if_integration=False,
         )
+        if not entry.setup_lock.locked():
+            async with entry.setup_lock:
+                if entry.state is not ConfigEntryState.LOADED:
+                    raise OperationNotAllowed(
+                        f"The config entry {entry.title} ({entry.domain}) with entry_id"
+                        f" {entry.entry_id} cannot forward setup for {domain} because it"
+                        f" is not loaded in the {entry.state} state"
+                    )
+                return await self._async_forward_entry_setup(entry, domain, True)
+        result = await self._async_forward_entry_setup(entry, domain, True)
+        # If the lock was held when we stated, and it was released during
+        # the platform setup, it means they did not await the setup call.
+        if not entry.setup_lock.locked():
+            _report_non_awaited_platform_forwards(entry, "async_forward_entry_setup")
+        return result
 
     async def _async_forward_entry_setup(
         self,
         entry: ConfigEntry,
         domain: Platform | str,
         preload_platform: bool,
-        non_locked_platform_forwards: bool,
     ) -> bool:
         """Forward the setup of an entry to a different component."""
         # Setup Component if not set up yet
@@ -2152,12 +2148,6 @@ class ConfigEntries:
 
         integration = loader.async_get_loaded_integration(self.hass, domain)
         await entry.async_setup(self.hass, integration=integration)
-
-        # Check again after setup to make sure the lock
-        # is still there because it could have been released
-        # unless we already reported it.
-        if not non_locked_platform_forwards and not entry.setup_lock.locked():
-            _report_non_locked_platform_forwards(entry)
         return True
 
     async def async_unload_platforms(
@@ -2221,7 +2211,7 @@ class ConfigEntries:
         # The component was not loaded.
         if entry.domain not in self.hass.config.components:
             return False
-        return entry.state == ConfigEntryState.LOADED
+        return entry.state is ConfigEntryState.LOADED
 
 
 @callback