The limit that multiplied by three

import java.util.*;

/**
 * Solution: the limit that multiplied by three.
 *
 * Four defects, and only one of them is about the algorithm. The other three
 * are about where the state lives, what it is keyed by, and when it is thrown
 * away — which is the pattern for rate limiters generally. None of the four
 * produced an error in staging, because staging runs one instance with a
 * handful of keys.
 */
public class Solution {

    /* ── infrastructure (unchanged) ─────────────────────────────────────── */

    static final class Ticker {
        private long millis;
        Ticker(long start) { this.millis = start; }
        long now() { return millis; }
        void advance(long ms) { millis += ms; }
    }

    record Request(String apiKey, String sourceIp) {}

    static final class SharedStore {
        private final Map<String, double[]> state = new HashMap<>();

        synchronized boolean consume(String key, double capacity, double perMs, long now) {
            double[] s = state.computeIfAbsent(key, k -> new double[] { capacity, now });
            s[0] = Math.min(capacity, s[0] + (now - s[1]) * perMs);
            s[1] = now;
            if (s[0] >= 1.0) { s[0] -= 1.0; return true; }
            return false;
        }

        synchronized void expire(long now, long maxIdleMs) {
            state.entrySet().removeIf(e -> now - (long) e.getValue()[1] > maxIdleMs);
        }

        synchronized int trackedKeys() { return state.size(); }
    }

    /* ── the limiter, fixed ─────────────────────────────────────────────── */

    static final int LIMIT_PER_MINUTE = 100;
    static final long WINDOW_MS = 60_000;

    static final class RateLimiter {
        private final Ticker clock;
        private final SharedStore store;

        /*
         * Defect 1. The counter lived in a HashMap on the instance, so each of
         * the three replicas enforced the full limit independently and the
         * documented 100/min was really 300/min. Worse, it loosened on every
         * scale-out — the protection weakened exactly when load was highest.
         *
         * Dividing the limit by the replica count is the tempting fix and is
         * wrong: it holds only while routing is even, and it has to be
         * re-tuned on every scale event. The state has to be shared.
         */
        RateLimiter(Ticker clock, SharedStore store) {
            this.clock = clock;
            this.store = store;
        }

        /*
         * Defect 2. A fixed window resets on an aligned boundary, so a client
         * could spend the whole limit at 59.999s and the whole limit again at
         * 60.000s. Clock-aligned traffic — cron, rounded retry backoffs,
         * dashboards — lands exactly there, so it was not a rare case.
         *
         * A token bucket has no boundary to exploit: the refill is continuous,
         * computed from elapsed time, and capacity bounds the burst explicitly.
         *
         * Defect 3. The key was the source IP, so every customer behind one NAT
         * gateway or corporate proxy shared a single quota. The limit is sold
         * per API key, so the API key is what it must be counted against.
         * (An IP limit is still useful as a separate, coarser layer against
         * volumetric abuse — but it is not this limit.)
         */
        boolean allow(Request request) {
            double capacity = LIMIT_PER_MINUTE;
            double perMs = LIMIT_PER_MINUTE / (double) WINDOW_MS;
            return store.consume(request.apiKey(), capacity, perMs, clock.now());
        }

        /*
         * Defect 4. The per-instance map was never pruned, so it retained an
         * entry for every key the process had ever seen — attacker-controlled
         * growth on a public endpoint. State now lives in the shared store,
         * which expires idle keys; nothing is retained here.
         *
         * The expiry has to be LONGER than the window. An evicted key comes
         * back with a full bucket, so expiring at exactly the window boundary
         * would hand out a free burst to anyone who paused briefly.
         */
        int trackedKeys() { return 0; }
    }

    static final class Instance {
        final RateLimiter limiter;
        Instance(Ticker clock, SharedStore store) { this.limiter = new RateLimiter(clock, store); }
    }

    static List<Instance> cluster(int n, Ticker clock, SharedStore store) {
        List<Instance> out = new ArrayList<>();
        for (int i = 0; i < n; i++) out.add(new Instance(clock, store));
        return out;
    }

    /* ── checks (unchanged from the starter) ────────────────────────────── */

    public static void main(String[] args) {
        List<String> failures = new ArrayList<>();
        final int CEILING = LIMIT_PER_MINUTE + 10;

        {
            var clock = new Ticker(0);
            var store = new SharedStore();
            var nodes = cluster(3, clock, store);
            var req = new Request("key-a", "10.0.0.1");
            int allowed = 0;
            for (int i = 0; i < 300; i++) if (nodes.get(i % 3).limiter.allow(req)) allowed++;
            if (allowed > CEILING)
                failures.add("1. three instances allowed " + allowed + " for a limit of "
                           + LIMIT_PER_MINUTE + " — the limit scales with the replica count");
        }

        {
            var clock = new Ticker(0);
            var store = new SharedStore();
            var nodes = cluster(3, clock, store);
            var req = new Request("key-b", "10.0.0.2");
            clock.advance(59_000);
            int allowed = 0;
            for (int i = 0; i < 100; i++) if (nodes.get(i % 3).limiter.allow(req)) allowed++;
            clock.advance(1_000);
            for (int i = 0; i < 100; i++) if (nodes.get(i % 3).limiter.allow(req)) allowed++;
            if (allowed > CEILING)
                failures.add("2. " + allowed + " requests allowed inside one second across a "
                           + "window boundary, for a limit of " + LIMIT_PER_MINUTE + "/min");
        }

        {
            var clock = new Ticker(0);
            var store = new SharedStore();
            var nodes = cluster(1, clock, store);
            var one = new Request("key-c", "203.0.113.9");
            var two = new Request("key-d", "203.0.113.9");
            int allowedOne = 0, allowedTwo = 0;
            for (int i = 0; i < 100; i++) {
                if (nodes.get(0).limiter.allow(one)) allowedOne++;
                if (nodes.get(0).limiter.allow(two)) allowedTwo++;
            }
            if (allowedOne < 90 || allowedTwo < 90)
                failures.add("3. two API keys behind one IP got " + allowedOne + " and "
                           + allowedTwo + " of 100 — they are sharing a quota");
        }

        {
            var clock = new Ticker(0);
            var store = new SharedStore();
            var nodes = cluster(1, clock, store);
            for (int i = 0; i < 50_000; i++)
                nodes.get(0).limiter.allow(new Request("key-" + i, "10.1." + (i / 256) + "." + (i % 256)));
            clock.advance(WINDOW_MS * 10);
            store.expire(clock.now(), WINDOW_MS * 2);
            int tracked = nodes.get(0).limiter.trackedKeys() + store.trackedKeys();
            if (tracked > 1_000)
                failures.add("4. still holding state for " + tracked + " keys ten windows after "
                           + "they went idle — this grows with every key ever seen");
        }

        if (failures.isEmpty()) {
            System.out.println("PASS");
        } else {
            failures.forEach(f -> System.out.println("  " + f));
            System.out.println("FAIL");
        }
    }
}