Rate Limiter

package com.practice.phonepe.questions.ratelimiter;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Single-process implementation -- only correct if there's exactly one server. Every caller
 * uses window-bucketed keys (e.g. "clientId:windowStartMillis"), so a key is never reused
 * once its window has passed; that's why there's no explicit expiry here. A real deployment
 * would use Redis instead, where INCR + EXPIRE are both atomic single commands and every
 * server pointed at the same Redis instance shares these same counts.
 */
public final class InMemoryRateLimitStore implements RateLimitStore {

    private final ConcurrentHashMap<String, AtomicLong> counters = new ConcurrentHashMap<>();

    @Override
    public long incrementAndGet(String key) {
        return counters.computeIfAbsent(key, k -> new AtomicLong(0)).incrementAndGet();
    }

    @Override
    public long get(String key) {
        AtomicLong counter = counters.get(key);
        return counter == null ? 0 : counter.get();
    }
}

package com.practice.phonepe.questions.ratelimiter;

/** How many requests a single entity may make within a window -- configurable per entity. */
public final class RateLimitConfig {

    private final int maxRequests;
    private final long windowMillis;

    public RateLimitConfig(int maxRequests, long windowMillis) {
        this.maxRequests = maxRequests;
        this.windowMillis = windowMillis;
    }

    public int getMaxRequests() {
        return maxRequests;
    }

    public long getWindowMillis() {
        return windowMillis;
    }
}

package com.practice.phonepe.questions.ratelimiter;

public interface RateLimiter {
    /** Throws RateLimitExceededException if clientId is over its configured limit. */
    void allowRequest(String clientId) throws RateLimitExceededException;
}

package com.practice.phonepe.questions.ratelimiter;

/** Thrown when an entity has exceeded its allowed request count for the current window. */
public final class RateLimitExceededException extends RuntimeException {

    private final long retryAfterMillis;

    public RateLimitExceededException(String message, long retryAfterMillis) {
        super(message);
        this.retryAfterMillis = retryAfterMillis;
    }

    public long getRetryAfterMillis() {
        return retryAfterMillis;
    }
}

package com.practice.phonepe.questions.ratelimiter;

/**
 * Wherever the request counters actually live. A single JVM's in-memory map only protects
 * that one process; multiple servers need to share one store so a client can't reset their
 * quota just by hitting a different instance. Both methods here map directly onto atomic
 * Redis commands (INCR, GET) -- swapping InMemoryRateLimitStore for a Redis-backed
 * implementation is the entire change needed to make this distributed.
 */
public interface RateLimitStore {

    /** Atomically increments the counter for key and returns the new value. */
    long incrementAndGet(String key);

    /** Reads the current counter for key, or 0 if it doesn't exist yet. */
    long get(String key);
}

package com.practice.phonepe.questions.ratelimiter;

import com.practice.phonepe.questions.ratelimiter.strategy.RateLimitAlgorithm;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public final class SimpleRateLimiter implements RateLimiter {

    private final RateLimitStore store;
    private final RateLimitAlgorithm algorithm;
    private final RateLimitConfig defaultConfig;
    private final Map<String, RateLimitConfig> clientConfigs = new ConcurrentHashMap<>();

    public SimpleRateLimiter(RateLimitStore store, RateLimitAlgorithm algorithm, RateLimitConfig defaultConfig) {
        this.store = store;
        this.algorithm = algorithm;
        this.defaultConfig = defaultConfig;
    }

    /** Different entities can have a different count-per-window; unconfigured ones get the default. */
    public void configureClient(String clientId, RateLimitConfig config) {
        clientConfigs.put(clientId, config);
    }

    @Override
    public void allowRequest(String clientId) {
        RateLimitConfig config = clientConfigs.getOrDefault(clientId, defaultConfig);

        boolean allowed;
        try {
            allowed = algorithm.allowRequest(store, clientId, config.getMaxRequests(), config.getWindowMillis());
        } catch (RuntimeException storeFailure) {
            // High availability: the rate limiter exists to protect the servers behind it --
            // if its own store is unreachable, blocking every request would be a worse outage
            // than the one it's guarding against. Fail open and let the request through.
            System.err.println("Rate limit store unavailable for '" + clientId + "', failing open: "
                    + storeFailure.getMessage());
            return;
        }

        if (!allowed) {
            throw new RateLimitExceededException(
                    "Rate limit exceeded for '" + clientId + "': max " + config.getMaxRequests()
                            + " requests per " + config.getWindowMillis() + "ms",
                    config.getWindowMillis());
        }
    }
}

package com.practice.phonepe.questions.ratelimiter.strategy;

import com.practice.phonepe.questions.ratelimiter.RateLimitStore;

/**
 * Simplest possible algorithm: bucket time into fixed windows, count requests in the current
 * one. One weakness worth knowing: a client can burst up to 2x the limit right around a
 * window boundary (max at the tail of one window, max again at the head of the next).
 */
public final class FixedWindowCounterStrategy implements RateLimitAlgorithm {

