Python Development Specialist

Thread-safe sliding-window-log rate limiter

1. Task restatement and assumptions

The goal is a standard-library-only, thread-safe, in-memory SlidingWindowRateLimiter. It records a timestamp log for each client and permits at most max_requests calls in any trailing window_seconds. It also needs quota inspection (remaining, reset_in) and idle-client eviction (purge_idle).

Assumptions:

2. Edge cases

3. Implementation — rate_limiter.py

"""Thread-safe, in-memory sliding-window-log rate limiter.

The limiter keeps, per client, a log (``deque``) of the monotonic timestamps of
*accepted* requests. A request is accepted iff fewer than ``max_requests``
timestamps fall inside the trailing ``window_seconds``.

Concurrency model:
    * One ``threading.Lock`` per client guards that client's log, so unrelated
      clients never contend with each other.
    * A single registry lock is taken only to create a new client entry or to
      purge idle ones; the hot path for an existing client never touches it.

asyncio:
    Every operation is pure in-memory bookkeeping with short, non-I/O critical
    sections, so it is safe to call directly from a single asyncio event-loop
    thread (no ``await`` / executor needed). Lock hold times are microseconds.

Example (Flask):
    limiter = SlidingWindowRateLimiter(max_requests=100, window_seconds=1.0)

    @app.before_request
    def throttle() -> Response | None:
        key = request.headers.get("X-API-Key") or request.remote_addr or "anon"
        if not limiter.allow(key):
            retry = limiter.reset_in(key)
            return Response("Too Many Requests", 429,
                            {"Retry-After": f"{math.ceil(retry)}"})
        return None
"""

from __future__ import annotations

import math
import threading
import time
from collections import deque
from collections.abc import Callable
from dataclasses import dataclass, field

__all__ = ["SlidingWindowRateLimiter", "IDLE_WINDOW_MULTIPLIER"]

#: A client idle for more than this many windows is eligible for purging.
IDLE_WINDOW_MULTIPLIER: int = 10


@dataclass(slots=True)
class _ClientState:
    """Mutable per-client state. All fields are guarded by ``lock``.

    Attributes:
        timestamps: Sorted (non-decreasing) accepted-request timestamps.
        lock: Per-client mutex.
        last_seen: Timestamp of the most recent ``allow()`` call.
        evicted: Set by ``purge_idle`` when this object is removed from the
            registry; any thread still holding a reference must retry.
    """

    timestamps: deque[float]
    last_seen: float
    lock: threading.Lock = field(default_factory=threading.Lock)
    evicted: bool = False


