From 206b47dfae3453609436c2eea516896d895661da Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 1 Oct 2026 19:14:24 +0200 Subject: [PATCH] feature-rc/SIENTIAPDE-2164- fix: never let a metadata write failure block the prediction --- .../sub_workflows/prediction_process.py | 34 +++- .../subworkflows/test_prediction_process.py | 170 +++++++++++++++++- 2 files changed, 190 insertions(+), 14 deletions(-) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 7f3bfac..e8c3064 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -276,15 +276,33 @@ class PredictionProcess: build one yet. Return: - The generated id, or `None` when there's nothing to write (see - `PredictionMetadata.write_prediction_metadata`). + The generated id, or `None` when there's nothing to write, including when the + activity call itself fails (see `PredictionMetadata.write_prediction_metadata`). """ - return await workflow.execute_activity_method( - Activities.write_prediction_metadata, - {**metadata, 'documents': [document] if document else []}, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(minutes=1), - ) + if not document: + # Skip the remote activity call entirely rather than dispatching it just to have + # `write_prediction_metadata` decide there's nothing to write — `mlflow_response_gate` + # and `mlflow_content_gate` don't build a `document` yet, so this is the common case + # today, not a rare edge case (review finding, 2026-10-01). + return None + + try: + return await workflow.execute_activity_method( + Activities.write_prediction_metadata, + {**metadata, 'documents': [document]}, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(minutes=1), + ) + except Exception as e: + # Business rule 1: writing metadata must never block or fail the prediction write. + # `write_prediction_metadata` already swallows Mongo-specific failures internally and + # returns `None`, but that protection doesn't cover the activity call itself failing + # to run at all (its `retry_policy` has `maximum_attempts=1` — no retry — so a + # timeout, a momentarily unavailable worker, or any other dispatch failure raises + # straight into the workflow). Catching it here, instead of just at the Mongo client + # level, is what actually makes the rule hold (review finding, 2026-10-01). + workflow.logger.warning(f'Failed to write prediction metadata, continuing anyway: {e}') + return None async def path_flag_handler( self, diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 92715a2..82abe8c 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -692,12 +692,13 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): @mark.asyncio @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) -async def test_path_flag_handler_repeat_no_document_writes_nothing( +async def test_path_flag_handler_repeat_no_document_skips_write_activity_call( workflow_mock, prediction_process ): - # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet) still - # calls write_prediction_metadata, with an empty `documents` list — the activity itself (not - # this workflow) decides there's nothing to write and returns `None`. + # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today, + # `mlflow_response_gate`/`mlflow_content_gate`) must not even dispatch the remote + # write_prediction_metadata activity — that would be a wasted Temporal round-trip on every + # REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01). data = {'test': 'data'} # Act @@ -721,9 +722,19 @@ async def test_path_flag_handler_repeat_no_document_writes_nothing( # Assert assert result is True + called_activities = [c.args[0] for c in workflow_mock.execute_activity_method.call_args_list] + assert Activities.write_prediction_metadata not in called_activities + # repeat_last_prediction still happens, just with metadata_id=None. workflow_mock.execute_activity_method.assert_any_call( - Activities.write_prediction_metadata, - {**metadata, 'documents': []}, + Activities.repeat_last_prediction, + { + **metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'model': 'test_model', + 'last_timestamp': '2024-01-01', + 'metadata_id': None, + }, retry_policy=ANY, start_to_close_timeout=ANY, ) @@ -811,6 +822,153 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): ) +@mark.asyncio +@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) +async def test_path_flag_handler_continue_no_document_skips_write_activity_call( + workflow_mock, prediction_process +): + # Arrange: same gap as the REPEAT case above (review finding, 2026-10-01) — no document must + # not dispatch the remote write_prediction_metadata activity. + data = {'test': 'data'} + + # Act + result = await prediction_process.path_flag_handler( + data, + 'continue', + { + 'metadata': metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'transform_table_name': 'test_transform_table', + 'model_id': 'test_model', + 'last_timestamp': '2024-01-01', + 'model_name': 'test_model_name', + 'model_config': {'retention': '30'}, + 'opc_output_config': {'test': 'config'}, + 'pi_web_api_output_config': { + 'endpoint': 'https://test-pi-server.com', + 'prediction_tags': {}, + 'confidence_tags': {}, + }, + 'prediction_store_policy': 'erl:1', + }, + 0.95, + '2024-01-01', + '', + ) + + # Assert + assert result is True + workflow_mock.execute_activity_method.assert_not_called() + workflow_mock.execute_child_workflow.assert_called_once_with( + 'subworkflow.format_and_export_prediction', + ANY, + ) + _, export_input = workflow_mock.execute_child_workflow.call_args.args + assert export_input['metadata_id'] is None + + +@mark.asyncio +@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) +async def test_path_flag_handler_continue_write_metadata_failure_does_not_block_prediction( + workflow_mock, prediction_process +): + # Arrange: write_prediction_metadata's retry_policy has maximum_attempts=1 — a timeout, an + # unavailable worker, or any dispatch failure raises straight into the workflow with no + # retry. Business rule 1 says this must never block the prediction write (review finding, + # 2026-10-01) — confirm the CONTINUE branch still proceeds instead of letting it propagate. + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]} + workflow_mock.execute_activity_method.side_effect = RuntimeError('activity dispatch failed') + + # Act + result = await prediction_process.path_flag_handler( + {'test': 'data'}, + 'continue', + { + 'metadata': metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'transform_table_name': 'test_transform_table', + 'model_id': 'test_model', + 'last_timestamp': '2024-01-01', + 'model_name': 'test_model_name', + 'model_config': {'retention': '30'}, + 'opc_output_config': {'test': 'config'}, + 'pi_web_api_output_config': { + 'endpoint': 'https://test-pi-server.com', + 'prediction_tags': {}, + 'confidence_tags': {}, + }, + 'prediction_store_policy': 'erl:1', + }, + 0.95, + '2024-01-01', + '', + document, + ) + + # Assert: the prediction write still happens, with metadata_id=None, instead of raising. + assert result is True + workflow_mock.execute_child_workflow.assert_called_once_with( + 'subworkflow.format_and_export_prediction', + ANY, + ) + _, export_input = workflow_mock.execute_child_workflow.call_args.args + assert export_input['metadata_id'] is None + + +@mark.asyncio +@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) +async def test_path_flag_handler_repeat_write_metadata_failure_does_not_block_prediction( + workflow_mock, prediction_process +): + # Arrange: only the write_prediction_metadata call (first) fails — repeat_last_prediction + # (second) must still run normally; its own failure is a separate, intentional + # non-best-effort path, not what this test is about. + document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]} + workflow_mock.execute_activity_method.side_effect = [ + RuntimeError('activity dispatch failed'), + MagicMock(), + ] + + # Act + result = await prediction_process.path_flag_handler( + {'test': 'data'}, + 'repeat', + { + 'metadata': metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'transform_table_name': 'test_transform_table', + 'model_id': 'test_model', + 'last_timestamp': '2024-01-01', + 'model_name': 'test_model_name', + 'model_config': {'retention': '30'}, + }, + 0.95, + '2024-01-01', + '', + document, + ) + + # Assert: repeat_last_prediction still runs, with metadata_id=None, instead of raising. + assert result is True + assert workflow_mock.execute_activity_method.call_count == 2 + workflow_mock.execute_activity_method.assert_any_call( + Activities.repeat_last_prediction, + { + **metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'model': 'test_model', + 'last_timestamp': '2024-01-01', + 'metadata_id': None, + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + + @mark.asyncio @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock) async def test_path_flag_handler_unknown(workflow_mock, prediction_process):