Pipelines I Trust Blindly: Idempotency, Retries, and Defensive Engineering in Practice

A pipeline I trust blindly is not a pipeline I believe in. It is a pipeline with documented properties that make trust rational: running it twice produces the same result as running it once, failures that are transient are retried and failures that are structural are surfaced immediately, and the state of the pipeline at any moment is visible to anyone who needs to know. Trust built on properties is durable. Trust built on confidence degrades every time the pipeline fails at 2 a.m.

Idempotency: making re-runs safe

An idempotent pipeline produces the same state in the destination regardless of how many times it runs for the same input. This property is not a nicety: it is what makes recovery from failure safe. When a pipeline fails halfway through, the correct response is to re-run it. If the pipeline is not idempotent, a re-run may duplicate data, overwrite partial state, or leave the destination in an inconsistent condition that is worse than the original failure.

-- Non-idempotent: inserts accumulate on re-run
INSERT INTO fct_orders (order_id, revenue, loaded_at)
SELECT order_id, revenue, CURRENT_TIMESTAMP
FROM stg_orders
WHERE order_date = '2024-08-01';

-- Idempotent: upsert on natural key
MERGE INTO fct_orders AS target
USING (
    SELECT order_id, revenue, CURRENT_TIMESTAMP AS loaded_at
    FROM stg_orders
    WHERE order_date = '2024-08-01'
) AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN
    UPDATE SET revenue = source.revenue, loaded_at = source.loaded_at
WHEN NOT MATCHED THEN
    INSERT (order_id, revenue, loaded_at)
    VALUES (source.order_id, source.revenue, source.loaded_at);

The upsert pattern works when a natural key exists. When it does not, the idempotent approach is delete-then-insert on the partition the pipeline is responsible for: delete all rows for the target date, then insert the fresh result. This is safe because the pipeline owns that partition exclusively.

# Delete-then-insert pattern for date-partitioned loads
def load_partition(conn, order_date: str, rows: list[dict]) -> None:
    with conn.begin():
        conn.execute(
            "DELETE FROM fct_orders WHERE order_date = :d",
            {"d": order_date}
        )
        if rows:
            conn.execute(
                "INSERT INTO fct_orders (order_id, order_date, revenue) VALUES (:order_id, :order_date, :revenue)",
                rows
            )
        # Both operations in the same transaction:
        # if the insert fails, the delete rolls back too.
        # Re-running starts from a clean partition every time.

Retries: distinguishing transient from structural

Not all failures should be retried. A pipeline that retries a structural failure (a schema change, a missing permission, a logic error in a query) is a pipeline that hides the failure for the duration of the retry budget and then fails with a worse signal than if it had stopped immediately. Retries are for transient failures: network timeouts, temporary service unavailability, rate limit responses. They are not for errors that will reproduce identically on re-attempt.

import time
import requests
from requests.exceptions import ConnectionError, Timeout

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
NON_RETRYABLE_STATUS_CODES = {400, 401, 403, 404, 422}

def fetch_with_retry(url: str, max_attempts: int = 4) -> dict:
    for attempt in range(1, max_attempts + 1):
        try:
            response = requests.get(url, timeout=30)

            if response.status_code in NON_RETRYABLE_STATUS_CODES:
                # Structural failure: stop immediately, do not retry
                raise ValueError(
                    f"Non-retryable HTTP {response.status_code} from {url}. "
                    "Fix the request before re-running."
                )

            if response.status_code in RETRYABLE_STATUS_CODES:
                if attempt == max_attempts:
                    raise RuntimeError(f"Exhausted {max_attempts} retries: HTTP {response.status_code}")
                backoff = 2 ** attempt
                print(f"Attempt {attempt} failed ({response.status_code}), retrying in {backoff}s")
                time.sleep(backoff)
                continue

            response.raise_for_status()
            return response.json()

        except (ConnectionError, Timeout) as e:
            if attempt == max_attempts:
                raise
            time.sleep(2 ** attempt)

    raise RuntimeError("Unreachable")

The exponential backoff doubles the wait on each attempt: 2 seconds, 4, 8, 16. For high-traffic APIs, adding jitter (a random offset) prevents multiple pipeline instances from retrying in lockstep and amplifying the load on the upstream service.

import random

def backoff_with_jitter(attempt: int, base: float = 2.0, cap: float = 60.0) -> float:
    sleep = min(cap, base ** attempt)
    jitter = random.uniform(0, sleep * 0.25)
    return sleep + jitter

Alerting: failing before the stakeholder does

An alert should fire when a condition is true that, if left unaddressed, will cause a stakeholder to notice a problem. This means the alert needs to fire before the stakeholder's next interaction with the data, with enough lead time to investigate and resolve. An alert that fires after the stakeholder has already seen stale numbers on a dashboard is a post-mortem notification, not an operational alert.

# Freshness alert: pipeline SLA check in Python
from datetime import datetime, timedelta
import smtplib

def check_pipeline_sla(conn, table: str, sla_hours: int) -> None:
    result = conn.execute(
        f"SELECT MAX(loaded_at) AS latest FROM {table}"
    ).fetchone()

    latest = result["latest"]
    hours_stale = (datetime.utcnow() - latest).total_seconds() / 3600

    if hours_stale > sla_hours:
        send_alert(
            subject=f"SLA breach: {table} is {hours_stale:.1f}h stale (SLA: {sla_hours}h)",
            body=(
                f"Table {table} was last loaded at {latest} UTC.\n"
                f"Current staleness: {hours_stale:.1f} hours.\n"
                f"SLA threshold: {sla_hours} hours.\n\n"
                f"Runbook: https://wiki.internal/pipelines/{table}/runbook"
            )
        )

Every alert should reference a runbook. An alert without a runbook trains people to respond by guessing, which produces inconsistent resolution and institutional knowledge that lives in the head of whoever was on-call the last time this happened. The runbook is short: what the alert means, the three most common causes, and the steps to resolve each one.

Defensive engineering as a discipline

Defensive engineering is the practice of writing pipelines that fail loudly, fail fast, and fail in ways that contain the blast radius. A pipeline that silently produces wrong output is the worst failure mode: it provides no signal that anything is wrong, and the cost of the error accumulates until a stakeholder or an audit discovers it.

def validate_output(df, expected_min_rows: int, expected_columns: list[str]) -> None:
    actual_columns = set(df.columns)
    missing = set(expected_columns) - actual_columns
    if missing:
        raise ValueError(f"Output missing expected columns: {missing}")

    if len(df) < expected_min_rows:
        raise ValueError(
            f"Output has {len(df)} rows, expected at least {expected_min_rows}. "
            "Refusing to load: result may be incomplete."
        )

    null_rates = df[expected_columns].isnull().mean()
    suspicious = null_rates[null_rates > 0.10]
    if not suspicious.empty:
        raise ValueError(
            f"Columns with >10% nulls in output: {suspicious.to_dict()}. "
            "Investigate before loading."
        )

Assertions at the output boundary of every pipeline stage are what produce the loud, fast failures that make recovery tractable. A pipeline that validates its output before writing it to the destination converts silent corruption into a visible error with a clear location. The error is easier to diagnose, cheaper to fix, and does not require a downstream stakeholder to discover it.