Building Resilient Real-Time Systems: WebSocket Backpressure and Connection Recovery

The Problem with Naive WebSocket Implementations

When I first deployed a system handling 300+ concurrent WebSocket connections, I made the assumption that as long as my FastAPI server could accept connections and my database could handle writes, everything would be fine. I was wrong.

The issue surfaced during a peak load test. Message throughput looked good on paper—sub-50ms sync times—but clients were silently dropping. Not crashing, not getting error messages. Just... gone. The server kept accepting new connections while old ones festered in half-closed states, consuming memory and file descriptors.

The root cause wasn't bandwidth or compute. It was backpressure. When a client's network buffer filled up faster than it could drain, the server had no way to know. It kept writing to a socket that wasn't actually delivering data. Meanwhile, the client's receive loop was blocked on a full buffer, unable to process new messages or send heartbeats. Deadlock.

Understanding Backpressure

Backpressure is what happens when a producer generates data faster than a consumer can accept it. In HTTP, this is mostly invisible—the request completes and the connection closes. In WebSockets, the connection is long-lived, so backpressure accumulates.

Here's the mechanics:

1. Server writes a message to a client's socket.
2. The OS puts that data in the socket's send buffer.
3. The client's network stack receives it and puts it in the client's receive buffer.
4. The client's application code reads from that buffer.
5. If step 4 is slow or blocked, the receive buffer fills up.
6. The OS tells the server's socket that it can't accept more data right now.
7. The server's write call blocks or returns a partial write.

If the server doesn't handle this—if it just keeps calling send() in a loop—it will block the entire event loop, preventing it from handling other clients or accepting new connections.

Implementing Backpressure Handling in FastAPI

FastAPI's WebSocket implementation uses asyncio, which gives us tools to detect and respond to backpressure. The key is to check the write buffer before sending, and to implement a queue with a maximum size.

import asyncio
from fastapi import FastAPI, WebSocket
from collections import deque
import json

app = FastAPI()

