Implement a sliding-window rate limiter and explain its memory cost.

“Log first, because it’s exact and it becomes my test oracle.”

import time
from collections import deque
from typing import Callable

Clock = Callable[[], float]


class SlidingLogLimiter:
    """At most `limit` allowed requests per key
    in any window (now - window, now]."""

    def __init__(self, limit: int, window: float,
                 clock: Clock = time.monotonic):
        if limit < 1 or window <= 0:
            raise ValueError("need limit >= 1, window > 0")
        self.limit, self.window = limit, window
        self.clock = clock
        self._logs: dict[str, deque[float]] = {}

    def allow(self, key: str) -> bool:
        now = self.clock()
        log = self._logs.setdefault(key, deque())
        cutoff = now - self.window
        # evict what has aged out
        while log and log[0] <= cutoff:
            log.popleft()
        if len(log) < self.limit:
            # only allowed requests are logged
            log.append(now)
            return True
        return False

    def sweep(self) -> None:
        """Drop keys with nothing left in the window."""
        cutoff = self.clock() - self.window
        idle = [k for k, log in self._logs.items()
                if log[-1] <= cutoff]
        for key in idle:
            del self._logs[key]


def test_boundary_is_half_open():
    t = [0.0]
    lim = SlidingLogLimiter(limit=3, window=10,
                            clock=lambda: t[0])
    got = [lim.allow("k") for _ in range(4)]
    assert got == [True, True, True, False]
    t[0] = 9.999
    assert not lim.allow("k")
    t[0] = 10.0
    # the t=0 entries are outside (0, 10]
    assert lim.allow("k")

“Time per call is amortized constant, since each timestamp is appended once and popped once. Memory is the answer they asked for. Each entry is an 8 B pointer in the deque’s block plus a 24 B float object, so about 32 B, and an empty deque is about 760 B of fixed overhead. So the worst case is keys * limit * 32 B: 50_000 keys * 1_000 limit is about 1.6 GB. Without sweep, keys that went quiet stay forever.”

“In one process with threads, I’d take a lock per key around evict, check and append, because the check and the append must be one step. It’s the same rule as the Redis script further down.”

“If the limit is small, I keep the log. If it’s large, I switch to the counter, which stores three integers per key whatever the limit. In Python 3.13, a tuple of three integers plus its dict slot measures about 200 B per key once the counts pass 256, so the same 50_000 keys at a limit of 1_000 cost about 10 MB against the log’s 1.6 GB. It still needs the same sweep for keys that went quiet; what no longer grows is the per-key cost.”

class SlidingCounterLimiter:
    def __init__(self, limit: int, window: float,
                 clock: Clock = time.monotonic):
        if limit < 1 or window <= 0:
            raise ValueError("need limit >= 1, window > 0")
        self.limit, self.window = limit, window
        self.clock = clock
        # key -> (window index, current, previous)
        self._state: dict[str, tuple[int, int, int]] = {}

    def allow(self, key: str) -> bool:
        now = self.clock()
        idx = int(now // self.window)
        w, cur, prev = self._state.get(key, (idx, 0, 0))
        if idx != w:
            # a skipped window means prev has aged out
            prev = cur if idx == w + 1 else 0
            cur, w = 0, idx
        overlap = 1 - (now % self.window) / self.window
        allowed = prev * overlap + cur + 1 <= self.limit
        if allowed:
            cur += 1
        self._state[key] = (w, cur, prev)
        return allowed

“The trade-off, in behavior, with limit=10 and window=60. Say every request of the previous window landed in its last second. Halfway through the current window the estimate is 5, so 5 more pass, and the true rolling window now holds 15. Now say they all landed in its first second instead. A second into the current window the estimate is still about 9.8, so the next request is refused, although the true rolling window is empty. So the counter can over-admit or refuse early; what it buys is memory that never grows with the limit.”

“The log is the oracle for that claim. I drive both limiters with the same timestamps and print where they disagree; the two bursts above show up at once, and random timestamps find others:”

def disagreements(times: list[float], limit: int = 10,
                  window: float = 60.0):
    t = [0.0]

    def clock() -> float:
        return t[0]

    log = SlidingLogLimiter(limit, window, clock)
    counter = SlidingCounterLimiter(limit, window, clock)
    out = []
    for now in sorted(times):
        t[0] = now
        exact, approx = log.allow("k"), counter.allow("k")
        if exact != approx:
            # record which one allowed it
            winner = "log" if exact else "counter"
            out.append((now, winner))
    return out

“There is a middle option. Keep a count per sub-window, say one-second buckets for window=60, kept in a ring. Memory is window / granularity integers per key whatever the limit, and the error is limited to the requests in the one bucket at the trailing edge. If the enterprise limit rises and the counter’s error is too loose, that is what I would switch to.”

“On refused requests: they stay out of the log. If they counted, a client retrying in a tight loop would keep its own window full and never get back in. Keeping them out also gives the log an exact Retry-After for free. When a request is refused, the oldest entry leaves the window at log[0] + window, so the wait is log[0] + window - now. The counter can only estimate it.”

“For a model API, the unit is tokens as well as requests: OpenAI’s rate limits guide measures tokens per minute beside requests per minute. Then the log stores (timestamp, cost) pairs and a running total per key. Subtract each entry’s cost as it ages out, and allow while total + cost <= limit. Retry-After becomes the time until enough cost has aged out, found by walking the log from the front.”

“Across nodes: the log becomes a Redis sorted set scored by timestamp, with ZREMRANGEBYSCORE, ZCARD and ZADD in one atomic script. Members need a unique suffix, because two requests in the same microsecond would otherwise collapse into one. Above 128 entries (the default zset-max-listpack-entries), Redis stores a sorted set as a skiplist plus a hash table, which costs more per entry than a Python deque. Before choosing, I’d measure rather than guess: fill one key to the enterprise limit, run MEMORY USAGE on it with SAMPLES 0 so it counts every entry rather than a sample, and multiply by the active keys. The counter becomes two keys, rl:{key}:{idx} and rl:{key}:{idx-1}. One Lua script reads both, computes the weighted estimate, and INCRs the current key with an EXPIRE of two windows only when the request is allowed. The check and the increment must be one step, or two nodes both take the last slot. Both scripts read the time from Redis TIME, so the nodes’ clocks do not matter.”