Build an event broadcast system with Redis pub/sub

Introduction

When you scale a real-time collaboration backend beyond a single process, every WebSocket connection lives on exactly one FastAPI instance — and an in-memory event bus inside that process can no longer reach the collaborators connected elsewhere. A user typing on Instance A never triggers an indicator on Instance B, presence updates go stale, and "live" features silently degrade the moment you add a second replica. By the end of this lesson you'll be able to design a Redis pub/sub-based event broadcast system that fans out typed events — new messages, typing indicators, presence updates, and LLM stream chunks — across multiple FastAPI instances to all subscribed WebSocket clients in a shared conversation session, applying echo suppression and per-subscription filters so each client receives exactly the events it cares about.

Key Terminology

• Broadcast event: A typed payload (e.g., NEW_MESSAGE, TYPING_INDICATOR, PRESENCE_UPDATE) carrying session_id, sender_id, and a timestamp, published once and delivered to every subscribed instance.
• Session channel: A Redis pub/sub channel namespaced by session ID (e.g., broadcast:session:abc123) that scopes event fan-out to the participants of one shared conversation.
• Echo suppression: The filter that drops events whose sender_id matches the local subscriber's user ID, preventing a user from receiving their own actions back through the broadcast loop.

Concepts

Why Redis Pub/Sub Over Streams or Polling

Before diving into implementation, understand why Redis pub/sub fits this use case better than alternatives. Redis Streams provide durability and consumer groups—features you need for message queuing but not for ephemeral real-time events. A typing indicator that arrives 30 seconds late is worse than one that never arrives. Pub/sub delivers fire-and-forget semantics with sub-millisecond latency within a Redis instance. The tradeoff is clear: if an instance is temporarily disconnected, it misses events published during that window. For presence updates, typing indicators, and new message notifications, this is acceptable because the next event naturally corrects state. Contrast this with the conversation message history itself, which uses vector clocks and Redis-backed persistence (covered in another goal) precisely because durability matters there. The event broadcast system complements that durable layer by handling the transient notification path.

Integration with Presence and Conversation Systems

The event broadcaster is not a standalone system—it is the distribution layer that the presence tracker and the concurrent message handler publish through. When the presence tracker detects a TTL expiration in the Redis sorted set, it publishes a PRESENCE_UPDATE event via the broadcaster. When the conversation state lock is released after an LLM response completes, the concurrent handler publishes an LLM_STREAM_END event. The typing indicator system from another goal publishes TYPING_INDICATOR events at a throttled rate (no more than once per second per user) to avoid flooding the pub/sub channel.

Failure Modes and Resilience

Redis pub/sub has no built-in retry or message persistence. If the Redis connection drops, the _listener_loop raises a ConnectionError that you must handle by reconnecting and re-subscribing to all active channels. A production-grade broadcaster wraps the listener loop in a retry decorator that exponentially backs off on connection failures, re-issues SUBSCRIBE commands for every channel in _active_channels, and logs a warning so the monitoring system can alert on prolonged disconnections. During the reconnection window, events are lost—but because the presence system uses Redis sorted sets with TTL and conversation state uses vector clocks, the system self-heals: the next presence heartbeat or message sync corrects any stale state caused by missed events. This eventual consistency model is a deliberate architectural choice that trades rare, transient inconsistency for the low latency and simplicity of pub/sub over heavier alternatives like Redis Streams with consumer groups.

Code Walkthrough

The Typed Event Model

Every event flowing through the broadcast system carries a type discriminator that subscribers use for filtering. Without typed events, every subscriber processes every event and discards irrelevant ones client-side—wasting bandwidth and CPU on JSON parsing. The EventType enum and BroadcastEvent data class establish a contract between publishers and subscribers that the rest of the system depends on.

The following code defines the EventType enum with members covering all real-time event categories in the collaboration system, and the BroadcastEvent dataclass that wraps every event payload with metadata including event_type, session_id, sender_id, and a Unix timestamp. The to_json and from_json methods handle serialization for Redis transport, ensuring that enum values survive the round trip through string-based pub/sub channels.

1import enum
2import json
3import time
4from dataclasses import dataclass, field, asdict
5from typing import Any, Optional
6
7class EventType(str, enum.Enum):
8    NEW_MESSAGE = "new_message"
9    TYPING_INDICATOR = "typing_indicator"
10    PRESENCE_UPDATE = "presence_update"
11    SESSION_JOIN = "session_join"
12    SESSION_LEAVE = "session_leave"
13    LLM_STREAM_CHUNK = "llm_stream_chunk"
14    LLM_STREAM_END = "llm_stream_end"
15    CONFLICT_RESOLVED = "conflict_resolved"
16
17@dataclass
18class BroadcastEvent:
19    event_type: EventType
20    session_id: str
21    sender_id: str
22    payload: dict[str, Any] = field(default_factory=dict)
23    timestamp: float = field(default_factory=time.time)
24    origin_instance: Optional[str] = None
25
26    def to_json(self) -> str:
27        data = asdict(self)
28        data["event_type"] = self.event_type.value
29        return json.dumps(data)
30
31    @classmethod
32    def from_json(cls, raw: str) -> "BroadcastEvent":
33        data = json.loads(raw)
34        data["event_type"] = EventType(data["event_type"])
35        return cls(**data)

