Bug(channels): concurrent subscribe/unsubscribe race condition causes silent event loss

[Vitaly312]

Description

Concurrent subscribing and unsubscribing of the last subscriber for the same channel leads to unsubscribing the plugin from the backend for this channel and events are suppressed for all subscribers.

URL to code causing the issue

No response

MCVE

import asyncio
from asyncio import Event
from collections.abc import Iterable

from anyio import fail_after

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


class StubMemoryChannelBackend(MemoryChannelsBackend):
    async def unsubscribe(self, channels: Iterable[str]) -> None:
        network_lag.set()
        await second_start_subscription.wait()
        return await super().unsubscribe(channels)


network_lag = Event()
second_start_subscription = Event()


async def test_phantom_channel():
    channels = ChannelsPlugin(arbitrary_channels_allowed=True, backend=StubMemoryChannelBackend())

    async def concurrent_subscribe():
        await network_lag.wait()
        second_start_subscription.set()
        subscriber = await channels.subscribe("42")
        await channels.wait_published("42", "42")
        try:
            with fail_after(0.1):
                await anext(subscriber.iter_events())
        except TimeoutError as e:
            raise ValueError("Expected event not received") from e

    async with channels:
        first_subscriber = await channels.subscribe("42")
        await asyncio.gather(concurrent_subscribe(), channels.unsubscribe(first_subscriber, "42"))


asyncio.run(test_phantom_channel())  # ValueError: Expected event not received

Steps to reproduce

No response

Screenshots

No response

Logs

Litestar Version

litestar: main branch
python: 3.12

Platform

• Linux
• Mac
• Windows
• Other (Please specify in the description above)

[isheng-eqi]

Submitted PR #4899 — adds an asyncio.Lock to protect the critical sections in subscribe() and unsubscribe(), making channel state checks + backend calls atomic.