Rate Limiter

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;

/** The entire public surface. Two lines. Everything hard is behind it. */
interface RateLimiter {
    boolean allow(String key);
}

/** Inject the clock, or a test for a 60-second window has to sleep for 60 seconds. */
interface Clock { long millis(); }

final class SystemClock implements Clock {
    public long millis() { return System.currentTimeMillis(); }
}

final class FakeClock implements Clock {
    private long now;
    FakeClock(long start) { this.now = start; }
    public long millis()      { return now; }
    void set(long ms)         { now = ms; }
    void advance(long ms)     { now += ms; }
}

/** Per-key state. lastSeen exists only so idle keys can be swept. */
abstract class KeyState { long lastSeen; }

/**
 * Everything the five algorithms share: one small state object per key, an
 * injected clock, a limit, a window, and a lock around ONE bucket.
 */
abstract class KeyedLimiter<S extends KeyState> implements RateLimiter {
    protected final ConcurrentHashMap<String, S> byKey = new ConcurrentHashMap<>();
    protected final Clock clock;
    protected final int limit;
    protected final long windowMs;

    KeyedLimiter(Clock clock, int limit, long windowMs) {
        this.clock = clock; this.limit = limit; this.windowMs = windowMs;
    }

    protected abstract S newState();
    protected abstract boolean tryTake(S state, long now);

    @Override public boolean allow(String key) {
        S state = byKey.computeIfAbsent(key, k -> newState());   // one bucket per key
        synchronized (state) {                                   // lock the BUCKET, not the map
            long now = clock.millis();
            state.lastSeen = now;
            return tryTake(state, now);                          // refill + check + take, atomically
        }
    }

    /** Without this the map grows forever — one entry per key ever seen. */
    public int sweepIdle(long idleMs) {
        long cutoff = clock.millis() - idleMs;
        int before = byKey.size();
        byKey.entrySet().removeIf(e -> e.getValue().lastSeen < cutoff);
        return before - byKey.size();
    }

    public int keyCount() { return byKey.size(); }
}

/** 1 · FIXED WINDOW — two numbers per key. Lets 2x the limit through at a boundary. */
final class FixedWindow extends KeyedLimiter<FixedWindow.W> {
    static final class W extends KeyState { long start = Long.MIN_VALUE; int count; }
    FixedWindow(Clock c, int limit, long windowMs) { super(c, limit, windowMs); }
    protected W newState() { return new W(); }

    protected boolean tryTake(W w, long now) {
        long windowStart = now - Math.floorMod(now, windowMs);   // derived from the clock, not scheduled
        if (w.start != windowStart) { w.start = windowStart; w.count = 0; }   // the reset instant
        if (w.count < limit) { w.count++; return true; }
        return false;
    }
}

/** 2 · SLIDING WINDOW LOG — exactly correct, and memory grows with the request rate. */
final class SlidingLog extends KeyedLimiter<SlidingLog.L> {
    static final class L extends KeyState { final ArrayDeque<Long> stamps = new ArrayDeque<>(); }
    SlidingLog(Clock c, int limit, long windowMs) { super(c, limit, windowMs); }
    protected L newState() { return new L(); }

    protected boolean tryTake(L l, long now) {
        long cutoff = now - windowMs;
        while (!l.stamps.isEmpty() && l.stamps.peekFirst() <= cutoff) l.stamps.pollFirst();  // lazy eviction
        if (l.stamps.size() < limit) { l.stamps.addLast(now); return true; }
        return false;
    }
}

/** 3 · SLIDING WINDOW COUNTER — three numbers per key, approximately right. */
final class SlidingCounter extends KeyedLimiter<SlidingCounter.C> {
    static final class C extends KeyState { long start = Long.MIN_VALUE; int curr, prev; }
    SlidingCounter(Clock c, int limit, long windowMs) { super(c, limit, windowMs); }
    protected C newState() { return new C(); }

    protected boolean tryTake(C c, long now) {
        long windowStart = now - Math.floorMod(now, windowMs);
        if (windowStart != c.start) {
            // one window on -> the old current becomes the previous; a longer gap -> nothing carries over
            c.prev  = (c.start != Long.MIN_VALUE && windowStart - c.start == windowMs) ? c.curr : 0;
            c.curr  = 0;
            c.start = windowStart;
        }
        double elapsed  = (now - windowStart) / (double) windowMs;         // 0.0 .. 1.0
        double estimate = c.prev * (1.0 - elapsed) + c.curr;               // the weighted share
        if (estimate < limit) { c.curr++; return true; }
        return false;
    }
}

/** 4 · TOKEN BUCKET — lazy refill, allows a burst up to capacity. The usual default. */
final class TokenBucket extends KeyedLimiter<TokenBucket.B> {
    private final double capacity;
    private final double ratePerMs;

    TokenBucket(Clock c, int limit, long windowMs) {
        super(c, limit, windowMs);
        this.capacity  = limit;                        // burst size; a real system tunes this separately
        this.ratePerMs = limit / (double) windowMs;    // 5 per 60s -> 0.0000833 tokens per ms
    }
    static final class B extends KeyState { double tokens = -1; long lastRefill; }
    protected B newState() { return new B(); }

    protected boolean tryTake(B b, long now) {
        if (b.tokens < 0) { b.tokens = capacity; b.lastRefill = now; }      // a new key starts full
        b.tokens    = Math.min(capacity, b.tokens + (now - b.lastRefill) * ratePerMs);   // LAZY refill
        b.lastRefill = now;                                                 // no timer anywhere
        if (b.tokens >= 1.0) { b.tokens -= 1.0; return true; }
        return false;
    }

