Fix allowing identical flows to be created before startup
diff --git a/homeassistant/helpers/discovery_flow.py b/homeassistant/helpers/discovery_flow.py
index 863fb58625ce..2bfccf46960e 100644
--- a/homeassistant/helpers/discovery_flow.py
+++ b/homeassistant/helpers/discovery_flow.py
@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from collections.abc import Coroutine
-from typing import Any
+from typing import Any, NamedTuple
 
 from homeassistant.const import EVENT_HOMEASSISTANT_STARTED
 from homeassistant.core import CoreState, Event, HomeAssistant, callback
@@ -20,17 +20,18 @@ def async_create_flow(
     hass: HomeAssistant, domain: str, context: dict[str, Any], data: Any
 ) -> None:
     """Create a discovery flow."""
-    if hass.state == CoreState.running:
+    dispatcher: FlowDispatcher | None = None
+    if DISCOVERY_FLOW_DISPATCHER in hass.data:
+        dispatcher = hass.data[DISCOVERY_FLOW_DISPATCHER]
+    elif hass.state != CoreState.running:
+        dispatcher = hass.data[DISCOVERY_FLOW_DISPATCHER] = FlowDispatcher(hass)
+        dispatcher.async_setup()
+
+    if not dispatcher or dispatcher.started:
         if init_coro := _async_init_flow(hass, domain, context, data):
             hass.async_create_task(init_coro)
         return
 
-    if DISCOVERY_FLOW_DISPATCHER not in hass.data:
-        dispatcher = hass.data[DISCOVERY_FLOW_DISPATCHER] = FlowDispatcher(hass)
-        dispatcher.async_setup()
-    else:
-        dispatcher = hass.data[DISCOVERY_FLOW_DISPATCHER]
-
     return dispatcher.async_create(domain, context, data)
 
 
