LLD: Pub/Sub¶

Difficulty: Advanced | Time: 40–50 minutes

1. Problem Statement¶

Design an in-process publish-subscribe broker. Publishers publish messages to named topics without knowing who — if anyone — is listening. Subscribers subscribe to a topic by name and receive every message published to it after they subscribe. The system supports many independent topics at once, and a subscriber can unsubscribe at any time. Design for many subscribers per topic and frequent, high-volume publish calls — this is a mechanism other in-process components build on, not a one-off feature.

2. Requirements¶

Functional (in scope):

• Create/address topics implicitly by name (publishing or subscribing to an unknown topic name just works — no separate "create topic" step)
• subscribe(topic, callback) registers a callback and returns a subscription handle/id
• unsubscribe(topic, subscription_id) removes that subscriber from that topic
• publish(topic, message) delivers the message to every subscriber currently on that topic
• Many topics, many subscribers per topic, frequent publishes — this must not degrade badly as either number grows
• One subscriber's callback misbehaving (throws, or is slow) must not prevent delivery to the other subscribers on the same topic

Explicitly out of scope for v1: durability (messages surviving a process restart), cross-process or cross-machine delivery, consumer groups / partitioning, persistence/replay of missed messages. That's the job of a real message broker — see Kafka Deep Dive for the distributed, durable version of this same idea. This page is about the in-process primitive: a fan-out mechanism living inside one process's memory.

3. Entities¶

