Rate Limiter Implementations (Go & Python)

import threading
import time
class TokenBucket:
    """Allows bursts up to capacity, then throttles to the refill rate."""
    def __init__(self, capacity: float, refill_rate: float) -> None:
        self._lock = threading.Lock()
        self.capacity = capacity
        self.tokens = capacity  # start full
        self.refill_rate = refill_rate  # tokens per second
        self.last_refill = time.monotonic()
    def allow(self) -> bool:
        """Return True if a single request may proceed right now."""
        with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            self.last_refill = now
            if self.tokens < 1:
                return False
            self.tokens -= 1
            return True
# asyncio variant: swap threading.Lock for asyncio.Lock and `allow` for an
# async def if callers are already inside an event loop instead of threads --
# the refill math is identical, only the mutual-exclusion primitive changes.

package ratelimit;
import java.util.concurrent.locks.ReentrantLock;
// TokenBucket allows bursts up to capacity, then throttles to the refill rate.
// Implements Limiter (defined below in the HTTP Middleware Wrapper section)
// explicitly -- unlike Go's implicit interface satisfaction or Python's
// structural Protocol above, Java requires "implements" up front.
public class TokenBucket implements Limiter {
    private final ReentrantLock lock = new ReentrantLock();
    private final double capacity;
    private double tokens;
    private final double refillRate; // tokens per second
    private long lastRefillNanos;
    public TokenBucket(double capacity, double refillRate) {
        this.capacity = capacity;
        this.tokens = capacity; // start full
        this.refillRate = refillRate;
        this.lastRefillNanos = System.nanoTime();
    }
    // allow reports whether a single request may proceed right now.
    @Override
    public boolean allow() {
        lock.lock();
        try {
            long now = System.nanoTime();
            double elapsed = (now - lastRefillNanos) / 1_000_000_000.0;
            tokens = Math.min(capacity, tokens + elapsed * refillRate);
            lastRefillNanos = now;
            if (tokens < 1) {
                return false;
            }
            tokens -= 1;
            return true;
        } finally {
            lock.unlock();
        }
    }
}
// Uses a ReentrantLock (java.util.concurrent.locks) as the direct analog of
// Go's sync.Mutex / Python's threading.Lock -- refill-then-check-then-consume
// is a compound operation needing one critical section, not a single atomic
// counter update, so a bare AtomicLong/AtomicDouble CAS loop wouldn't suffice.

import threading
import time
class LeakyBucket:
    """Queues requests and lets them "leak" out at a fixed rate.
    Unlike TokenBucket, it smooths bursts rather than passing them through.
    """
    def __init__(self, capacity: int, leak_rate: float) -> None:
        """leak_rate: seconds between each leak (1 request drained)."""
        self._lock = threading.Lock()
        self.capacity = capacity
        self.queue = 0  # current queue depth
        self.leak_rate = leak_rate
        self.last_leak = time.monotonic()
    def allow(self) -> bool:
        """Return True if the request can be queued (admitted to the bucket).
        Does NOT mean the request is processed immediately -- only that it
        was admitted to the queue for eventual draining at leak_rate.
        """
        with self._lock:
            self._drain(time.monotonic())
            if self.queue >= self.capacity:
                return False  # queue full, reject
            self.queue += 1
            return True
    def _drain(self, now: float) -> None:
        """Remove completed "leaks" based on elapsed time since last_leak."""
        elapsed = now - self.last_leak
        leaked = int(elapsed / self.leak_rate)
        if leaked <= 0:
            return
        leaked = min(leaked, self.queue)
        self.queue -= leaked
        self.last_leak += leaked * self.leak_rate