class ConnectionManager:
    def __init__(self, max_queue_size: int = 100):
        self.active_connections: dict[str, WebSocket] = {}
        self.send_queues: dict[str, asyncio.Queue] = {}
        self.max_queue_size = max_queue_size

    async def connect(self, client_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[client_id] = websocket
        self.send_queues[client_id] = asyncio.Queue(maxsize=self.max_queue_size)
        # Start the sender task for this client
        asyncio.create_task(self._sender(client_id))

    async def disconnect(self, client_id: str):
        if client_id in self.active_connections:
            del self.active_connections[client_id]
        if client_id in self.send_queues:
            del self.send_queues[client_id]

    async def _sender(self, client_id: str):
        """Dedicated task that drains the send queue for a client."""
        websocket = self.active_connections[client_id]
        queue = self.send_queues[client_id]
        
        try:
            while True:
                message = await queue.get()
                await websocket.send_json(message)
        except Exception as e:
            print(f"Sender error for {client_id}: {e}")
            await self.disconnect(client_id)

    async def broadcast(self, message: dict):
        """Broadcast to all clients, dropping slow ones."""
        dead_clients = []
        
        for client_id, queue in self.send_queues.items():
            try:
                # Non-blocking put with timeout
                queue.put_nowait(message)
            except asyncio.QueueFull:
                # Client is too slow; mark for disconnection
                print(f"Queue full for {client_id}, dropping connection")
                dead_clients.append(client_id)
        
        # Clean up slow clients
        for client_id in dead_clients:
            try:
                await self.active_connections[client_id].close(code=1008)  # Policy violation
            except:
                pass
            await self.disconnect(client_id)

manager = ConnectionManager(max_queue_size=50)

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(client_id, websocket)
    try:
        while True:
            data = await websocket.receive_json()
            # Process incoming data
            await manager.broadcast({"from": client_id, "data": data})
    except Exception as e:
        print(f"Connection error: {e}")
    finally:
        await manager.disconnect(client_id)

This pattern separates reading from writing. Each client has its own send queue and a dedicated _sender task. The main connection handler only reads. This prevents a slow client from blocking the read loop of other clients.

Connection Health Monitoring

Backpressure handling prevents stalls, but we still need to detect dead connections. A client might disconnect abruptly without sending a close frame, leaving a zombie connection on the server.

I implemented a heartbeat mechanism:

import time

class ConnectionManager:
    def __init__(self, max_queue_size: int = 100, heartbeat_interval: float = 30.0):
        self.active_connections: dict[str, WebSocket] = {}
        self.send_queues: dict[str, asyncio.Queue] = {}
        self.last_heartbeat: dict[str, float] = {}
        self.heartbeat_interval = heartbeat_interval
        self.max_queue_size = max_queue_size

    async def connect(self, client_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[client_id] = websocket
        self.send_queues[client_id] = asyncio.Queue(maxsize=self.max_queue_size)
        self.last_heartbeat[client_id] = time.time()
        asyncio.create_task(self._sender(client_id))
        asyncio.create_task(self._heartbeat(client_id))

    async def _heartbeat(self, client_id: str):
        """Send periodic heartbeats to detect stale connections."""
        try:
            while client_id in self.active_connections:
                await asyncio.sleep(self.heartbeat_interval)
                self.last_heartbeat[client_id] = time.time()
                try:
                    queue = self.send_queues[client_id]
                    queue.put_nowait({"type": "ping"})
                except asyncio.QueueFull:
                    # Can't even queue a ping; connection is dead
                    await self.disconnect(client_id)
                    break
        except asyncio.CancelledError:
            pass

    async def disconnect(self, client_id: str):
        if client_id in self.active_connections:
            try:
                await self.active_connections[client_id].close()
            except:
                pass
            del self.active_connections[client_id]
        if client_id in self.send_queues:
            del self.send_queues[client_id]
        if client_id in self.last_heartbeat:
            del self.last_heartbeat[client_id]

The client should respond to pings with pongs. If a client stops responding, the heartbeat task will eventually fail to queue a message and the connection will be cleaned up.

Graceful Degradation Under Load

When a system is overloaded, you have a choice: either drop new connections, or drop slow existing connections. I chose the latter—new clients can always connect, but if they can't keep up, they get disconnected.

async def broadcast(self, message: dict, drop_slow: bool = True):
    """Broadcast with optional backpressure-based disconnection."""
    dead_clients = []
    
    for client_id, queue in self.send_queues.items():
        try:
            queue.put_nowait(message)
        except asyncio.QueueFull:
            if drop_slow:
                dead_clients.append(client_id)
            else:
                # Alternative: wait for queue to drain (blocks)
                try:
                    await asyncio.wait_for(queue.put(message), timeout=0.5)
                except asyncio.TimeoutError:
                    dead_clients.append(client_id)
    
    for client_id in dead_clients:
        try:
            await self.active_connections[client_id].close(code=1008)
        except:
            pass
        await self.disconnect(client_id)

This ensures the server stays responsive. Slow clients get dropped, not the entire system.

Monitoring and Observability

You need visibility into queue depths and connection lifetimes:

class ConnectionManager:
    def get_stats(self) -> dict:
        stats = {
            "total_connections": len(self.active_connections),
            "queue_depths": {},
            "queue_avg": 0,
        }
        
        depths = []
        for client_id, queue in self.send_queues.items():
            depth = queue.qsize()
            depths.append(depth)
            stats["queue_depths"][client_id] = depth
        
        if depths:
            stats["queue_avg"] = sum(depths) / len(depths)
            stats["queue_max"] = max(depths)
            stats["queue_min"] = min(depths)
        
        return stats

@app.get("/stats")
async def get_stats():
    return manager.get_stats()

I log queue depths every 30 seconds. If the average approaches the max queue size, it's a signal to either reduce broadcast frequency or scale horizontally.

Lessons from Production

When I first deployed this to handle 300+ concurrent connections, I learned three things:

1. Queue size matters. Too small and you drop clients unnecessarily. Too large and you mask real performance problems. I landed on 50 as a reasonable middle ground for my use case.

2. Heartbeats aren't free. Sending a ping every 30 seconds to 300 clients is 10 pings per second. That's manageable, but if you're already at capacity, heartbeats can push you over the edge. Monitor this.

3. Slow clients are contagious. If one client is slow and you don't drop it, its queue will fill up and stay full, consuming memory. The moment you detect this, disconnect it. Don't wait for a timeout.

The system now runs at sub-50ms sync times consistently, and I haven't seen a zombie connection in months.