연습 문제

레이트 리미터는 코딩 테스트보다 시스템 설계 면접과 실무 코드에서 더 자주 등장합니다. 아래 네 문제는 앞 장의 알고리즘을 조금씩 변형한 것입니다. 먼저 직접 풀어 본 뒤 풀이와 비교해 보세요.

문제 1. 로그 감사: 처음으로 한도를 넘은 시각

API 서버가 남긴 요청 시각 목록(초 단위 정수, 오름차순)이 있습니다. 규칙은 "어떤 window초 구간에도 요청이 limit개를 넘으면 안 된다"입니다. 처음으로 규칙을 어긴 요청의 시각을 돌려주고, 어긴 적이 없으면 None을 돌려주세요.

풀이 방향: 슬라이딩 창 로그를 그대로 씁니다. 창 밖으로 나간 시각을 앞에서 버린 뒤 현재 요청을 넣고, 개수가 limit을 넘는 순간이 답입니다. 각 시각은 한 번 들어가고 한 번 나가므로 전체 O(n)입니다.

from collections import deque


def first_violation(times: list[int], limit: int, window: int) -> int | None:
    recent: deque[int] = deque()
    for t in times:
        while recent and recent[0] <= t - window:
            recent.popleft()
        recent.append(t)
        if len(recent) > limit:
            return t
    return None


print(first_violation([1, 2, 11, 12, 13, 14], limit=3, window=10))  # 14
print(first_violation([1, 5, 11, 15], limit=2, window=10))          # None

시각 14에서 구간 (4, 14]에 11, 12, 13, 14 네 개가 들어 있어 처음으로 한도 3을 넘습니다.

문제 2. 비용이 다른 요청과 Retry-After

검색 API는 토큰 5개, 조회 API는 토큰 1개를 쓰는 토큰 버킷이 있습니다. 요청 목록 (시각, 비용)이 주어지면, 허용된 요청에는 None을, 거절된 요청에는 Retry-After 헤더에 넣을 정수 초를 돌려주세요. 버킷은 가득 찬 상태로 시각 0에 시작합니다.

풀이 방향: 게으른 리필 후 비용과 비교합니다. 부족한 토큰이 차는 시간은 (cost - tokens) / rate이고, 헤더는 정수 초이므로 올림합니다. 내림하면 클라이언트가 너무 일찍 재시도해 또 거절당합니다. 용량보다 비싼 요청은 영원히 통과할 수 없으므로 따로 처리합니다.

import math


def simulate(requests: list[tuple[float, float]], rate: float,
             capacity: float) -> list[int | None]:
    tokens, last = capacity, 0.0
    out: list[int | None] = []
    for t, cost in requests:
        if cost > capacity:
            raise ValueError(f"cost {cost} can never fit in capacity {capacity}")
        tokens = min(capacity, tokens + (t - last) * rate)
        last = t
        if tokens >= cost:
            tokens -= cost
            out.append(None)
        else:
            out.append(math.ceil((cost - tokens) / rate))
    return out


print(simulate([(0, 5), (0, 5), (1, 1), (2, 5)], rate=2, capacity=10))
# [None, None, None, 1]

시각 2에는 토큰이 3개뿐이라 2개가 모자랍니다. 초당 2개씩 차므로 1초 뒤에 다시 시도하면 됩니다.

문제 3. 두 단계 한도를 함께 지키기

한 사용자에게 "1초에 3번"과 "1분에 5번"을 동시에 적용하세요. 두 규칙을 모두 통과한 요청만 허용합니다.

풀이 방향: 규칙마다 로그를 두되, 먼저 모든 규칙을 검사하고 전부 통과했을 때만 모든 로그에 기록합니다. 첫 규칙에서 기록을 남긴 뒤 두 번째 규칙에서 거절하면, 거절된 요청이 첫 규칙의 한도를 갉아먹는 버그가 생깁니다. 토큰 버킷 여러 개를 쓸 때도 "모두 확인한 뒤 모두 차감"이 원칙입니다.

from collections import deque


class MultiLimit:
    def __init__(self, rules: list[tuple[int, float]]) -> None:
        self.rules = rules                          # [(limit, window), ...]
        self.logs = [deque() for _ in rules]

    def allow(self, now: float) -> bool:
        for (limit, window), log in zip(self.rules, self.logs):
            while log and log[0] <= now - window:
                log.popleft()
            if len(log) >= limit:
                return False                        # 아무것도 기록하지 않고 거절
        for log in self.logs:
            log.append(now)                         # 모두 통과했을 때만 기록
        return True


limiter = MultiLimit([(3, 1.0), (5, 60.0)])
print([limiter.allow(t) for t in [0, 0.1, 0.2, 0.3, 1.5, 1.6, 1.7, 3.0]])
# [True, True, True, False, True, True, False, False]

문제 4. 클라이언트 쪽 재시도 간격

429를 받은 클라이언트의 다음 재시도까지 기다릴 시간을 계산하세요. 서버가 Retry-After를 주었다면 그보다 일찍 보내면 안 되고, 주지 않았다면 지수 백오프에 지터(무작위 흔들림)를 더합니다.

풀이 방향: 모든 클라이언트가 똑같이 1초, 2초, 4초 뒤에 재시도하면 같은 순간에 다시 몰립니다. 0부터 min(cap, base × 2^attempt) 사이에서 무작위로 고르는 "full jitter"가 재시도를 고르게 흩어 줍니다.

import random


def next_delay(attempt: int, retry_after: float | None,
               base: float = 0.5, cap: float = 30.0) -> float:
    backoff = random.uniform(0, min(cap, base * 2 ** attempt))   # full jitter
    if retry_after is not None:
        return max(retry_after, backoff)
    return backoff


random.seed(7)
for attempt in range(5):
    print(attempt, round(next_delay(attempt, None), 2))   # 상한 0.5, 1, 2, 4, 8 안의 무작위 값
print(next_delay(0, retry_after=12))                       # 12

정리

• 로그 감사는 deque 슬라이딩 창으로 O(n)에 풉니다.
• Retry-After는 부족한 토큰이 차는 시간을 올림해 계산합니다.
• 여러 한도는 "모두 검사한 뒤 모두 기록"해야 거절된 요청이 한도를 갉아먹지 않습니다.
• 클라이언트는 Retry-After를 지키고, 없으면 지터를 더한 지수 백오프로 재시도합니다.