fix(logs): do not log the MongoDB credentials

DataManager logged the full connection string, user and password included,
at startup. Log it through redact_uri (mongodb://***@host:port/...).

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

ingestor/ingestor.py

@@ -268,12 +268,14 @@ class Ingestor(SientiaMonitoring):
 
             overleases = list(self.ingestor_manager.managed_tags.keys())[1:]
 
-            await self.ingestor_manager.drop_slot_leases(overleases)
-
+            # Unsubscribe before giving the leases back: another pod may take a released
+            # slot right away, and must not find this pod still reading it
             for lease in overleases:
                 await self.ingestor_manager.unsubscribe_slot(lease)
                 self.ingestor_manager.managed_tags.pop(lease)
 
+            await self.ingestor_manager.drop_slot_leases(overleases)
+
             await self.emit_metric(
                 metric_object=metrics.SLOTS_MANAGED,
                 method='set',

ingestor/managers/data_manager.py

@@ -12,6 +12,22 @@ from sientia_do.temporal.constants import now
 import ingestor.metrics as metrics
 
 
+def redact_uri(uri: str) -> str:
+    """
+    Returns ``uri`` without the user and password, safe to log
+    (``mongodb://user:pass@host:27017/db`` -> ``mongodb://***@host:27017/db``).
+    """
+    prefix, sep, rest = uri.partition('://')
+    if not sep:
+        prefix, rest = '', uri
+    ends = [i for i in (rest.find('/'), rest.find('?')) if i != -1]
+    end = min(ends) if ends else len(rest)
+    authority, tail = rest[:end], rest[end:]
+    if '@' not in authority:
+        return uri
+    return f'{prefix}{sep}***@{authority.rpartition("@")[2]}{tail}'
+
+
 class DataManager(SientiaMonitoring):
     """
     Manages data persistence operations for the OPC Ingestor.
@@ -66,7 +82,8 @@ class DataManager(SientiaMonitoring):
         )
 
         logger.info(
-            f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
+            'Trying to initializing DataManager with MongoDB servers: '
+            f'{redact_uri(mongo_connection_string)}'
         )
 
         self.connection_string = mongo_connection_string
@@ -82,7 +99,9 @@ class DataManager(SientiaMonitoring):
 
         self.metadata = metadata
 
-        logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
+        logger.info(
+            f'DataManager initialized with MongoDB servers: {redact_uri(self.connection_string)}'
+        )
 
     def shutdown(self):
         """

ingestor/managers/ingestor_manager.py

@@ -538,26 +538,84 @@ 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.
+
+        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.
 
-        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)
+        # list() snapshots the readers: the loop awaits, so a reader added or removed
+        # meanwhile must not break the iteration. SonarQube flags it as an unnecessary
+        # list() (S7504); the issue is accepted in SonarQube for this reason.
+        for server, reader in list(self.opc_managers.items()):
+            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_<slot>:<server> 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."""
+        for slot in slots:
+            self.logger.warning(f'Slot {slot} is no longer configured; unsubscribing it')
+            await self.unsubscribe_slot(slot)
+            self.managed_tags.pop(slot, None)
 
     async def update_slot_config(self):
         """
@@ -567,7 +625,7 @@ class IngestorManager(SientiaMonitoring):
         This method performs the following steps:
         1. Renews the lease for each managed slot using the resource manager.
         2. Fetches the latest configuration for each slot.
-        3. Logs and removes slots whose configurations are no longer available.
+        3. Logs, unsubscribes and removes slots whose configurations are no longer available.
         4. Updates the configuration of slots if changes are detected.
         5. Unsubscribes and re-subscribes to slots with updated configurations.
         6. Removes slots from the managed tags if they are no longer valid.
@@ -600,8 +658,9 @@ class IngestorManager(SientiaMonitoring):
                 continue
             self.managed_tags[slot] = update
 
-        for slot in removed_slots:
-            self.managed_tags.pop(slot, None)
+        # Unsubscribe right away: leaving it to update_ingestor_manager's snapshot diff
+        # would leak the subscription if anything in between failed
+        await self._unsubscribe_removed_slots(removed_slots)
 
     def parse_slot_config(self, slot: str, raw: dict) -> dict | None:
         """
@@ -764,7 +823,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
 

ingestor/metrics.py

@@ -93,7 +93,7 @@ OPC_MANAGERS_ACTIVE = Gauge(
 )
 OPC_SUBSCRIPTION_ERRORS = Counter(
     'ingestor_opc_subscription_errors_total',
-    'Errors when trying to subscribe to OPC tags',
+    'Errors when trying to subscribe to or unsubscribe from OPC tags',
     ['pod_id', 'server', 'slot'],
 )
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "2.1.0-rc1"
+version = "2.2.0-rc1"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"

requirements.txt

@@ -1,9 +1,8 @@
 asyncua
 redis
-# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
-# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
-# Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
-sientia_do==1.12.2
+# 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1
+# defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'").
+sientia_do==1.12.5
 prometheus_client
 pymongo
 nats-py>=2.6

tests/unit/conftest.py

@@ -0,0 +1,20 @@
+from unittest.mock import AsyncMock, MagicMock
+
+from pytest import fixture
+
+
+@fixture
+def reader_double():
+    """Factory of OPC reader doubles holding subscriptions for the given slots.
+
+    Like the real readers (ServerReader, OpcManager), `subscriptions` is keyed by slot
+    and `unsubscribe` drops the slot from it.
+    """
+
+    def make(*slots):
+        reader = MagicMock()
+        reader.subscriptions = {slot: MagicMock() for slot in slots}
+        reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot))
+        return reader
+
+    return make

tests/unit/managers/test_data_manager.py

@@ -1,9 +1,10 @@
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
+import pytest
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
-from ingestor.managers.data_manager import DataManager
+from ingestor.managers.data_manager import DataManager, redact_uri
 
 metadata = {
     'metadata': {
@@ -54,6 +55,48 @@ def test___init___success(mongodb_repository):
     assert data_manager.connection_string == 'mongodb://localhost:27017'
 
 
+@patch('ingestor.managers.data_manager.MongoDBRepository')
+def test___init___does_not_log_the_credentials(mongodb_repository):
+    logger_mock = MagicMock()
+    uri = 'mongodb://root:s3cr3t@mongodb.mongodb.svc.cluster.local:27017'
+
+    data_manager = DataManager(
+        metadata=metadata['metadata'],
+        mongo_connection_string=uri,
+        mongo_database='sientia',
+        logger=logger_mock,
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+    logged = ' '.join(str(c) for c in logger_mock.info.call_args_list)
+    assert 's3cr3t' not in logged
+    assert 'root' not in logged
+    logger_mock.info.assert_any_call(
+        'DataManager initialized with MongoDB servers: '
+        'mongodb://***@mongodb.mongodb.svc.cluster.local:27017'
+    )
+    assert data_manager.connection_string == uri
+
+
+@pytest.mark.parametrize(
+    ('uri', 'expected'),
+    [
+        ('mongodb://localhost:27017', 'mongodb://localhost:27017'),
+        (
+            'mongodb://u:p@h1:27017,h2:27017/db?authSource=admin',
+            'mongodb://***@h1:27017,h2:27017/db?authSource=admin',
+        ),
+        ('mongodb://u:p@h:27017?authSource=admin', 'mongodb://***@h:27017?authSource=admin'),
+        ('mongodb+srv://u@cluster.example.com', 'mongodb+srv://***@cluster.example.com'),
+        ('u:p@h:27017', '***@h:27017'),
+        ('mongodb://h:27017/db?x=a@b', 'mongodb://h:27017/db?x=a@b'),
+    ],
+)
+def test_redact_uri(uri, expected):
+    assert redact_uri(uri) == expected
+
+
 def test_shutdown(data_manager):
     close_mock = MagicMock()
     data_manager.mongo_repository.close = close_mock

tests/unit/managers/test_ingestor_manager.py

@@ -1,13 +1,16 @@
 import gc
-from unittest.mock import AsyncMock, MagicMock, call, patch
+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
 from ingestor.managers.ingestor_manager import IngestorManager
+from ingestor.managers.opc_manager import OpcManager
+from ingestor.managers.server_reader import ServerReader
 
 metadata = {
     'metadata': {
@@ -548,21 +551,153 @@ async def test_get_slot_leases_1_failure(ingestor_manager):
 
 
 @mark.asyncio
-async def test_unsubscribe_slot(ingestor_manager):
+async def test_unsubscribe_slot(ingestor_manager, reader_double):
     ingestor_manager.managed_tags = {
         'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}},
         'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}},
     }
     ingestor_manager.opc_managers = {
-        'server1': AsyncMock(),
-        'server2': AsyncMock(),
-        'server3': AsyncMock(),
+        'server1': reader_double('slot1', 'slot2'),
+        'server2': reader_double('slot1'),
+        'server3': reader_double('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, reader_double):
+    """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_double('slot1', 'slot3'),
+        'server2': reader_double('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, reader_double):
+    """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_double('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
+async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, reader_double):
+    """A reader that fails to unsubscribe is notified and forgotten; the others still go"""
+    broken = reader_double('slot1')
+    broken.unsubscribe = AsyncMock(side_effect=ConnectionError('session closed'))
+    healthy = reader_double('slot1')
+    ingestor_manager.opc_managers = {'server1': broken, 'server2': healthy}
+
+    await ingestor_manager.unsubscribe_slot('slot1')
+
+    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',
+        message='Failed to unsubscribe slot slot1 from server1: session closed',
+        block='opc_manager',
+        level=NotificationLevel.ERROR,
+        attachment_content=ANY,
+    )
+
+
+@mark.asyncio
+async def test_unsubscribe_slot_with_real_readers(ingestor_manager):
+    """Contract: both real readers keep `subscriptions` keyed by slot, and
+    unsubscribe_slot closes the slot's subscription on each of them"""
+    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',
+            'subscription_period_ms': 1000.0,
+        },
+        pool=pool,
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+        metadata={},
+    )
+    server_reader.engine.emit_metric = AsyncMock()
+    await server_reader.create_subscription('slot1')
+
+    with patch('ingestor.managers.opc_manager.metrics'):
+        legacy_reader = OpcManager(
+            name='legacy',
+            url='opc.tcp://legacy:4840',
+            data_manager=AsyncMock(),
+            subscription_period_ms=1000,
+            logger=MagicMock(),
+            server_uri='opc.tcp://legacy:4840',
+            notification_handler=MagicMock(),
+            metadata={},
+            metrics_controller=AsyncMock(),
+        )
+    legacy_reader.emit_metric = AsyncMock()
+    legacy_subscription = AsyncMock()
+    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')
+
+    v2_subscription.delete.assert_awaited_once()
+    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')
+
+
+@mark.asyncio
+async def test_update_slot_config_unsubscribes_removed_slot(ingestor_manager, reader_double):
+    """A slot deleted from Redis is unsubscribed right away, not left to the main loop"""
+    ingestor_manager.managed_tags = {
+        'slot1': parse_slot(raw_slot_v1('Same')),
+        'slot2': parse_slot(raw_slot_v1('Same')),
+    }
+    reader = reader_double('slot1', 'slot2')
+    ingestor_manager.opc_managers = {'test_server': reader}
+    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
+        side_effect=lambda slot: raw_slot_v1('Same') if slot == 'slot1' else None
+    )
+    ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
+
+    await ingestor_manager.update_slot_config()
+
+    reader.unsubscribe.assert_called_once_with('slot2')
+    assert list(ingestor_manager.managed_tags) == ['slot1']
+    assert list(reader.subscriptions) == ['slot1']
 
 
 @mark.asyncio
@@ -646,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

@@ -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')
@@ -229,6 +231,10 @@ async def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_sta
     ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_called_once_with(
         ['tag2', 'tag3']
     )
+    # The extra slots are unsubscribed before their leases are given back
+    calls = [c[0] for c in ingestor_manager_started.ingestor_manager.mock_calls]
+    assert calls.index('drop_slot_leases') > calls.index('unsubscribe_slot')
+    assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['tag1']
 
 
 @mark.asyncio
@@ -332,3 +338,52 @@ 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 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
+
+
+@mark.asyncio
+async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started, reader_double):
+    """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_double('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, reader_double):
+    """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_double('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']