Design an in-memory pub/sub message queue

TL;DR

• You build an append-only log per partition, a cursor per consumer group, and one worker thread per (group, partition) — which is where the ordering guarantee comes from and where it stops.
• Three decisions carry the interview: the offset lives in a store, never in a thread (so replay and rebalancing are trivial), the buffer is bounded (so a stalled consumer becomes backpressure, not an out-of-memory kill), and commit happens after the ack (so failure means redelivery, not loss).
• Patterns that earn their place: Mediator (the broker), Strategy (partitioner, retry), Producer-Consumer over a Condition, Observer/Event Bus at the edges.

Problem statement

"Design an in-memory publish/subscribe message queue. Producers publish to a named topic; the topic is split into partitions so several consumers can work in parallel. Consumers join named groups: every group sees every message, but within a group each message goes to exactly one consumer. Delivery is at-least-once with acknowledgement, failures are retried with backoff and eventually parked in a dead-letter queue, and a group can rewind and replay. It must be correct with many producers and consumers running at once, and shut down without losing anything."

Requirements

Functional

• Topics with a configurable partition count; a key routes a message to a partition.
• Publish; subscribe by (topic, group, consumer); push delivery to each consumer.
• Consumer groups: every group has its own cursor, so two groups read the same records independently.
• Offsets stored per (group, topic, partition), with seek and replay from any offset.
• At-least-once delivery: the offset advances only after the consumer acks.
• Retry with exponential backoff, then a dead-letter queue so one poison record cannot stall a partition.
• Retention by size (a bounded buffer per partition) and by age.
• Concurrent producers and consumers, plus graceful shutdown that drains before stopping.

Non-functional and constraints

• Per-key ordering is guaranteed; global ordering across partitions is explicitly not.
• A publish never blocks forever: it either finds room, sheds, or fails with a clear error.
• No message is lost between an ack and a commit; a crash mid-delivery causes redelivery, never loss.
• In-memory, single process, standard library only. Time and IDs injected; tests are deterministic and use barriers rather than sleeps.

Out of scope: durability to disk, cross-process transport, exactly-once semantics, and consumer-driven pull APIs. The distributed version of all of this is Design a distributed message queue.

Clarifying questions and assumptions

Question to ask	Assumption taken here
Push or pull delivery?	Push: the broker owns a worker thread per (group, partition). A pull API is the same log with the loop moved into the consumer.
Does every subscriber get every message?	Every group does. Inside a group, exactly one consumer gets each record — that is the whole point of groups.
What ordering do we promise?	Per key, because a key always maps to one partition and one worker. Never across partitions; say this before you are asked.
At-least-once, at-most-once, or effectively-once?	At-least-once. The offset advances after the ack, so a failure mid-processing means redelivery. Effectively-once = at-least-once plus an idempotent consumer.
Is an ack explicit or implicit?	Implicit: returning from on_message acks, raising nacks. A real system passes an ack callable so a consumer can finish asynchronously.
What happens when a consumer is slower than the producer?	The partition buffer fills, then either the producer blocks (BLOCK) or the oldest records are shed (DROP_OLDEST). Both are counted; neither is silent.
Can records be replayed after they are acked?	Yes, until pressure reclaims them. Records are freed when the buffer is full, not the moment they are acked.

Core entities and relationships

• Message — what a producer hands over: id, topic, payload, optional key and headers, and a creation time from the injected Clock.
• Record — a message once it has a place in the log: (topic, partition, offset) plus the message. Immutable and shared by every group.
• Partition — the append-only buffer, the offset counter, and the single Condition that makes producers and workers meet. 1 → * records.
• Topic — a name, a list of partitions and a Partitioner. 1 → * partitions.
• Broker — the mediator: topic registry, group registry, OffsetStore, DeadLetterQueue, publish, subscribe, replay, drain and shutdown.
• ConsumerGroup — a set of consumers sharing one cursor per partition, plus the rebalance that deals partitions out again when a consumer joins. 1 → * DeliveryWorker.
• DeliveryWorker — one thread bound to one (group, partition): fetch, deliver, retry, dead-letter, commit.
• OffsetStore — (group, topic, partition) → next offset, with seek for replay and a compare-and-set advance for commits.
• RetryPolicy and RetentionPolicy — the two frozen policy objects; DeadLetterQueue — where exhausted records park.

Class diagram

Structure: the log and how a message finds its place in it.

classDiagram
    class Broker {
        +offsets: OffsetStore
        +dlq: DeadLetterQueue
        +state: BrokerState
        -_lock: RLock
        -_progress: Condition
        +create_topic(name, partitions, retention) Topic
        +publish(topic, payload, key) Record
        +subscribe(topic, group, consumer) ConsumerGroup
        +replay(group, topic, from_offset)
        +drain(timeout) bool
        +close(timeout)
    }
    class Topic {
        +name: str
        +partition_count: int
        +route(key) Partition
        +partitions() List~Partition~
    }
    class Partition {
        +topic: str
        +index: int
        +dropped: int
        -_condition: Condition
        -_base_offset: int
        -_next_offset: int
        -_low_water: int
        +append(message) Record
        +fetch(offset, timeout, stop) Record
        +set_low_water(offset)
    }
    class Record {
        +topic: str
        +partition: int
        +offset: int
        +message: Message
    }
    class Message {
        +id: str
        +topic: str
        +payload: str
        +key: str
        +created: float
    }
    class RetentionPolicy {
        +max_messages: int
        +max_age_seconds: float
        +on_full: FullPolicy
        +block_timeout: float
    }
    class OffsetStore {
        -_lock: Lock
        +committed(group, topic, partition) int
        +seek(group, topic, partition, offset)
        +advance(group, topic, partition, expected, new) bool
        +low_water(topic, partition, groups) int
    }
    class Partitioner {
        <<interface>>
        +partition_for(key, partition_count) int
    }
    class KeyHashPartitioner
    class RoundRobinPartitioner
    class StickyPartitioner
    Broker "1" *-- "many" Topic
    Topic "1" *-- "many" Partition
    Partition "1" o-- "many" Record
    Record --> "1" Message
    Topic --> "1" Partitioner
    Partition --> "1" RetentionPolicy
    Broker --> "1" OffsetStore
    Partitioner <|.. KeyHashPartitioner
    Partitioner <|.. RoundRobinPartitioner
    Partitioner <|.. StickyPartitioner

