Add Retries and Failure Alerts to Automation Jobs
The nightly job fails on a network blip, so a retry gets added. Then it fails differently:
for attempt in range(5):
try:
process(document)
break
except Exception:
time.sleep(2)
A corrupt PDF is now retried five times before failing anyway, adding ten seconds per bad file across four hundred of them. A rate-limited API is hammered five times in ten seconds, extending the ban. And the alerting added afterwards sends one email per failed document, so a bad night produces four hundred emails and everyone filters the channel.
Root Cause
Retrying everything is the mistake. Errors fall into two groups and only one benefits from another attempt: transient failures — a timeout, a connection reset, a 503, a locked file — where the same call may succeed shortly afterwards, and permanent failures — a corrupt document, a missing field, a 404, a validation error — where it never will. Catching bare Exception erases the distinction, so permanent failures consume the retry budget and delay the real outcome. The alerting problem has the same shape: a hundred instances of one broken upstream service are one incident, not a hundred, and paging per failure guarantees the notifications are muted before the night they matter.
Minimal Diagnostic
Before tuning anything, find out what the job is actually failing on.
# stdlib only
import json
import re
from collections import Counter
from pathlib import Path
LOG = Path("logs/run.jsonl")
def failure_profile(path: Path) -> None:
kinds, messages, retried = Counter(), Counter(), Counter()
for line in path.read_text(encoding="utf-8").splitlines():
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
if record.get("level") != "ERROR":
continue
kinds[record.get("error_type", "unknown")] += 1
normalised = re.sub(r"\d+", "N", str(record.get("message", ""))[:70])
messages[normalised] += 1
retried[record.get("attempts", 1)] += 1
total = sum(kinds.values())
print(f"{total} failure(s) in {path.name}")
for kind, count in kinds.most_common(8):
print(f" {count:>5} {kind}")
print(" attempts distribution:", dict(sorted(retried.items())))
for message, count in messages.most_common(5):
print(f" {count:>5} {message}")
if __name__ == "__main__":
failure_profile(LOG)
412 failure(s) in run.jsonl
389 PdfReadError
18 ReadTimeout
5 HTTPError
attempts distribution: {5: 412}
389 EOF marker not found
18 HTTPSConnectionPool host='api.internal', read timeout=N
Every failure was retried five times, but 389 of them are corrupt PDFs that could never succeed — roughly 32 minutes of the run spent sleeping between doomed attempts.
Fix: Classify, Then Retry Only the Transient
Separate the decision from the mechanism, and make backoff jittered so retries do not synchronise.
# pip install requests
import logging
import random
import time
from dataclasses import dataclass
from typing import Callable, TypeVar
import requests
log = logging.getLogger("pipeline")
T = TypeVar("T")
class PermanentError(Exception):
"""The item will never succeed; quarantine it."""
TRANSIENT_EXCEPTIONS = (requests.Timeout, requests.ConnectionError, TimeoutError, OSError)
TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504}
def classify(error: BaseException) -> tuple[bool, float | None]:
"""Returns (retryable, suggested delay from the server, if any)."""
if isinstance(error, requests.HTTPError) and error.response is not None:
status = error.response.status_code
if status in TRANSIENT_STATUS:
after = error.response.headers.get("Retry-After") # changed: honour the server
return True, float(after) if after and after.isdigit() else None
return False, None # changed: other 4xx never retried
if isinstance(error, PermanentError):
return False, None
return isinstance(error, TRANSIENT_EXCEPTIONS), None
@dataclass(frozen=True)
class RetryPolicy:
attempts: int = 4
base_delay: float = 1.0
max_delay: float = 30.0
budget_seconds: float = 120.0
def with_retries(operation: Callable[[], T], policy: RetryPolicy = RetryPolicy(),
label: str = "operation") -> T:
started = time.monotonic()
last: BaseException | None = None
for attempt in range(1, policy.attempts + 1):
try:
return operation()
except BaseException as error: # noqa: BLE001 — classified below
retryable, server_delay = classify(error)
last = error
if not retryable or attempt == policy.attempts:
raise
delay = server_delay if server_delay is not None else min(
policy.base_delay * 2 ** (attempt - 1), policy.max_delay)
delay *= 0.5 + random.random() # changed: jitter, so retries spread
if time.monotonic() - started + delay > policy.budget_seconds:
log.warning("%s: retry budget exhausted after %d attempt(s)", label, attempt)
raise
log.info("%s: attempt %d failed (%s), retrying in %.1fs",
label, attempt, type(error).__name__, delay)
time.sleep(delay)
raise last # unreachable, keeps type checkers happy
The retry budget is as important as the attempt count. Four attempts with exponential backoff can span a minute per item, and four hundred items make a job that never finishes; capping the total time spent retrying one item bounds the whole run. Jitter matters whenever several workers retry at once — without it they back off in lockstep and hit the recovering service simultaneously, which is how a brief outage becomes a long one.
Variant Fix 1: Stop When the Whole Run Is Failing
If most items fail the same way, the problem is not the items. A circuit breaker turns a doomed four-hour run into a two-minute failure:
# stdlib only
from dataclasses import dataclass, field
@dataclass
class Circuit:
threshold: float = 0.5 # fraction of recent items failing
window: int = 40 # how many recent items to consider
minimum: int = 20 # do not trip on a tiny sample
recent: list[bool] = field(default_factory=list)
def record(self, ok: bool) -> None:
self.recent.append(ok)
del self.recent[:-self.window]
def should_stop(self) -> bool:
if len(self.recent) < self.minimum:
return False
failure_rate = 1 - sum(self.recent) / len(self.recent)
return failure_rate >= self.threshold
class RunAborted(RuntimeError):
"""Most recent items failed; the problem is upstream."""
Aborting is the kinder failure. A job that processes four hundred documents into a downstream system with a 90% failure rate has done real damage by the time it finishes; one that stops after forty has produced a clear signal and left the rest untouched, ready for a rerun once the upstream problem is fixed.
Variant Fix 2: Alerting Once per Incident
Group failures by cause and send one message, with counts:
# stdlib only
import hashlib
import json
import os
import re
import urllib.request
from collections import Counter
from pathlib import Path
STATE = Path("state/alerted.json")
def fingerprint(error_type: str, message: str) -> str:
normalised = re.sub(r"[0-9a-f]{8,}|\d+", "N", message)[:120]
return hashlib.sha256(f"{error_type}|{normalised}".encode()).hexdigest()[:12]
def alert_once(failures: list[tuple[str, str]], run_id: str, webhook: str | None = None) -> list[str]:
grouped = Counter(fingerprint(kind, message) for kind, message in failures)
examples = {}
for kind, message in failures:
examples.setdefault(fingerprint(kind, message), (kind, message))
seen = json.loads(STATE.read_text()) if STATE.exists() else {}
sent = []
for key, count in grouped.most_common():
if seen.get(key) == run_id:
continue # already alerted this run
kind, message = examples[key]
text = (f"[{run_id}] {count} x {kind}: {message[:160]}"
f"{' (recurring)' if key in seen else ''}")
if webhook:
request = urllib.request.Request(
webhook, data=json.dumps({"text": text}).encode(),
headers={"Content-Type": "application/json"})
urllib.request.urlopen(request, timeout=10).read()
seen[key], _ = run_id, sent.append(text)
STATE.parent.mkdir(parents=True, exist_ok=True)
STATE.write_text(json.dumps(seen))
return sent
Normalising digits and hex strings out of the message before hashing is what makes grouping work: EOF marker not found in invoice-0041.pdf and the same error on invoice-0042.pdf produce one fingerprint and one alert saying 389 x PdfReadError. Marking a fingerprint seen in a previous run as (recurring) distinguishes a new problem from one already being worked on, which is the distinction that decides whether anyone needs to look tonight.
What Makes Retries Safe at All
None of this is safe unless the operation is idempotent. Retrying a step that appends a row, sends an email or increments a counter turns a transient failure into duplicated work, which is usually worse than the failure.
Three patterns make an operation safe to repeat. Write to a temporary name and rename into place, so a partial write is never visible and a repeat simply overwrites. Derive a deterministic key from the input — a content hash, or the source filename and a run date — and use it as the primary key or the object name, so a second attempt replaces rather than adds. And check before acting where the destination allows it: an upload that first asks whether the object already exists costs one request and removes a whole class of duplication.
Where the downstream system genuinely cannot be made idempotent, retry at a higher level instead: record the item as needing attention and reprocess it in a later run, with a human deciding. That is slower and it is correct, which is the right trade for anything that touches money or customer communication.
Verification
Test the classification and the backoff rather than hoping.
# pip install requests
import time
import requests
def test_permanent_not_retried() -> None:
calls = []
def operation():
calls.append(time.monotonic())
response = requests.Response()
response.status_code = 404
raise requests.HTTPError(response=response)
try:
with_retries(operation, RetryPolicy(attempts=4), label="404")
except requests.HTTPError:
pass
assert len(calls) == 1, f"permanent error attempted {len(calls)} times, expected 1"
def test_transient_retried_with_growing_delay() -> None:
calls = []
def operation():
calls.append(time.monotonic())
if len(calls) < 3:
raise requests.ConnectionError("reset")
return "ok"
assert with_retries(operation, RetryPolicy(attempts=4, base_delay=0.05)) == "ok"
assert len(calls) == 3, f"{len(calls)} attempt(s), expected 3"
gaps = [b - a for a, b in zip(calls, calls[1:])]
assert gaps[1] > gaps[0], f"delay did not grow: {gaps}"
def test_budget_caps_total_time() -> None:
started = time.monotonic()
def operation():
raise TimeoutError("slow")
try:
with_retries(operation, RetryPolicy(attempts=20, base_delay=0.1, budget_seconds=0.5))
except TimeoutError:
pass
assert time.monotonic() - started < 2.0, "retry budget did not cap the total time"
if __name__ == "__main__":
test_permanent_not_retried()
test_transient_retried_with_growing_delay()
test_budget_caps_total_time()
print("retry policy verified")
The first test is the one that matters most, because the bug it catches — a permanent error being retried — costs time silently rather than failing. It also fails the moment someone widens the transient list to include a case that should not be there.
FAQ
Should I use a library like tenacity instead?
Yes, for anything beyond this. The classification function is still yours to write; that is the part no library can supply.
How many attempts is right? Three or four, with a time budget. Beyond that, the failure is not transient and more attempts only delay the answer.
Where should alerts go? Somewhere with an owner. A channel nobody owns is the same as no alerting, and a grouped alert makes a low-traffic channel viable.
Should a partial failure fail the whole run? Only past a threshold. Record the quarantined items, succeed if the rate is acceptable, and let the circuit breaker handle the rest.
Related
- Scheduling and Logging Automation Jobs — the scheduling workflow end to end
- Run Python Automation with a GitHub Actions Schedule — where the job runs
- Validating Document Data with Schemas — turning bad data into a clear permanent error
- Generating Reports from Pipeline Data — reporting what the run did