Make Broken Pipelines a Type Error: Typed Observability for Proxy Health in a Collection SDK

A proxy pool dies quietly. Not with an outage page, but with a slow drift: success rates slide a percent a week, a scraping SDK's users add try/except around calls that never used to fail, and someone pastes a JSON dump into a chat thread asking "does this look right?" By the time anything is obviously broken, the evidence of when it started breaking — and through which exits — has been aggregated away into a green checkmark on a dashboard built from averages.

The root cause is usually not missing monitoring. It's untyped, unstructured telemetry: proxy errors logged as free-text strings, exit health as a dict someone mutates from three modules, metrics with fields that appear and disappear between versions. My argument: in a collection SDK, the telemetry schema is an API, and it should be typed like one — because the questions you'll need to answer during an incident are exactly the questions a type system forces you to make explicit, and the IDE catches the malformed pipelines before they ship.

What goes wrong with stringly-typed telemetry

The typical pattern looks innocent:

logger.info(f"fetch failed: {url} via {proxy_ip} status={status}")

Six months later you have 40 formats of that line, three spellings of "timeout", and no way to compute "block rate per exit for origin X over the last hour" without a regex-and-prayer pipeline. The failure modes cluster:

• Unanswerable questions. Was that 403 from a residential exit or the datacenter tier? Which session token was riding it? The data to answer existed at runtime and evaporated at the log call.
• Silent schema drift. A field gets renamed in v1.3; dashboards keep rendering, now charting None.
• Wrong-but-valid aggregations. If "blocked" lumps 429s and 403s together, your burn-list logic averages two signals with opposite meanings (slow down vs. go away).

The typed event schema

The fix is a small closed set of typed events, one per state change in the proxy layer. Here is a schema that has survived real incidents:

# telemetry.py
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Iterator, Optional, Protocol


class ProxyTier(Enum):
    DATACENTER = "datacenter"
    RESIDENTIAL = "residential"
    ISP = "isp"
    MOBILE = "mobile"


class FailureKind(Enum):
    RATE_LIMIT = "rate_limit"        # 429: origin throttles exit
    BLOCK = "block"                  # 403/forbidden page: exit burned for origin
    PROXY_CONNECT = "proxy_connect"  # cannot reach the gateway/exit at all
    EXIT_TIMEOUT = "exit_timeout"    # connected, but exit too slow
    AUTH = "auth"                    # provider rejected credentials


@dataclass(frozen=True)
class ProxyEvent:
    """One state change in the proxy layer. Frozen: events are facts."""
    ts: float                      # unix seconds
    request_id: str                # joins events across a single fetch's life
    exit_id: str                   # provider-visible identity of the exit
    origin: str                    # target site label
    tier: ProxyTier
    ok: bool
    failure: Optional[FailureKind] = None   # None iff ok
    latency_ms: Optional[float] = None
    attempt: int = 1               # which retry attempt produced this event


class Sink(Protocol):
    def write(self, event: ProxyEvent) -> None: ...

Two choices in this schema earn their keep:

request_id joins the retry chain. When a fetch takes four attempts through three exits, you get four events sharing one request_id — the question "how many exits did it take to land this row" becomes a groupby instead of forensics.

FailureKind separates what averages hide. RATE_LIMIT means pace (back off, maybe switch), BLOCK means identity (this exit is finished for this origin), PROXY_CONNECT means infra (the exit never participated). Any consumer of this enum — burn lists, cooldowns, dashboards — is forced by the type checker to handle each case it cares about and can no longer blur them accidentally.

The in-memory store that answers incident questions

With typed events, the analytics you need during an incident are plain functions — no regex, no ETL:

