ChannelsPlugin.subscribe is not cancellation-safe: a subscriber is leaked into _channels if the request is cancelled mid-subscribe

[lesnik512]

Summary

ChannelsPlugin.subscribe registers the new subscriber into self._channels synchronously, before it awaits. If the calling task is cancelled during one of those awaits (most importantly the history fetch), subscribe raises CancelledError, but the subscriber is already registered in _channels — and because subscribe never returned, the caller has no reference to it and can never call unsubscribe. The subscriber is leaked permanently into _channels, and the backend stays subscribed to its channel.

The relevant ordering in ChannelsPlugin.subscribe:

# litestar/channels/plugin.py
subscriber = self._subscriber_class(...)
for channel in channels:
    ...
    channel_subscribers.add(subscriber)          # (1) synchronous register, BEFORE any await
if channels_to_subscribe:
    await self._backend.subscribe(channels_to_subscribe)   # (2)
if history:
    await self.put_subscriber_history(            # (3) real I/O (e.g. redis xrevrange) — cancellation window
        subscriber=subscriber, limit=history, channels=channels
    )
return subscriber

A cancellation at (2) or (3) leaves the subscriber registered at (1) but unreachable by the caller.

ChannelsPlugin.start_subscription has the same flaw: it does subscriber = await self.subscribe(...) and only enters its try/finally (which unsubscribes) after that returns. A cancellation inside subscribe skips the finally, so the context manager leaks too.

Why this matters in practice

It surfaces with SSE/long-lived subscriptions served by an ASGI server that cancels (or drops) the request task when the client disconnects. A client that connects and disconnects while subscribe is still fetching history leaks a subscriber on every occurrence. Under bursty connect/disconnect (e.g. reconnect storms), _channels and the backend's subscribed-channel set grow without bound — observed as steadily rising process memory and Redis/KeyDB CPU (the stream backend keeps XREAD-scanning the orphaned channels), cleared only by a restart.

MRE

Self-contained, no external services. The in-memory backend's get_history is normally synchronous, so it is subclassed to suspend — standing in for any backend whose history fetch awaits real I/O (e.g. RedisChannelsStreamBackend.get_history → xrevrange). The leak itself is in ChannelsPlugin, not the backend.

import asyncio

from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend


class SlowHistoryBackend(MemoryChannelsBackend):
    """Stand-in for any backend whose get_history suspends (e.g. redis xrevrange)."""

    def __init__(self, history: int = 1) -> None:
        super().__init__(history=history)
        self.in_history = asyncio.Event()
        self.release = asyncio.Event()

    async def get_history(self, channel: str, limit: int | None = None) -> list[bytes]:
        self.in_history.set()
        await self.release.wait()           # suspend, as a real backend would on I/O
        return await super().get_history(channel, limit)


async def main() -> None:
    backend = SlowHistoryBackend(history=1)
    plugin = ChannelsPlugin(backend=backend, arbitrary_channels_allowed=True)
    channel = "user:1"

    task = asyncio.create_task(plugin.subscribe([channel], history=1))
    await backend.in_history.wait()                     # subscribe is now suspended in get_history

    registered_before = len(plugin._channels.get(channel, set()))

    task.cancel()                                       # client disconnects mid-subscribe
    await asyncio.gather(task, return_exceptions=True)

    leaked = len(plugin._channels.get(channel, set()))

    print(f"registered while subscribe was still awaiting : {registered_before}")
    print(f"subscribers left in _channels after cancel    : {leaked}")
    print(f"backend still subscribed to channel           : {channel in backend._channels}")
    assert leaked == 0, "subscriber leaked into _channels, unreachable for unsubscribe"


asyncio.run(main())

Output:

registered while subscribe was still awaiting : 1
subscribers left in _channels after cancel    : 1
backend still subscribed to channel           : True
AssertionError: subscriber leaked into _channels, unreachable for unsubscribe

Expected vs actual

• Expected: a subscribe cancelled before it returns leaves no registered subscriber (nothing the caller can clean up should be left behind).
• Actual: the subscriber stays in _channels (and the backend stays subscribed) forever.

Proposed fix

Make subscribe cancellation-safe by registering only after the awaits. The history fetch is
side-effect-free (it just reads and fills the subscriber's own queue), and backend.subscribe
performs no real suspension for the stream backends, so the registration can be moved to the end:

subscriber = self._subscriber_class(...)
if history:
    await self.put_subscriber_history(subscriber=subscriber, limit=history, channels=channels)
channels_to_subscribe = set()
for channel in channels:
    ...
    channel_subscribers.add(subscriber)   # register after the only real await
if channels_to_subscribe:
    await self._backend.subscribe(channels_to_subscribe)
return subscriber

A drop during the history fetch now simply discards the unregistered subscriber. (Trade-off: events published in the brief window between the history snapshot and registration are not captured; this is comparable to the existing history-vs-live ordering races.) An alternative is to keep the current order but guard it so a cancellation between registration and return triggers unsubscribe. start_subscription should be fixed to match.

Happy to open a PR.

Environment

• Litestar 2.21.1
• Python 3.13 / 3.14

[isheng-eqi]

Submitted PR #4899 — reorders subscribe() so history fetch happens before registration, and backend.subscribe() before adding subscriber to channel sets. Cancellation during awaits now safely discards the unregistered subscriber.