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

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()
+        A reader that fails to unsubscribe is handled by _unsubscribe_reader: the other
-        raises when the server session is gone) is notified and logged, and the slot is
+        readers are still unsubscribed and the main loop goes on.
-        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.
 
         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_<slot>:<server> 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:
+            if slot in reader.subscriptions:
-                continue
+                await self._unsubscribe_reader(server, reader, slot)
-            try:
+
-                await reader.unsubscribe(slot)
+    async def _unsubscribe_reader(
-            except Exception as e:
+        self, server: str, reader: ServerReader | OpcManager, slot: str
-                trace = traceback.format_exc()
+    ) -> None:
-                await self.send_notification_async(
+        """
-                    metadata=self.metadata,
+        Unsubscribe one reader from a slot without letting a failure stop the main loop.
-                    notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}',
+
-                    message=f'Failed to unsubscribe slot {slot} from {server}: {e}',
+        A failure (e.g. the legacy OpcManager, whose delete() raises when the server
-                    block='opc_manager',
+        session is gone) increments OPC_SUBSCRIPTION_ERRORS, sends
-                    level=NotificationLevel.ERROR,
+        OPC_UNSUBSCRIBE_ERROR_<slot>:<server> and is logged. The slot is dropped from the
-                    attachment_content=trace,
+        reader's books, as the v2 engine already does, so it can be subscribed again.
-                )
+
-                self.logger.error(trace)
+        Known limitation: if delete() failed while the session is still alive (e.g. a
-                reader.subscriptions.pop(slot, None)
+        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
 

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()
+    await server_reader.create_subscription('slot1')
-    server_reader.engine.subscriptions['slot1'] = v2_subscription
-    server_reader.engine._servers['slot1'] = '1'
 
     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()}

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