WebSocket Events

WebSocket Events

A WebSocket endpoint receives an undifferentiated stream of messages. An HTTP application gets routing for free — method plus path picks the handler. Over a socket you have to build that yourself, and the shape you choose early is hard to change later because clients depend on it.

This page covers two directions:

Inbound — a client sends {"type": "message", ...} and something has to decide which function handles it.

Outbound — an order is placed by an HTTP request, a job finishes in a worker, a row changes in the database, and connected clients need to know.

A message envelope

Decide on an envelope before writing any handlers. The one that works:

{"type": "chat.message", "id": "c7f1", "data": {"text": "hello"}}

type selects the handler. data carries the payload, kept in its own object so the envelope can gain fields without colliding with application keys. id is a client-generated correlation id, echoed in any reply, so a client that sends three requests can match three responses — there is no request/response pairing in the protocol itself.

Use dotted namespaces (chat.message, chat.typing, presence.ping). They group naturally in logs, and they give you an obvious prefix to authorize on.

Version the envelope from day one, via the subprotocol or a v field. Clients are cached, installed, and sometimes never updated; you will eventually be serving two message formats at once.

Dispatching inbound messages

The naive version is a chain of if statements in on_receive. It works until there are six types, and then it never gets refactored. Start with a registry.

a dispatch table on the consumer

import json
from typing import Any, Awaitable, Callable

from sillo.websockets import WebSocketConsumer


class Dispatcher(WebSocketConsumer):
    handlers: dict[str, str] = {}

    @classmethod
    def handles(cls, message_type: str):
        """Register a method as the handler for a message type."""
        def decorator(func):
            cls.handlers[message_type] = func.__name__
            return func
        return decorator

    async def on_receive(self, websocket, data):
        try:
            envelope = json.loads(data) if isinstance(data, str) else data
        except (json.JSONDecodeError, TypeError):
            await self.reply(websocket, None, "error", {"detail": "invalid JSON"})
            return

        message_type = envelope.get("type")
        method_name = self.handlers.get(message_type)
        if method_name is None:
            await self.reply(
                websocket, envelope.get("id"), "error",
                {"detail": f"unknown type {message_type!r}"},
            )
            return

        try:
            await getattr(self, method_name)(websocket, envelope)
        except Exception:
            self.logger.exception("handler for %s failed", message_type)
            await self.reply(
                websocket, envelope.get("id"), "error", {"detail": "internal error"}
            )

    async def reply(self, websocket, correlation_id, message_type, payload):
        await websocket.send_json(
            {"type": message_type, "id": correlation_id, "data": payload}
        )

Then handlers are declarative:

using the dispatcher

class ChatConsumer(Dispatcher):
    handlers = {}          # each subclass needs its own registry

    async def on_connect(self, websocket):
        await websocket.accept()
        self.room = f"room:{websocket.path_params['room_id']}"
        await self.join_group(self.room)

    @Dispatcher.handles("chat.message")
    async def handle_message(self, websocket, envelope):
        text = str(envelope["data"].get("text", ""))[:2000]
        await self.broadcast(
            {"type": "chat.message", "data": {"text": text}}, group_name=self.room
        )
        await self.reply(websocket, envelope.get("id"), "chat.ack", {"ok": True})

    @Dispatcher.handles("chat.typing")
    async def handle_typing(self, websocket, envelope):
        await self.broadcast(
            {"type": "chat.typing", "data": {}}, group_name=self.room
        )

Note handlers = {} redeclared on the subclass. Without it, every subclass shares the base class’s dictionary and registers into the same table — one consumer’s message types silently become available on another’s endpoint.

The catch-all except Exception is load-bearing. Without it, one bad message propagates out of on_receive and terminates the connection, as described in Consumers.

Validating message payloads

Type-based dispatch tells you which handler; it says nothing about whether the payload is well-formed. Since Pydantic is already a dependency, use it.

per-type schemas

from pydantic import BaseModel, Field, ValidationError


class ChatMessage(BaseModel):
    text: str = Field(min_length=1, max_length=2000)


class Reaction(BaseModel):
    message_id: str
    emoji: str = Field(max_length=8)


SCHEMAS = {"chat.message": ChatMessage, "chat.react": Reaction}


async def parse(envelope: dict):
    schema = SCHEMAS.get(envelope.get("type"))
    if schema is None:
        raise ValueError(f"unknown type {envelope.get('type')!r}")
    return schema.model_validate(envelope.get("data") or {})

