gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature-rc/SIENTIAPDE-2164- fix: never let a metadata write failure block the prediction

Browse Source
This commit is contained in:
Carla Pinto
2026-10-01 19:14:24 +02:00
parent 9c628e70a6
commit 206b47dfae
2 changed files with 190 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -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,

170
tests/laborious/workflows/subworkflows/test_prediction_process.py
View File

@@ -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):