Ensure startup can proceed if async_get_integration raises (#50799)

* Ensure startup can proceed if async_get_integration raises

There were cases where the event would never get set and
startup would deadlock because the second attempt to load
the integration would block forever

* pylint

* reorder

homeassistant/loader.py

@@ -506,31 +506,35 @@ async def async_get_integration(hass: HomeAssistant, domain: str) -> Integration
 
     event = cache[domain] = asyncio.Event()
 
+    try:
+        integration = await _async_get_integration(hass, domain)
+    except Exception:  # pylint: disable=broad-except
+        # Remove event from cache.
+        cache.pop(domain)
+        event.set()
+        raise
+
+    cache[domain] = integration
+    event.set()
+    return integration
+
+
+async def _async_get_integration(hass: HomeAssistant, domain: str) -> Integration:
     # Instead of using resolve_from_root we use the cache of custom
     # components to find the integration.
-    integration = (await async_get_custom_components(hass)).get(domain)
+    if integration := (await async_get_custom_components(hass)).get(domain):
-    if integration is not None:
         validate_custom_integration_version(integration)
         _LOGGER.warning(CUSTOM_WARNING, integration.domain)
-        cache[domain] = integration
-        event.set()
         return integration
 
     from homeassistant import components  # pylint: disable=import-outside-toplevel
 
-    integration = await hass.async_add_executor_job(
+    if integration := await hass.async_add_executor_job(
         Integration.resolve_from_root, hass, components, domain
-    )
+    ):
-    event.set()
+        return integration
 
-    if not integration:
+    raise IntegrationNotFound(domain)
-        # Remove event from cache.
-        cache.pop(domain)
-        raise IntegrationNotFound(domain)
-
-    cache[domain] = integration
-
-    return integration
 
 
 class LoaderError(Exception):

tests/test_loader.py

@@ -467,3 +467,32 @@ async def test_get_custom_components_safe_mode(hass):
     """Test that we get empty custom components in safe mode."""
     hass.config.safe_mode = True
     assert await loader.async_get_custom_components(hass) == {}
+
+
+async def test_custom_integration_missing_version(hass, caplog):
+    """Test trying to load a custom integration without a version twice does not deadlock."""
+    test_integration_1 = loader.Integration(
+        hass, "custom_components.test1", None, {"domain": "test1"}
+    )
+    with patch("homeassistant.loader.async_get_custom_components") as mock_get:
+        mock_get.return_value = {
+            "test1": test_integration_1,
+        }
+
+        with pytest.raises(loader.IntegrationNotFound):
+            await loader.async_get_integration(hass, "test1")
+
+        with pytest.raises(loader.IntegrationNotFound):
+            await loader.async_get_integration(hass, "test1")
+
+
+async def test_custom_integration_missing(hass, caplog):
+    """Test trying to load a custom integration that is missing twice not deadlock."""
+    with patch("homeassistant.loader.async_get_custom_components") as mock_get:
+        mock_get.return_value = {}
+
+        with pytest.raises(loader.IntegrationNotFound):
+            await loader.async_get_integration(hass, "test1")
+
+        with pytest.raises(loader.IntegrationNotFound):
+            await loader.async_get_integration(hass, "test1")