Skip to content

Design a notification service (LLD)

TL;DR

  • You build an admission pipeline (preferences → dedup → rate limit → template → enqueue), a bounded priority queue, and a dispatcher that sends, retries, falls back to the next channel and finally dead-letters.
  • Three decisions carry the interview: two different idempotency guards (a dedup window on the request side, a delivery ledger on the send side), a circuit breaker per provider so a failing SMS vendor cannot slow email down, and preferences re-checked at send time, not at enqueue time.
  • Strategy, Decorator, Pipeline and Null Object earn their place. A thread-per-notification model is discussed and deliberately not used.

Problem statement

"Design the service other teams call when they want to tell a user something. It renders a template, respects the user's channel preferences and opt-outs, does not spam them, sends over email, SMS, push or in-app, retries when a provider fails, falls back to another channel, and gives up into a dead letter queue. Some sends are scheduled, some are urgent. Focus on the classes, the delivery path, and what stops a user getting the same message twice."

Requirements

Functional

  • Four channels — email, SMS, push, in-app — behind one sender interface, each with a provider stub.
  • Templates with {placeholder} slots, registered per (event, channel); a payload that does not fill every placeholder is an error, not a blank.
  • Per-user preferences: muted events, disabled channels, addresses per channel, and an ordered channel list that doubles as the fallback chain.
  • A dedup window: the same idempotency key inside it produces one notification, not two.
  • A per-user rate limit that a CRITICAL priority walks past.
  • A bounded queue ordered by due time then priority, so scheduled sends and urgent sends both work and an outage cannot exhaust memory.
  • Retry with exponential backoff and jitter; then fallback to the next allowed channel; then the dead letter queue.
  • Delivery status per notification, with the attempt history.

Non-functional and constraints

  • At-least-once transport must not become at-least-twice delivery: the same message on the same channel is handed to a provider at most once.
  • A failing provider must be isolated, not merely retried.
  • In-memory, single process; deterministic and testable with an injected clock, ID generators and a seeded jitter source.

Out of scope: the real providers, digests and quiet hours (both named as extensions), open and click tracking, template A/B testing, cross-region delivery.

Clarifying questions and assumptions

Question to ask Assumption taken here
Do we send on every allowed channel, or one? One, with the rest as fallbacks. Blasting all four is a product decision that turns "fallback" into "spam".
Where does the idempotency key come from? The caller. If they omit it, dedup_key() falls back to user:event, so a forgetful caller still gets deduplication.
Is a rate-limited notification dropped or delayed? Dropped, with a reason the caller can see. Delaying it silently is how you deliver "your code is 2FA…" ten minutes late.
What if the queue is full? notify raises QueueFullError. Backpressure the caller can see beats an unbounded queue that dies as an out-of-memory kill.
What if the user opts out while a notification is queued? The dispatcher re-checks at send time and marks it SUPPRESSED. Checking only at enqueue time is how people get mail after unsubscribing.
Should a retry be a new notification? No. One Notification is one delivery chain: attempts, channel switches and history all hang off it.
Do we need workers and threads? The dispatcher exposes run_once(); a worker is a thread calling it in a loop. That keeps the tests deterministic and the concurrency real.

Core entities and relationships

  • NotificationRequest — what a caller asks for: user, event, payload, priority, idempotency key, optional send_after. Frozen.
  • Notification — one delivery chain on one channel, with its fallbacks, attempt count, DeliveryStatus and DeliveryAttempt history. The only mutable domain object.
  • Template 1 → 1 (event, channel); TemplateEngine renders it into a frozen RenderedMessage.
  • UserPreferences — muted events, disabled channels, addresses, and channel_order, which is the fallback chain.
  • Pipeline 1 → * Stage: PreferenceStage, DedupStage, RateLimitStage, TemplateStage, EnqueueStage. Each narrows the context or stops it.
  • DedupStore (request side) and DeliveryLedger (send side) — two different idempotency guards, described below.
  • NotificationQueue — bounded, ordered by (due_at, -priority). Dispatcher — the worker step. NotificationService — the facade.
  • ChannelSender — the provider interface; RecordingSender, FlakySender and NullSender implement it, and CircuitBreakerSender decorates any of them.
  • RetryPolicy — ExponentialBackoff or NoRetry. CircuitBreaker — one per provider.

Class diagram

The admission path: request in, notification queued or suppressed.

classDiagram
    class NotificationService {
        +notify(user_id, event, payload, priority, idempotency_key, send_after, channels) NotifyResult
        +stages() List~str~
        +pending() int
    }
    class Pipeline {
        +run(ctx) PipelineContext
        +names() List~str~
    }
    class Stage {
        <<interface>>
        +name: str
        +process(ctx) bool
    }
    class PreferenceStage
    class DedupStage
    class RateLimitStage
    class TemplateStage
    class EnqueueStage
    class PipelineContext {
        +request: NotificationRequest
        +channels: List~Channel~
        +messages: Dict~Channel,RenderedMessage~
        +suppressions: List~Suppression~
        +drop(channel, reason, detail)
    }
    class NotificationRequest {
        +user_id: str
        +event: str
        +priority: Priority
        +idempotency_key: str
        +send_after: float
        +dedup_key() str
    }
    class UserPreferences {
        +channel_order: tuple
        +muted_events: frozenset
        +addresses: Dict~Channel,str~
        +allows(event, channel) bool
    }
    class TemplateEngine {
        +register(template) Template
        +render(template, notification_id, recipient, payload, idempotency_key) RenderedMessage
    }
    class Template {
        +event: str
        +channel: Channel
        +placeholders() frozenset
    }
    class DedupStore {
        +claim(key) bool
    }
    class TokenBucketRateLimiter {
        +allow(key) bool
    }
    class NotifyResult {
        +notification: Notification
        +suppressions: tuple
        +accepted() bool
    }
    NotificationService --> Pipeline
    Pipeline "1" o-- "many" Stage
    Stage <|.. PreferenceStage
    Stage <|.. DedupStage
    Stage <|.. RateLimitStage
    Stage <|.. TemplateStage
    Stage <|.. EnqueueStage
    Pipeline ..> PipelineContext
    PipelineContext --> NotificationRequest
    PipelineContext --> UserPreferences
    TemplateStage --> TemplateEngine
    TemplateEngine "1" o-- "many" Template
    DedupStage --> DedupStore
    RateLimitStage --> TokenBucketRateLimiter
    NotificationService ..> NotifyResult : returns

