SIENTIAPDE-1975: align the gateway write path with the Laborious

- GW1: write commands (C2) carry only the server_id; ServerConfigResolver
  resolves the connection (command override, live pooled connection,
  connectors, opc-servers) with a 60 s cache and a 10 s negative cache.
- GW2: write_status keeps the worst outcome (ok < error < session_bad,
  pending/NULL replaced), same rule as the Laborious, built with SQLAlchemy
  Core (no SQL formatting).
- GW3: update_write_status never raises; outcome (updated, kept,
  row_not_found, invalid, error) logged and counted
  (opc_write_status_update_total); runs off the event loop.
- GW4: the reason of a failed write is logged; opc_write_status_by_model
  shows the last outcome; a bad timestamp is a failed write, not an
  exception.
- GW5: the NATS first connect gives up after 5 s (nats-py's defaults block
  the main loop for about 4 minutes while NATS is down); reconnection after
  it is unbounded; idle_sweep restarts a connection closed for good; stop()
  is safe on a closed connection.

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

ingestor/adapters/nats_consumer.py

@@ -3,12 +3,14 @@
 Thin adapter: decode the message JSON and hand the dict to WriteEngine,
 which is already raise-safe against malformed commands. No metrics of its
 own — the engine records write metrics; a malformed-JSON drop is logged
