Use Redis pub/sub for real-time event broadcasting

Introduction

When you run multiple FastAPI pods behind a load balancer, mutating shared state on one pod — registering a new model, invalidating a cache key, flipping a routing weight — means every other pod must learn about it within milliseconds. Wiring each pod to call every other pod's internal endpoint couples the publisher to the topology and breaks as the cluster scales. Redis pub/sub gives you a server-side fan-out primitive that broadcasts an event to every subscribed pod in a single round-trip, with no coupling between sender and receivers.

By the end of this lesson, you will build a RedisEventBus class, wire it into FastAPI WebSocket handlers for cross-pod token streaming, and know when at-most-once delivery is the right contract versus when Redis Streams is the safer fallback.

Key Terminology

• pub/sub — Redis primitive where PUBLISH fans a message to every currently-connected SUBSCRIBEr in one round-trip; matters because it is the cheapest way to broadcast cache invalidations and token streams across pods.
• at-most-once delivery — semantics where a subscriber disconnected at publish time misses the message permanently with no offset, ack, or replay; matters because it forces the design rule that no event whose loss costs money or correctness can ride pub/sub.
• thundering herd — failure mode where every pod reconnects in the same millisecond after a Sentinel failover and overwhelms the new primary; matters because exponential backoff with jitter is the only thing that prevents a recoverable failover from becoming a full outage.
• Redis Streams — append-only log primitive (XADD / XREADGROUP / XACK) that gives at-least-once delivery and replay; matters because it is the fallback when subscribers will be temporarily disconnected and event loss is unacceptable.
• back-pressure boundary — bounded asyncio.Queue per WebSocket that drops events when full instead of blocking the listener; matters because one wedged browser must not stall the cross-pod event loop.

Concepts

At-most-once semantics and the fallback for disconnected subscribers

Pub/sub does not buffer. A subscriber that drops for two seconds during a deploy misses every event published in those two seconds, permanently — no offset, no ack, no replay. This is at-most-once delivery, and for cache invalidation it is exactly right: a missed invalidate just means the next request refills from the source of truth. For events you cannot afford to lose — billing usage, deploy notifications, audit trails — at-most-once is the wrong contract.

The fallback when subscribers will be temporarily disconnected is Redis Streams. XADD events:model-registry * model_id m-42 appends an entry to a log; XREAD COUNT 100 STREAMS events:model-registry $ reads new entries. With consumer groups (XREADGROUP GROUP api consumer-1 ...), each entry is delivered to exactly one consumer in the group, with explicit XACK for at-least-once semantics. A subscriber that was offline can resume from its last acknowledged ID and replay everything it missed. Streams cap memory with XADD ... MAXLEN ~ 100000 so the log does not grow unbounded.

The rule of thumb: if losing an event would cause a user-visible bug or a billing miscount, you want Streams. If losing one means the next request is 50ms slower, you want pub/sub.

Production Considerations

• Treating PUBLISH return 0 as success. It tells you nobody listened. Wire it to a metric so you notice when an entire subscriber fleet has died.
• Sharing one Redis connection between publish and subscribe. A subscribed connection is read-only; you will get ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context.
• Subscribing inside a request handler. Subscriptions are pod-lifetime, not request-lifetime. Set them up in the startup hook once and let the listener task own them.
• Using pub/sub for anything that needs replay. The day you ship "show me the last 50 deploys" you will discover the cost of choosing pub/sub over Streams.
• Letting the listener task die silently. Wrap _listen in a supervisor that logs and restarts on exception, otherwise a single bad payload kills cross-pod events forever.
• Unbounded per-connection queues. A wedged WebSocket client should drop messages, not consume infinite memory. Always pass maxsize and handle QueueFull.
• Forgetting jitter on reconnect. Synchronous reconnect storms are how a Sentinel failover becomes a full outage.
• Publishing huge payloads. Pub/sub messages traverse Redis memory; a 5 MB JSON blob multiplied by 1,000 subscribers is 5 GB of bandwidth. Publish IDs and let consumers fetch the body from storage.

Code Walkthrough

Now that you understand at-most-once delivery and when Redis Streams is the safer fallback, the implementation below builds on those constraints directly. The RedisEventBus class encapsulates two responsibilities: serialising Python events to a JSON wire format on publish, and dispatching incoming messages to in-process listeners on receive. Keeping both paths in one class means there is exactly one place where the event schema is defined.

