The 100 problems, solved / Iterators, streams and counters / 023 · Design a playlist shuffler with a no-repeat window

Design a playlist shuffler with a no-repeat window

next() is uniformly random among tracks not played in the last k calls; add back and repeat.

Concepts this design leans on

01The design, in the six-step framework

1. Requirements and use cases

Google has asked this for years under several names: least recently used random shuffle, no repeat within K plays, Spotify shuffle. The tracks are arbitrary objects, so they cannot be compared or sorted, and every eligible track must be equally likely.

  • A listener calls next and gets a track chosen uniformly among tracks not played in the last k calls.
  • A listener calls back to return to the previous track, and next then replays forward through the history.
  • A listener calls repeat to get the current track again; this is not a new play.
  • A playlist owner adds or removes a track between plays.
  • Variant: loop shuffle, where every track plays once per pass and the order stays constant within a pass.

Out of scope: skip weighting, smart ordering by mood, and audio. Non-functional constraints: next is O(1); memory beyond the playlist is O(k), plus the bounded back history; randomness is injected so tests are deterministic; and k must be less than the number of tracks, or nothing is eligible, which the constructor rejects.

2. Core entities and relationships

  • Track: any hashable object; never compared except for equality.
  • Shuffler: the public object; owns the partitioned array, the cooldown queue and the history.
  • Eligible region: the prefix of the array holding tracks that may play.
  • Cooldown queue: a deque of the last k played tracks, oldest first.
  • History: a bounded list plus a cursor for back and forward.
  • LoopShuffler: the per-pass variant with an incremental Fisher-Yates shuffle.

The shuffler composes everything except tracks, which it aggregates. The key move is the single array split into two regions by an integer: slots below eligible_end are eligible and slots above it are cooling. A dictionary from track to slot gives O(1) removal. This is the swap-with-last trick from the random set, combined with a recency queue.

Public APIStateShuffler[T]tracks: list[T]slot: dict[T, int]eligible_end: intcooldown: deque[T]history: Historyrng: Randomnext() -> Tback() -> Trepeat() -> Tadd(track)remove(track)LoopShuffler[T]order: list[T]pos: intrng: Randomnext() -> THistory[T]played: deque[T]cursor: intpush(t)back() -> Tforward() -> T | Nonecurrent() -> T

3. Interfaces and key methods

import random
from typing import Generic, Hashable, List, Optional, TypeVar

T = TypeVar("T", bound=Hashable)


class Shuffler(Generic[T]):
    def __init__(self, tracks: List[T], k: int, rng: Optional[random.Random] = None) -> None:
        """Requires 0 <= k < len(tracks); tracks must be distinct."""

    def next(self) -> T:
        """Uniform over tracks not among the last k plays; replays history after back()."""

    def back(self) -> T:
        """The previous track; raises IndexError at the start of history."""

    def repeat(self) -> T:
        """The current track again; does not count as a play."""

    def add(self, track: T) -> None: ...
    def remove(self, track: T) -> None: ...

Invariants: every track is in exactly one region; the cooling region holds at most k tracks; slot[t] is the index of t in the array; and the eligible region is never empty, because k is less than the track count.

4. Design patterns and the choices behind them

Two regions of one array

The force is uniform choice among eligible tracks in O(1). Rejection sampling (pick any track and retry if cooling) has expected O(n / (n − k)) tries, which is unbounded as k nears n. Rebuilding an eligible list is O(n). A partitioned array is O(1): pick a random index below eligible_end, swap that track to the boundary, and shrink the region. The track that finishes its cooldown is swapped back into the eligible region. Uniformity holds because each eligible track occupies exactly one slot in the range.

P(track t is chosen) = 1 / eligible_end for every eligible t, where eligible_end = n − (tracks cooling)

History as a cursor, not as plays

The force is that back must not create a new play or disturb the cooldown. A separate history with a cursor keeps those concepts apart: back moves the cursor, and next first replays forward if the cursor is behind the end, then falls through to a random pick. This is the same structure as a browser's history, and a bounded deque caps memory. It keeps the iterator position separate from the collection.

Injected randomness

The force is testing a random class. A seeded generator gives exact sequences, and a statistical test over many draws checks uniformity. See designing for unit tests. Defining precisely what "next" and "repeat" count as is method design, and it should come before code.

5. Core flow and code

Flow for tracks a, b, c, d with k = 1:

  1. The array is a b c d and eligible_end is 4.
  2. next picks index 1 (b). b swaps with the last eligible slot, the region shrinks to 3, and b joins the cooldown. The array is a d c | b.
  3. next again: pick among a, d, c, say c. c moves to the boundary and joins the cooldown, which now exceeds k, so b, the oldest, rejoins the eligible region. c cools.
import random
from collections import deque
from typing import Deque, Dict, Generic, Hashable, List, Optional, TypeVar

T = TypeVar("T", bound=Hashable)


class History(Generic[T]):
    def __init__(self, limit: int = 1000) -> None:
        self.played: Deque[T] = deque(maxlen=limit)
        self.cursor = -1

    def push(self, track: T) -> None:
        self.played.append(track)
        self.cursor = len(self.played) - 1

    def back(self) -> T:
        if self.cursor <= 0:
            raise IndexError("start of history")
        self.cursor -= 1
        return self.played[self.cursor]

    def forward(self) -> Optional[T]:
        if self.cursor + 1 < len(self.played):
            self.cursor += 1
            return self.played[self.cursor]
        return None

    def current(self) -> T:
        if self.cursor < 0:
            raise IndexError("nothing played yet")
        return self.played[self.cursor]


