fix: set change during iteration when dispatching listeners

An existing listener may add new listeners to process ServiceInfo
when it sees a record. We need to make a copy of the listeners
set before iterating them to avoid `set changed size during iteration`

Fixes
```
2024-04-12 16:31:25.699 ERROR (MainThread) [homeassistant] Error doing job: Exception in callback _SelectorDatagramTransport._read_ready()
Traceback (most recent call last):
  File "/opt/homebrew/Cellar/python@3.12/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
  File "/opt/homebrew/Cellar/python@3.12/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/selector_events.py", line 1248, in _read_ready
    self._protocol.datagram_received(data, addr)
  File "src/zeroconf/_listener.py", line 86, in zeroconf._listener.AsyncListener.datagram_received
  File "src/zeroconf/_listener.py", line 104, in zeroconf._listener.AsyncListener.datagram_received
  File "src/zeroconf/_listener.py", line 175, in zeroconf._listener.AsyncListener._process_datagram_at_time
  File "src/zeroconf/_handlers/record_manager.py", line 161, in zeroconf._handlers.record_manager.RecordManager.async_updates_from_response
  File "src/zeroconf/_handlers/record_manager.py", line 70, in zeroconf._handlers.record_manager.RecordManager.async_updates_complete
RuntimeError: set changed size during iteration
```

src/zeroconf/_handlers/record_manager.py

@@ -56,7 +56,7 @@ def async_updates(self, now: _float, records: List[RecordUpdate]) -> None:
 
         This method will be run in the event loop.
         """
-        for listener in self.listeners:
+        for listener in self.listeners.copy():
             listener.async_update_records(self.zc, now, records)
 
     def async_updates_complete(self, notify: bool) -> None:
@@ -67,7 +67,7 @@ def async_updates_complete(self, notify: bool) -> None:
 
         This method will be run in the event loop.
         """
-        for listener in self.listeners:
+        for listener in self.listeners.copy():
             listener.async_update_records_complete()
         if notify:
             self.zc.async_notify_all()

tests/test_handlers.py

@@ -1762,3 +1762,77 @@ def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.Recor
     )
 
     await aiozc.async_close()
+
+
+@pytest.mark.asyncio
+async def test_async_updates_iteration_safe():
+    """Ensure we can safely iterate over the async_updates."""
+
+    aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
+    zc: Zeroconf = aiozc.zeroconf
+    updated = []
+    good_bye_answer = r.DNSPointer(
+        "myservicelow_tcp._tcp.local.",
+        const._TYPE_PTR,
+        const._CLASS_IN | const._CLASS_UNIQUE,
+        0,
+        'goodbye.local.',
+    )
+
+    class OtherListener(r.RecordUpdateListener):
+        """A RecordUpdateListener that does not implement update_records."""
+
+        def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.RecordUpdate]) -> None:
+            """Update multiple records in one shot."""
+            updated.extend(records)
+
+    other = OtherListener()
+
+    class ListenerThatAddsListener(r.RecordUpdateListener):
+        """A RecordUpdateListener that does not implement update_records."""
+
+        def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.RecordUpdate]) -> None:
+            """Update multiple records in one shot."""
+            updated.extend(records)
+            zc.async_add_listener(other, None)
+
+    zc.async_add_listener(ListenerThatAddsListener(), None)
+    await asyncio.sleep(0)  # flush out any call soons
+
+    # This should not raise RuntimeError: set changed size during iteration
+    zc.record_manager.async_updates(
+        now=current_time_millis(), records=[r.RecordUpdate(good_bye_answer, None)]
+    )
+
+    assert len(updated) == 1
+    await aiozc.async_close()
+
+
+@pytest.mark.asyncio
+async def test_async_updates_complete_iteration_safe():
+    """Ensure we can safely iterate over the async_updates_complete."""
+
+    aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
+    zc: Zeroconf = aiozc.zeroconf
+
+    class OtherListener(r.RecordUpdateListener):
+        """A RecordUpdateListener that does not implement update_records."""
+
+        def async_update_records_complete(self) -> None:
+            """Update multiple records in one shot."""
+
+    other = OtherListener()
+
+    class ListenerThatAddsListener(r.RecordUpdateListener):
+        """A RecordUpdateListener that does not implement update_records."""
+
+        def async_update_records_complete(self) -> None:
+            """Update multiple records in one shot."""
+            zc.async_add_listener(other, None)
+
+    zc.async_add_listener(ListenerThatAddsListener(), None)
+    await asyncio.sleep(0)  # flush out any call soons
+
+    # This should not raise RuntimeError: set changed size during iteration
+    zc.record_manager.async_updates_complete(False)
+    await aiozc.async_close()