From cd1be2430acf871600778f5006650a54ad8046ff Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 May 2026 11:45:08 -0300 Subject: [PATCH] SIENTIAPDE-1811 Update .gitignore, requirements, and enhance OPC UA error handling - Added new entries to .gitignore for openspec and cursor directories. - Updated sientia-dataops-library dependency version in requirements-light.txt from 1.10.4 to 1.12.0. - Enhanced OPC UA communication by refining reconnect logic and error handling in opc_repository.py, including the introduction of a reconnect flag and improved session management. - Updated tests to cover new reconnect scenarios and ensure robust error handling for protocol states. --- .gitignore | 4 +- docs/opc-communication.md | 30 ++- laborious/utils/repository/opc_repository.py | 194 ++++++++++++++---- laborious/worker/worker.py | 1 + requirements-light.txt | 2 +- .../utils/repository/test_opc_repository.py | 96 ++++++++- 6 files changed, 266 insertions(+), 61 deletions(-) diff --git a/.gitignore b/.gitignore index c35c913..c849e0a 100644 --- a/.gitignore +++ b/.gitignore @@ -53,4 +53,6 @@ catboost_info/ .mypy_cache/ mlruns/ -relatorio* \ No newline at end of file +relatorio* +openspec/ +.cursor/ \ No newline at end of file diff --git a/docs/opc-communication.md b/docs/opc-communication.md index 3084d05..15d78e7 100644 --- a/docs/opc-communication.md +++ b/docs/opc-communication.md @@ -12,7 +12,7 @@ Worker (long-lived) ├── 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* + └── 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) @@ -26,8 +26,8 @@ One worker process holds one `OpcRepository` instance per configured server. Mul |-------|----------| | 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* | `_start_reconnect_on_bad` → `_run_reconnect_on_bad` → `_reconnect_locked()` (respects `reconnection_interval`) | -| Write | `write_data()` checks `_session_ready`, validates, then one `get_node` + `write_value` | +| 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 @@ -36,7 +36,7 @@ Requested session and secure-channel lifetime: **10 minutes** (`OPC_UA_SESSION_A ### Reconnection interval -`OPC_RECONNECTION_INTERVAL` is in **seconds** (default `120`). It gates **background** reconnect after Tier-1 `Bad*` (`last_reconnection_time` is updated only in `_reconnect_locked()`). It limits load on the OPC server when many workflows fail at once. +`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 @@ -61,15 +61,17 @@ Public `connect()` / `disconnect()` acquire the lock and call `_connect_locked() **Write path (`write_data`):** -1. If `_session_ready` is cleared → **fail immediately** (`opc_error_kind=reconnect_in_progress`). -2. `validate_connection()` checks `protocol.state` only (read-only). -3. Single `get_node` + `write_value` (no retry). +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_on_bad`):** +**Reconnect path (`_run_reconnect`):** -1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows. +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). @@ -77,9 +79,15 @@ A second `_connect_locked()` while a session is already open raises `OpcSessionA **Future threads:** replace `asyncio.Lock` / `Event` with `threading` primitives or route all OPC I/O through one dedicated loop. -## Tier-1 `Bad*` errors and reconnect +## Reconnect triggers -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: +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 diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index b164ce1..2ecc27a 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -195,8 +195,18 @@ class OpcRepository(SientiaMonitoring): self._connection_lock = asyncio.Lock() self._session_ready = asyncio.Event() self._reconnect_task: asyncio.Task[None] | None = None + self._allow_reconnect = True def _opc_debug_tags(self, session_id: str) -> dict[str, str]: + """ + Build Prometheus/log label tags for OPC session-scoped metrics. + + Args: + session_id (str): OPC UA session token string. + + Return: + dict[str, str]: Labels pod_id, server_name, runtime, opc_server_id, session_id. + """ return { 'pod_id': str(getattr(self, 'pod_id', 'unknown')), 'server_name': self.server_name, @@ -234,6 +244,12 @@ class OpcRepository(SientiaMonitoring): ).total_seconds() > self.reconnection_interval def _not_connected_error(self) -> dict[str, Any]: + """ + Build the standard error payload when validate_connection finds no open protocol. + + Return: + dict[str, Any]: Notification fields for OPC_CONNECTION_NOT_READY. + """ return { 'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}', 'message': f'OPC server {self.id} is not connected', @@ -491,8 +507,12 @@ class OpcRepository(SientiaMonitoring): async def disconnect(self) -> None: """ Gracefully disconnect from the OPC server under the connection lock. + + Disables background reconnect so late writes during worker shutdown do not + respawn sessions. """ async with self._connection_lock: + self._allow_reconnect = False await self._disconnect_locked() async def validate_connection(self) -> tuple[bool, dict[str, Any]]: @@ -509,58 +529,77 @@ class OpcRepository(SientiaMonitoring): self.error(f'OPC server {self.id} is not connected', self.metadata) return False, self._not_connected_error() - async def _start_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + def _reconnect_task_in_progress(self) -> bool: """ - Schedule a background reconnect if interval and task state allow it. + Return whether a background reconnect task is currently running. + + Return: + bool: True when a reconnect task exists and has not finished. + """ + return self._reconnect_task is not None and not self._reconnect_task.done() + + async def _start_reconnect(self, reason: str, session_id: str) -> None: + """ + Schedule a background reconnect when allowed by interval and task state. + + Clears _session_ready before starting the task. No-op when _allow_reconnect is + False, the reconnection window has not elapsed, or a reconnect is already running. Args: - opc_status (str): OPC UA status name that triggered reconnect. + reason (str): Trigger for reconnect (OPC status name or synthetic reason). session_id (str): Session token before failure. """ + if not self._allow_reconnect: + return if not self._reconnection_window_elapsed(): self.warning( f'OPC reconnect skipped reason=reconnection_window opc_server_id={self.id} ' - f'opc_status={opc_status}', + f'reconnect_reason={reason}', self.metadata, ) return - if self._reconnect_task is not None and not self._reconnect_task.done(): + if self._reconnect_task_in_progress(): self.warning( f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} ' - f'opc_status={opc_status}', + f'reconnect_reason={reason}', self.metadata, ) return self._session_ready.clear() self.info( - f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} ' + f'OPC reconnect scheduled reconnect_reason={reason} opc_server_id={self.id} ' f'old_session_id={session_id}', self.metadata, ) - self._reconnect_task = asyncio.create_task( - self._run_reconnect_on_bad(opc_status, session_id) - ) + self._reconnect_task = asyncio.create_task(self._run_reconnect(reason, session_id)) - async def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None: + async def _run_reconnect(self, reason: str, session_id: str) -> None: """ - Tear down and re-establish the OPC UA session under the connection lock. + Background task that tears down and re-establishes the OPC UA session. Args: - opc_status (str): OPC UA status that triggered reconnect. - session_id (str): Previous session token string. + reason (str): Trigger for reconnect (OPC status or ProtocolClosed). + session_id (str): Previous session token string for logging. """ try: async with self._connection_lock: self.info( - f'OPC reconnect started opc_status={opc_status} opc_server_id={self.id} ' + f'OPC reconnect started reconnect_reason={reason} opc_server_id={self.id} ' f'old_session_id={session_id}', self.metadata, ) - await self._reconnect_locked() + success, error = await self._reconnect_locked() + if not success: + self.error( + f'OPC reconnect failed reconnect_reason={reason} opc_server_id={self.id}', + self.metadata, + ) + if error: + self.error(error.get('message', ''), self.metadata) except Exception: self.error( - f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}', + f'OPC reconnect task failed opc_server_id={self.id} reconnect_reason={reason}', self.metadata, ) self.error(traceback.format_exc(), self.metadata) @@ -593,6 +632,14 @@ class OpcRepository(SientiaMonitoring): async def _emit_opc_write_metric( self, session_id: str, result: str, metadata: dict[str, Any] | None ) -> None: + """ + Emit opc_write_attempts_total for a single write attempt outcome. + + Args: + session_id (str): OPC UA session token string, or "unknown". + result (str): Outcome label (OK, OPC status name, ProtocolClosed, etc.). + metadata (dict[str, Any] | None): Write context for model_id/model_name labels. + """ await self.emit_metric( metrics.OPC_WRITE_ATTEMPTS_TOTAL, { @@ -611,6 +658,20 @@ class OpcRepository(SientiaMonitoring): opc_error_kind: str | None = None, opc_status: str | None = None, ) -> dict[str, Any]: + """ + Build a structured error dict returned from failed write_data paths. + + Args: + notification_id (str): Stable notification identifier. + message (str): Human-readable failure message. + level (NotificationLevel): Severity for downstream notifications. + attachment_content (str | None): Optional traceback or diagnostic text. + opc_error_kind (str | None): Classifier (session_bad, connection_lost, etc.). + opc_status (str | None): OPC UA status name or synthetic reason. + + Return: + dict[str, Any]: Error payload consumed by the OPC activity layer. + """ payload: dict[str, Any] = { 'notification_id': notification_id, 'message': message, @@ -656,7 +717,7 @@ class OpcRepository(SientiaMonitoring): f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}', metadata, ) - await self._start_reconnect_on_bad(opc_status, session_id) + await self._start_reconnect(opc_status, session_id) return False, self._write_failure_payload( notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}', message=f'Failed to {phase} on OPC server: {exc} | metadata: {metadata}', @@ -665,32 +726,89 @@ class OpcRepository(SientiaMonitoring): opc_status=opc_status, ) + async def _write_reconnect_in_progress( + self, metadata: dict[str, Any] + ) -> tuple[bool, dict[str, Any]]: + """ + Fail a write because a background reconnect task is already running. + + Args: + metadata (dict[str, Any]): Write context passed through to the activity. + + Return: + tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind reconnect_in_progress). + """ + await self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata) + self.warning( + f'OPC write rejected reconnect_in_progress opc_server_id={self.id} ' + f'model_id={metadata.get("model_id", "unknown")} ' + f'model_name={metadata.get("model_name", "unknown")}', + metadata, + ) + return False, { + 'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}', + 'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}', + 'block': 'opc_repository', + 'level': NotificationLevel.WARNING, + 'opc_error_kind': 'reconnect_in_progress', + } + + async def _write_connection_lost( + self, metadata: dict[str, Any], opc_status: str + ) -> tuple[bool, dict[str, Any]]: + """ + Fail a write after scheduling reconnect for a closed or stale session. + + Args: + metadata (dict[str, Any]): Write context passed through to the activity. + opc_status (str): Synthetic reason (ProtocolClosed, SessionNotReady). + + Return: + tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind connection_lost). + """ + await self._emit_opc_write_metric('unknown', opc_status, metadata) + return False, self._write_failure_payload( + notification_id=f'OPC_WRITE_CONNECTION_LOST_{self.id}', + message=f'OPC write skipped: connection lost ({opc_status}) | metadata: {metadata}', + level=NotificationLevel.WARNING, + opc_error_kind='connection_lost', + opc_status=opc_status, + ) + async def write_data( self, node: str, value: Any, data_type: str, metadata: dict[str, Any] ) -> tuple[bool, dict[str, Any]]: """ - Write data to OPC server with a single attempt and Tier-1 Bad* reconnect scheduling. - """ - if not self._session_ready.is_set(): - await self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata) - self.warning( - f'OPC write rejected reconnect_in_progress opc_server_id={self.id} ' - f'model_id={metadata.get("model_id", "unknown")} ' - f'model_name={metadata.get("model_name", "unknown")}', - metadata, - ) - return False, { - 'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}', - 'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}', - 'block': 'opc_repository', - 'level': NotificationLevel.WARNING, - 'opc_error_kind': 'reconnect_in_progress', - } + Write data to OPC server with a single attempt and background reconnect scheduling. - is_connected, error = await self.validate_connection() + Reconnect is scheduled on Tier-1 Bad*, closed protocol, or stale session readiness. + There is no retry within the same call. + + Args: + node (str): OPC UA node id to write. + value (Any): Value to convert and send. + data_type (str): Logical type key (float, int, bool, str, double). + metadata (dict[str, Any]): Activity context (model_id, model_name, etc.). + + Return: + tuple[bool, dict[str, Any]]: (True, {response_time}) on success, or + (False, structured error info) on failure. + """ + if self._reconnect_task_in_progress(): + return await self._write_reconnect_in_progress(metadata) + + if not self._session_ready.is_set(): + session_id = _opc_authentication_token_str(self.client) + await self._start_reconnect('SessionNotReady', session_id) + if self._reconnect_task_in_progress(): + return await self._write_reconnect_in_progress(metadata) + return await self._write_connection_lost(metadata, 'SessionNotReady') + + is_connected, _error = await self.validate_connection() if not is_connected: - await self._emit_opc_write_metric('unknown', 'NotConnected', metadata) - return False, error + session_id = _opc_authentication_token_str(self.client) + await self._start_reconnect('ProtocolClosed', session_id) + return await self._write_connection_lost(metadata, 'ProtocolClosed') start_time = time.time() session_id = _opc_authentication_token_str(self.client) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index fe78753..69e6d2f 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -205,6 +205,7 @@ async def main(): activities.cleanup_minio_objects_expired, activities.repeat_last_prediction, activities.export_data_to_postgres, + activities.export_payload_to_postgres, activities.write_metrics, # Pi Web API activities.write_pi_web_api_data, diff --git a/requirements-light.txt b/requirements-light.txt index 91a9c33..f105052 100644 --- a/requirements-light.txt +++ b/requirements-light.txt @@ -3,7 +3,7 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.10.4 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.0 prometheus-client botocore boto3 diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index f1fd81a..365a400 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -257,6 +257,7 @@ async def test_disconnect(opc_repository, mock_client): opc_repository._disconnection_fallback.assert_called_once() assert opc_repository.client is None + assert opc_repository._allow_reconnect is False @pytest.mark.asyncio @@ -331,14 +332,19 @@ async def test_write_data_validate_connection_do_nothing(opc_repository): @pytest.mark.asyncio async def test_write_data_validate_connection_failed(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(False, {})) - opc_repository.client = AsyncMock() + opc_repository.client = MagicMock() + opc_repository.client.uaclient.protocol = MagicMock(state='closed') + opc_repository._start_reconnect = AsyncMock() - result = await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_not_called() - assert result == (False, {}) + opc_repository._start_reconnect.assert_called_once() + assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed' + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + assert error_data['opc_status'] == 'ProtocolClosed' @pytest.mark.asyncio @@ -439,7 +445,7 @@ def test_is_reconnectable_opcua_bad(): async def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, mock_client): opc_repository.validate_connection = AsyncMock(return_value=(True, {})) opc_repository.client = mock_client - opc_repository._start_reconnect_on_bad = AsyncMock() + opc_repository._start_reconnect = AsyncMock() mock_node = AsyncMock() mock_client.get_node = MagicMock(return_value=mock_node) mock_node.write_value.side_effect = BadSessionIdInvalid() @@ -449,7 +455,7 @@ async def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_reposit ) mock_node.write_value.assert_called_once() - opc_repository._start_reconnect_on_bad.assert_called_once() + opc_repository._start_reconnect.assert_called_once() assert is_success is False assert error_data['opc_error_kind'] == 'session_bad' assert error_data['opc_status'] == 'BadSessionIdInvalid' @@ -476,15 +482,85 @@ async def test_write_data_reconnect_in_progress_immediate(opc_repository): @pytest.mark.asyncio -async def test_start_reconnect_on_bad_skips_within_interval(opc_repository): +async def test_start_reconnect_skips_within_interval(opc_repository): opc_repository.last_reconnection_time = datetime.now() opc_repository.reconnection_interval = 3600 - await opc_repository._start_reconnect_on_bad('BadSessionIdInvalid', 'tok') + await opc_repository._start_reconnect('BadSessionIdInvalid', 'tok') assert opc_repository._reconnect_task is None +@pytest.mark.asyncio +async def test_write_data_protocol_closed_schedules_reconnect(opc_repository): + opc_repository.client = MagicMock() + opc_repository.client.uaclient.protocol = MagicMock(state='closed') + opc_repository._start_reconnect = AsyncMock() + + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + opc_repository._start_reconnect.assert_called_once() + assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed' + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + assert error_data['opc_status'] == 'ProtocolClosed' + + +@pytest.mark.asyncio +async def test_write_data_protocol_closed_skips_reconnect_within_interval(opc_repository): + opc_repository.client = MagicMock() + opc_repository.client.uaclient.protocol = MagicMock(state='closed') + opc_repository.last_reconnection_time = datetime.now() + opc_repository.reconnection_interval = 3600 + + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + assert opc_repository._reconnect_task is None + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + + +@pytest.mark.asyncio +async def test_write_data_after_failed_reconnect_schedules_again(opc_repository): + opc_repository._session_ready.clear() + opc_repository.reconnection_interval = 0 + opc_repository.last_reconnection_time = None + opc_repository._reconnect_locked = AsyncMock( + return_value=(False, {'message': 'connect failed'}) + ) + + await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + await asyncio.sleep(0.1) + assert opc_repository._reconnect_locked.call_count == 1 + assert not opc_repository._reconnect_task_in_progress() + + await opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata']) + await asyncio.sleep(0.1) + assert opc_repository._reconnect_locked.call_count == 2 + + +@pytest.mark.asyncio +async def test_write_data_after_disconnect_does_not_schedule_reconnect(opc_repository, mock_client): + opc_repository.client = mock_client + proto = MagicMock() + proto.state = 'closed' + mock_client.uaclient = MagicMock(protocol=proto) + opc_repository._disconnection_fallback = AsyncMock(return_value=[]) + await opc_repository.disconnect() + + is_success, error_data = await opc_repository.write_data( + 'ns=2;s=TestNode', 42.0, 'float', metadata['metadata'] + ) + + assert opc_repository._reconnect_task is None + assert is_success is False + assert error_data['opc_error_kind'] == 'connection_lost' + + @pytest.mark.asyncio async def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client): opc_repository.validate_connection = AsyncMock(return_value=(True, {}))