Skip to content

← Seams, Modules, Packaging and Tooling step 13 of 36

Hard End-to-End

A rate limiter you can test without sleeping

Build a sliding-window rate limiter whose every dependency arrives through its constructor, then drive it through a scripted timeline.

Define, in this module:

class Clock(Protocol):
    def now(self) -> float: ...


class RateLimitStore(Protocol):
    def get(self, key: str) -> list[float] | None: ...
    def set(self, key: str, hits: list[float]) -> None: ...


class RateLimiter:
    def __init__(self, clock: Clock, store: RateLimitStore) -> None: ...
    def allow(self, key: str, limit: int, window_seconds: float) -> bool: ...

allow implements a sliding window log:

  1. Read the current time from the injected clock.
  2. Load this key’s recorded hit timestamps from the store. A key that has never been seen returns None, not an empty list — that is what a real key-value store does, and handling it is part of the exercise.
  3. Discard every timestamp that is not strictly greater than now - window_seconds. A hit exactly window_seconds old has left the window.
  4. If fewer than limit survive, record now and allow. Otherwise deny and record nothing.
  5. Write the surviving list back either way.

Then the graded entrypoint:

def solve(
    times: list[float],
    keys: list[str],
    limit: int,
    window_seconds: float,
) -> list[bool]:

Construct one RateLimiter over your own FakeClock and an in-memory store, replay the (time, key) pairs in order — setting the fake clock to each time before calling allow — and return the decision for each request.

No patching, no sleeping. Nothing in your solution may call time.time(), time.sleep() or unittest.mock. That is not an arbitrary restriction: it is the whole point. A rate limiter that reaches for the system clock can only be tested by waiting, so its window-rollover behaviour — the one thing that is actually hard — gets tested by nobody. With the clock injected, the boundary case is a two-line assertion that runs in microseconds.

The boundary is deliberately sharp. With limit=1 and window_seconds=10, a hit at t=0 and a request at t=10 must be allowed: 0 > 10 - 10 is false, so the old hit is outside the window. Get the comparison backwards and you have an off-by-one that only appears under production traffic.

Your submission must pass mypy --strict. Both Protocols need full annotations, your fakes must satisfy them structurally — no inheriting from the Protocol — and store.get returning list[float] | None must be narrowed before use.

Loading visualization…