Behaviour: how a record reaches a consumer, and what happens when it does not.

classDiagram
    class Broker
    class ConsumerGroup {
        +name: str
        -_lock: Lock
        +add_consumer(consumer)
        +assignment() Dict~str,List~
        +pause()
        +resume()
        +stats() Dict~str,int~
    }
    class DeliveryWorker {
        +group: str
        +state: DeliveryState
        +delivered: int
        +retried: int
        +dead_lettered: int
        +skipped: int
        +start() DeliveryWorker
        +stop()
    }
    class Consumer {
        <<interface>>
        +name: str
        +on_message(record: Record)
    }
    class RecordingConsumer {
        +wait_for(count, timeout) bool
    }
    class FlakyConsumer {
        +fail_times: int
        +poison: str
    }
    class RetryPolicy {
        +max_attempts: int
        +base_delay: float
        +multiplier: float
        +should_retry(attempt) bool
        +delay_for(attempt) float
    }
    class DeadLetterQueue {
        -_lock: Lock
        +add(letter)
        +letters(group) List~DeadLetter~
    }
    class DeadLetter {
        +record: Record
        +group: str
        +attempts: int
        +error: str
    }
    class DeliveryState {
        <<enumeration>>
        PENDING
        IN_FLIGHT
        ACKED
        RETRY_SCHEDULED
        DEAD_LETTERED
    }
    class OffsetStore
    class Partition
    Broker "1" *-- "many" ConsumerGroup
    ConsumerGroup "1" *-- "many" DeliveryWorker
    DeliveryWorker --> "1" Consumer
    DeliveryWorker --> "1" Partition
    DeliveryWorker --> "1" RetryPolicy
    DeliveryWorker --> "1" OffsetStore
    DeliveryWorker --> DeadLetterQueue : parks poison records
    DeadLetterQueue "1" o-- "many" DeadLetter
    DeliveryWorker --> DeliveryState
    Consumer <|.. RecordingConsumer
    RecordingConsumer <|-- FlakyConsumer

Design patterns applied

Pattern	Where	Why it earns its place
Mediator	Broker	Producers and consumers never hold a reference to each other. Adding a group, replaying, or shutting down is one object's job, and that object is the only place that knows the wiring.
Producer-Consumer	Partition._condition	The textbook rendezvous: append notifies, fetch waits on a predicate. wait_for with a timeout and a stop event means a worker never spins and never misses a wake-up.
Strategy	Partitioner, RetryPolicy	Routing and backoff are the two rules an interviewer will ask you to change ("now make it sticky", "now cap the backoff"). Each is one object with no effect on the broker.
Iterator (cursor)	OffsetStore	An offset is an external iterator over a partition. Because it lives in a store and not in a thread, replay is an assignment and rebalancing loses nothing.
Event Bus / Observer	subscribe + push delivery	Consumers register interest by name and are called back; the broker fans out. The difference from a plain in-process Event Bus is durability of position: a bus forgets, a log remembers.
Template Method	DeliveryWorker._deliver	The retry/dead-letter skeleton is fixed; the varying step is the consumer's on_message.
Null Object (light)	An empty group list	A topic with no subscribers keeps its records so a late group can replay them, rather than special-casing "nobody is listening".
Singleton	Broker.instance()	Offered, not imposed. A process usually wants one broker; the constructor stays public so each test owns its own broker and its own threads.

Deliberately not used: a State class per DeliveryState. Five values with guarded transitions are an enum and a for loop over attempts; five classes would hide a nine-line method. Also not used: a thread per consumer. Threads are bound to partitions, not consumers, because the partition is the unit of ordering — bind them to consumers and you either lose ordering or need a lock per key.

Key flows

Publish, wake, deliver, ack, commit — and the retry branch that keeps a poison record from stalling everyone.

sequenceDiagram
    autonumber
    actor P as Producer
    participant B as Broker
    participant T as Topic
    participant PART as Partition
    participant W as DeliveryWorker
    participant C as Consumer
    participant DLQ as DeadLetterQueue
    P->>B: publish(orders, alice:paid, key=alice)
    B->>T: route(key)
    T-->>B: partition 1 (crc32 of the key)
    B->>PART: append(message)
    PART->>PART: lock, assign offset, notify_all
    PART-->>P: Record(orders, 1, offset)
    W->>PART: fetch(offset) - wakes from the Condition
    PART-->>W: Record
    W->>C: on_message(record)
    alt returns normally
        C-->>W: ack
        W->>B: advance(expected=offset, new=offset+1)
    else raises
        C-->>W: nack
        loop while attempts remain
            W->>W: wait delay_for(attempt)
            W->>C: on_message(record)
        end
        W->>DLQ: add(DeadLetter)
        W->>B: commit past it so the partition keeps moving
    end

One delivery attempt, as a state machine. RETRY_SCHEDULED is the only state that costs wall-clock time, and it is the one you must be able to bound in the room.

stateDiagram-v2
    [*] --> PENDING : appended to the partition
    PENDING --> IN_FLIGHT : worker fetches and calls on_message
    IN_FLIGHT --> ACKED : returned normally
    IN_FLIGHT --> RETRY_SCHEDULED : raised, attempts remain
    RETRY_SCHEDULED --> IN_FLIGHT : backoff elapsed
    IN_FLIGHT --> DEAD_LETTERED : attempts exhausted
    ACKED --> [*]
    DEAD_LETTERED --> [*]

Implementation

Write the log first, then the cursor, then the thread that joins them. The enums pin the vocabulary; DeliveryState is the one that becomes the state diagram and FullPolicy the one that becomes the concurrency section.

code/lld/pub_sub_system/models.py — enums

