Distributing Token Buckets Across Async Workers

An OTA enforces its rate limit per API client, not per process, so the moment you scale a sync from one worker to a pool each worker’s local token bucket only knows about its own traffic — six workers each pacing to 10 requests/second collectively hammer booking_com at 60 and earn a wave of 429s. This page builds a distributed token bucket in Redis so the entire pool draws from one shared allowance per endpoint, refilled and consumed atomically inside a single Lua script. It is the horizontal-scale companion to the single-worker pacing covered in the guide to handling OTA API rate limits.

The audience is the engineer whose availability and rate pushes fan out across an asyncio pool or several containers, all sharing one OTA credential. The goal is a fleet that collectively stays under the documented ceiling — say 20 requests/second on the ARI endpoint and a slower 5/second on the rate-plan endpoint — without any single worker needing to know how many peers exist.

Prerequisites & environment

The bucket sits in front of every outbound OTA call, so pin these so the refill arithmetic and atomicity hold across the fleet:

Python 3.11+ — for asyncio and the async Redis client.
redis 5.x (redis-py) with asyncio — the bucket state (tokens, last-refill timestamp) lives in one hash per endpoint; the refill-and-consume runs server-side as a Lua script so it is atomic across all workers.
httpx 0.27+ — the async client making the paced calls; acquire a token, then send.
tenacity 8.x — for the bounded retry that still wraps a genuine 429, since a shared bucket reduces but never fully eliminates limit responses.
structlog 24.x — throttle waits and 429s log as key=value keyed by channel slug and endpoint.
A shared Redis reachable by every worker. Rate-limit budgets should be treated as a per-credential resource, scoped alongside the OAuth2 token refresh strategies the same worker uses to authenticate.

Get the OTA’s published limits per endpoint before coding — most channel managers document different ceilings for availability, rates, and reservations, and a single global bucket either starves the fast endpoint or overruns the slow one.

Figure 1: Every worker consumes a token from a shared per-endpoint bucket via one atomic Lua script before it may call the OTA, so the whole fleet stays under each documented ceiling.

Step-by-step implementation

Four steps: express each endpoint’s limit as a bucket config, do the refill-and-consume atomically in Lua, wrap it in an acquire() coroutine that sleeps precisely when empty, and still bracket the actual call with a bounded 429 retry as a backstop.

Step 1 — Model each endpoint’s limit as a bucket

A token bucket has two parameters: a refill rate (tokens per second) and a capacity (the burst ceiling). Encode one config per channel-and-endpoint pair, because an OTA almost always meters availability, rates, and reservations independently.

python

from pydantic import BaseModel, Field

class BucketConfig(BaseModel):
    channel: str
    endpoint: str
    rate_per_sec: float = Field(gt=0)   # steady-state refill rate
    capacity: float = Field(gt=0)       # max burst = bucket depth

    @property
    def key(self) -> str:
        # one bucket per channel+endpoint — never a single global bucket
        return f"bkt:{self.channel}:{self.endpoint}"

BUCKETS = {
    ("booking_com", "availability"): BucketConfig(
        channel="booking_com", endpoint="availability", rate_per_sec=20, capacity=40),
    ("booking_com", "rate_plans"): BucketConfig(
        channel="booking_com", endpoint="rate_plans", rate_per_sec=5, capacity=10),
}

Splitting the buckets by endpoint is what stops a burst of availability pushes from starving the slower rate-plan endpoint: a single global bucket sized for the fast limit would let rate-plan calls sail past their own lower ceiling and trip a 429 the shared counter never saw coming.

Step 2 — Refill and consume atomically in Lua

The refill and the consume must be one indivisible operation, or two workers reading “1 token left” both consume it. Push the whole computation into Redis as a Lua script: it lazily adds tokens for the elapsed time, then either deducts one or reports the wait.

python

# KEYS[1] = bucket key; ARGV = rate, capacity, now_ms, requested(=1)
REFILL_CONSUME = """
local rate      = tonumber(ARGV[1])
local capacity  = tonumber(ARGV[2])
local now_ms    = tonumber(ARGV[3])
local want      = tonumber(ARGV[4])
local state = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(state[1]) or capacity
local ts     = tonumber(state[2]) or now_ms
-- lazily refill for the time elapsed since the last touch
local elapsed = math.max(0, now_ms - ts) / 1000.0
tokens = math.min(capacity, tokens + elapsed * rate)
if tokens >= want then
  tokens = tokens - want
  redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now_ms)
  redis.call('PEXPIRE', KEYS[1], math.ceil(capacity / rate * 2000))
  return {1, 0}                      -- granted, no wait
else
  local deficit = want - tokens
  local wait_ms = math.ceil(deficit / rate * 1000)
  redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now_ms)
  return {0, wait_ms}                -- denied, sleep this long
end
"""

Running refill-and-consume as one server-side script is the entire correctness argument: because Redis executes a Lua script atomically, no two workers can ever interleave their read-modify-write on the same bucket, so the shared token count can never be double-spent no matter how many workers hit it at once.

Step 3 — Wrap it in an acquire() that sleeps precisely when empty

The Python side becomes a thin loop: call the script, and if it returns a wait, sleep exactly that long before retrying. Sleeping the script-computed interval — rather than a fixed poll — means the worker wakes the instant a token is actually available.

python

import time
import asyncio
import redis.asyncio as redis
import structlog

log = structlog.get_logger()

