diff --git a/CLAUDE.md b/CLAUDE.md index 358a73c..c1a07f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -60,7 +60,7 @@ 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`; same rule as the Laborious' `laborious/utils/write_status.py` — keep both in sync). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded. +**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. diff --git a/README.md b/README.md index 4694bc6..ee7ea36 100644 --- a/README.md +++ b/README.md @@ -91,8 +91,9 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975): 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. The Laborious applies the same rule to - the outcomes it records itself, so the order of the answers does not matter. + `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`. diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 44fb7c6..aa88ff1 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -10,8 +10,8 @@ import ingestor.metrics as metrics _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. +# 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} @@ -78,8 +78,9 @@ class DataSink: 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. + 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.