Handling Routing Engine Rate Limits with Circuit Breakers

A nightly batch that map-matches a whole fleet will, sooner or later, overwhelm whatever engine sits behind it — a hosted API returns HTTP 429 once you cross the quota, and a single self-hosted OSRM or Valhalla container returns 503 or simply slows to a crawl when its request queue saturates. This page gives you one self-contained defence: a token-bucket rate limiter that keeps the client under the request rate, wrapped in a circuit breaker that fails fast when the engine is already unhealthy. It extends the benchmarking and batch patterns in choosing a routing engine and applies to any of the engines surveyed in the routing integration overview.

Compatibility and Configuration Requirements

The Rate Limiter and Circuit Breaker

The class below is copy-paste ready. It composes a token-bucket limiter (throttles the request rate) with a circuit breaker (fails fast when the engine is unhealthy) and adds exponential backoff with jitter on rate-limit responses. Every parameter choice is commented inline.

import time
import random
import threading
from typing import Callable, Optional


class RoutingEngineGuard:
    """
    Token-bucket rate limiter + circuit breaker for a routing engine HTTP API.

    Wrap every OSRM / Valhalla / GraphHopper call with `guard.call(fn)`.

    Parameters
    ----------
    rate_per_sec : float
        Sustainable requests per second. Set to the hosted provider's
        documented quota, or the benchmarked sustainable rate of a
        self-hosted container. This is the token refill rate.
    bucket_size : int
        Maximum burst of requests allowed at once. Larger buckets tolerate
        bursty batch jobs; keep it near rate_per_sec (1-2 s of tokens) so a
        burst cannot blow far past the average quota.
    failure_threshold : int
        Consecutive failures before the breaker trips OPEN. 5 is a good
        default; lower it for a fragile engine, raise it for a robust API
        that returns occasional transient errors.
    reset_timeout_s : float
        Seconds the breaker stays OPEN before allowing one HALF-OPEN probe.
        30 s lets a saturated engine drain its queue before you retry.
    max_backoff_s : float
        Ceiling for exponential backoff on 429/503 responses, so a long
        outage does not push a single retry sleep into the minutes.
    """

    _CLOSED = "closed"
    _OPEN = "open"
    _HALF_OPEN = "half_open"

    def __init__(
        self,
        rate_per_sec: float = 20.0,
        bucket_size: int = 40,
        failure_threshold: int = 5,
        reset_timeout_s: float = 30.0,
        max_backoff_s: float = 60.0,
    ):
        self.rate_per_sec = rate_per_sec
        self.bucket_size = bucket_size
        self.failure_threshold = failure_threshold
        self.reset_timeout_s = reset_timeout_s
        self.max_backoff_s = max_backoff_s

        # Token bucket state
        self._tokens = float(bucket_size)
        self._last_refill = time.monotonic()

        # Circuit breaker state
        self._state = self._CLOSED
        self._consecutive_failures = 0
        self._opened_at = 0.0

        self._lock = threading.Lock()

    # -- Token bucket -------------------------------------------------------
    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(
            self.bucket_size,
            self._tokens + elapsed * self.rate_per_sec,
        )
        self._last_refill = now

    def _acquire_token(self, timeout_s: float = 30.0) -> None:
        """Block until a token is available or raise on timeout."""
        deadline = time.monotonic() + timeout_s
        while True:
            with self._lock:
                self._refill()
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    return
                # Time until the next token becomes available
                wait = (1.0 - self._tokens) / self.rate_per_sec
            if time.monotonic() + wait > deadline:
                raise TimeoutError("Rate limiter: no token before timeout")
            time.sleep(min(wait, 0.5))

    # -- Circuit breaker ----------------------------------------------------
    def _before_call(self) -> None:
        with self._lock:
            if self._state == self._OPEN:
                if time.monotonic() - self._opened_at >= self.reset_timeout_s:
                    self._state = self._HALF_OPEN  # allow one probe
                else:
                    raise CircuitOpenError("Circuit is OPEN; failing fast")

    def _on_success(self) -> None:
        with self._lock:
            self._consecutive_failures = 0
            self._state = self._CLOSED

    def _on_failure(self) -> None:
        with self._lock:
            self._consecutive_failures += 1
            if (
                self._state == self._HALF_OPEN
                or self._consecutive_failures >= self.failure_threshold
            ):
                self._state = self._OPEN
                self._opened_at = time.monotonic()

    # -- Public entry point -------------------------------------------------
    def call(
        self,
        fn: Callable[[], "httpx.Response"],
        max_retries: int = 4,
        retry_after_of: Optional[Callable[[Exception], Optional[float]]] = None,
    ):
        """
        Execute fn() under the rate limiter and circuit breaker.

        fn must perform the HTTP request and raise on a non-recoverable error.
        A RateLimitedError (defined by the caller) signals 429/503 and triggers
        backoff without immediately counting toward the failure threshold.
        """
        self._before_call()

        attempt = 0
        while True:
            self._acquire_token()
            try:
                result = fn()
                self._on_success()
                return result
            except RateLimitedError as exc:
                # Honour Retry-After if the server sent one, else backoff
                server_wait = exc.retry_after
                backoff = server_wait if server_wait is not None else (
                    min(self.max_backoff_s, (2 ** attempt))
                    + random.uniform(0, 0.5)  # jitter avoids thundering herd
                )
                attempt += 1
                if attempt > max_retries:
                    self._on_failure()
                    raise
                time.sleep(backoff)
            except Exception:
                # Hard failure (connection error, 5xx) counts toward the breaker
                self._on_failure()
                raise


