Katkat
Back

Implement a Leaky Bucket Rate Limiter

BoxSoftware Engineer · Onsite
EasySoftware EngineerOnsiteCoding & Algorithms

Last updated

This question evaluates understanding of rate-limiting algorithms, time-based state management, and concurrent programming concepts, including synchronization and test design for boundary and timing behaviors.

Implement a leaky bucket rate limiter.

You are given a class template for a rate limiter that controls whether incoming requests should be accepted or rejected. The limiter has:

  • capacity: the maximum number of requests that can be waiting in the bucket.
  • leak_rate_per_second: the number of requests that leave the bucket per second.

Design and implement a class with an API similar to:

class LeakyBucketRateLimiter:
 def __init__(self, capacity: int, leak_rate_per_second: float):
 pass

 def allow_request(self) -> bool:
 pass

allow_request() should return:

  • True if the request is accepted into the bucket.
  • False if the bucket is already full after accounting for leaked requests.

Requirements:

  1. First implement a single-threaded version.
  2. Write tests for normal behavior, boundary cases, and time-based leakage.
  3. Then extend the implementation so that it is safe when multiple threads call allow_request() concurrently.
  4. Reuse as much of the original implementation as possible when adding thread-safety.