using it in a handler

@Dispatcher.handles("chat.message")
async def handle_message(self, websocket, envelope):
    try:
        payload = await parse(envelope)
    except ValidationError as exc:
        await self.reply(websocket, envelope.get("id"), "error",
                         {"detail": exc.errors()})
        return
    await self.broadcast(
        {"type": "chat.message", "data": {"text": payload.text}},
        group_name=self.room,
    )

Every message is untrusted input, and there is no framework validation

An HTTP route with request_model= gets a 422 for free. A WebSocket message gets nothing — whatever the client sends arrives in on_receive unexamined.

Three things follow. Validate every payload before use, exactly as you would an HTTP body. Enforce a size limit yourself, because a client can send a 50 MB frame and the default configuration will happily buffer it. And re-check authorization per message, not only at connect time: a connection opened an hour ago belongs to a user whose permissions may have changed, or whose session may have been revoked.

async def on_receive(self, websocket, data):
    if len(data) > 64 * 1024:
        await websocket.close(code=status.WS_1009_MESSAGE_TOO_BIG)
        return
    if not await still_authorized(self.user, self.room):
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    ...

Rate limiting per connection

A socket that is open is a socket that can be flooded. The HTTP rate limiter does not apply — it works per request, and a WebSocket is one request that never ends.

a token bucket per connection

import time


class RateLimited(Dispatcher):
    rate = 10          # messages
    per = 1.0          # seconds

    async def on_connect(self, websocket):
        await super().on_connect(websocket)
        self._tokens = float(self.rate)
        self._last = time.monotonic()

    def _allow(self) -> bool:
        now = time.monotonic()
        self._tokens = min(
            self.rate, self._tokens + (now - self._last) * (self.rate / self.per)
        )
        self._last = now
        if self._tokens < 1:
            return False
        self._tokens -= 1
        return True

    async def on_receive(self, websocket, data):
        if not self._allow():
            await websocket.send_json({"type": "error", "data": {"detail": "slow down"}})
            return
        await super().on_receive(websocket, data)

State lives on the instance, and instances are per-connection, so this is naturally scoped correctly. Replying rather than disconnecting is usually kinder — a well-behaved client backs off, and a misbehaving one can be closed after N violations.

Outbound: bridging domain events

The more valuable direction. Something happens elsewhere in the application and connected clients should learn about it.

ChannelBox.group_send is callable from anywhere in the process, so the simplest bridge is a direct call:

broadcasting from an HTTP handler

from sillo.websockets import ChannelBox


@app.post("/orders/{order_id}/ship")
async def ship_order(request, response):
    order = await Order.get(id=request.path_params["order_id"])
    await order.mark_shipped()

    await ChannelBox.group_send(
        f"order:{order.id}",
        {"type": "order.shipped", "data": {"id": order.id, "at": order.shipped_at}},
    )
    return response.json({"ok": True})

That couples the order endpoint to the WebSocket layer. For anything beyond a couple of call sites, put sillo’s event emitter in between so the domain code emits and the transport subscribes.

decoupling via the event emitter

from sillo.events import AsyncEventEmitter
from sillo.websockets import ChannelBox

events = AsyncEventEmitter()


@events.on("order.shipped")
async def push_order_shipped(payload):
    await ChannelBox.group_send(
        f"order:{payload['id']}",
        {"type": "order.shipped", "data": payload},
    )


@events.on("order.shipped")
async def notify_by_email(payload):
    await mailer.send_shipping_notice(payload["email"])

Now the endpoint emits once and knows nothing about sockets or email:

await events.emit("order.shipped", {"id": order.id, "email": order.email})

Two properties of emit worth knowing. It is fire-and-forget — it schedules the listeners and returns a dict immediately rather than awaiting them, so a slow listener does not delay the HTTP response, and a failing listener does not fail the request. And listeners run in priority order, HIGHEST through LOWEST, defaulting to NORMAL:

@events.on("order.shipped", priority=EventPriority.HIGHEST)
async def audit_first(payload):
    await audit.record(payload)

Use emit_async when you do need to wait for listeners to finish before continuing. The full emitter API is covered in Events.

Naming inbound and outbound types consistently

Use the same vocabulary in both directions. If clients send chat.message, the broadcast that results should also be chat.message — the client’s rendering code then handles its own messages and other people’s identically, which is what you want, since group_send includes the sender anyway.