class DeliveryState(StrEnum):
    """The lifecycle of one attempt to hand a record to one consumer group."""

    PENDING = "pending"  # appended to the partition, nobody has read it yet
    IN_FLIGHT = "in_flight"  # a worker is inside consumer.on_message
    ACKED = "acked"  # returned normally; the offset moves past it
    RETRY_SCHEDULED = "retry_scheduled"  # raised, attempts left, waiting out the backoff
    DEAD_LETTERED = "dead_lettered"  # attempts exhausted; parked in the DLQ, offset moves on


class BrokerState(StrEnum):
    RUNNING = "running"
    DRAINING = "draining"  # close() called: no new publishes, workers finishing
    STOPPED = "stopped"


class FullPolicy(StrEnum):
    """What a partition does when its bounded buffer is full."""

    BLOCK = "block"  # backpressure: the producer waits for the slowest group
    DROP_OLDEST = "drop_oldest"  # shed: the newest data matters more than the oldest

code/lld/pub_sub_system/models.py — errors

class TopicNotFoundError(NotFoundError):
    """Publish or subscribe to a topic that was never created."""


class TopicExistsError(ConflictError):
    """create_topic called twice with the same name."""


class BrokerClosedError(InvalidStateError):
    """The broker is draining or stopped; it no longer accepts work."""


class BackpressureError(ConflictError):
    """A bounded partition stayed full for the whole publish timeout."""


class OffsetOutOfRangeError(ValidationError):
    """The requested offset was trimmed by retention, or is beyond the log end."""


class SubscriptionError(ValidationError):
    """An impossible subscription (unknown topic, duplicate consumer name, empty group)."""

A Message is what the producer wrote; a Record is where it ended up. Keeping them separate is what lets the same message be re-read from an offset forever without the producer's object being mutable.

code/lld/pub_sub_system/models.py — messages and records

@dataclass(frozen=True, slots=True)
class Message:
    """What a producer hands to the broker. The key decides the partition."""

    id: str
    topic: str
    payload: str
    key: str | None = None
    headers: dict[str, str] = field(default_factory=dict)
    created: float = 0.0  # epoch seconds from the injected Clock


@dataclass(frozen=True, slots=True)
class Record:
    """A message once it has a place in the log. Immutable, shared by every group."""

    topic: str
    partition: int
    offset: int
    message: Message

    @property
    def key(self) -> str | None:
        return self.message.key

    @property
    def payload(self) -> str:
        return self.message.payload

    def __str__(self) -> str:
        return f"{self.topic}/{self.partition}@{self.offset} {self.payload}"


@dataclass(frozen=True, slots=True)
class DeadLetter:
    """A record that exhausted its retries, plus why."""

    record: Record
    group: str
    attempts: int
    error: str
    failed_at: float

Both policies are frozen dataclasses with the arithmetic on them, not scattered through the worker. RetentionPolicy carries two independent bounds — size, which creates backpressure, and age, which does not care whether anyone read the record.

code/lld/pub_sub_system/models.py — policies

@dataclass(frozen=True, slots=True)
class RetryPolicy:
    """Exponential backoff with a ceiling. Deterministic: no jitter in the handbook build."""

    max_attempts: int = 3
    base_delay: float = 0.001
    multiplier: float = 2.0
    max_delay: float = 0.05

    def __post_init__(self) -> None:
        if self.max_attempts < 1 or self.base_delay < 0:
            raise ValidationError("max_attempts must be >= 1 and base_delay >= 0")

    def should_retry(self, attempt: int) -> bool:
        return attempt < self.max_attempts

    def delay_for(self, attempt: int) -> float:
        return min(self.max_delay, self.base_delay * (self.multiplier ** (attempt - 1)))


@dataclass(frozen=True, slots=True)
class RetentionPolicy:
    """Two independent bounds: how many records a partition holds, and for how long.

    ``max_messages`` is the bounded buffer that creates backpressure; space is
    reclaimed when the slowest subscribed group commits past the oldest record.
    ``max_age_seconds`` trims regardless of consumers -- the reason a slow group
    can be fast-forwarded and lose records.
    """

    max_messages: int = 1024
    max_age_seconds: float | None = None
    on_full: FullPolicy = FullPolicy.BLOCK
    block_timeout: float = 1.0

    def __post_init__(self) -> None:
        if self.max_messages < 1:
            raise ValidationError("max_messages must be at least 1")

The partitioner is the only Strategy a producer chooses, and zlib.crc32 instead of hash() is a detail worth saying out loud: CPython salts string hashes per process, so hash() would route the same key differently after a restart.

code/lld/pub_sub_system/strategies.py

class KeyHashPartitioner:
    """Same key, same partition, forever -- which is what makes per-key ordering work.

    ``zlib.crc32`` and not ``hash()``: CPython salts string hashes per process,
    so ``hash()`` would send the same key to different partitions after a restart.
    """

    def __init__(self, fallback: RoundRobinPartitioner | None = None) -> None:
        self._fallback = fallback or RoundRobinPartitioner()

    def partition_for(self, key: str | None, partition_count: int) -> int:
        if key is None:
            return self._fallback.partition_for(None, partition_count)
        return zlib.crc32(key.encode("utf-8")) % partition_count