    @Override
    public boolean allowRequest(RateLimitStore store, String clientId, int maxRequests, long windowMillis) {
        long windowStart = (System.currentTimeMillis() / windowMillis) * windowMillis;
        String key = clientId + ":" + windowStart;
        long count = store.incrementAndGet(key);
        return count <= maxRequests;
    }
}

package com.practice.phonepe.questions.ratelimiter.strategy;

import com.practice.phonepe.questions.ratelimiter.RateLimitStore;

public interface RateLimitAlgorithm {
    /** true if this request is allowed under maxRequests-per-windowMillis for clientId. */
    boolean allowRequest(RateLimitStore store, String clientId, int maxRequests, long windowMillis);
}

package com.practice.phonepe.questions.ratelimiter.strategy;

import com.practice.phonepe.questions.ratelimiter.RateLimitStore;

/**
 * Smooths out fixed window's boundary-burst problem without paying for a full request log:
 * estimate the current count as a weighted blend of the previous window's count and this
 * window's count so far, weighted by how far into the current window we are.
 */
public final class SlidingWindowCounterStrategy implements RateLimitAlgorithm {

    @Override
    public boolean allowRequest(RateLimitStore store, String clientId, int maxRequests, long windowMillis) {
        long now = System.currentTimeMillis();
        long currentWindowStart = (now / windowMillis) * windowMillis;
        long previousWindowStart = currentWindowStart - windowMillis;

        double elapsedInCurrent = now - currentWindowStart;
        double weightOfPrevious = (windowMillis - elapsedInCurrent) / (double) windowMillis;

        long previousCount = store.get(clientId + ":" + previousWindowStart);
        long currentCount = store.incrementAndGet(clientId + ":" + currentWindowStart);

        double estimatedCount = previousCount * weightOfPrevious + currentCount;
        return estimatedCount <= maxRequests;
    }
}

package com.practice.phonepe.questions.ratelimiter.strategy;

import com.practice.phonepe.questions.ratelimiter.RateLimitStore;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.ConcurrentHashMap;

/**
 * The simplest sliding window to reason about: keep every request's timestamp, evict
 * anything older than windowMillis, and just count what's left -- no weighting math, no
 * approximation, exact. The cost is memory: one entry per request currently inside the
 * window, instead of one counter.
 *
 * Like TokenBucketStrategy, this keeps its own local state rather than going through
 * RateLimitStore -- a list of timestamps doesn't fit a simple increment-a-counter contract.
 * A distributed version of this does exist (Redis sorted sets: ZADD + ZREMRANGEBYSCORE +
 * ZCARD), it just needs a different store shape than the one built here.
 */
public final class SlidingWindowLogStrategy implements RateLimitAlgorithm {

    private final ConcurrentHashMap<String, Deque<Long>> requestLogs = new ConcurrentHashMap<>();

    @Override
    public boolean allowRequest(RateLimitStore store, String clientId, int maxRequests, long windowMillis) {
        Deque<Long> timestamps = requestLogs.computeIfAbsent(clientId, id -> new ArrayDeque<>());

        synchronized (timestamps) {
            long now = System.currentTimeMillis();
            evictOld(timestamps, now, windowMillis);

            if (timestamps.size() < maxRequests) {
                timestamps.addLast(now);
                return true;
            }
            return false;
        }
    }

    private void evictOld(Deque<Long> timestamps, long now, long windowMillis) {
        while (!timestamps.isEmpty() && timestamps.peekFirst() <= now - windowMillis) {
            timestamps.pollFirst();
        }
    }
}

package com.practice.phonepe.questions.ratelimiter.strategy;


import com.practice.phonepe.questions.ratelimiter.RateLimitStore;

import java.util.concurrent.ConcurrentHashMap;

/**
 * Unlike the other two strategies, this one ignores the RateLimitStore parameter entirely.
 * Token bucket needs a compound read-modify-write -- tokens AND a last-refill timestamp,
 * recomputed together based on elapsed time -- and a simple incrementAndGet/get pair can't
 * express that atomically. So it keeps its own per-client state locally instead, which means
 * it's node-local only: two servers each running this strategy would enforce the limit
 * independently, not as one shared quota. Making it truly distributed would need either a
 * richer store primitive (an atomic compare-and-swap on a compound value) or a Redis Lua
 * script doing the refill-and-deduct server-side in a single round trip.
 *
 * maxRequests is treated as both the bucket's capacity (burst size) and the number of tokens
 * that refill over windowMillis -- so it fully refills once per window, same meaning as
 * "maxRequests per windowMillis" in the other two strategies.
 */
public final class TokenBucketStrategy implements RateLimitAlgorithm {

    private static final class Bucket {
        double availableTokens;
        long lastRefillTimestamp;

        Bucket(double availableTokens, long lastRefillTimestamp) {
            this.availableTokens = availableTokens;
            this.lastRefillTimestamp = lastRefillTimestamp;
        }
    }

    private final ConcurrentHashMap<String, Bucket> buckets = new ConcurrentHashMap<>();

