From 4235ce8edb95557b4e10ade418a1c656b7ea8483 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Wed, 7 Oct 2026 16:15:06 -0300 Subject: [PATCH] SIENTIAPDE-1974: fix(slots): guard every reader unsubscribe and count failures Second code review round (xhigh) of 057c1fd + 085e609: 5 findings. Fixed: - manage_server's subscribe-failure path called reader.unsubscribe() directly; with the legacy OpcManager a dead session made it raise out of the except block and stop the main loop. Both paths now go through _unsubscribe_reader, the single guarded per-reader unsubscribe. - An unsubscribe failure also increments OPC_SUBSCRIPTION_ERRORS (same tags as the subscription failures), besides the OPC_UNSUBSCRIBE_ERROR notification. - The contract test creates the subscriptions through the readers' real create_subscription (fake pool / client) instead of writing engine internals, and checks the pool pin/unpin pair. Documented (no code change): - If delete() fails while the session is still alive (e.g. a timeout), the server-side subscription may keep delivering data until that session ends (reconnection or pod restart); noted in _unsubscribe_reader. - manage_leases now unsubscribes before releasing the leases: if Redis fails while releasing them, the slots stay unread until LEASE_TTL expires, which is preferred to two pods reading the same slot. 284 unit tests; the 2 new behavior tests fail on 085e609. Co-Authored-By: Claude Opus 5.5 --- ingestor/managers/ingestor_manager.py | 70 ++++++++++++++------ tests/unit/managers/test_ingestor_manager.py | 54 +++++++++++++-- tests/unit/test_ingestor.py | 6 +- 3 files changed, 100 insertions(+), 30 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 5520ee8..a141e3c 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -549,10 +549,8 @@ class IngestorManager(SientiaMonitoring): also left the subscription open on the OPC server. Unsubscribing a slot that no reader holds is a no-op. - A reader that fails to unsubscribe (e.g. the legacy OpcManager, whose delete() - raises when the server session is gone) is notified and logged, and the slot is - dropped from its books, as the v2 engine already does, so it can be subscribed - again. The other readers are still unsubscribed and the main loop goes on. + A reader that fails to unsubscribe is handled by _unsubscribe_reader: the other + readers are still unsubscribed and the main loop goes on. Args: slot (str): The name of the slot to unsubscribe. @@ -560,28 +558,56 @@ class IngestorManager(SientiaMonitoring): Side Effects: - Removes the slot's subscription from every OPC reader that has one - Cleans up subscription resources on the OPC servers - - Sends OPC_UNSUBSCRIBE_ERROR_: when a reader fails """ # list() snapshots the readers: the loop awaits, so a reader added or removed # meanwhile must not break the iteration. for server, reader in list(self.opc_managers.items()): # NOSONAR - if slot not in reader.subscriptions: - continue - try: - await reader.unsubscribe(slot) - except Exception as e: - trace = traceback.format_exc() - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}', - message=f'Failed to unsubscribe slot {slot} from {server}: {e}', - block='opc_manager', - level=NotificationLevel.ERROR, - attachment_content=trace, - ) - self.logger.error(trace) - reader.subscriptions.pop(slot, None) + if slot in reader.subscriptions: + await self._unsubscribe_reader(server, reader, slot) + + async def _unsubscribe_reader( + self, server: str, reader: ServerReader | OpcManager, slot: str + ) -> None: + """ + Unsubscribe one reader from a slot without letting a failure stop the main loop. + + A failure (e.g. the legacy OpcManager, whose delete() raises when the server + session is gone) increments OPC_SUBSCRIPTION_ERRORS, sends + OPC_UNSUBSCRIBE_ERROR_: and is logged. The slot is dropped from the + reader's books, as the v2 engine already does, so it can be subscribed again. + + Known limitation: if delete() failed while the session is still alive (e.g. a + timeout), the server-side subscription may keep delivering data until that + session ends (reconnection or pod restart). + + Args: + server (str): The reader's key in opc_managers + reader (ServerReader | OpcManager): The OPC reader + slot (str): The slot to unsubscribe + """ + try: + await reader.unsubscribe(slot) + except Exception as e: + await self.emit_metric( + metric_object=metrics.OPC_SUBSCRIPTION_ERRORS, + tags={ + 'pod_id': self.pod_id, + 'server': server, + 'slot': slot, + }, + ) + trace = traceback.format_exc() + await self.send_notification_async( + metadata=self.metadata, + notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}', + message=f'Failed to unsubscribe slot {slot} from {server}: {e}', + block='opc_manager', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + self.logger.error(trace) + reader.subscriptions.pop(slot, None) async def _unsubscribe_removed_slots(self, slots: list[str]) -> None: """Unsubscribe and forget slots whose configuration is gone from Redis.""" @@ -796,7 +822,7 @@ class IngestorManager(SientiaMonitoring): self.logger.error(trace) self.logger.warning(f'Removing subscription from server {server} for slot {slot}') - await self.opc_managers[server].unsubscribe(slot) + await self._unsubscribe_reader(server, self.opc_managers[server], slot) return 2 return 0 diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 4c14b29..a91b1a5 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -4,6 +4,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel +import ingestor.metrics as metrics from ingestor.config import GatewayConfig from ingestor.connection.pool import OpcConnectionPool from ingestor.io.slots import parse_slot @@ -607,6 +608,10 @@ async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, healthy.unsubscribe.assert_called_once_with('slot1') assert 'slot1' not in broken.subscriptions + ingestor_manager.emit_metric.assert_called_once_with( + metric_object=metrics.OPC_SUBSCRIPTION_ERRORS, + tags={'pod_id': ingestor_manager.pod_id, 'server': 'server1', 'slot': 'slot1'}, + ) ingestor_manager.send_notification_async.assert_called_once_with( metadata=ingestor_manager.metadata, notification_id='OPC_UNSUBSCRIBE_ERROR_slot1:server1', @@ -619,12 +624,18 @@ async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, @mark.asyncio async def test_unsubscribe_slot_with_real_readers(ingestor_manager): - """Contract: both real readers expose `subscriptions` keyed by slot, and + """Contract: both real readers keep `subscriptions` keyed by slot, and unsubscribe_slot closes the slot's subscription on each of them""" - pool = MagicMock(get=AsyncMock(), evict=AsyncMock()) + v2_subscription = AsyncMock() + client = MagicMock(create_subscription=AsyncMock(return_value=v2_subscription)) + pool = MagicMock(get=AsyncMock(return_value=client), evict=AsyncMock()) server_reader = ServerReader( name='kepserver', - server_config={'server_id': '1', 'url': 'opc.tcp://kepserver:49320'}, + server_config={ + 'server_id': '1', + 'url': 'opc.tcp://kepserver:49320', + 'subscription_period_ms': 1000.0, + }, pool=pool, sink=AsyncMock(), logger=MagicMock(), @@ -633,9 +644,7 @@ async def test_unsubscribe_slot_with_real_readers(ingestor_manager): metadata={}, ) server_reader.engine.emit_metric = AsyncMock() - v2_subscription = AsyncMock() - server_reader.engine.subscriptions['slot1'] = v2_subscription - server_reader.engine._servers['slot1'] = '1' + await server_reader.create_subscription('slot1') with patch('ingestor.managers.opc_manager.metrics'): legacy_reader = OpcManager( @@ -649,9 +658,15 @@ async def test_unsubscribe_slot_with_real_readers(ingestor_manager): metadata={}, metrics_controller=AsyncMock(), ) + legacy_reader.emit_metric = AsyncMock() legacy_subscription = AsyncMock() - legacy_reader.subscriptions['slot1'] = legacy_subscription + legacy_reader.client = MagicMock( + create_subscription=AsyncMock(return_value=legacy_subscription) + ) + await legacy_reader.create_subscription('slot1') + assert 'slot1' in server_reader.subscriptions + assert 'slot1' in legacy_reader.subscriptions ingestor_manager.opc_managers = {'kepserver': server_reader, 'legacy': legacy_reader} await ingestor_manager.unsubscribe_slot('slot1') @@ -660,6 +675,7 @@ async def test_unsubscribe_slot_with_real_readers(ingestor_manager): legacy_subscription.delete.assert_awaited_once() assert 'slot1' not in server_reader.subscriptions assert 'slot1' not in legacy_reader.subscriptions + pool.pin.assert_called_once_with('1') pool.unpin.assert_called_once_with('1') @@ -765,6 +781,30 @@ async def test_manage_server_create_subscription_failure(ingestor_manager): ingestor_manager.opc_managers['server1'].subscribe.assert_not_called() +@mark.asyncio +async def test_manage_server_subscribe_and_unsubscribe_failure(ingestor_manager): + """subscribe() fails and so does the cleanup unsubscribe (legacy reader, session + gone): manage_server reports the failure and nothing escapes to the main loop""" + reader = AsyncMock() + reader.subscriptions = {} + reader.subscribe.side_effect = Exception('Subscription error') + reader.unsubscribe.side_effect = ConnectionError('session closed') + ingestor_manager.opc_managers = {'server1': reader} + server_config = {'tags': {'tag1': 'config1'}} + + result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config) + + assert result == 2 + reader.unsubscribe.assert_called_once_with('slot1') + notified = [ + c.kwargs['notification_id'] for c in ingestor_manager.send_notification_async.call_args_list + ] + assert notified == [ + 'OPC_SUBSCRIPTION_ERROR_slot1:server1', + 'OPC_UNSUBSCRIBE_ERROR_slot1:server1', + ] + + @mark.asyncio async def test_manage_server(ingestor_manager): ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()} diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index de8361f..5c404b3 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -341,11 +341,15 @@ async def test_update_ingestor_manager_no_ingestor_manager(ingestor_manager_star def _manager_with_real_unsubscribe(managed_tags, readers): - """A mocked IngestorManager whose unsubscribe_slot is the real one.""" + """A mocked IngestorManager whose unsubscribe path (unsubscribe_slot and + _unsubscribe_reader) is the real one.""" manager = AsyncMock(update_opc_servers=AsyncMock(), subscribe_to_tags=AsyncMock()) manager.managed_tags = managed_tags manager.opc_managers = readers manager.unsubscribe_slot = lambda slot: IngestorManager.unsubscribe_slot(manager, slot) + manager._unsubscribe_reader = lambda server, reader, slot: IngestorManager._unsubscribe_reader( + manager, server, reader, slot + ) return manager