Groups

Groups

ChannelBox is a registry of named groups, each holding a set of channels. It exists so you can say “send this to everyone in room 42” without tracking connections yourself.

the shape of it

ChannelBox.CHANNEL_GROUPS == {
    "room:42": {<Channel a>: Ellipsis, <Channel b>: Ellipsis},
    "lobby":   {<Channel c>: Ellipsis},
}

Groups are dictionaries keyed by channel, with ... as a placeholder value — a dict used as an ordered set. Everything on ChannelBox is a classmethod operating on that one class-level dictionary.

When to reach for groups

Use them for fan-out to a set of connections that changes over time: chat rooms, document collaborators, subscribers to a live metric, everyone watching a build.

Do not use them as a connection registry keyed by user. send_to scans every group looking for a UUID, which is O(all connections) per message — keep your own {user_id: channels} mapping for direct messaging.

Do not use them at all if you run more than one worker process without a fan-out layer. That constraint is important enough to state before anything else.

Groups are per-process and are not shared

CHANNEL_GROUPS is a plain class attribute — a dictionary in one Python process’s memory. It is not shared between uvicorn workers, between gunicorn processes, or between machines.

With --workers 4, a client connected to worker 1 will never receive a broadcast issued from worker 3. Roughly three quarters of your messages vanish, and nothing raises. Development with a single worker looks perfect.

The same applies to show_history and to any HTTP endpoint that broadcasts: the endpoint runs on whichever worker the load balancer picked, and reaches only that worker’s connections.

If you need more than one process, put a message bus underneath — the Redis pattern at the end of this page is about sixty lines. Decide this before you ship, because every broadcast call site changes.

Joining and leaving

membership

from sillo.websockets import Channel, ChannelBox

channel = Channel(websocket=ws, payload_type="json")

await ChannelBox.add_channel_to_group(channel, "room:42")
await ChannelBox.remove_channel_from_group(channel, "room:42")

A channel can be in any number of groups at once. Groups are created on first join and deleted when their last member leaves — there is no explicit create or destroy.

The return values are not what they look like

add_channel_to_group returns a ChannelAddStatusEnum:

Return	What it actually means
CHANNEL_ADDED	The group did not exist and was created
CHANNEL_EXIST	The group already existed

Both cases add the channel. The status describes the group, not the channel, despite the names:

await ChannelBox.add_channel_to_group(a, "room")   # CHANNEL_ADDED
await ChannelBox.add_channel_to_group(b, "room")   # CHANNEL_EXIST — b was added
len(ChannelBox.CHANNEL_GROUPS["room"])             # 2

Do not branch on this value expecting it to tell you whether the channel was already present. It never does.

remove_channel_from_group raises UnboundLocalError

The function assigns its return status inside two if branches. When the channel is not in the named group and the group still has other members, neither branch executes and the bare return hits an unassigned local:

stray = Channel(websocket=ws, payload_type="json")
await ChannelBox.remove_channel_from_group(stray, "room")   # 'room' has other members
# UnboundLocalError: cannot access local variable 'channel_remove_status'
# where it is not associated with a value

This is not an exotic path. It fires when a disconnect handler runs twice, when cleanup runs after a connection that never finished joining, or when two code paths both try to tidy up.

Wrap every removal until this is fixed:

try:
    await ChannelBox.remove_channel_from_group(channel, group)
except UnboundLocalError:
    pass

The other three statuses do work: CHANNEL_REMOVED when a member leaves a group that still has others, GROUP_REMOVED when the last member leaves, and GROUP_DOES_NOT_EXIST when the group is not there at all.

Broadcasting

group_send

status = await ChannelBox.group_send(
    group_name="room:42",
    payload={"type": "message", "text": "hello"},
    save_history=False,
)

Return	Meaning
GROUP_SEND	The group had at least one channel and the loop ran
NO_SUCH_GROUP	The group was missing or empty

GROUP_SEND is not a delivery receipt. Channel._send catches RuntimeError — which is what a closed socket raises — logs it at DEBUG, and continues. A broadcast to a group of five hundred where four hundred have gone away still returns GROUP_SEND.