package ratelimit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
// LeakyBucket queues requests and lets them "leak" out at a fixed rate.
// Unlike TokenBucket, it smooths bursts rather than passing them through.
public class LeakyBucket implements Limiter {
    private final ReentrantLock lock = new ReentrantLock();
    private final int capacity; // max queued requests
    private final AtomicInteger queue = new AtomicInteger(0); // current queue depth
    private final long leakRateNanos; // time between each leak (1 request drained)
    private long lastLeakNanos;
    public LeakyBucket(int capacity, long leakRateMillis) {
        this.capacity = capacity;
        this.leakRateNanos = leakRateMillis * 1_000_000L;
        this.lastLeakNanos = System.nanoTime();
    }
    // allow reports whether the request can be queued (accepted into the bucket).
    // It does NOT mean the request is processed immediately -- only that it was
    // admitted to the queue for eventual draining at leakRate.
    @Override
    public boolean allow() {
        lock.lock();
        try {
            drain(System.nanoTime());
            if (queue.get() >= capacity) {
                return false; // queue full, reject
            }
            queue.incrementAndGet();
            return true;
        } finally {
            lock.unlock();
        }
    }
    // drain removes completed "leaks" based on elapsed time since lastLeakNanos.
    private void drain(long now) {
        long elapsed = now - lastLeakNanos;
        int leaked = (int) (elapsed / leakRateNanos);
        if (leaked <= 0) {
            return;
        }
        int current = queue.get();
        if (leaked > current) {
            leaked = current;
        }
        queue.addAndGet(-leaked);
        lastLeakNanos += leaked * leakRateNanos;
    }
}
// queue is an AtomicInteger purely so reads outside the lock (metrics,
// health checks) stay tear-free -- the compound drain+check+increment
// inside allow() still runs under the ReentrantLock for correctness.

import threading
import time
class FixedWindow:
    """Counts requests per discrete time window."""
    def __init__(self, limit: int, window_size: float) -> None:
        self._lock = threading.Lock()
        self.limit = limit
        self.window_size = window_size
        self.window_start = time.monotonic()
        self.count = 0
    def allow(self) -> bool:
        with self._lock:
            now = time.monotonic()
            if now - self.window_start >= self.window_size:
                # New window: reset counter and boundary.
                self.window_start = now
                self.count = 0
            if self.count >= self.limit:
                return False
            self.count += 1
            return True

package ratelimit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
// FixedWindow counts requests per discrete time window.
public class FixedWindow implements Limiter {
    private final ReentrantLock lock = new ReentrantLock();
    private final int limit;
    private final long windowSizeNanos;
    private long windowStartNanos;
    private final AtomicInteger count = new AtomicInteger(0);
    public FixedWindow(int limit, long windowSizeMillis) {
        this.limit = limit;
        this.windowSizeNanos = windowSizeMillis * 1_000_000L;
        this.windowStartNanos = System.nanoTime();
    }
    @Override
    public boolean allow() {
        lock.lock();
        try {
            long now = System.nanoTime();
            if (now - windowStartNanos >= windowSizeNanos) {
                // New window: reset counter and boundary.
                windowStartNanos = now;
                count.set(0);
            }
            if (count.get() >= limit) {
                return false;
            }
            count.incrementAndGet();
            return true;
        } finally {
            lock.unlock();
        }
    }
}

import threading
import time
from collections import deque
class SlidingWindowLog:
    """Exact (no boundary burst) but stores one timestamp per request
    within the window -- memory scales with request volume.
    """
    def __init__(self, limit: int, window_size: float) -> None:
        self._lock = threading.Lock()
        self.limit = limit
        self.window_size = window_size
        self.timestamps: deque[float] = deque()
    def allow(self) -> bool:
        with self._lock:
            now = time.monotonic()
            cutoff = now - self.window_size
            self._prune_before(cutoff)
            if len(self.timestamps) >= self.limit:
                return False
            self.timestamps.append(now)
            return True
    def _prune_before(self, cutoff: float) -> None:
        """Drop timestamps older than cutoff.
        Timestamps are appended in increasing order, so the surviving
        entries are always a suffix -- popleft() from a deque is O(1) per
        expired entry, not a full O(n) rebuild with allocation per call.
        """
        while self.timestamps and self.timestamps[0] < cutoff:
            self.timestamps.popleft()

