How It Works

Sliding window counter

Instead of a log, keep just two counters: the previous window and the current one. Assume the previous window's requests were spread evenly, and count only the fraction of it that still overlaps the last 60 seconds.

estimate = previous count × (1 - time into current window / window length) + current count

With a limit of "10 per 60 seconds", 8 requests in the previous window (0 to 60 s) and 3 so far in the current one (60 to 120 s):

As time passes the previous window fades out. If its traffic was bunched at one end the estimate is slightly off, but each key needs only two numbers.

Token bucket

Tokens flow into the bucket at rate per second, up to capacity. Each request takes one token and is rejected when fewer than one remains. No timer is needed: when a request arrives, add "time since last check × rate" in one go (lazy refill).

Let us trace rate = 1 per second and capacity = 3.

The bucket starts full, so a burst of three passes within 0.4 seconds, after which the client is held to one request per second. After a quiet period the bucket refills and absorbs the next burst.

def token_bucket(times: list[float], rate: float, capacity: float) -> list[bool]:
    tokens, last = capacity, times[0] if times else 0.0
    result = []
    for now in times:
        tokens = min(capacity, tokens + (now - last) * rate)   # lazy refill
        last = now
        if tokens >= 1:
            tokens -= 1
            result.append(True)
        else:
            result.append(False)
    return result


print(token_bucket([0, 0.2, 0.4, 0.6, 1.5, 3.5, 3.6, 3.7], rate=1, capacity=3))
# [True, True, True, False, True, True, True, False]

Leaky bucket

Picture a bucket with a hole in the bottom. Incoming requests are placed in a fixed-size queue, and a worker takes them out at a constant rate. When the queue is full, new requests are dropped. With a queue of 3 and one request processed per second, five requests arriving together at time 0 result in three queued and two rejected; the queued ones are handled at 1, 2 and 3 seconds.

from collections import deque


def leaky_bucket(arrivals: list[int], queue_size: int, leak_per_tick: int) -> None:
    queue: deque[int] = deque()
    for tick, count in enumerate(arrivals):
        for _ in range(leak_per_tick):           # drain at a constant rate first
            if queue:
                print(f"t={tick}: processed request {queue.popleft()}")
        for i in range(count):                   # accept new work only if there is room
            if len(queue) < queue_size:
                queue.append(tick * 100 + i)
            else:
                print(f"t={tick}: rejected request {tick * 100 + i}")


leaky_bucket([5, 0, 0, 0], queue_size=3, leak_per_tick=1)

Unlike a token bucket, output never exceeds the rate, which makes it a good fit for shielding a downstream system that needs a steady pace. The cost is latency while requests wait in the queue. Note that the "meter" form of the leaky bucket, which measures without queueing, makes exactly the same decisions as a token bucket.

Summary

• A fixed window is simple, but up to twice the limit can pass around a boundary.
• A sliding window log is exact; a sliding window counter approximates it with two numbers.
• A token bucket enforces an average rate while allowing bursts up to its capacity.
• A leaky bucket smooths output with a queue at the cost of waiting time.