Build real-time notification system with Redis pub/sub

Introduction

When you scale FastAPI to four replicas behind a load balancer, your single-process WebSocket fan-out quietly breaks: a notification published on replica A has no way to reach a socket pinned to replica B, so the message is silently dropped and the user never hears about the payment that just confirmed. Redis pub/sub is the smallest cross-pod bus that fixes this without dragging in Kafka — each replica subscribes only to the topics it currently serves, and any service that can talk to Redis can publish.

By the end you'll be able to wire a ConnectionManager to a Redis pub/sub reader task, scope subscriptions per authenticated user, publish notifications from any service, and reason about the at-most-once semantics so you don't accidentally promise durable delivery on a fire-and-forget bus.

Key Terminology

• Pub/sub channel — a named Redis topic (e.g. notify:user:42) that publishers write to and subscribers listen on; channel matching is exact unless you use PSUBSCRIBE, which we avoid because pattern fan-out is harder to ACL-scope.
• Lazy subscription — subscribing to a user's channel only when their first socket connects to this replica and unsubscribing on their last disconnect; keeps Redis subscription counts proportional to real fan-out load instead of accumulating dead listeners.
• At-most-once delivery — the guarantee Redis pub/sub provides: if a subscriber is restarting or wedged, in-flight messages on its channels are lost; there is no inbox waiting for a reconnect.
• Fan-out — the act of delivering one published message to every interested subscriber; in this design fan-out happens twice — Redis fans out to replicas, and each replica fans out locally to the sockets it owns.
• Backpressure — the implicit queue between the reader loop and a slow WebSocket's send buffer; left unbounded it turns one stalled client into a memory leak that takes the whole replica down.

Concepts

The cross-replica fan-out problem

Picture four FastAPI pods behind a load balancer. A user's browser is pinned to replica B; two minutes later a billing worker on replica A wants to push payment_confirmed to that user. Replica A has no local socket for the user, so without a shared bus the notification is lost. The fix is a thin indirection — replicas don't push to each other, they publish to Redis, and every replica subscribes to the topics it cares about (see Code Walkthrough).

Topic conventions

Keep the namespace flat and predictable so ACLs, dashboards, and grep all stay simple: notify:user:{user_id} for direct messages across a user's tabs and devices, notify:org:{org_id} for organization-wide broadcasts, notify:room:{room_id} for chat-scoped fan-out. Avoid nesting like notify.user.42.activity.feed — exact-match channels are cheaper than pattern subscriptions and easier to scope with Redis ACLs.

Lazy subscribe, lazy unsubscribe

Subscribe per user, not per socket. Two tabs from the same user equals one Redis subscription with two local sockets; otherwise you double-deliver. Check first = not self._sockets[user_id] before calling subscribe, and symmetrically unsubscribe when the last socket goes away. A long-running replica that subscribes greedily accumulates dead channels and pays for fan-out it no longer needs.

Auth scoping at connect time

A socket must only register interest in topics the bearer owns. Validate the JWT during the WebSocket handshake and refuse to subscribe to a channel outside the claimant's scope — the subject claim must match user_id for notify:user:*, and a membership claim must cover org_id for notify:org:*. Doing this check before the first _pubsub.subscribe() means a forged socket can't even register interest in a topic, let alone receive from it.

At-most-once and backpressure

Redis pub/sub is fire-and-forget. If a replica is mid-restart or its reader loop is wedged for a few seconds, messages on its subscribed channels are gone. That is fine for transient notifications ("model warmed up", "typing…") and wrong for anything you'd have to explain to a customer ("why didn't I get the invoice email?") — for durable fan-out, move to Redis Streams with consumer groups. The other implicit queue is between the reader and a slow socket's send buffer; bound every send with asyncio.wait_for so one stalled client can't pile up messages forever.

Code Walkthrough

The snippets below demonstrate lazy per-user subscription, the cross-replica reader loop with bounded local fan-out, and a thin REST publisher that any service can call.