class RoundRobinPartitioner:
    """Even spread for keyless messages. The counter needs a lock: many producers."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._next = 0

    def partition_for(self, key: str | None, partition_count: int) -> int:
        with self._lock:
            chosen = self._next % partition_count
            self._next += 1
            return chosen


class StickyPartitioner:
    """Fills one partition until a batch is full, then moves on -- fewer, larger batches."""

    def __init__(self, batch_size: int = 16) -> None:
        self.batch_size = batch_size
        self._lock = threading.Lock()
        self._current = 0
        self._count = 0

    def partition_for(self, key: str | None, partition_count: int) -> int:
        if key is not None:
            return zlib.crc32(key.encode("utf-8")) % partition_count
        with self._lock:
            if self._count >= self.batch_size:
                self._current = (self._current + 1) % partition_count
                self._count = 0
            self._count += 1
            return self._current % partition_count

Now the centre of the design. Partition owns the only Condition in the package: producers append and notify, workers park on fetch, and a commit that reclaims space notifies the producers back.

code/lld/pub_sub_system/storage.py — the partition

class Partition:
    """An append-only log segment with a bounded buffer.

    Invariants, all protected by ``_condition``:

    * ``_next_offset`` never goes backwards, so an offset identifies a record forever;
    * ``_records[0].offset == _base_offset``, so index arithmetic replaces a scan;
    * a record is only discarded when every subscribed group has committed past
      it, or when ``max_age_seconds`` says it is stale.
    """

    def __init__(self, topic: str, index: int, retention: RetentionPolicy, clock: Clock | None = None) -> None:
        self.topic = topic
        self.index = index
        self.retention = retention
        self._clock = clock or SystemClock()
        self._condition = threading.Condition()
        self._records: deque[Record] = deque()
        self._base_offset = 0
        self._next_offset = 0
        self._low_water = 0  # the oldest offset any subscribed group still needs
        self._closed = False
        self.dropped = 0  # records lost to DROP_OLDEST or to age-based retention

    def append(self, message: Message) -> Record:
        """Producer side. Blocks (or sheds) when the buffer is full, then notifies workers."""
        with self._condition:
            if self._closed:
                raise BackpressureError(f"{self.topic}/{self.index} is closed")
            self._trim_expired_locked()
            if len(self._records) >= self.retention.max_messages:
                self._make_room_locked()
            record = Record(self.topic, self.index, self._next_offset, message)
            self._records.append(record)
            self._next_offset += 1
            self._condition.notify_all()  # wake every worker parked on this partition
            return record

    def _make_room_locked(self) -> None:
        """Reclaim what every group has acked; if that is not enough, block or shed."""
        if self._has_room_locked():
            return
        if self.retention.on_full is FullPolicy.DROP_OLDEST:
            self._records.popleft()
            self._base_offset += 1
            self.dropped += 1
            return
        room = self._condition.wait_for(
            lambda: self._closed or self._has_room_locked(),
            timeout=self.retention.block_timeout,
        )
        if self._closed:
            # The wait also ends on close. Waking for that reason is not room:
            # without this check a parked producer appends past max_messages.
            raise BackpressureError(f"{self.topic}/{self.index} was closed while the producer waited")
        if not room:
            raise BackpressureError(
                f"{self.topic}/{self.index} full ({self.retention.max_messages}) "
                f"after {self.retention.block_timeout}s: the slowest group is not committing"
            )

    def _has_room_locked(self) -> bool:
        self._reclaim_locked()
        return len(self._records) < self.retention.max_messages

    def _reclaim_locked(self) -> None:
        """Drop only records that every subscribed group has committed past.

        Records are *not* freed the moment they are acked: keeping them until
        the buffer is under pressure is what makes replay possible at all.
        """
        while self._records and self._records[0].offset < self._low_water:
            self._records.popleft()
            self._base_offset += 1

    def fetch(self, offset: int, timeout: float, stop: threading.Event | None = None) -> Record | None:
        """Consumer side. Waits for ``offset`` to exist; ``None`` means timeout or shutdown."""
        with self._condition:
            ready = self._condition.wait_for(
                lambda: self._closed or offset < self._next_offset or (stop is not None and stop.is_set()),
                timeout=timeout,
            )
            if not ready or (stop is not None and stop.is_set()) or offset >= self._next_offset:
                return None
            if offset < self._base_offset:
                raise OffsetOutOfRangeError(
                    f"offset {offset} was trimmed from {self.topic}/{self.index}; "
                    f"earliest is {self._base_offset}"
                )
            return self._records[offset - self._base_offset]

    def set_low_water(self, offset: int) -> None:
        """Record how far the slowest group has got, and wake any blocked producer."""
        with self._condition:
            self._low_water = max(self._low_water, min(offset, self._next_offset))
            self._condition.notify_all()

    def _trim_expired_locked(self) -> None:
        if self.retention.max_age_seconds is None:
            return
        cutoff = self._clock.now() - self.retention.max_age_seconds
        while self._records and self._records[0].message.created < cutoff:
            self._records.popleft()
            self._base_offset += 1
            self.dropped += 1

    def wake(self) -> None:
        """Used by shutdown and rebalancing to unpark workers immediately."""
        with self._condition:
            self._condition.notify_all()

    def close(self) -> None:
        with self._condition:
            self._closed = True
            self._condition.notify_all()

    def earliest_offset(self) -> int:
        with self._condition:
            return self._base_offset

    def next_offset(self) -> int:
        with self._condition:
            return self._next_offset

    def size(self) -> int:
        with self._condition:
            return len(self._records)

    def low_water(self) -> int:
        with self._condition:
            return self._low_water

The offset store is deliberately dull, and that is the point — progress is data, not thread state, so replay is an assignment and a rebalance is a restart.

code/lld/pub_sub_system/storage.py — offsets

class OffsetStore:
    """Committed offsets keyed by ``(group, topic, partition)``.

    The committed value is the offset of the *next* record to read, so an empty
    entry and "start from the beginning" are the same thing: zero.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._offsets: dict[tuple[str, str, int], int] = {}

    def committed(self, group: str, topic: str, partition: int) -> int:
        with self._lock:
            return self._offsets.get((group, topic, partition), 0)

    def seek(self, group: str, topic: str, partition: int, offset: int) -> None:
        """Unconditional move, used for replay while the group is paused."""
        with self._lock:
            self._offsets[(group, topic, partition)] = max(0, offset)

    def advance(self, group: str, topic: str, partition: int, expected: int, new: int) -> bool:
        """Compare-and-set: a concurrent seek must not be clobbered by a late ack."""
        key = (group, topic, partition)
        with self._lock:
            if self._offsets.get(key, 0) != expected:
                return False
            self._offsets[key] = new
            return True

    def low_water(self, topic: str, partition: int, groups: list[str]) -> int:
        """The oldest offset any group still needs. Below it, records can be dropped."""
        if not groups:
            return 0  # no subscribers yet: keep everything so a late group can replay
        with self._lock:
            return min(self._offsets.get((group, topic, partition), 0) for group in groups)

    def snapshot(self) -> dict[tuple[str, str, int], int]:
        with self._lock:
            return dict(self._offsets)

