LeetLLM
MediumRate LimitingPython 3

Notification Rate Limiter

Throttle noisy notification channels while deduping repeated webhook or automation events.

35m3 sample tests8 hidden tests

Implement NotificationLimiter, a per-channel notification throttle with dedupe keys.

Requirements

  • Constructor receives limit, window_seconds, and injected now. Raise ValueError if limit <= 0 or window_seconds <= 0.
  • allow(channel, dedupe_key) returns (allowed, retry_after_seconds).
  • A channel may send at most limit unique dedupe keys per rolling window.
  • Dedupe keys are scoped per channel (the same key may be allowed on another channel).
  • Repeating the same dedupe key inside the window is blocked without consuming a new slot.
  • Expired records are removed on every call. A record expires when now - timestamp >= window_seconds (half-open window).
  • retry_after_seconds is 0 when allowed.
  • On capacity deny: retry_after_seconds = max(0, window_seconds - (now - oldest_active_timestamp)).
  • On duplicate-key deny: the same formula using that key's stored timestamp.

Example

This trace separates duplicate denial from capacity denial, then crosses the half-open expiry boundary so a new key can enter.

python
1clock = {"now": 0.0} 2limiter = NotificationLimiter(2, 10.0, lambda: clock["now"]) 3assert limiter.allow("slack", "a") == (True, 0.0) 4assert limiter.allow("slack", "a") == (False, 10.0) # duplicate: full window remains 5assert limiter.allow("slack", "b") == (True, 0.0) 6assert limiter.allow("slack", "c") == (False, 10.0) # channel is at capacity 7clock["now"] = 10.0 8assert limiter.allow("slack", "c") == (True, 0.0)

Constraints

  • Use injected time. Keep tests free of sleeps.
  • Keep channels isolated.

Editor