SIENTIAPDE-1974: fix(read): mark the shutdown snapshot loop for SonarQube (S7504)

SonarQube S7504 flagged `for slot in list(self.engine.subscriptions)` in
ServerReader.shutdown as an unnecessary list() call. It is necessary:
engine.unsubscribe() pops from engine.subscriptions, so iterating the dict
directly raises "dictionary changed size during iteration" and leaves the
remaining slots subscribed. Kept the snapshot with an explanatory comment and
`# NOSONAR`, the same convention as OpcConnectionPool.close_all().

Adds a regression test where unsubscribe() really pops: it fails with
RuntimeError without the list() and passes with it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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)

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