Websockety w FastAPI na więcej niż jednym workerze: jak dotrzeć do każdego połączenia

FastAPI sprawia, że websockety wyglądają na kilka linijek kodu. Dodajesz endpoint z @app.websocket, trzymasz otwarte połączenia na liście i gdy coś się dzieje, wysyłasz coś do każdego z nich. Na laptopie działa to od pierwszego razu: status zamówienia zmienia się na „wysłane” i widzi to każda otwarta strona.

Potem robisz deploy z więcej niż jednym workerem i niektóre strony przestają się aktualizować. Nie wszystkie, i właśnie dlatego tak trudno znaleźć przyczynę. Ten wpis wyjaśnia dlaczego, czego wymaga działanie websocketów na wielu workerach i serwerach i jak wysyłać aktualizacje na żywo z FastAPI w ogóle bez utrzymywania serwera websocket.

Dlaczego jeden worker działa, a cztery nie

Typowa pierwsza wersja trzyma połączenia w pamięci:

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 żyje wewnątrz jednego procesu. Na produkcji uruchamiasz ich kilka, jako workery jednego serwera albo jako kopie aplikacji za load balancerem, i każdy ma własną listę. Przeglądarka łączy się z tym workerem, którego wybrał load balancer. Gdy żądanie na workerze 2 wywołuje broadcast(), wiadomość dostają tylko przeglądarki połączone z workerem 2. Pozostałe nigdy się o niej nie dowiadują.

Przy jednym workerze błędu nie widać. Przy czterech mniej więcej trzy strony na cztery tracą każdą aktualizację, a to, których stron to dotyczy, zmienia się za każdym razem, gdy ktoś połączy się ponownie.

Czego wymaga działanie na wielu workerach

Żeby dotrzeć do każdego połączenia, workery muszą dzielić się każdym zdarzeniem:

• Wspólny kanał między workerami. Każdy worker subskrybuje kanał pub/sub, zwykle Redis, i tam publikowany jest każdy broadcast. Potem każdy worker wysyła go do przeglądarek, które ma u siebie. To kolejna usługa do utrzymywania, a każde zdarzenie pokonuje teraz dodatkowy przeskok.
• Sticky sessions albo ich brak. Websocket zostaje na workerze, na którym został otwarty, ale żądania HTTP wokół niego już nie. Wszystko, co worker przechowuje o połączeniu, na przykład kim jest użytkownik albo do jakich pokoi dołączył, musi działać niezależnie od tego, który worker odpowie jako następny.
• Rozpoznanie, kto się łączy. Przeglądarka nie może ustawić nagłówków dla websocketu, więc endpoint musi odczytać plik cookie sesji albo token w adresie, sprawdzić go i zdecydować, których pokoi ten użytkownik może słuchać.
• Deploye i restarty. Restart workera zrywa każde połączenie, które on utrzymuje, i wszystkie łączą się ponownie naraz. Strona, która była w tym czasie rozłączona, traci to, co wysłano w międzyczasie, chyba że coś przechowa dla niej ostatnie zdarzenia, żeby mogła je nadrobić.
• Timeouty po drodze. Load balancery i proxy zamykają połączenia, które wyglądają na bezczynne, więc endpoint potrzebuje heartbeatu, a proxy musi być skonfigurowane tak, żeby w ogóle przepuszczało websockety.

Nic z tego nie jest egzotyczne. To warstwa czasu rzeczywistego, którą trzeba zbudować i utrzymywać obok aplikacji, a każda usługa, która chce mieć aktualizacje na żywo, buduje ją od nowa.

Albo niech połączenia utrzymuje chmura czasu rzeczywistego

Hostowana usługa czasu rzeczywistego przejmuje połączenia od Twoich workerów. Przeglądarki łączą się z usługą. Twoje endpointy FastAPI wysyłają każde zdarzenie jednym krótkim żądaniem HTTP, z tego workera, który akurat obsługuje żądanie, a usługa dostarcza je każdej stronie, która powinna je dostać. Dodawaj workery, restartuj je albo przenoś, a w dostarczaniu nic się nie zmienia.

FastAPI w WordPlus Cloud

Z WordPlus Cloud Twoja aplikacja raz deklaruje swoje pokoje, każdy z regułą określającą, kto może do niego dołączyć, i montuje jeden 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 odpowiada stronom, gdy proszą o dołączenie do pokoju, przekazując użytkownika ustalonego przez Twoją aplikację, i serwuje pliki SDK dla przeglądarki, więc nie trzeba dodawać ani Node, ani bundlera. Strona renderuje realtime.head() dla swojego użytkownika, a potem dołącza do pokoju w czystym JavaScripcie:

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

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

A dowolny endpoint, na dowolnym workerze, wysyła zdarzenie:

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

Działa to tak samo z jednym workerem, jak i z czterdziestoma. A do tego:

• Twoje reguły nigdy nie blokują event loopa. To zwykłe funkcje uruchamiane w wątku roboczym, więc zapytanie do bazy danych w jednej z nich nie stanowi problemu.
• Nic nie ginie przy ponownym połączeniu. Pokój przechowuje swoje ostatnie zdarzenia, więc strona, która wraca, dostaje to, co ją ominęło, w odpowiedniej kolejności.
• Jedno połączenie dla wszystkich kart. Karty jednej przeglądarki dzielą jedno połączenie, więc osoba z trzema otwartymi kartami jest liczona tylko raz.
• Zapieczętowane w drodze. To, co wysyła Twoja aplikacja, podróżuje zaszyfrowane kluczami, które mają tylko Twój serwer i Twoi użytkownicy.
• Więcej niż zdarzenia. Informacja, kto jest online, rozmowy, które dzwonią w każdej karcie, rozmowy grupowe i transmisje na żywo działają przez to samo połączenie, z gotowym ekranem rozmowy, czatem i odtwarzaczem z Better Messages albo z Twoim własnym interfejsem na tym samym API.

Przewodnik dla FastAPI zawiera całą konfigurację, a demo na żywo to mały sklep zbudowany na niej, w którym status zamówienia zmienia się na „wysłane” w drugim oknie w chwili, gdy klikasz.

WordPlus Cloud for Apps ma cenę według jednoczesnych połączeń, od 1000 wzwyż, rozliczaną miesięcznie lub rocznie. Zobacz, co daje aplikacjom albo cennik dla aplikacji. Chcesz zapytać o swoją integrację? Napisz do nas.