Keep server-only types clearly distinguishable: presence.join, order.shipped, system.reconnect. A client that receives an unknown type should ignore it rather than error, which is what makes adding new server events a non-breaking change.

Reconnection and missed messages

Connections drop. Phones change networks, laptops sleep, load balancers recycle, and deployments restart every worker. Any design that assumes a connection lasts is a design that loses messages.

Nothing in ChannelBox addresses this. A broadcast issued while a client is reconnecting is gone — it was written to a socket that no longer exists, and _send swallowed the error. Message history is per-process and, with the default manager, unbounded and prone to being wiped wholesale.

The pattern that works is a sequence number plus a catch-up query.

messages carry a monotonic cursor

@app.post("/rooms/{room_id}/messages")
async def post_message(request, response):
    body = await request.json
    message = await Message.create(room_id=..., text=body["text"])

    await ChannelBox.group_send(
        f"room:{message.room_id}",
        {"type": "chat.message", "seq": message.id, "data": {"text": message.text}},
    )
    return response.json({"id": message.id})

the client asks for what it missed

async def on_connect(self, websocket):
    await websocket.accept()
    await self.join_group(self.room)

    since = websocket.query_params.get("since")
    if since:
        missed = await Message.filter(
            room_id=self.room_id, id__gt=int(since)
        ).order_by("id").limit(500)
        for m in missed:
            await websocket.send_json(
                {"type": "chat.message", "seq": m.id, "data": {"text": m.text}}
            )

The client stores the highest seq it has seen and passes it as since on reconnect. The database is the source of truth; the socket is a low-latency optimisation over polling, not the record.

Two details matter. Join the group before running the catch-up query, so a message arriving mid-catch-up is delivered rather than falling into the gap between the query and the subscription — the client will occasionally see a duplicate, which is why messages carry an id it can deduplicate on. And cap the catch-up: a client returning after a week should get a bounded page and a “history truncated” marker, not thirty thousand frames.

This also means at-least-once delivery, not exactly-once. Design clients to be idempotent per seq and the whole class of problems disappears.

What not to do

Do not dispatch with a chain of if statements. It never gets refactored.

Do not share a handlers dict between consumer subclasses. Redeclare it per class.

Do not trust any field of an incoming message. There is no framework-level validation on a socket.

Do not check authorization only at connect. Long-lived connections outlive permissions.

Do not let an exception escape on_receive. It terminates the connection.

Do not skip a size limit. A single frame can be arbitrarily large.

Do not rely on the HTTP rate limiter. It does not see per-message traffic.

Do not do slow work in a handler. The connection’s receive loop is blocked for the duration; emit an event and return.

Do not assume emit waits. Use emit_async when ordering matters.

Ordering guarantees

Within one connection, WebSocket frames arrive in the order they were sent — TCP guarantees that, and neither sillo nor the ASGI server reorders them. So a client that sends A then B has them handled A then B.

Across connections there is no ordering at all. Two clients posting simultaneously reach the server in whatever order the event loop happens to schedule, and a broadcast triggered by the second can be written to a socket before the broadcast triggered by the first. If order matters — document edits, financial transactions, anything with a “last write wins” rule — assign the sequence server-side and let clients sort by it, rather than trusting arrival order.

Fan-out order is also not the order of a gather. The moment you parallelise sends for throughput, two clients in the same room can receive the same two messages in different orders. Sequence numbers fix this too, which is one more reason to add them early.

Performance notes

Dispatch through a dictionary is O(1) and irrelevant to your latency budget. What matters is what handlers do.

Anything slow inside on_receive stalls that connection’s entire receive loop — the client can send, but nothing is read until the handler returns. Push slow work to a background task or a job queue and reply immediately with an acknowledgement.

Validation is cheap; pydantic-core is Rust. Building a schema per message is not — construct them once at import time.

Broadcast fan-out is the real cost. One message from one client that triggers a broadcast to a thousand-member group is a thousand sends. At ten messages per second from fifty active clients, that is 500,000 sends per second, which is not achievable. Coalesce: batch typing indicators on a timer, aggregate presence changes, and send diffs rather than full state.

Related

• Consumers — the class these patterns extend
• Groups — group_send and its delivery semantics
• Channels — what a group contains
• WebSockets — connections, routing, and close codes
• Events — the full application event emitter API
• Rate Limiting — the HTTP limiter, and why it does not cover sockets