Error Recovery in Long-Running Scripts

By the end of this lesson you can:

• Checkpoint progress so a restarted job resumes instead of starting over.
• Isolate per-item errors so one bad record doesn't kill the batch.
• Retry transient failures with backoff (a reusable decorator).
• Write idempotent steps and shut down gracefully on Ctrl-C / signals.

You're processing 100,000 records. At record 73,000 a network blip raises an exception. A naive script dies — and a re-run starts at record 1, redoing 3 hours of work (and maybe double-charging, double-emailing, double-uploading).

1. Checkpointing — resume, don't restart

import json
from pathlib import Path

CHECKPOINT = Path("progress.json")

def load_done() -> set:
    if CHECKPOINT.exists():
        return set(json.loads(CHECKPOINT.read_text()))
    return set()

def mark_done(done: set, item_id) -> None:
    done.add(item_id)
    CHECKPOINT.write_text(json.dumps(list(done)))   # persist after each item

done = load_done()
for item in all_items:
    if item["id"] in done:
        continue                 # already processed — skip on resume
    process(item)
    mark_done(done, item["id"])  # record progress immediately

Persist progress as you go. On restart, skip what's already marked done. The job resumes at record 73,001, not record 1.

2. Per-item isolation — keep going

failures = []
for item in items:
    try:
        process(item)
        mark_done(done, item["id"])
    except Exception as e:
        log.exception("item %s failed", item["id"])
        failures.append((item["id"], str(e)))   # collect, don't crash

log.info("done: %d ok, %d failed", len(items) - len(failures), len(failures))

Wrap each item so one failure is logged and recorded, not fatal. At the end you have a list of failures to retry later — far better than a half-finished crash.

3. Retry with backoff — a reusable decorator

import time, functools, random

def retry(attempts=3, base=1.0, exceptions=(Exception,)):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for n in range(1, attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    if n == attempts:
                        raise
                    wait = base * 2 ** (n - 1) + random.uniform(0, 0.5)
                    log.warning("attempt %d failed (%s); retry in %.1fs", n, e, wait)
                    time.sleep(wait)
        return wrapper
    return deco

@retry(attempts=4, exceptions=(ConnectionError, TimeoutError))
def fetch(item):
    ...

This generalises Lesson 24's retry into a decorator you can slap on any flaky operation. Retry only the errors that might be transient — don't retry a ValueError that'll fail identically every time.

4. Idempotency — safe to re-run

# NOT idempotent: re-running double-charges
charge_customer(amount)

# idempotent: a key makes a repeat a no-op
def charge_once(customer, amount, idempotency_key):
    if already_charged(idempotency_key):     # check first
        return                               # safe re-run
    charge_customer(amount)
    record_charge(idempotency_key)

Graceful shutdown

import signal

stop = False
def handle_signal(signum, frame):
    global stop
    stop = True
    log.info("shutdown requested — finishing current item then stopping")

signal.signal(signal.SIGINT, handle_signal)    # Ctrl-C
signal.signal(signal.SIGTERM, handle_signal)   # kill / scheduler stop

for item in items:
    if stop:
        break                  # exit cleanly between items, checkpoint intact
    process(item)

Catch the stop signal and finish the current item before exiting — so you never leave a half-written record. The checkpoint means the next run continues seamlessly.

Goal: process a large list of items with all four habits — checkpoint, isolate, retry, graceful stop — so it can crash, be killed, or hit bad items and still finish correctly on the next run.

import json, time, signal, logging, functools, random
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s",
                    datefmt="%H:%M:%S")
log = logging.getLogger("batch")

CHECKPOINT = Path("progress.json")
FAILURES = Path("failures.json")
_stop = False

def _handle(signum, frame):
    global _stop; _stop = True
    log.info("stop requested — will finish current item")
signal.signal(signal.SIGINT, _handle)

def retry(attempts=3, base=0.5):
    def deco(fn):
        @functools.wraps(fn)
        def w(*a, **k):
            for n in range(1, attempts + 1):
                try: return fn(*a, **k)
                except (ConnectionError, TimeoutError) as e:
                    if n == attempts: raise
                    time.sleep(base * 2 ** (n - 1) + random.uniform(0, 0.3))
            return None
        return w
    return deco

@retry(attempts=4)
def process(item: dict) -> None:
    # …real work: call an API, write a row, etc. (idempotent!)…
    if item["id"] == 5:
        raise ValueError("bad data in item 5")     # permanent → not retried
    time.sleep(0.05)

def run(items: list[dict]) -> None:
    done = set(json.loads(CHECKPOINT.read_text())) if CHECKPOINT.exists() else set()
    failures = json.loads(FAILURES.read_text()) if FAILURES.exists() else {}
    log.info("resuming: %d already done", len(done))

    for item in items:
        if _stop:
            log.info("graceful stop at item %s", item["id"]); break
        if item["id"] in done:
            continue
        try:
            process(item)
            done.add(item["id"])
            CHECKPOINT.write_text(json.dumps(list(done)))   # checkpoint each
        except Exception as e:
            log.exception("item %s failed", item["id"])
            failures[str(item["id"])] = str(e)
            FAILURES.write_text(json.dumps(failures))
    log.info("finished: %d done, %d failed", len(done), len(failures))

run([{"id": i} for i in range(10)])

12:00:00 INFO resuming: 0 already done
12:00:00 ERROR item 5 failed
ValueError: bad data in item 5
12:00:01 INFO finished: 9 done, 1 failed
# Re-run later: "resuming: 9 already done" — only item 5 is retried.

Read the code

All four habits work together: each success is checkpointed immediately, item 5's permanent error is isolated (logged to failures.json, batch continues), transient errors are retried via the decorator (but ValueError isn't — it'd just fail again), and Ctrl-C triggers a graceful stop between items. Kill it mid-run and the re-run picks up exactly where it left off. This skeleton is how you make any long pipeline — including the capstone — survive the real world.

Long jobs must survive failure. The four habits: checkpoint progress (persist after each item, skip done ones on resume); isolate per-item errors (log + collect, never crash the batch); retry transient errors with backoff (a reusable decorator — but only retryable exceptions); and make steps idempotent (re-running is harmless — the safety net under retries and resumes). Add graceful shutdown (catch SIGINT/SIGTERM, stop cleanly between items) and a failures-re-run pass. Then a crash at hour 3 costs you the current item, not the whole job.