    double tokensFor(String key) { B b = byKey.get(key); return b == null ? capacity : b.tokens; }
}

/** 5 · LEAKY BUCKET — a fixed queue drained at a constant rate: smooths instead of bursting. */
final class LeakyBucket extends KeyedLimiter<LeakyBucket.Q> {
    private final double capacity;
    private final double leakPerMs;

    LeakyBucket(Clock c, int limit, long windowMs) {
        super(c, limit, windowMs);
        this.capacity  = limit;
        this.leakPerMs = limit / (double) windowMs;
    }
    static final class Q extends KeyState { double level; long lastLeak = Long.MIN_VALUE; }
    protected Q newState() { return new Q(); }

    protected boolean tryTake(Q q, long now) {
        if (q.lastLeak == Long.MIN_VALUE) q.lastLeak = now;
        q.level    = Math.max(0.0, q.level - (now - q.lastLeak) * leakPerMs);   // drain by elapsed time
        q.lastLeak = now;
        if (q.level + 1.0 <= capacity) { q.level += 1.0; return true; }         // joins the queue
        return false;                                                           // overflow -> dropped
    }
}

public class Main {
    static int fire(RateLimiter limiter, String key, int n) {
        int ok = 0;
        for (int i = 0; i < n; i++) if (limiter.allow(key)) ok++;
        return ok;
    }

    /** Five at 12:00:59, five at 12:01:01 — the whole problem, in six lines. */
    static void boundary(String label, Function<Clock, RateLimiter> make) {
        FakeClock clock = new FakeClock(0);
        RateLimiter limiter = make.apply(clock);
        clock.set(59_000);                          // 12:00:59 — last second of the window
        int first = fire(limiter, "user-42", 5);
        clock.set(61_000);                          // 12:01:01 — two seconds later
        int second = fire(limiter, "user-42", 5);
        System.out.printf("  %-18s %d at :59  +  %d at 1:01  =  %2d in 2 seconds%s%n",
                label, first, second, first + second, first + second > 5 ? "   <-- above the limit" : "");
    }

    public static void main(String[] args) throws Exception {
        System.out.println("== the boundary · limit 5 per 60s ==");
        boundary("fixed window",    c -> new FixedWindow(c, 5, 60_000));
        boundary("sliding log",     c -> new SlidingLog(c, 5, 60_000));
        boundary("sliding counter", c -> new SlidingCounter(c, 5, 60_000));
        boundary("token bucket",    c -> new TokenBucket(c, 5, 60_000));
        boundary("leaky bucket",    c -> new LeakyBucket(c, 5, 60_000));

        System.out.println();
        System.out.println("== lazy refill · tokens return because time passed, not because a thread ran ==");
        FakeClock clock = new FakeClock(0);
        TokenBucket bucket = new TokenBucket(clock, 5, 60_000);
        System.out.printf("  t=0s   allowed %d of 8   tokens left %.2f%n",
                fire(bucket, "user-42", 8), bucket.tokensFor("user-42"));
        clock.advance(30_000);
        System.out.printf("  t=30s  allowed %d of 8   tokens left %.2f%n",
                fire(bucket, "user-42", 8), bucket.tokensFor("user-42"));

        System.out.println();
        System.out.println("== per key · and the map that grows forever ==");
        FakeClock c2 = new FakeClock(0);
        TokenBucket perKey = new TokenBucket(c2, 5, 60_000);
        System.out.println("  user-42 hammered 20x -> allowed " + fire(perKey, "user-42", 20));
        System.out.println("  user-7  first call   -> " + perKey.allow("user-7"));
        System.out.println("  keys held: " + perKey.keyCount());
        c2.advance(10 * 60_000);
        System.out.println("  swept " + perKey.sweepIdle(5 * 60_000) + " idle keys -> keys held: " + perKey.keyCount());

        System.out.println();
        System.out.println("== 8 threads, 1 key, limit 10 ==");
        FakeClock c3 = new FakeClock(0);
        RateLimiter shared = new TokenBucket(c3, 10, 60_000);
        ExecutorService pool = Executors.newFixedThreadPool(8);
        List<Future<Integer>> futures = new ArrayList<>();
        for (int t = 0; t < 8; t++) futures.add(pool.submit(() -> fire(shared, "user-42", 5)));
        int total = 0;
        for (Future<Integer> f : futures) total += f.get();
        pool.shutdown();
        System.out.println("  40 requests attempted -> " + total + " allowed, never more than the limit");
    }
}

/* Expected output:
== the boundary · limit 5 per 60s ==
  fixed window       5 at :59  +  5 at 1:01  =  10 in 2 seconds   <-- above the limit
  sliding log        5 at :59  +  0 at 1:01  =   5 in 2 seconds
  sliding counter    5 at :59  +  1 at 1:01  =   6 in 2 seconds   <-- above the limit
  token bucket       5 at :59  +  0 at 1:01  =   5 in 2 seconds
  leaky bucket       5 at :59  +  0 at 1:01  =   5 in 2 seconds

== lazy refill · tokens return because time passed, not because a thread ran ==
  t=0s   allowed 5 of 8   tokens left 0.00
  t=30s  allowed 2 of 8   tokens left 0.50

== per key · and the map that grows forever ==
  user-42 hammered 20x -> allowed 5
  user-7  first call   -> true
  keys held: 2
  swept 2 idle keys -> keys held: 0

== 8 threads, 1 key, limit 10 ==
  40 requests attempted -> 10 allowed, never more than the limit

The sliding counter's 6 is not a bug: it is the approximation. It weights the
previous window by the fraction still inside the moving window, so one extra
request slips through one second past the boundary instead of five.
*/