Retry a failed step without repeating the experiment

Recover a transient record-write failure, then reuse a completed result only when its inputs and artifacts still match.

Problem

When you need this. An orchestration retry can accidentally repeat expensive fitting or overwrite earlier outputs. Here the model has already been evaluated when the final completion record write fails. Repeating the whole flow would create another training run even though the candidate is intact.

The exercise fails that one write deliberately, lets Prefect retry the task once and invokes the same pipeline again. The second invocation should return the original run after checking its saved cohort and model.

Concepts

A retry repeats a failed operation. Idempotency means that repeating the logical request has the intended effect without creating another logical result. The pipeline identity hashes the selected application contract, tracking wrapper, pipeline implementation and dependency pins.

The record-writing task permits one retry for an OSError, excluding a file-already-exists conflict. Schema and contract errors are permanent for these inputs and are not retried. The deliberate fault occurs before the completion record is created, so the retry can create it exclusively.

A completed record is useful only while its files still match. The rerun checks the cohort digest, model digest and fresh validation result. An exclusive lock refuses concurrent invocations. An incomplete workspace remains preserved; recovery uses a new workspace rather than guessing whether an interrupted write completed.

Architecture

Working code

Download the shared application and tests, unzip it and enter mlops-return-workflow/. Keep the same directory throughout the course. Create the environment once. Install the pinned platform dependencies for this workshop.

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-platform.txt

On PowerShell, activate with .venv\Scripts\Activate.ps1.

Continue from Connect preparation, training and evaluation with Prefect; keep the same lab directory and outputs.

Restrict retries to the failed operation

The controlled failure is a completion record write after training. The task’s retry policy excludes input errors and conflicting existing completion records.

def retry_transient(task, task_run, state):
    try:state.result()
    except FileExistsError:return False
    except OSError:return True
    except Exception:return False
    return False

The retry count is bounded to one. The attempt counter shows two write attempts, while the tracker contains one training execution.

Check an identical rerun

The second call computes the same identity, checks the completed files and returns the existing run ID.

def pipeline_identity(source):
    return f.digest({'contract':f.contract(source),'trainer':w.sha(w.HERE/'tracking.py'),
                     'pipeline':w.sha(Path(__file__)),'dependencies':w.sha(w.HERE/'requirements-platform.txt')})

def run(source, workspace, inject_once=False):
    source=Path(source).resolve();workspace=Path(workspace).resolve();workspace.mkdir(parents=True,exist_ok=True)
    identity=pipeline_identity(source);folder=workspace/identity
    folder.mkdir(exist_ok=True)
    # A single-process lock makes two simultaneous invocations fail closed.
    lock=folder/'running.lock'
    try:handle=lock.open('x')
    except FileExistsError:raise ValueError('pipeline already running; inspect owner before recovery')
    with handle:
        try:
            receipt=folder/'complete.json'
            if receipt.exists():
                result=w.strict_json(receipt)
                model=folder/'tracker/runs'/result['run_id']/'workspace/outputs/candidate.json'
                cohort=folder/'cohort.json'
                snapshot=folder/'tracker/runs'/result['run_id']/'workspace'
                manifest=w.strict_json(snapshot/'contract.json')
                if w.sha(snapshot/'scenario.json')!=manifest['dataset_sha256']:
                    raise ValueError('saved source dataset changed')
                for name,digest in manifest['implementation'].items():
                    if w.sha(snapshot/name)!=digest:raise ValueError('saved implementation changed')
                checked=w.evaluate(cohort,model)
                if (result['pipeline_id']!=identity or w.sha(model)!=result['model_sha256']
                    or w.sha(cohort)!=result['prepared_cohort_sha256']
                    or checked['accuracy']!=result['validation_accuracy']):
                    raise ValueError('completed result identity or artifact changed')
                return dict(result,reused=True)
            # A failed attempt is kept; recovery uses an explicit fresh workspace.
            if any(p.name!='running.lock' for p in folder.iterdir()):
                raise ValueError('incomplete attempt retained; choose a fresh workspace')
            result=train_flow(source,folder,inject_once)
            return dict(result,reused=False,workspace=str(folder))
        finally:lock.unlink()

python course.py retries --workspace work/course

{
  "receipt_attempts": 2,
  "same_run_reused": true,
  "training_executions_in_attempt": 1,
  "run_id": "a758683696ac4e79a050118607d23426",
  "failure": "controlled transient receipt-write failure; original attempt logged"
}

The retained result reports two write attempts and one training execution. The identical call reuses that run. Changing inputs produces a different identity; corrupting the saved cohort causes refusal.

Recorded commands retain outputs in recorded-platform/; your commands use work/. Read the actual command records. Use a fresh work directory for another attempt; original outputs are preserved.

Failure cases

test_completed_retry_does_not_repeat_training checks the actual task retry and MLflow run count. test_completed_cohort_tampering_is_refused checks the reuse boundary.

test_incomplete_attempt_is_retained leaves an interrupted marker. The pipeline refuses to mix another attempt into that directory. A stale lock requires an operator to establish that its process has ended before recovery.

python -m unittest -v test_platform.py

Running and scaling the workshop

This lock and exclusive-file strategy applies to one local filesystem. It is not a distributed lease and has no automatic stale-lock timeout.

A remote object store needs conditional writes or a transactional state service. Retrying a non-idempotent remote action also requires an action-specific idempotency key. None is implied by this record-writing example.

Key takeaways

• Retry the operation that failed.
• Verify saved artifacts before returning a completed result.
• Preserving an incomplete attempt is safer than silently rewriting it.

Best practices

Classify transient dependency failures separately from invalid inputs.

Record attempts and keep retry budgets finite.

References

• Prefect: retries