Websocket in FastAPI oltre un solo worker: raggiungere ogni connessione

FastAPI fa sembrare i websocket una questione di poche righe. Aggiungi un endpoint con @app.websocket, tieni le connessioni aperte in una lista e invii a ciascuna quando succede qualcosa. Sul tuo portatile funziona al primo colpo: l’ordine passa a spedito e ogni pagina aperta lo vede.

Poi fai il deploy con più di un worker, e alcune pagine smettono di aggiornarsi. Non tutte, ed è proprio questo a renderlo difficile da scovare. Questo articolo spiega perché, cosa serve per far funzionare i websocket su più worker e più server, e come inviare aggiornamenti in tempo reale da FastAPI senza gestire nessun server websocket.

Perché con un worker funziona e con quattro no

La prima versione tipica tiene le connessioni in 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)

Quella lista vive dentro un solo processo. In produzione ne fai girare diversi, come worker di un unico server o come copie della tua app dietro un load balancer, e ognuno ha la sua lista. Un browser si collega al worker che il load balancer ha scelto. Quando una richiesta sul worker 2 chiama broadcast(), lo ricevono solo i browser collegati al worker 2. Gli altri non ne sanno mai nulla.

Con un worker il bug è invisibile. Con quattro, circa tre pagine su quattro perdono ogni aggiornamento, e quali siano cambia ogni volta che qualcuno si riconnette.

Cosa serve con più worker

Per raggiungere ogni connessione, i worker devono condividere ogni evento:

• Un canale condiviso tra i worker. Ogni worker si iscrive a un canale pub/sub, di solito Redis, e ogni broadcast viene pubblicato lì. Poi ogni worker lo invia ai browser che tiene. È un servizio in più da gestire, e ora ogni evento fa un passaggio in più.
• Sticky session o no. Un websocket resta sul worker su cui si è aperto, ma le richieste HTTP intorno a lui no. Qualsiasi cosa un worker conservi su una connessione, come chi è l’utente o in quali stanze è entrato, deve funzionare a prescindere da quale worker risponda dopo.
• Sapere chi si sta collegando. Un browser non può impostare header su un websocket, quindi l’endpoint deve leggere il cookie di sessione o un token nell’indirizzo, verificarlo e decidere quali stanze questo utente può ascoltare.
• Deploy e riavvii. Riavviare un worker fa cadere ogni connessione che tiene, e tutte si riconnettono nello stesso momento. Una pagina rimasta scollegata perde ciò che è stato inviato nel frattempo, a meno che qualcosa non conservi gli eventi recenti perché possa recuperarli.
• Timeout che si mettono di mezzo. Load balancer e proxy chiudono le connessioni che sembrano inattive, quindi l’endpoint ha bisogno di un heartbeat, e il proxy va configurato anche solo per lasciar passare i websocket.

Niente di tutto questo è esotico. È uno strato in tempo reale da costruire e tenere in funzione accanto alla tua app, e ogni servizio che vuole aggiornamenti in tempo reale lo ricostruisce da capo.

Oppure lascia che sia un cloud in tempo reale a tenere le connessioni

Un servizio in tempo reale gestito toglie le connessioni ai tuoi worker. I browser si collegano al servizio. I tuoi endpoint FastAPI inviano ogni evento con una breve richiesta HTTP, da qualunque worker gestisca la richiesta, e il servizio lo consegna a ogni pagina che deve riceverlo. Aggiungi worker, riavviali o spostali, e nella consegna non cambia nulla.

FastAPI su WordPlus Cloud

Con WordPlus Cloud, la tua app dichiara le sue stanze una volta sola, con una regola su chi può entrare in ciascuna, e monta un unico 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")

Il router risponde alle pagine quando chiedono di entrare in una stanza, con l’utente che trova la tua app, e serve i file dell’SDK per il browser, così non c’è Node né un bundler da aggiungere. Una pagina renderizza realtime.head() per il suo utente, poi entra nella stanza con JavaScript puro:

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

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

E qualsiasi endpoint, su qualsiasi worker, invia:

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

Funziona allo stesso modo con un worker o con quaranta. In più:

• Le tue regole non bloccano mai l’event loop. Sono semplici funzioni, eseguite su un worker thread, quindi una query al database dentro una di esse va benissimo.
• Nulla si perde alla riconnessione. Una stanza conserva i suoi ultimi eventi, così una pagina che torna riceve ciò che si è persa, in ordine.
• Una sola connessione per tutte le schede. Le schede di un browser condividono un’unica connessione, così una persona con tre schede aperte conta una volta sola.
• Sigillato lungo il tragitto. Ciò che la tua app invia viaggia cifrato con chiavi che hanno solo il tuo server e i tuoi utenti.
• Non solo eventi. Chi è online, chiamate che squillano in ogni scheda, chiamate di gruppo e live streaming arrivano sulla stessa connessione, con la schermata di chiamata, la chat e il player pronti all’uso di Better Messages, oppure con la tua interfaccia sulla stessa API.

La guida per FastAPI contiene tutta la configurazione, e la demo dal vivo è un piccolo negozio costruito su di essa, dove un ordine passa a spedito in un’altra finestra mentre fai clic.

WordPlus Cloud for Apps ha un prezzo in base alle connessioni simultanee, a partire da 1.000, con fatturazione mensile o annuale. Scopri cosa fa per le app, o i prezzi per le app. Domande sulla tua integrazione? Scrivici.