The delivery path: queue, dispatcher, senders and their two decorators.

classDiagram
    class NotificationQueue {
        -_lock: Lock
        +put(notification)
        +take(now) Notification
    }
    class Notification {
        +channel: Channel
        +status: DeliveryStatus
        +attempts: int
        +fallbacks: tuple
        +claim()
        +succeed(provider_message_id, at)
        +schedule_retry(due_at)
        +switch_to(channel, message, due_at)
        +dead_letter()
    }
    class DeliveryAttempt {
        +attempt: int
        +ok: bool
        +error: str
    }
    class Dispatcher {
        +run_once() Notification
        +drain(limit) List~Notification~
        +sender_for(channel) ChannelSender
        +dead_letters() List~Notification~
    }
    class DeliveryLedger {
        +claim(key, channel) bool
        +release(key, channel)
        +delivered() int
    }
    class ChannelSender {
        <<interface>>
        +channel: Channel
        +send(message) str
    }
    class RecordingSender
    class FlakySender
    class NullSender
    class CircuitBreakerSender {
        +send(message) str
    }
    class CircuitBreaker {
        -_lock: Lock
        +allow() bool
        +record_success()
        +record_failure()
        +state CircuitState
    }
    class RetryPolicy {
        <<interface>>
        +max_attempts: int
        +delay(attempt) float
    }
    class ExponentialBackoff
    class NoRetry
    class PreferenceStore {
        +get(user_id) UserPreferences
        +set(preferences) UserPreferences
    }
    class DeliveryStatus {
        <<enumeration>>
        QUEUED
        SENDING
        RETRYING
        DELIVERED
        SUPPRESSED
        DEAD_LETTER
    }
    NotificationQueue "1" o-- "many" Notification
    Notification "1" o-- "many" DeliveryAttempt
    Notification --> DeliveryStatus
    Dispatcher --> NotificationQueue
    Dispatcher --> DeliveryLedger
    Dispatcher --> RetryPolicy
    Dispatcher --> PreferenceStore
    Dispatcher --> ChannelSender
    ChannelSender <|.. RecordingSender
    ChannelSender <|.. FlakySender
    ChannelSender <|.. NullSender
    ChannelSender <|.. CircuitBreakerSender
    CircuitBreakerSender --> ChannelSender : wraps
    CircuitBreakerSender --> CircuitBreaker
    RetryPolicy <|.. ExponentialBackoff
    RetryPolicy <|.. NoRetry

Design patterns applied

Pattern Where Why it earns its place
Pipeline Pipeline over five Stages Quiet hours is a sixth stage in a list. Without it, notify becomes a 60-line function with five nested ifs and no way to test one rule alone.
Strategy ChannelSender, RetryPolicy The provider and the backoff are the two things that differ per channel. NoRetry for in-app and ExponentialBackoff for SMS is configuration, not code.
Decorator CircuitBreakerSender wrapping any ChannelSender Provider isolation is orthogonal to what a provider does, so it wraps rather than subclasses. Stack another decorator for metrics and nothing else changes.
Null Object NullSender An unconfigured channel accepts and discards. sender_for never returns None, so the dispatcher has no null check and a missing provider is a no-op instead of an AttributeError in a worker at 3 a.m.
Template Method (lightweight) Notification.claim / succeed / fail / schedule_retry The status machine lives on the entity; the dispatcher decides which transition, never how to apply it.
Facade NotificationService.notify Callers pass a user, an event and a payload. They never see the pipeline, the queue or the ledger.
Dependency injection Clock, two IdGenerators, random.Random for jitter, every collaborator Backoff delays are asserted exactly, and the circuit breaker's cooldown is tested by moving a FakeClock, not by sleeping.
Value object RenderedMessage, NotificationRequest, Suppression A provider stub receives a frozen message and cannot reach back into the domain.

What was deliberately not used: a thread per notification. It is the first thing people reach for and it fails at both ends — thousands of threads under load, and no way to bound work or apply backpressure. A bounded queue plus a fixed worker pool gives you a queue depth you can alarm on, a natural place to shed load, and a run_once() you can call from a test without threads at all.

Key flows

Delivery: claim, re-check, send, and the three ways to fail.

sequenceDiagram
    autonumber
    participant W as Worker
    participant Q as NotificationQueue
    participant P as PreferenceStore
    participant L as DeliveryLedger
    participant B as CircuitBreakerSender
    participant Pr as Provider
    W->>Q: take(now)
    Q-->>W: notification (due, highest priority)
    W->>P: get(user_id)
    P-->>W: preferences
    alt channel no longer allowed
        W->>W: switch to a fallback, or mark SUPPRESSED
    else still wanted
        W->>L: claim(idempotency_key, channel)
        alt already claimed
            L-->>W: False - another worker sent it
            W->>W: mark DELIVERED (deduplicated)
        else first claim
            L-->>W: True
            W->>B: send(message)
            alt circuit open
                B-->>W: ChannelUnavailableError (provider not called)
            else provider called
                B->>Pr: deliver
                Pr-->>B: provider message id or error
                B-->>W: result, breaker updated
            end
            opt failed
                W->>L: release(key, channel)
                W->>Q: re-queue with backoff, fall back, or dead-letter
            end
        end
    end