    @Override
    public boolean allowRequest(RateLimitStore store, String clientId, int maxRequests, long windowMillis) {
        Bucket bucket = buckets.computeIfAbsent(clientId,
                id -> new Bucket(maxRequests, System.currentTimeMillis()));

        synchronized (bucket) {
            refill(bucket, maxRequests, windowMillis);
            if (bucket.availableTokens >= 1) {
                bucket.availableTokens -= 1;
                return true;
            }
            return false;
        }
    }

    private void refill(Bucket bucket, int maxRequests, long windowMillis) {
        long now = System.currentTimeMillis();
        long elapsed = now - bucket.lastRefillTimestamp;
        if (elapsed <= 0) {
            return;
        }
        double refillPerMillis = maxRequests / (double) windowMillis;
        bucket.availableTokens = Math.min(maxRequests, bucket.availableTokens + elapsed * refillPerMillis);
        bucket.lastRefillTimestamp = now;
    }
}

package com.practice.phonepe.questions.ratelimiter;

import com.practice.phonepe.questions.ratelimiter.strategy.FixedWindowCounterStrategy;
import com.practice.phonepe.questions.ratelimiter.strategy.SlidingWindowCounterStrategy;
import com.practice.phonepe.questions.ratelimiter.strategy.SlidingWindowLogStrategy;
import com.practice.phonepe.questions.ratelimiter.strategy.TokenBucketStrategy;

public final class Demo {

    public static void main(String[] args) {
        RateLimitStore store = new InMemoryRateLimitStore();

        System.out.println("-- Fixed window: client-A limited to 2 requests/sec --");
        SimpleRateLimiter fixedWindowLimiter =
                new SimpleRateLimiter(store, new FixedWindowCounterStrategy(), new RateLimitConfig(5, 1000));
        fixedWindowLimiter.configureClient("client-A", new RateLimitConfig(2, 1000));
        for (int i = 0; i < 4; i++) {
            attempt(fixedWindowLimiter, "client-A", i);
        }

        System.out.println("-- Sliding window: client-B limited to 3 requests/sec, different algorithm --");
        SimpleRateLimiter slidingWindowLimiter =
                new SimpleRateLimiter(store, new SlidingWindowCounterStrategy(), new RateLimitConfig(3, 1000));
        for (int i = 0; i < 5; i++) {
            attempt(slidingWindowLimiter, "client-B", i);
        }

        System.out.println("-- Sliding window log: client-E limited to 3 requests per 500ms, exact eviction --");
        SimpleRateLimiter slidingLogLimiter =
                new SimpleRateLimiter(store, new SlidingWindowLogStrategy(), new RateLimitConfig(3, 500));
        for (int i = 0; i < 4; i++) {
            attempt(slidingLogLimiter, "client-E", i); // burst: first 3 allowed, 4th denied
        }
        System.out.println("   waiting 520ms so all 3 timestamps age out of the 500ms window...");
        sleep(520);
        attempt(slidingLogLimiter, "client-E", 4); // window is empty again -> allowed

        System.out.println("-- Token bucket: client-D allows a burst of 3, then refills gradually --");
        SimpleRateLimiter tokenBucketLimiter =
                new SimpleRateLimiter(store, new TokenBucketStrategy(), new RateLimitConfig(3, 1000));
        for (int i = 0; i < 4; i++) {
            attempt(tokenBucketLimiter, "client-D", i); // burst: first 3 allowed, 4th denied
        }
        System.out.println("   waiting 350ms for ~1 token to refill (3 tokens / 1000ms)...");
        sleep(350);
        attempt(tokenBucketLimiter, "client-D", 4); // should be allowed again

        System.out.println("-- Distribution: two limiter instances (simulating two servers) sharing one store --");
        System.out.println("   client-C limited to 3 requests/sec, total across BOTH servers");
        RateLimitConfig sharedLimit = new RateLimitConfig(3, 1000);
        SimpleRateLimiter serverA = new SimpleRateLimiter(store, new FixedWindowCounterStrategy(), sharedLimit);
        SimpleRateLimiter serverB = new SimpleRateLimiter(store, new FixedWindowCounterStrategy(), sharedLimit);
        for (int i = 0; i < 4; i++) {
            boolean useServerA = i % 2 == 0;
            SimpleRateLimiter server = useServerA ? serverA : serverB;
            String serverName = useServerA ? "server-A" : "server-B";
            try {
                server.allowRequest("client-C");
                System.out.println("request " + i + " via " + serverName + " -> ALLOWED");
            } catch (RateLimitExceededException e) {
                System.out.println("request " + i + " via " + serverName + " -> DENIED: " + e.getMessage());
            }
        }
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private static void attempt(RateLimiter limiter, String clientId, int requestNumber) {
        try {
            limiter.allowRequest(clientId);
            System.out.println("request " + requestNumber + " -> ALLOWED");
        } catch (RateLimitExceededException e) {
            System.out.println("request " + requestNumber + " -> DENIED: " + e.getMessage());
        }
    }
}