class Shuffler(Generic[T]):
    def __init__(self, tracks: List[T], k: int, rng: Optional[random.Random] = None) -> None:
        if len(set(tracks)) != len(tracks):
            raise ValueError("tracks must be distinct")
        if not 0 <= k < len(tracks):
            raise ValueError("need 0 <= k < number of tracks")
        self.tracks = list(tracks)
        self.slot: Dict[T, int] = {t: i for i, t in enumerate(self.tracks)}
        self.k = k
        self.eligible_end = len(self.tracks)
        self.cooldown: Deque[T] = deque()
        self.history: History[T] = History()
        self.rng = rng or random.Random()

    def _swap(self, i: int, j: int) -> None:
        a, b = self.tracks[i], self.tracks[j]
        self.tracks[i], self.tracks[j] = b, a
        self.slot[a], self.slot[b] = j, i

    def next(self) -> T:
        replay = self.history.forward()
        if replay is not None:
            return replay
        i = self.rng.randrange(self.eligible_end)
        track = self.tracks[i]
        self.eligible_end -= 1
        self._swap(i, self.eligible_end)        # chosen track leaves the eligible region
        self.cooldown.append(track)
        if len(self.cooldown) > self.k:
            released = self.cooldown.popleft()  # oldest cooling track becomes eligible
            self._swap(self.slot[released], self.eligible_end)
            self.eligible_end += 1
        self.history.push(track)
        return track

    def back(self) -> T:
        return self.history.back()

    def repeat(self) -> T:
        return self.history.current()

    def add(self, track: T) -> None:
        if track in self.slot:
            return
        self.tracks.append(track)
        self.slot[track] = len(self.tracks) - 1
        self._swap(self.eligible_end, len(self.tracks) - 1)
        self.eligible_end += 1

    def remove(self, track: T) -> None:
        if len(self.tracks) - 1 <= self.k:
            raise ValueError("too few tracks would remain for this k")
        i = self.slot[track]
        if i < self.eligible_end:
            self.eligible_end -= 1
            self._swap(i, self.eligible_end)
            i = self.eligible_end
        else:
            self.cooldown.remove(track)
        self._swap(i, len(self.tracks) - 1)
        self.tracks.pop()
        del self.slot[track]


class LoopShuffler(Generic[T]):
    """Every track once per pass. The swap happens as we go, so a pass stores no extra list."""

    def __init__(self, tracks: List[T], rng: Optional[random.Random] = None) -> None:
        self.order = list(tracks)
        self.pos = 0
        self.rng = rng or random.Random()

    def next(self) -> T:
        if self.pos == len(self.order):
            self.pos = 0
        j = self.rng.randrange(self.pos, len(self.order))
        self.order[self.pos], self.order[j] = self.order[j], self.order[self.pos]
        self.pos += 1
        return self.order[self.pos - 1]


if __name__ == "__main__":
    s = Shuffler(list("abcde"), k=3, rng=random.Random(9))
    plays = [s.next() for _ in range(5000)]
    for i in range(len(plays)):
        assert plays[i] not in plays[max(0, i - 3):i]      # no repeat within k
    assert 0.17 < plays.count("a") / 5000 < 0.23           # fair over time

    t = Shuffler(list("abcd"), k=1, rng=random.Random(2))
    first, second = t.next(), t.next()
    assert t.repeat() == second and t.back() == first and t.next() == second
    t.add("e")
    t.remove("a")
    assert all(t.next() in "bcde" for _ in range(50))
    assert sorted(t.tracks) == list("bcde")

    loop = LoopShuffler(list("abc"), random.Random(1))
    assert sorted(loop.next() for _ in range(3)) == list("abc")

6. Extensions, concurrency and what interviewers probe

  • What does uniform mean here? Uniform among eligible tracks per call. With k = n − 1 only one track is eligible, so the sequence becomes a fixed cycle, which is correct.
  • Space follow-up. For loop shuffle, the reported optimisation avoids a shuffled copy: LoopShuffler swaps as it goes, and a "go back" uses the positions before pos, so one array suffices.
  • Arbitrary objects. The tracks need hashing for the slot map. If they cannot be hashed, assign integer ids at construction and shuffle ids.
  • Removing a cooling track. It must leave the cooldown queue as well, which is O(k) in the code; a linked structure makes it O(1).
  • Thread safety. A player has one consumer. If two devices share a session, a lock around next is enough, because the swaps and queue change together.
  • Common mistake. Releasing the oldest track before picking, which allows a replay at distance k, or using rejection sampling and being unable to bound its running time.

02Where this question was reported

  • LeetCode Discuss · LeetCode 384 (Medium), Google-tagged, freq 16.6
  • LeetCode Discuss · Tracks are arbitrary objects; returned track cannot recur in the next K plays
  • LeetCode Discuss · Pick next song on each call; all eligible (not in last k) songs equally likely
  • LeetCode Discuss · 'Least recently used random shuffle playlist'
  • Glassdoor · No repeats per loop, constant order per loop, go back and repeat current; follow-up: optimize space (two lists)
  • Glassdoor · 'Design Spotify playlist algorithm including shuffle'
  • freshersnow.com · Shuffle with k-iteration cooldown; follow-up iterator with next()/hasNext()
  • Glassdoor · Create a deck-of-cards class, write shuffle and poker-hand scoring (snippet only; URL mapping from search cluster)