package ratelimit;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
// SlidingWindowLog is exact (no boundary burst) but stores one timestamp
// per request within the window -- memory scales with request volume.
public class SlidingWindowLog implements Limiter {
    private final ReentrantLock lock = new ReentrantLock();
    private final int limit;
    private final long windowSizeNanos;
    private final ConcurrentLinkedDeque<Long> timestamps = new ConcurrentLinkedDeque<>();
    private final AtomicInteger size = new AtomicInteger(0);
    public SlidingWindowLog(int limit, long windowSizeMillis) {
        this.limit = limit;
        this.windowSizeNanos = windowSizeMillis * 1_000_000L;
    }
    @Override
    public boolean allow() {
        lock.lock();
        try {
            long now = System.nanoTime();
            long cutoff = now - windowSizeNanos;
            pruneBefore(cutoff);
            if (size.get() >= limit) {
                return false;
            }
            timestamps.addLast(now);
            size.incrementAndGet();
            return true;
        } finally {
            lock.unlock();
        }
    }
    // pruneBefore drops timestamps older than cutoff. Timestamps are appended
    // in increasing order, so the surviving deque is always a suffix -- this
    // is O(k) where k is the number of expired entries, not a full O(n) scan
    // with allocation per call.
    private void pruneBefore(long cutoff) {
        Long head;
        while ((head = timestamps.peekFirst()) != null && head < cutoff) {
            timestamps.pollFirst();
            size.decrementAndGet();
        }
    }
}
// ConcurrentLinkedDeque gives lock-free peekFirst()/pollFirst()/addLast(),
// but the overall prune-then-check-then-append sequence still needs the
// ReentrantLock -- otherwise two threads could both pass the size check
// before either appends, admitting one request too many past the limit.

from typing import Callable, Protocol
class Limiter(Protocol):
    """Satisfied by TokenBucket, LeakyBucket, FixedWindow, and
    SlidingWindowLog -- any algorithm above can be plugged into the
    middleware below.
    """
    def allow(self) -> bool: ...
def rate_limit_middleware(limiter: Limiter, app: Callable) -> Callable:
    """WSGI middleware: reject requests with 429 when the underlying
    Limiter denies them.
    In production this would key limiters per-client (see
    PerClientMiddleware below) rather than share one global limiter
    across all traffic.
    """
    def wrapped_app(environ, start_response):
        if not limiter.allow():
            start_response(
                "429 Too Many Requests",
                [("Content-Type", "text/plain"), ("Retry-After", "1")],
            )
            return [b"429 Too Many Requests"]
        return app(environ, start_response)
    return wrapped_app
# Flask equivalent, same decision, different plumbing:
#
#   @app.before_request
#   def check_rate_limit():
#       if not limiter.allow():
#           resp = make_response("429 Too Many Requests", 429)
#           resp.headers["Retry-After"] = "1"
#           return resp

// File: Limiter.java
package ratelimit;
// Limiter is satisfied by TokenBucket, LeakyBucket, FixedWindow, and
// SlidingWindowLog -- any algorithm above can be plugged into RateLimitFilter.
// Java needs an explicit "implements Limiter" on each class (see above);
// there's no structural typing the way Go's interfaces or Python's Protocol
// give you for free.
public interface Limiter {
    boolean allow();
}
// File: RateLimitFilter.java
package ratelimit;
import jakarta.servlet.Filter;
import jakarta.servlet.FilterChain;
import jakarta.servlet.FilterConfig;
import jakarta.servlet.ServletException;
import jakarta.servlet.ServletRequest;
import jakarta.servlet.ServletResponse;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;
// RateLimitFilter wraps any Limiter as a jakarta.servlet.Filter, rejecting
// requests with 429 when the underlying Limiter denies them. In production
// this would key limiters per-client (see PerClientFilter below) rather than
// share one global limiter across all traffic. Uses jakarta.servlet (Jakarta
// EE 9+ / Servlet 6.0, the namespace shipped by Tomcat 10+ and Spring Boot
// 3+) rather than the legacy javax.servlet package -- pick whichever matches
// your container, the doFilter logic below is identical either way.
public class RateLimitFilter implements Filter {
    private final Limiter limiter;
    public RateLimitFilter(Limiter limiter) {
        this.limiter = limiter;
    }
    @Override
    public void init(FilterConfig filterConfig) {
        // No setup needed -- limiter is already constructed.
    }
    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {
        if (!limiter.allow()) {
            HttpServletResponse httpResponse = (HttpServletResponse) response;
            httpResponse.setHeader("Retry-After", "1");
            httpResponse.sendError(429, "429 Too Many Requests"); // no SC_TOO_MANY_REQUESTS constant in the servlet API
            return;
        }
        chain.doFilter(request, response);
    }
    @Override
    public void destroy() {
        // No resources to release.
    }
}

