Token Bucket Rate Limiting for Webhook Senders

A consumer that publishes “100 requests per second, bursts to 200” has given you a contract, and a dispatcher that ignores it will be throttled, blocked, or de-provisioned. Sending as fast as your worker pool allows and treating the resulting 429 responses as retryable failures is the worst possible strategy: it converts every rate-limit rejection into extra retry traffic aimed at the endpoint that just told you to slow down. This guide implements the producer-side control described in Webhook Rate Limiting & Backpressure as running code — a token bucket held in Redis, refilled and consumed by a single atomic Lua script, with one bucket per destination endpoint. It is the outbound mirror of applying backpressure to webhook consumers: there you protect yourself from an over-eager sender, here you are the sender.

The distributed part is what makes this non-trivial. A token bucket inside a single Python process is thirty lines and a time.monotonic() call. A token bucket that stays correct while twelve dispatcher replicas draw from it concurrently, across restarts and rolling deploys, needs shared state and an operation that cannot interleave.

Prerequisites

• Redis 7 reachable from every dispatcher replica, with EVALSHA scripting available (Redis Cluster works if all bucket keys for one endpoint hash to one slot — use a hash tag such as wh:{ep_42}:bucket).
• Python 3.11+ with redis-py 5.x (redis.asyncio) and httpx 0.27 for outbound delivery.
• The consumer’s documented limit: sustained requests per second, burst allowance, and whether the limit is per API key, per endpoint, or per tenant. If it is undocumented, measure it — see benchmarking webhook throughput with k6.
• A dispatch path where a delivery can wait or be requeued without holding an inbound HTTP request open.
• A retry policy already in place, because a delivery that cannot get a token within its wait budget must be rescheduled rather than dropped — see choosing retry budgets and max attempts.

Step 1: Translate the published limit into rate and capacity

A token bucket has exactly two knobs, and they answer two different questions. rate is tokens added per second and sets the long-run average — it maps directly onto the consumer’s documented sustained limit. capacity is the maximum number of tokens the bucket can hold and therefore the largest burst you may fire after a quiet period.

The mistake is setting capacity equal to the burst number the consumer publishes. A consumer that advertises “bursts to 200” usually means its own limiter tolerates 200 in flight; if you also allow 200 and another sender is active, you jointly exceed it. Size capacity to a fraction of the published burst — 50% is a sane default when you are not the only sender — and leave headroom.

A capacity equal to one second of rate (capacity = rate) makes the bucket behave almost like a strict pacer with no burst tolerance. Capacity equal to ten seconds of rate lets a batch of ten seconds’ worth of events leave instantly after idle time, which is usually what you want for webhook fan-out — events arrive clumped, and a small burst allowance is the difference between “delivered immediately” and “queued behind an artificial pacer”.

Step 2: Make refill and consume atomic with a Lua script

The bucket operation is read tokens, compute refill from elapsed time, compare against the request, decrement, and write back. Doing that as separate Redis commands from multiple replicas is a textbook lost-update race: two dispatchers both read tokens = 1, both decide they may send, both write tokens = 0, and two requests leave against one token. Redis runs a Lua script to completion without interleaving other commands, which makes the whole read-modify-write one atomic step.

-- consume.lua
-- KEYS[1] = bucket hash key, e.g. wh:bucket:ep_42
-- ARGV[1] = refill rate in tokens per second
-- ARGV[2] = bucket capacity in tokens
-- ARGV[3] = tokens requested (normally 1)
-- Returns: {allowed(0|1), tokens_remaining, seconds_until_enough}
local rate      = tonumber(ARGV[1])
local capacity  = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])

-- Use the Redis server clock so replica clock skew cannot corrupt the refill.
local t   = redis.call('TIME')
local now = tonumber(t[1]) + (tonumber(t[2]) / 1000000)

local state  = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(state[1])
local ts     = tonumber(state[2])

if tokens == nil or ts == nil then
  tokens = capacity          -- a fresh bucket starts full
  ts = now
end

local elapsed = now - ts
if elapsed < 0 then
  elapsed = 0                -- guard against a clock step backwards
end
tokens = math.min(capacity, tokens + (elapsed * rate))

local allowed = 0
local wait = 0.0
if tokens >= requested then
  tokens = tokens - requested
  allowed = 1
