LRU Cache Implementation & Rate Limiter

LRU Cache: concept & data-structure choice ▶ 0:00:02 – 0:19:40

LRU = "Least Recently Used". When the cache is full, evict the item that has gone untouched the longest. Keep the order in a doubly linked list and find nodes through a hash map, so both get and put are O(1).

The problem

A cache is fast in-memory storage with a limited capacity. While there is space, just insert. When a new item arrives and the cache is full, the replacement strategy decides who leaves. LRU removes the least recently accessed item. Here "access" means a get or a put of that key.

Building it step by step

1. Only a hash map? Lookup is O(1), but the map has no idea which key is the oldest. Finding it means scanning everything, which is O(n).
2. Hash map + an ordered list? Keep keys in order from LRU at one end to MRU at the other. On access, move the key to the MRU end. On overflow, remove from the LRU end.
3. Which list? An array needs O(n) to remove from the middle. Ganesh asked this in class. A singly linked list cannot unlink a node unless you know the node before it, so you would scan again. A doubly linked list has a prev pointer, so unlinking any node is O(1).
4. How do we reach a node in O(1)? The hash map stores key → node, so it holds a reference (the node's address) into the linked list. Harshit asked this in class. "Where is my Aadhaar card? I go straight to its drawer, I don't search every document."

Walk-through (capacity 3, head = MRU)

Every step uses three O(1) operations: remove node, add to head, and remove last. "Move to head" is just remove + add to head.

LRU Cache: implementation ▶ 0:19:40 – 1:21:24

Three small classes: Node, DoublyLinkedList (add_to_head, remove_node, move_to_head) and LRUCache (get, put), with dummy head and tail nodes.

Design decisions the teacher made

• Write your own doubly linked list. Interviewers usually will not let you use a library for it, and it is short. Build it from first principles so you can rebuild it under pressure.
• Dummy (sentinel) head and tail nodes. The most common linked-list bug is the null-pointer exception on empty-list and one-node edge cases. With two dummy nodes that always exist, every real node always has a real prev and next, so there are no None checks. The list is empty when head.next is tail.
• Add at the head, evict from the tail. The teacher first considered "add at the tail" and then switched. Inserting at the head is the more natural habit.
• Unit operations: add_to_head and remove_node. move_to_head = remove + add. Small pieces mean fewer pointer bugs.
• Size comes from the hash map: len(lookup) is the number of entries. No separate counter is needed.
• Add first, then evict if over capacity. The other order (check, evict, then add) is equally correct. The teacher prefers "add, then trim", the same trick used with heaps in Top-K problems.

Pointer surgery, visualised

add_to_head(node), going from head ⇄ first to head ⇄ node ⇄ first. There are 4 links to set:

1. node.next = head.next (secure the old first node before you lose it)
2. node.prev = head
3. head.next.prev = node (old first points back to node)
4. head.next = node

remove_node(node), going from A ⇄ node ⇄ B to A ⇄ B. There are 2 links to set:

1. node.prev.next = node.next
2. node.next.prev = node.prev

Python code (no OrderedDict)

from typing import Optional


class Node:
    def __init__(self, key: int = 0, value: int = 0):
        self.key = key            # key is stored so eviction can delete it from the map
        self.value = value
        self.prev: Optional["Node"] = None
        self.next: Optional["Node"] = None


class DoublyLinkedList:
    """head side = most recently used (MRU), tail side = least recently used (LRU)."""
    def __init__(self):
        self.head = Node()        # dummy (sentinel) nodes: no None checks anywhere
        self.tail = Node()
        self.head.next = self.tail
        self.tail.prev = self.head          # the line the teacher first forgot

    def add_to_head(self, node: Node) -> None:
        # 4 pointer changes: node <-> old first node, and head <-> node
        node.next = self.head.next
        node.prev = self.head
        self.head.next.prev = node
        self.head.next = node

    def remove_node(self, node: Node) -> None:
        node.prev.next = node.next
        node.next.prev = node.prev

    def move_to_head(self, node: Node) -> None:
        self.remove_node(node)              # move = remove + add to head
        self.add_to_head(node)

    def remove_last(self) -> Node:
        last = self.tail.prev               # real last node sits just before dummy tail
        self.remove_node(last)
        return last

    def print(self) -> None:
        cur, out = self.head.next, []
        while cur is not self.tail:         # tail (not None) is our stop marker
            out.append(f"{cur.key}:{cur.value}")
            cur = cur.next
        print(" -> ".join(out) or "(empty)")


class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = DoublyLinkedList()         # keeps the ORDER
        self.lookup: dict[int, Node] = {}       # key -> node, for O(1) jumps

    def get(self, key: int) -> int:
        if key not in self.lookup:
            return -1
        node = self.lookup[key]
        self.cache.move_to_head(node)           # reading also counts as "using"
        return node.value

    def put(self, key: int, value: int) -> None:
        if key in self.lookup:                  # existing key: update + make it MRU
            node = self.lookup[key]
            node.value = value
            self.cache.move_to_head(node)
            return
        node = Node(key, value)                 # new key: add first ...
        self.lookup[key] = node
        self.cache.add_to_head(node)
        if len(self.lookup) > self.capacity:    # ... then evict if we went over
            lru = self.cache.remove_last()
            del self.lookup[lru.key]            # remove from BOTH places


# usage
c = LRUCache(3)
c.put(1, 10); c.put(2, 20); c.put(3, 100)
c.cache.print()          # 3:100 -> 2:20 -> 1:10
c.put(4, 200)            # full -> evicts key 1 (least recently used)
c.cache.print()          # 4:200 -> 3:100 -> 2:20
c.put(2, 40)             # update -> 2 jumps to the front
print(c.get(3))          # 100  (3 moves to the front)
c.cache.print()          # 3:100 -> 2:40 -> 4:200
print(c.get(1))          # -1

classDiagram
  class LRUCache {
    +capacity
    +cache
    +lookup
    +get(key)
    +put(key, value)
  }
  class DoublyLinkedList {
    +head
    +tail
    +add_to_head(node)
    +remove_node(node)
    +move_to_head(node)
    +remove_last()
  }
  class Node {
    +key
    +value
    +prev
    +next
  }
  LRUCache *-- DoublyLinkedList
  DoublyLinkedList *-- Node
  LRUCache --> Node

LRUCache has a DoublyLinkedList, and the list dies with the cache, so this is composition. Ajay answered this in class.

Bugs hit during live coding (and the fixes)

The Python shortcut: OrderedDict

Python's collections.OrderedDict is a hash map plus a doubly linked list internally, so LRU becomes a few lines. Mention it in the interview, but expect to be asked to write the manual version.

from collections import OrderedDict


class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.data: OrderedDict[int, int] = OrderedDict()   # remembers order

    def get(self, key: int) -> int:
        if key not in self.data:
            return -1
        self.data.move_to_end(key)          # end = most recently used
        return self.data[key]

    def put(self, key: int, value: int) -> None:
        self.data[key] = value
        self.data.move_to_end(key)
        if len(self.data) > self.capacity:
            self.data.popitem(last=False)   # front = least recently used


# usage
c = LRUCache(2)
c.put(1, 1); c.put(2, 2); c.get(1); c.put(3, 3)
print(list(c.data))      # [1, 3]  (2 was evicted)

Rate Limiter: what and why ▶ 1:21:24 – 1:30:00 · re-recorded intro 26 Apr 0:00:05 – 0:01:30

A rate limiter controls how many requests a user (or IP, or API key) can make in a given time window, and rejects the rest.

The problem

Every API call costs you compute and database work. Normal users call your "fetch weather" API now and then. An abuser runs while True: call_api() for five days. The outside world is "cruel". Without protection, abusers (or bots) consume your resources and real users are starved. This also covers a DDoS (Distributed Denial of Service) attack: many machines everywhere send something like 1,000 requests per second, so a real user cannot even open a site such as the CBSE results page.

Seen in real life: OTP "resend after 1 minute" and "too many attempts, blocked for 30 minutes". Instagram's "too many requests" when posting or commenting too fast. Shubham's company saw a hacker hitting the OTP API back-to-back to brute-force codes. Stripe, AWS, GitHub, Twitter, NGINX and Redis all use rate limiting.

Core goals: protect the system from abuse, overload and DDoS, and share capacity fairly between users.

Where it sits (high-level picture)

1. Clients → API gateway or proxy. Every request passes through here. Gateways often provide generic rate limiting out of the box.
2. A custom rate-limiter service checks rules stored in a DB or config: per API, per customer tier (premium users get higher limits), stricter rules for sensitive endpoints.
3. Allowed requests go on to the microservices. Rejected ones get HTTP 429 Too Many Requests.

Companies build their own limiter instead of using only the gateway's because they want customised rules, such as a premium-customer experience, not just DDoS protection. Rate limiting is also a full HLD topic of its own.

Token Bucket algorithm ▶ 1:30:00 – 1:37:43

Each user has a bucket of tokens. A request spends one token. Tokens drip back in at a fixed rate, up to a maximum capacity.

State per bucket

• capacity: the most tokens the bucket can ever hold (the allowed burst size).
• tokens: tokens available right now. It starts equal to capacity.
• refill rate: tokens added per unit of time, e.g. 5 per minute.
• last refill time: needed to work out how many tokens were earned since then.

Algorithm (lazy refill)

1. A request arrives for a user. Look up (or create) their bucket.
2. Refill first: earned = (now - last_refill) × rate, then tokens = min(capacity, tokens + earned), then last_refill = now. Nothing runs in the background. The refill is calculated only when the user shows up.
3. If tokens ≥ 1: spend one and allow. Otherwise reject (429).

Teacher's numbers: capacity 10, rate 5 per minute. The user drained the bucket to 0 at 6:00. They return at 6:15 and have earned 15 × 5 = 75 tokens, but get min(10, 0 + 75) = 10. If they return at 6:01 instead, they get only 5. The min with capacity is what stops someone from "saving up" a huge burst.

See TokenBucketStrategy in the code in the next section.

Other algorithms & the LLD (Strategy + Factory) ▶ 1:39:32 – 1:41:07

The rate-limiter service holds one interchangeable RateLimitingStrategy (Strategy pattern). A factory builds the chosen algorithm from config.

The teacher named the algorithms to read up on (token bucket, fixed window, sliding window) and promised the sliding window with a double-ended queue for the next session. That session was the lost 26 Apr recording, so the rows marked "not in recording" below are standard reference material.

LLD: how the patterns fit

• Strategy: RateLimitingStrategy is an interface with allow_request(user_id). Token bucket, fixed window and sliding log are interchangeable implementations. RateLimiter depends only on the interface (Dependency Inversion) and can swap algorithms at runtime.
• Factory: RateLimiterFactory.create("token_bucket", ...) hides the construction details. The algorithm name can come from the rules DB or config.
• Thread safety (NFR): many requests for the same user can arrive at the same time, so each strategy guards its counters with a lock.

import threading
import time
from abc import ABC, abstractmethod
from collections import deque


class RateLimitingStrategy(ABC):
    @abstractmethod
    def allow_request(self, user_id: str) -> bool: ...


class TokenBucketStrategy(RateLimitingStrategy):
    """Each user has a bucket of `capacity` tokens, refilled at `refill_rate` tokens/sec."""

    class Bucket:
        def __init__(self, capacity: int):
            self.tokens = float(capacity)          # start full
            self.last_refill = time.monotonic()

    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.buckets: dict[str, TokenBucketStrategy.Bucket] = {}
        self.lock = threading.Lock()

    def _refill(self, b: "TokenBucketStrategy.Bucket") -> None:
        # Lazy refill: computed only when the user shows up
        now = time.monotonic()
        earned = (now - b.last_refill) * self.refill_rate
        b.tokens = min(self.capacity, b.tokens + earned)   # never above capacity
        b.last_refill = now

    def allow_request(self, user_id: str) -> bool:
        with self.lock:
            b = self.buckets.setdefault(user_id, self.Bucket(self.capacity))
            self._refill(b)
            if b.tokens >= 1:
                b.tokens -= 1
                return True
            return False


class FixedWindowStrategy(RateLimitingStrategy):
    """At most `limit` requests per user in each fixed window (e.g. 10:00:00-10:00:59)."""

    def __init__(self, limit: int, window_sec: float):
        self.limit = limit
        self.window = window_sec
        self.counters: dict[str, tuple[int, int]] = {}   # user -> (window_id, count)
        self.lock = threading.Lock()

    def allow_request(self, user_id: str) -> bool:
        with self.lock:
            window_id = int(time.monotonic() // self.window)
            wid, count = self.counters.get(user_id, (window_id, 0))
            if wid != window_id:              # a new window started -> reset
                count = 0
            if count < self.limit:
                self.counters[user_id] = (window_id, count + 1)
                return True
            self.counters[user_id] = (window_id, count)
            return False


class SlidingWindowLogStrategy(RateLimitingStrategy):
    """Keep recent request times in a deque; allow if fewer than `limit` in the last window."""

    def __init__(self, limit: int, window_sec: float):
        self.limit = limit
        self.window = window_sec
        self.logs: dict[str, deque[float]] = {}
        self.lock = threading.Lock()

    def allow_request(self, user_id: str) -> bool:
        with self.lock:
            now = time.monotonic()
            log = self.logs.setdefault(user_id, deque())
            while log and log[0] <= now - self.window:   # drop times that slid out
                log.popleft()
            if len(log) < self.limit:
                log.append(now)
                return True
            return False


class RateLimiterFactory:
    @staticmethod
    def create(kind: str, **cfg) -> RateLimitingStrategy:
        if kind == "token_bucket":
            return TokenBucketStrategy(cfg["capacity"], cfg["refill_rate"])
        if kind == "fixed_window":
            return FixedWindowStrategy(cfg["limit"], cfg["window_sec"])
        if kind == "sliding_log":
            return SlidingWindowLogStrategy(cfg["limit"], cfg["window_sec"])
        raise ValueError(f"unknown strategy: {kind}")


class RateLimiter:
    """The service in front of the API. It does not care WHICH algorithm is plugged in."""

    def __init__(self, strategy: RateLimitingStrategy):
        self.strategy = strategy

    def set_strategy(self, strategy: RateLimitingStrategy) -> None:
        self.strategy = strategy             # swap the algorithm at runtime

    def handle(self, user_id: str) -> str:
        return "200 OK" if self.strategy.allow_request(user_id) else "429 Too Many Requests"


# usage
limiter = RateLimiter(RateLimiterFactory.create("token_bucket", capacity=3, refill_rate=1))
print([limiter.handle("abuser") for _ in range(5)])
# ['200 OK', '200 OK', '200 OK', '429 Too Many Requests', '429 Too Many Requests']
print(limiter.handle("real_user"))      # 200 OK  (own bucket, the abuser cannot starve it)
time.sleep(1.1)
print(limiter.handle("abuser"))         # 200 OK  (about 1 token refilled after ~1 s)

limiter.set_strategy(RateLimiterFactory.create("sliding_log", limit=2, window_sec=1))
print([limiter.handle("u1") for _ in range(3)])
# ['200 OK', '200 OK', '429 Too Many Requests']

classDiagram
  class RateLimitingStrategy {
    +allow_request(user_id) bool
  }
  class TokenBucketStrategy {
    +capacity
    +refill_rate
    +buckets
    +allow_request(user_id) bool
  }
  class FixedWindowStrategy {
    +limit
    +window
    +counters
    +allow_request(user_id) bool
  }
  class SlidingWindowLogStrategy {
    +limit
    +window
    +logs
    +allow_request(user_id) bool
  }
  class RateLimiter {
    +strategy
    +set_strategy(strategy)
    +handle(user_id) str
  }
  class RateLimiterFactory {
    +create(kind) RateLimitingStrategy
  }
  TokenBucketStrategy ..|> RateLimitingStrategy
  FixedWindowStrategy ..|> RateLimitingStrategy
  SlidingWindowLogStrategy ..|> RateLimitingStrategy
  RateLimiter o-- RateLimitingStrategy
  RateLimiterFactory --> RateLimitingStrategy