Notification lifecycle. Every arrow is a method on Notification, which is why no service can invent a transition.

stateDiagram-v2
    [*] --> QUEUED : pipeline accepted it
    QUEUED --> SENDING : worker claims it
    QUEUED --> SUPPRESSED : preferences changed while queued
    SENDING --> DELIVERED : provider accepted
    SENDING --> DELIVERED : ledger says another worker already sent it
    SENDING --> RETRYING : attempt failed, attempts left
    RETRYING --> SENDING : backoff elapsed
    SENDING --> QUEUED : attempts exhausted, fall back to the next channel
    SENDING --> DEAD_LETTER : no attempts and no channels left
    DELIVERED --> [*]
    SUPPRESSED --> [*]
    DEAD_LETTER --> [*]

Circuit breaker. One per provider, so a bad SMS vendor costs SMS traffic and nothing else.

stateDiagram-v2
    [*] --> CLOSED
    CLOSED --> OPEN : failures reach the threshold
    OPEN --> HALF_OPEN : cooldown elapsed
    HALF_OPEN --> CLOSED : the probe succeeded
    HALF_OPEN --> OPEN : the probe failed

Implementation

Write the vocabulary first; in this problem the enums are half the design.

Priority.bypasses_rate_limit and the DeliveryStatus set are the two decisions readers will question, so put them where they can see them:

code/lld/notification_service/models.py — the vocabulary
class Channel(StrEnum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"
    IN_APP = "in_app"


class Priority(StrEnum):
    LOW = "low"
    NORMAL = "normal"
    HIGH = "high"
    CRITICAL = "critical"

    @property
    def rank(self) -> int:
        """Higher wins. The queue orders by due time first, then by this."""
        return {"low": 0, "normal": 1, "high": 2, "critical": 3}[self.value]

    @property
    def bypasses_rate_limit(self) -> bool:
        """A password reset is never dropped because you posted too much today."""
        return self is Priority.CRITICAL


class DeliveryStatus(StrEnum):
    QUEUED = "queued"  # accepted, waiting for a worker
    SENDING = "sending"  # claimed by a worker, provider call in flight
    RETRYING = "retrying"  # the attempt failed, backoff is running
    DELIVERED = "delivered"
    SUPPRESSED = "suppressed"  # preferences changed while it sat in the queue
    DEAD_LETTER = "dead_letter"  # every channel and every attempt exhausted


class SuppressionReason(StrEnum):
    OPTED_OUT = "opted_out"
    NO_CHANNEL = "no_channel"
    DUPLICATE = "duplicate"
    RATE_LIMITED = "rate_limited"


class CircuitState(StrEnum):
    CLOSED = "closed"  # calls flow through
    OPEN = "open"  # provider is failing; fail fast without calling it
    HALF_OPEN = "half_open"  # cooldown elapsed; let one probe through

Notification is the only mutable object in the package, and every transition is a named method:

code/lld/notification_service/models.py — the notification
@dataclass(slots=True)
class Notification:
    """One delivery chain: a channel, its fallbacks, and the attempts so far."""

    id: str
    request_id: str
    user_id: str
    event: str
    channel: Channel
    message: RenderedMessage
    payload: dict[str, str] = field(default_factory=dict)  # kept so a fallback can re-render
    priority: Priority = Priority.NORMAL
    status: DeliveryStatus = DeliveryStatus.QUEUED
    attempts: int = 0
    due_at: float = 0.0
    fallbacks: tuple[Channel, ...] = ()
    history: list[DeliveryAttempt] = field(default_factory=list)
    provider_message_id: str | None = None
    last_error: str = ""

    def claim(self) -> None:
        if self.status not in (DeliveryStatus.QUEUED, DeliveryStatus.RETRYING):
            raise NotificationStateError(f"{self.id} is {self.status}, not claimable")
        self.status = DeliveryStatus.SENDING
        self.attempts += 1

    def succeed(self, provider_message_id: str, at: float) -> None:
        self.status = DeliveryStatus.DELIVERED
        self.provider_message_id = provider_message_id
        self.history.append(DeliveryAttempt(self.attempts, self.channel, at, ok=True))

    def fail(self, error: str, at: float) -> None:
        self.last_error = error
        self.history.append(DeliveryAttempt(self.attempts, self.channel, at, ok=False, error=error))

    def schedule_retry(self, due_at: float) -> None:
        self.status = DeliveryStatus.RETRYING
        self.due_at = due_at

    def switch_to(self, channel: Channel, message: RenderedMessage, due_at: float) -> None:
        """Fallback: same notification, next channel, attempt counter reset."""
        self.channel = channel
        self.message = message
        self.fallbacks = tuple(c for c in self.fallbacks if c is not channel)
        self.attempts = 0
        self.status = DeliveryStatus.QUEUED
        self.due_at = due_at

    def dead_letter(self) -> None:
        self.status = DeliveryStatus.DEAD_LETTER

    def suppress(self) -> None:
        self.status = DeliveryStatus.SUPPRESSED

    def summary(self) -> str:
        return f"{self.id} {self.event} via {self.channel}: {self.status} after {self.attempts} attempt(s)"


@dataclass(frozen=True, slots=True)
class NotifyResult:
    """What ``notify`` answers: what was accepted and what was dropped, and why."""

    request: NotificationRequest
    notification: Notification | None
    suppressions: tuple[Suppression, ...] = ()

    def accepted(self) -> bool:
        return self.notification is not None

    def line(self) -> str:
        if self.notification is not None:
            return f"{self.request.event} -> {self.notification.channel} ({self.notification.status})"
        reasons = ", ".join(s.reason for s in self.suppressions)
        return f"{self.request.event} -> suppressed ({reasons})"

The three collaborators the pipeline leans on. DedupStore.claim is one compare-and-set under a lock; TokenBucketRateLimiter refills from the injected clock, so there is no timer thread anywhere:

code/lld/notification_service/pipeline.py — engine, dedup and rate limit
class TemplateEngine:
    """Registry plus renderer. A missing placeholder is an error, not an empty string."""

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

    def register(self, template: Template) -> Template:
        with self._lock:
            self._templates[(template.event, template.channel)] = template
        return template

    def template_for(self, event: str, channel: Channel) -> Template:
        with self._lock:
            try:
                return self._templates[(event, channel)]
            except KeyError:
                raise TemplateNotFoundError(f"no {channel} template for event {event!r}") from None

    def has(self, event: str, channel: Channel) -> bool:
        with self._lock:
            return (event, channel) in self._templates

    def render(
        self,
        template: Template,
        notification_id: str,
        recipient: str,
        payload: dict[str, str],
        idempotency_key: str,
    ) -> RenderedMessage:
        missing = sorted(template.placeholders() - payload.keys())
        if missing:
            raise RenderError(f"template {template.id} needs {', '.join(missing)}")
        fill = PLACEHOLDER.sub(lambda m: str(payload[m.group(1)]), template.subject)
        body = PLACEHOLDER.sub(lambda m: str(payload[m.group(1)]), template.body)
        return RenderedMessage(notification_id, template.channel, recipient, fill, body, idempotency_key)


class DedupStore:
    """A sliding window on the *request* side: the same key twice inside it is one send."""

    def __init__(self, clock: Clock, window_seconds: float = 300.0) -> None:
        self._clock = clock
        self._window = window_seconds
        self._lock = threading.Lock()
        self._seen: dict[str, float] = {}

    def claim(self, key: str) -> bool:
        """True the first time; False while an earlier claim is still inside the window."""
        now = self._clock.now()
        with self._lock:
            previous = self._seen.get(key)
            if previous is not None and now - previous < self._window:
                return False
            self._seen[key] = now
            return True

    def forget(self, key: str) -> None:
        with self._lock:
            self._seen.pop(key, None)


class TokenBucketRateLimiter:
    """Per-user budget. Refills continuously from the injected clock — no timer thread."""

    def __init__(self, clock: Clock, capacity: int = 5, refill_per_second: float = 0.2) -> None:
        self._clock = clock
        self._capacity = capacity
        self._refill = refill_per_second
        self._lock = threading.Lock()
        self._buckets: dict[str, tuple[float, float]] = {}

    def allow(self, key: str) -> bool:
        now = self._clock.now()
        with self._lock:
            tokens, last = self._buckets.get(key, (float(self._capacity), now))
            tokens = min(self._capacity, tokens + (now - last) * self._refill)
            if tokens < 1.0:
                self._buckets[key] = (tokens, now)
                return False
            self._buckets[key] = (tokens - 1.0, now)
            return True

    def tokens(self, key: str) -> float:
        with self._lock:
            return round(self._buckets.get(key, (float(self._capacity), 0.0))[0], 3)

Now the pipeline. Note the order: three cheap filters, then rendering. The brief's ordering renders first; render last, because rendering is the expensive stage and most drops happen before it — say that trade out loud rather than silently reordering.

code/lld/notification_service/pipeline.py — the five stages
@dataclass(slots=True)
class PipelineContext:
    """What flows through the stages. Stages narrow it; they never widen it."""

    request: NotificationRequest
    preferences: UserPreferences
    notification_id: str
    channels: list[Channel] = field(default_factory=list)
    messages: dict[Channel, RenderedMessage] = field(default_factory=dict)
    suppressions: list[Suppression] = field(default_factory=list)
    notification: Notification | None = None

    def drop(self, channel: Channel | None, reason: SuppressionReason, detail: str = "") -> None:
        self.suppressions.append(Suppression(channel, reason, detail))


class Stage(Protocol):
    """One question, asked once. ``False`` stops the pipeline."""

    name: str

    def process(self, ctx: PipelineContext) -> bool: ...


class NotificationSink(Protocol):
    """Whatever the last stage hands the notification to (the queue, in practice)."""

    def put(self, notification: Notification) -> None: ...


class PreferenceStage:
    """Opt-outs and the fallback order. First, because it drops the most traffic."""

    name = "preferences"

    def process(self, ctx: PipelineContext) -> bool:
        wanted = ctx.request.channels or tuple(ctx.preferences.channel_order)
        ctx.channels = [c for c in wanted if ctx.preferences.allows(ctx.request.event, c)]
        if ctx.channels:
            return True
        muted = ctx.request.event in ctx.preferences.muted_events
        ctx.drop(
            None,
            SuppressionReason.OPTED_OUT if muted else SuppressionReason.NO_CHANNEL,
            f"{ctx.request.user_id} has no usable channel for {ctx.request.event}",
        )
        return False


class DedupStage:
    """The same key inside the window is the same notification, not a second one."""

    name = "dedup"

    def __init__(self, store: DedupStore) -> None:
        self._store = store

    def process(self, ctx: PipelineContext) -> bool:
        if self._store.claim(ctx.request.dedup_key()):
            return True
        ctx.drop(None, SuppressionReason.DUPLICATE, ctx.request.dedup_key())
        return False


class RateLimitStage:
    """Per-user budget, which a CRITICAL request walks past."""

    name = "rate_limit"

    def __init__(self, limiter: TokenBucketRateLimiter) -> None:
        self._limiter = limiter

    def process(self, ctx: PipelineContext) -> bool:
        if ctx.request.priority.bypasses_rate_limit or self._limiter.allow(ctx.request.user_id):
            return True
        ctx.drop(None, SuppressionReason.RATE_LIMITED, ctx.request.user_id)
        return False


class TemplateStage:
    """Render each surviving channel. Deliberately *after* the three cheap filters:
    rendering is the expensive step and most drops happen before it."""

    name = "template"

    def __init__(self, engine: TemplateEngine) -> None:
        self._engine = engine

    def process(self, ctx: PipelineContext) -> bool:
        usable: list[Channel] = []
        for channel in ctx.channels:
            if not self._engine.has(ctx.request.event, channel):
                ctx.drop(channel, SuppressionReason.NO_CHANNEL, "no template")
                continue
            ctx.messages[channel] = self._engine.render(
                self._engine.template_for(ctx.request.event, channel),
                ctx.notification_id,
                ctx.preferences.addresses[channel],
                ctx.request.payload,
                ctx.request.dedup_key(),
            )
            usable.append(channel)
        ctx.channels = usable
        if usable:
            return True
        ctx.drop(None, SuppressionReason.NO_CHANNEL, "no template for any allowed channel")
        return False


class EnqueueStage:
    """Build one notification for the preferred channel; the rest become fallbacks."""

    name = "enqueue"

    def __init__(self, queue: NotificationSink, clock: Clock, ids: IdGenerator) -> None:
        self._queue = queue
        self._clock = clock
        self._ids = ids

    def process(self, ctx: PipelineContext) -> bool:
        primary, *fallbacks = ctx.channels
        notification = Notification(
            id=ctx.notification_id,
            request_id=ctx.request.id,
            user_id=ctx.request.user_id,
            event=ctx.request.event,
            channel=primary,
            message=ctx.messages[primary],
            payload=dict(ctx.request.payload),
            priority=ctx.request.priority,
            due_at=max(self._clock.now(), ctx.request.send_after),
            fallbacks=tuple(fallbacks),
        )
        ctx.notification = notification
        self._queue.put(notification)
        return True


class Pipeline:
    """Runs the stages in order and stops at the first ``False``."""

    def __init__(self, stages: Sequence[Stage]) -> None:
        self._stages = tuple(stages)

    def names(self) -> list[str]:
        return [s.name for s in self._stages]

    def run(self, ctx: PipelineContext) -> PipelineContext:
        for stage in self._stages:
            if not stage.process(ctx):
                return ctx
        return ctx

Senders are a Protocol, so a provider is anything with a channel and a send. NullSender is the Null Object that removes every null check downstream:

code/lld/notification_service/channels.py — the senders
class ChannelSender(Protocol):
    """One provider. Returns the provider's message id, or raises to signal failure."""

    channel: Channel

    def send(self, message: RenderedMessage) -> str: ...


class RecordingSender:
    """The stub every channel uses in this package: it records and returns an id.

    Real senders differ only in what they do with ``message``; the pipeline,
    the retry decorator and the breaker never learn which one they hold.
    """

    def __init__(self, channel: Channel, prefix: str) -> None:
        self.channel = channel
        self._prefix = prefix
        self._lock = threading.Lock()
        self.sent: list[RenderedMessage] = []

    def send(self, message: RenderedMessage) -> str:
        with self._lock:
            self.sent.append(message)
            return f"{self._prefix}-{len(self.sent)}"

    def count(self) -> int:
        with self._lock:
            return len(self.sent)


class NullSender:
    """Null Object: a channel with no provider configured.

    It accepts and discards, so the dispatcher never needs
    ``if sender is None`` and a missing provider degrades to a no-op instead of
    an AttributeError at three in the morning.
    """

    def __init__(self, channel: Channel) -> None:
        self.channel = channel

    def send(self, message: RenderedMessage) -> str:
        return "null"


class FlakySender:
    """Fails its first ``failures`` calls, then succeeds. Deterministic by design."""

    def __init__(self, channel: Channel, failures: int, error: str = "provider timeout") -> None:
        self.channel = channel
        self._remaining = failures
        self._error = error
        self._lock = threading.Lock()
        self.calls = 0

    def send(self, message: RenderedMessage) -> str:
        with self._lock:
            self.calls += 1
            if self._remaining > 0:
                self._remaining -= 1
                raise ChannelUnavailableError(f"{self.channel}: {self._error}")
            return f"{self.channel}-{self.calls}"

The breaker and its decorator. The decorator is nine lines and it is the whole of provider isolation:

code/lld/notification_service/channels.py — the circuit breaker
class CircuitBreaker:
    """One breaker per provider. CLOSED -> OPEN -> HALF_OPEN -> CLOSED.

    ``_lock`` guards the failure counter and the state. The point is isolation:
    an SMS provider that starts timing out must not hold up email, and once it
    is known bad every call fails in nanoseconds instead of waiting on a socket.
    """

    def __init__(self, clock: Clock, threshold: int = 3, cooldown: float = 30.0) -> None:
        self._clock = clock
        self._threshold = threshold
        self._cooldown = cooldown
        self._lock = threading.Lock()
        self._failures = 0
        self._state = CircuitState.CLOSED
        self._opened_at = 0.0

    @property
    def state(self) -> CircuitState:
        with self._lock:
            return self._refresh()

    def allow(self) -> bool:
        with self._lock:
            return self._refresh() is not CircuitState.OPEN

    def record_success(self) -> None:
        with self._lock:
            self._failures = 0
            self._state = CircuitState.CLOSED

    def record_failure(self) -> None:
        with self._lock:
            if self._refresh() is CircuitState.HALF_OPEN:
                self._trip()  # the probe failed: straight back to OPEN
                return
            self._failures += 1
            if self._failures >= self._threshold:
                self._trip()

    def _trip(self) -> None:
        self._state = CircuitState.OPEN
        self._opened_at = self._clock.now()

    def _refresh(self) -> CircuitState:
        """Caller holds the lock. Moves OPEN to HALF_OPEN once the cooldown elapses."""
        if self._state is CircuitState.OPEN and self._clock.now() - self._opened_at >= self._cooldown:
            self._state = CircuitState.HALF_OPEN
        return self._state


class CircuitBreakerSender:
    """Decorator: fails fast while the provider is known bad."""

    def __init__(self, inner: ChannelSender, breaker: CircuitBreaker) -> None:
        self.channel = inner.channel
        self._inner = inner
        self._breaker = breaker

    def send(self, message: RenderedMessage) -> str:
        if not self._breaker.allow():
            raise ChannelUnavailableError(f"{self.channel}: circuit open, not calling the provider")
        try:
            provider_id = self._inner.send(message)
        except Exception:
            self._breaker.record_failure()
            raise
        self._breaker.record_success()
        return provider_id

The queue is the backpressure boundary and the ledger is the idempotency guard. These two small classes are where most of the correctness lives:

code/lld/notification_service/services.py — queue and ledger
class NotificationQueue:
    """Bounded, due-time ordered, priority-broken ties. The backpressure boundary.

    ``_lock`` guards the heap and the counter. ``put`` refuses past capacity
    rather than growing without limit, because an unbounded queue turns a
    provider outage into an out-of-memory kill.
    """

    def __init__(self, capacity: int = 1000) -> None:
        self._capacity = capacity
        self._lock = threading.Lock()
        self._heap: list[tuple[float, int, int, Notification]] = []
        self._sequence = itertools.count()

    def put(self, notification: Notification) -> None:
        with self._lock:
            if len(self._heap) >= self._capacity:
                raise QueueFullError(f"queue is full at {self._capacity}; shed load or scale workers")
            entry = (
                notification.due_at,
                -notification.priority.rank,
                next(self._sequence),
                notification,
            )
            heapq.heappush(self._heap, entry)

    def take(self, now: float) -> Notification | None:
        """Pop the earliest due notification, or None when nothing is ready yet."""
        with self._lock:
            if not self._heap or self._heap[0][0] > now:
                return None
            return heapq.heappop(self._heap)[3]

    def __len__(self) -> int:
        with self._lock:
            return len(self._heap)


class DeliveryLedger:
    """``(idempotency key, channel)`` may be handed to a provider at most once.

    The queue already gives one notification to one worker; this is the second
    line of defence, for the case a message is redelivered after a crash. It is
    what turns at-least-once transport into effectively-once delivery.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._claims: set[tuple[str, Channel]] = set()

    def claim(self, key: str, channel: Channel) -> bool:
        with self._lock:
            if (key, channel) in self._claims:
                return False
            self._claims.add((key, channel))
            return True

    def release(self, key: str, channel: Channel) -> None:
        """A failed attempt gives the claim back so the retry can proceed."""
        with self._lock:
            self._claims.discard((key, channel))

    def delivered(self) -> int:
        with self._lock:
            return len(self._claims)

Dispatcher.run_once is one worker step. Read it as five decisions in a row: still wanted, already sent, send, recover, record.

code/lld/notification_service/services.py — the dispatcher
class Dispatcher:
    """One worker step: claim, re-check preferences, send, retry, fall back, dead-letter."""

    def __init__(
        self,
        queue: NotificationQueue,
        senders: Mapping[Channel, ChannelSender],
        engine: TemplateEngine,
        preferences: PreferenceStore,
        ledger: DeliveryLedger,
        retry: RetryPolicy | None = None,
        clock: Clock | None = None,
    ) -> None:
        self._queue = queue
        self._senders = dict(senders)
        self._engine = engine
        self._preferences = preferences
        self._ledger = ledger
        self._retry = retry or ExponentialBackoff()
        self._clock = clock or SystemClock()
        self._lock = threading.Lock()
        self._dead_letters: list[Notification] = []

    def sender_for(self, channel: Channel) -> ChannelSender:
        """Null Object: an unconfigured channel is a no-op, never an exception."""
        return self._senders.get(channel) or NullSender(channel)

    def run_once(self) -> Notification | None:
        """Deliver one notification, or return None when nothing is due."""
        notification = self._queue.take(self._clock.now())
        if notification is None:
            return None
        if not self._still_wanted(notification):
            return notification
        notification.claim()
        key = notification.message.idempotency_key
        if not self._ledger.claim(key, notification.channel):
            # Someone already sent this exact message on this channel. Do not send it twice.
            notification.succeed("deduplicated", self._clock.now())
            return notification
        try:
            provider_message_id = self.sender_for(notification.channel).send(notification.message)
        except Exception as exc:  # any provider failure, including an open circuit
            self._ledger.release(key, notification.channel)
            notification.fail(str(exc), self._clock.now())
            self._recover(notification)
        else:
            notification.succeed(provider_message_id, self._clock.now())
        return notification

    def drain(self, limit: int = 10_000) -> list[Notification]:
        """Run until the queue has nothing due. Deterministic; used by the demo and tests."""
        done: list[Notification] = []
        for _ in range(limit):
            notification = self.run_once()
            if notification is None:
                return done
            done.append(notification)
        return done

    def dead_letters(self) -> list[Notification]:
        with self._lock:
            return list(self._dead_letters)

    def _still_wanted(self, notification: Notification) -> bool:
        """Preferences can change while a notification sits in the queue.

        Checking at send time rather than at enqueue time is the difference
        between honouring an opt-out and emailing someone who just left.
        """
        preferences = self._preferences.get(notification.user_id)
        if preferences.allows(notification.event, notification.channel):
            return True
        for channel in notification.fallbacks:
            if preferences.allows(notification.event, channel) and self._engine.has(
                notification.event, channel
            ):
                self._switch(notification, channel, preferences)
                return False
        notification.suppress()
        return False

    def _recover(self, notification: Notification) -> None:
        """Retry the same channel, then fall back, then dead-letter."""
        if notification.attempts < self._retry.max_attempts:
            notification.schedule_retry(self._clock.now() + self._retry.delay(notification.attempts))
            self._queue.put(notification)
            return
        preferences = self._preferences.get(notification.user_id)
        for channel in notification.fallbacks:
            if preferences.allows(notification.event, channel) and self._engine.has(
                notification.event, channel
            ):
                self._switch(notification, channel, preferences)
                return
        notification.dead_letter()
        with self._lock:
            self._dead_letters.append(notification)

    def _switch(self, notification: Notification, channel: Channel, preferences: UserPreferences) -> None:
        """Re-render for the new channel: an SMS body is not an email body."""
        message = self._engine.render(
            self._engine.template_for(notification.event, channel),
            notification.id,
            preferences.addresses[channel],
            notification.payload,
            notification.message.idempotency_key,
        )
        notification.switch_to(channel, message, self._clock.now())
        self._queue.put(notification)

Running python -m lld.notification_service.demo walks every branch:

pipeline: preferences -> dedup -> rate_limit -> template -> enqueue
first: order_shipped -> push (queued) | repeat: order_shipped -> suppressed (duplicate)
push calls=2, email sends=1, breaker=open
delivery: n-1 order_shipped via email: delivered after 1 attempt(s)
attempts: [(1, 'push', False), (2, 'push', False), (1, 'email', True)]
rate limit, capacity 2: order_shipped -> push (queued) | order_shipped -> suppressed (rate_limited)
critical bypasses the limit: order_shipped -> push (queued)
after the opt-out: {'suppressed': 2}
u-2 has push only: n-6 order_shipped via push: dead_letter after 2 attempt(s)
dead letters: ['n-6']
queue depth=0, distinct deliveries=1

Trace the third and fourth lines. Push fails twice, which is max_attempts, so the notification switches to email and is delivered on its first attempt there — and those two failures also tripped the breaker, so the next push notification fails without a network call at all. The last user has only push configured, so the same failures end in the dead letter queue.

Concurrency and edge cases

Which lock protects what. Six small locks, none held across a call into another component:

  1. NotificationQueue._lock guards the heap. take pops under it and returns; put refuses past capacity under it.
  2. DeliveryLedger._lock guards the claim set. Claim, send, release is the sequence, and the claim is what serialises two workers holding the same message.
  3. CircuitBreaker._lock guards the failure count and the state, including the timed OPEN → HALF_OPEN move, which happens lazily inside _refresh rather than on a timer thread.
  4. DedupStore._lock, TokenBucketRateLimiter._lock and PreferenceStore._lock each guard one dict.

The provider call itself happens outside every lock. That matters more here than anywhere else on this page: a network call to an SMS vendor can take a full round trip, and a lock held across it would serialise every worker on the slowest provider.

Two idempotency guards, and they are not the same thing. DedupStore sits in the pipeline and answers "has this caller already asked for this?" over a time window; it stops a retrying caller from creating two notifications. DeliveryLedger sits in the dispatcher and answers "has this exact message already gone to this channel?"; it stops an at-least-once queue from handing the same notification to two workers and delivering twice. Drop either one and you have a real bug: without the store you get two notifications, without the ledger you get one notification delivered twice. The test fires 12 concurrent notify calls with one key (one accepted, eleven suppressed), then drains 30 distinct notifications through four worker threads and asserts 31 sends with 31 distinct idempotency keys.

Failure isolation. A retry alone does not help when a provider is down — it multiplies the load on it. The breaker trips after a threshold of failures and every subsequent call fails in nanoseconds instead of waiting on a socket; a same-datacenter round trip is about 500 µs and a failing provider is far worse, so failing fast is thousands of times cheaper than trying. After the cooldown one probe goes through, and a single failure puts it straight back to OPEN.

Backpressure. The queue is bounded and notify raises QueueFullError rather than growing. A worker pool of 10 threads at roughly 1k operations per second each is about 10k sends/s of headroom; a queue that grows past capacity means the providers are slower than the callers, and the correct answer is to shed load at the door, not to buffer until the process dies.

Mid-flight preference changes. _still_wanted runs at send time, not at enqueue time. If the channel is no longer allowed it tries a fallback; if nothing is allowed it marks the notification SUPPRESSED. The test opts a user out while their notification sits in the queue and asserts it is never sent.

Other edge cases handled: a payload missing a placeholder (RenderError, before anything is enqueued); no template for a fallback channel (that channel is dropped, the rest survive); a channel with no address configured; an unconfigured provider (Null Object); a scheduled send that is not yet due (take returns None); a CRITICAL request walking past the rate limit; a notification claimed twice (NotificationStateError); and jitter that is seeded, so every backoff assertion is an exact number.

Common mistake

Treating "retry" and "idempotency" as the same feature. Retries are what you do when a send fails; idempotency is what saves you when a send succeeded and you did not find out — the provider accepted the SMS and the acknowledgement was lost. Retrying that without a delivery ledger sends it twice. Say the sentence "at-least-once transport plus an idempotent consumer equals effectively-once" and then point at the ledger.

Extensibility and follow-ups

  • Quiet hours: a sixth Stage that pushes send_after to the end of the quiet window instead of dropping. The queue already orders by due time, so nothing else changes.
  • Digests: a stage that appends to a per-user bucket instead of enqueuing, plus a scheduled job that renders the bucket into one notification. The dispatcher never learns digests exist.
  • A/B template tests: TemplateEngine.template_for becomes a strategy that picks a variant by a hash of the user id; the rendered message already carries an id you can join analytics on.
  • Open and click tracking: a TrackingSender decorator that rewrites links before delegating. That is a third decorator on the same interface.
  • Per-channel retry policies: Dispatcher takes one RetryPolicy today; make it a map keyed by channel and give in-app NoRetry.
  • Real workers and a real broker: run_once() becomes the body of a consumer loop against SQS or Kafka. The ledger becomes a table with a unique key on (idempotency_key, channel) — the same design, one process wider. That is where this becomes Design a notification system.

Interview tip

Draw the pipeline as five boxes before you write a single class. It gives the interviewer a shape to hang follow-ups on ("where would quiet hours go?", "where would digests go?"), and every one of those answers is then "a new stage, here" — which is exactly the extensibility signal being graded.

Tests

tests/test_notification_service.py has 14 cases over one wired-up Rig. The fallback test walks the whole recovery ladder in one function:

code/lld/notification_service/tests/test_notification_service.py — retry, fallback, dead letter
def test_retries_then_falls_back_then_dead_letters() -> None:
    """Push fails forever: two attempts, then email; with no email, the dead letter queue."""
    rig = build(push_sender=FlakySender(Channel.PUSH, failures=99, error="APNs 503"))
    result = rig.service.notify("u-1", EVENT, PAYLOAD, idempotency_key="A-1")
    rig.pump()

    notification = result.notification
    assert notification.status is DeliveryStatus.DELIVERED and notification.channel is Channel.EMAIL
    assert [(a.attempt, a.channel, a.ok) for a in notification.history] == [
        (1, Channel.PUSH, False),
        (2, Channel.PUSH, False),
        (1, Channel.EMAIL, True),
    ]
    assert rig.email.count() == 1 and rig.dispatcher.dead_letters() == []

    lonely = build(push_sender=FlakySender(Channel.PUSH, failures=99))
    lonely.preferences.set(UserPreferences("u-2", (Channel.PUSH,), addresses={Channel.PUSH: "d-2"}))
    stranded = lonely.service.notify("u-2", EVENT, PAYLOAD, idempotency_key="B-1")
    lonely.pump()
    assert stranded.notification.status is DeliveryStatus.DEAD_LETTER
    assert [n.id for n in lonely.dispatcher.dead_letters()] == [stranded.notification.id]

The concurrency test covers both idempotency guards at once — many callers with one key, then many workers on one queue:

code/lld/notification_service/tests/test_notification_service.py — idempotency under threads
def test_one_idempotency_key_under_concurrency_produces_exactly_one_send() -> None:
    """Twelve callers race with the same key; twelve workers race to deliver."""
    rig = build(capacity=100, queue_capacity=200)

    def submit(i: int) -> bool:
        return rig.service.notify("u-1", EVENT, PAYLOAD, idempotency_key="A-1").accepted()

    with ThreadPoolExecutor(max_workers=8) as pool:
        accepted = list(pool.map(submit, range(12)))
    assert accepted.count(True) == 1 and accepted.count(False) == 11

    # Now 30 distinct notifications, drained by four worker threads at once.
    for i in range(30):
        rig.service.notify("u-1", EVENT, PAYLOAD, idempotency_key=f"B-{i}")
    with ThreadPoolExecutor(max_workers=4) as pool:
        list(pool.map(lambda _: rig.dispatcher.drain(), range(4)))

    assert rig.push.count() == 31  # 30 distinct plus the single deduplicated one
    assert rig.ledger.delivered() == 31 and len(rig.queue) == 0
    assert len({m.idempotency_key for m in rig.push.sent}) == 31  # nothing sent twice

The rest cover: the happy path with exact rendered text and recipient; each pipeline stage dropping for its own reason via parametrize (muted, duplicate, rate limited, no channel); a payload that fails to render and the Null Object for an unconfigured channel; the breaker walking CLOSED → OPEN → HALF_OPEN → CLOSED with proof the provider was not called while open; preferences changed while queued, both to a fallback and to SUPPRESSED; due-time-then-priority ordering plus QueueFullError; and exact backoff delays from a seeded jitter source. Run them with uv run pytest code/lld/notification_service -q.

45-minute pacing

Minutes What to do What to say or write
0–5 Clarify One channel or all of them? Where does the idempotency key come from? Dropped or delayed when rate limited? Out of scope: real providers, digests, tracking.
5–10 The pipeline Draw five boxes: preferences, dedup, rate limit, template, enqueue. Say why rendering comes last.
10–17 Entities NotificationRequest vs Notification; the status set; UserPreferences.channel_order doubling as the fallback chain.
17–24 Class diagram The delivery half: queue, dispatcher, ChannelSender and the two decorators. Mark the six locks.
24–33 The delivery path Write run_once: still wanted, ledger claim, send, recover. Then _recover: retry, fall back, dead-letter.
33–39 Idempotency and isolation Two guards and why both exist; the breaker's three states and the cooldown probe.
39–43 Tests The recovery ladder test and the 12-callers-one-key race.
43–45 Extensions Quiet hours and digests as new stages; a real broker as the HLD hand-off.