import threading
from typing import Callable, Dict
class PerClientMiddleware:
    """Maintains one Limiter per client key (e.g. IP address), lazily
    created on first request. limiter_factory lets the caller choose
    which of the four algorithms to instantiate per client.
    """
    def __init__(self, limiter_factory: Callable[[], "Limiter"]) -> None:
        self._lock = threading.Lock()
        self.limiters: Dict[str, "Limiter"] = {}
        self.limiter_factory = limiter_factory
    def wrap(self, app: Callable) -> Callable:
        def wrapped_app(environ, start_response):
            key = self._client_key(environ)
            with self._lock:
                limiter = self.limiters.get(key)
                if limiter is None:
                    limiter = self.limiter_factory()
                    self.limiters[key] = limiter
            if not limiter.allow():
                start_response(
                    "429 Too Many Requests",
                    [("Content-Type", "text/plain"), ("Retry-After", "1")],
                )
                return [b"429 Too Many Requests"]
            return app(environ, start_response)
        return wrapped_app
    @staticmethod
    def _client_key(environ) -> str:
        return environ.get("REMOTE_ADDR", "unknown")
# Example wiring:
#
#   mw = PerClientMiddleware(lambda: TokenBucket(capacity=20, refill_rate=5))
#   app = mw.wrap(api_handler)  # burst 20, refill 5/sec, per client
#
# Note: limiters dict above grows unbounded as new clients appear -- a
# production version needs an eviction policy (e.g. LRU from lru-cache.md,
# or a TTL sweep) to bound memory under a churn of unique client keys.

package ratelimit;
import jakarta.servlet.Filter;
import jakarta.servlet.FilterChain;
import jakarta.servlet.FilterConfig;
import jakarta.servlet.ServletException;
import jakarta.servlet.ServletRequest;
import jakarta.servlet.ServletResponse;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
// PerClientFilter maintains one Limiter per client key (e.g. IP address),
// lazily created on first request. limiterFactory lets the caller choose
// which of the four algorithms to instantiate per client.
public class PerClientFilter implements Filter {
    private final Map<String, Limiter> limiters = new ConcurrentHashMap<>();
    private final Supplier<Limiter> limiterFactory;
    public PerClientFilter(Supplier<Limiter> limiterFactory) {
        this.limiterFactory = limiterFactory;
    }
    @Override
    public void init(FilterConfig filterConfig) {
        // No setup needed.
    }
    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {
        String key = clientKey((HttpServletRequest) request);
        Limiter limiter = limiters.computeIfAbsent(key, k -> limiterFactory.get());
        if (!limiter.allow()) {
            HttpServletResponse httpResponse = (HttpServletResponse) response;
            httpResponse.setHeader("Retry-After", "1");
            httpResponse.sendError(429, "429 Too Many Requests"); // no SC_TOO_MANY_REQUESTS constant in the servlet API
            return;
        }
        chain.doFilter(request, response);
    }
    @Override
    public void destroy() {
        // No resources to release.
    }
    private static String clientKey(HttpServletRequest request) {
        return request.getRemoteAddr();
    }
}
// ConcurrentHashMap.computeIfAbsent gives lock-free-on-the-common-path,
// exactly-once-per-key construction of each client's Limiter -- the direct
// analog of Go's mutex-guarded map lookup and Python's lock-guarded dict.get.
// Example wiring (web.xml or Servlet 6.0 annotation-based registration):
//   PerClientFilter filter = new PerClientFilter(
//       () -> new TokenBucket(20, 5)); // burst 20, refill 5/sec, per client
//   FilterRegistration.Dynamic reg =
//       servletContext.addFilter("rateLimit", filter);
//   reg.addMappingForUrlPatterns(null, false, "/api/*");
// Note: limiters map above grows unbounded as new clients appear -- a
// production version needs an eviction policy (e.g. LRU from lru-cache.md,
// or a TTL sweep) to bound memory under a churn of unique client keys.