@@ -49,13 +50,28 @@ def _async_init_flow(
     return hass.config_entries.flow.async_init(domain, context=context, data=data)
 
 
+class PendingFlowKey(NamedTuple):
+    """Key for pending flows."""
+
+    domain: str
+    source: str
+
+
+class PendingFlowValue(NamedTuple):
+    """Value for pending flows."""
+
+    context: dict[str, Any]
+    data: Any
+
+
 class FlowDispatcher:
     """Dispatch discovery flows."""
 
     def __init__(self, hass: HomeAssistant) -> None:
         """Init the discovery dispatcher."""
         self.hass = hass
-        self.pending_flows: list[tuple[str, dict[str, Any], Any]] = []
+        self.started = False
+        self.pending_flows: dict[PendingFlowKey, list[PendingFlowValue]] = {}
 
     @callback
     def async_setup(self) -> None:
@@ -64,10 +80,16 @@ def async_setup(self) -> None:
 
     async def _async_start(self, event: Event) -> None:
         """Start processing pending flows."""
-        self.hass.data.pop(DISCOVERY_FLOW_DISPATCHER)
-
-        init_coros = [_async_init_flow(self.hass, *flow) for flow in self.pending_flows]
-
+        pending_flows = self.pending_flows
+        self.pending_flows = {}
+        self.started = True
+        init_coros = [
+            _async_init_flow(
+                self.hass, flow_key.domain, flow_values.context, flow_values.data
+            )
+            for flow_key, flows in pending_flows.items()
+            for flow_values in flows
+        ]
         await gather_with_concurrency(
             FLOW_INIT_LIMIT,
             *[init_coro for init_coro in init_coros if init_coro is not None],
@@ -76,4 +98,8 @@ async def _async_start(self, event: Event) -> None:
     @callback
     def async_create(self, domain: str, context: dict[str, Any], data: Any) -> None:
         """Create and add or queue a flow."""
-        self.pending_flows.append((domain, context, data))
+        key = PendingFlowKey(domain, context["source"])
+        values = PendingFlowValue(context, data)
+        existing = self.pending_flows.setdefault(key, [])
+        if not any(existing_values.data == data for existing_values in existing):
+            existing.append(values)
diff --git a/tests/helpers/test_discovery_flow.py b/tests/helpers/test_discovery_flow.py
index 549848e5c7b0..4019be803154 100644
--- a/tests/helpers/test_discovery_flow.py
+++ b/tests/helpers/test_discovery_flow.py
@@ -56,8 +56,11 @@ async def test_async_create_flow_deferred_until_started(hass, mock_flow_init):
     ]
 
 
-async def test_async_create_flow_checks_existing_flows(hass, mock_flow_init):
-    """Test existing flows prevent an identical one from being creates."""
+async def test_async_create_flow_checks_existing_flows_after_startup(
+    hass, mock_flow_init
+):
+    """Test existing flows prevent an identical ones from being after startup."""
+    hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
     with patch(
         "homeassistant.data_entry_flow.FlowManager.async_has_matching_flow",
         return_value=True,
@@ -69,3 +72,26 @@ async def test_async_create_flow_checks_existing_flows(hass, mock_flow_init):
             {"properties": {"id": "aa:bb:cc:dd:ee:ff"}},
         )
         assert not mock_flow_init.mock_calls
+
+
+async def test_async_create_flow_checks_existing_flows_before_startup(
+    hass, mock_flow_init
+):
+    """Test existing flows prevent an identical ones from being created before startup."""
+    hass.state = CoreState.stopped
+    for _ in range(2):
+        discovery_flow.async_create_flow(
+            hass,
+            "hue",
+            {"source": config_entries.SOURCE_HOMEKIT},
+            {"properties": {"id": "aa:bb:cc:dd:ee:ff"}},
+        )
+    hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
+    await hass.async_block_till_done()
+    assert mock_flow_init.mock_calls == [
+        call(
+            "hue",
+            context={"source": "homekit"},
+            data={"properties": {"id": "aa:bb:cc:dd:ee:ff"}},
+        )
+    ]

## Proposed change
<!--
  Describe the big picture of your changes here to communicate to the
  maintainers why we should accept this pull request. If it fixes a bug
  or resolves a feature request, be sure to link to that issue in the
  additional information section.
-->
The check for identical discovery flows only worked after the started event because we were still holding the tasks as pending so there was nothing to compare again. We now check against pending flows as well before the started event.

If startup took a while we could have ended up with quite the thundering herd when the started event does fire.

## Type of change
<!--
  What type of change does your PR introduce to Home Assistant?
  NOTE: Please, check only 1! box!
  If your PR requires multiple boxes to be checked, you'll most likely need to
  split it into multiple PRs. This makes things easier and faster to code review.
-->

- [ ] Dependency upgrade
- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New integration (thank you!)
- [ ] New feature (which adds functionality to an existing integration)
- [ ] Deprecation (breaking change to happen in the future)
- [ ] Breaking change (fix/feature causing existing functionality to break)
- [ ] Code quality improvements to existing code or addition of tests

## Additional information
<!--
  Details are important, and help maintainers processing your PR.
  Please be sure to fill out additional details, if applicable.
-->

- This PR fixes or closes issue: fixes #
- This PR is related to issue: 
- Link to documentation pull request: 

## Checklist
<!--
  Put an `x` in the boxes that apply. You can also fill these out after
  creating the PR. If you're unsure about any of them, don't hesitate to ask.
  We're here to help! This is simply a reminder of what we are going to look
  for before merging your code.
-->

- [x] The code change is tested and works locally.
- [ ] Local tests pass. **Your PR cannot be merged unless tests pass**
- [ ] There is no commented out code in this PR.
- [ ] I have followed the [development checklist][dev-checklist]
- [ ] The code has been formatted using Black (`black --fast homeassistant tests`)
- [x] Tests have been added to verify that the new code works.

If user exposed functionality or configuration variables are added/changed:

- [ ] Documentation added/updated for [www.home-assistant.io][docs-repository]

If the code communicates with devices, web services, or third-party tools:

- [ ] The [manifest file][manifest-docs] has all fields filled out correctly.  
      Updated and included derived files by running: `python3 -m script.hassfest`.
- [ ] New or updated dependencies have been added to `requirements_all.txt`.  
      Updated by running `python3 -m script.gen_requirements_all`.
- [ ] For the updated dependencies - a link to the changelog, or at minimum a diff between library versions is added to the PR description.
- [ ] Untested files have been added to `.coveragerc`.

<!--
  This project is very active and we have a high turnover of pull requests.

  Unfortunately, the number of incoming pull requests is higher than what our
  reviewers can review and merge so there is a long backlog of pull requests
  waiting for review. You can help here!
  
  By reviewing another pull request, you will help raise the code quality of
  that pull request and the final review will be faster. This way the general
  pace of pull request reviews will go up and your wait time will go down.
  
  When picking a pull request to review, try to choose one that hasn't yet
  been reviewed.

  Thanks for helping out!
-->

To help with the load of incoming pull requests:

- [ ] I have reviewed two other [open pull requests][prs] in this repository.

[prs]: https://github.com/home-assistant/core/pulls?q=is%3Aopen+is%3Apr+-author%3A%40me+-draft%3Atrue+-label%3Awaiting-for-upstream+sort%3Acreated-desc+review%3Anone+-status%3Afailure

<!--
  Thank you for contributing <3

  Below, some useful links you could explore:
-->
[dev-checklist]: https://developers.home-assistant.io/docs/en/development_checklist.html
[manifest-docs]: https://developers.home-assistant.io/docs/en/creating_integration_manifest.html
[quality-scale]: https://developers.home-assistant.io/docs/en/next/integration_quality_scale_index.html
[docs-repository]: https://github.com/home-assistant/home-assistant.io