Three properties to internalise:

Sends are sequential. The loop awaits each channel in turn. One slow client delays everyone behind it. For large groups, gather instead:

concurrent fan-out with a timeout

import asyncio

channels = list(ChannelBox.CHANNEL_GROUPS.get(group, {}))
await asyncio.gather(
    *(asyncio.wait_for(c._send(payload), timeout=5) for c in channels),
    return_exceptions=True,
)

return_exceptions=True matters — without it, one timeout cancels the rest of the fan-out.

There is no sender exclusion. group_send has no exclude parameter. Filter by UUID if you need “everyone but me”.

The payload is shared. Every channel gets the same object, so payload type must be uniform across the group, and a mutable payload mutated by one recipient’s send path affects the others.

Broadcasting from an HTTP handler

Groups are global to the process, so any code can broadcast — a REST endpoint, a background task, a webhook receiver.

pushing from an ordinary route

@app.post("/rooms/{room_id}/announce")
async def announce(request, response):
    body = await request.json
    await ChannelBox.group_send(
        f"room:{request.path_params['room_id']}",
        {"type": "announcement", "text": body["text"]},
        save_history=True,
    )
    return response.json({"sent": True})

This is the pattern that makes WebSockets useful in an otherwise request/response application: the HTTP side writes, the WebSocket side notifies. It is also the pattern most affected by the multi-worker problem above — this endpoint reaches only the connections held by the worker that served the POST.

Message history

Passing save_history=True stores the payload before sending.

history

await ChannelBox.group_send("room:42", payload, save_history=True)

messages = await ChannelBox.show_history("room:42")   # list[ChannelMessageDC]
everything = await ChannelBox.show_history()          # dict[str, list[...]]

await ChannelBox.flush_history("room:42")
await ChannelBox.flush_history()                      # everything

Each stored message is a ChannelMessageDC with payload, a uuid, and a UTC created timestamp. The uuid and timestamp are per-message — generated by default_factory rather than evaluated once at import.

replaying history to a new joiner

async def on_connect(self, websocket):
    await websocket.accept()
    await self.join_group(self.room)
    for message in await ChannelBox.show_history(self.room):
        await websocket.send_json(message.payload)

The default history manager's size limit does not work

InMemoryHistoryManager checks sys.getsizeof(self._history[group]) against its limit. sys.getsizeof on a list returns the size of the list’s own pointer array — not the size of the objects it contains.

Fifty messages of 100 KB each — 5 MB of payloads — measure as 472 bytes. The 1 MiB default limit is never reached, and history grows without bound until the process runs out of memory.

When the limit does trip, at roughly 130,000 messages regardless of size, the handler clears the entire group’s history rather than dropping the oldest entries. Every replay for that room returns nothing.

Do not use the default manager with save_history=True in production. Either disable history entirely:

from sillo.websockets import ChannelBox, NoOpHistoryManager

ChannelBox.set_history_manager(NoOpHistoryManager())

or supply a bounded one:

from collections import deque

from sillo.websockets import BaseHistoryManager


class BoundedHistoryManager(BaseHistoryManager):
    def __init__(self, max_messages: int = 200):
        self.max_messages = max_messages
        self._history: dict[str, deque] = {}

    async def save_message(self, group_name, message):
        bucket = self._history.setdefault(
            group_name, deque(maxlen=self.max_messages)
        )
        bucket.append(message)

    async def get_history(self, group_name=None):
        if group_name:
            return list(self._history.get(group_name, ()))
        return {k: list(v) for k, v in self._history.items()}

    async def flush_history(self, group_name=None):
        if group_name:
            self._history.pop(group_name, None)
        else:
            self._history.clear()


ChannelBox.set_history_manager(BoundedHistoryManager(max_messages=200))

A deque with maxlen drops the oldest message on overflow, which is what “keep the last N” should mean.

BaseHistoryManager is a three-method ABC, so a Redis- or database-backed implementation is straightforward — and is what you want anyway once you have more than one process, since history has exactly the same per-process problem as groups.

Inspection and shutdown

operational helpers

