7. Sliding-Window Token Budget Limiter for API Usage

Problem

An API gateway enforces a rolling token budget per client: at any moment, the total tokens consumed by requests within the last window_seconds must not exceed budget.

Requests arrive in timestamp order; each one is either allowed (if it fits within the current rolling budget) or rejected — and a rejected request's tokens never count against the budget, since it never actually ran.

Source: src/7_token_budget_rate_limiter.py

def allowed_requests(requests: list[tuple[str, int, int]], window_seconds: int, budget: int) -> list[str]:
    ...

Examples:
>>> allowed_requests([("r1", 0, 50), ("r2", 5, 60), ("r3", 8, 40), ("r4", 20, 30)], 10, 100)
['r1', 'r3', 'r4']

>>> allowed_requests([("a", 0, 60), ("b", 1, 60)], 10, 100)
['a']

Step-by-Step Approach

1. Recognize this as a sliding-window budget problem: only allowed requests within the trailing window_seconds count toward the current budget check.
2. Maintain a deque of (timestamp, tokens) for requests that were allowed and are still inside the window, plus a running_sum of their tokens.
3. For each incoming request, first evict from the front of the deque any entries whose timestamp has fallen out of the (ts - window_seconds, ts] window, subtracting their tokens from running_sum as they're evicted.
4. Check whether running_sum + this request's tokens fits within budget.
5. If it fits, append (ts, tokens) to the deque, add to running_sum, and mark the request allowed. If it doesn't fit, skip it entirely — do not add it to the deque or running_sum, since a rejected request never consumed budget.
6. Return the collected allowed request_ids in their original order.

The key insight is that both the eviction and the accumulation happen with a monotonically advancing pointer at the front of the deque, so each request is added and removed from the window at most once — giving O(n) total work instead of re-summing history for every request.

Reference solution

from collections import deque


def allowed_requests(requests: list[tuple[str, int, int]], window_seconds: int, budget: int) -> list[str]:
    # deque of (timestamp, tokens) for allowed requests still inside the
    # window; running_sum tracks their total so we never rescan history.
    window: deque[tuple[int, int]] = deque()
    running_sum = 0
    allowed: list[str] = []
    for request_id, ts, tokens in requests:
        # evict entries that fell out of the (ts - window_seconds, ts] window
        while window and window[0][0] <= ts - window_seconds:
            old_ts, old_tokens = window.popleft()
            running_sum -= old_tokens
        if running_sum + tokens <= budget:
            window.append((ts, tokens))
            running_sum += tokens
            allowed.append(request_id)
        # rejected requests are simply skipped: they never enter the window
    return allowed

Key Functions & Tricks

• collections.deque() — O(1) append/popleft at both ends, ideal for a sliding window
• window[0][0] <= ts - window_seconds — eviction condition defining the trailing (ts - window, ts] window
• running_sum — maintained incrementally so budget checks are O(1) instead of re-summing the deque
• skip on rejection — rejected requests are never appended, so they correctly never count toward future budget checks

How to Recognize This Pattern