Rate limiter sınıfı: token bucket'tan thread-safe çoklu kullanıcıya ve sliding window log'a

TokenBucket bucket = new TokenBucket(new TokenBucket.Limit(3, 1_000), clock.now());
bucket.tryConsume(0)      -> true, true, true   // 3'lük patlama
bucket.tryConsume(0)      -> false
bucket.tryConsume(999)    -> false              // henüz yeni token yok
bucket.tryConsume(1_000)  -> true

var limiter = new TokenBucketRateLimiter(
        user -> user.startsWith("premium:") ? new TokenBucket.Limit(5, 100) : new TokenBucket.Limit(2, 100),
        clock::now);
limiter.tryAcquire("free:1")     // 2 kez true
limiter.tryAcquire("premium:1")  // 5 kez true
limiter.evictIdle()              // dolu kovaları siler, silinen sayısını döner

while (true) {
    Slot<S> slot = slots.computeIfAbsent(userId, id -> new Slot<>(newState(id, now())));
    synchronized (slot) {
        if (slot.retired) continue;          // silinmiş: yenisini al
        return tryAcquire(slot.state, now());
    }
}

var log = new SlidingWindowLogRateLimiter(3, 10_000, clock::now);   // herhangi 10 sn'de en fazla 3
// t=0: 1 istek, t=4000: 2 istek -> doldu
// t=9999: reddedilir (t=0'daki istek hâlâ (−1, 9999] penceresinde)
// t=10000: kabul (t=0'daki istek pencereden çıktı)

/** Common contract for every limiter in this task. */
public interface RateLimiter {
    boolean tryAcquire(String userId);
}

import java.util.Objects;

/** Level 1: a single token bucket. Not thread-safe; callers synchronize. */
public final class TokenBucket {

    /** {@code capacity} tokens at most, one new token every {@code refillIntervalMillis}. */
    public record Limit(long capacity, long refillIntervalMillis) {
        public Limit {
            if (capacity <= 0 || refillIntervalMillis <= 0) {
                throw new IllegalArgumentException("capacity and refill interval must be positive");
            }
        }
    }

    private final Limit limit;
    private long tokens;
    private long lastRefillMillis;

    public TokenBucket(Limit limit, long nowMillis) {
        this.limit = Objects.requireNonNull(limit);
        this.tokens = limit.capacity();          // a new bucket starts full
        this.lastRefillMillis = nowMillis;
    }

    public boolean tryConsume(long nowMillis) {
        refill(nowMillis);
        if (tokens == 0) {
            return false;
        }
        tokens--;
        return true;
    }

    public boolean isFullAt(long nowMillis) {
        refill(nowMillis);
        return tokens == limit.capacity();
    }

    /** Lazy refill: compute what accumulated since the last refill instead of running a timer. */
    private void refill(long nowMillis) {
        long elapsed = nowMillis - lastRefillMillis;
        if (elapsed <= 0) {
            return;                               // same instant or clock moved back
        }
        long newTokens = elapsed / limit.refillIntervalMillis();
        if (newTokens == 0) {
            return;
        }
        if (newTokens >= limit.capacity() - tokens) {
            tokens = limit.capacity();            // full: drop the leftover time
            lastRefillMillis = nowMillis;
        } else {
            tokens += newTokens;                  // keep the fractional part of the interval
            lastRefillMillis += newTokens * limit.refillIntervalMillis();
        }
    }
}

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.LongSupplier;

/**
 * Levels 2-3: per-user state in a ConcurrentHashMap, one lock per user, safe eviction of idle state.
 * Subclasses only describe the algorithm (Template Method), which is what Level 4 reuses.
 */
public abstract class KeyedRateLimiter<S> implements RateLimiter {

    /** Holder that can be retired by eviction; a thread holding a stale holder must retry. */
    private static final class Slot<S> {
        final S state;
        boolean retired;                          // guarded by the slot's monitor

        Slot(S state) {
            this.state = state;
        }
    }

    private final ConcurrentHashMap<String, Slot<S>> slots = new ConcurrentHashMap<>();
    private final LongSupplier clockMillis;

