Design a leaderboard class

01The design, in the six-step framework

1. Requirements and use cases

LeetCode 1244 has a three-method class, and an interviewing.io mock with a Google engineer steered the candidate towards keeping a binary heap inside it. Everything here turns on a trade-off: writes are frequent, reads ask for the top k, and scores change, which is what makes a plain heap awkward.

• A game adds to a player's score; a new player starts from zero.
• A caller asks for the sum of the top k scores, the LeetCode contract, or the top k players with scores.
• A caller resets a player, removing them from the board.
• Ties break deterministically, by player id, so two calls agree.
• A caller can read a player's score.

Out of scope: persistence, time-windowed boards, sharding across machines, which the system design version covers. Constraints: add_score is O(log n); top is O(k log n) with k far smaller than n; memory stays O(n) even after millions of score updates, so stale heap entries must be cleaned up; the internals are hidden so the heap can be replaced by a balanced tree.

2. Core entities and relationships

• Leaderboard: the public class; it holds the current scores and a max-heap with lazy deletion.
• Entry: an immutable (score, player) result returned to callers.
• ScanLeaderboard: the simplest correct implementation, kept as an interview baseline and a test oracle.
• Ranking: the protocol both implement, so callers depend on the promise and not the structure.

There are no composition edges to speak of, and that is right: the structures are private fields, not classes. A Player class holding one integer would be ceremony.

3. Interfaces and key methods

import heapq, random, threading
from dataclasses import dataclass
from typing import List, Protocol

@dataclass(frozen=True)
class Entry:
    player: str
    score: int

class Ranking(Protocol):
    def add_score(self, player: str, delta: int) -> int:
        """Add delta to the player's score; returns the new score."""
    def score(self, player: str) -> int:
        """The current score, 0 if unknown."""
    def top(self, k: int) -> List[Entry]:
        """The k best, by score then player id. Does not change the ranking."""
    def top_sum(self, k: int) -> int:
        """Sum of the top k scores (the LeetCode contract)."""
    def reset(self, player: str) -> None:
        """Remove the player from the board."""

Invariants: scores is the only source of truth; a heap entry is valid exactly when its score equals scores[player]; every live player has at least one valid heap entry; top leaves the set of valid entries unchanged; the heap size stays within a constant factor of the player count.

4. Design patterns and the choices behind them

Choosing a structure by the operation mix

State the mix first: if reads are rare, the scan wins and the extra code is not justified. The heap version matches what the mock interviewer asked for, and the tree is the answer to "what would you really ship". Knowing all three is the point of priority queues in object design.

Lazy deletion

The force: a heap cannot update or delete an arbitrary entry cheaply. So do not try. On each change, push a new entry and leave the old one in place. An entry is stale if its score no longer matches the dictionary. top pops until it has k valid entries, discarding stale ones for good, then pushes the valid ones back. Each stale entry is removed only once, so its cost is charged to the update that created it. Duplicates need care: a player may have the same score twice after going up and down, so a seen set makes sure each player appears only once in the result.

Compaction keeps memory bounded

A player updated a million times leaves a million entries if top is never called. When the heap exceeds twice the live count plus a floor, rebuild it from the dictionary in O(n). Doubling means the rebuild is amortised O(1) per update, the same argument as a growing array.

A hidden structure behind a small API

Callers see five methods and an Entry; nothing returns the heap. That encapsulation is what lets the tree replace the heap later, and the method design keeps top free of visible side effects even though it reorders internals. ScanLeaderboard earns its place as a test oracle: run random operations on both and compare.

5. Core flow and code

1. add_score updates the dictionary and pushes a new heap entry with a negated score, because heapq is a min-heap.
2. If the heap has grown too large, _compact rebuilds it.
3. top(k) pops entries, skipping stale ones and players already seen, until it has k.
4. It pushes the k valid entries back and returns them.
5. reset deletes the player from the dictionary; their heap entries become stale automatically.