Multi-Instance Event Flow

Before examining the broadcaster implementation, visualize how events traverse from one user's action to another user's WebSocket across different server instances. The Redis pub/sub layer sits between the publishing instance and all subscribing instances, including the publisher itself.

This sequence diagram traces a cross-instance message delivery path where User A's WebSocket payload flows through FastAPI Instance 1, gets persisted with a vector clock for conflict-free ordering, then fans out via PUBLISH session:{id} on Redis Pub/Sub. Both local and remote subscribers receive the event_json, but each instance applies filtering logic—Instance 1 skips the sender to avoid echo, while Instance 2 matches the EventType and routes the new_message event to User B's WebSocket. This pub/sub fan-out pattern ensures horizontal scaling without sticky sessions.

Loading diagram...

Instance 1 publishes the event after processing User A's message. Redis delivers the event to all subscribers on the session:{id} channel, including Instance 1 itself. Instance 1's local filter skips re-delivery to User A (the sender), while Instance 2's filter matches User B's subscription and forwards the event over their WebSocket. This echo-suppression is handled by comparing sender_id against each subscriber's user ID.

The EventBroadcaster Service

The core service manages Redis pub/sub connections, maintains local subscription registries, and routes incoming events through per-subscriber filter functions. The EventBroadcaster class uses a dedicated redis.asyncio pub/sub connection separate from the main Redis connection pool, because Redis pub/sub connections enter a special mode where they can only execute SUBSCRIBE, UNSUBSCRIBE, and PING commands.

The following implementation defines the EventBroadcaster class with publish_event() for sending events, subscribe() for registering local WebSocket handlers with optional EventType filters, unsubscribe() for cleanup, and a _listener_loop() that runs as a background asyncio.Task dispatching incoming pub/sub messages to matching local subscribers. The SubscriptionFilter callable type alias defines the per-subscriber filtering contract.

1import asyncio
2import logging
3import uuid
4from collections import defaultdict
5from typing import Callable, Awaitable
6
7import redis.asyncio as aioredis
8
9from .events import BroadcastEvent, EventType
10
11logger = logging.getLogger(__name__)
12SubscriptionFilter = Callable[[BroadcastEvent], bool]
13EventHandler = Callable[[BroadcastEvent], Awaitable[None]]
14
15class EventBroadcaster:
16    def __init__(self, redis_url: str, instance_id: str | None = None):
17        self._redis = aioredis.from_url(redis_url)
18        self._pubsub = self._redis.pubsub()
19        self._instance_id = instance_id or uuid.uuid4().hex[:12]
20        self._subscribers: dict[str, dict[str, tuple[EventHandler, SubscriptionFilter | None]]] = defaultdict(dict)
21        self._active_channels: set[str] = set()
22        self._listener_task: asyncio.Task | None = None
23
24    def _channel_key(self, session_id: str) -> str:
25        return f"broadcast:session:{session_id}"
26
27    async def start(self) -> None:
28        self._listener_task = asyncio.create_task(self._listener_loop())
29        logger.info("EventBroadcaster started on instance %s", self._instance_id)
30
31    async def publish_event(self, event: BroadcastEvent) -> int:
32        event.origin_instance = self._instance_id
33        channel = self._channel_key(event.session_id)
34        receivers = await self._redis.publish(channel, event.to_json())
35        logger.debug("Published %s to %s (%d receivers)", event.event_type.value, channel, receivers)
36        return receivers
37
38    async def subscribe(
39        self, session_id: str, handler: EventHandler,
40        event_filter: SubscriptionFilter | None = None,
41    ) -> str:
42        sub_id = uuid.uuid4().hex[:16]
43        channel = self._channel_key(session_id)
44        self._subscribers[channel][sub_id] = (handler, event_filter)
45        if channel not in self._active_channels:
46            await self._pubsub.subscribe(channel)
47            self._active_channels.add(channel)
48            logger.info("Subscribed to Redis channel %s", channel)
49        return sub_id
50
51    async def unsubscribe(self, session_id: str, sub_id: str) -> None:
52        channel = self._channel_key(session_id)
53        self._subscribers[channel].pop(sub_id, None)
54        if not self._subscribers[channel]:
55            await self._pubsub.unsubscribe(channel)
56            self._active_channels.discard(channel)
57            del self._subscribers[channel]
58
59    async def _listener_loop(self) -> None:
60        try:
61            async for message in self._pubsub.listen():
62                if message["type"] != "message":
63                    continue
64                try:
65                    event = BroadcastEvent.from_json(message["data"])
66                except (json.JSONDecodeError, KeyError, ValueError) as exc:
67                    logger.warning("Malformed broadcast event: %s", exc)
68                    continue
69                channel = message["channel"]
70                if isinstance(channel, bytes):
71                    channel = channel.decode()
72                subs = list(self._subscribers.get(channel, {}).items())
73                for sub_id, (handler, evt_filter) in subs:
74                    if evt_filter is not None and not evt_filter(event):
75                        continue
76                    try:
77                        await handler(event)
78                    except Exception:
79                        logger.exception("Handler %s failed for event %s", sub_id, event.event_type.value)
80        except asyncio.CancelledError:
81            logger.info("Listener loop cancelled, cleaning up")
82        finally:
83            await self._pubsub.unsubscribe()
84
85    async def shutdown(self) -> None:
86        if self._listener_task:
87            self._listener_task.cancel()
88            await asyncio.gather(self._listener_task, return_exceptions=True)
89        await self._pubsub.close()
90        await self._redis.close()

