- One call per pytest.raises block (S5778): build the guarded start()
before the block and nest the patch outside it.
- The NATS callbacks stay async on purpose: nats-py rejects callbacks
that are not coroutine functions. Comment it and assert it in the
callback test, so nobody drops the async keyword to please S7503.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- 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>
Second code review round (xhigh) of 057c1fd + 085e609: 5 findings.
Fixed:
- manage_server's subscribe-failure path called reader.unsubscribe() directly;
with the legacy OpcManager a dead session made it raise out of the except
block and stop the main loop. Both paths now go through _unsubscribe_reader,
the single guarded per-reader unsubscribe.
- An unsubscribe failure also increments OPC_SUBSCRIPTION_ERRORS (same tags as
the subscription failures), besides the OPC_UNSUBSCRIBE_ERROR notification.
- The contract test creates the subscriptions through the readers' real
create_subscription (fake pool / client) instead of writing engine
internals, and checks the pool pin/unpin pair.
Documented (no code change):
- If delete() fails while the session is still alive (e.g. a timeout), the
server-side subscription may keep delivering data until that session ends
(reconnection or pod restart); noted in _unsubscribe_reader.
- manage_leases now unsubscribes before releasing the leases: if Redis fails
while releasing them, the slots stay unread until LEASE_TTL expires, which is
preferred to two pods reading the same slot.
284 unit tests; the 2 new behavior tests fail on 085e609.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Code review (code-review skill, xhigh) of 057c1fd: 6 findings, all fixed.
- unsubscribe_slot handles a failing reader: the legacy OpcManager's delete()
raises when the server session is gone, which still stopped the main loop.
The failure is notified (OPC_UNSUBSCRIBE_ERROR_<slot>:<server>) and logged,
the slot is dropped from that reader's books (as the v2 engine already does)
so it can be subscribed again, and the other readers are still unsubscribed.
Repo convention: a single server failure must not kill the pod.
- update_slot_config unsubscribes a slot deleted from Redis right away
(_unsubscribe_removed_slots) instead of leaving it to the snapshot diff in
update_ingestor_manager, which leaked the subscription if anything in between
failed. The later diff call is now a no-op.
- manage_leases unsubscribes the extra slots before giving their leases back,
so another pod never takes a slot this pod is still reading.
- unsubscribe_slot iterates a snapshot of the readers (list(...) # NOSONAR, as
in pool.close_all / ServerReader.shutdown): the loop awaits, so a reader
added or removed meanwhile must not break the iteration.
- Contract test with the real readers (ServerReader over the real
SubscriptionEngine, and the legacy OpcManager): both expose `subscriptions`
keyed by slot and both are unsubscribed (subscription deleted, pool unpinned).
- The reader test double moves to tests/unit/conftest.py (reader_double).
283 unit tests; the 3 new behavior tests fail on 057c1fd.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The main loop snapshots managed_tags, then two steps remove a slot from
managed_tags before update_ingestor_manager compares the snapshot with the
current state and calls unsubscribe_slot for every slot that disappeared:
- manage_leases drops an extra lease ("Extra slots available"): it already
unsubscribed the slot, so the second call is redundant;
- update_slot_config finds the slot gone from Redis (deleted by the
orchestrator): nothing unsubscribed it yet.
unsubscribe_slot looked the slot up in managed_tags[slot] to find its servers,
so both cases raised KeyError, which stopped the main loop and restarted the
container (seen on dev on 2026-10-06 18:30 during a rollout, ~40 s without
reads). In the second case the subscription also stayed open on the OPC server.
The same code is on main, so production has it too.
unsubscribe_slot now asks the readers themselves: every reader (ServerReader or
the legacy OpcManager, both keep `subscriptions` by slot) holding a
subscription for the slot unsubscribes it. A slot no reader holds is a no-op.
This also covers a resubscription whose new config dropped a server: the old
server's subscription for that slot is closed too.
Tests: unit tests for the new lookup and the double call, plus two regressions
at the Ingestor level with the real unsubscribe_slot (dropped extra lease;
slot removed from Redis). Without the fix they fail with the same KeyError at
ingestor_manager.py:558. 280 unit tests.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
SonarQube S7504 flagged `for slot in list(self.engine.subscriptions)` in
ServerReader.shutdown as an unnecessary list() call. It is necessary:
engine.unsubscribe() pops from engine.subscriptions, so iterating the dict
directly raises "dictionary changed size during iteration" and leaves the
remaining slots subscribed. Kept the snapshot with an explanatory comment and
`# NOSONAR`, the same convention as OpcConnectionPool.close_all().
Adds a regression test where unsubscribe() really pops: it fails with
RuntimeError without the list() and passes with it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The rc3 OpcManager is restored unchanged (with its tests and
scripts/generate-and-push-opc-cert.sh) for validation and as a fallback until
the end of SIENTIAPDE-1974; it is removed only once it is unused and the test
suite passes the same without it.
- READ_ENGINE=v2 (default): ServerReader/SubscriptionEngine over the shared
pool, certificate from MinIO. READ_ENGINE=legacy: the rc3 OpcManager with
its own connection, slot v1 only (a v2 slot is rejected and notified) and
the certificate files of the Gitea clone. An invalid value falls back to v2
with a warning. Writes/gRPC always use the pool.
- pool: OPC clients carry the pod id as application/session name and product
URI (as OpcManager did), so each pod's session is recognizable in the
server console; application_uri stays the server_uri (certificate SAN).
- README/CLAUDE.md document the plan B and how to remove it.
Tests: 275 unit tests pass (restored OpcManager tests unchanged); ruff clean.
Local e2e: both engines read the same slot v1 from an asyncua server and write
identical documents to the same collections.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- 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>
Swaps the legacy OpcManager read path for the gateway's SubscriptionEngine over
the shared OpcConnectionPool. The OPC client certificate now always comes from
MinIO + Mongo (CertProvider, Basic256Sha256) — no more certs/ shipped in the
Gitea repository, so a new version branch no longer breaks the reads.
- io/slots.py: parse_slot accepts slot v2 (names[]) and converts the legacy
v1 (tag_name + topics; cert_path => use_security, path dropped), so gateway
and Orchestrator can be deployed in any order.
- managers/server_reader.py: new per-server reader with the OpcManager
interface the IngestorManager already drives (connect, create_subscription,
subscribe, unsubscribe, check_cycles, check_opc_listenning, shutdown).
- IngestorManager owns the read path: lazy-MinIO CertProvider, the pool and a
read DataSink, built without touching the network. Slots from Redis are
normalized on lease and refresh; a malformed slot is notified and the last
good config is kept.
- GatewayRuntime borrows the pool (writes/gRPC) instead of building it, and no
longer closes it on stop: a Postgres/NATS failure can't stop the reads.
- pool: pin/unpin per active subscription so release_idle never evicts a
subscribed connection; public evict() for lost servers / changed config.
- SubscriptionEngine pins on subscribe, unpins on unsubscribe, and drops a
subscription locally even when the server-side delete fails.
- DataSink: model_id only written when the name has one, keeping v1 documents
identical to the legacy ones.
- config: CERT_BUCKET default is the legacy streamlit-connectors bucket (D7).
- Removes OpcManager, its tests and scripts/generate-and-push-opc-cert.sh;
README/CLAUDE.md document the MinIO certificate and the slot v2 format.
Tests: 218 unit tests pass; ruff check/format clean. Local end-to-end against
an asyncua server: v1 slot with two topics -> one doc per collection, legacy
doc shape, idle sweep keeps the subscription, lost-server shutdown and
re-initialization work.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- data_sink.py: _IDENTIFIER_RE uses \w+/re.ASCII instead of the
equivalent-but-verbose [A-Za-z0-9_]+ character class (verified same
match behavior). NOSONAR on update_write_status's 'async with no
await inside' - deliberate, per the ponytail comment already in the
method body (uniform async interface across DataSink, no async pg driver).
- test_grpc_server.py: move make_request() out of the pytest.raises
block so exactly one call can raise the expected exception.
- sonar-project.properties: exclude ingestor/proto/ (generated grpc
code) from analysis - its ReadTag/etc. names are dictated by
grpcio-tools, not a style choice, and renaming them breaks the wire.
main's requirements.txt unpins asyncua (PR #30), so CI installs 2.0.1,
where DataValue's field was renamed StatusCode_ -> StatusCode. The old
kwarg made every OPC write raise TypeError at runtime - masked locally
by a stale asyncua 1.0.6 in the dev conda env. Also uses ua.DateTime
(a datetime subclass) and ua.UInt32 to satisfy the new stricter types,
with no behavior change (verified via binary encode/decode round-trip).
GatewayRuntime builds and owns the F4 object graph (CertProvider/pool ->
WriteEngine/NatsWriteConsumer -> ActiveReadEngine -> optional GrpcServer),
reusing the existing ingestor's logger/notification_handler/metrics_controller
and DataManager (CertProvider shares its Mongo connection instead of opening
a second one). app.py starts it after prepare_ingestor, sweeps idle OPC
connections each loop iteration, and stops it on shutdown. A NATS connect
failure at startup is logged/notified but never crashes the app -- the read
path keeps working and idle_sweep() retries the consumer start.
ingestor.py needed no changes: data_manager is already reachable via
ingestor.ingestor_manager.data_manager.
Also documents the new MinIO/Postgres env vars this wiring reads (via
sientia_do's build_postgres_config for Postgres) in .env.example, and
updates test_app.py's fixture to mock GatewayConfig/GatewayRuntime so the
existing app tests keep exercising only what they mocked before.
IngestorManager.__del__ calls asyncio.run(self.shutdown()). Adding
tests/unit/connection/test_pool.py shifted GC timing enough that a
cyclic-referenced IngestorManager instance from an earlier test was
swept mid test___del__, double-calling the patched asyncio mock and
failing the assertion. Deterministic gc.collect() after each test in
this file keeps finalizers scoped to their own test.
Enhance release workflow to trigger only on merged pull requests and streamline data insertion in DataManager
- Updated the release workflow to execute only when a pull request is merged.
- Simplified the document insertion logic in DataManager by consolidating the dictionary construction.
- Refactored OPC Manager tests to utilize asynchronous mocks for improved accuracy in testing.