else
  wait = (requested - tokens) / rate
end

redis.call('HSET', KEYS[1], 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', KEYS[1], math.ceil(capacity / rate) + 60)

-- Lua numbers are truncated to integers on the wire; send floats as strings.
return {allowed, tostring(tokens), tostring(wait)}

Three details in that script are load-bearing. Reading the clock with redis.call('TIME') gives every replica one authoritative clock, so a pod whose NTP has drifted two seconds cannot mint or destroy tokens. Returning wait lets a caller sleep for exactly as long as the deficit needs instead of polling blindly. And the EXPIRE on every call means a bucket for an endpoint you deleted last month evaporates on its own rather than living in Redis forever.

Step 3: Gate every dispatch through a per-endpoint bucket

Register the script once per client so redis-py uses EVALSHA and only ships the source on a NOSCRIPT miss. Key the bucket by endpoint identifier, never by URL string — URLs change, and two subscriptions pointing at the same host still deserve separate accounting unless the consumer’s limit is per host.

import asyncio
import random
import time
from dataclasses import dataclass
from pathlib import Path

import httpx
import redis.asyncio as aioredis

CONSUME_LUA = Path("consume.lua").read_text()


@dataclass(frozen=True)
class BucketConfig:
    rate: float          # sustained tokens per second
    capacity: float      # burst ceiling in tokens


class TokenBucketLimiter:
    def __init__(self, redis: aioredis.Redis, prefix: str = "wh:bucket:") -> None:
        self.redis = redis
        self.prefix = prefix
        self._script = redis.register_script(CONSUME_LUA)

    async def try_acquire(
        self, endpoint_id: str, cfg: BucketConfig, tokens: float = 1.0
    ) -> tuple[bool, float, float]:
        allowed, remaining, wait = await self._script(
            keys=[f"{self.prefix}{endpoint_id}"],
            args=[cfg.rate, cfg.capacity, tokens],
        )
        return bool(int(allowed)), float(remaining), float(wait)

    async def acquire(
        self,
        endpoint_id: str,
        cfg: BucketConfig,
        tokens: float = 1.0,
        max_wait: float = 5.0,
    ) -> float:
        """Block until a token is available. Returns seconds spent waiting."""
        started = time.monotonic()
        while True:
            ok, _remaining, wait = await self.try_acquire(endpoint_id, cfg, tokens)
            if ok:
                return time.monotonic() - started
            if time.monotonic() - started + wait > max_wait:
                raise TimeoutError(
                    f"no token for {endpoint_id} within {max_wait}s "
                    f"(needs {wait:.2f}s more)"
                )
            # Jitter the poll so concurrent waiters do not wake together.
            await asyncio.sleep(min(wait, 0.25) + random.uniform(0, 0.05))

Wrap the outbound call so no code path can send without a token:

from prometheus_client import Histogram

LIMITER_WAIT_SECONDS = Histogram(
    "webhook_limiter_wait_seconds",
    "Seconds a delivery spent waiting for a token",
    ["endpoint_id"],
)


class RateLimitedDispatcher:
    def __init__(self, limiter: TokenBucketLimiter, client: httpx.AsyncClient) -> None:
        self.limiter = limiter
        self.client = client

    async def deliver(
        self, endpoint_id: str, url: str, body: bytes, headers: dict[str, str],
        cfg: BucketConfig,
    ) -> httpx.Response:
        waited = await self.limiter.acquire(endpoint_id, cfg, max_wait=5.0)
        if waited > 0.5:
            # Sustained waiting means the queue is offered faster than the limit.
            LIMITER_WAIT_SECONDS.labels(endpoint_id=endpoint_id).observe(waited)
        return await self.client.post(url, content=body, headers=headers, timeout=10.0)

When acquire raises TimeoutError, do not drop the event and do not tighten the loop. Requeue it with a delay of roughly the reported deficit; the delivery has not failed, it simply has not started. Treating a token timeout as a delivery failure inflates your error rate and, if it feeds the retry counter, burns attempts on an event that never left the process.

Step 4: Adapt the rate from 429 and Retry-After

Static configuration goes stale. Consumers lower their limits, add per-tenant quotas, or throttle during their own incidents, and the first you hear about it is a 429. A static bucket keeps sending at the configured rate and collects rejections indefinitely; an adaptive bucket treats the 429 as the authoritative signal it is.

Two mechanisms work together. Retry-After is a hard instruction — set a cooldown key with exactly that TTL and refuse to acquire tokens for that endpoint until it expires. Separately, multiplicatively decrease the stored rate (halve it, with a floor) so that when the cooldown ends you are not immediately back at the rate that triggered the throttle. Recovery is additive: a clean streak grows the rate by a small percentage per interval, back toward the configured ceiling but never above it.

COOLDOWN_PREFIX = "wh:cooldown:"
RATE_PREFIX = "wh:rate:"


class AdaptiveRate:
    """Multiplicative decrease on 429, additive increase on a clean streak."""

    def __init__(
        self,
        redis: aioredis.Redis,
        ceiling: float,
        floor: float = 1.0,
        decrease: float = 0.5,
        increase: float = 1.10,
    ) -> None:
        self.redis = redis
        self.ceiling = ceiling
        self.floor = floor
        self.decrease = decrease
        self.increase = increase

    async def current(self, endpoint_id: str) -> float:
        raw = await self.redis.get(f"{RATE_PREFIX}{endpoint_id}")
        return float(raw) if raw else self.ceiling

    async def in_cooldown(self, endpoint_id: str) -> int:
        ttl = await self.redis.ttl(f"{COOLDOWN_PREFIX}{endpoint_id}")
        return max(0, ttl)

    async def on_throttled(self, endpoint_id: str, retry_after: str | None) -> float:
        seconds = 1
        if retry_after and retry_after.isdigit():
            seconds = max(1, min(int(retry_after), 300))
        await self.redis.setex(f"{COOLDOWN_PREFIX}{endpoint_id}", seconds, "1")
        new_rate = max(self.floor, await self.current(endpoint_id) * self.decrease)
        await self.redis.set(f"{RATE_PREFIX}{endpoint_id}", new_rate, ex=3600)
        return new_rate

    async def on_success_window(self, endpoint_id: str) -> float:
        """Call from a periodic task, e.g. once a minute per active endpoint."""
        if await self.in_cooldown(endpoint_id):
            return await self.current(endpoint_id)
        new_rate = min(self.ceiling, await self.current(endpoint_id) * self.increase)
        await self.redis.set(f"{RATE_PREFIX}{endpoint_id}", new_rate, ex=3600)
        return new_rate

Wire it into the dispatcher so the effective config is read per delivery:

    async def deliver_adaptive(
        self, endpoint_id: str, url: str, body: bytes, headers: dict[str, str],
        adaptive: AdaptiveRate, capacity: float,
    ) -> httpx.Response:
        cooldown = await adaptive.in_cooldown(endpoint_id)
        if cooldown:
            raise TimeoutError(f"{endpoint_id} in cooldown for {cooldown}s")
        cfg = BucketConfig(rate=await adaptive.current(endpoint_id), capacity=capacity)
        await self.limiter.acquire(endpoint_id, cfg, max_wait=5.0)
        response = await self.client.post(url, content=body, headers=headers, timeout=10.0)
        if response.status_code == 429:
            await adaptive.on_throttled(endpoint_id, response.headers.get("Retry-After"))
        return response

The rate floor is not optional. Without it, an endpoint that returns 429 for an unrelated reason — a misconfigured WAF rule, say — halves its way to a rate of 0.0001 tokens per second and effectively stops receiving events forever, with no error anywhere to explain it. Keep the floor high enough that a wrongly-throttled endpoint still trickles, and alert when any endpoint’s rate sits at the floor for more than a few minutes. Persistent throttling is also a legitimate breaker input; feed it into per-endpoint circuit breaker state machines so a consumer that is throttling everything gets stopped rather than trickled.

Verification and testing

The property to assert is not “requests were limited” but “the observed rate over a window never exceeded the configured rate, no matter how many concurrent callers there were”.

import asyncio
import time

import fakeredis.aioredis
import pytest


@pytest.mark.asyncio
async def test_burst_is_bounded_by_capacity():
    redis = fakeredis.aioredis.FakeRedis()
    limiter = TokenBucketLimiter(redis)
    cfg = BucketConfig(rate=10.0, capacity=5.0)
    granted = 0
    for _ in range(20):                     # ask 20 times with no delay
        ok, _rem, _wait = await limiter.try_acquire("ep_burst", cfg)
        granted += int(ok)
    assert granted <= 6, f"burst leaked: {granted} tokens from a capacity of 5"


@pytest.mark.asyncio
async def test_sustained_rate_holds_under_concurrency():
    redis = fakeredis.aioredis.FakeRedis()
    limiter = TokenBucketLimiter(redis)
    cfg = BucketConfig(rate=20.0, capacity=20.0)
    started = time.monotonic()

    async def worker() -> int:
        sent = 0
        while time.monotonic() - started < 2.0:
            ok, _rem, _wait = await limiter.try_acquire("ep_rate", cfg)
            if ok:
                sent += 1
            await asyncio.sleep(0.005)
        return sent

    counts = await asyncio.gather(*[worker() for _ in range(8)])
    total = sum(counts)
    # 2 seconds at 20/s plus a full 20-token bucket = 60 ceiling, allow slack.
    assert total <= 66, f"eight workers overdrew the shared bucket: {total}"


@pytest.mark.asyncio
async def test_429_halves_the_rate_and_sets_cooldown():
    redis = fakeredis.aioredis.FakeRedis()
    adaptive = AdaptiveRate(redis, ceiling=100.0, floor=1.0)
    assert await adaptive.current("ep_1") == 100.0
    new_rate = await adaptive.on_throttled("ep_1", retry_after="30")
    assert new_rate == 50.0
    assert 0 < await adaptive.in_cooldown("ep_1") <= 30


@pytest.mark.asyncio
async def test_rate_never_falls_below_the_floor():
    redis = fakeredis.aioredis.FakeRedis()
    adaptive = AdaptiveRate(redis, ceiling=100.0, floor=2.0)
    for _ in range(20):
        await adaptive.on_throttled("ep_2", retry_after="1")
    assert await adaptive.current("ep_2") == 2.0

Against a live Redis, confirm the bucket is doing arithmetic you recognise:

# Watch a bucket drain and refill while the dispatcher runs.
watch -n 1 'redis-cli HGETALL wh:bucket:ep_42'

# Verify the script is cached and being called via EVALSHA, not shipped each time.
redis-cli INFO commandstats | grep -E 'cmdstat_(eval|evalsha)'

If cmdstat_eval grows in step with cmdstat_evalsha, something is re-registering the script per request — usually a limiter constructed inside the request handler rather than once at startup.

Failure modes and gotchas

• Non-atomic buckets that look correct in tests. A GET/compare/SET implementation passes every single-process test and overdraws immediately under concurrency, because the race window is microseconds wide and only opens under real parallelism. If you cannot use Lua, WATCH/MULTI/EXEC with a retry loop is the only safe alternative — but expect contention aborts at high rates. Test with at least eight concurrent callers, as in the second test above.
• Client wall-clock used for refill. Passing time.time() from each dispatcher into the script lets a pod with a fast clock mint free tokens and a pod with a slow clock starve. Every replica must read the same clock; redis.call('TIME') inside the script is the cheapest way to guarantee that.
• One global bucket instead of one per endpoint. A single shared bucket means a busy consumer’s traffic consumes the allowance of every other consumer, and one slow endpoint throttles deliveries to healthy ones. Key by endpoint (or by whatever unit the consumer’s limit is actually enforced on) and accept the extra keys — they expire on their own.
• Waiters polling in lockstep. If every blocked delivery sleeps the exact returned wait, they all wake at the same instant, hit the script simultaneously, and one wins while the rest recompute the same deadline. Add a small random component to the sleep, and cap the sleep so a long deficit is re-evaluated against a rate that may have changed under it.
• Rate limiting applied after the retry decision. If a token timeout is counted as a failed attempt, a busy endpoint exhausts each event’s attempt budget without ever sending a request. Acquire the token first; only a request that actually left the process may increment the attempt counter.

Frequently Asked Questions

Related

• Applying Backpressure to Webhook Consumers — the receiving side of the same conversation.
• Choosing Retry Budgets and Max Attempts — why a token timeout must not consume an attempt.
• Per-Endpoint Circuit Breaker State Machines — stopping traffic entirely when throttling is no longer enough.
• Webhook Rate Limiting & Backpressure — the parent overview covering limiter choice and feedback design.