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 <noreply@anthropic.com>

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):
         """

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

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']