Per-Subscription Filtering in Practice

The filter callable registered alongside each subscribe() is where per-subscriber routing logic lives. When a WebSocket connection joins a session, the connection manager constructs a closure that captures the joining user's user_id and the set of EventType values that client cares about, then passes it as event_filter. Inside _listener_loop, every incoming event is run through that closure: it returns False when event.sender_id == user_id (echo suppression — the sender already has the event locally) or when event.event_type is not in the accepted set (the client doesn't care about this category), and True otherwise. Only events that pass the filter are dispatched to the matching route_to_websocket handler, which fetches the live WebSocket from the connection manager and forwards event_type, payload, sender_id, and timestamp as a JSON frame; if the user disconnected between filter and dispatch, the handler returns silently. The sub_id returned by subscribe() is stored against the WebSocket session so that the connection manager can call unsubscribe() during disconnect cleanup, releasing the Redis subscription when the last local listener for a channel goes away.

You'll know the broadcaster works when you connect two clients to the same session_id on different FastAPI instances, send a message from Client A, and Client B receives a new_message frame within sub-second latency while Client A does NOT receive its own event back.

Do's and Don'ts

Do's

1. ✓Do declare EventType with dual inheritance from both str and enum.Enum — this makes each member a native string at runtime, so to_json's data["event_type"] = self.event_type.value passes cleanly through json.dumps, and from_json can reconstruct the enum with EventType(data["event_type"]) on the subscriber side. A plain enum.Enum without str raises a TypeError in json.dumps because enum members are not natively JSON-serializable, breaking the Redis round-trip entirely.
2. ✓Do populate origin_instance on every BroadcastEvent and check it before routing to local WebSocket connections — Redis pub/sub delivers a published message back to the publishing instance's own subscriber; filtering on origin_instance is what suppresses that echo and prevents the sender from receiving a duplicate of their own event. Omitting this field means every publish results in an extra delivery to the originating client with no way to detect it.
3. ✓Do namespace Redis pub/sub channels by session_id (e.g., session:{session_id}) — scoping each instance's subscription to its active conversation sessions limits deserialization and routing work to events that instance actually needs to handle, rather than consuming the entire cluster's event stream. A flat global channel forces every replica to parse every LLM_STREAM_CHUNK and TYPING_INDICATOR from every session cluster-wide.

Don'ts

1. ✗Don't skip the EventType enum reconstruction in from_json — after json.loads, data["event_type"] is a plain str; calling cls(**data) without first converting it via EventType(data["event_type"]) leaves BroadcastEvent.event_type holding a string, silently breaking every subscriber filter that compares event.event_type == EventType.TYPING_INDICATOR using enum equality, since "typing_indicator" == EventType.TYPING_INDICATOR evaluates to True for a str-enum but the type contract is violated for any code doing isinstance or pattern-matching.
2. ✗Don't defer all per-subscription filtering to the WebSocket send layer — if every FastAPI instance fans out all eight EventType values to every local subscriber and discards irrelevant ones only after full JSON deserialization, high-frequency events like LLM_STREAM_CHUNK impose unnecessary parse and dispatch overhead on every replica for every session; apply the EventType discriminator at the delivery routing step, before any WebSocket send is attempted, so subscribers receive only the event categories they registered for.
3. ✗Don't publish all session events to a single shared Redis channel — collapsing all conversation traffic into one channel forces every FastAPI instance to receive, deserialize, and discard events for sessions it has no active WebSocket connections for, coupling each replica's CPU load to cluster-wide traffic volume instead of its own active sessions; use the session:{session_id} channel shape shown in the sequence diagram so each instance subscribes only to the sessions it hosts.