FastAPI-websockets met meer dan één worker: elke verbinding bereiken

Met FastAPI lijken websockets een kwestie van een paar regels. Je voegt een endpoint toe met @app.websocket, houdt de open verbindingen bij in een lijst, en stuurt naar elk ervan als er iets gebeurt. Op je laptop werkt het meteen: de bestelling krijgt de status verzonden en elke open pagina ziet het.

Dan deploy je met meer dan één worker, en sommige pagina’s worden niet meer bijgewerkt. Niet allemaal, en juist dat maakt het lastig te vinden. In dit artikel lees je waarom, wat er nodig is om websockets over meerdere workers en servers te laten werken, en hoe je vanuit FastAPI live updates verstuurt zonder überhaupt een websocketserver te draaien.

Waarom één worker werkt en vier niet

De gebruikelijke eerste versie houdt de verbindingen in het geheugen bij:

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)

Die lijst leeft binnen één proces. In productie draai je er meerdere, als workers van één server of als kopieën van je app achter een load balancer, en elk heeft een eigen lijst. Een browser maakt verbinding met de worker die de load balancer toevallig koos. Als een verzoek op worker 2 broadcast() aanroept, krijgen alleen de browsers die met worker 2 verbonden zijn het. De andere horen er nooit van.

Met één worker is de bug onzichtbaar. Met vier gaat elke update aan ongeveer drie van de vier pagina’s voorbij, en welke dat zijn, verandert telkens als iemand opnieuw verbinding maakt.

Wat er nodig is met meerdere workers

Om elke verbinding te bereiken, moeten de workers elk event met elkaar delen:

• Een gedeeld kanaal tussen de workers. Elke worker abonneert zich op een pub/sub-kanaal, meestal Redis, en elke broadcast wordt daar gepubliceerd. Elke worker stuurt hem vervolgens naar de browsers die hij vasthoudt. Dat is nog een dienst om te draaien, en elk event maakt nu een extra tussenstap.
• Sticky sessions of niet. Een websocket blijft bij de worker waarop hij geopend werd, maar de HTTP-verzoeken eromheen niet. Alles wat een worker over een verbinding bijhoudt, zoals wie de gebruiker is of aan welke ruimtes hij deelneemt, moet werken, welke worker er ook als volgende antwoordt.
• Weten wie er verbinding maakt. Een browser kan geen headers meegeven aan een websocket, dus het endpoint moet de sessiecookie of een token in het adres uitlezen, controleren en bepalen naar welke ruimtes deze gebruiker mag luisteren.
• Deploys en herstarts. Een herstart van een worker verbreekt elke verbinding die hij vasthoudt, en ze maken allemaal tegelijk opnieuw verbinding. Een pagina die even weg was, mist wat er intussen is verstuurd, tenzij iets de recente events bewaart, zodat ze die kan inhalen.
• Timeouts die in de weg zitten. Load balancers en proxy’s sluiten verbindingen die inactief lijken, dus het endpoint heeft een heartbeat nodig, en de proxy moet zo ingesteld zijn dat hij websockets überhaupt doorlaat.

Niets hiervan is exotisch. Het is een realtimelaag die je naast je app moet bouwen en draaiende houden, en elke dienst die live updates wil, bouwt hem opnieuw.

Of laat een realtime cloud de verbindingen vasthouden

Een gehoste realtime dienst haalt de verbindingen weg uit je workers. De browsers maken verbinding met de dienst. Je FastAPI-endpoints versturen elk event in één kort HTTP-verzoek, vanuit de worker die het verzoek afhandelt, welke dat ook is, en de dienst bezorgt het aan elke pagina die het moet krijgen. Voeg workers toe, herstart ze of verplaats ze, en aan de bezorging verandert niets.

FastAPI op WordPlus Cloud

Met WordPlus Cloud legt je app zijn ruimtes één keer vast, elk met een regel voor wie erin mag, en voegt één router toe:

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")

De router beantwoordt de pagina’s als ze vragen om aan een ruimte deel te nemen, met de gebruiker die je app vaststelt, en serveert de browserbestanden van de SDK, dus Node en een bundler hoef je niet toe te voegen. Een pagina rendert realtime.head() voor de gebruiker en neemt dan in gewoon JavaScript deel aan de ruimte:

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

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

En elk endpoint, op elke worker, verstuurt:

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

Dat werkt hetzelfde met één worker of met veertig. En daarnaast:

• Je regels blokkeren de event loop nooit. Het zijn gewone functies, uitgevoerd op een worker thread, dus een databasequery in een regel kan gewoon.
• Niets kwijt bij opnieuw verbinden. Een ruimte bewaart de laatste events, dus een pagina die terugkomt, krijgt wat ze gemist heeft, in de juiste volgorde.
• Eén verbinding voor alle tabbladen. De tabbladen van een browser delen één verbinding, zodat iemand met drie tabbladen open maar één keer telt.
• Verzegeld onderweg. Wat je app verstuurt, reist versleuteld met sleutels die alleen je server en je gebruikers hebben.
• Meer dan events. Wie er online is, gesprekken die in elk tabblad overgaan, groepsgesprekken en livestreams lopen over dezelfde verbinding, met een kant-en-klaar belscherm, chat en speler van Better Messages, of je eigen interface op dezelfde API.

De FastAPI-handleiding beschrijft de hele installatie, en de live demo is een kleine webwinkel die erop draait, waar een bestelling in een ander venster de status verzonden krijgt terwijl je klikt.

WordPlus Cloud for Apps is geprijsd op gelijktijdige verbindingen, vanaf 1.000, maandelijks of jaarlijks gefactureerd. Bekijk wat het doet voor apps, of de prijzen voor apps. Vragen over je integratie? Schrijf ons.