FastAPI websocketi s više od jednog workera: kako doći do svake veze

Uz FastAPI websocketi izgledaju kao posao od nekoliko redaka. Dodate endpoint s @app.websocket, otvorene veze čuvate u listi i svakoj od njih šaljete kad se nešto dogodi. Na prijenosnom računalu to radi iz prve: narudžba se označi kao poslana i to vidi svaka otvorena stranica.

Zatim napravite deploy s više od jednog workera, a neke se stranice prestanu ažurirati. Ne sve, i upravo je zato problem teško pronaći. Ovaj članak objašnjava zašto, što je potrebno da websocketi rade preko više workera i poslužitelja i kako iz FastAPI-ja slati ažuriranja uživo, a da uopće ne pokrećete websocket poslužitelj.

Zašto jedan worker radi, a četiri ne

Uobičajena prva verzija veze čuva u memoriji:

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)

Ta lista živi unutar jednog procesa. U produkciji ih pokrećete nekoliko, kao workere jednog poslužitelja ili kao kopije svoje aplikacije iza load balancera, i svaki ima vlastitu listu. Preglednik se spaja na onaj worker koji je odabrao load balancer. Kad zahtjev na workeru 2 pozove broadcast(), dobiju ga samo preglednici spojeni na worker 2. Ostali za njega nikad ne saznaju.

S jednim workerom greška se ne vidi. S četiri otprilike tri stranice od četiri propuste svako ažuriranje, a koje su to stranice, mijenja se svaki put kad se netko ponovno spoji.

Što je potrebno za rad preko više workera

Da bi došli do svake veze, workeri moraju dijeliti svaki događaj:

• Zajednički kanal među workerima. Svaki se worker pretplaćuje na pub/sub kanal, obično Redis, i ondje se objavljuje svaki broadcast. Svaki ga worker zatim šalje preglednicima koje drži. To je još jedna usluga koju treba održavati u radu, a svaki događaj sada ima jedan skok više.
• Sticky sessions ili ne. Websocket ostaje na workeru na kojem se otvorio, ali HTTP zahtjevi oko njega ne ostaju. Sve što worker čuva o vezi, primjerice tko je korisnik ili kojim se sobama pridružio, mora raditi bez obzira na to koji worker odgovori sljedeći.
• Znati tko se spaja. Preglednik na websocketu ne može postaviti zaglavlja, pa endpoint mora pročitati kolačić sesije ili token u adresi, provjeriti ga i odlučiti koje sobe taj korisnik smije slušati.
• Deploy i ponovno pokretanje. Ponovno pokretanje workera prekida sve veze koje drži, a sve se one odjednom ponovno spajaju. Stranica koja nije bila spojena propušta ono što je za to vrijeme poslano, osim ako nešto ne čuva nedavne događaje kako bi ih ona mogla nadoknaditi.
• Timeouti koji smetaju. Load balanceri i proxyji zatvaraju veze koje izgledaju neaktivno, pa endpointu treba heartbeat, a proxy uopće mora biti podešen tako da propušta websockete.

Ništa od toga nije egzotično. To je sloj za rad u stvarnom vremenu koji treba izgraditi i održavati u radu uz vašu aplikaciju, a svaka usluga koja želi ažuriranja uživo gradi ga iznova.

Ili neka veze drži oblak u stvarnom vremenu

Hostana usluga za rad u stvarnom vremenu preuzima veze od vaših workera. Preglednici se spajaju na uslugu. Vaši FastAPI endpointi svaki događaj šalju jednim kratkim HTTP zahtjevom, s onog workera koji obrađuje zahtjev, a usluga ga isporučuje svakoj stranici koja ga treba dobiti. Dodajte workere, ponovno ih pokrenite ili premjestite, a u isporuci se ništa ne mijenja.

FastAPI na WordPlus Cloudu

S WordPlus Cloudom vaša aplikacija jednom definira svoje sobe, svaku s pravilom o tome tko joj se smije pridružiti, i uključuje jedan 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")

Router odgovara stranicama kad se traže pridružiti sobi, s korisnikom kojeg pronađe vaša aplikacija, i poslužuje datoteke SDK-a za preglednik, pa ne treba dodavati ni Node ni bundler. Stranica renderira realtime.head() za svog korisnika, a zatim se pridružuje sobi u običnom JavaScriptu:

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

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

A šalje bilo koji endpoint, na bilo kojem workeru:

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

To radi jednako s jednim workerom ili s četrdeset. Osim toga:

• Vaša pravila nikad ne blokiraju event loop. To su obične funkcije koje se izvršavaju u worker threadu, pa upit bazi podataka u nekoj od njih nije problem.
• Ništa se ne gubi pri ponovnom spajanju. Soba čuva svoje zadnje događaje, pa stranica koja se vrati dobiva ono što je propustila, redom.
• Jedna veza za sve kartice. Kartice jednog preglednika dijele jednu vezu, pa se osoba s tri otvorene kartice broji jednom.
• Zapečaćeno na putu. Ono što vaša aplikacija šalje putuje šifrirano ključevima koje drže samo vaš poslužitelj i vaši korisnici.
• Više od događaja. Tko je na mreži, pozivi koji zvone u svakoj kartici, grupni pozivi i prijenosi uživo dolaze na istoj vezi, s gotovim zaslonom poziva, chatom i playerom iz Better Messagesa ili s vašim vlastitim sučeljem na istom API-ju.

Vodič za FastAPI sadrži cijelo podešavanje, a demo uživo je mala trgovina izgrađena na njemu, u kojoj se narudžba, čim kliknete, u drugom prozoru označi kao poslana.

WordPlus Cloud for Apps ima cijenu prema istodobnim vezama, od 1.000 naviše, uz mjesečnu ili godišnju naplatu. Pogledajte što nudi aplikacijama ili cijene za aplikacije. Pitanja o vašoj integraciji? Pišite nam.