Des websockets FastAPI au-delà d’un seul worker : atteindre chaque connexion

Avec FastAPI, les websockets semblent tenir en quelques lignes. Vous ajoutez un endpoint avec @app.websocket, vous gardez les connexions ouvertes dans une liste, et vous prévenez chacune d’elles quand quelque chose se passe. Sur votre ordinateur portable, cela fonctionne du premier coup : la commande passe à « expédiée » et chaque page ouverte le voit.

Puis vous déployez avec plus d’un worker, et certaines pages cessent de se mettre à jour. Pas toutes, et c’est bien ce qui rend le problème difficile à trouver. Cet article explique pourquoi, ce qu’il faut pour que les websockets fonctionnent sur plusieurs workers et plusieurs serveurs, et comment envoyer des mises à jour en direct depuis FastAPI sans faire tourner le moindre serveur websocket.

Pourquoi un worker fonctionne et quatre non

La première version habituelle garde les connexions en mémoire :

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)

Cette liste vit à l’intérieur d’un seul processus. En production, vous en faites tourner plusieurs, comme workers d’un même serveur ou comme copies de votre application derrière un load balancer, et chacun a sa propre liste. Un navigateur se connecte au worker que le load balancer a choisi, quel qu’il soit. Quand une requête sur le worker 2 appelle broadcast(), seuls les navigateurs connectés au worker 2 reçoivent le message. Les autres n’en entendent jamais parler.

Avec un seul worker, le bug est invisible. Avec quatre, environ trois pages sur quatre manquent chaque mise à jour, et ce ne sont plus les mêmes dès que quelqu’un se reconnecte.

Ce qu’il faut avec plusieurs workers

Pour atteindre chaque connexion, les workers doivent partager chaque événement :

• Un canal partagé entre les workers. Chaque worker s’abonne à un canal pub/sub, en général Redis, et chaque message diffusé y est publié. Chaque worker le transmet ensuite aux navigateurs dont il tient les connexions. C’est un service de plus à faire tourner, et chaque événement fait désormais un saut de plus.
• Avec ou sans sticky sessions. Un websocket reste sur le worker où il s’est ouvert, mais les requêtes HTTP qui l’entourent, non. Tout ce qu’un worker garde sur une connexion, comme l’identité de l’utilisateur ou les salons qu’il a rejoints, doit fonctionner quel que soit le worker qui répond ensuite.
• Savoir qui se connecte. Un navigateur ne peut pas ajouter d’en-têtes à un websocket, si bien que l’endpoint doit lire le cookie de session ou un token placé dans l’adresse, le vérifier, et décider quels salons cet utilisateur peut écouter.
• Déploiements et redémarrages. Redémarrer un worker coupe toutes les connexions qu’il tient, et elles se reconnectent toutes en même temps. Une page qui était absente manque ce qui a été envoyé entre-temps, à moins que quelque chose ne garde les événements récents pour qu’elle les rattrape.
• Des timeouts sur le chemin. Les load balancers et les proxys ferment les connexions qui semblent inactives, si bien que l’endpoint a besoin d’un heartbeat, et que le proxy doit avant tout être configuré pour laisser passer les websockets.

Rien de tout cela n’est exotique. C’est une couche temps réel à construire et à faire tourner à côté de votre application, et chaque service qui veut des mises à jour en direct la reconstruit à son tour.

Ou confier les connexions à un cloud temps réel

Un service temps réel hébergé retire les connexions de vos workers. Les navigateurs se connectent au service. Vos endpoints FastAPI envoient chaque événement par une seule courte requête HTTP, depuis le worker qui traite la requête, quel qu’il soit, et le service le distribue à chaque page qui doit le recevoir. Ajoutez des workers, redémarrez-les ou déplacez-les, et rien ne change dans la distribution.

FastAPI sur WordPlus Cloud

Avec WordPlus Cloud, votre application déclare ses salons une fois pour toutes, avec pour chacun une règle qui dit qui peut le rejoindre, et monte un seul 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")

Le router répond aux pages quand elles demandent à rejoindre un salon, avec l’utilisateur que trouve votre application, et sert les fichiers du SDK destinés au navigateur, si bien qu’il n’y a ni Node ni bundler à ajouter. Une page rend realtime.head() pour son utilisateur, puis rejoint le salon en JavaScript pur :

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

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

Et n’importe quel endpoint, sur n’importe quel worker, envoie :

@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}

Cela tourne de la même façon avec un worker ou avec quarante. Au-delà :

• Vos règles ne bloquent jamais la boucle d’événements. Ce sont de simples fonctions, exécutées dans un thread de travail, si bien qu’une requête à la base de données dans l’une d’elles ne pose aucun problème.
• Rien de perdu à la reconnexion. Un salon garde ses derniers événements, si bien qu’une page qui revient reçoit ce qu’elle a manqué, dans l’ordre.
• Une seule connexion pour tous les onglets. Les onglets d’un même navigateur partagent une seule connexion, si bien qu’une personne avec trois onglets ouverts ne compte qu’une fois.
• Scellé en chemin. Ce que votre application envoie voyage chiffré, avec des clés que seuls votre serveur et vos utilisateurs détiennent.
• Plus que des événements. La liste des personnes en ligne, les appels qui sonnent dans chaque onglet, les appels de groupe et les diffusions en direct passent par la même connexion, avec un écran d’appel, un chat et un lecteur prêts à l’emploi venus de Better Messages, ou votre propre interface sur la même API.

Le guide FastAPI détaille toute la mise en place, et la démo en ligne est une petite boutique construite dessus, où une commande passe à « expédiée » dans une autre fenêtre au moment où vous cliquez.

WordPlus Cloud for Apps a un prix selon les connexions simultanées, à partir de 1 000, facturé au mois ou à l’année. Découvrez ce qu’il apporte aux applications, ou consultez les tarifs pour les applications. Des questions sur votre intégration ? Écrivez-nous.