Bug(channels): single exception in backend.publish breaks all subsequent events publishing

[Vitaly312]

Description

If single exception(e.g. network failure) happens in the backend.publish, backend._pub_worker dies, causing ignoring all subsequent events that will be published + deadlock on shutdown(since the plugin tries to empty the queue before shutdown)

URL to code causing the issue

No response

MCVE

import asyncio
from collections.abc import Iterable
from unittest.mock import AsyncMock

from anyio import fail_after

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


class StubMemoryChannelBackend(MemoryChannelsBackend):
    def __init__(self, history: int = 0) -> None:
        super().__init__(history)
        self._counter = 0

    async def publish(self, data: bytes, channels: Iterable[str]) -> None:
        if self._counter == 0:
            raise RuntimeError("Network lag")
        self._counter += 1
        await super().publish(data, channels)


async def test_deadlock_on_shutdown():
    plugin = ChannelsPlugin(arbitrary_channels_allowed=True, backend=StubMemoryChannelBackend())
    with fail_after(1):
        async with plugin:
            plugin.publish("test", "test")
    # TimeoutError (during __aexit__ --> _on_shutdown)


async def test_single_exception_suppress_all_publishes():
    plugin = ChannelsPlugin(arbitrary_channels_allowed=True, backend=StubMemoryChannelBackend())
    async with plugin:
        plugin.publish("test", "test")
        sub = await plugin.subscribe("42")
        plugin.publish("42", "42")  # Stub backend does not break that, but ...
        try:
            await asyncio.wait_for(sub._queue.get(), timeout=1)  # TimeoutError
        except TimeoutError:
            print(
                "Backend available now but event isn't published "
                "because _pub_worker is broken since it doesn't handle exceptions"
            )
            plugin._on_shutdown = AsyncMock() # just to exit from context manager without the deadlock


asyncio.run(test_deadlock_on_shutdown())
asyncio.run(test_single_exception_suppress_all_publishes())

Steps to reproduce

No response

Screenshots

No response

Logs

Litestar Version

the main branch(v3)

Platform

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