Filter stale replayed BLE advertisements in Matter BLE proxy (#172773)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

homeassistant/components/matter/ble_proxy.py

@@ -23,6 +23,7 @@ from matter_ble_proxy import (
 )
 
 from homeassistant.components.bluetooth import (
+    MONOTONIC_TIME,
     BluetoothScanningMode,
     async_ble_device_from_address,
     async_register_callback,
@@ -51,11 +52,18 @@ class HaBluetoothScanSource(BleScanSource):
         if self._cancel is not None:
             return
 
+        # Drop HA's synchronous replay of stale history on register; otherwise a
+        # rotating peripheral's old addresses each become a parallel connect candidate.
+        # `MONOTONIC_TIME` is the clock that stamps `service_info.time`.
+        scan_start = MONOTONIC_TIME()
+
         @callback
         def _on_advertisement(
             service_info: BluetoothServiceInfoBleak,
             _change: object,
         ) -> None:
+            if service_info.time < scan_start:
+                return
             try:
                 callback_fn(_to_advertisement_data(service_info))
             except Exception:

tests/components/matter/test_ble_proxy.py

@@ -22,7 +22,7 @@ from homeassistant.components.matter.ble_proxy import (
 from homeassistant.core import HomeAssistant
 
 
-def _make_service_info() -> BluetoothServiceInfoBleak:
+def _make_service_info(time: float | None = None) -> BluetoothServiceInfoBleak:
     """Return a real BluetoothServiceInfoBleak with realistic field values."""
     address = "AA:BB:CC:DD:EE:FF"
     name = "TestDevice"
@@ -37,7 +37,7 @@ def _make_service_info() -> BluetoothServiceInfoBleak:
         device=BLEDevice(name=name, address=address, details={}),
         advertisement=None,
         connectable=True,
-        time=monotonic_time_coarse(),
+        time=monotonic_time_coarse() if time is None else time,
         tx_power=0,
         raw=None,
     )
@@ -152,6 +152,43 @@ async def test_scan_source_callback_forwards_advertisement(
     assert forwarded[0].address == "AA:BB:CC:DD:EE:FF"
 
 
+@pytest.mark.parametrize(
+    ("advert_time", "expected_count"),
+    [
+        pytest.param(999.0, 0, id="stale-before-scan-start-dropped"),
+        pytest.param(1000.0, 1, id="equal-scan-start-forwarded"),
+        pytest.param(1001.0, 1, id="fresh-after-scan-start-forwarded"),
+    ],
+)
+async def test_scan_source_drops_replayed_history(
+    hass: HomeAssistant, advert_time: float, expected_count: int
+) -> None:
+    """Adverts older than the registration instant (HA history replay) are dropped."""
+    forwarded: list[AdvertisementData] = []
+    captured: dict[str, object] = {}
+
+    def fake_register(hass_, cb, _matcher, _mode):
+        captured["cb"] = cb
+        return MagicMock()
+
+    source = HaBluetoothScanSource(hass)
+    with (
+        patch(
+            "homeassistant.components.matter.ble_proxy.async_register_callback",
+            side_effect=fake_register,
+        ),
+        patch(
+            "homeassistant.components.matter.ble_proxy.MONOTONIC_TIME",
+            return_value=1000.0,
+        ),
+    ):
+        await source.start(forwarded.append)
+
+    captured["cb"](_make_service_info(time=advert_time), object())
+
+    assert len(forwarded) == expected_count
+
+
 async def test_scan_source_callback_swallows_exceptions(
     hass: HomeAssistant, caplog: pytest.LogCaptureFixture
 ) -> None: