From 057c1fd1ce78a9d77607c114e451f0458da71e17 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Wed, 7 Oct 2026 15:55:20 -0300 Subject: [PATCH] SIENTIAPDE-1974: fix(slots): unsubscribe a slot from the readers that hold it The main loop snapshots managed_tags, then two steps remove a slot from managed_tags before update_ingestor_manager compares the snapshot with the current state and calls unsubscribe_slot for every slot that disappeared: - manage_leases drops an extra lease ("Extra slots available"): it already unsubscribed the slot, so the second call is redundant; - update_slot_config finds the slot gone from Redis (deleted by the orchestrator): nothing unsubscribed it yet. unsubscribe_slot looked the slot up in managed_tags[slot] to find its servers, so both cases raised KeyError, which stopped the main loop and restarted the container (seen on dev on 2026-10-06 18:30 during a rollout, ~40 s without reads). In the second case the subscription also stayed open on the OPC server. The same code is on main, so production has it too. unsubscribe_slot now asks the readers themselves: every reader (ServerReader or the legacy OpcManager, both keep `subscriptions` by slot) holding a subscription for the slot unsubscribes it. A slot no reader holds is a no-op. This also covers a resubscription whose new config dropped a server: the old server's subscription for that slot is closed too. Tests: unit tests for the new lookup and the double call, plus two regressions at the Ingestor level with the real unsubscribe_slot (dropped extra lease; slot removed from Redis). Without the fix they fail with the same KeyError at ingestor_manager.py:558. 280 unit tests. Co-Authored-By: Claude Opus 5.5 --- ingestor/managers/ingestor_manager.py | 24 +++++---- tests/unit/managers/test_ingestor_manager.py | 42 +++++++++++++-- tests/unit/test_ingestor.py | 54 ++++++++++++++++++++ 3 files changed, 106 insertions(+), 14 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 0be1773..caffaa9 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -538,26 +538,28 @@ class IngestorManager(SientiaMonitoring): async def unsubscribe_slot(self, slot: str): """ - Unsubscribes a specific slot from all associated OPC servers. + Unsubscribes a specific slot from every OPC reader that holds a subscription for it. - This method removes all subscriptions for a given slot across all - OPC servers that were managing it. It ensures clean cleanup of - resources when slots are released or reconfigured. + The readers' own subscriptions are the source of truth, not managed_tags: the + main loop removes a slot from managed_tags before update_ingestor_manager asks + to unsubscribe it, both when manage_leases drops an extra lease (which already + unsubscribed it) and when update_slot_config finds the slot gone from Redis + (which did not). Looking the slot up in managed_tags raised KeyError in both + cases, which stopped the main loop and restarted the container; the second case + also left the subscription open on the OPC server. Unsubscribing a slot that no + reader holds is a no-op. Args: slot (str): The name of the slot to unsubscribe. - Raises: - KeyError: If the specified slot does not exist in the managed tags. - Side Effects: - - Removes subscriptions from all OPC servers for the specified slot + - Removes the slot's subscription from every OPC reader that has one - Cleans up subscription resources on the OPC servers """ - for server in self.managed_tags[slot].keys(): - if server in self.opc_managers: - await self.opc_managers[server].unsubscribe(slot) + for reader in self.opc_managers.values(): + if slot in reader.subscriptions: + await reader.unsubscribe(slot) async def update_slot_config(self): """ diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index df27f66..f39b2ef 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -547,6 +547,14 @@ async def test_get_slot_leases_1_failure(ingestor_manager): assert result == {} +def _reader(*slots): + """An OPC reader double whose unsubscribe drops the slot, like the real readers.""" + reader = MagicMock() + reader.subscriptions = dict.fromkeys(slots, MagicMock()) + reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot)) + return reader + + @mark.asyncio async def test_unsubscribe_slot(ingestor_manager): ingestor_manager.managed_tags = { @@ -554,15 +562,43 @@ async def test_unsubscribe_slot(ingestor_manager): 'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}}, } ingestor_manager.opc_managers = { - 'server1': AsyncMock(), - 'server2': AsyncMock(), - 'server3': AsyncMock(), + 'server1': _reader('slot1', 'slot2'), + 'server2': _reader('slot1'), + 'server3': _reader('slot2'), } await ingestor_manager.unsubscribe_slot('slot1') ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1') ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot1') ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called() + assert list(ingestor_manager.opc_managers['server1'].subscriptions) == ['slot2'] + + +@mark.asyncio +async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager): + """The slot is already gone from managed_tags: the readers still holding it are unsubscribed""" + ingestor_manager.managed_tags = {'slot1': {'server1': {'tags': 'config1'}}} + ingestor_manager.opc_managers = { + 'server1': _reader('slot1', 'slot3'), + 'server2': _reader('slot3'), + } + + await ingestor_manager.unsubscribe_slot('slot3') + + ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot3') + ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot3') + + +@mark.asyncio +async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager): + """A second unsubscribe of the same slot (lease already dropped) does nothing and never raises""" + ingestor_manager.managed_tags = {} + ingestor_manager.opc_managers = {'server1': _reader('slot2')} + + await ingestor_manager.unsubscribe_slot('slot2') + await ingestor_manager.unsubscribe_slot('slot2') + + ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot2') @mark.asyncio diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 8bbccc4..bd2e03c 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,8 +1,10 @@ +from copy import deepcopy from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from ingestor.ingestor import Ingestor +from ingestor.managers.ingestor_manager import IngestorManager @patch('ingestor.ingestor.getenv') @@ -332,3 +334,55 @@ async def test_update_ingestor_manager_no_ingestor_manager(ingestor_manager_star ingestor_manager_started.ingestor_manager = None await ingestor_manager_started.update_ingestor_manager({}) + + +def _manager_with_real_unsubscribe(managed_tags, readers): + """A mocked IngestorManager whose unsubscribe_slot 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) + return manager + + +def _reader(*slots): + reader = MagicMock() + reader.subscriptions = dict.fromkeys(slots, MagicMock()) + reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot)) + return reader + + +@mark.asyncio +async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started): + """Regression: manage_leases drops slot 2, then update_ingestor_manager sees slot 2 in + the snapshot only and unsubscribes it again. That used to raise KeyError and stop + the main loop (container restart).""" + reader = _reader('1', '2') + ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( + {'1': {'kepserver': {}}, '2': {'kepserver': {}}}, {'kepserver': reader} + ) + snapshot = deepcopy(ingestor_manager_started.ingestor_manager.managed_tags) + + await ingestor_manager_started.manage_leases(0, 0, 1) + await ingestor_manager_started.update_ingestor_manager(snapshot) + + reader.unsubscribe.assert_called_once_with('2') + assert list(reader.subscriptions) == ['1'] + assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['1'] + + +@mark.asyncio +async def test_slot_removed_from_redis_is_unsubscribed(ingestor_manager_started): + """Regression: update_slot_config drops a slot deleted by the orchestrator from + managed_tags without unsubscribing it; update_ingestor_manager must close its + subscription instead of raising KeyError.""" + reader = _reader('1', '3') + ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe( + {'1': {'kepserver': {}}}, {'kepserver': reader} + ) + snapshot = {'1': {'kepserver': {}}, '3': {'kepserver': {}}} + + await ingestor_manager_started.update_ingestor_manager(snapshot) + + reader.unsubscribe.assert_called_once_with('3') + assert list(reader.subscriptions) == ['1']