The worker is the at-least-once contract in code. Read the offset fresh every turn (a replay may have moved it), deliver, retry with backoff, dead-letter when the attempts run out, and only then commit — with a compare-and-set so a concurrent seek is never clobbered.

code/lld/pub_sub_system/services.py — the delivery worker

class DeliveryWorker:
    """One thread per ``(group, partition)``. That pairing *is* the ordering guarantee.

    Because exactly one worker owns a partition inside a group, records on that
    partition are delivered in offset order and no two consumers in the group
    ever see the same record. Ordering across partitions is not promised -- say
    that out loud before the interviewer asks.
    """

    def __init__(
        self,
        group: str,
        partition: Partition,
        consumer: Consumer,
        offsets: OffsetStore,
        dlq: DeadLetterQueue,
        retry: RetryPolicy,
        clock: Clock,
        on_progress: Broker | None = None,
    ) -> None:
        self.group = group
        self.partition = partition
        self.consumer = consumer
        self.state = DeliveryState.PENDING
        self.delivered = 0
        self.retried = 0
        self.dead_lettered = 0
        self.skipped = 0  # records lost to retention before this group reached them
        self._offsets = offsets
        self._dlq = dlq
        self._retry = retry
        self._clock = clock
        self._broker = on_progress
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> DeliveryWorker:
        self._thread = threading.Thread(
            target=self._run, name=f"{self.group}-{self.partition.topic}-{self.partition.index}", daemon=True
        )
        self._thread.start()
        return self

    def stop(self) -> None:
        self._stop.set()
        self.partition.wake()  # unpark immediately instead of waiting out FETCH_TIMEOUT
        if self._thread is not None:
            self._thread.join()
            self._thread = None

    def _run(self) -> None:
        topic, index = self.partition.topic, self.partition.index
        while not self._stop.is_set():
            # Re-read the offset every turn: a replay may have moved it under us.
            offset = self._offsets.committed(self.group, topic, index)
            try:
                record = self.partition.fetch(offset, FETCH_TIMEOUT, self._stop)
            except OffsetOutOfRangeError:
                earliest = self.partition.earliest_offset()
                self.skipped += earliest - offset
                self._offsets.seek(self.group, topic, index, earliest)
                continue
            if record is None:
                continue
            acked = self._deliver(record)
            if self._stop.is_set() and not acked:
                return  # at-least-once: no commit, so the record is redelivered on restart
            self._offsets.advance(self.group, topic, index, expected=offset, new=offset + 1)
            if self._broker is not None:
                self._broker.report_progress(self.group, topic, index, offset + 1)

    def _deliver(self, record: Record) -> bool:
        """Retry with exponential backoff, then dead-letter so the partition keeps moving."""
        for attempt in range(1, self._retry.max_attempts + 1):
            self.state = DeliveryState.IN_FLIGHT
            try:
                self.consumer.on_message(record)
            except Exception as exc:  # a consumer bug must not kill the worker
                if not self._retry.should_retry(attempt):
                    self.state = DeliveryState.DEAD_LETTERED
                    self.dead_lettered += 1
                    self._dlq.add(
                        DeadLetter(
                            record=record,
                            group=self.group,
                            attempts=attempt,
                            error=f"{type(exc).__name__}: {exc}",
                            failed_at=self._clock.now(),
                        )
                    )
                    return False
                self.state = DeliveryState.RETRY_SCHEDULED
                self.retried += 1
                if self._stop.wait(self._retry.delay_for(attempt)):
                    return False  # shutting down mid-backoff: leave the offset uncommitted
            else:
                self.state = DeliveryState.ACKED
                self.delivered += 1
                return True
        return False

A group is a rebalance policy plus a bag of workers. Because the cursor lives in the store, stopping every worker and dealing the partitions out again is a safe, boring operation.

code/lld/pub_sub_system/services.py — the consumer group

class ConsumerGroup:
    """A set of consumers that share one cursor per partition.

    Adding a consumer rebalances: the workers stop, partitions are dealt out
    again, and new workers start from the *stored* offsets. Nothing is lost,
    because progress lives in the OffsetStore and never in a thread.
    """

    def __init__(
        self,
        name: str,
        topic: Topic,
        offsets: OffsetStore,
        dlq: DeadLetterQueue,
        retry: RetryPolicy,
        clock: Clock,
        broker: Broker,
    ) -> None:
        self.name = name
        self.topic = topic
        self._offsets = offsets
        self._dlq = dlq
        self._retry = retry
        self._clock = clock
        self._broker = broker
        self._lock = threading.Lock()  # serialises rebalances
        self._consumers: list[Consumer] = []
        self._workers: list[DeliveryWorker] = []
        self._totals = {"delivered": 0, "retried": 0, "dead_lettered": 0, "skipped": 0}

    def add_consumer(self, consumer: Consumer) -> None:
        with self._lock:
            if any(c.name == consumer.name for c in self._consumers):
                raise SubscriptionError(f"consumer {consumer.name!r} is already in group {self.name!r}")
            self._consumers.append(consumer)
            self._rebalance_locked()

    def assignment(self) -> dict[str, list[int]]:
        with self._lock:
            plan: dict[str, list[int]] = {c.name: [] for c in self._consumers}
            for worker in self._workers:
                plan[worker.consumer.name].append(worker.partition.index)
            return plan

    def pause(self) -> None:
        with self._lock:
            self._stop_workers_locked()

    def resume(self) -> None:
        with self._lock:
            self._rebalance_locked()

    def close(self) -> None:
        with self._lock:
            self._stop_workers_locked()

    def stats(self) -> dict[str, int]:
        """Counters survive rebalances: retired workers fold their totals into the group."""
        with self._lock:
            return {name: total + self._live_locked(name) for name, total in self._totals.items()}

    def _live_locked(self, name: str) -> int:
        return sum(getattr(worker, name) for worker in self._workers)

    def _stop_workers_locked(self) -> None:
        for worker in self._workers:
            worker.stop()
            for name in self._totals:
                self._totals[name] += getattr(worker, name)
        self._workers = []

    def _rebalance_locked(self) -> None:
        self._stop_workers_locked()
        if not self._consumers:
            return
        for partition in self.topic.partitions():
            consumer = self._consumers[partition.index % len(self._consumers)]
            worker = DeliveryWorker(
                self.name, partition, consumer, self._offsets, self._dlq, self._retry, self._clock, self._broker
            )
            self._workers.append(worker.start())

