Prevent config entries from being reloaded concurrently (#72636)

* Prevent config entries being reloaded concurrently

- Fixes Config entry has already been setup when
  two places try to reload the config entry at the
  same time.

- This comes up quite a bit:
  https://github.com/home-assistant/core/issues?q=is%3Aissue+sort%3Aupdated-desc+%22Config+entry+has+already+been+setup%22+is%3Aclosed

* Make sure plex creates mocks in the event loop

* drop reload_lock, already inherits

homeassistant/config_entries.py

@@ -186,6 +186,7 @@ class ConfigEntry:
         "reason",
         "_async_cancel_retry_setup",
         "_on_unload",
+        "reload_lock",
     )
 
     def __init__(
@@ -275,6 +276,9 @@ class ConfigEntry:
         # Hold list for functions to call on unload.
         self._on_unload: list[CALLBACK_TYPE] | None = None
 
+        # Reload lock to prevent conflicting reloads
+        self.reload_lock = asyncio.Lock()
+
     async def async_setup(
         self,
         hass: HomeAssistant,
@@ -1005,12 +1009,13 @@ class ConfigEntries:
         if (entry := self.async_get_entry(entry_id)) is None:
             raise UnknownEntry
 
-        unload_result = await self.async_unload(entry_id)
+        async with entry.reload_lock:
+            unload_result = await self.async_unload(entry_id)
 
-        if not unload_result or entry.disabled_by:
-            return unload_result
+            if not unload_result or entry.disabled_by:
+                return unload_result
 
-        return await self.async_setup(entry_id)
+            return await self.async_setup(entry_id)
 
     async def async_set_disabled_by(
         self, entry_id: str, disabled_by: ConfigEntryDisabler | None

tests/components/plex/conftest.py

@@ -381,7 +381,7 @@ def hubs_music_library_fixture():
 
 
 @pytest.fixture(name="entry")
-def mock_config_entry():
+async def mock_config_entry():
     """Return the default mocked config entry."""
     return MockConfigEntry(
         domain=DOMAIN,

tests/test_config_entries.py

@@ -1497,7 +1497,7 @@ async def test_reload_entry_entity_registry_works(hass):
     )
     await hass.async_block_till_done()
 
-    assert len(mock_unload_entry.mock_calls) == 1
+    assert len(mock_unload_entry.mock_calls) == 2
 
 
 async def test_unique_id_persisted(hass, manager):
@@ -3080,3 +3080,41 @@ async def test_deprecated_disabled_by_str_set(hass, manager, caplog):
     )
     assert entry.disabled_by is config_entries.ConfigEntryDisabler.USER
     assert " str for config entry disabled_by. This is deprecated " in caplog.text
+
+
+async def test_entry_reload_concurrency(hass, manager):
+    """Test multiple reload calls do not cause a reload race."""
+    entry = MockConfigEntry(domain="comp", state=config_entries.ConfigEntryState.LOADED)
+    entry.add_to_hass(hass)
+
+    async_setup = AsyncMock(return_value=True)
+    loaded = 1
+
+    async def _async_setup_entry(*args, **kwargs):
+        await asyncio.sleep(0)
+        nonlocal loaded
+        loaded += 1
+        return loaded == 1
+
+    async def _async_unload_entry(*args, **kwargs):
+        await asyncio.sleep(0)
+        nonlocal loaded
+        loaded -= 1
+        return loaded == 0
+
+    mock_integration(
+        hass,
+        MockModule(
+            "comp",
+            async_setup=async_setup,
+            async_setup_entry=_async_setup_entry,
+            async_unload_entry=_async_unload_entry,
+        ),
+    )
+    mock_entity_platform(hass, "config_flow.comp", None)
+    tasks = []
+    for _ in range(15):
+        tasks.append(asyncio.create_task(manager.async_reload(entry.entry_id)))
+    await asyncio.gather(*tasks)
+    assert entry.state is config_entries.ConfigEntryState.LOADED
+    assert loaded == 1