class EventStore:
    def __init__(self, max_events: int = 200_000):
        self._events: list[ProxyEvent] = []
        self._max = max_events

    def write(self, event: ProxyEvent) -> None:
        self._events.append(event)
        if len(self._events) > self._max:
            self._events = self._events[-self._max // 2:]

    def events(self, since: Optional[float] = None) -> Iterator[ProxyEvent]:
        cutoff = since if since is not None else 0.0
        return (e for e in self._events if e.ts >= cutoff)

    def block_rate_per_exit(self, window_s: float = 3600.0) -> dict[str, float]:
        """Which exits are being refused — the 2 a.m. question."""
        since = time.time() - window_s
        per_exit: dict[str, list[bool]] = {}
        for e in self.events(since):
            if e.origin and e.failure in (FailureKind.BLOCK, FailureKind.RATE_LIMIT):
                per_exit.setdefault(e.exit_id, []).append(e.ok)
        return {k: 1.0 - (sum(v) / len(v)) for k, v in per_exit.items()}

    def exits_per_request(self, window_s: float = 3600.0) -> float:
        """Retry multiplier: how hard the fleet is working per delivered row."""
        since = time.time() - window_s
        per_req: dict[str, set[str]] = {}
        for e in self.events(since):
            per_req.setdefault(e.request_id, set()).add(e.exit_id)
        if not per_req:
            return 0.0
        return sum(len(v) for v in per_req.values()) / len(per_req)

    def failure_by_tier(self, window_s: float = 3600.0) -> dict[ProxyTier, dict[FailureKind, int]]:
        since = time.time() - window_s
        out: dict[ProxyTier, dict[FailureKind, int]] = {}
        for e in self.events(since):
            if not e.ok and e.failure is not None:
                out.setdefault(e.tier, {}).setdefault(e.failure, 0)
                out[e.tier][e.failure] += 1
        return out

Notice what failure_by_tier buys you: when residential exits start failing with BLOCK while datacenter exits fail with RATE_LIMIT, you are looking at an origin that upgraded its fingerprinting (residential IPs now get challenged as suspicious) — a completely different response than a provider gateway having a bad night (PROXY_CONNECT spiking on one tier). Same dashboard tile in a stringly-typed world; opposite playbooks in reality.

Emitting events from the fetch path

The emitter is deliberately boring, which is the point — boring, typed, and impossible to get wrong:

import aiohttp
import itertools

_counter = itertools.count(1)

async def fetch_typed(session: aiohttp.ClientSession, url: str, origin: str,
                      exit_id: str, tier: ProxyTier,
                      store: EventStore) -> str:
    request_id = f"req-{next(_counter)}"
    t0 = time.monotonic()
    try:
        async with session.get(url, proxy=f"http://user:[e-mail]:7000") as r:
            latency = (time.monotonic() - t0) * 1000
            failure = None
            if r.status == 429:
                failure = FailureKind.RATE_LIMIT
            elif r.status in (403, 401):
                failure = FailureKind.BLOCK
            store.write(ProxyEvent(
                ts=time.time(), request_id=request_id, exit_id=exit_id,
                origin=origin, tier=tier, ok=r.status == 200,
                failure=failure, latency_ms=round(latency, 1),
            ))
            if r.status != 200:
                raise RuntimeError(f"HTTP {r.status}")
            return await r.text()
    except (aiohttp.ClientProxyConnectionError, asyncio.TimeoutError) as exc:
        kind = (FailureKind.EXIT_TIMEOUT
                if isinstance(exc, asyncio.TimeoutError)
                else FailureKind.PROXY_CONNECT)
        store.write(ProxyEvent(
            ts=time.time(), request_id=request_id, exit_id=exit_id,
            origin=origin, tier=tier, ok=False, failure=kind,
        ))
        raise

Because ProxyEvent is frozen with non-optional types (plus failure: None iff ok), an attempt to write an event with ok=True and a failure kind — the classic lie your metrics tell during partial outages — fails loudly in review, and mypy --strict catches a missing tier before the code merges. Ship a py.typed marker with the package and every downstream consumer gets the same guarantees.

From types to alerts

The last step is turning the schema into thresholds that page a human for the right reason. With the functions above, three alerts cover most proxy-layer incidents:

• Block rate concentration: top-decile block_rate_per_exit() above 50% while the median stays low → specific exits are burned; rotate the burn list, don't touch concurrency.
• Retry multiplier drift: exits_per_request() above 1.5 for an hour → the fleet is collectively over-asking; lower global rate, regardless of which scripts look innocent.
• Tier-skewed PROXY_CONNECT: connect failures above 5% on one tier only → provider problem; fail over or pause that tier before you spend another gigabyte on dead dials.

Each alert maps to a distinct playbook, and each exists only because the schema made the distinction a type rather than a convention. That's the quiet bargain of typed telemetry: you pay a little structure up front, and in exchange, the difference between "slow down," "switch exits," and "call the provider" is never averaged away exactly when you need it most.

---

Disclosure: I use Thordata's residential proxies for this project. New users get 500MB free — code thor020 (10% off): https://www.thordata.com/?ls=uXcSHJzx&lk=02-tele