    protected KeyedRateLimiter(LongSupplier clockMillis) {
        this.clockMillis = Objects.requireNonNull(clockMillis);
    }

    protected abstract S newState(String userId, long nowMillis);

    protected abstract boolean tryAcquire(S state, long nowMillis);

    /** True if dropping the state is indistinguishable from keeping it. */
    protected abstract boolean isIdle(S state, long nowMillis);

    @Override
    public final boolean tryAcquire(String userId) {
        Objects.requireNonNull(userId, "userId");
        while (true) {
            Slot<S> slot = slots.computeIfAbsent(userId, id -> new Slot<>(newState(id, clockMillis.getAsLong())));
            synchronized (slot) {                 // only this user's requests contend here
                if (slot.retired) {
                    continue;                     // evicted between lookup and lock: take the new slot
                }
                return tryAcquire(slot.state, clockMillis.getAsLong());
            }
        }
    }

    /** Removes idle per-user state; call periodically. Returns how many users were evicted. */
    public final int evictIdle() {
        int evicted = 0;
        for (var entry : slots.entrySet()) {
            Slot<S> slot = entry.getValue();
            synchronized (slot) {
                if (!slot.retired && isIdle(slot.state, clockMillis.getAsLong())) {
                    slot.retired = true;
                    slots.remove(entry.getKey(), slot);
                    evicted++;
                }
            }
        }
        return evicted;
    }

    public final int trackedUsers() {
        return slots.size();
    }
}

import java.util.Objects;
import java.util.function.Function;
import java.util.function.LongSupplier;

/** Levels 2-3: token bucket per user, limits chosen per user (e.g. by plan). */
public final class TokenBucketRateLimiter extends KeyedRateLimiter<TokenBucket> {

    private final Function<String, TokenBucket.Limit> limitPolicy;

    public TokenBucketRateLimiter(Function<String, TokenBucket.Limit> limitPolicy, LongSupplier clockMillis) {
        super(clockMillis);
        this.limitPolicy = Objects.requireNonNull(limitPolicy);
    }

    @Override
    protected TokenBucket newState(String userId, long nowMillis) {
        return new TokenBucket(limitPolicy.apply(userId), nowMillis);
    }

    @Override
    protected boolean tryAcquire(TokenBucket bucket, long nowMillis) {
        return bucket.tryConsume(nowMillis);
    }

    @Override
    protected boolean isIdle(TokenBucket bucket, long nowMillis) {
        return bucket.isFullAt(nowMillis);        // a full bucket behaves exactly like a new one
    }
}

import java.util.ArrayDeque;
import java.util.function.LongSupplier;

/** Level 4: exact limit of {@code maxRequests} in any sliding window of {@code windowMillis}. */
public final class SlidingWindowLogRateLimiter extends KeyedRateLimiter<ArrayDeque<Long>> {

    private final int maxRequests;
    private final long windowMillis;

    public SlidingWindowLogRateLimiter(int maxRequests, long windowMillis, LongSupplier clockMillis) {
        super(clockMillis);
        if (maxRequests <= 0 || windowMillis <= 0) {
            throw new IllegalArgumentException("maxRequests and windowMillis must be positive");
        }
        this.maxRequests = maxRequests;
        this.windowMillis = windowMillis;
    }

    @Override
    protected ArrayDeque<Long> newState(String userId, long nowMillis) {
        return new ArrayDeque<>(maxRequests);
    }

    @Override
    protected boolean tryAcquire(ArrayDeque<Long> log, long nowMillis) {
        pruneOlderThanWindow(log, nowMillis);
        if (log.size() >= maxRequests) {
            return false;
        }
        log.addLast(nowMillis);                   // only accepted requests are logged
        return true;
    }

    @Override
    protected boolean isIdle(ArrayDeque<Long> log, long nowMillis) {
        pruneOlderThanWindow(log, nowMillis);
        return log.isEmpty();
    }

