Sliding-Window Rate Limiter

Problem Overview

Build a rate limiter for an ordered stream of request timestamps. The initial version uses one shared stream; later versions associate requests with user and experience identifiers. A request may pass only when counting it would remain within the configured maximum for the relevant sliding window.

This problem may also be described as “Rate Limiter,” “Log Rate Limiter,” or a LeetCode 359 / 362-style rate limiter. The Roblox variant differs from both LeetCode versions: each request produces a Boolean allow/deny result, and the follow-up applies several independent windows simultaneously.

from typing import List

def rate_limiter(
    requestTimestamps: List[int],
    windowLength: int,
    maxRequests: int,
) -> List[bool]:
    """
    Return True for each accepted request and False for each denied request.
    Only accepted requests count against future capacity.
    """
    pass

requestTimestamps = [1, 2, 3, 4, 5, 6]
windowLength = 3
maxRequests = 2

rate_limiter(requestTimestamps, windowLength, maxRequests)
# [True, True, False, True, True, False]

requestTimestamps = [1, 2, 3, 4, 5, 6]
windowLength = 3
maxRequests = 2

[true, true, false, true, true, false]

from typing import List

def per_entity_rate_limiter(
    requestTimestamps: List[int],
    userIds: List[int],
    experienceIds: List[str],
    windowLength: int,
    maxRequests: int,
) -> List[bool]:
    """
    Return True when the request is allowed by both:
    1. its user's sliding window
    2. its experience's sliding window
    """
    pass

requestTimestamps = [1, 2, 3, 4, 5]
userIds = [1, 1, 2, 1, 2]
experienceIds = ["A", "A", "A", "A", "B"]
windowLength = 3
maxRequests = 1

per_entity_rate_limiter(
    requestTimestamps,
    userIds,
    experienceIds,
    windowLength,
    maxRequests,
)
# [True, False, False, True, True]

Follow-up Variant: More Request Fields

A follow-up may introduce dimensions such as an IP address, a device ID, or several other request attributes, then require a separate rate limit for each one.

The same idea extends to every dimension: maintain a queue map for each field, and accept a request only when all of its field-specific queues remain below their limits.

from collections import defaultdict, deque
from typing import Deque, Dict, Hashable, List, Mapping

def multi_field_rate_limiter(
    requestTimestamps: List[int],
    fieldsByName: Mapping[str, List[Hashable]],
    windowLength: int,
    maxRequests: int,
) -> List[bool]:
    windows: Dict[str, Dict[Hashable, Deque[int]]] = {
        field_name: defaultdict(deque)
        for field_name in fieldsByName
    }
    decisions = []

    for i, timestamp in enumerate(requestTimestamps):
        current_windows = []

        for field_name, values in fieldsByName.items():
            window = windows[field_name][values[i]]
            _cleanup(window, timestamp, windowLength)
            current_windows.append(window)

        if all(len(window) < maxRequests for window in current_windows):
            decisions.append(True)
            for window in current_windows:
                window.append(timestamp)
        else:
            decisions.append(False)

    return decisions

Adding an IP dimension requires only another entry in the field mapping:

multi_field_rate_limiter(
    requestTimestamps,
    {
        "user": userIds,
        "experience": experienceIds,
        "ip": ipAddresses,
    },
    windowLength,
    maxRequests,
)

If each field uses its own limit, keep a separate (windowLength, maxRequests) configuration for each dimension:

limits = {
    "user": (60, 100), # windowLength, maxRequests
    "experience": (60, 500),
    "ip": (60, 50),
}

The corresponding queue cleanup and capacity check should then use that dimension’s configured pair.

Test Cases

assert rate_limiter([1, 2, 3, 4, 5, 6], 3, 2) == [
    True, True, False, True, True, False
]

# Requests exactly on the left boundary expire.
assert rate_limiter([1, 4], 3, 1) == [True, True]

# Denied requests are not stored.
assert rate_limiter([1, 2, 3, 4], 3, 2) == [True, True, False, True]

assert per_entity_rate_limiter(
    [1, 2, 3, 4, 5],
    [1, 1, 2, 1, 2],
    ["A", "A", "A", "A", "B"],
    3,
    1,
) == [True, False, False, True, True]

# Experience cap can deny a request even when the user is new.
assert per_entity_rate_limiter(
    [10, 11],
    [1, 2],
    ["game-1", "game-1"],
    10,
    1,
) == [True, False]