The nouns in the problem statement: Topic (really just a name — a key in the broker's internal map, not a class of its own here), Publisher (any caller of publish() — not a class either, just a role), Subscriber (an interface: something with an on_message callback), Subscription (the handle returned by subscribe(), needed to unsubscribe later), and the Broker (a.k.a. PubSubSystem) that owns the topic → subscriber-list mapping and does the dispatching.

4. Class Design¶

classDiagram
    class Subscriber {
        <<interface>>
        +on_message(message)* void
    }

    class Subscription {
        +subscription_id: str
        +topic: str
        +callback: Callable
    }

    class Broker {
        -_subscribers: Dict~str, Dict~str, Subscription~~
        -_lock: Lock
        -_executor: ThreadPoolExecutor
        +subscribe(topic, callback) str
        +unsubscribe(topic, subscription_id) void
        +publish(topic, message) void
        +shutdown() void
    }

    class Publisher {
        <<role>>
        publish(topic, message)
    }

    Broker o-- Subscription : holds, one map per topic
    Publisher --> Broker : calls publish()
    Subscription --> Subscriber : wraps a callback

This is Observer generalized from "one subject, one observer list" to "N independent subjects (topics), each with its own observer list, addressed by name instead of by object reference." The ParkingLot in the Observer pattern writeup has exactly one _observers: list[Observer]. Here the broker owns _subscribers: dict[str, dict[str, Subscription]] — topic name maps to a collection of subscriptions.

Why a dict of dicts, not a dict of lists: the inner collection needs to support unsubscribe(topic, subscription_id) — find-and-remove one specific subscriber out of potentially thousands. A list[Subscriber] makes that an O(n) linear scan; a dict[str, Subscription] keyed by subscription_id makes it O(1). At "many subscribers per topic" scale, that difference is the entire point of the exercise (see Patterns Applied below) — pick the data structure for the operation you'll actually be doing frequently, not just the one that's simplest to write.

5. Patterns Applied¶

• Observer is the whole pattern here — one topic publishing a message is exactly "notify interested parties without hard-coding who they are." See Design Patterns — Observer. The generalization from the textbook version is going from one subject to N named subjects (topics) sharing one broker, which is why the broker's core state is a map of observer collections, not a single one.

• What breaks in the naive Observer implementation at this scale, concretely:

  • The textbook Observer example uses self._observers: list[Observer] and self._observers.append(observer) for subscribe. Removing one specific subscriber (list.remove(observer)) is O(n) — acceptable for a handful of dashboard widgets watching one ParkingLot, but not when a topic can have thousands of subscribers and subscribe/unsubscribe churn is frequent (e.g. UI components mounting/unmounting, each subscribing on mount and unsubscribing on unmount). At that scale, O(n) unsubscribe under contention becomes a measurable bottleneck. The fix is keying the collection by a subscription_id in a dict/set, making both subscribe and unsubscribe O(1).
  • The textbook example calls obs.update(event) directly in a loop inside _notify(), on the calling thread. That's fine when observers are trusted, fast, local objects. It is not fine at pub/sub scale where subscribers are arbitrary, possibly slow, possibly buggy callbacks — the whole reason this page treats dispatch strategy (sync in-line vs. thread pool) as a first-class design decision rather than an implementation detail. See Core Code and Concurrency below.
  • Iterating the live observer list while another thread mutates it (a subscribe or unsubscribe arriving mid-publish) is exactly the "RuntimeError: set/dict changed size during iteration" failure mode — not present in the single-threaded textbook example, front and center here.

6. Core Code¶

from __future__ import annotations

import logging
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Callable

logger = logging.getLogger(__name__)

Message = object          # any payload; kept generic on purpose
Callback = Callable[[Message], None]


@dataclass(frozen=True)
class Subscription:
    subscription_id: str
    topic: str
    callback: Callback


class Broker:
    """In-process publish-subscribe broker.

    Delivery is asynchronous (dispatched via a thread pool) and best-effort
    at-least-once within the process: a subscriber that's registered when
    publish() is called will be handed the message exactly once, unless its
    own callback raises after partially processing it (out of scope to
    de-duplicate that — see Edge Cases).
    """

    def __init__(self, max_workers: int = 16):
        self._subscribers: dict[str, dict[str, Subscription]] = {}
        self._lock = threading.Lock()
        self._executor = ThreadPoolExecutor(max_workers=max_workers)

    def subscribe(self, topic: str, callback: Callback) -> str:
        subscription_id = str(uuid.uuid4())
        sub = Subscription(subscription_id, topic, callback)
        with self._lock:
            self._subscribers.setdefault(topic, {})[subscription_id] = sub
        return subscription_id

    def unsubscribe(self, topic: str, subscription_id: str) -> None:
        with self._lock:
            topic_subs = self._subscribers.get(topic)
            if topic_subs is not None:
                topic_subs.pop(subscription_id, None)   # no-op if already gone

    def publish(self, topic: str, message: Message) -> None:
        # Snapshot under the lock, then release before dispatching — a publish
        # must never hold the lock while calling into subscriber code (see
        # Concurrency). list() copies the values out of the dict at this instant;
        # subscribes/unsubscribes that happen after this line don't affect this
        # publish's delivery set.
        with self._lock:
            subs = list(self._subscribers.get(topic, {}).values())

        if not subs:
            return                                        # zero subscribers: no-op, not an error

        for sub in subs:
            self._executor.submit(self._deliver, sub, message)

    def _deliver(self, sub: Subscription, message: Message) -> None:
        # Runs on a worker thread. Exception isolation: one subscriber's
        # callback throwing must not prevent delivery to any other subscriber,
        # nor propagate back to the publisher's thread.
        try:
            sub.callback(message)
        except Exception:
            logger.exception(
                "subscriber %s on topic %r raised while handling a message; "
                "isolated, other subscribers unaffected",
                sub.subscription_id, sub.topic,
            )

    def shutdown(self, wait: bool = True) -> None:
        self._executor.shutdown(wait=wait)

Why ThreadPoolExecutor.submit per subscriber, not a synchronous loop calling every callback in turn: a synchronous loop means publish() doesn't return until the slowest subscriber's callback finishes, and one subscriber that hangs blocks delivery to every other subscriber on that publish call, and blocks the publisher itself. Dispatching each delivery as an independent submitted task means the publisher returns as soon as the snapshot is taken and the tasks are queued — a slow or hung subscriber only delays itself.

7. Edge Cases¶

8. Concurrency¶

This is the crux of the exercise — "Observer at scale, thread-safe subscriber management." Three operations touch the same shared state (_subscribers) from different threads at will: subscribe, unsubscribe, and publish.

The bug a naive version has: iterate the live subscriber collection directly inside publish() while another thread calls subscribe() or unsubscribe() on the same topic. Mutating a dict/set while another thread iterates it raises RuntimeError: dict changed size during iteration (or, without that safety check, silently skips or double-visits entries) — the same class of race as Race Conditions, just surfacing as a crash instead of a lost update.

sequenceDiagram
    participant P as Publisher
    participant B as Broker
    participant D as _subscribers dict\n(guarded by _lock)
    participant Pool as ThreadPoolExecutor
    participant S1 as Subscriber A
    participant S2 as Subscriber B

    P->>B: publish(topic, message)
    activate B
    B->>B: acquire _lock
    B->>D: list(_subscribers[topic].values())
    D-->>B: snapshot [subA, subB]
    B->>B: release _lock
    Note over B: lock released BEFORE any subscriber code runs
    B->>Pool: submit(_deliver, subA, message)
    B->>Pool: submit(_deliver, subB, message)
    B-->>P: return (does not wait for callbacks)
    deactivate B
    par async delivery
        Pool->>S1: sub.callback(message)
    and
        Pool->>S2: sub.callback(message)
    end

The fix — snapshot, then release, then dispatch:

with self._lock:
    subs = list(self._subscribers.get(topic, {}).values())   # copy, under the lock
# lock is released here — before any subscriber callback runs
for sub in subs:
    self._executor.submit(self._deliver, sub, message)

list(...) copies the values out of the dict while the lock is held, so the copy is internally consistent — no other thread can be mid-mutation of that dict at the moment of the copy. Once the copy exists, the lock is released immediately; nothing about calling _deliver (or even just submit-ing it) happens while the lock is held.

Why the lock must not be held across dispatch: this is the exact same lesson as "don't hold a lock across I/O" from Locks — a subscriber's callback is arbitrary code the broker doesn't control, potentially slow, potentially making a network call. If publish() held self._lock for the duration of calling every subscriber's callback:

• every other publish() call — even to a different topic, since this implementation uses one broker-wide lock — would serialize behind it, meaning the whole system's publish throughput degrades to the speed of the single slowest subscriber callback anywhere.
• subscribe()/unsubscribe() calls would also block for that same duration, since they need the same lock.
• worst case, a subscriber callback that itself calls broker.subscribe() or broker.unsubscribe() (e.g. "unsubscribe myself after handling this message") would deadlock outright — the callback is running on a thread that's waiting to re-acquire a lock its own call stack is already holding.

Copy-then-release avoids all three: the lock's critical section is just "copy references out of a dict," which is fast and bounded, never proportional to subscriber count or subscriber behavior.

Why a broker-wide lock instead of a per-topic lock: subscribe/unsubscribe/snapshot are all O(1)-ish dict operations already — the critical section is short regardless of topic count, so a single lock's contention cost is low compared to the complexity of managing one lock per topic (including the bookkeeping of creating/tearing down per-topic locks as topics come and go). If profiling showed the single lock was a bottleneck under many concurrent topics, sharding to a per-topic lock (or a dict of locks keyed by topic name) would be the natural next step — the same "coarse vs. fine-grained" trade-off discussed in the Parking Lot exercise.

9. Extensibility¶

Interview Questions¶

Key Takeaways¶

Previous: Notification System (LLD) | Next: Task Scheduler