class SlidingWindowRateLimiter:
    """Per-client sliding-window-log rate limiter.

    Complexity:
        ``allow``, ``remaining``, ``reset_in``: amortized O(1) time — each
        timestamp is appended once and popped from the left at most once.
        Space: O(C * max_requests) for C active clients.
        ``purge_idle``: O(C).
    """

    __slots__ = ("_max_requests", "_window", "_idle_timeout", "_clock",
                 "_clients", "_registry_lock")

    def __init__(
        self,
        max_requests: int,
        window_seconds: float,
        *,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """Create a rate limiter.

        Args:
            max_requests: Maximum accepted calls per client per window (> 0).
            window_seconds: Length of the trailing window in seconds (> 0,
                finite). ``int`` values are accepted and converted to float.
            clock: Zero-arg callable returning monotonic seconds. Override
                only for testing/simulation.

        Raises:
            ValueError: If ``max_requests`` is not a positive int or
                ``window_seconds`` is not a positive finite number.
        """
        self._max_requests: int = _validate_max_requests(max_requests)
        self._window: float = _validate_window(window_seconds)
        self._idle_timeout: float = IDLE_WINDOW_MULTIPLIER * self._window
        self._clock: Callable[[], float] = clock
        self._clients: dict[str, _ClientState] = {}
        self._registry_lock = threading.Lock()

    # ------------------------------------------------------------------ props
    @property
    def max_requests(self) -> int:
        """int: Maximum accepted calls per window."""
        return self._max_requests

    @property
    def window_seconds(self) -> float:
        """float: Window length in seconds."""
        return self._window

    def __len__(self) -> int:
        """Return the number of clients currently tracked."""
        return len(self._clients)

    # ----------------------------------------------------------------- public
    def allow(self, client_id: str, *, now: float | None = None) -> bool:
        """Attempt to consume one request slot for ``client_id``.

        Args:
            client_id: Non-empty identifier of the caller (API key, user id).
            now: Optional timestamp override (same clock as ``clock``).

        Returns:
            ``True`` if the call is permitted (its timestamp is recorded),
            ``False`` if the client already has ``max_requests`` calls in the
            trailing window (nothing is recorded).

        Raises:
            TypeError: If ``client_id`` is not a non-empty ``str`` or ``now``
                is not a real number.
            ValueError: If ``now`` is NaN or infinite.
        """
        _validate_client_id(client_id)
        t = self._resolve_now(now)

        while True:
            state = self._get_or_create(client_id, t)
            with state.lock:
                if state.evicted:
                    # Purged between lookup and lock acquisition; retry so we
                    # never write into an orphaned state object.
                    continue
                ts = state.timestamps
                if ts and t < ts[-1]:
                    t = ts[-1]  # Clamp to keep the log sorted (A7).
                if t > state.last_seen:
                    state.last_seen = t
                self._evict_expired(ts, t)
                if len(ts) >= self._max_requests:
                    return False
                ts.append(t)
                return True

    def remaining(self, client_id: str, *, now: float | None = None) -> int:
        """Return how many more calls ``client_id`` may make right now.

        Read-only with respect to client activity: does not create state for
        unknown clients and does not refresh idle status.

        Args:
            client_id: Non-empty identifier of the caller.
            now: Optional timestamp override.

        Returns:
            Calls still allowed in the current window, in
            ``[0, max_requests]``.

        Raises:
            TypeError: If ``client_id`` is not a non-empty ``str`` or ``now``
                is not a real number.
            ValueError: If ``now`` is NaN or infinite.
        """
        _validate_client_id(client_id)
        count, _ = self._snapshot(client_id, self._resolve_now(now))
        return max(0, self._max_requests - count)

    def reset_in(self, client_id: str, *, now: float | None = None) -> float:
        """Return seconds until at least one slot frees up for ``client_id``.

        Args:
            client_id: Non-empty identifier of the caller.
            now: Optional timestamp override.

        Returns:
            ``0.0`` if the client is not currently exhausted, otherwise the
            (positive) seconds until its oldest logged request leaves the
            window. Suitable for an HTTP ``Retry-After`` header.

        Raises:
            TypeError: If ``client_id`` is not a non-empty ``str`` or ``now``
                is not a real number.
            ValueError: If ``now`` is NaN or infinite.
        """
        _validate_client_id(client_id)
        t = self._resolve_now(now)
        count, oldest = self._snapshot(client_id, t)
        if count < self._max_requests or oldest is None:
            return 0.0
        return max(0.0, oldest + self._window - t)

    def purge_idle(self, now: float | None = None) -> int:
        """Evict clients idle for more than ``10 * window_seconds``.

        Safe to call from a background thread or opportunistically from
        request handlers. Clients whose lock is currently held are, by
        definition, active and are skipped without blocking.

        Args:
            now: Optional timestamp override.

        Returns:
            Number of clients purged.

        Raises:
            TypeError: If ``now`` is not a real number.
            ValueError: If ``now`` is NaN or infinite.
        """
        t = self._resolve_now(now)
        purged = 0
        with self._registry_lock:
            for client_id, state in list(self._clients.items()):
                if not state.lock.acquire(blocking=False):
                    continue
                try:
                    if t - state.last_seen > self._idle_timeout:
                        state.evicted = True
                        del self._clients[client_id]
                        purged += 1
                finally:
                    state.lock.release()
        return purged

    # ---------------------------------------------------------------- private
    def _resolve_now(self, now: float | None) -> float:
        """Return ``now`` validated as a finite float, or the clock value."""
        if now is None:
            return float(self._clock())
        if isinstance(now, bool) or not isinstance(now, (int, float)):
            raise TypeError(
                f"now must be a real number or None, got {type(now).__name__}"
            )
        if not math.isfinite(now):
            raise ValueError(f"now must be finite, got {now!r}")
        return float(now)

    def _get_or_create(self, client_id: str, now: float) -> _ClientState:
        """Return the state for ``client_id``, creating it if absent.

        Lock-free fast path for existing clients; double-checked creation
        under the registry lock otherwise.
        """
        state = self._clients.get(client_id)
        if state is not None:
            return state
        with self._registry_lock:
            state = self._clients.get(client_id)
            if state is None:
                state = _ClientState(
                    timestamps=deque(maxlen=self._max_requests),
                    last_seen=now,
                )
                self._clients[client_id] = state
            return state

    def _snapshot(self, client_id: str, now: float) -> tuple[int, float | None]:
        """Return ``(live_count, oldest_live_timestamp)`` without creating state."""
        state = self._clients.get(client_id)
        if state is None:
            return 0, None
        with state.lock:
            if state.evicted:
                return 0, None
            ts = state.timestamps
            self._evict_expired(ts, now)
            return len(ts), (ts[0] if ts else None)

    def _evict_expired(self, timestamps: deque[float], now: float) -> None:
        """Pop timestamps that have left the window. Caller holds the lock."""
        cutoff = now - self._window
        while timestamps and timestamps[0] <= cutoff:
            timestamps.popleft()


# -------------------------------------------------------------- validation
def _validate_max_requests(value: object) -> int:
    """Ensure ``value`` is a positive ``int`` (not ``bool``)."""
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ValueError(f"max_requests must be a positive int, got {value!r}")
    return value


def _validate_window(value: object) -> float:
    """Ensure ``value`` is a positive, finite real number (not ``bool``)."""
    if (
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not math.isfinite(value)
        or value <= 0
    ):
        raise ValueError(
            f"window_seconds must be a positive finite number, got {value!r}"
        )
    return float(value)


def _validate_client_id(client_id: object) -> None:
    """Ensure ``client_id`` is a non-empty ``str``."""
    if not isinstance(client_id, str):
        raise TypeError(
            f"client_id must be a non-empty str, got {type(client_id).__name__}"
        )
    if not client_id:
        raise TypeError("client_id must be a non-empty str, got empty string")


# ------------------------------------------------------------------ demo
if __name__ == "__main__":
    import random

    random.seed(42)
    limiter = SlidingWindowRateLimiter(max_requests=5, window_seconds=1.0)
    api_keys = ["key_alice", "key_bob", "key_mallory"]

    # Simulated 3-second traffic trace: mallory is abusive (~20 rps),
    # alice moderate (~4 rps), bob bursty then silent.
    events: list[tuple[float, str]] = []
    for key, rps in (("key_alice", 4), ("key_mallory", 20)):
        events += [(random.uniform(0, 3), key) for _ in range(rps * 3)]
    events += [(0.10 + i * 0.01, "key_bob") for i in range(8)]
    events.sort()

    stats: dict[str, list[int]] = {k: [0, 0] for k in api_keys}
    for ts, key in events:
        stats[key][0 if limiter.allow(key, now=ts) else 1] += 1

    print(f"{'client':<12} {'allowed':>8} {'denied':>7}")
    for key, (ok, denied) in stats.items():
        print(f"{key:<12} {ok:>8} {denied:>7}")

    t_end = 3.0
    for key in api_keys:
        print(f"{key}: remaining={limiter.remaining(key, now=t_end)} "
              f"reset_in={limiter.reset_in(key, now=t_end):.3f}s")

    # Bob's last call was at ~0.17s; well past 10 windows later he is purged.
    print(f"tracked before purge: {len(limiter)}")
    print(f"purged at t=12.0: {limiter.purge_idle(now=12.0)}  "
          f"(tracked now: {len(limiter)})")

    # Background purger pattern for long-running servers:
    #   stop = threading.Event()
    #   def _janitor() -> None:
    #       while not stop.wait(limiter.window_seconds * 10):
    #           limiter.purge_idle()
    #   threading.Thread(target=_janitor, daemon=True).start()

4. Tests — test_rate_limiter.py

"""pytest suite for SlidingWindowRateLimiter."""

from __future__ import annotations

import math
import threading
import time
from collections.abc import Callable

import pytest

from rate_limiter import SlidingWindowRateLimiter


@pytest.fixture
def limiter() -> SlidingWindowRateLimiter:
    return SlidingWindowRateLimiter(max_requests=3, window_seconds=1.0)


# --------------------------------------------------------- constructor args
@pytest.mark.parametrize("bad", [0, -1, 1.5, True, "3", None])
def test_invalid_max_requests_raises_value_error(bad: object) -> None:
    with pytest.raises(ValueError, match="max_requests"):
        SlidingWindowRateLimiter(max_requests=bad, window_seconds=1.0)  # type: ignore[arg-type]


@pytest.mark.parametrize("bad", [0, 0.0, -0.5, math.nan, math.inf, True, "1", None])
def test_invalid_window_seconds_raises_value_error(bad: object) -> None:
    with pytest.raises(ValueError, match="window_seconds"):
        SlidingWindowRateLimiter(max_requests=1, window_seconds=bad)  # type: ignore[arg-type]


def test_int_window_is_accepted_and_coerced_to_float() -> None:
    rl = SlidingWindowRateLimiter(max_requests=1, window_seconds=2)
    assert rl.window_seconds == 2.0 and isinstance(rl.window_seconds, float)


# ------------------------------------------------------------- client_id
@pytest.mark.parametrize("bad", [None, 123, b"key", "", ["k"]])
@pytest.mark.parametrize("method", ["allow", "remaining", "reset_in"])
def test_invalid_client_id_raises_type_error(
    limiter: SlidingWindowRateLimiter, method: str, bad: object
) -> None:
    with pytest.raises(TypeError, match="client_id"):
        getattr(limiter, method)(bad, now=0.0)


@pytest.mark.parametrize("bad", ["1.0", True, object()])
def test_non_numeric_now_raises_type_error(
    limiter: SlidingWindowRateLimiter, bad: object
) -> None:
    with pytest.raises(TypeError, match="now"):
        limiter.allow("a", now=bad)  # type: ignore[arg-type]


@pytest.mark.parametrize("bad", [math.nan, math.inf])
def test_non_finite_now_raises_value_error(
    limiter: SlidingWindowRateLimiter, bad: float
) -> None:
    with pytest.raises(ValueError, match="finite"):
        limiter.allow("a", now=bad)


# ------------------------------------------------------------- happy path
def test_allows_up_to_max_then_denies(limiter: SlidingWindowRateLimiter) -> None:
    results = [limiter.allow("a", now=0.1 * i) for i in range(5)]
    assert results == [True, True, True, False, False]


def test_window_slides_and_frees_slots(limiter: SlidingWindowRateLimiter) -> None:
    for t in (0.0, 0.2, 0.4):
        assert limiter.allow("a", now=t)
    assert not limiter.allow("a", now=0.99)
    assert limiter.allow("a", now=1.0)       # 0.0 expired exactly at boundary
    assert not limiter.allow("a", now=1.1)   # 0.2, 0.4, 1.0 still live
    assert limiter.allow("a", now=1.2)       # 0.2 expired


def test_denied_calls_do_not_consume_quota(limiter: SlidingWindowRateLimiter) -> None:
    for _ in range(3):
        limiter.allow("a", now=0.0)
    for _ in range(50):
        assert not limiter.allow("a", now=0.5)
    assert limiter.remaining("a", now=1.0) == 3


def test_clients_are_isolated(limiter: SlidingWindowRateLimiter) -> None:
    for _ in range(3):
        limiter.allow("alice", now=0.0)
    assert not limiter.allow("alice", now=0.0)
    assert limiter.allow("bob", now=0.0)
    assert limiter.remaining("bob", now=0.0) == 2


# ------------------------------------------------------- remaining/reset_in
def test_remaining_for_unknown_client_is_max_and_creates_no_state(
    limiter: SlidingWindowRateLimiter,
) -> None:
    assert limiter.remaining("ghost", now=0.0) == 3
    assert limiter.reset_in("ghost", now=0.0) == 0.0
    assert len(limiter) == 0


def test_remaining_decrements_and_floors_at_zero(limiter: SlidingWindowRateLimiter) -> None:
    seen = []
    for _ in range(5):
        limiter.allow("a", now=0.0)
        seen.append(limiter.remaining("a", now=0.0))
    assert seen == [2, 1, 0, 0, 0]


def test_reset_in_is_zero_when_not_exhausted(limiter: SlidingWindowRateLimiter) -> None:
    limiter.allow("a", now=0.0)
    assert limiter.reset_in("a", now=0.0) == 0.0


def test_reset_in_reports_time_until_oldest_expires(
    limiter: SlidingWindowRateLimiter,
) -> None:
    for t in (0.0, 0.3, 0.6):
        limiter.allow("a", now=t)
    assert limiter.reset_in("a", now=0.75) == pytest.approx(0.25)
    assert limiter.reset_in("a", now=1.0) == 0.0  # slot freed at boundary
    assert limiter.allow("a", now=1.0)


def test_retrying_after_reset_in_succeeds(limiter: SlidingWindowRateLimiter) -> None:
    for _ in range(3):
        limiter.allow("a", now=10.0)
    wait = limiter.reset_in("a", now=10.4)
    assert not limiter.allow("a", now=10.4)
    assert limiter.allow("a", now=10.4 + wait)


# --------------------------------------------------------------- memory
def test_log_never_exceeds_max_requests() -> None:
    rl = SlidingWindowRateLimiter(max_requests=100, window_seconds=1.0)
    for i in range(10_000):
        rl.allow("a", now=i * 0.0001)
    assert len(rl._clients["a"].timestamps) <= 100


def test_backwards_time_is_clamped_and_log_stays_sorted(
    limiter: SlidingWindowRateLimiter,
) -> None:
    limiter.allow("a", now=5.0)
    assert limiter.allow("a", now=1.0)  # treated as t=5.0
    ts = list(limiter._clients["a"].timestamps)
    assert ts == sorted(ts) == [5.0, 5.0]


# ---------------------------------------------------------------- purge
def test_purge_idle_evicts_only_idle_clients(limiter: SlidingWindowRateLimiter) -> None:
    limiter.allow("idle", now=0.0)
    limiter.allow("active", now=9.5)
    assert limiter.purge_idle(now=10.5) == 1
    assert len(limiter) == 1
    assert limiter.remaining("idle", now=10.5) == 3


def test_purge_boundary_is_strictly_greater(limiter: SlidingWindowRateLimiter) -> None:
    limiter.allow("a", now=0.0)
    assert limiter.purge_idle(now=10.0) == 0     # exactly 10 windows: kept
    assert limiter.purge_idle(now=10.0001) == 1


def test_denied_calls_count_as_activity(limiter: SlidingWindowRateLimiter) -> None:
    for _ in range(3):
        limiter.allow("a", now=0.0)
    limiter.allow("a", now=0.5)  # denied but active
    assert limiter.purge_idle(now=10.2) == 0


def test_queries_do_not_count_as_activity(limiter: SlidingWindowRateLimiter) -> None:
    limiter.allow("a", now=0.0)
    limiter.remaining("a", now=9.9)
    limiter.reset_in("a", now=9.9)
    assert limiter.purge_idle(now=10.5) == 1


def test_purged_client_starts_fresh(limiter: SlidingWindowRateLimiter) -> None:
    for _ in range(3):
        limiter.allow("a", now=0.0)
    limiter.purge_idle(now=100.0)
    assert limiter.allow("a", now=100.0)
    assert limiter.remaining("a", now=100.0) == 2


def test_purge_on_empty_limiter_returns_zero(limiter: SlidingWindowRateLimiter) -> None:
    assert limiter.purge_idle(now=0.0) == 0


def test_injected_clock_is_used_by_default() -> None:
    fake_now = [0.0]
    clock: Callable[[], float] = lambda: fake_now[0]
    rl = SlidingWindowRateLimiter(max_requests=1, window_seconds=1.0, clock=clock)
    assert rl.allow("a")
    assert not rl.allow("a")
    fake_now[0] = 1.0
    assert rl.allow("a")


# ----------------------------------------------------------- concurrency
def _run_threads(n: int, target: Callable[[int], None]) -> None:
    barrier = threading.Barrier(n)
    errors: list[BaseException] = []

    def wrapper(idx: int) -> None:
        barrier.wait()
        try:
            target(idx)
        except BaseException as exc:  # surface failures to the main thread
            errors.append(exc)

    threads = [threading.Thread(target=wrapper, args=(i,)) for i in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert not errors, errors


def test_concurrent_single_client_admits_exactly_max_requests() -> None:
    rl = SlidingWindowRateLimiter(max_requests=100, window_seconds=1.0)
    allowed = 0
    counter_lock = threading.Lock()

    def worker(_: int) -> None:
        nonlocal allowed
        local = sum(rl.allow("shared", now=0.5) for _ in range(50))
        with counter_lock:
            allowed += local

    _run_threads(24, worker)  # 1,200 attempts
    assert allowed == 100
    assert rl.remaining("shared", now=0.5) == 0


def test_concurrent_many_clients_each_get_exact_quota() -> None:
    rl = SlidingWindowRateLimiter(max_requests=10, window_seconds=1.0)
    clients = [f"client-{i}" for i in range(200)]
    results: dict[str, int] = {c: 0 for c in clients}
    lock = threading.Lock()

    def worker(_: int) -> None:
        for c in clients:
            ok = sum(rl.allow(c, now=0.0) for _ in range(3))
            with lock:
                results[c] += ok

    _run_threads(20, worker)  # 60 attempts per client
    assert all(v == 10 for v in results.values())
    assert len(rl) == 200


def test_concurrent_purge_and_allow_do_not_corrupt_state() -> None:
    rl = SlidingWindowRateLimiter(max_requests=5, window_seconds=0.01)
    stop = threading.Event()

    def purger() -> None:
        while not stop.is_set():
            rl.purge_idle(now=1e9)  # everything looks idle

    p = threading.Thread(target=purger)
    p.start()
    try:
        _run_threads(20, lambda i: [rl.allow(f"c{j % 7}", now=float(j))
                                    for j in range(2_000)])
    finally:
        stop.set()
        p.join()
    for state in rl._clients.values():
        assert len(state.timestamps) <= 5
        assert not state.evicted


# ------------------------------------------------------------- benchmark
def test_benchmark_100k_allow_calls_single_client_under_one_second() -> None:
    rl = SlidingWindowRateLimiter(max_requests=100, window_seconds=1.0)
    start = time.perf_counter()
    for _ in range(100_000):
        rl.allow("bench")
    elapsed = time.perf_counter() - start
    assert elapsed < 1.0, f"100k allow() calls took {elapsed:.3f}s"


def test_benchmark_with_sliding_evictions_under_one_second() -> None:
    # 1,000 rps simulated against max=100/1s → constant left-pops.
    rl = SlidingWindowRateLimiter(max_requests=100, window_seconds=1.0)
    start = time.perf_counter()
    allowed = sum(rl.allow("bench", now=i * 0.001) for i in range(100_000))
    elapsed = time.perf_counter() - start
    assert elapsed < 1.0, f"100k allow() calls took {elapsed:.3f}s"
    assert allowed == pytest.approx(10_000, abs=100)  # ~100 per simulated second

Run with:

pip install pytest            # test-only dependency
pytest -q test_rate_limiter.py
python rate_limiter.py        # demo

Dependencies: the runtime uses only the standard library, so no requirements.txt is needed. pytest is required for the tests only.

5. Design decisions and complexity

• Per-client locks plus a registry lock. An existing client's hot path is one lock-free dict.get followed by one per-client Lock. Unrelated clients never contend. The registry lock is held only briefly, to insert a new client or during purge_idle.
• Eviction flag instead of lock hierarchies. Without it, purge_idle could delete a state object that an allow() caller has already looked up. The evicted flag, checked under the client lock, turns that race into a cheap retry. Lock order is always registry → client (purge) or client alone (allow), so deadlock is impossible.
• Non-blocking acquire in purge. A contended client is busy, so it cannot be idle. Skipping it keeps the purger from ever stalling request threads.
• deque(maxlen=max_requests). Rejected calls are never appended, so the log naturally stays at ≤ max_requests. maxlen is a belt-and-braces hard cap.
• Clock injection. Both the clock constructor argument and the now keyword give deterministic tests and simulation. Production uses time.monotonic, which is immune to wall-clock jumps.

Complexity:

Throughput note: at the target of 1,000 requests/second per client across 10k clients, total load is about 10M calls/second. That exceeds what a single CPython process can execute because of the GIL. The design itself will not be the bottleneck, but at that scale you would shard clients across processes by hashing client_id, or move to a shared store such as Redis.