class Leaderboard:
    MIN_COMPACT = 64

    def __init__(self) -> None:
        self.scores: dict = {}
        self.heap: list = []            # (-score, player)
        self._lock = threading.Lock()

    def score(self, player: str) -> int:
        return self.scores.get(player, 0)

    def add_score(self, player: str, delta: int) -> int:
        with self._lock:
            new = self.scores.get(player, 0) + delta
            self.scores[player] = new
            heapq.heappush(self.heap, (-new, player))
            if len(self.heap) > max(self.MIN_COMPACT, 2 * len(self.scores)):
                self._compact()
            return new

    def _compact(self) -> None:
        self.heap = [(-s, p) for p, s in self.scores.items()]
        heapq.heapify(self.heap)

    def _valid(self, item: tuple) -> bool:
        neg, player = item
        return self.scores.get(player) == -neg

    def top(self, k: int) -> List[Entry]:
        if k < 0:
            raise ValueError("k must not be negative")
        with self._lock:
            result, kept, seen = [], [], set()
            while self.heap and len(result) < k:
                item = heapq.heappop(self.heap)
                neg, player = item
                if self._valid(item) and player not in seen:
                    seen.add(player)
                    kept.append(item)
                    result.append(Entry(player, -neg))
            for item in kept:
                heapq.heappush(self.heap, item)
            return result

    def top_sum(self, k: int) -> int:
        return sum(e.score for e in self.top(k))

    def reset(self, player: str) -> None:
        with self._lock:
            self.scores.pop(player, None)

class ScanLeaderboard:
    """Baseline: O(1) writes, O(n log k) reads. Also the oracle in tests."""
    def __init__(self) -> None:
        self.scores: dict = {}

    def score(self, player: str) -> int:
        return self.scores.get(player, 0)

    def add_score(self, player: str, delta: int) -> int:
        self.scores[player] = self.scores.get(player, 0) + delta
        return self.scores[player]

    def top(self, k: int) -> List[Entry]:
        best = heapq.nsmallest(k, self.scores.items(), key=lambda kv: (-kv[1], kv[0]))
        return [Entry(p, s) for p, s in best]

    def top_sum(self, k: int) -> int:
        return sum(e.score for e in self.top(k))

    def reset(self, player: str) -> None:
        self.scores.pop(player, None)

if __name__ == "__main__":
    lb = Leaderboard()
    for player, delta in [(1, 73), (2, 56), (3, 39), (4, 51), (5, 4)]:
        lb.add_score(str(player), delta)
    assert lb.top_sum(1) == 73
    lb.reset("1"); lb.reset("2")
    lb.add_score("2", 51)
    assert lb.top_sum(3) == 141             # LeetCode's sample: 51 + 51 + 39

    rng, a, b = random.Random(1), Leaderboard(), ScanLeaderboard()
    for _ in range(5000):                   # model test against the oracle
        player, op = str(rng.randint(0, 30)), rng.random()
        if op < 0.1:
            a.reset(player); b.reset(player)
        elif op < 0.9:
            delta = rng.randint(-5, 20)
            a.add_score(player, delta); b.add_score(player, delta)
        else:
            k = rng.randint(0, 10)
            assert a.top(k) == b.top(k)
    assert len(a.heap) <= max(a.MIN_COMPACT, 2 * len(a.scores)) + 1

6. Extensions, concurrency and what interviewers probe

• "Why does top push entries back?" Popping is the only way to read a heap in order, and the next call needs those entries. Discarded stale entries stay gone, which is what makes the cost amortised.
• "What rank is this player?" A heap cannot answer. A balanced tree with subtree counts does in O(log n), which is the strongest argument for it.
• "Scores only ever increase." Then a player's old entries are always lower and can never be mistaken for current ones, and a fixed-size top-k heap maintained on every update gives O(1) reads.
• "Weekly boards." Key the structure by period, one Leaderboard per week, and drop whole objects; expiring individual entries is the harder way round.
• Concurrency: top mutates the heap even though it looks like a read, so a read-write lock would be a bug here; one exclusive lock is correct. That is a good point to make out loud.
• Common mistake: calling heapify on every update, which is O(n), and returning a player twice when they have two entries with equal scores.

02Where this question was reported

• LeetCode Discuss · LeetCode 1244 (Medium), Google-tagged, freq 10.6
• interviewing.io · Interviewer steers toward keeping a binary heap inside the class