Resilience Engineering

29.1 A Taxonomy of Failures (18 min)

Resilient code starts with naming what can go wrong, and deciding in advance what to do about each case.

Two principles follow:

1. Retry only what time can fix (rows 1–3).
2. Isolate everything else: one bad series should cost you that series, not the whole run.

29.2 Retries and Timeouts, Configured Centrally (18 min)

dlt's requests helper, and therefore RESTClient, retries automatically. Its behaviour is controlled by configuration, not code. Add to .dlt/config.toml:

[runtime]
log_level = "WARNING"
request_timeout = 30              # seconds before a request is abandoned
request_max_attempts = 5          # total attempts per request (1 = no retries)
request_backoff_factor = 1        # exponential backoff: ~1s, 2s, 4s, ...
request_max_retry_delay = 60      # never wait longer than this between attempts

These map to environment variables like RUNTIME__REQUEST_MAX_ATTEMPTS, so tests and schedulers can override them (Day 23).

Watching retries happen

import time
from dlt.sources.helpers import requests

t0 = time.perf_counter()
try:
    requests.get("https://httpbin.org/status/503", timeout=10)     # always answers 503
except Exception as e:
    print(type(e).__name__)
print(f"gave up after {time.perf_counter() - t0:.1f}s")

Run it with different settings and compare the elapsed time:

RUNTIME__REQUEST_MAX_ATTEMPTS=1 python day29/retry_demo.py
RUNTIME__REQUEST_MAX_ATTEMPTS=4 RUNTIME__REQUEST_BACKOFF_FACTOR=1 python day29/retry_demo.py

