sientia-dataops-laborious_temporal/docs/opc-communication.md

OPC UA communication (Laborious)

Laborious exports predictions to OPC UA servers through OpcRepository (laborious/utils/repository/opc_repository.py) and the Temporal activity layer in laborious/activities/opc.py.

Implementation plan for session/channel recovery on Tier-1 Bad* errors: .cursor/plans/opc_bad_reconnect_ac4c6045.plan.md.

Architecture

Worker (long-lived)
  └── OpcRepository per OPC server id (from OPC_CONFIG / env)
        ├── connect / disconnect / validate_connection (read-only)
        ├── _connect_locked / _reconnect_locked (under _connection_lock)
        ├── write_data (single attempt per call)
        └── background reconnect on Tier-1 Bad*, protocol closed, or stale session

Temporal activity write_opc_data
  └── OPC.manage_output_tags → write_data per tag (sequential per activity)

One worker process holds one OpcRepository instance per configured server. Multiple Temporal activities can call write_data concurrently on the same repository.

Connection lifecycle

Phase	Behavior
Startup	init_opc() creates repositories and calls connect() → _connect_locked()
Steady state	validate_connection() is read-only (protocol.state only); _session_ready is checked in write_data
Tier-1 Bad* / protocol closed	_start_reconnect → _run_reconnect → _reconnect_locked() (respects reconnection_interval)
Write	write_data() checks reconnect task, _session_ready, validates protocol, then one get_node + write_value
Shutdown	close() disconnects all repositories

Session and channel timeouts

Requested session and secure-channel lifetime: 10 minutes (OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS in opc_repository.py). The server may revise these values; negotiated values are logged after connect and exposed as opc_session_revised_timeout_milliseconds.

Reconnection interval

OPC_RECONNECTION_INTERVAL is in seconds (default 120). It gates background reconnect after Tier-1 Bad*, closed protocol, or stale session (last_reconnection_time is updated only in _reconnect_locked()). It limits load on the OPC server when many workflows fail at once.

Concurrency: connection lock and session readiness

To allow multiple concurrent writes when the session is healthy, but block all writes while the connection is being torn down or re-established:

Primitive	Role
_connection_lock (asyncio.Lock)	Held for the entire disconnect → connect path. Only one connection-maintenance task at a time.
_session_ready (asyncio.Event)	Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect.

Connection methods (caller holds _connection_lock for _*_locked helpers):

Method	Role
_create_client()	Create asyncua Client + optional set_security; raises if client already exists
_open_session()	client.connect() + metrics; raises if session already open or client missing
_connect_locked()	_create_client() (when needed) + _open_session(); raises if already connected
_disconnect_locked()	Teardown session and clear client
_reconnect_locked()	_disconnect_locked() + _connect_locked(); sets last_reconnection_time

Public connect() / disconnect() acquire the lock and call _connect_locked() / _disconnect_locked().

Write path (write_data):

1. If a reconnect task is in flight → fail immediately (opc_error_kind=reconnect_in_progress).
2. If _session_ready is cleared and no task is running → schedule reconnect (SessionNotReady); fail with connection_lost or reconnect_in_progress if a task started.
3. validate_connection() checks protocol.state only (read-only). If closed → schedule reconnect (ProtocolClosed) and fail with opc_error_kind=connection_lost.
4. Single get_node + write_value (no retry). Tier-1 Bad* on write also schedules reconnect.

Reconnect path (_run_reconnect):

1. _start_reconnect clears _session_ready and schedules the task when the interval allows and _allow_reconnect is true.
2. async with _connection_lock: → _reconnect_locked().
3. _session_ready is set on successful _open_session().
4. disconnect() sets _allow_reconnect=False so shutdown does not respawn sessions.

A second _connect_locked() while a session is already open raises OpcSessionAlreadyConnectedError (disconnect first).

asyncua note: Concurrent write_value on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes with an optional asyncio.Semaphore(1) while keeping the connection lock semantics above.

