A failed ETL job is easy to spot. The harder question is: where can we safely resume without duplicating data or silently skipping work?

Consider a two-stage retail-media pipeline. Stage 1 normalizes raw campaign exports into Silver; stage 2 aggregates Silver into Gold. Silver finishes, Gold fails, and the scheduler wants to retry. Replaying both stages might overwrite valid data or duplicate append-only outputs. Skipping Silver without checking its artifact might use a partial or stale snapshot.

This article walks through a read-only, fail-closed restart planner added to my public retail-media-analytics-pipeline portfolio repository on November 7, 2026. The repository uses simulated advertising data; the recovery planner is not a production deployment and does not automatically restart Spark.

The architecture: evidence before action

The basic flow is:

Bronze inputs | v Silver normalization -- stage events --> JSONL event log | v Gold aggregation -- stage events --> JSONL event log | v Monitor validates history | v Read-only recovery planner | v Human verifies artifacts | v Manual rerun, if safe

The important separation is observation versus execution. An event log tells us what the application reported. It cannot, by itself, prove that the output files are complete, that the input snapshot has not changed, or that another worker is still writing.

The repository's monitor summarizes JSONL stage events. Its recovery module consumes that validated summary and returns a proposed restart boundary or refuses to plan.

A small state model

Suppose the monitor produces this illustrative summary:

summary = { "expected_stages": 2, "runs": [{ "run_id": "demo-001", "stages": [ {"stage": 1, "status": "succeeded"}, {"stage": 2, "status": "failed"}, ], }], }

The planner considers only four useful terminal or intermediate states: succeeded, failed, not_started, plus running or stalled states that block recovery. An unrecognized state is not interpreted optimistically.

The critical invariant is that only a contiguous prefix of successful stages can be skipped. If stage 2 failed, stage 1 may be skipped only after the operator explicitly verifies stage 1's output. If the run is still active, the planner refuses. If downstream activity appears after the proposed restart boundary, the history is ambiguous and the planner refuses.

Implement the decision boundary

Here is a shortened, illustrative version of the core decision logic. It assumes that a separate monitor has already validated event syntax and ordering; the full repository implementation also checks run-ID uniqueness, stage inventory, and input types.

class RecoveryNotSafe(ValueError): pass def choose_restart(statuses, verified_upstream): if not statuses: raise RecoveryNotSafe("No stages supplied") allowed = {"succeeded", "failed", "not_started"} if any(s not in allowed for s in statuses): raise RecoveryNotSafe("Active or unknown stage") if all(s == "succeeded" for s in statuses): return {"action": "none", "restart_at_stage": None} first = next( i for i, status in enumerate(statuses, start=1) if status != "succeeded" ) # Everything after the failed/not-started boundary  # must have no activity.  if any(s != "not_started" for s in statuses[first:]): raise RecoveryNotSafe("Unexpected downstream activity") required = list(range(1, first)) if sorted(verified_upstream) != required: raise RecoveryNotSafe("Upstream artifacts not verified") return { "action": "manual_restart", "restart_at_stage": first, "skipped_stages": required, }

Notice the indexing: first is a 1-based stage number, while statuses[first:] correctly selects stages after that stage in Python's zero-based list. This is a small detail that is easy to get wrong in orchestration code.

In the real module, the verified_stage_outputs argument must be a list of distinct integer stage numbers matching the skipped prefix exactly. A boolean is not accepted as an integer stage number. That strictness prevents ambiguous operator input from turning into an unsafe decision.

Calling the real implementation

The public module exposes plan_recovery:

from src.pipeline_recovery import plan_recovery plan = plan_recovery( summary, "demo-001", verified_stage_outputs=[1], ) print(plan["restart_at_stage"]) # 2

The command-line entry point reads a real event file through the monitor:

python -m src.pipeline_recovery \ --events data/outputs/pipeline_events.jsonl \ --run-id YOUR_RUN_ID \ --expected-stages 2 \ --verified-stage 1

The --verified-stage 1 flag is an operator assertion, not an automatic file-integrity check. Before using it, verify the Silver dataset's source snapshot, schema, partition coverage, row counts or reconciliation totals, and completeness. If any check is inconclusive, do not assert verification.

Test the refusal paths, not just the happy path

A recovery planner should be judged as much by what it refuses as by what it permits. These small examples exercise the illustrative function above:

import pytest def test_verified_silver_can_be_skipped(): plan = choose_restart( ["succeeded", "failed"], [1] ) assert plan["restart_at_stage"] == 2 def test_missing_verification_is_rejected(): with pytest.raises(RecoveryNotSafe): choose_restart(["succeeded", "failed"], []) def test_active_run_is_rejected(): with pytest.raises(RecoveryNotSafe): choose_restart(["running", "not_started"], []) def test_completed_run_needs_no_restart(): assert choose_restart( ["succeeded", "succeeded"], [] )["action"] == "none"

These are illustrative pytest tests, not a claim that this exact snippet is committed to the repository. The actual project contains standard-library unittest coverage for the real implementation, including failed-first-stage, verified-upstream, and rejection scenarios.

A useful additional test matrix covers duplicate run IDs, incomplete stage inventories, out-of-order events, duplicate verification values, unexpected downstream success, and unknown statuses. For safety-critical recovery, malformed evidence should result in an explicit error rather than a guessed restart point.

Why a plan is not a safe rerun

Even a perfect stage-boundary decision cannot guarantee idempotency. Consider a Gold write that appended half its output before failing. Restarting Gold could double-count the first half. Likewise, a Silver stage marked successful yesterday may not match today's changed input snapshot.

Before executing a retry, confirm:

  • Process ownership: the previous worker has stopped; no lock or lease is still held.
  • Input consistency: the retry reads the intended immutable snapshot or checkpoint.
  • Output semantics: append, overwrite, merge, and partition replacement behave as expected.
  • Downstream consumers: partial results have not already been published or consumed.
  • Observability: the retry gets a distinct attempt identifier and auditable events.

For distributed pipelines, use a durable orchestrator and transactional table format where appropriate. A local JSONL planner is useful for demonstrating decision logic, but it is not a replacement for leases, checkpoints, transactional commits, or a workflow engine.

Where this fits in a medallion pipeline

In Bronze/Silver/Gold architectures, recovery boundaries often align with durable, validated datasets. But a layer name alone does not establish a checkpoint. A stage is skippable only if its output is complete and belongs to the exact logical run being recovered.

For retail-media analytics, that matters because delayed conversions, attribution changes, and backfilled campaign exports can change aggregates. A previously successful Silver dataset might be technically readable yet semantically wrong for a new Gold calculation. The planner intentionally leaves that business validation to the operator.

Takeaways

Fail closed when the evidence is ambiguous. Validate event history before making a restart decision. Require explicit verification before skipping upstream work. Separate planning from execution, and treat idempotency as a storage-and-workflow property—not something an event log can prove.

The November 7 implementation commit includes the planner, documentation, and synthetic unit tests. It is a practical starting point for thinking about safer retries, not a claim of production-grade recovery.

Further reading: Apache Spark Structured Streaming fault tolerance, Delta Lake ACID transactions, and Apache Airflow retries and task instances.