The broker mediates and owns the two things nobody else can: the registries, and drain, which is a real barrier built on a Condition that every commit signals.

code/lld/pub_sub_system/services.py — the broker

class Broker:
    """Mediator: producers and consumers know the broker, never each other.

    One instance is enough for a process, and ``instance()`` provides it, but the
    constructor stays public so a test owns its own broker and its own threads.
    """

    _instance_lock: ClassVar[threading.Lock] = threading.Lock()
    _instance: ClassVar[Broker | None] = None

    def __init__(
        self,
        clock: Clock | None = None,
        ids: IdGenerator | None = None,
        retry: RetryPolicy | None = None,
        partitioner: Partitioner | None = None,
    ) -> None:
        self._clock = clock or SystemClock()
        self._ids = ids or SequentialIdGenerator("m")
        self._retry = retry or RetryPolicy()
        self._partitioner = partitioner
        self._lock = threading.RLock()  # guards the topic and group registries
        self._progress = threading.Condition()  # signalled after every committed offset
        self._topics: dict[str, Topic] = {}
        self._groups: dict[tuple[str, str], ConsumerGroup] = {}
        self.offsets = OffsetStore()
        self.dlq = DeadLetterQueue()
        self.state = BrokerState.RUNNING

    @classmethod
    def instance(cls) -> Broker:
        if cls._instance is None:
            with cls._instance_lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    @classmethod
    def reset_instance(cls) -> None:
        with cls._instance_lock:
            cls._instance = None

    def create_topic(self, name: str, partitions: int = 1, retention: RetentionPolicy | None = None) -> Topic:
        with self._lock:
            self._require_running()
            if name in self._topics:
                raise TopicExistsError(f"topic {name!r} already exists")
            topic = Topic(name, partitions, retention, self._clock, self._partitioner)
            self._topics[name] = topic
            return topic

    def topic(self, name: str) -> Topic:
        with self._lock:
            try:
                return self._topics[name]
            except KeyError:
                raise TopicNotFoundError(f"unknown topic {name!r}") from None

    def publish(self, topic: str, payload: str, key: str | None = None, headers: dict[str, str] | None = None) -> Record:
        """Route by key, append under the partition lock, and wake the parked workers."""
        target = self.topic(topic)
        self._require_running()
        message = Message(
            id=self._ids.next_id(),
            topic=topic,
            payload=payload,
            key=key,
            headers=dict(headers or {}),
            created=self._clock.now(),
        )
        return target.route(key).append(message)  # may block: that is backpressure

    def subscribe(self, topic: str, group: str, consumer: Consumer) -> ConsumerGroup:
        target = self.topic(topic)
        if not group:
            raise SubscriptionError("group name must be non-empty")
        with self._lock:
            self._require_running()
            key = (group, topic)
            if key not in self._groups:
                self._groups[key] = ConsumerGroup(
                    group, target, self.offsets, self.dlq, self._retry, self._clock, self
                )
            consumer_group = self._groups[key]
        consumer_group.add_consumer(consumer)  # outside the registry lock: it starts threads
        return consumer_group

    def group_names(self, topic: str) -> list[str]:
        with self._lock:
            return [group for (group, subscribed) in self._groups if subscribed == topic]

    def group(self, name: str, topic: str) -> ConsumerGroup:
        with self._lock:
            try:
                return self._groups[(name, topic)]
            except KeyError:
                raise SubscriptionError(f"group {name!r} is not subscribed to {topic!r}") from None

    def report_progress(self, group: str, topic: str, partition: int, offset: int) -> None:
        """A worker committed. Reclaim space in the partition and wake anyone draining."""
        low = self.offsets.low_water(topic, partition, self.group_names(topic))
        self.topic(topic).partition(partition).set_low_water(low)
        with self._progress:
            self._progress.notify_all()

    def lag(self, group: str, topic: str) -> int:
        target = self.topic(topic)
        return sum(
            p.next_offset() - self.offsets.committed(group, topic, p.index) for p in target.partitions()
        )

    def drain(self, timeout: float = 2.0) -> bool:
        """Barrier for tests and shutdown: return once every group has caught up."""
        with self._progress:
            return self._progress.wait_for(self._all_caught_up, timeout=timeout)

    def _all_caught_up(self) -> bool:
        with self._lock:
            pairs = list(self._groups)
        return all(self.lag(group, topic) == 0 for group, topic in pairs)

    def replay(self, group: str, topic: str, from_offset: int = 0) -> None:
        """Pause the group, rewind every partition, resume. Offsets are just numbers."""
        with self._lock:
            consumer_group = self._groups.get((group, topic))
        if consumer_group is None:
            raise SubscriptionError(f"group {group!r} is not subscribed to {topic!r}")
        consumer_group.pause()
        target = self.topic(topic)
        for partition in target.partitions():
            self.offsets.seek(group, topic, partition.index, max(from_offset, partition.earliest_offset()))
        consumer_group.resume()

    def close(self, timeout: float = 2.0) -> None:
        """Graceful shutdown: stop accepting work, let the workers finish, then stop them."""
        with self._lock:
            if self.state is not BrokerState.RUNNING:
                return
            self.state = BrokerState.DRAINING
            groups, topics = list(self._groups.values()), list(self._topics.values())
        self.drain(timeout)
        for consumer_group in groups:
            consumer_group.close()
        for topic in topics:
            topic.close()
        with self._lock:
            self.state = BrokerState.STOPPED

    def _require_running(self) -> None:
        if self.state is not BrokerState.RUNNING:
            raise BrokerClosedError(f"broker is {self.state}")