Future threads: replace asyncio.Lock / Event with threading primitives or route all OPC I/O through one dedicated loop.

Reconnect triggers

Background reconnect is scheduled when:

• validate_connection() sees a closed or missing protocol (ProtocolClosed).
• _session_ready is clear after a failed reconnect (SessionNotReady).
• A write raises a Tier-1 UaStatusCodeError in RECONNECTABLE_OPC_BAD_NAMES.

For Tier-1 Bad* when the server invalidates the session (e.g. BadSessionIdInvalid) but the client still sees transport as open, write_data fails once, records the OPC status in metrics, and schedules reconnect if:

• The exception is a UaStatusCodeError whose name is in RECONNECTABLE_OPC_BAD_NAMES (see plan), and
• reconnection_interval has elapsed since last_reconnection_time, and
• No reconnect task is already running.

There is no write retry: the failed export is not sent again in the same activity.

Prediction confidence and PostgreSQL comments

prediction_confidence	Meaning
(unchanged)	Successful OPC export
12	Generic OPC write failure (OPC_WRITTING_ERROR_CONFIDENCE)
14	Tier-1 session/channel Bad* on export (OPC_SESSION_BAD_CONFIDENCE)
14	Write while reconnect in progress (OPC_SESSION_BAD_CONFIDENCE, comment OPC UA reconnect in progress)
13	PI Web API write failure (separate path)

Session/channel errors use a stable comment for counting:

OPC UA session/channel error: BadSessionIdInvalid

Reconnect-in-progress exports use:

OPC UA reconnect in progress

Example SQL:

SELECT count(*) FROM predictions WHERE prediction_confidence = 14;
SELECT count(*) FROM predictions WHERE comments LIKE 'OPC UA session/channel error:%';

Prometheus metrics (opc_*)

Defined in laborious/metrics.py. Do not rename in production without a dashboard migration.

Metric	Purpose
opc_connections_initiated_total	Connection attempts
opc_connections_failed_total	Failed connects
opc_connection_status	Gauge 1=connected, 0=disconnected
opc_session_created_total	Session established after connect
opc_session_closed_total	Disconnect initiated
opc_session_revised_timeout_milliseconds	Negotiated session timeout (ms)
opc_write_attempts_total	Per write; label result = OK or exception name
opc_write_inter_arrival_over_session_timeout_total	Successful writes spaced longer than revised session timeout

Legacy activity metrics: laborious_prediction_opc_writing_count, laborious_prediction_opc_writing_response_time_monitor.

Environment variables

Variable	Default	Description
OPC_CONFIG	—	JSON map of server configs (overrides single-server env)
OPC_ID	1	Server id
OPC_URL	opc.tcp://localhost:4840	Endpoint
OPC_SERVER_NAME	default_server	Label for metrics/logs
OPC_SERVER_URI	same as URL	Application URI / cert SAN
OPC_CERT_PATH	—	Client certificate (secure mode)
OPC_PRIVATE_KEY_PATH	—	Client private key
OPC_SERVER_CERT_PATH	—	Server certificate
OPC_RECONNECTION_INTERVAL	120	Minimum seconds between reconnects

Operations checklist

• Correlate BadSessionIdInvalid in opc_write_attempts_total with opc_session_closed_total / opc_session_created_total (reconnect may finish after the row is stored with confidence 14).
• Use confidence 14 and comment prefix for session invalidation rates; use 12 for other OPC failures.
• Respect OPC_RECONNECTION_INTERVAL under parallel load; bursts of confidence 14 are expected until the next successful cycle.

Related tests

• Unit: tests/laborious/utils/repository/test_opc_repository.py
• Unit: tests/laborious/activities/test_opc.py
• E2E (mock OPC): e2e/test_predictions_batch_format_export.py
• E2E (in-process asyncua server + real OpcRepository): e2e/test_opc_real_server.py — scenarios 3.1.2, 3.2.2, 3.2.4, 3.2.5
• Scenarios: e2e/scenarios.md