SIENTIAPDE-1974: fix(read): code-review fixes for the slot v2 read path

- config: GatewayConfig.certs_from_env() reads only CERT_BUCKET/CERT_CACHE_DIR;
  IngestorManager uses it, so a malformed write/gRPC env var (GRPC_PORT,
  WRITE_MIN_PAUSE_S, IDLE_CONN_MAX_S) can no longer stop the reads.
- certs: CertProvider raises CertificateNotConfiguredError with an actionable
  message when no platform_certificates document exists; downloads once under
  an asyncio.Lock, caches the paths, writes atomically (tmp + os.replace);
  invalidate() forces a fresh download.
- pool: server_id normalized to str (1 and '1' share a session); warns when a
  get() asks for other url/server_uri/use_security than the pooled client was
  opened with (client reused, the slot config wins); invalidates the cert
  cache when a secured connect fails.
- SubscriptionEngine: tracks nodes per subscription and forgets them on
  unsubscribe (no more cycle alarms / tag count for removed tags), keeping
  addresses still monitored by another slot.
- ServerReader.shutdown(evict) / IngestorManager.remove_server(evict): a
  server that only left this pod's slots keeps the shared connection for
  NATS writes/gRPC; lost servers and config changes still evict.
- IngestorManager warns once per server when a slot still sets
  server_cert_path (server certificate pinning is no longer applied).
- slots: CONNECTION_FIELDS exported and reused by ServerReader.

Accepted, not changed: read metrics label server_name with the server_id
(same as the write path, which only knows the id); a malformed slot is
notified each lease attempt (alerts are deduplicated by notification_id).

Tests: 236 unit tests pass (+18); ruff clean; local e2e against an asyncua
server incl. slot moved (connection kept and reused by the write path) and
lost-server eviction/re-initialization.

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

ingestor/config.py

@@ -31,6 +31,16 @@ class GatewayConfig:
     write_min_pause_s: float = 0.0
     idle_conn_max_s: float = 300.0
 