class DistributedBucket:
    def __init__(self, r: redis.Redis):
        self._script = r.register_script(REFILL_CONSUME)

    async def acquire(self, cfg: BucketConfig, max_wait: float = 5.0) -> None:
        deadline = time.monotonic() + max_wait
        while True:
            now_ms = int(time.time() * 1000)
            granted, wait_ms = await self._script(
                keys=[cfg.key],
                args=[cfg.rate_per_sec, cfg.capacity, now_ms, 1])
            if granted:
                return
            if time.monotonic() >= deadline:
                raise TimeoutError(f"rate budget exhausted for {cfg.key}")
            log.info("rate_throttle_wait", key=cfg.key, wait_ms=wait_ms)
            await asyncio.sleep(wait_ms / 1000.0)   # wake exactly when a token refills

Bounding the loop with max_wait keeps a saturated bucket from blocking a coroutine forever: if the fleet is so far over budget that a token won’t arrive in five seconds, that push should surface as backpressure to the scheduler rather than pile up an unbounded queue of sleeping coroutines.

Step 4 — Still wrap the real call in a bounded 429 retry

A shared bucket keeps the fleet under the documented limit, but OTAs enforce hidden sub-limits and occasionally return 429 anyway. Keep a short tenacity retry that honours Retry-After around the actual request as a backstop.

python

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RateLimited(Exception): ...

@retry(retry=retry_if_exception_type(RateLimited),
       wait=wait_exponential(multiplier=1, max=30),
       stop=stop_after_attempt(4), reraise=True)
async def push_availability(bucket: DistributedBucket, client: httpx.AsyncClient,
                            cfg: BucketConfig, payload: dict) -> httpx.Response:
    await bucket.acquire(cfg)                    # pace against the shared budget first
    resp = await client.post("/ari/availability", json=payload)
    if resp.status_code == 429:
        # respect the server's own hint; the bucket underestimated a hidden sub-limit
        retry_after = float(resp.headers.get("Retry-After", "2"))
        await asyncio.sleep(retry_after)
        raise RateLimited(f"429 for {cfg.channel} property {payload.get('property_id')}")
    resp.raise_for_status()
    return resp

Keeping the exponential-backoff retry even with a correct bucket is defence in depth: the bucket enforces the limit you know about, and the Retry-After-aware backoff — the same discipline as implementing exponential backoff in Python — absorbs the ones the OTA never published.

Gotchas & production notes

Clock skew between workers is harmless; clock skew from Redis is not. The Lua script uses each worker’s now_ms, so a worker whose clock runs fast would over-refill. Prefer passing Redis’s own time via redis.call('TIME', ...) inside the script if your fleet’s NTP discipline is weak — a single authoritative clock removes refill drift entirely.
Size capacity to a real burst, not to the per-second rate. Setting capacity equal to rate_per_sec forbids any burst and serializes the fleet into a lockstep drip; setting it far too high lets a cold bucket dump a huge burst that instantly trips the OTA. Match capacity to the OTA’s documented burst allowance, typically one to two seconds of rate.
Retry-After from the OTA must win over the bucket’s wait. When you do get a 429, the server is telling you the truth about a limit your bucket underestimated; sleeping its Retry-After and only then re-acquiring keeps you from immediately re-consuming a token and re-tripping the same limit. Feed persistent 429s into error categorization and retry logic so a limit that never clears becomes an alert, not an infinite wait.
Reservations and inventory pulls draw from the same credential — bucket them too. It is easy to pace only the write endpoints and forget that a nightly batch reconciliation pulling rosters shares the same rate ceiling; give every endpoint the credential touches its own bucket or the read path will silently blow the write path’s budget.

Verification snippet

The invariant to prove is fleet-wide: N workers hammering one bucket for T seconds must consume no more than capacity + rate * T tokens. Drive many coroutines through a fakeredis-backed bucket and count grants.

python

import asyncio
import time
import fakeredis.aioredis

async def test_fleet_respects_shared_limit() -> None:
    r = fakeredis.aioredis.FakeRedis()
    bucket = DistributedBucket(r)
    cfg = BucketConfig(channel="booking_com", endpoint="availability",
                       rate_per_sec=20, capacity=40)
    grants = {"n": 0}

    async def worker() -> None:
        end = time.monotonic() + 1.0        # run each worker for ~1 second
        while time.monotonic() < end:
            try:
                await bucket.acquire(cfg, max_wait=0.2)
                grants["n"] += 1
            except TimeoutError:
                pass

    await asyncio.gather(*(worker() for _ in range(10)))
    # ceiling over ~1s: initial burst (40) + refill (20/s * 1s) with slack
    assert grants["n"] <= 40 + 20 + 5

asyncio.run(test_fleet_respects_shared_limit())

A green run establishes the guarantee the whole design exists for: ten concurrent workers, none of which knew the others existed, together consumed no more tokens than the single shared ceiling permits — so the fleet paces as one client against the OTA instead of multiplying its request rate by the worker count.

Handling OTA API Rate Limits — the parent guide to staying inside an OTA’s request ceilings
Implementing Exponential Backoff in Python — the Retry-After-aware backoff that backstops the bucket on a real 429
Caching Access Tokens in Redis with Proactive Refresh — the same atomic Redis-and-Lua discipline applied to shared credentials
Error Categorization & Retry Logic — where a limit that never clears becomes a terminal alert rather than an endless wait
Batch Reconciliation Workflows — the nightly read path that must share the same per-credential budget

← Back to Handling OTA API Rate Limits