(httpbin.org is a public testing service. If it's slow or down, the timings are still instructive.)

Retry budgets. Retries multiply: 50 series × 5 attempts × 30 s timeout could take over two hours during an outage. That's why 29.4 adds a circuit breaker.

29.3 Throttling and Batching (20 min)

Retries react to overload; throttling prevents it. Public statistical APIs are shared infrastructure, and being polite keeps you from being blocked.

A minimal throttle

Create macro_ingest/resilience.py (we'll add to it all day):

"""Resilience helpers shared by all macro_ingest sources."""
import time


class Throttle:
    """Ensure at least `min_interval` seconds between successive calls to wait()."""

    def __init__(self, min_interval=0.3):
        self.min_interval = min_interval
        self._last = 0.0

    def wait(self):
        elapsed = time.monotonic() - self._last
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self._last = time.monotonic()

time.monotonic() is a clock that never jumps backwards (unlike wall-clock time), which makes it the right tool for measuring intervals.

Batching: fewer, larger requests

StatCan's vector endpoints accept many vectors per POST. One request for 40 vectors is far kinder, and more reliable, than 40 requests. But don't make batches enormous: if one huge request fails, everything in it fails.

def chunked(items, size):
    """Split a list into consecutive chunks of at most `size` items."""
    for i in range(0, len(items), size):
        yield items[i:i + size]

print(list(chunked([1, 2, 3, 4, 5], 2)))     # [[1, 2], [3, 4], [5]]

Check each provider's documentation for published request limits and service windows. StatCan's WDS user guide, for example, describes its usage limits and when data is refreshed. Encode the limits you find in configuration (min_interval, batch_size) rather than in code.

29.4 Isolation: Error Logs and Circuit Breakers (28 min)

The error log: failures become data

Instead of letting an exception end the run, catch it per series, record it, and move on. Add to resilience.py:

from datetime import datetime, timezone


class PayloadError(ValueError):
    """The API answered, but the content is unusable (failure types 5–6)."""


class ErrorLog:
    """Collects per-series failures during a run; loaded afterwards as a table."""

    def __init__(self):
        self.records = []

    def add(self, source, series_id, error, stage="fetch"):
        self.records.append({
            "source": source,
            "series_id": series_id,
            "stage": stage,
            "error_type": type(error).__name__,
            "message": str(error)[:500],
            "occurred_at": datetime.now(timezone.utc),
        })

    def __len__(self):
        return len(self.records)

The runner (29.7) loads ErrorLog.records into a raw_dlt.extraction_errors table after each source runs, whether the run succeeded or not. Now "what failed last Tuesday?" is a SQL query.

The circuit breaker: stop hammering a dead API

If the first several series all fail, the API is probably down. Retrying the remaining 45 series wastes time and load. A circuit breaker trips after N consecutive failures:

class CircuitOpen(RuntimeError):
    """Raised when too many consecutive failures suggest a sustained outage."""


class CircuitBreaker:
    def __init__(self, max_consecutive_failures=3):
        self.max = max_consecutive_failures
        self.failures = 0

    def success(self):
        self.failures = 0

    def failure(self):
        self.failures += 1
        if self.failures >= self.max:
            raise CircuitOpen(f"{self.failures} consecutive failures — stopping this source")

Putting it together in a resource

@dlt.resource(name="boc_observations", write_disposition="merge",
              primary_key=("series_id", "date"))
def observations():
    cursors = dlt.current.resource_state().setdefault("last_date", {})
    throttle, breaker = Throttle(min_interval), CircuitBreaker(max_failures)

    for sid in series_ids:
        start = ...                                     # per-series cursor (Day 27)
        throttle.wait()
        try:
            rows = fetch_series(client, sid, start)     # may raise HTTP/payload errors
        except CircuitOpen:
            raise
        except Exception as e:
            error_log.add("boc", sid, e)
            breaker.failure()                           # may raise CircuitOpen
            continue                                    # cursor NOT advanced
        breaker.success()
        if rows:
            cursors[sid] = max(r["date"] for r in rows).isoformat()
        yield rows

Why these choices?

• A failed series doesn't advance its cursor, so next run it will be requested from the same point and nothing is skipped.
• CircuitOpen is re-raised so the whole source fails (and its state isn't committed), and the runner reports an outage.
• One bad code (404) costs one error row, and the other series load normally.

Try it 29.4

Without writing an API call, test the circuit breaker's logic in the Python REPL: record failure, failure, success, failure, failure, failure with max_consecutive_failures=3. On which call does it trip?

from macro_ingest.resilience import CircuitBreaker, CircuitOpen
b = CircuitBreaker(3)
for i, ok in enumerate([False, False, True, False, False, False], start=1):
    try:
        b.success() if ok else b.failure()
    except CircuitOpen as e:
        print(f"tripped on call {i}: {e}")

29.5 Validating Payloads (15 min)

HTTP 200 only means "the server answered". Before trusting the content, check it. Put the fetch-and-validate logic in plain functions (no dlt), so they're easy to test in 29.6.

Create macro_ingest/parsers.py:

"""Fetch-and-validate functions: plain Python, easy to test."""
from datetime import date, timedelta

from .resilience import PayloadError


def fetch_valet_series(client, series_id, start):
    """Return long rows for one Valet series, or raise on any problem."""
    resp = client.get(f"observations/{series_id}/json",
                      params={"start_date": start.isoformat()})
    resp.raise_for_status()
    payload = resp.json()

    if "observations" not in payload:
        raise PayloadError(f"{series_id}: no 'observations' in response")
    if series_id not in payload.get("seriesDetail", {}):
        raise PayloadError(f"{series_id}: series missing from 'seriesDetail'")

    tomorrow = date.today() + timedelta(days=1)
    rows = []
    for obs in payload["observations"]:
        cell = obs.get(series_id)
        try:
            d = date.fromisoformat(obs["d"])
            value = float(cell.get("v"))
        except (KeyError, AttributeError, TypeError, ValueError):
            continue                                      # unusable value: skip quietly
        if d > tomorrow:
            raise PayloadError(f"{series_id}: observation dated in the future ({d})")
        rows.append({"date": d, "series_id": series_id, "value": value, "source": "boc"})
    return rows


def parse_wds_items(payload, error_log):
    """Split a WDS vector response into long rows; log FAILED items."""
    if not isinstance(payload, list):
        raise PayloadError("WDS response is not a list")
    rows = []
    for item in payload:
        obj = item.get("object")
        if item.get("status") != "SUCCESS" or not isinstance(obj, dict):
            error_log.add("statcan", str(obj)[:60], PayloadError(f"item status {item.get('status')}"),
                          stage="item")
            continue
        sid = f"v{obj.get('vectorId')}"
        for dp in obj.get("vectorDataPoint", []):
            if dp.get("value") is None:
                continue
            rows.append({"date": date.fromisoformat(dp["refPer"][:10]), "series_id": sid,
                         "value": float(dp["value"]), "source": "statcan"})
    return rows

What gets skipped quietly (individual unusable values, as since Day 2) versus what fails the series (structural problems that suggest the API changed or broke) is a deliberate, documented choice.

29.6 Testing Failures Offline with requests-mock (25 min)

You can't wait for Valet to go down to test your outage handling. Mocking replaces real HTTP responses with scripted ones, so tests run offline, instantly and deterministically.

pip install pytest requests-mock
pip freeze > requirements.txt

Create tests/test_resilience.py:

import os
from datetime import date

import pytest
import requests_mock as rm

os.environ["RUNTIME__REQUEST_MAX_ATTEMPTS"] = "2"       # keep tests fast
os.environ["RUNTIME__REQUEST_BACKOFF_FACTOR"] = "0"

from macro_ingest.clients import valet_client            # built in 30.2
from macro_ingest.parsers import fetch_valet_series, parse_wds_items
from macro_ingest.resilience import CircuitBreaker, CircuitOpen, ErrorLog, PayloadError

BASE = "https://www.bankofcanada.ca/valet/"
GOOD = {"seriesDetail": {"V39079": {"label": "Overnight"}},
        "observations": [{"d": "2026-07-01", "V39079": {"v": "2.25"}},
                         {"d": "2026-07-02", "V39079": {"v": "Bank holiday"}}]}


def test_valid_payload_parses_and_skips_bad_values():
    with rm.Mocker() as m:
        m.get(f"{BASE}observations/V39079/json", json=GOOD)
        rows = fetch_valet_series(valet_client(), "V39079", date(2026, 7, 1))
    assert len(rows) == 1 and rows[0]["value"] == 2.25


def test_unknown_series_raises_http_error():
    with rm.Mocker() as m:
        m.get(f"{BASE}observations/NOPE/json", status_code=404)
        with pytest.raises(Exception):
            fetch_valet_series(valet_client(), "NOPE", date(2026, 7, 1))


def test_transient_503_then_success_recovers():
    with rm.Mocker() as m:
        m.get(f"{BASE}observations/V39079/json",
              [{"status_code": 503}, {"json": GOOD}])      # first call fails, second succeeds
        rows = fetch_valet_series(valet_client(), "V39079", date(2026, 7, 1))
    assert len(rows) == 1


def test_missing_observations_is_a_payload_error():
    with rm.Mocker() as m:
        m.get(f"{BASE}observations/V39079/json", json={"seriesDetail": {}})
        with pytest.raises(PayloadError):
            fetch_valet_series(valet_client(), "V39079", date(2026, 7, 1))


def test_wds_failed_items_are_logged_not_fatal():
    log = ErrorLog()
    payload = [
        {"status": "SUCCESS", "object": {"vectorId": 41690973, "vectorDataPoint": [
            {"refPer": "2026-07-01", "value": 169.0}]}},
        {"status": "FAILED", "object": "Vector not found"},
    ]
    rows = parse_wds_items(payload, log)
    assert len(rows) == 1 and len(log) == 1


def test_circuit_breaker_trips_after_consecutive_failures():
    b = CircuitBreaker(2)
    b.failure()
    with pytest.raises(CircuitOpen):
        b.failure()

python -m pytest tests/test_resilience.py -v

Each test encodes one row of the 29.1 failure table. When you change the extractor later, these tests tell you immediately whether its resilience still holds.

If a mocked test unexpectedly reaches the real internet or hangs, check that the URL in m.get(...) exactly matches what the client requests (base URL + path, including the trailing slash).

29.7 Run Summaries and Alerts (13 min)

A resilient pipeline reports. After every run, ingest.py (Day 30) will:

1. load the ErrorLog into raw_dlt.extraction_errors;
2. append one row per source to raw_dlt.ingest_runs (status, duration, error count);
3. send an alert if anything failed.

A minimal alert helper, sending to any "incoming webhook" (Slack, Microsoft Teams and Discord all offer these) when one is configured, and printing otherwise. Add to resilience.py:

import os

import requests as plain_requests      # a single, non-retried attempt is fine for alerts


def notify(message):
    """Send an alert to the webhook in NOTIFICATIONS__WEBHOOK_URL, else print it."""
    url = os.environ.get("NOTIFICATIONS__WEBHOOK_URL")
    if not url:
        print(f"[ALERT] {message}")
        return
    try:
        plain_requests.post(url, json={"text": message}, timeout=10)
    except Exception as e:                # an alert failure must never crash the pipeline
        print(f"[ALERT not delivered: {e}] {message}")

Store the webhook URL as a Codespaces secret named NOTIFICATIONS__WEBHOOK_URL. It's a secret: anyone with it can post to your channel.

What deserves an alert?

Alert fatigue is real. If the channel pings for every trivial event, people stop reading it. Alert on things that need a human, and log everything else.

Day 29 checklist

• I can classify failures and decide which to retry
• Retry and timeout behaviour is set in config.toml, overridable by env vars
• I throttle and batch requests politely
• One failing series is recorded and skipped, never fatal, and its cursor doesn't advance
• A circuit breaker stops a source during sustained outages
• Payloads are validated beyond HTTP status, including StatCan's per-item status
• I test failure handling offline with mocks
• Runs produce summaries, and real problems produce alerts

git add .
git commit -m "Day 29: resilience helpers, payload validation, mocked failure tests, alerts"
git push