Running python -m lld.pub_sub_system.demo publishes six order events into two partitions, feeds three groups, retries a transient failure, dead-letters a poison record and replays one group:

published alice:created      -> partition 1 offset 0
published bob:created        -> partition 0 offset 0
published alice:paid         -> partition 1 offset 1
published bob:cancelled      -> partition 0 offset 1
published alice:shipped      -> partition 1 offset 2
published carol:created      -> partition 1 offset 3
billing received 6 of 6 (its own cursor)
audit   received 6 of 6 (an independent cursor)
alice keeps her order (one partition, one worker): ['alice:created', 'alice:paid', 'alice:shipped']
shipping stats: {'delivered': 5, 'retried': 7, 'dead_lettered': 1, 'skipped': 0}
shipping retried alice:paid 2 times before it acked
dead letter after 3 attempts: bob:cancelled (ConsumerError: cannot process bob:cancelled)
lag before replay: billing=0
billing after replay from offset 0: 12 records
broker state=stopped, dead letters=1

Concurrency and edge cases

Which lock protects what. Four, and they never nest in a cycle:

1. Partition._condition guards the deque, _base_offset, _next_offset and _low_water. It is both the mutex and the rendezvous: append assigns the offset and calls notify_all; fetch calls wait_for(offset < next_offset or stopping). An uncontended lock costs about 17 ns, so partitioning is what buys throughput — four partitions are four independent locks, not one queue with four readers.
2. OffsetStore._lock guards the cursor map. Commits go through advance(expected, new), a compare-and-set, so a replay that moved the cursor while a record was in flight is not clobbered by the late ack.
3. ConsumerGroup._lock serialises rebalances, so two threads joining at once cannot both deal partitions out.
4. Broker._lock (an RLock) guards the topic and group registries, and Broker._progress is a separate Condition signalled after every commit — that is what makes drain() a barrier instead of a poll loop.

Ordering. One worker per (group, partition) is the entire guarantee. Same key → same partition → same worker → offset order. Across partitions there is no order at all, and a candidate who promises one has not understood the model.

Backpressure versus retention. A record is not freed when it is acked; it is freed when the buffer is under pressure and every group has committed past it. Sizing, using the chat fan-out numbers: 2B messages/day is 2B / 86,400 s ≈ 23k msg/s; spread over four partitions that is roughly 5.8k msg/s each, so a 1,024-record buffer absorbs about 1,024 / 5,800 ≈ 0.18 s of consumer hiccup. That is the honest claim — a bounded buffer buys you a couple of hundred milliseconds, not minutes. Past that, BLOCK slows the producer down and DROP_OLDEST sheds; both increment dropped.

Retention overtaking a slow consumer. Age-based trimming ignores consumers, so a group can find its offset below earliest_offset(). fetch raises OffsetOutOfRangeError, the worker seeks forward, and the gap is counted in skipped — visible data loss beats silent data loss.

Subscribe during publish. Joining a group takes ConsumerGroup._lock outside the broker's registry lock, stops the workers, re-deals partitions and restarts from the stored offsets. Records published during the rebalance sit in the partition and are picked up on the next fetch. A late group starts at zero and reads the whole backlog.

Graceful shutdown. close() flips the state to DRAINING (publishes now raise BrokerClosedError), waits on drain, then stops the workers and closes the partitions. A worker interrupted mid-backoff returns without committing, which is the at-least-once contract: on restart, that record is delivered again.

Common mistake

Committing the offset before calling the consumer, because it makes the loop simpler. That turns at-least-once into at-most-once, and the failure mode is a silently dropped payment. The other half of the same mistake is an unbounded queue: it looks like it removes the backpressure problem, but it only moves it to the memory allocator, where it fails much later and much worse.

Extensibility and follow-ups

• Persistence: Partition is already an append-only structure with a base offset. Add a write-ahead log behind an interface, flush on append, and rebuild the deque on start; the offset store becomes a second small file. Nothing above the partition changes.
• Effectively-once: at-least-once plus an idempotent consumer. Give the consumer the record id as an idempotency key and have it deduplicate; the broker keeps its simple contract.
• Delayed and priority topics: a delayed topic sorts by "visible at" and fetch waits until the clock passes it — the Condition already supports a timed wait. A priority topic is several partitions with a weighted worker, and you should say out loud that priority and per-key ordering are in tension.
• Consumer lag as an SLI: Broker.lag(group, topic) is already the raw number; export it per partition and alert on its derivative rather than its value.
• Pull-based consumers: expose fetch(offset, max_records) and delete the worker. The log does not change; only who owns the loop does.
• Going distributed is the hand-off: replication and leader election per partition, a rebalance protocol between processes, and durability guarantees — see Messaging, queues and Kafka internals. A single Kafka broker sustains roughly 100 MB/s in, which at ~1 KB per message is about 100k messages/s — that is the scale at which this in-process design stops being the answer.

Interview tip

Volunteer the ordering boundary before you are asked: "per key, because a key maps to one partition and one worker; nothing across partitions." Then say where the offset lives. Those two sentences tell the interviewer you have thought about a real queue rather than a list with a lock around it.

Tests

tests/test_pub_sub_system.py has 22 cases. Every asynchronous assertion uses a barrier — RecordingConsumer.wait_for or Broker.drain — so nothing waits on a fixed delay and nothing is timing-dependent. The one exception is the close-while-blocked test, which parks a producer for 20 ms so the shutdown path it is aiming at is the one that runs.

The group test asserts the two guarantees at once: independent cursors across groups, and per-key order inside one:

code/lld/pub_sub_system/tests/test_pub_sub_system.py — consumer groups

def test_two_groups_each_get_every_record_on_their_own_cursor(broker: Broker) -> None:
    broker.create_topic(ORDERS, partitions=2)
    billing, audit = RecordingConsumer("billing-1"), RecordingConsumer("audit-1")
    broker.subscribe(ORDERS, "billing", billing)
    broker.subscribe(ORDERS, "audit", audit)

    for key, event in [("alice", "created"), ("bob", "created"), ("alice", "paid")]:
        broker.publish(ORDERS, f"{key}:{event}", key=key)
    assert broker.drain(timeout=1.0)

    assert sorted(billing.payloads()) == sorted(audit.payloads())
    assert len(billing.payloads()) == 3
    assert billing.keys_in_order("alice") == ["alice:created", "alice:paid"]  # per-key order
    assert broker.lag("billing", ORDERS) == 0