groups = await ChannelBox.show_groups()      # the live dict — not a copy
await ChannelBox.flush_groups()              # drop all groups, sockets stay open
await ChannelBox.close_all_connections()     # close every socket, then clear

show_groups() returns the actual dictionary. Mutating what you get back mutates the registry; iterate over list(...) if you might trigger a disconnect while looping.

close_all_connections() belongs in a shutdown hook so clients get a clean close frame rather than a dropped TCP connection:

graceful shutdown

@app.on_shutdown
async def close_sockets():
    await ChannelBox.close_all_connections()

flush_groups() clears membership without closing anything, which leaves open sockets that belong to no group. It is useful between tests and rarely what you want at runtime.

Scaling out with Redis

The shape of the fix: keep local groups for delivery, and use pub/sub to tell every process what to deliver.

a minimal fan-out layer

import asyncio
import json

import redis.asyncio as redis

from sillo.websockets import ChannelBox

_redis = redis.from_url("redis://localhost")


async def publish(group: str, payload: dict) -> None:
    """Broadcast to every worker instead of only this one."""
    await _redis.publish("ws-fanout", json.dumps({"group": group, "payload": payload}))


async def fanout_listener() -> None:
    pubsub = _redis.pubsub()
    await pubsub.subscribe("ws-fanout")
    async for message in pubsub.listen():
        if message["type"] != "message":
            continue
        envelope = json.loads(message["data"])
        await ChannelBox.group_send(envelope["group"], envelope["payload"])


@app.on_startup
async def start_fanout():
    app.state["fanout"] = asyncio.create_task(fanout_listener())

Every process subscribes; every process delivers to its own local members. Call publish() instead of ChannelBox.group_send() from your application code — the listener is the only caller of group_send.

Two caveats. Each process receives every message regardless of whether it has members in that group, so at high volume you want per-group channels rather than one firehose. And pub/sub is at-most-once — a worker that is restarting misses messages entirely, which is why history belongs in a shared store rather than being reconstructed from the bus.

What not to do

Do not run multiple workers with in-memory groups. Most broadcasts will silently miss most clients.

Do not branch on CHANNEL_ADDED vs CHANNEL_EXIST. They describe the group, not the channel.

Do not call remove_channel_from_group unguarded. It raises UnboundLocalError on a realistic path.

Do not read GROUP_SEND as delivery. Dead sockets are skipped silently.

Do not use save_history=True with the default manager. Its size limit does not work and overflow wipes the group.

Do not mutate what show_groups() returns. It is the live registry.

Do not use send_to for direct messaging. It scans every connection.

Do not mix payload types in one group. A single payload goes to all members.

Performance notes

group_send is O(group size) and sequential. At a thousand members and one millisecond per send, a broadcast takes a second — during which that task is doing nothing else.

_clean_expired() sweeps every group and every channel, and runs on each remove_channel_from_group. High connection churn makes this the hot path; manage membership yourself if you are seeing it in profiles.

send_to iterates all groups and all channels. It is the single most expensive operation on ChannelBox and has an easy replacement — your own dictionary.

History with the default manager grows without bound. Measure resident memory on a long-running process before trusting it.

API reference

Method	Signature	Notes
add_channel_to_group	(channel, group_name="default")	Status describes the group
remove_channel_from_group	(channel, group_name)	Can raise UnboundLocalError
group_send	(group_name="default", payload={}, save_history=False)	Sequential; no sender exclusion
show_groups	() -> dict	The live registry
flush_groups	()	Clears membership; sockets stay open
show_history	(group_name="")	List for one group, dict for all
flush_history	(group_name=None)
set_history_manager	(manager: BaseHistoryManager)
close_all_connections	()	Closes sockets, then clears groups
CHANNEL_GROUPS	dict[str, dict[Channel, Ellipsis]]	Class attribute; process-local
HISTORY_SIZE	int	From CHANNEL_BOX_HISTORY_SIZE, default 1 MiB

Related

• Channels — what goes into a group
• Consumers — join_group, leave_group, broadcast
• WebSockets — connections and routing
• Events — dispatching by message type
• Startup & Shutdown — where close_all_connections belongs