-here. No reconnect logic — nats-py handles reconnection internally.
+here. Once connected, nats-py reconnects by itself, forever.
 """
 
+import asyncio
 import json
+import math
+import traceback
 
-import nats
 from nats.aio.client import Client
 from nats.aio.subscription import Subscription
 from sientia_do.observability.logger import Logger
@@ -24,18 +26,55 @@ class NatsWriteConsumer:
         logger: Logger,
         subject: str = 'opc.write',
         queue: str = 'gateway-writers',
+        connect_timeout_s: float = 5.0,
     ):
         self.nats_url = nats_url
         self.write_engine = write_engine
         self.logger = logger
         self.subject = subject
         self.queue = queue
+        self.connect_timeout_s = connect_timeout_s
         self.nc: Client | None = None
         self.sub: Subscription | None = None
 
+    @property
+    def closed(self) -> bool:
+        """True when there is no connection, or it was closed for good."""
+        return self.nc is None or self.nc.is_closed
+
     async def start(self) -> None:
-        self.nc = await nats.connect(servers=self.nats_url)
-        self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        """Connect and subscribe, giving up after ``connect_timeout_s``.
+
+        It runs from the main loop (startup and ``GatewayRuntime.idle_sweep``), so it
+        must never block it: nats-py's defaults retry the first connection 60 times,
+        about 2 minutes, longer than the heartbeat TTL, and with
+        ``max_reconnect_attempts=-1`` they would retry it forever. Once connected,
+        reconnection is unbounded, so an outage never closes the connection for good.
+
+        Raises:
+            Exception: When the connection or the subscription failed; nothing is left
+                open in that case.
+        """
+        nc = Client()
+        try:
+            await asyncio.wait_for(
+                nc.connect(
+                    servers=self.nats_url,
+                    connect_timeout=math.ceil(self.connect_timeout_s),
+                    max_reconnect_attempts=-1,
+                    name='opc-gateway',
+                    # One log line per event instead of nats-py's traceback per attempt.
+                    error_cb=self._on_error,
+                    disconnected_cb=self._on_disconnected,
+                    reconnected_cb=self._on_reconnected,
+                ),
+                timeout=self.connect_timeout_s,
+            )
+            self.sub = await nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        except (Exception, asyncio.CancelledError):
+            await self._discard(nc)
+            raise
+        self.nc = nc
         self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
 
     async def _handle(self, msg) -> None:
@@ -45,11 +84,39 @@ class NatsWriteConsumer:
             preview = msg.data[:200]
             self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
             return
-        await self.write_engine.write(payload)
+        try:
+            await self.write_engine.write(payload)
+        except Exception as e:
+            # The engine is raise-safe; this only keeps an unexpected bug from going to
+            # nats-py's error callback without context.
+            self.logger.error(
+                f'NatsWriteConsumer: unexpected error handling a write command ({e}):\n'
+                f'{traceback.format_exc()}'
+            )
+
+    async def _on_error(self, e: Exception) -> None:
+        self.logger.warning(f'NATS error: {type(e).__name__} {e}')
+
+    async def _on_disconnected(self) -> None:
+        self.logger.warning('Disconnected from NATS; reconnecting')
+
+    async def _on_reconnected(self) -> None:
+        self.logger.info('Reconnected to NATS')
+
+    async def _discard(self, nc: Client) -> None:
+        try:
+            await nc.close()
+        except Exception as e:
+            self.logger.warning(f'Error closing a failed NATS connection: {e}')
 
     async def stop(self) -> None:
         if self.nc is None:
             return
+        if self.nc.is_closed:
+            # Closed for good already: unsubscribe/drain would raise ConnectionClosedError
+            # and abort the shutdown sequence.
+            self.nc = None
+            return
         if self.sub is not None:
             await self.sub.unsubscribe()
         await self.nc.drain()

ingestor/connection/pool.py

@@ -76,6 +76,16 @@ class OpcConnectionPool(SientiaMonitoring):
                 f'from the settings it was opened with'
             )
 
+    def config_for(self, server_id: str) -> dict | None:
+        """Connection settings of the pooled client for `server_id`, or None when the
+        server has no live pooled connection. Lets a writer reuse exactly the settings
+        the read path opened the connection with."""
+        server_id = str(server_id)
+        if server_id not in self._connections:
+            return None
+        cfg = self._configs.get(server_id)
+        return dict(cfg) if cfg is not None else None
+
     def _cached(self, server_id: str) -> Client | None:
         entry = self._connections.get(server_id)
         if entry is None:

ingestor/connection/servers.py

@@ -0,0 +1,120 @@
+"""Connection settings of the OPC server a write command targets (SIENTIAPDE-1975 GW1).
+
+The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`); the
+connection (url, server URI, security) belongs to the gateway, which owns every OPC
+session. It is resolved with the same rule the orchestrator uses to build the read slots
+(SIENTIAPDE-1974): the tag database `connectors` first, the legacy `opc-servers` otherwise.
+"""
+
+from typing import Any
+
+from sientia_do.observability.logger import Logger
+
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+
+CONNECTORS = 'connectors'
+OPC_SERVERS = 'opc-servers'
+
+
+def _id_forms(server_id: str) -> list[Any]:
+    """`server_id` as stored: the tag database keeps ints, opc-servers strings or ints."""
+    forms: list[Any] = [server_id]
+    if server_id.isdigit():
+        forms.append(int(server_id))
+    return forms
+
+
+def _connector_cfg(connector: dict[str, Any]) -> dict[str, Any] | None:
+    if not connector.get('url'):
+        return None
+    return {
+        'url': connector['url'],
+        'server_uri': connector.get('server_uri'),
+        'use_security': bool(connector.get('use_security', False)),
+    }
+
+
+def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None:
+    if not server.get('url'):
+        return None
+    return {
+        'url': server['url'],
+        'server_uri': server.get('uri'),
+        # The certificate comes from MinIO now: a cert_path only says the server uses
+        # security, as in the orchestrator's slot (SIENTIAPDE-1974 RF4).
+        'use_security': bool(server.get('cert_path')),
+    }
+
+
+class ServerConfigResolver:
+    """Resolve and cache the connection settings of an OPC server by `server_id`.
+
+    Order: settings carried by the command itself (an explicit override; the C2
+    command of the Laborious carries none), the live pooled connection (what the read
+    path opened, so writes never fight the reads over settings), then MongoDB. Lookups
+    are cached for `ttl_s`; a server that is not found, for `missing_ttl_s`, so a
+    pipeline pointing at a removed server does not query MongoDB on every command."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        mongo,
+        logger: Logger,
+        ttl_s: float = 60.0,
+        missing_ttl_s: float = 10.0,
+        metadata: dict | None = None,
+    ):
+        self.pool = pool
+        self.mongo = mongo
+        self.logger = logger
+        self.ttl_s = ttl_s
+        self.missing_ttl_s = missing_ttl_s
+        self.metadata = metadata or {}
+        self.clock = MonotonicClock()
+        self._cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
+
+    async def resolve(self, server_id: str, cmd: dict) -> dict[str, Any] | None:
+        """Connection settings for `server_id`, or None when no source knows it.
+
+        Raises:
+            Exception: When MongoDB cannot be read (the caller reports the write as
+                failed).
+        """
+        if cmd.get('url'):
+            return {
+                'url': cmd['url'],
+                'server_uri': cmd.get('server_uri'),
+                'use_security': cmd.get('use_security', False),
+            }
+        server_id = str(server_id)
+        pooled = self.pool.config_for(server_id)
+        if pooled is not None:
+            return pooled
+        now = self.clock.elapsed()
+        cached = self._cache.get(server_id)
+        if cached is not None:
+            cfg, cached_at = cached
+            if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s):
+                return dict(cfg) if cfg is not None else None
+        cfg = await self._lookup(server_id)
+        self._cache[server_id] = (cfg, now)
+        return dict(cfg) if cfg is not None else None
+
+    async def _lookup(self, server_id: str) -> dict[str, Any] | None:
+        ids = {'$in': _id_forms(server_id)}
+        connector = await self.mongo.find_one(
+            CONNECTORS,
+            {'connector_id': ids, 'type': 'opc_ua', 'status': 'Active'},
+            self.metadata,
+        )
+        cfg = _connector_cfg(connector) if connector else None
+        if cfg is not None:
+            return cfg
+        server = await self.mongo.find_one(OPC_SERVERS, {'id': ids}, self.metadata)
+        cfg = _opc_server_cfg(server) if server else None
+        if cfg is None:
+            self.logger.warning(
+                f'OPC server {server_id} not found in {CONNECTORS} nor {OPC_SERVERS}'
+            )
+        return cfg

ingestor/engines/write.py

@@ -18,6 +18,7 @@ from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 import ingestor.metrics as metrics
 from ingestor.connection.clock import MonotonicClock
 from ingestor.connection.pool import OpcConnectionPool
+from ingestor.connection.servers import ServerConfigResolver
 from ingestor.converters import ConversionError, to_typed_value
 from ingestor.io.data_sink import DataSink
 
@@ -77,9 +78,13 @@ class WriteEngine(SientiaMonitoring):
         write_timestamp: bool = True,
         write_quality: bool = True,
         min_pause_s: float = 0.0,
+        servers: ServerConfigResolver | None = None,
     ):
         self.pool = pool
         self.sink = sink
+        # Resolves the server connection by server_id (SIENTIAPDE-1975 GW1). Without it,
+        # only the connection carried by the command itself is used.
+        self.servers = servers
         self.write_timestamp = write_timestamp
         self.write_quality = write_quality
         self.min_pause_s = min_pause_s
@@ -105,11 +110,24 @@ class WriteEngine(SientiaMonitoring):
         except ConversionError as e:
             return await self._emit_error(cmd, 'conversion', str(e), 'error')
 
-        dv = self._build_datavalue(cmd, typed)
+        try:
+            dv = self._build_datavalue(cmd, typed)
+        except (KeyError, TypeError, ValueError) as e:
+            # write_timestamp on with a missing or non-ISO timestamp.
+            return await self._emit_error(cmd, 'bad_timestamp', repr(e), 'error')
+
+        try:
+            server_cfg = await self._server_cfg(cmd)
+        except Exception as e:
+            return await self._emit_error(cmd, 'server_lookup', str(e), 'error')
+        if server_cfg is None:
+            return await self._emit_error(
+                cmd, 'server_not_found', f'OPC server {cmd["server_id"]} is not configured', 'error'
+            )
 
         start = self.clock.elapsed()
         try:
-            client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd))
+            client = await self.pool.get(cmd['server_id'], server_cfg)
             node = client.get_node(cmd['addr'])
             await node.write_value(dv)
         except Exception as e:
@@ -127,10 +145,14 @@ class WriteEngine(SientiaMonitoring):
                 now = self.clock.elapsed()
         self._last_write_at = now
 
-    def _server_cfg(self, cmd: dict) -> dict:
-        # ponytail: minimal cfg built straight from the C2 message. Resolving
-        # the full connector config (url/security) from the connectors DB by
-        # server_id is F4 wiring, not this task — pool is mocked in tests.
+    async def _server_cfg(self, cmd: dict) -> dict | None:
+        """Connection settings of the target server: resolved by server_id (connection
+        carried by the command, live pooled connection, connectors, opc-servers), or only
+        the ones carried by the command when no resolver is wired."""
+        if self.servers is not None:
+            return await self.servers.resolve(str(cmd['server_id']), cmd)
+        if not cmd.get('url'):
+            return None
         return {
             'url': cmd.get('url'),
             'use_security': cmd.get('use_security', False),
@@ -156,6 +178,12 @@ class WriteEngine(SientiaMonitoring):
         return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES
 
     async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str):
+        # The reason only lives here: the row gets write_status alone (SIENTIAPDE-1975).
+        self.logger.error(
+            f'OPC write failed ({status}, {error_type}): server={cmd.get("server_id")} '
+            f'addr={cmd.get("addr")} model_id={cmd.get("model_id")} '
+            f'timestamp={cmd.get("timestamp")}: {message}'
+        )
         tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
         await self.emit_metric(
             metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type}
@@ -164,6 +192,17 @@ class WriteEngine(SientiaMonitoring):
             metric_object=metrics.OPC_WRITE_TOTAL,
             tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'},
         )
+        # Same series as a success (status='ok'), so it always shows the last outcome.
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
+            method='set',
+            value=0,
+            tags={
+                'pod_id': self.pod_id,
+                'model_id': cmd.get('model_id', 'unknown'),
+                'status': 'ok',
+            },
+        )
         result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message)
         await self.sink.update_write_status(cmd, result)
         return result

ingestor/gateway_runtime.py

@@ -22,6 +22,7 @@ from ingestor.adapters.grpc_server import GrpcServer
 from ingestor.adapters.nats_consumer import NatsWriteConsumer
 from ingestor.config import GatewayConfig
 from ingestor.connection.pool import OpcConnectionPool
+from ingestor.connection.servers import ServerConfigResolver
 from ingestor.engines.active_read import ActiveReadEngine
 from ingestor.engines.write import WriteEngine
 from ingestor.io.data_sink import DataSink
@@ -61,6 +62,12 @@ class GatewayRuntime(SientiaMonitoring):
         )
         self.sink = DataSink(data_manager, postgres)
 
+        # Write commands carry only the server_id (SIENTIAPDE-1975): the connection is
+        # resolved here, from the live pooled connection or the connectors/opc-servers
+        # documents, with the same rule the orchestrator uses for the read slots.
+        self.servers = ServerConfigResolver(
+            self.pool, data_manager.mongo_repository, logger, metadata=self.metadata
+        )
         self.write_engine = WriteEngine(
             self.pool,
             self.sink,
@@ -70,6 +77,7 @@ class GatewayRuntime(SientiaMonitoring):
             write_timestamp=config.write_timestamp,
             write_quality=config.write_quality,
             min_pause_s=config.write_min_pause_s,
+            servers=self.servers,
         )
         self.nats_consumer = NatsWriteConsumer(
             config.nats_url,
@@ -110,6 +118,11 @@ class GatewayRuntime(SientiaMonitoring):
 
     async def idle_sweep(self) -> None:
         await self.pool.release_idle(self.config.idle_conn_max_s)
+        if self._nats_started and self.nats_consumer.closed:
+            # Reconnection is unbounded once connected; a closed connection means it
+            # was closed for good (e.g. by the server), so subscribe again.
+            self.logger.warning('GatewayRuntime: NATS connection closed; starting again')
+            self._nats_started = False
         if not self._nats_started:
             await self._start_nats()
 

ingestor/io/data_sink.py

@@ -1,6 +1,7 @@
+import asyncio
 import re
 
-from sqlalchemy import text
+from sqlalchemy import Select, Update, case, column, select, table, update
 
 import ingestor.metrics as metrics
 
@@ -8,6 +9,34 @@ import ingestor.metrics as metrics
 # match before a trailing '\n', which is not what we want for a SQL identifier.
 _IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
 
+# Worst-wins precedence of predictions.write_status (SIENTIAPDE-1975), from best to worst;
+# NULL, 'pending' and unknown values rank 0. Same rule as the Laborious
+# (laborious/utils/write_status.py): keep both in sync.
+STATUS_RANK = {'ok': 1, 'error': 2, 'session_bad': 3}
+
+
+def write_status_statements(
+    schema: str, table_name: str, status: str, timestamp, model_id
+) -> tuple[Update, Select]:
+    """The UPDATE that records `status` unless the row has a worse one, and the SELECT that
+    tells whether the row exists. Built with SQLAlchemy Core: identifiers and values are
+    never formatted into the SQL text."""
+    predictions = table(
+        table_name,
+        column('write_status'),
+        column('timestamp'),
+        column('model_id'),
+        schema=schema,
+    )
+    row = (predictions.c.timestamp == timestamp) & (predictions.c.model_id == model_id)
+    current_rank = case(STATUS_RANK, value=predictions.c.write_status, else_=0)
+    return (
+        update(predictions)
+        .where(row, current_rank <= STATUS_RANK.get(status, 0))
+        .values(write_status=status),
+        select(1).select_from(predictions).where(row),
+    )
+
 
 class DataSink:
     """Gateway egress wrapper: fans a reading out to every configured name/collection."""
@@ -43,33 +72,71 @@ class DataSink:
                 },
             )
 
-    async def update_write_status(self, cmd: dict, result) -> None:  # NOSONAR
-        """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id.
+    async def update_write_status(self, cmd: dict, result) -> str:
+        """Record a write outcome on the prediction row (C3), keeping the worst one.
 
-        schema/table_name are per-model SQL identifiers, not values — they can't be
-        bound params, so they're checked against an allowlist before interpolation.
-        status/timestamp/model_id ARE bound params, never f-stringed.
+        The row is found by timestamp + model_id. A prediction is written to several
+        tags, each answered by its own command, so a better outcome never replaces a
+        worse one: ``ok`` < ``error`` < ``session_bad``, and ``pending``/NULL are
+        replaced by any answer. The Laborious runs the same statement for the outcomes it
+        records itself (SIENTIAPDE-1975), so the order of the answers does not matter.
+
+        Never raises: a failure here must not reach the NATS callback nor hide the OPC
+        write itself, which already happened. The outcome is logged and counted instead.
+
+        Return:
+            str: ``updated``, ``kept`` (the row already has a worse outcome),
+            ``row_not_found`` (e.g. the Laborious INSERT failed), ``invalid`` (bad or
+            missing schema/table/timestamp/model_id) or ``error``.
         """
-        schema = self._validate_identifier(cmd['schema'])
-        table = self._validate_identifier(cmd['table_name'])
-        # schema/table are allowlist-validated identifiers above; status/timestamp/model_id
-        # are bound params below, never interpolated — not a real injection vector.
-        stmt = text(
-            f'UPDATE {schema}.{table} SET write_status = :status '  # noqa: S608 # nosec B608
-            'WHERE timestamp = :timestamp AND model_id = :model_id'
+        try:
+            schema = self._validate_identifier(cmd.get('schema'))
+            table = self._validate_identifier(cmd.get('table_name'))
+            if cmd.get('timestamp') is None or cmd.get('model_id') is None:
+                raise ValueError('missing timestamp or model_id')
+        except ValueError as e:
+            return await self._status_outcome('invalid', cmd, f'write_status not recorded: {e}')
+
+        statements = write_status_statements(
+            schema, table, result.status, cmd['timestamp'], cmd['model_id']
         )
-        params = {
-            'status': result.status,
-            'timestamp': cmd['timestamp'],
-            'model_id': cmd['model_id'],
-        }
-        # ponytail: sync engine call inside async def — same pattern sientia_do's own
-        # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
+        try:
+            # The engine is synchronous (no async pg driver here): run it off the event
+            # loop, which also serves the subscriptions of the read path.
+            outcome = await asyncio.to_thread(self._update_sync, *statements)
+        except Exception as e:
+            return await self._status_outcome('error', cmd, f'write_status UPDATE failed: {e}')
+        detail = {
+            'updated': f'write_status={result.status} recorded',
+            'kept': f'write_status={result.status} not recorded: the row has a worse outcome',
+            'row_not_found': f'write_status={result.status} not recorded: no prediction row',
+        }[outcome]
+        return await self._status_outcome(outcome, cmd, detail)
+
+    def _update_sync(self, update_stmt: Update, exists_stmt: Select) -> str:
         with self.postgres.engine.begin() as connection:
-            connection.execute(stmt, params)
+            if connection.execute(update_stmt).rowcount:
+                return 'updated'
+            row = connection.execute(exists_stmt).first()
+        return 'kept' if row is not None else 'row_not_found'
+
+    async def _status_outcome(self, outcome: str, cmd: dict, detail: str) -> str:
+        message = (
+            f'{detail} (server={cmd.get("server_id")} addr={cmd.get("addr")} '
+            f'model_id={cmd.get("model_id")} timestamp={cmd.get("timestamp")})'
+        )
+        if outcome in ('updated', 'kept'):
+            self.data_manager.logger.debug(message)
+        else:
+            self.data_manager.logger.warning(message)
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_UPDATE_TOTAL,
+            tags={'pod_id': self.data_manager.pod_id, 'result': outcome},
+        )
+        return outcome
 
     @staticmethod
-    def _validate_identifier(name: str) -> str:
+    def _validate_identifier(name: object) -> str:
         if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name):
             raise ValueError(f'invalid SQL identifier: {name!r}')
         return name

ingestor/metrics.py

@@ -156,6 +156,12 @@ OPC_WRITE_STATUS_BY_MODEL = Gauge(
     'Last write status per model (1=ok,0=error)',
     ['pod_id', 'model_id', 'status'],
 )
+OPC_WRITE_STATUS_UPDATE_TOTAL = Counter(
+    'opc_write_status_update_total',
+    'write_status UPDATEs on the prediction rows',
+    # result: updated|kept|row_not_found|invalid|error
+    ['pod_id', 'result'],
+)
 # --- Gateway: Active read metrics ---
 OPC_ACTIVE_READ_TOTAL = Counter(
     'opc_active_read_total',