Redis Pub/Sub Bridge

The single-server ceiling

One FastAPI process can hold tens of thousands of WebSocket connections, but it cannot hold a million. To scale horizontally — multiple servers behind a load balancer — you hit a structural problem: connections in the same room may live on different servers. Server A's broadcast cannot reach the user on Server B unless they share a message bus.

Redis Pub/Sub is the standard answer

Each server publishes outgoing messages to a Redis channel and subscribes to that same channel. Redis fans out to all subscribers. Each server then broadcasts the message to its local connections only. This pattern works for thousands of servers and millions of connections, with no central state.

cwkPippa's parallel pattern

cwkPippa is intentionally single-server (Mac Studio at home). But Pippa's Council pattern — where one user's question fans out to multiple brain processes and the responses converge — is the same shape. The transport differs (subprocess pipes, not Redis), but "fan out, fan in" is universal in distributed real-time.

Code

Redis-bridged manager·python

import json
import redis.asyncio as aioredis
from fastapi import FastAPI, WebSocket
from typing import Dict, Set
import asyncio

class RedisManager:
    def __init__(self, redis_url: str):
        self.redis = aioredis.from_url(redis_url)
        self.pubsub = self.redis.pubsub()
        self.local: Dict[str, Set[WebSocket]] = {}
        self._task: asyncio.Task | None = None

    async def start(self):
        await self.pubsub.psubscribe('ws:room:*')
        self._task = asyncio.create_task(self._listen())

    async def stop(self):
        if self._task:
            self._task.cancel()
        await self.pubsub.aclose()
        await self.redis.aclose()

    async def join(self, ws: WebSocket, room: str):
        await ws.accept()
        self.local.setdefault(room, set()).add(ws)

    def leave(self, ws: WebSocket, room: str):
        members = self.local.get(room)
        if members:
            members.discard(ws)
            if not members:
                self.local.pop(room, None)

    async def publish(self, room: str, message: dict):
        '''All servers (including this one) will fan it out locally.'''
        await self.redis.publish(f'ws:room:{room}', json.dumps(message))

    async def _listen(self):
        async for msg in self.pubsub.listen():
            if msg.get('type') != 'pmessage':
                continue
            channel = msg['channel'].decode()
            room = channel.split(':', 2)[2]
            data = json.loads(msg['data'])
            for ws in list(self.local.get(room, ())):
                try:
                    await ws.send_json(data)
                except Exception:
                    self.leave(ws, room)

Topology·text

  Server A              Redis              Server B
  (A, B in #general)   pub/sub            (C, D in #general)
        |                |                       |
  user A sends "hi"      |                       |
        |--- publish --->|                       |
        |                |--- pmessage --------->|
        |                |                       v
        |                |                C and D get "hi"
        v                |
  A and B get "hi"
  (via local fan-out)

The single-server ceiling

Redis Pub/Sub is the standard answer

cwkPippa's parallel pattern

Code

External links

Exercise

Progress

Comments 0