    /** The window is (now - windowMillis, now]. */
    private void pruneOlderThanWindow(ArrayDeque<Long> log, long nowMillis) {
        while (!log.isEmpty() && log.peekFirst() <= nowMillis - windowMillis) {
            log.pollFirst();
        }
    }
}

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public final class RateLimitersTest {

    /** Controllable clock; tests never sleep. */
    static final class FakeClock {
        final AtomicLong millis = new AtomicLong();
        long now() { return millis.get(); }
        void set(long value) { millis.set(value); }
    }

    public static void main(String[] args) throws Exception {
        Map<String, Callable<Void>> tests = new LinkedHashMap<>();
        tests.put("level1_burstThenRefill", wrap(RateLimitersTest::level1_burstThenRefill));
        tests.put("level1_refillKeepsFractionAndCapsAtCapacity", wrap(RateLimitersTest::level1_refillKeepsFractionAndCapsAtCapacity));
        tests.put("level2_usersAreIsolatedAndPolicyPerUser", wrap(RateLimitersTest::level2_usersAreIsolatedAndPolicyPerUser));
        tests.put("level2_onlyFullBucketsAreEvicted", wrap(RateLimitersTest::level2_onlyFullBucketsAreEvicted));
        tests.put("level3_concurrentRequestsNeverExceedCapacity", RateLimitersTest::level3_concurrentRequestsNeverExceedCapacity);
        tests.put("level3_evictionDuringTrafficKeepsLimit", RateLimitersTest::level3_evictionDuringTrafficKeepsLimit);
        tests.put("level4_slidingLogIsExactAtWindowEdge", wrap(RateLimitersTest::level4_slidingLogIsExactAtWindowEdge));
        tests.put("level4_tokenBucketAllowsMoreInSameWindow", wrap(RateLimitersTest::level4_tokenBucketAllowsMoreInSameWindow));

        int passed = 0;
        for (var test : tests.entrySet()) {
            test.getValue().call();
            System.out.println("PASS " + test.getKey());
            passed++;
        }
        System.out.println(passed + "/" + tests.size() + " tests passed");
    }

    static void level1_burstThenRefill() {
        var clock = new FakeClock();
        var bucket = new TokenBucket(new TokenBucket.Limit(3, 1_000), clock.now());
        check(bucket.tryConsume(0) && bucket.tryConsume(0) && bucket.tryConsume(0), "burst of 3");
        check(!bucket.tryConsume(0), "4th rejected");
        check(!bucket.tryConsume(999), "no token before interval");
        check(bucket.tryConsume(1_000), "one token after interval");
        check(!bucket.tryConsume(1_000), "only one");
    }

    static void level1_refillKeepsFractionAndCapsAtCapacity() {
        var bucket = new TokenBucket(new TokenBucket.Limit(2, 1_000), 0);
        bucket.tryConsume(0);
        bucket.tryConsume(0);
        check(bucket.tryConsume(1_500), "1 token at 1500");
        // the 500 ms remainder is kept, so the next token arrives at 2000, not 2500
        check(bucket.tryConsume(2_000), "token at 2000");
        check(bucket.isFullAt(1_000_000), "long idle -> full, not overflowing");
        check(bucket.tryConsume(1_000_000) && bucket.tryConsume(1_000_000) && !bucket.tryConsume(1_000_000), "cap 2");
    }

    static void level2_usersAreIsolatedAndPolicyPerUser() {
        var clock = new FakeClock();
        var limiter = new TokenBucketRateLimiter(
                user -> user.startsWith("premium:") ? new TokenBucket.Limit(5, 100) : new TokenBucket.Limit(2, 100),
                clock::now);
        check(countAllowed(limiter, "free:1", 10) == 2, "free plan");
        check(countAllowed(limiter, "free:2", 10) == 2, "other user unaffected");
        check(countAllowed(limiter, "premium:1", 10) == 5, "premium plan");
    }

    static void level2_onlyFullBucketsAreEvicted() {
        var clock = new FakeClock();
        var limiter = new TokenBucketRateLimiter(user -> new TokenBucket.Limit(2, 1_000), clock::now);
        limiter.tryAcquire("a");
        limiter.tryAcquire("b");
        limiter.tryAcquire("b");
        check(limiter.evictIdle() == 0, "nobody is full yet");
        clock.set(1_000);   // a: 1+1 = 2 (full), b: 0+1 = 1
        check(limiter.evictIdle() == 1 && limiter.trackedUsers() == 1, "only a evicted");
        check(countAllowed(limiter, "a", 5) == 2, "evicted user starts with a full bucket, same as before");
    }

    static Void level3_concurrentRequestsNeverExceedCapacity() throws Exception {
        var clock = new FakeClock();                // frozen: no refill during the test
        var limiter = new TokenBucketRateLimiter(user -> new TokenBucket.Limit(100, 1_000), clock::now);
        int allowed = hammer(limiter, "hot-user", 8, 2_000, () -> {});
        check(allowed == 100, "exactly capacity, got " + allowed);
        return null;
    }

    static Void level3_evictionDuringTrafficKeepsLimit() throws Exception {
        var clock = new FakeClock();
        var limiter = new TokenBucketRateLimiter(user -> new TokenBucket.Limit(50, 1_000), clock::now);
        // Evicting a full bucket is lossless, so even with constant eviction the total stays 50.
        int allowed = hammer(limiter, "user", 8, 1_000, limiter::evictIdle);
        check(allowed == 50, "limit kept under eviction, got " + allowed);
        return null;
    }

    static void level4_slidingLogIsExactAtWindowEdge() {
        var clock = new FakeClock();
        var limiter = new SlidingWindowLogRateLimiter(3, 10_000, clock::now);
        clock.set(0);
        limiter.tryAcquire("u");
        clock.set(4_000);
        limiter.tryAcquire("u");
        limiter.tryAcquire("u");
        check(!limiter.tryAcquire("u"), "3 in window");
        clock.set(9_999);
        check(!limiter.tryAcquire("u"), "first request still inside (0, 9999]");
        clock.set(10_000);
        check(limiter.tryAcquire("u"), "request at 0 left the window at 10000");
        check(!limiter.tryAcquire("u"), "full again");
    }

    static void level4_tokenBucketAllowsMoreInSameWindow() {
        // Both are configured as "5 requests per 10 seconds"; one request every 100 ms for 10 s.
        var clock = new FakeClock();
        var bucket = new TokenBucketRateLimiter(user -> new TokenBucket.Limit(5, 2_000), clock::now);
        var log = new SlidingWindowLogRateLimiter(5, 10_000, clock::now);
        int bucketAllowed = 0;
        int logAllowed = 0;
        for (long t = 0; t < 10_000; t += 100) {
            clock.set(t);
            if (bucket.tryAcquire("u")) bucketAllowed++;
            if (log.tryAcquire("u")) logAllowed++;
        }
        check(logAllowed == 5, "log is exact: " + logAllowed);
        check(bucketAllowed == 9, "bucket = burst 5 + refills at 2s,4s,6s,8s: " + bucketAllowed);
    }

    // ---------- helpers ----------

    static int countAllowed(RateLimiter limiter, String user, int attempts) {
        int allowed = 0;
        for (int i = 0; i < attempts; i++) {
            if (limiter.tryAcquire(user)) allowed++;
        }
        return allowed;
    }

    /** Runs threads * callsPerThread requests at once while {@code background} runs in a loop. */
    static int hammer(RateLimiter limiter, String user, int threads, int callsPerThread, Runnable background)
            throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(threads + 1);
        try {
            var start = new CountDownLatch(1);
            var done = new AtomicBoolean(false);
            Future<?> backgroundTask = pool.submit(() -> {
                while (!done.get()) background.run();
            });
            List<Future<Integer>> results = new ArrayList<>();
            for (int i = 0; i < threads; i++) {
                results.add(pool.submit(() -> {
                    start.await();
                    return countAllowed(limiter, user, callsPerThread);
                }));
            }
            start.countDown();
            int total = 0;
            for (Future<Integer> f : results) total += f.get();
            done.set(true);
            backgroundTask.get();
            return total;
        } finally {
            pool.shutdownNow();
        }
    }

    static Callable<Void> wrap(Runnable test) {
        return () -> { test.run(); return null; };
    }

    static void check(boolean condition, String message) {
        if (!condition) {
            throw new AssertionError(message);
        }
    }
}