Code snippetpython
1import asyncio
2from collections import defaultdict
3from typing import DefaultDict
4import redis.asyncio as redis
5from fastapi import WebSocket
6
7class ConnectionManager:
8    def __init__(self, redis_url: str):
9        self._sockets: DefaultDict[str, set[WebSocket]] = defaultdict(set)
10        self._redis = redis.from_url(redis_url, decode_responses=True)
11        self._pubsub = self._redis.pubsub()
12        self._reader_task: asyncio.Task | None = None
13        self._lock = asyncio.Lock()
14
15    async def start(self) -> None:
16        self._reader_task = asyncio.create_task(self._reader())
17
18    async def connect(self, user_id: str, ws: WebSocket) -> None:
19        await ws.accept()
20        async with self._lock:
21            first = not self._sockets[user_id]
22            self._sockets[user_id].add(ws)
23            if first:
24                await self._pubsub.subscribe(f"notify:user:{user_id}")  # lazy
25
26    async def disconnect(self, user_id: str, ws: WebSocket) -> None:
27        async with self._lock:
28            self._sockets[user_id].discard(ws)
29            if not self._sockets[user_id]:
30                del self._sockets[user_id]
31                await self._pubsub.unsubscribe(f"notify:user:{user_id}")
32
33    async def _reader(self) -> None:
34        async for msg in self._pubsub.listen():
35            if msg["type"] != "message":
36                continue  # skip subscribe/unsubscribe confirmations
37            user_id = msg["channel"].split(":")[-1]
38            await self._fanout_local(user_id, msg["data"])
39
40    async def _fanout_local(self, user_id: str, payload: str) -> None:
41        sockets = list(self._sockets.get(user_id, ()))  # snapshot outside lock
42        for ws in sockets:
43            try:
44                await asyncio.wait_for(ws.send_text(payload), timeout=2.0)
45            except Exception:
46                await self._force_close(user_id, ws)
47
48    async def _force_close(self, user_id: str, ws: WebSocket) -> None:
49        try:
50            await ws.close(code=1011)
51        finally:
52            async with self._lock:
53                self._sockets[user_id].discard(ws)

Code snippetpython
1from fastapi import APIRouter, HTTPException
2from pydantic import BaseModel
3import redis.asyncio as redis
4
5router = APIRouter()
6_pub = redis.from_url(REDIS_URL, decode_responses=True)
7
8class NotifyIn(BaseModel):
9    user_id: str
10    type: str
11    payload: dict
12
13@router.post("/notify", status_code=202)
14async def notify(body: NotifyIn) -> dict:
15    if not body.user_id.isalnum():
16        raise HTTPException(400, "bad user_id")  # block colons/wildcards
17    channel = f"notify:user:{body.user_id}"
18    delivered = await _pub.publish(channel, body.json())
19    return {"channel": channel, "delivered_to": delivered}

You'll know it works when: two browser tabs for the same user, served by different replicas, both receive a single POST /notify payload within a second; delivered_to reflects the number of replicas with active subscriptions (not sockets); and killing one replica mid-flight does not crash the publisher — the message is simply lost on that replica, exactly as at-most-once predicts.

Do's and Don'ts

Do's

1. ✓Do subscribe lazily per user — check first = not self._sockets[user_id] before subscribing and unsubscribe on the last disconnect, so Redis subscription counts track real fan-out load instead of leaking forever.
2. ✓Do validate JWT scope before the first subscribe — a user with a valid token for user:42 must never be able to register interest in notify:user:43; auth at connect time is the only defense.
3. ✓Do bound every send_text with asyncio.wait_for — two seconds is generous for a WebSocket; anything longer means the client is broken and you should drop it, not buffer for it.

Don'ts

1. ✗Don't use PSUBSCRIBE notify:user:* "for simplicity" — every replica then receives every notification and your Redis CPU becomes the bottleneck before you hit a thousand users.
2. ✗Don't treat publish() returning 0 as a bug — it just means no replica was subscribed at that instant, which is exactly what at-most-once means; if you need delivery guarantees, switch to Redis Streams.
3. ✗Don't hold the manager lock across await ws.send_text() — snapshot the socket set inside the lock and send outside, or one slow client will block every connect and disconnect on the replica.