417 lines
18 KiB
Python
417 lines
18 KiB
Python
"""The `import_model` workflow: a `.sientia` object becomes a registered model, or a durable ERROR.
|
|
|
|
Shape of this workflow, and the reasons it has that shape:
|
|
|
|
- **The claim comes first.** The record already exists — the frontend inserted it through the BFF —
|
|
so the workflow's first act is to take ownership of it. Everything after that failure is recorded
|
|
*on* the record; a failed claim is recorded nowhere, because the workflow owns no row.
|
|
- **Fail-fast, and nothing else.** No rollback, no compensation, no resume, no reconciliation, no
|
|
`continue_as_new`. A failed import stays failed; trying again is a new import with a new upload.
|
|
- **Only the status writes retry**, and only because `UPDATE ... SET <fixed values> WHERE id` is
|
|
idempotent. Every provisioning call keeps `maximum_attempts=1`, because a repeated registry call or
|
|
Mongo insert can write twice.
|
|
- **One `finally`, two actions**: cleanup, then the single terminal write that classifies the
|
|
outcome once. No `except` block writes a verdict of its own.
|
|
- **The workflow holds ids, never contents.** The decrypted bundle never crosses an activity
|
|
boundary, and the password reaches this module only as an envelope it cannot open.
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
from temporalio import workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from typing import Any
|
|
|
|
from sientia_do.temporal.policies import retry_policy
|
|
|
|
from laborious.activities.activities import Activities
|
|
from laborious.activities.model_import import (
|
|
IMPORT_STATUS_HINT_RETRY_POLICY,
|
|
IMPORT_STATUS_RETRY_POLICY,
|
|
derive_run_name,
|
|
)
|
|
from laborious.activities.model_import_errors import ImportInputError, failure_location
|
|
from laborious.utils.bundle.steps import ImportStep
|
|
|
|
# `start_to_close_timeout` is per attempt. The terminal write additionally gets a
|
|
# `schedule_to_close_timeout` larger than its retry envelope (10 attempts, ≈151 s of backoff, ≈251 s
|
|
# if every attempt also burns its timeout) — set it smaller and Temporal cuts the retries short,
|
|
# which is the one configuration mistake that would silently undo design D20.
|
|
STATUS_WRITE_TIMEOUT = timedelta(seconds=10)
|
|
STATUS_WRITE_ENVELOPE = timedelta(seconds=300)
|
|
DOWNLOAD_TIMEOUT = timedelta(seconds=600)
|
|
OPEN_TIMEOUT = timedelta(seconds=900)
|
|
EXPERIMENT_TIMEOUT = timedelta(seconds=60)
|
|
ARTIFACT_UPLOAD_TIMEOUT = timedelta(seconds=900)
|
|
REGISTRATION_TIMEOUT = timedelta(seconds=120)
|
|
DOCUMENT_TIMEOUT = timedelta(seconds=60)
|
|
CLEANUP_TIMEOUT = timedelta(seconds=120)
|
|
RETENTION_TIMEOUT = timedelta(seconds=60)
|
|
|
|
# Fields no import may carry. The project is a frontend concern: the model listing is the MongoDB
|
|
# `models` document, no migrated database holds a project or model table, and nothing in this
|
|
# repository reads a project-to-model link (design D22). An input naming one is rejected rather than
|
|
# accepted and ignored, so no caller believes a project link was recorded. The same goes for a
|
|
# relational target: there is none, and for a plaintext password, which does not belong in an event
|
|
# history whatever else is true.
|
|
REJECTED_INPUT_FIELDS = (
|
|
'project_id',
|
|
'project_name',
|
|
'provisioning_target',
|
|
'schema',
|
|
'table_name',
|
|
'status_table',
|
|
'password',
|
|
)
|
|
|
|
DIGEST_LENGTH = 64
|
|
_HEX_DIGITS = frozenset('0123456789abcdef')
|
|
|
|
|
|
@workflow.defn(name='import_model')
|
|
class ImportModel:
|
|
"""Turn an encrypted `.sientia` object in MinIO into a registered model, or into an ERROR."""
|
|
|
|
@workflow.run
|
|
async def run(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Execute the import.
|
|
|
|
Args:
|
|
input_data: the import request.
|
|
Required keys:
|
|
- import_run_id (int): primary key of the import log row the frontend inserted
|
|
- object_key (str): the uploaded object's key
|
|
- expected_digest (str): the SHA-256 the uploader computed for those bytes
|
|
- password_envelope (str): the encrypted bundle password, never plaintext
|
|
Optional keys:
|
|
- bucket (str): defaults to the configured import bucket
|
|
|
|
Returns:
|
|
dict: `import_run_id`, `run_id`, `model_name`, `version` and the model document's id.
|
|
|
|
Raises:
|
|
ImportLogRowNotClaimableError: the record is not this workflow's — the run ends FAILED
|
|
with that error type and nothing is written to any row.
|
|
Exception: whatever failed. The record carries the failing step, its code and its
|
|
sentence; this exception carries the technical truth.
|
|
"""
|
|
workflow_id = workflow.info().workflow_id
|
|
import_run_id = self._require_import_run_id(input_data.get('import_run_id'))
|
|
metadata: dict[str, Any] = {
|
|
'model_name': '-',
|
|
'model_id': str(import_run_id),
|
|
'workflow_name': 'import_model',
|
|
'schedule_name': '-',
|
|
}
|
|
base = {
|
|
'metadata': metadata,
|
|
'import_run_id': import_run_id,
|
|
'workflow_id': workflow_id,
|
|
}
|
|
|
|
# Class A. Outside the try/finally on purpose: a claim that matches nothing leaves this
|
|
# workflow owning no row, so there is nothing to clean up and nothing to record.
|
|
claim = await workflow.execute_activity_method(
|
|
Activities.claim_import_status,
|
|
base,
|
|
retry_policy=IMPORT_STATUS_RETRY_POLICY,
|
|
start_to_close_timeout=STATUS_WRITE_TIMEOUT,
|
|
schedule_to_close_timeout=STATUS_WRITE_ENVELOPE,
|
|
)
|
|
|
|
current_step = ImportStep.RECEIVED
|
|
failed_step: str | None = None
|
|
gate: int | None = None
|
|
succeeded = False
|
|
source: dict[str, Any] = {}
|
|
result: dict[str, Any] = {'import_run_id': import_run_id}
|
|
|
|
try:
|
|
self._validate_input(input_data, claim)
|
|
|
|
current_step = ImportStep.DOWNLOAD
|
|
await self._hint(base, current_step)
|
|
await self._ensure_retention(base, input_data)
|
|
download = await workflow.execute_activity_method(
|
|
Activities.download_import_bundle,
|
|
{
|
|
**base,
|
|
'bucket': input_data.get('bucket'),
|
|
'object_key': input_data['object_key'],
|
|
'expected_digest': input_data['expected_digest'],
|
|
},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=DOWNLOAD_TIMEOUT,
|
|
)
|
|
|
|
current_step = ImportStep.DECRYPTION
|
|
await self._hint(base, current_step)
|
|
opened = await workflow.execute_activity_method(
|
|
Activities.open_import_bundle,
|
|
{
|
|
**base,
|
|
'path': download['path'],
|
|
'password_envelope': input_data['password_envelope'],
|
|
},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=OPEN_TIMEOUT,
|
|
)
|
|
source = opened['metadata']
|
|
parameters = opened['parameters']
|
|
model_name = source['model_name']
|
|
metadata['model_name'] = model_name
|
|
run_name = derive_run_name(model_name, source['model_version'])
|
|
|
|
# The names go on the record before anything exists in MLflow, so a person watching the
|
|
# import list sees *which* model is being imported while it is still running.
|
|
await self._record_names(base, source, run_name)
|
|
|
|
current_step = ImportStep.EXPERIMENT_CREATION
|
|
await self._hint(base, current_step, source)
|
|
experiment = await workflow.execute_activity_method(
|
|
Activities.create_import_experiment,
|
|
{**base, 'experiment_name': source['experiment_name']},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=EXPERIMENT_TIMEOUT,
|
|
)
|
|
|
|
current_step = ImportStep.ARTIFACT_UPLOAD
|
|
await self._hint(base, current_step, source)
|
|
run = await workflow.execute_activity_method(
|
|
Activities.upload_import_artifacts,
|
|
{
|
|
**base,
|
|
'experiment_id': experiment['experiment_id'],
|
|
'run_name': run_name,
|
|
'extracted_dir': opened['extracted_dir'],
|
|
'parameters': parameters,
|
|
'source': source,
|
|
},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=ARTIFACT_UPLOAD_TIMEOUT,
|
|
)
|
|
|
|
current_step = ImportStep.REGISTRATION
|
|
await self._hint(base, current_step, source)
|
|
version = await workflow.execute_activity_method(
|
|
Activities.register_import_model_version,
|
|
{**base, 'model_name': model_name, 'run_id': run['run_id']},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=REGISTRATION_TIMEOUT,
|
|
)
|
|
|
|
current_step = ImportStep.MODEL_DOCUMENT
|
|
await self._hint(base, current_step, source)
|
|
document = await workflow.execute_activity_method(
|
|
Activities.write_import_model_document,
|
|
{
|
|
**base,
|
|
'model_name': model_name,
|
|
'target': parameters['target_variable'],
|
|
},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=DOCUMENT_TIMEOUT,
|
|
)
|
|
|
|
succeeded = True
|
|
result = {
|
|
'import_run_id': import_run_id,
|
|
'run_id': run['run_id'],
|
|
'model_name': model_name,
|
|
'version': version['version'],
|
|
'document_id': document.get('id'),
|
|
}
|
|
except BaseException as error: # NOSONAR - classified once below, then re-raised untouched
|
|
located_step, located_gate = failure_location(error)
|
|
failed_step = located_step or current_step.value
|
|
gate = located_gate
|
|
raise
|
|
finally:
|
|
cleanup = await self._cleanup(base)
|
|
await self._record_terminal(
|
|
base,
|
|
succeeded=succeeded,
|
|
step=failed_step or current_step.value,
|
|
gate=gate,
|
|
source=source,
|
|
cleanup_failed=not cleanup.get('cleaned', True),
|
|
)
|
|
|
|
return result
|
|
|
|
# ------------------------------------------------------------------ validation
|
|
|
|
@staticmethod
|
|
def _require_import_run_id(value: Any) -> int:
|
|
"""The one check that runs before the claim, because the claim needs its result.
|
|
|
|
A missing or unusable id means there is no record to claim and none to write: the run ends
|
|
FAILED with this error type, and the failure is visible in the workflow's terminal state, its
|
|
error type, the error metric and a notification — everywhere except a row, which is the
|
|
point.
|
|
"""
|
|
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
|
raise ImportInputError('import_run_id must be a positive integer')
|
|
return value
|
|
|
|
@staticmethod
|
|
def _validate_input(input_data: dict[str, Any], claim: dict[str, Any]) -> None:
|
|
"""Validate the rest of the request, and check it against the record it just claimed.
|
|
|
|
This runs **after** the claim so its failure lands on the record rather than only in the
|
|
logs: by now the row is this workflow's, so a bad request is reported to the person who made
|
|
it. Everything here fails at step `received`.
|
|
"""
|
|
unknown = [name for name in REJECTED_INPUT_FIELDS if name in input_data]
|
|
if unknown:
|
|
raise ImportInputError(
|
|
f'the import request carries fields this workflow does not accept: '
|
|
f'{", ".join(sorted(unknown))}'
|
|
)
|
|
|
|
for required in ('object_key', 'expected_digest', 'password_envelope'):
|
|
if not input_data.get(required):
|
|
raise ImportInputError(f'the import request is missing {required}')
|
|
|
|
digest = str(input_data['expected_digest']).lower()
|
|
if len(digest) != DIGEST_LENGTH or not set(digest) <= _HEX_DIGITS:
|
|
raise ImportInputError('expected_digest must be a 64-character hexadecimal SHA-256')
|
|
|
|
object_key = input_data['object_key']
|
|
file_name = claim.get('file_name')
|
|
if file_name and file_name != object_key:
|
|
raise ImportInputError('the record names a different object than the request')
|
|
|
|
request_digest = claim.get('request_digest')
|
|
if request_digest and request_digest != digest:
|
|
raise ImportInputError('the record names a different digest than the request')
|
|
|
|
# --------------------------------------------------------------- status writes
|
|
|
|
async def _hint(
|
|
self,
|
|
base: dict[str, Any],
|
|
step: ImportStep,
|
|
source: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""Record the step about to be attempted. A hint: its failure never fails the import."""
|
|
try:
|
|
await workflow.execute_activity_method(
|
|
Activities.record_import_status,
|
|
{**base, 'step': step.value, 'source': source or None},
|
|
retry_policy=IMPORT_STATUS_HINT_RETRY_POLICY,
|
|
start_to_close_timeout=STATUS_WRITE_TIMEOUT,
|
|
)
|
|
except BaseException as error: # NOSONAR - a lost hint must not fail a working import
|
|
await self._report_status_write_failure(base, error)
|
|
|
|
async def _record_names(
|
|
self, base: dict[str, Any], source: dict[str, Any], run_name: str
|
|
) -> None:
|
|
"""Write the experiment and run names. Also a hint: the run is created either way."""
|
|
try:
|
|
await workflow.execute_activity_method(
|
|
Activities.record_import_names,
|
|
{
|
|
**base,
|
|
'experiment_name': source['experiment_name'],
|
|
'run_name': run_name,
|
|
'source': source,
|
|
'step': ImportStep.EXPERIMENT_CREATION.value,
|
|
},
|
|
retry_policy=IMPORT_STATUS_HINT_RETRY_POLICY,
|
|
start_to_close_timeout=STATUS_WRITE_TIMEOUT,
|
|
)
|
|
except BaseException as error: # NOSONAR - the names are for a human reading the list
|
|
await self._report_status_write_failure(base, error)
|
|
|
|
async def _record_terminal(
|
|
self,
|
|
base: dict[str, Any],
|
|
*,
|
|
succeeded: bool,
|
|
step: str | None,
|
|
gate: int | None,
|
|
source: dict[str, Any],
|
|
cleanup_failed: bool,
|
|
) -> None:
|
|
"""The single authoritative write, and the last thing this workflow does.
|
|
|
|
Its own failure is reported through the channel with the verdict it could not write, and it
|
|
never turns a completed import into a failed one — nor masks the cause of a real failure.
|
|
"""
|
|
try:
|
|
await workflow.execute_activity_method(
|
|
Activities.record_import_terminal_status,
|
|
{
|
|
**base,
|
|
'succeeded': succeeded,
|
|
'step': step,
|
|
'gate': gate,
|
|
'source': source or None,
|
|
'cleanup_failed': cleanup_failed,
|
|
},
|
|
retry_policy=IMPORT_STATUS_RETRY_POLICY,
|
|
start_to_close_timeout=STATUS_WRITE_TIMEOUT,
|
|
schedule_to_close_timeout=STATUS_WRITE_ENVELOPE,
|
|
)
|
|
except BaseException as error: # NOSONAR - the outcome stands even when it cannot be written
|
|
await self._report_status_write_failure(
|
|
base,
|
|
error,
|
|
verdict=f'succeeded={succeeded} step={step} gate={gate}',
|
|
)
|
|
|
|
async def _report_status_write_failure(
|
|
self,
|
|
base: dict[str, Any],
|
|
error: BaseException,
|
|
verdict: str | None = None,
|
|
) -> None:
|
|
"""Route a status write that could not be made through the one failure channel."""
|
|
try:
|
|
await workflow.execute_activity_method(
|
|
Activities.report_import_status_write_failure,
|
|
{**base, 'detail': str(error), 'verdict': verdict},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=STATUS_WRITE_TIMEOUT,
|
|
)
|
|
except BaseException: # NOSONAR - reporting the report is where this stops
|
|
pass
|
|
|
|
# ----------------------------------------------------------- side-effect steps
|
|
|
|
@staticmethod
|
|
async def _ensure_retention(base: dict[str, Any], input_data: dict[str, Any]) -> None:
|
|
"""Ensure the bundle bucket's expiry rule, without making it a reason to fail an import.
|
|
|
|
The rule is applied by the importer rather than assumed from a Helm value, so the acceptance
|
|
criterion is verifiable from the code that depends on it. But a bucket policy the importer
|
|
cannot set is an operational problem, not a bad import: failing a valid model import because
|
|
a lifecycle API refused would be the wrong trade, so this is reported and dropped like a
|
|
lost status hint.
|
|
"""
|
|
try:
|
|
await workflow.execute_activity_method(
|
|
Activities.ensure_import_bundle_retention,
|
|
{**base, 'bucket': input_data.get('bucket')},
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=RETENTION_TIMEOUT,
|
|
)
|
|
except BaseException: # NOSONAR - retention is not a precondition of a correct import
|
|
pass
|
|
|
|
@staticmethod
|
|
async def _cleanup(base: dict[str, Any]) -> dict[str, Any]:
|
|
"""Remove this import's temporary files. Runs on every path, before the terminal write."""
|
|
try:
|
|
return await workflow.execute_activity_method(
|
|
Activities.cleanup_import_files,
|
|
base,
|
|
retry_policy=retry_policy,
|
|
start_to_close_timeout=CLEANUP_TIMEOUT,
|
|
)
|
|
except BaseException: # NOSONAR - cleanup never replaces the failure that brought us here
|
|
return {'cleaned': False}
|