The retry test uses a consumer that fails once for every payload and always for one poison value, which pins down all three outcomes — retried and acked, dead-lettered, and the partition still making progress:

code/lld/pub_sub_system/tests/test_pub_sub_system.py — retry and dead letters

def test_a_transient_failure_is_retried_and_a_poison_record_is_dead_lettered(broker: Broker) -> None:
    broker.create_topic(ORDERS, partitions=1)
    consumer = FlakyConsumer("shipping-1", fail_times=1, poison="bob:cancelled")
    group = broker.subscribe(ORDERS, "shipping", consumer)

    broker.publish(ORDERS, "alice:paid", key="alice")
    broker.publish(ORDERS, "bob:cancelled", key="bob")
    broker.publish(ORDERS, "alice:shipped", key="alice")
    assert broker.drain(timeout=1.0)

    assert consumer.payloads() == ["alice:paid", "alice:shipped"]  # the poison never acked
    assert consumer.attempts("alice:paid") == 2  # one failure, then success
    [letter] = broker.dlq.letters("shipping")
    assert letter.attempts == 3 and letter.record.payload == "bob:cancelled"
    assert group.stats()["dead_lettered"] == 1
    assert broker.lag("shipping", ORDERS) == 0  # the DLQ let the partition move on

The backpressure test proves the bounded buffer is real: with nobody consuming, the third publish into a two-record partition raises; once a group has consumed and committed, the space is reclaimed and the publish succeeds.

code/lld/pub_sub_system/tests/test_pub_sub_system.py — backpressure

def test_a_full_partition_blocks_the_producer_then_gives_up(broker: Broker) -> None:
    retention = RetentionPolicy(max_messages=2, on_full=FullPolicy.BLOCK, block_timeout=0.02)
    broker.create_topic(ORDERS, partitions=1, retention=retention)
    broker.publish(ORDERS, "first", key="alice")
    broker.publish(ORDERS, "second", key="alice")

    with pytest.raises(BackpressureError):  # nobody is consuming, so nothing can be reclaimed
        broker.publish(ORDERS, "third", key="alice")

    drain_me = RecordingConsumer("billing-1")
    broker.subscribe(ORDERS, "billing", drain_me)
    assert drain_me.wait_for(2, timeout=1.0)
    assert broker.drain(timeout=1.0)
    broker.publish(ORDERS, "third", key="alice")  # the acked records were reclaimed
    assert drain_me.wait_for(3, timeout=1.0)

The concurrency test drives 400 publishes through eight threads into four partitions with two consumers in one group, then asserts that every (partition, offset) pair is unique and that the group delivered all 400 payloads exactly once.

code/lld/pub_sub_system/tests/test_pub_sub_system.py — concurrency

def test_concurrent_producers_lose_no_record_and_assign_unique_offsets(broker: Broker) -> None:
    topic = broker.create_topic(ORDERS, partitions=4, retention=RetentionPolicy(max_messages=512))
    consumers = [RecordingConsumer(f"billing-{i}") for i in range(2)]
    for consumer in consumers:
        broker.subscribe(ORDERS, "billing", consumer)

    def publish(i: int) -> tuple[int, int]:
        record = broker.publish(ORDERS, f"event-{i:03d}", key=f"user-{i}")
        return record.partition, record.offset

    with ThreadPoolExecutor(max_workers=8) as pool:
        placed = list(pool.map(publish, range(400)))

    assert len(set(placed)) == 400  # every (partition, offset) pair is unique
    assert sum(p.next_offset() for p in topic.partitions()) == 400
    assert broker.drain(timeout=2.0)
    delivered = [payload for c in consumers for payload in c.payloads()]
    assert sorted(delivered) == [f"event-{i:03d}" for i in range(400)]  # each exactly once

The rest cover: key-to-partition stability across 50 calls; four invalid requests through parametrize; a duplicate consumer name; the PENDING → IN_FLIGHT → ACKED walk with the committed offset; replay redelivering everything; a late subscriber reading the backlog; DROP_OLDEST shedding three records and fast-forwarding the slow group; age-based trimming on a FakeClock; a rebalance that neither loses nor redelivers; graceful shutdown then BrokerClosedError; the compare-and-set on the offset store; a raising consumer that does not kill its worker; and round-robin spreading for keyless messages. Run them with uv run pytest code/lld/pub_sub_system -q.

45-minute pacing

Minutes	What to do	What to say or write
0–5	Clarify	Push or pull? Group semantics? What ordering do we promise? At-least-once? What happens when a consumer is slow? Out of scope: durability, cross-process, exactly-once.
5–10	Entities	Nouns: Broker, Topic, Partition, Record, ConsumerGroup, OffsetStore, DeliveryWorker, DLQ. State the ordering boundary here, unprompted.
10–18	Class diagram	Log on the left (Topic → Partition → Record), delivery on the right (Group → Worker → Consumer), with the OffsetStore between them.
18–34	Code	Partition.append with the Condition → fetch with wait_for → OffsetStore.advance (compare-and-set) → DeliveryWorker._deliver with retry and DLQ.
34–40	Concurrency	Name the four locks. Walk the bounded buffer and the two full policies with the ~0.18 s of headroom arithmetic. Explain the drain barrier and shutdown.
40–45	Extensions	Persistence behind an interface, effectively-once with an idempotency key, lag as an SLI, and the hand-off to the distributed queue.

Related

• Event Bus — the in-process cousin that forgets instead of remembering
• Observer — the notification mechanic underneath push delivery
• Mediator — why producers and consumers never reference each other
• Concurrency for LLD in Python — Condition, wait_for and the producer-consumer rendezvous
• Messaging, queues and Kafka internals — partitions, consumer groups and delivery semantics at scale
• Design a distributed message queue — the same problem once one process is not enough