diff --git a/ingestor/managers/server_reader.py b/ingestor/managers/server_reader.py index 21c76f7..3335d2f 100644 --- a/ingestor/managers/server_reader.py +++ b/ingestor/managers/server_reader.py @@ -83,7 +83,9 @@ class ServerReader: also dropped, so the next initialization reconnects. Without it (the slot just moved to another pod) the connection stays for writes/gRPC: the connection is shared, and the idle sweep closes it if nobody uses it anymore.""" - for slot in list(self.engine.subscriptions): + # list() snapshots the keys: unsubscribe() pops from engine.subscriptions, so + # iterating the dict directly would raise "changed size during iteration". + for slot in list(self.engine.subscriptions): # NOSONAR await self.engine.unsubscribe(slot) if evict: await self.pool.evict(self.server_id) diff --git a/tests/unit/managers/test_server_reader.py b/tests/unit/managers/test_server_reader.py index b2cc54f..b34876e 100644 --- a/tests/unit/managers/test_server_reader.py +++ b/tests/unit/managers/test_server_reader.py @@ -142,3 +142,23 @@ async def test_shutdown_without_evict_keeps_the_shared_connection(): pool.evict.assert_not_awaited() for c in reader.engine.emit_metric.await_args_list: assert c.kwargs['metric_object'] is not metrics.OPC_CONNECTION_STATUS + + +@pytest.mark.asyncio +async def test_shutdown_unsubscribes_every_slot_while_engine_pops_them(): + """The real unsubscribe() pops from engine.subscriptions: shutdown must iterate a + snapshot (list()), or the loop breaks with "dictionary changed size during + iteration" and the remaining slots stay subscribed (SonarQube S7504 does not apply).""" + reader, pool = make_reader() + subs = {'1': MagicMock(), '2': MagicMock(), '3': MagicMock()} + reader.engine.subscriptions = subs + + async def pop(slot): + subs.pop(slot) + + reader.engine.unsubscribe = AsyncMock(side_effect=pop) + + await reader.shutdown() + + assert subs == {} + assert reader.engine.unsubscribe.await_count == 3