+    @classmethod
+    def certs_from_env(cls) -> 'GatewayConfig':
+        """Only the certificate settings — what the read path needs. Plain strings,
+        no number parsing, so a malformed write/gRPC env var (GRPC_PORT,
+        WRITE_MIN_PAUSE_S, ...) can only disable the runtime, never the reads."""
+        return cls(
+            cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
+            cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
+        )
+
     @classmethod
     def from_env(cls) -> 'GatewayConfig':
         return cls(

ingestor/connection/certs.py

@@ -1,3 +1,5 @@
+import asyncio
+import os
 from os import getenv
 from pathlib import Path
 
@@ -8,10 +10,20 @@ from ingestor.config import _env_bool
 COLLECTION_NAME = 'platform_certificates'
 
 
+class CertificateNotConfiguredError(RuntimeError):
+    """No usable platform OPC client certificate is registered (Mongo + MinIO)."""
+
+
 class CertProvider:
     """Downloads the platform's client cert + private key (Mongo metadata,
     MinIO bytes) into a writable cache dir — pod root FS is read-only (§12),
-    so `cache_dir` must be an emptyDir/tmpfs mount."""
+    so `cache_dir` must be an emptyDir/tmpfs mount.
+
+    The certificate is single and shared by every connection (reads, writes,
+    gRPC), so it is downloaded once and cached; `invalidate()` forces a fresh
+    download (the pool calls it when a secured connection fails, e.g. after the
+    certificate was replaced in MinIO). The lock serializes downloads so two
+    connections never rewrite the cache files at the same time."""
 
     def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None):
         self.mongo = mongo
@@ -19,19 +31,44 @@ class CertProvider:
         self.cache_dir = cache_dir
         self.bucket = bucket
         self.metadata = metadata or {}
+        self._lock = asyncio.Lock()
+        self._paths: tuple[str, str] | None = None
 
     async def fetch(self) -> tuple[str, str]:
+        async with self._lock:
+            if self._paths is None:
+                self._paths = await self._download_all()
+            return self._paths
+
+    def invalidate(self) -> None:
+        self._paths = None
+
+    async def _download_all(self) -> tuple[str, str]:
         doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata)
+        try:
+            cert_object = doc['client_cert']['minio_object_name']
+            key_object = doc['private_key']['minio_object_name']
+        except (KeyError, TypeError) as e:
+            raise CertificateNotConfiguredError(
+                f'No OPC client certificate configured: the Mongo collection '
+                f"'{COLLECTION_NAME}' needs a document with client_cert.minio_object_name "
+                f'and private_key.minio_object_name (objects in the MinIO bucket '
+                f"'{self.bucket}'). Register the platform certificate before using "
+                f'secured OPC servers.'
+            ) from e
         Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
-        cert_path = await self._download(doc['client_cert']['minio_object_name'])
+        cert_path = await self._download(cert_object)
-        key_path = await self._download(doc['private_key']['minio_object_name'])
+        key_path = await self._download(key_object)
         return cert_path, key_path
 
     async def _download(self, object_name: str) -> str:
         data = await self.minio.download_file(object_name, self.bucket, self.metadata)
-        local_path = str(Path(self.cache_dir) / Path(object_name).name)
+        local_path = Path(self.cache_dir) / Path(object_name).name
-        Path(local_path).write_bytes(data)
+        # Atomic replace: a connection loading the previous file never sees it half-written.
-        return local_path
+        tmp_path = local_path.with_name(local_path.name + '.tmp')
+        tmp_path.write_bytes(data)
+        os.replace(tmp_path, local_path)
+        return str(local_path)
 
 
 class LazyMinio:

ingestor/connection/pool.py

@@ -30,6 +30,8 @@ class OpcConnectionPool(SientiaMonitoring):
         # callbacks and never calls get(), so without a pin release_idle() would
         # evict it as idle and silently stop the reads (SIENTIAPDE-1974 G4).
         self._pins: dict[str, int] = {}
+        # Connection settings each pooled client was opened with (see _check_cfg).
+        self._configs: dict[str, dict] = {}
         # ponytail: one Lock per server_id, never cleaned up — server set is
         # small and bounded. Add eviction of stale locks only if it grows.
         self._locks: dict[str, asyncio.Lock] = {}
@@ -45,16 +47,35 @@ class OpcConnectionPool(SientiaMonitoring):
         """Return the cached client for `server_id`, connecting one if needed.
 
         Double-checked under a per-server lock so concurrent callers for the
-        same new server share one connection (no leaked, untracked sessions)."""
+        same new server share one connection (no leaked, untracked sessions).
+        `server_id` is normalized to str, so 1 and '1' share the same session."""
+        server_id = str(server_id)
         client = self._cached(server_id)
         if client is not None:
+            self._check_cfg(server_id, server_cfg)
             return client
         async with self._locks.setdefault(server_id, asyncio.Lock()):
             client = self._cached(server_id)  # re-check: a racer may have connected
             if client is not None:
+                self._check_cfg(server_id, server_cfg)
                 return client
             return await self._open(server_id, server_cfg)
 
+    def _check_cfg(self, server_id: str, server_cfg: dict) -> None:
+        """Warn when a caller asks for a server with other connection settings than
+        the pooled client was opened with. The client is reused as is: reconnecting
+        on every mismatch would let one malformed write command drop the reads.
+        Config changes from the slot evict the client first (ServerReader.shutdown)."""
+        opened = self._configs.get(server_id, {})
+        diff = [
+            k for k in ('url', 'server_uri', 'use_security') if opened.get(k) != server_cfg.get(k)
+        ]
+        if diff:
+            self.logger.warning(
+                f'OPC server {server_id}: reusing the pooled connection, but {diff} differ '
+                f'from the settings it was opened with'
+            )
+
     def _cached(self, server_id: str) -> Client | None:
         entry = self._connections.get(server_id)
         if entry is None:
@@ -67,6 +88,10 @@ class OpcConnectionPool(SientiaMonitoring):
         try:
             client = await self._connect(server_cfg)
         except Exception:
+            if server_cfg.get('use_security'):
+                # The cached certificate may be the cause (replaced in MinIO,
+                # rejected by the server): download it again on the next attempt.
+                self.cert_provider.invalidate()
             await self.emit_metric(
                 metric_object=metrics.OPC_CONNECTIONS_FAILED,
                 tags={'pod_id': self.pod_id, 'server_name': server_id},
@@ -74,6 +99,7 @@ class OpcConnectionPool(SientiaMonitoring):
             raise
 
         self._connections[server_id] = (client, self.clock.elapsed())
+        self._configs[server_id] = dict(server_cfg)
         await self._record_connected(server_id, server_cfg['url'])
         return client
 
@@ -102,10 +128,12 @@ class OpcConnectionPool(SientiaMonitoring):
 
     def pin(self, server_id: str) -> None:
         """Mark one more active subscription on `server_id` (never evicted as idle)."""
+        server_id = str(server_id)
         self._pins[server_id] = self._pins.get(server_id, 0) + 1
 
     def unpin(self, server_id: str) -> None:
         """Drop one active subscription mark from `server_id`."""
+        server_id = str(server_id)
         count = self._pins.get(server_id, 0) - 1
         if count > 0:
             self._pins[server_id] = count
@@ -127,6 +155,7 @@ class OpcConnectionPool(SientiaMonitoring):
     async def evict(self, server_id: str):
         """Disconnect `server_id` now (lost server or changed config); the next
         get() reconnects. No-op when the server has no pooled client."""
+        server_id = str(server_id)
         if server_id in self._connections:
             await self._evict(server_id)
 
@@ -139,6 +168,7 @@ class OpcConnectionPool(SientiaMonitoring):
 
     async def _evict(self, server_id: str):
         client, _ = self._connections.pop(server_id)
+        self._configs.pop(server_id, None)
         await self._disconnect_with_retries(client, server_id)
         await self.emit_metric(
             metric_object=metrics.OPC_DISCONNECTIONS_TOTAL,

ingestor/engines/subscription.py

@@ -35,6 +35,9 @@ class SubscriptionEngine(SientiaMonitoring):
         self.subscriptions: dict = {}
         self._clients: dict = {}
         self._servers: dict = {}
+        # Addresses each subscription monitors, so unsubscribe can forget them
+        # (otherwise check_cycles keeps alarming on tags nobody reads anymore).
+        self._sub_nodes: dict[str, set[str]] = {}
         self.non_receive_count = 0
 
         SientiaMonitoring.__init__(
@@ -75,6 +78,7 @@ class SubscriptionEngine(SientiaMonitoring):
         server_id = self._servers[subscription]
         addr_nodes = [client.get_node(n) for n in nodes]
         self.nodes.update(nodes)
+        self._sub_nodes.setdefault(subscription, set()).update(nodes)
         self._init_cycle_rules(nodes, collect_period, server_id)
 
         await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
@@ -102,6 +106,7 @@ class SubscriptionEngine(SientiaMonitoring):
         server_id = self._servers.pop(subscription, None)
         if server_id is not None:
             self.pool.unpin(server_id)
+        self._forget_nodes(subscription)
         try:
             await sub.delete()
         except Exception as e:
@@ -109,6 +114,22 @@ class SubscriptionEngine(SientiaMonitoring):
             # dropped from the books either way so the slot can be re-subscribed.
             self.logger.warning(f'Failed to delete subscription {subscription}: {e}')
         self.logger.info(f'Unsubscribed from {subscription}.')
+        if server_id is not None:
+            await self.emit_metric(
+                metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+                method='set',
+                value=len(self.nodes),
+                tags={'pod_id': self.pod_id, 'server_name': server_id},
+            )
+
+    def _forget_nodes(self, subscription: str) -> None:
+        """Drop the subscription's addresses, except those another subscription
+        of this engine still monitors."""
+        nodes = self._sub_nodes.pop(subscription, set())
+        still_monitored = set().union(*self._sub_nodes.values())
+        for node in nodes:
+            if node not in still_monitored:
+                self.nodes.pop(node, None)
 
     async def datachange_notification(self, node, _val, data):
         """asyncua subscription callback: fan the new value out to every name

ingestor/io/slots.py

@@ -11,6 +11,8 @@ any order (SIENTIAPDE-1974):
 """
 
 SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms')
+# What a connection/subscription needs (the pool keys clients by server_id itself).
+CONNECTION_FIELDS = tuple(f for f in SERVER_FIELDS if f != 'server_id')
 
 
 def _parse_names(address: str, tag: dict) -> list[dict]:

ingestor/managers/ingestor_manager.py

@@ -102,7 +102,7 @@ class IngestorManager(SientiaMonitoring):
         # Read path (SIENTIAPDE-1974): pooled connections with the certificate from
         # MinIO. Nothing here touches the network at construction (MinIO is lazy), so
         # a MinIO/Postgres/NATS outage can't stop the reads before they start.
-        config = gateway_config or GatewayConfig.from_env()
+        config = gateway_config or GatewayConfig.certs_from_env()
         self.cert_provider = build_cert_provider(
             mongo=self.data_manager.mongo_repository,
             config=config,
@@ -115,6 +115,7 @@ class IngestorManager(SientiaMonitoring):
             self.cert_provider, logger, notification_handler, metrics_controller
         )
         self.sink = DataSink(self.data_manager, None)
+        self._server_cert_warned: set[str] = set()
         self.resource_manager = ResourceManager(
             host=redis_host,
             port=redis_port,
@@ -198,12 +199,18 @@ class IngestorManager(SientiaMonitoring):
         await self.pool.close_all()
         self.data_manager.shutdown()
 
-    async def remove_server(self, server: str):
+    async def remove_server(self, server: str, evict: bool = True):
         """
         Removes an OPC server from the ingestor.
+
+        Args:
+            server (str): The server name.
+            evict (bool): Also drop the pooled connection (lost server, failed
+                subscription). False when the server merely left this pod's slots:
+                the shared connection stays for writes/gRPC until the idle sweep.
         """
         if server in self.opc_managers:
-            await self.opc_managers[server].shutdown()
+            await self.opc_managers[server].shutdown(evict=evict)
             del self.opc_managers[server]
         for slot, _config in self.managed_tags.items():
             self.managed_tags[slot].pop(server, None)
@@ -269,9 +276,10 @@ class IngestorManager(SientiaMonitoring):
         for server in servers:
             if server not in registered_servers:
                 self.logger.warning(
-                    f'Server {server} not found in managed tags. Desconnecting from server.'
+                    f'Server {server} not found in managed tags. Unsubscribing from server.'
                 )
-                await self.remove_server(server)
+                # Slot moved away: keep the shared connection for writes/gRPC.
+                await self.remove_server(server, evict=False)
 
         await self.emit_metric(
             metric_object=metrics.OPC_MANAGERS_ACTIVE,
@@ -559,7 +567,7 @@ class IngestorManager(SientiaMonitoring):
             (logged and notified).
         """
         try:
-            return parse_slot(raw)
+            parsed = parse_slot(raw)
         except (AttributeError, KeyError, TypeError, ValueError) as e:
             self.send_notification(
                 metadata=self.metadata,
@@ -570,6 +578,20 @@ class IngestorManager(SientiaMonitoring):
             )
             self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}')
             return None
+        self._warn_server_cert_path(raw)
+        return parsed
+
+    def _warn_server_cert_path(self, raw: dict) -> None:
+        """Logs once per server when a slot still pins a server certificate
+        (`server_cert_path`): connections now come from the pool with the platform
+        certificate from MinIO, and that file-based pin is no longer applied."""
+        for name, server in raw.items():
+            if server.get('server_cert_path') and name not in self._server_cert_warned:
+                self._server_cert_warned.add(name)
+                self.logger.warning(
+                    f'OPC server {name}: server_cert_path is set in the slot but is no '
+                    f'longer applied (server certificate validation is disabled)'
+                )
 
     async def drop_slot_leases(self, ids: list[str]) -> None:
         """

ingestor/managers/server_reader.py

@@ -20,8 +20,7 @@ import ingestor.metrics as metrics
 from ingestor.connection.pool import OpcConnectionPool
 from ingestor.engines.subscription import SubscriptionEngine
 from ingestor.io.data_sink import DataSink
-
+from ingestor.io.slots import CONNECTION_FIELDS
-_CONNECTION_FIELDS = ('url', 'server_uri', 'use_security', 'subscription_period_ms')
 
 
 class ServerReader:
@@ -42,7 +41,7 @@ class ServerReader:
         self.name = name
         self.server_id = str(server_config['server_id'])
         self.url = server_config['url']
-        self.server_cfg = {k: server_config[k] for k in _CONNECTION_FIELDS if k in server_config}
+        self.server_cfg = {k: server_config[k] for k in CONNECTION_FIELDS if k in server_config}
         self.pool = pool
         self.logger = logger
         self.config: dict | None = None
@@ -77,22 +76,27 @@ class ServerReader:
     async def check_opc_listenning(self) -> bool:
         return await self.engine.check_opc_listenning(self.server_id)
 
-    async def shutdown(self):
+    async def shutdown(self, evict: bool = True):
-        """Drop every subscription and the pooled connection, so the next
+        """Drop every subscription of this server.
-        initialization reconnects (lost server or changed server config)."""
+
+        With `evict` (lost server or changed server config) the pooled connection is
+        also dropped, so the next initialization reconnects. Without it (the slot just
+        moved to another pod) the connection stays for writes/gRPC: the connection is
+        shared, and the idle sweep closes it if nobody uses it anymore."""
         for slot in list(self.engine.subscriptions):
             await self.engine.unsubscribe(slot)
-        await self.pool.evict(self.server_id)
+        if evict:
-        await self.engine.emit_metric(
+            await self.pool.evict(self.server_id)
-            metric_object=metrics.OPC_CONNECTION_STATUS,
+            await self.engine.emit_metric(
-            method='set',
+                metric_object=metrics.OPC_CONNECTION_STATUS,
-            value=0,
+                method='set',
-            tags={
+                value=0,
-                'pod_id': self.engine.pod_id,
+                tags={
-                'server_name': self.server_id,
+                    'pod_id': self.engine.pod_id,
-                'server_url': self.url,
+                    'server_name': self.server_id,
-            },
+                    'server_url': self.url,
-        )
+                },
+            )
         await self.engine.emit_metric(
             metric_object=metrics.OPC_TAGS_SUBSCRIBED,
             method='set',

tests/unit/connection/test_certs.py

@@ -70,3 +70,79 @@ def test_build_cert_provider_is_lazy_and_uses_config(monkeypatch):
     assert provider.cache_dir == 'certs-cache'
     assert isinstance(provider.minio, certs.LazyMinio)
     minio_cls.assert_not_called()  # MinIO is only reached when a cert is needed
+
+
+def _provider(tmp_path, doc=None, data=b'pem'):
+    from unittest.mock import MagicMock
+
+    mongo = MagicMock()
+    mongo.find_one = AsyncMock(
+        return_value=doc
+        if doc is not None
+        else {
+            'client_cert': {'minio_object_name': 'platform_certificates/c.pem'},
+            'private_key': {'minio_object_name': 'platform_certificates/k.pem'},
+        }
+    )
+    minio = MagicMock()
+    minio.download_file = AsyncMock(return_value=data)
+    return CertProvider(mongo=mongo, minio=minio, cache_dir=str(tmp_path), bucket='b'), minio
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'doc',
+    [
+        None,
+        {},
+        {'client_cert': {'minio_object_name': 'c.pem'}},
+        {'client_cert': None, 'private_key': {}},
+    ],
+)
+async def test_fetch_without_platform_certificate_raises_clear_error(tmp_path, doc):
+    from ingestor.connection.certs import CertificateNotConfiguredError
+
+    provider, minio = _provider(tmp_path)
+    provider.mongo.find_one.return_value = doc
+
+    with pytest.raises(CertificateNotConfiguredError, match='platform_certificates'):
+        await provider.fetch()
+    minio.download_file.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fetch_is_cached_until_invalidated(tmp_path):
+    provider, minio = _provider(tmp_path)
+
+    first = await provider.fetch()
+    second = await provider.fetch()
+    assert first == second
+    assert minio.download_file.await_count == 2  # cert + key, once
+
+    provider.invalidate()
+    await provider.fetch()
+    assert minio.download_file.await_count == 4
+
+
+@pytest.mark.asyncio
+async def test_concurrent_fetches_download_once(tmp_path):
+    import asyncio
+
+    provider, minio = _provider(tmp_path)
+
+    results = await asyncio.gather(*(provider.fetch() for _ in range(5)))
+
+    assert len(set(results)) == 1
+    assert minio.download_file.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_download_replaces_file_atomically(tmp_path):
+    provider, minio = _provider(tmp_path, data=b'new')
+    (tmp_path / 'c.pem').write_bytes(b'old')
+
+    cert_path, _ = await provider.fetch()
+
+    assert (tmp_path / 'c.pem').read_bytes() == b'new'
+    assert cert_path == str(tmp_path / 'c.pem')
+    assert not list(tmp_path.glob('*.tmp'))

tests/unit/connection/test_pool.py

@@ -248,3 +248,63 @@ async def test_evict_unknown_server_is_noop():
     await pool.evict('nope')
 
     pool.emit_metric.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_get_normalizes_server_id_to_str():
+    pool = make_pool()
+    fake_client = AsyncMock()
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        c1 = await pool.get(1, {'url': 'opc.tcp://h:4840', 'use_security': False})
+        c2 = await pool.get('1', {'url': 'opc.tcp://h:4840', 'use_security': False})
+
+    assert c1 is c2
+    assert fake_client.connect.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_get_with_different_cfg_reuses_client_and_warns():
+    pool = make_pool()
+    fake_client = AsyncMock()
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
+        client = await pool.get('srv', {'url': 'opc.tcp://10.0.0.5:4840', 'use_security': False})
+
+    assert client is fake_client
+    assert fake_client.connect.await_count == 1
+    pool.logger.warning.assert_called_once()
+    assert 'url' in pool.logger.warning.call_args.args[0]
+
+
+@pytest.mark.asyncio
+async def test_secured_connect_failure_invalidates_cert_cache():
+    from unittest.mock import MagicMock
+
+    cert_provider = MagicMock()
+    cert_provider.fetch = AsyncMock(return_value=('cert.pem', 'key.pem'))
+    pool = make_pool(cert_provider=cert_provider)
+    fake_client = AsyncMock()
+    fake_client.connect.side_effect = RuntimeError('BadCertificateUntrusted')
+    with (
+        patch('ingestor.connection.pool.Client', return_value=fake_client),
+        patch('ingestor.connection.pool.apply_security', AsyncMock()),
+    ):
+        with pytest.raises(RuntimeError):
+            await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': True})
+
+    cert_provider.invalidate.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_insecure_connect_failure_keeps_cert_cache():
+    from unittest.mock import MagicMock
+
+    cert_provider = MagicMock()
+    pool = make_pool(cert_provider=cert_provider)
+    fake_client = AsyncMock()
+    fake_client.connect.side_effect = RuntimeError('unreachable')
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        with pytest.raises(RuntimeError):
+            await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
+
+    cert_provider.invalidate.assert_not_called()

tests/unit/engines/test_subscription.py

@@ -255,3 +255,46 @@ async def test_unsubscribe_tolerates_dead_connection():
     assert 'slot1' not in eng.subscriptions
     eng.pool.unpin.assert_called_once_with('1')
     eng.logger.warning.assert_called_once()
+
+
+async def _subscribed_engine(slots):
+    """Engine with one subscription per slot: {slot: [addresses]}."""
+    eng = make_engine()
+    eng.pool = MagicMock(
+        get=AsyncMock(
+            return_value=MagicMock(create_subscription=AsyncMock(return_value=AsyncMock()))
+        )
+    )
+    for slot, addresses in slots.items():
+        await eng.create_subscription('1', {'subscription_period_ms': 500}, slot)
+        nodes = {
+            a: {'frequency': '1000', 'names': [{'collection': 'c', 'tag_name': a}]}
+            for a in addresses
+        }
+        await eng.subscribe(slot, nodes, 10)
+    return eng
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_forgets_the_slot_nodes():
+    eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=3']})
+
+    await eng.unsubscribe('s1')
+
+    # no more cycle checks / tag counting for addresses nobody reads anymore
+    assert set(eng.nodes) == {'ns=2;i=3'}
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+        method='set',
+        value=1,
+        tags={'pod_id': eng.pod_id, 'server_name': '1'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_keeps_node_shared_with_another_slot():
+    eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=2']})
+
+    await eng.unsubscribe('s1')
+
+    assert set(eng.nodes) == {'ns=2;i=2'}

tests/unit/managers/test_ingestor_manager.py

@@ -314,7 +314,9 @@ async def test_update_opc_servers(metrics, ingestor_manager):
     assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'}
     assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'}
 
-    ingestor_manager.remove_server.assert_any_call('server4')
+    # server4 left this pod's slots: unsubscribe only, the connection stays for writes
+    ingestor_manager.remove_server.assert_any_call('server4', evict=False)
+    # server5 failed to initialize: removed with the default (evict)
     ingestor_manager.remove_server.assert_any_call('server5')
     assert ingestor_manager.remove_server.call_count == 2
 
@@ -738,3 +740,55 @@ async def test_update_slot_config_keeps_last_good_config_when_malformed(ingestor
     assert ingestor_manager.managed_tags == {'slot1': good}
     _, kwargs = ingestor_manager.send_notification.call_args
     assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_slot1'
+
+
+@mark.asyncio
+async def test_remove_server_forwards_evict(ingestor_manager):
+    reader = AsyncMock()
+    ingestor_manager.opc_managers = {'server1': reader}
+    ingestor_manager.managed_tags = {'slot1': {'server1': {}}}
+
+    await ingestor_manager.remove_server('server1', evict=False)
+
+    reader.shutdown.assert_awaited_once_with(evict=False)
+    assert 'server1' not in ingestor_manager.opc_managers
+    assert ingestor_manager.managed_tags == {'slot1': {}}
+
+
+@patch('ingestor.managers.ingestor_manager.DataManager')
+@patch('ingestor.managers.ingestor_manager.ResourceManager')
+def test___init___ignores_malformed_write_env(
+    resource_manager_mock, data_manager_mock, monkeypatch
+):
+    """A malformed write/gRPC-only env var must not stop the read path (review #1)."""
+    monkeypatch.setenv('GRPC_PORT', '')
+    monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0,5')
+    monkeypatch.setenv('CERT_BUCKET', 'streamlit-connectors')
+
+    ingestor = IngestorManager(
+        redis_data={'host': 'localhost', 'port': 6379},
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        poll_interval=5,
+        mongo_connection_string='mongodb://localhost:27017',
+        mongo_database='sientia',
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metadata=metadata['metadata'],
+        metrics_controller=MagicMock(),
+    )
+
+    assert ingestor.cert_provider.bucket == 'streamlit-connectors'
+
+
+def test_parse_slot_config_warns_once_about_server_cert_path(ingestor_manager):
+    raw = raw_slot_v1()
+    raw['server1']['server_cert_path'] = '/app/certs/kepserver.der'
+
+    ingestor_manager.parse_slot_config('1', raw)
+    ingestor_manager.parse_slot_config('1', raw)  # every loop refreshes the slot
+
+    warnings = [
+        c for c in ingestor_manager.logger.warning.call_args_list if 'server_cert_path' in str(c)
+    ]
+    assert len(warnings) == 1

tests/unit/managers/test_server_reader.py

@@ -127,3 +127,18 @@ async def test_shutdown_unsubscribes_all_slots_evicts_and_zeroes_metrics():
         value=0,
         tags={'pod_id': 'pod', 'server_name': '1'},
     )
+
+
+@pytest.mark.asyncio
+async def test_shutdown_without_evict_keeps_the_shared_connection():
+    """Slot moved to another pod: drop the subscriptions, keep the pooled
+    connection for NATS writes/gRPC (review #7)."""
+    reader, pool = make_reader()
+    reader.engine.subscriptions.update({'1': MagicMock()})
+
+    await reader.shutdown(evict=False)
+
+    reader.engine.unsubscribe.assert_awaited_once_with('1')
+    pool.evict.assert_not_awaited()
+    for c in reader.engine.emit_metric.await_args_list:
+        assert c.kwargs['metric_object'] is not metrics.OPC_CONNECTION_STATUS

tests/unit/test_config.py

@@ -69,3 +69,19 @@ def test_bool_parsing_edge_cases(monkeypatch, value, expected):
     cfg = GatewayConfig.from_env()
 
     assert cfg.grpc_enabled is expected
+
+
+def test_certs_from_env_ignores_malformed_numeric_vars(monkeypatch):
+    import pytest
+
+    monkeypatch.setenv('GRPC_PORT', '')
+    monkeypatch.setenv('IDLE_CONN_MAX_S', 'abc')
+    monkeypatch.setenv('CERT_BUCKET', 'my-bucket')
+    monkeypatch.setenv('CERT_CACHE_DIR', 'certs-cache')
+
+    cfg = GatewayConfig.certs_from_env()
+
+    assert cfg.cert_bucket == 'my-bucket'
+    assert cfg.cert_cache_dir == 'certs-cache'
+    with pytest.raises(ValueError):
+        GatewayConfig.from_env()  # the full config still fails, only the runtime uses it