Websockets en FastAPI con más de un worker: llegar a todas las conexiones

FastAPI hace que los websockets parezcan cosa de unas pocas líneas. Añades un endpoint con @app.websocket, guardas las conexiones abiertas en una lista y envías a cada una cuando pasa algo. En tu portátil funciona a la primera: el pedido pasa a enviado y todas las páginas abiertas lo ven.

Luego despliegas con más de un worker, y algunas páginas dejan de actualizarse. No todas, y eso es lo que lo hace difícil de encontrar. Este artículo explica por qué, qué hace falta para que los websockets funcionen entre workers y servidores, y cómo enviar actualizaciones en directo desde FastAPI sin mantener ningún servidor websocket.

Por qué funciona con un worker y no con cuatro

La primera versión habitual guarda las conexiones en memoria:

connections: list[WebSocket] = []

@app.websocket("/ws")
async def updates(websocket: WebSocket):
    await websocket.accept()
    connections.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        connections.remove(websocket)

async def broadcast(message: str):
    for connection in connections:
        await connection.send_text(message)

Esa lista vive dentro de un solo proceso. En producción ejecutas varios, como workers de un mismo servidor o como copias de tu app detrás de un load balancer, y cada uno tiene su propia lista. Un navegador se conecta al worker que haya elegido el load balancer. Cuando una petición en el worker 2 llama a broadcast(), solo lo reciben los navegadores conectados al worker 2. Los demás nunca se enteran.

Con un worker, el bug es invisible. Con cuatro, más o menos tres de cada cuatro páginas se pierden cada actualización, y cuáles son cambia cada vez que alguien se reconecta.

Qué hace falta entre workers

Para llegar a todas las conexiones, los workers tienen que compartir cada evento:

• Un canal compartido entre workers. Cada worker se suscribe a un canal de pub/sub, normalmente Redis, y cada broadcast se publica ahí. Después, cada worker lo envía a los navegadores que tiene conectados. Es un servicio más que mantener, y ahora cada evento da un salto extra.
• Con sticky sessions o sin ellas. Un websocket se queda en el worker en el que se abrió, pero las peticiones HTTP que lo rodean no. Todo lo que un worker guarda sobre una conexión, como quién es el usuario o a qué salas se ha unido, tiene que funcionar sea cual sea el worker que responda después.
• Saber quién se conecta. Un navegador no puede poner cabeceras en un websocket, así que el endpoint tiene que leer la cookie de sesión o un token en la URL, comprobarlo y decidir qué salas puede escuchar ese usuario.
• Deploys y reinicios. Reiniciar un worker corta todas las conexiones que tiene, y todas se reconectan a la vez. Una página que estaba desconectada se pierde lo que se envió mientras tanto, a menos que algo guarde los eventos recientes para que pueda ponerse al día.
• Timeouts por el camino. Los load balancers y los proxies cierran las conexiones que parecen inactivas, así que el endpoint necesita un heartbeat y, antes que nada, el proxy tiene que estar configurado para dejar pasar los websockets.

Nada de esto es exótico. Es una capa de tiempo real que construir y mantener en marcha junto a tu app, y cada servicio que quiere actualizaciones en directo la vuelve a construir.

O deja que una nube en tiempo real mantenga las conexiones

Un servicio de tiempo real alojado saca las conexiones de tus workers. Los navegadores se conectan al servicio. Tus endpoints de FastAPI envían cada evento en una petición HTTP corta, desde el worker que atienda la petición, y el servicio lo entrega a cada página que deba recibirlo. Añade workers, reinícialos o muévelos, y la entrega no cambia en nada.

FastAPI en WordPlus Cloud

Con WordPlus Cloud, tu app declara sus salas una sola vez, con una regla para quién puede unirse a cada una, y monta un router:

from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware
from wordplus_realtime import Realtime
import wordplus_realtime.fastapi as wr

realtime = Realtime.from_environment()   # WORDPLUS_SITE, WORDPLUS_SECRET, WORDPLUS_LOCAL_KEY

# The room of an order, order:1042: its customer may join, and the last 20 updates are kept.
realtime.room("order:{id}", lambda user, p: user is not None and owner_of(p["id"]) == user, history=20)

async def current_user(request: Request):
    return request.session.get("user")   # your sign-in

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET)
app.include_router(wr.router(realtime, current_user), prefix="/api/realtime")

El router responde a las páginas cuando piden unirse a una sala, con el usuario que encuentra tu app, y sirve los archivos del SDK para el navegador, así que no hay que añadir Node ni ningún bundler. Una página renderiza realtime.head() para su usuario y luego se une a la sala con JavaScript puro:

import { connect } from '@wordplus/realtime';

const rt = await connect( window.wordplusRealtimeConfig );
rt.join( 'order:1042' ).on( 'updated', ( data ) => showStatus( data.status ) );

Y cualquier endpoint, en cualquier worker, envía:

@app.post("/orders/{pk}/ship")
async def ship(pk: int):
    order = await mark_shipped(pk)
    await realtime.to(f"order:{pk}").aemit("updated", {"status": order.status})
    return {"ok": True}

Eso funciona igual con un worker que con cuarenta. Más allá de eso:

• Tus reglas nunca bloquean el event loop. Son funciones normales que se ejecutan en un worker thread, así que puedes hacer una consulta a la base de datos dentro de una sin problema.
• Nada se pierde al reconectar. Una sala guarda sus últimos eventos, así que una página que vuelve recibe lo que se perdió, en orden.
• Una conexión para todas las pestañas. Las pestañas de un navegador comparten una conexión, así que una persona con tres pestañas abiertas cuenta una sola vez.
• Sellado en tránsito. Lo que envía tu app viaja cifrado con claves que solo tienen tu servidor y tus usuarios.
• Más que eventos. Quién está en línea, llamadas que suenan en todas las pestañas, llamadas en grupo y transmisiones en directo llegan por la misma conexión, con una pantalla de llamada, un chat y un reproductor listos para usar de Better Messages, o tu propia interfaz sobre la misma API.

La guía de FastAPI tiene toda la configuración, y la demo en vivo es una pequeña tienda construida sobre ella, en la que un pedido pasa a enviado en otra ventana en cuanto haces clic.

WordPlus Cloud for Apps tiene un precio según las conexiones simultáneas, a partir de 1.000, con facturación mensual o anual. Descubre lo que hace para apps, o consulta los precios para apps. ¿Preguntas sobre tu integración? Escríbenos.