class CircuitOpenError(RuntimeError):
    """Raised when the breaker is OPEN and the call is rejected fast."""


class RateLimitedError(RuntimeError):
    """Raised by the caller's fn() on an HTTP 429/503 rate-limit response."""

    def __init__(self, retry_after: Optional[float] = None):
        super().__init__("Routing engine rate limited")
        self.retry_after = retry_after

Wire it to a real engine call by having fn() raise RateLimitedError on a throttle response:

import httpx

guard = RoutingEngineGuard(rate_per_sec=20.0, bucket_size=40)


def osrm_match_call(coords_path: str, base_url="http://localhost:5000"):
    def fn():
        r = httpx.get(f"{base_url}/match/v1/driving/{coords_path}",
                      params={"geometries": "polyline"}, timeout=30.0)
        if r.status_code in (429, 503):
            retry_after = r.headers.get("Retry-After")
            raise RateLimitedError(float(retry_after) if retry_after else None)
        r.raise_for_status()
        return r.json()

    return guard.call(fn)

Applying the Guard to a Nightly Batch

In a fleet batch, one guard instance fronts the engine and every trip request passes through it. Because the token bucket serialises access to the shared rate, a thread pool can fan out trips without collectively overrunning the quota — the limiter absorbs the coordination.

from concurrent.futures import ThreadPoolExecutor, as_completed

guard = RoutingEngineGuard(rate_per_sec=20.0, bucket_size=40, failure_threshold=5)


def match_trip(trip):
    """Match one trip; skipped fast if the breaker is already OPEN."""
    coords_path = ";".join(f"{lon:.6f},{lat:.6f}" for lon, lat in trip.coords)
    try:
        return trip.id, osrm_match_call(coords_path)
    except CircuitOpenError:
        return trip.id, None          # engine unhealthy — requeue for later
    except TimeoutError:
        return trip.id, None          # no token in time — shed and retry


def run_batch(trips, workers=8):
    results, deferred = {}, []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(match_trip, t): t for t in trips}
        for fut in as_completed(futures):
            trip_id, matched = fut.result()
            if matched is None:
                deferred.append(futures[fut])
            else:
                results[trip_id] = matched
    return results, deferred

Trips that come back None — rejected by an OPEN breaker or a token timeout — are collected in deferred and reprocessed after the reset timeout, rather than lost. Keep workers modest: eight threads against a rate_per_sec of 20 keeps every worker busy without any single one starving, because the bucket, not the pool, sets the pace.

Execution and Tuning Guidelines

Run one RoutingEngineGuard instance per engine endpoint — the token bucket and breaker state are per-server, so sharing one instance across two engines would let one engine’s outage starve the other.

• rate_per_sec and bucket_size. The refill rate is your average quota; the bucket is your burst allowance. Keep the bucket to roughly one to two seconds of tokens (bucket_size ≈ 1–2 × rate_per_sec). Raising bucket_size lets a batch surge through faster but risks a burst that trips a stricter provider quota; lowering it smooths traffic at the cost of throughput.
• failure_threshold. How many consecutive hard failures trip the breaker. Lower it (2–3) for a fragile self-hosted container so you stop hammering it quickly; raise it (8–10) for a robust hosted API where isolated 5xx responses are noise you should ride through.
• reset_timeout_s. How long the breaker stays open before a probe. Raise it when the engine needs time to drain a saturated queue or reload tiles; lower it when outages are brief and you want to recover throughput fast. Too short and you probe before the engine has recovered, flapping between OPEN and HALF-OPEN.
• max_backoff_s. Caps exponential backoff so a long outage does not produce a multi-minute sleep on a single retry. Pair it with the jitter term already in the code to avoid a thundering herd when many workers retry at once.
• Acquire timeout. The _acquire_token timeout bounds how long a caller waits for a token. For a batch job, keep it generous; for a real-time path, keep it short and handle the TimeoutError by shedding load.

Observing breaker state in production

A circuit breaker that trips silently is only marginally better than no breaker at all — you want to know when the engine went unhealthy, for how long, and how often it flaps. Emit a metric each time the state changes and each time a call is rejected fast, then alert on sustained OPEN time. Expose the current state cheaply so a health endpoint or dashboard can read it without touching the engine:

def snapshot(self) -> dict:
    """Cheap, lock-guarded view of guard state for metrics/health checks."""
    with self._lock:
        self._refill()
        return {
            "state": self._state,
            "tokens_available": round(self._tokens, 2),
            "consecutive_failures": self._consecutive_failures,
            "seconds_open": (
                round(time.monotonic() - self._opened_at, 1)
                if self._state != self._CLOSED else 0.0
            ),
        }

Poll snapshot() on your metrics interval and record state as a labelled gauge. A breaker that opens and closes many times per minute (flapping) signals that reset_timeout_s is too short for the engine to recover — raise it until the state settles. A breaker stuck OPEN for minutes is a genuine outage that should page, not a tuning problem.

Common Pitfalls

Up: Choosing a Routing Engine: OSRM vs Valhalla vs GraphHopper | Routing Engine Integration for Fleet Telematics

Related

• Choosing a Routing Engine: OSRM vs Valhalla vs GraphHopper — the benchmark harness this guard protects from tripping provider throttles
• Routing Engine Integration for Fleet Telematics — where rate limiting and caching sit in the overall integration architecture
• OSRM Integration for Fleet Map Matching — the /match endpoint the wrapper example calls
• Hidden Markov Model Map Matching in Python — the matching algorithm behind the engines you are throttling