diff --git a/CLAUDE.md b/CLAUDE.md index 44c07c9..c1a07f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -60,6 +60,8 @@ Any change to slot/lease counts should be traced through `manage_leases`/`manage **ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop. +**Writes** (SIENTIAPDE-1975): `NatsWriteConsumer` → `WriteEngine` → pool, then `DataSink.update_write_status`. Commands (C2) carry only `server_id`; `ServerConfigResolver` (`connection/servers.py`) resolves the connection (command override → live pooled connection → `connectors` → `opc-servers`). `write_status` keeps the worst outcome (`ok` < `error` < `session_bad`); the gateway is the only one that updates it (the Laborious inserts the row as `pending` and never records OPC outcomes). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded. + **Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications. **Error handling convention**: broad `except Exception` + `traceback.print_exc()` + notification-handler alert is intentional throughout (see `ruff.lint.ignore` for `BLE001` in `pyproject.toml`) — this is an always-on service where a single tag/server failure must not kill the pod; failures are surfaced via notifications/metrics instead of raised. diff --git a/README.md b/README.md index f062786..ee7ea36 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,8 @@ Gateway for on-demand and command-driven interactions with OPC UA servers: - **NATS write consumer**: consumes write commands from `opc.write` (see `NATS_WRITE_SUBJECT`) and executes them against the target OPC server - (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`). + (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`). See + [Write commands](#write-commands-c2-and-write_status-c3) below. - **On-demand gRPC `ReadTag`**: a gRPC server (`ingestor/adapters/grpc_server.py`, `ingestor/proto/opc_gateway.proto`) for synchronous, single-tag reads. - **Scheduled active reads**: a polling read engine (`ingestor/engines/active_read.py`) @@ -70,6 +71,37 @@ See `docs/deploy-notes.md` for the deployment-facing details (env vars, health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in `.env.example` for the full list of new environment variables. +### Write commands (C2) and `write_status` (C3) + +The Laborious publishes one command per tag (SIENTIAPDE-1975): + +```json +{"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float", + "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data", + "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00", + "write_timestamp": false, "write_quality": false} +``` + +- **Server connection**: the command carries only `server_id`. The connection is + resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): `url`/ + `server_uri`/`use_security` in the command when present (override), else the + live pooled connection opened by the reads, else the `connectors` document + (`type: opc_ua`, `status: Active`), else the `opc-servers` document + (`use_security = bool(cert_path)`) — the orchestrator's rule for the read slots. + Lookups are cached for 60 s. An unknown server is a failed write (`server_not_found`). +- **Outcome**: `UPDATE ... SET write_status` on the prediction row (by `timestamp` + + `model_id`), keeping the worst outcome: `ok` < `error` < `session_bad`, and + `pending`/NULL are replaced by any answer, so the order of the answers does not + matter. The gateway is the only one that updates the column: the Laborious inserts the + row as `pending` (merged with its PI Web API outcome) and never records OPC outcomes. + The UPDATE never raises; its outcome (`updated`, `kept`, `row_not_found`, + `invalid`, `error`) is logged and counted in `opc_write_status_update_total`. + The reason of a failed write is logged by `WriteEngine`. +- **NATS connection**: the first connection gives up after 5 s (it runs on the main + loop: nats-py's defaults would block it for minutes while NATS is down); once + connected, it reconnects forever. A connection closed for good is started again + by `GatewayRuntime.idle_sweep`. + ## 📋 Prerequisites - Python 3.11+ @@ -290,6 +322,9 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: ### OPC Gateway Metrics - `opc_write_total`: Active write attempts (NATS-driven), by result - `opc_write_latency_seconds`: Active write latency +- `opc_write_status_by_model`: Last write outcome per model (1 = ok, 0 = failed; label `status`) +- `opc_write_status_update_total`: `write_status` UPDATEs, by result + (`updated`, `kept`, `row_not_found`, `invalid`, `error`) - `opc_active_read_total`: Scheduled/on-demand read attempts, by result - `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...) diff --git a/ingestor/adapters/nats_consumer.py b/ingestor/adapters/nats_consumer.py index 97ed2fe..b985142 100644 --- a/ingestor/adapters/nats_consumer.py +++ b/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,41 @@ 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()}' + ) + + # These stay async with nothing to await: nats-py rejects callbacks that are not + # coroutine functions (InvalidCallbackTypeError on connect). + 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() diff --git a/ingestor/connection/pool.py b/ingestor/connection/pool.py index d95abc3..8c1b164 100644 --- a/ingestor/connection/pool.py +++ b/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: diff --git a/ingestor/connection/servers.py b/ingestor/connection/servers.py new file mode 100644 index 0000000..fa4b431 --- /dev/null +++ b/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 diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 675cb74..959a7d8 100644 --- a/ingestor/engines/write.py +++ b/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 diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py index 59a6a88..987ed46 100644 --- a/ingestor/gateway_runtime.py +++ b/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() diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 8555e2c..aa88ff1 100644 --- a/ingestor/io/data_sink.py +++ b/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. The gateway is the only writer that updates the +# column: the Laborious only sets the value of the rows it inserts ('pending', PI outcome). +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,72 @@ 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, so the order of the answers does not matter. Only the + gateway updates the column (SIENTIAPDE-1975): the Laborious publishes the commands + and sets ``pending`` (merged with its PI Web API outcome) on the row it inserts. + + 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 diff --git a/ingestor/metrics.py b/ingestor/metrics.py index 9b7e50b..b7fcd76 100644 --- a/ingestor/metrics.py +++ b/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', diff --git a/tests/unit/adapters/test_nats_consumer.py b/tests/unit/adapters/test_nats_consumer.py index 7410dab..1485d1a 100644 --- a/tests/unit/adapters/test_nats_consumer.py +++ b/tests/unit/adapters/test_nats_consumer.py @@ -1,3 +1,5 @@ +import asyncio +import inspect import json from unittest.mock import AsyncMock, MagicMock, patch @@ -37,20 +39,99 @@ async def test_handle_invalid_json_drops_message_and_logs_error(): consumer.logger.error.assert_called_once() +CLIENT = 'ingestor.adapters.nats_consumer.Client' + + +def _fake_nc(connect=None): + nc = MagicMock() + nc.connect = connect or AsyncMock() + nc.subscribe = AsyncMock() + nc.close = AsyncMock() + nc.drain = AsyncMock() + nc.is_closed = False + return nc + + +async def _hang(*_args, **_kwargs): + await asyncio.sleep(10) + + @pytest.mark.asyncio async def test_start_subscribes_with_subject_and_queue(): consumer = make_consumer(subject='opc.write', queue='gateway-writers') - mock_nc = AsyncMock() - mock_nc.subscribe = AsyncMock() + nc = _fake_nc() - with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: - mock_nats.connect = AsyncMock(return_value=mock_nc) + with patch(CLIENT, return_value=nc): await consumer.start() - mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222') - mock_nc.subscribe.assert_awaited_once_with( - 'opc.write', queue='gateway-writers', cb=consumer._handle - ) + kwargs = nc.connect.await_args.kwargs + assert kwargs['servers'] == 'nats://localhost:4222' + # Reconnection is unbounded once connected: an outage never closes it for good. + assert kwargs['max_reconnect_attempts'] == -1 + nc.subscribe.assert_awaited_once_with('opc.write', queue='gateway-writers', cb=consumer._handle) + assert consumer.closed is False + + +@pytest.mark.asyncio +async def test_start_gives_up_after_the_connect_timeout(): + """start() runs on the main loop: an unreachable NATS must not block it.""" + consumer = make_consumer(connect_timeout_s=0.1) + nc = _fake_nc(connect=AsyncMock(side_effect=_hang)) + # Outer guard: fails the test instead of hanging it if start() ignores its timeout. + start = asyncio.wait_for(consumer.start(), timeout=5) + + with patch(CLIENT, return_value=nc): + with pytest.raises(TimeoutError): + await start + + nc.close.assert_awaited_once() + assert consumer.closed is True + + +@pytest.mark.asyncio +async def test_start_closes_the_connection_when_subscribe_fails(): + consumer = make_consumer() + nc = _fake_nc() + nc.subscribe.side_effect = RuntimeError('permissions violation') + + with patch(CLIENT, return_value=nc), pytest.raises(RuntimeError): + await consumer.start() + + nc.close.assert_awaited_once() + assert consumer.closed is True + + +@pytest.mark.asyncio +async def test_handle_unexpected_engine_error_is_logged_not_raised(): + consumer = make_consumer() + consumer.write_engine.write.side_effect = RuntimeError('bug') + msg = MagicMock(data=json.dumps({'server_id': '1'}).encode()) + + await consumer._handle(msg) + + assert 'bug' in consumer.logger.error.call_args.args[0] + + +@pytest.mark.asyncio +async def test_connection_events_are_logged_in_one_line(): + consumer = make_consumer() + nc = _fake_nc() + with patch(CLIENT, return_value=nc): + await consumer.start() + callbacks = nc.connect.await_args.kwargs + # nats-py raises InvalidCallbackTypeError on connect for a non-coroutine callback. + for name in ('error_cb', 'disconnected_cb', 'reconnected_cb'): + assert inspect.iscoroutinefunction(callbacks[name]), name + + await callbacks['error_cb'](OSError('connection refused')) + await callbacks['disconnected_cb']() + await callbacks['reconnected_cb']() + + warnings = [c.args[0] for c in consumer.logger.warning.call_args_list] + assert warnings == [ + 'NATS error: OSError connection refused', + 'Disconnected from NATS; reconnecting', + ] @pytest.mark.asyncio @@ -63,14 +144,28 @@ async def test_stop_before_start_does_not_raise(): @pytest.mark.asyncio async def test_stop_after_start_drains_and_unsubscribes(): consumer = make_consumer() - mock_nc = AsyncMock() - mock_sub = AsyncMock() - mock_nc.subscribe = AsyncMock(return_value=mock_sub) + nc = _fake_nc() + sub = AsyncMock() + nc.subscribe = AsyncMock(return_value=sub) - with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: - mock_nats.connect = AsyncMock(return_value=mock_nc) + with patch(CLIENT, return_value=nc): await consumer.start() await consumer.stop() - mock_sub.unsubscribe.assert_awaited_once() - mock_nc.drain.assert_awaited_once() + sub.unsubscribe.assert_awaited_once() + nc.drain.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stop_on_a_connection_closed_for_good_does_not_raise(): + consumer = make_consumer() + nc = _fake_nc() + with patch(CLIENT, return_value=nc): + await consumer.start() + nc.is_closed = True + consumer.sub.unsubscribe = AsyncMock(side_effect=RuntimeError('connection closed')) + + await consumer.stop() # no raise + + nc.drain.assert_not_awaited() + assert consumer.closed is True diff --git a/tests/unit/connection/test_pool.py b/tests/unit/connection/test_pool.py index d23dc03..296ed49 100644 --- a/tests/unit/connection/test_pool.py +++ b/tests/unit/connection/test_pool.py @@ -342,3 +342,23 @@ async def test_connect_server_uri_still_wins_for_application_uri(): # must match the certificate SAN URI (BadCertificateUriInvalid otherwise) assert fake_client.application_uri == 'urn:sientia:opc-ingestor' assert fake_client.name == pool.pod_id + + +@pytest.mark.asyncio +async def test_config_for_returns_a_copy_of_the_live_connection_settings(): + pool = make_pool() + cfg = {'url': 'opc.tcp://h:4840', 'server_uri': 'urn:h', 'use_security': False} + with patch('ingestor.connection.pool.Client', return_value=AsyncMock()): + await pool.get(1, cfg) + + found = pool.config_for('1') + assert found == cfg + found['url'] = 'mutated' + assert pool.config_for(1)['url'] == 'opc.tcp://h:4840' + + await pool.evict('1') + assert pool.config_for('1') is None + + +def test_config_for_unknown_server_is_none(): + assert make_pool().config_for('missing') is None diff --git a/tests/unit/connection/test_servers.py b/tests/unit/connection/test_servers.py new file mode 100644 index 0000000..3f98131 --- /dev/null +++ b/tests/unit/connection/test_servers.py @@ -0,0 +1,151 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from ingestor.connection.servers import CONNECTORS, OPC_SERVERS, ServerConfigResolver + +CONNECTOR = { + 'connector_id': 1, + 'type': 'opc_ua', + 'status': 'Active', + 'name': 'kepserver', + 'url': 'opc.tcp://kepserver:49320', + 'server_uri': 'urn:kepserver', + 'use_security': True, +} +OPC_SERVER = { + 'id': '1', + 'server_name': 'kepserver', + 'url': 'opc.tcp://legacy:49320', + 'uri': 'urn:legacy', + 'cert_path': '/app/certs/cert.pem', +} + + +def make_resolver(connector=None, opc_server=None, pooled=None, ttl_s=60.0): + pool = MagicMock() + pool.config_for = MagicMock(return_value=pooled) + mongo = MagicMock() + + async def find_one(collection, _filters, _metadata): + return {CONNECTORS: connector, OPC_SERVERS: opc_server}[collection] + + mongo.find_one = AsyncMock(side_effect=find_one) + return ServerConfigResolver(pool, mongo, MagicMock(), ttl_s=ttl_s) + + +@pytest.mark.asyncio +async def test_command_connection_overrides_every_other_source(): + resolver = make_resolver(connector=CONNECTOR) + + cfg = await resolver.resolve('1', {'url': 'opc.tcp://h:1', 'server_uri': 'urn:h'}) + + assert cfg == {'url': 'opc.tcp://h:1', 'server_uri': 'urn:h', 'use_security': False} + resolver.mongo.find_one.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_live_pooled_connection_wins_over_mongo(): + """Writes reuse exactly the settings the read path opened the connection with.""" + pooled = {'url': 'opc.tcp://read:1', 'server_uri': 'urn:read', 'use_security': True} + resolver = make_resolver(connector=CONNECTOR, pooled=pooled) + + assert await resolver.resolve('1', {}) == pooled + resolver.pool.config_for.assert_called_once_with('1') + resolver.mongo.find_one.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_tag_database_connector_first(): + resolver = make_resolver(connector=CONNECTOR, opc_server=OPC_SERVER) + + cfg = await resolver.resolve('1', {}) + + assert cfg == { + 'url': 'opc.tcp://kepserver:49320', + 'server_uri': 'urn:kepserver', + 'use_security': True, + } + collection, filters, _ = resolver.mongo.find_one.await_args_list[0].args + assert collection == CONNECTORS + # connector_id is an int in the tag database; the command carries a string. + assert filters == {'connector_id': {'$in': ['1', 1]}, 'type': 'opc_ua', 'status': 'Active'} + + +@pytest.mark.asyncio +async def test_opc_servers_fallback_without_tag_database(): + resolver = make_resolver(connector=None, opc_server=OPC_SERVER) + + cfg = await resolver.resolve('1', {}) + + # Same rule as the orchestrator's read slot: a cert_path means security. + assert cfg == { + 'url': 'opc.tcp://legacy:49320', + 'server_uri': 'urn:legacy', + 'use_security': True, + } + collection, filters, _ = resolver.mongo.find_one.await_args_list[1].args + assert collection == OPC_SERVERS + assert filters == {'id': {'$in': ['1', 1]}} + + +@pytest.mark.asyncio +async def test_opc_server_without_cert_path_is_insecure(): + resolver = make_resolver(opc_server={**OPC_SERVER, 'cert_path': ''}) + + assert (await resolver.resolve('1', {}))['use_security'] is False + + +@pytest.mark.asyncio +async def test_non_numeric_server_id_is_looked_up_as_is(): + resolver = make_resolver(opc_server={**OPC_SERVER, 'id': 'kep-a'}) + + await resolver.resolve('kep-a', {}) + + _, filters, _ = resolver.mongo.find_one.await_args_list[0].args + assert filters['connector_id'] == {'$in': ['kep-a']} + + +@pytest.mark.asyncio +async def test_unknown_server_is_cached_briefly(): + resolver = make_resolver(ttl_s=60.0) + resolver.missing_ttl_s = 10.0 + resolver.clock = MagicMock() + resolver.clock.elapsed.side_effect = [0.0, 5.0, 11.0] + + assert await resolver.resolve('9', {}) is None + assert await resolver.resolve('9', {}) is None # within missing_ttl_s: no lookup + assert await resolver.resolve('9', {}) is None # expired: looked up again + assert resolver.mongo.find_one.await_count == 4 # 2 lookups (connectors + opc-servers) + assert resolver.logger.warning.call_count == 2 + + +@pytest.mark.asyncio +async def test_documents_without_url_are_ignored(): + resolver = make_resolver(connector={**CONNECTOR, 'url': ''}, opc_server=OPC_SERVER) + + assert (await resolver.resolve('1', {}))['url'] == 'opc.tcp://legacy:49320' + + +@pytest.mark.asyncio +async def test_lookups_are_cached_for_the_ttl(): + resolver = make_resolver(connector=CONNECTOR, ttl_s=60.0) + resolver.clock = MagicMock() + resolver.clock.elapsed.side_effect = [0.0, 30.0, 61.0] + + first = await resolver.resolve('1', {}) + first['url'] = 'mutated by the caller' + second = await resolver.resolve('1', {}) # within the TTL: cached, and a copy + await resolver.resolve('1', {}) # expired: looked up again + + assert second['url'] == 'opc.tcp://kepserver:49320' + assert resolver.mongo.find_one.await_count == 2 + + +@pytest.mark.asyncio +async def test_mongo_errors_propagate_to_the_caller(): + resolver = make_resolver() + resolver.mongo.find_one = AsyncMock(side_effect=RuntimeError('mongo down')) + + with pytest.raises(RuntimeError, match='mongo down'): + await resolver.resolve('1', {}) diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index ad54c30..a913151 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -7,8 +7,17 @@ from asyncua import ua import ingestor.metrics as metrics from ingestor.engines.write import WriteEngine +SERVER_CFG = {'url': 'opc.tcp://kepserver:49320', 'server_uri': 'urn:kep', 'use_security': True} + + +def make_resolver(cfg=SERVER_CFG): + resolver = MagicMock() + resolver.resolve = AsyncMock(return_value=cfg) + return resolver + def make_engine(**kwargs): + kwargs.setdefault('servers', make_resolver()) eng = WriteEngine( pool=AsyncMock(), sink=AsyncMock(), @@ -257,3 +266,104 @@ async def test_no_rate_limit_when_min_pause_is_zero(): await eng.write(base_cmd()) await eng.write(base_cmd()) sleep_mock.assert_not_awaited() + + +# --- SIENTIAPDE-1975: server resolution (GW1) and error reporting (GW4) --- + + +@pytest.mark.asyncio +async def test_write_uses_the_resolved_server_connection(): + """The C2 command of the Laborious carries only the server_id.""" + eng = make_engine() + eng.pool.get.return_value = make_client() + + await eng.write(base_cmd()) + + eng.servers.resolve.assert_awaited_once_with('srv1', base_cmd()) + eng.pool.get.assert_awaited_once_with('srv1', SERVER_CFG) + + +@pytest.mark.asyncio +async def test_write_unknown_server_records_error_without_connecting(): + eng = make_engine(servers=make_resolver(cfg=None)) + + result = await eng.write(base_cmd()) + + assert result.status == 'error' + assert result.opc_status_code == 'server_not_found' + eng.pool.get.assert_not_awaited() + eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result) + + +@pytest.mark.asyncio +async def test_write_server_lookup_failure_records_error(): + resolver = make_resolver() + resolver.resolve.side_effect = RuntimeError('mongo down') + eng = make_engine(servers=resolver) + + result = await eng.write(base_cmd()) + + assert result.status == 'error' + assert result.opc_status_code == 'server_lookup' + assert 'mongo down' in result.message + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_without_resolver_uses_the_command_connection(): + eng = make_engine(servers=None) + eng.pool.get.return_value = make_client() + + await eng.write(base_cmd(url='opc.tcp://h:4840', server_uri='urn:h', use_security=True)) + + eng.pool.get.assert_awaited_once_with( + 'srv1', {'url': 'opc.tcp://h:4840', 'use_security': True, 'server_uri': 'urn:h'} + ) + + +@pytest.mark.asyncio +async def test_write_without_resolver_or_url_is_server_not_found(): + eng = make_engine(servers=None) + + result = await eng.write(base_cmd()) + + assert result.opc_status_code == 'server_not_found' + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_error_logs_the_reason_and_sets_the_model_status(): + eng = make_engine() + node = make_node() + node.write_value.side_effect = RuntimeError('BadWriteNotSupported') + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + logged = eng.logger.error.call_args.args[0] + assert 'RuntimeError' in logged and 'BadWriteNotSupported' in logged + assert 'addr=ns=2;i=5' in logged and 'model_id=m1' in logged + status_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_BY_MODEL + ) + # Same series as a success, so it shows the last outcome. + assert status_call.kwargs['value'] == 0 + assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'} + + +@pytest.mark.asyncio +@pytest.mark.parametrize('timestamp', [None, 'not-a-timestamp']) +async def test_write_bad_timestamp_records_error_without_raising(timestamp): + eng = make_engine(write_timestamp=True) + cmd = base_cmd(timestamp=timestamp) + if timestamp is None: + del cmd['timestamp'] + + result = await eng.write(cmd) + + assert result.status == 'error' + assert result.opc_status_code == 'bad_timestamp' + eng.pool.get.assert_not_awaited() + eng.sink.update_write_status.assert_awaited_once() diff --git a/tests/unit/io/test_data_sink_status.py b/tests/unit/io/test_data_sink_status.py index a626036..590796a 100644 --- a/tests/unit/io/test_data_sink_status.py +++ b/tests/unit/io/test_data_sink_status.py @@ -1,89 +1,149 @@ -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.dialects import postgresql +import ingestor.metrics as metrics from ingestor.engines.write import WriteResult -from ingestor.io.data_sink import DataSink +from ingestor.io.data_sink import DataSink, write_status_statements + +CMD = { + 'server_id': '1', + 'addr': 'ns=2;s=Pred', + 'schema': 'sientia_data', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', +} -def make_sink(): +def _result(status): + return WriteResult(ok=status == 'ok', status=status, opc_status_code='x', message='') + + +def make_sink(rowcount=1, existing_row=True): postgres = MagicMock() connection = MagicMock() + connection.execute.return_value.rowcount = rowcount + connection.execute.return_value.first.return_value = (1,) if existing_row else None postgres.engine.begin.return_value.__enter__.return_value = connection - sink = DataSink(data_manager=MagicMock(), postgres=postgres) + data_manager = MagicMock() + data_manager.emit_metric = AsyncMock() + sink = DataSink(data_manager=data_manager, postgres=postgres) return sink, connection +def _outcome_metric(sink): + calls = [ + c + for c in sink.data_manager.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_UPDATE_TOTAL + ] + assert len(calls) == 1 + return calls[0].kwargs['tags']['result'] + + @pytest.mark.asyncio -async def test_update_write_status_targets_model_row(): +async def test_update_write_status_targets_model_row_with_precedence(): sink, connection = make_sink() - cmd = { - 'schema': 'sientia_data', - 'table_name': 'pred_m1', - 'model_id': '1', - 'timestamp': '2026-07-13T00:00:00Z', - } - result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') - await sink.update_write_status(cmd, result) + outcome = await sink.update_write_status(CMD, _result('ok')) + assert outcome == 'updated' assert connection.execute.call_count == 1 - stmt, params = connection.execute.call_args[0] - sql = str(stmt) - assert 'update' in sql.lower() and 'write_status' in sql.lower() - assert 'sientia_data.pred_m1' in sql + (stmt,) = connection.execute.call_args[0] + compiled = stmt.compile(dialect=postgresql.dialect()) + sql = str(compiled) + assert sql.startswith('UPDATE sientia_data.pred_m1 SET write_status=') + assert 'CASE sientia_data.pred_m1.write_status WHEN' in sql # values must be bound params, never interpolated into the SQL text itself - assert params == {'status': 'ok', 'timestamp': '2026-07-13T00:00:00Z', 'model_id': '1'} - assert 'ok' not in sql - assert '2026-07-13T00:00:00Z' not in sql + assert '2026-07-13T00:00:00Z' not in sql and "'ok'" not in sql + assert compiled.params['write_status'] == 'ok' + assert '2026-07-13T00:00:00Z' in compiled.params.values() + assert _outcome_metric(sink) == 'updated' @pytest.mark.asyncio -async def test_update_write_status_rejects_bad_schema(): - sink, connection = make_sink() - cmd = { - 'schema': 'a; DROP TABLE x', - 'table_name': 'pred_m1', - 'model_id': '1', - 'timestamp': '2026-07-13T00:00:00Z', - } - result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') +async def test_update_write_status_kept_when_the_row_has_a_worse_outcome(): + sink, _ = make_sink(rowcount=0, existing_row=True) - with pytest.raises(ValueError): - await sink.update_write_status(cmd, result) - - connection.execute.assert_not_called() + assert await sink.update_write_status(CMD, _result('ok')) == 'kept' + assert _outcome_metric(sink) == 'kept' @pytest.mark.asyncio -async def test_update_write_status_rejects_trailing_newline_identifier(): - sink, connection = make_sink() - cmd = { - 'schema': 'sientia_data\n', - 'table_name': 'pred_m1', - 'model_id': '1', - 'timestamp': '2026-07-13T00:00:00Z', - } - result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') +async def test_update_write_status_row_not_found_is_logged(): + """E.g. the Laborious INSERT failed and it wrote to OPC anyway (SIENTIAPDE-1975 D10).""" + sink, _ = make_sink(rowcount=0, existing_row=False) - with pytest.raises(ValueError): - await sink.update_write_status(cmd, result) - - connection.execute.assert_not_called() + assert await sink.update_write_status(CMD, _result('error')) == 'row_not_found' + assert 'no prediction row' in sink.data_manager.logger.warning.call_args.args[0] + assert _outcome_metric(sink) == 'row_not_found' @pytest.mark.asyncio -async def test_update_write_status_rejects_bad_table_name(): +async def test_update_write_status_database_error_is_not_raised(): sink, connection = make_sink() - cmd = { - 'schema': 'sientia_data', - 'table_name': 'pred_m1; DROP TABLE x', - 'model_id': '1', - 'timestamp': '2026-07-13T00:00:00Z', - } - result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') + connection.execute.side_effect = RuntimeError('column "write_status" does not exist') - with pytest.raises(ValueError): - await sink.update_write_status(cmd, result) + assert await sink.update_write_status(CMD, _result('ok')) == 'error' + assert 'write_status' in sink.data_manager.logger.warning.call_args.args[0] + assert _outcome_metric(sink) == 'error' + +@pytest.mark.parametrize( + ('field', 'value'), + [ + ('schema', 'a; DROP TABLE x'), + ('schema', 'sientia_data\n'), + ('table_name', 'pred_m1; DROP TABLE x'), + ('schema', None), + ('timestamp', None), + ('model_id', None), + ], +) +@pytest.mark.asyncio +async def test_update_write_status_rejects_invalid_commands(field, value): + sink, connection = make_sink() + + outcome = await sink.update_write_status({**CMD, field: value}, _result('error')) + + assert outcome == 'invalid' connection.execute.assert_not_called() + assert _outcome_metric(sink) == 'invalid' + + +def _apply(initial, answers): + """Run the real statements on SQLite for each answer; return the final status.""" + engine = create_engine('sqlite://') + with engine.begin() as connection: + connection.execute( + text('CREATE TABLE pred (model_id INTEGER, timestamp TEXT, write_status TEXT)') + ) + connection.execute( + text('INSERT INTO pred VALUES (1, :ts, :status)'), + {'ts': '2026-07-13T00:00:00Z', 'status': initial}, + ) + for answer in answers: + update_stmt, exists_stmt = write_status_statements( + None, 'pred', answer, '2026-07-13T00:00:00Z', 1 + ) + connection.execute(update_stmt) + assert connection.execute(exists_stmt).first() is not None + return connection.execute(text('SELECT write_status FROM pred')).scalar() + + +@pytest.mark.parametrize( + ('initial', 'answers', 'expected'), + [ + ('pending', ['ok'], 'ok'), + ('pending', ['error', 'ok'], 'error'), + ('pending', ['ok', 'error'], 'error'), + ('pending', ['session_bad', 'error', 'ok'], 'session_bad'), + ('error', ['ok'], 'error'), + (None, ['ok'], 'ok'), + ], +) +def test_update_statement_keeps_the_worst_answer(initial, answers, expected): + assert _apply(initial, answers) == expected diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py index 8da8ddb..06b0fc2 100644 --- a/tests/unit/test_gateway_runtime.py +++ b/tests/unit/test_gateway_runtime.py @@ -39,6 +39,7 @@ def mocks(monkeypatch): patched['pool'].release_idle = AsyncMock() patched['NatsWriteConsumer'].return_value.start = AsyncMock() patched['NatsWriteConsumer'].return_value.stop = AsyncMock() + patched['NatsWriteConsumer'].return_value.closed = False patched['GrpcServer'].return_value.start = AsyncMock() patched['GrpcServer'].return_value.stop = AsyncMock() return patched @@ -173,3 +174,47 @@ async def test_idle_sweep_releases_idle_connections(mocks): await runtime.idle_sweep() mocks['pool'].release_idle.assert_awaited_once_with(42.0) + + +def test_write_engine_resolves_servers_from_the_pool_and_mongo(mocks): + """SIENTIAPDE-1975 GW1: write commands carry only the server_id.""" + data_manager = MagicMock() + runtime = gr.GatewayRuntime( + config=GatewayConfig(), + data_manager=data_manager, + pool=mocks['pool'], + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(), + ) + + _, kwargs = mocks['WriteEngine'].call_args + assert kwargs['servers'] is runtime.servers + assert runtime.servers.pool is mocks['pool'] + assert runtime.servers.mongo is data_manager.mongo_repository + + +@pytest.mark.asyncio +async def test_idle_sweep_restarts_a_closed_nats_connection(mocks): + runtime = make_runtime(pool=mocks['pool']) + nats_mock = mocks['NatsWriteConsumer'].return_value + await runtime.start() + assert nats_mock.start.await_count == 1 + + nats_mock.closed = True # closed for good (e.g. by the server) + await runtime.idle_sweep() + + assert nats_mock.start.await_count == 2 + assert runtime._nats_started is True + runtime.logger.warning.assert_called_once() + + +@pytest.mark.asyncio +async def test_idle_sweep_leaves_a_live_nats_connection_alone(mocks): + runtime = make_runtime(pool=mocks['pool']) + nats_mock = mocks['NatsWriteConsumer'].return_value + await runtime.start() + + await runtime.idle_sweep() + + assert nats_mock.start.await_count == 1