Code snippetpython
1import asyncio
2import json
3import logging
4from collections import defaultdict
5from typing import Any, Awaitable, Callable
6
7import redis.asyncio as aioredis
8
9logger = logging.getLogger(__name__)
10Handler = Callable[[dict[str, Any]], Awaitable[None]]
11
12class RedisEventBus:
13    """Async pub/sub event bus for cross-pod broadcast."""
14
15    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
16        self._redis = aioredis.from_url(redis_url, decode_responses=True)
17        self._pubsub = self._redis.pubsub()
18        self._handlers: dict[str, list[Handler]] = defaultdict(list)
19        self._listener_task: asyncio.Task | None = None
20
21    async def publish(self, channel: str, event: dict[str, Any]) -> int:
22        payload = json.dumps(event, default=str)
23        receivers = await self._redis.publish(channel, payload)
24        if receivers == 0:
25            logger.warning("event published with no subscribers channel=%s", channel)
26        return receivers
27
28    def on(self, channel: str, handler: Handler) -> None:
29        self._handlers[channel].append(handler)
30
31    async def start(self) -> None:
32        if not self._handlers:
33            return
34        await self._pubsub.subscribe(*self._handlers.keys())
35        self._listener_task = asyncio.create_task(self._listen())
36
37    async def _listen(self) -> None:
38        async for message in self._pubsub.listen():
39            if message["type"] != "message":
40                continue
41            try:
42                event = json.loads(message["data"])
43            except json.JSONDecodeError:
44                logger.exception("invalid JSON on channel %s", message["channel"])
45                continue
46            for handler in self._handlers[message["channel"]]:
47                asyncio.create_task(handler(event))

The async client (redis.asyncio) is mandatory inside FastAPI — a synchronous subscribe call would block the event loop. One connection drives publishes while a separate pubsub() connection drives subscriptions; mixing them on the same connection breaks because a subscribed connection cannot issue normal commands. publish JSON-encodes with default=str to tolerate datetime and UUID values, and logs a warning when PUBLISH returns 0 — the signal that nobody was listening. The _listen loop dispatches each decoded payload via create_task so a slow handler cannot delay the next incoming message.

The WebSocket layer wires this bus to connected browsers using a bounded asyncio.Queue per connection, enforcing the back-pressure boundary described in the Concepts section:

Code snippetpython
1from fastapi import FastAPI, WebSocket, WebSocketDisconnect
2
3app = FastAPI()
4bus = RedisEventBus()
5
6async def _enqueue(event: dict[str, Any], queue: asyncio.Queue) -> None:
7    try:
8        queue.put_nowait(event)
9    except asyncio.QueueFull:
10        pass  # slow client: drop rather than block the shared listener
11
12@app.websocket("/ws/model-events")
13async def model_events(websocket: WebSocket) -> None:
14    await websocket.accept()
15    queue: asyncio.Queue = asyncio.Queue(maxsize=64)
16    bus.on("events:model-registry", lambda e: _enqueue(e, queue))
17    try:
18        while True:
19            event = await queue.get()
20            await websocket.send_json(event)
21    except WebSocketDisconnect:
22        pass

Each WebSocket connection owns a queue capped at 64 entries. When the queue is full, put_nowait raises QueueFull and the event is silently dropped for that subscriber — one wedged browser cannot stall the listener task or block delivery to every other connection. This is the same pattern used to fan out streaming LLM token deltas: the worker publishing each delta to events:chat:<session_id> never needs to know which pods have active WebSocket connections, and at-most-once delivery is acceptable because a missed token just produces a visible gap rather than a correctness failure.

Confirm that publishing to events:model-registry from one FastAPI process delivers the JSON event to a WebSocket client connected through a separate process, and that closing that client's connection does not interrupt delivery to any remaining subscribers.

In your domain

Building on the walkthrough above, this section applies the lesson to your discipline.

The RedisEventBus shape generalises: a publisher process emits a JSON event on a channel and every subscribed pod reacts in-process. The concrete worked example below shows where this fan-out lands in your day-to-day stack — what gets published, who subscribes, and where at-most-once delivery is acceptable versus where you must promote the event to Redis Streams.

Do's and Don'ts

Do's

1. ✓Do use redis.asyncio instead of the synchronous redis client inside FastAPI — a synchronous subscribe or publish call blocks the event loop, stalling every concurrent WebSocket connection and in-flight request on that pod until the call returns.
2. ✓Do dispatch each decoded event via asyncio.create_task(handler(event)) inside _listen — awaiting the handler inline means one slow WebSocket send or downstream write delays every subsequent message arriving on every channel for the entire pod.
3. ✓Do pass default=str to json.dumps when serialising payloads in publish — without it, a datetime or UUID field raises TypeError at publish time and silently drops the event before it reaches any subscriber.

Don'ts

1. ✗Don't reuse the pubsub() connection for PUBLISH calls — a connection in subscribe mode cannot issue normal Redis commands; mixing them causes the publish to fail because redis.asyncio treats subscribed connections as read-only command pipes.
2. ✗Don't call bus.on() after bus.start() — start() subscribes only the channels already present in _handlers at call time via self._pubsub.subscribe(*self._handlers.keys()); a handler registered afterward receives no messages because its channel is never passed to Redis.
3. ✗Don't use await queue.put(event) inside _enqueue for slow WebSocket clients — put blocks until the bounded queue has space, which means one wedged browser stalls the shared _listen task and delays delivery to every other connected client on that pod.