fix(event): suppress native asyncio.QueueShutDown on Python 3.13+

packages/reflex-base/src/reflex_base/event/processor/event_processor.py

@@ -35,11 +35,15 @@
     class QueueShutDown(asyncio.QueueShutDown):  # pyright: ignore[reportRedeclaration]
         """Exception raised when trying to put an item into a shut down queue."""
 
+    _QUEUE_SHUTDOWN_ERRORS: tuple[type[BaseException], ...] = (asyncio.QueueShutDown,)
+
 else:
 
     class QueueShutDown(Exception):  # noqa: N818
         """Exception raised when trying to put an item into a shut down queue."""
 
+    _QUEUE_SHUTDOWN_ERRORS = (QueueShutDown,)
+
 
 _StreamItemT = TypeVar("_StreamItemT")
 
@@ -299,7 +303,11 @@ async def stop(self, graceful_shutdown_timeout: float | None = None) -> None:
             self._queue_task.cancel()
             try:
                 await self._queue_task
-            except (asyncio.CancelledError, QueueShutDown, RuntimeError):
+            except (
+                asyncio.CancelledError,
+                RuntimeError,
+                *_QUEUE_SHUTDOWN_ERRORS,
+            ):
                 pass
             except Exception as ex:
                 telemetry.send_error(ex, context="backend")
@@ -642,7 +650,7 @@ async def _process_queue(self):
         if (queue := self._queue) is None:
             msg = "Event processor is not running, call .start(...) first."
             raise RuntimeError(msg)
-        with contextlib.suppress(QueueShutDown):
+        with contextlib.suppress(*_QUEUE_SHUTDOWN_ERRORS):
             while True:
                 entry = await queue.get()
                 if (

tests/units/reflex_base/event/processor/test_event_processor.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 import pytest
+from pytest_mock import MockerFixture
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor.event_processor import (
     EventProcessor,
@@ -226,6 +227,36 @@ async def test_stop_idempotent(processor: EventProcessor):
     await processor.stop()
 
 
+@pytest.mark.skipif(
+    not hasattr(asyncio, "QueueShutDown"),
+    reason="asyncio.Queue.shutdown() requires Python 3.13+",
+)
+async def test_native_queue_shutdown_is_suppressed(
+    processor: EventProcessor, mocker: MockerFixture
+):
+    """Native queue shutdown exceptions do not surface as processor errors.
+
+    Args:
+        processor: The event processor fixture.
+        mocker: The pytest mock fixture.
+    """
+    send_error = mocker.patch("reflex.utils.telemetry.send_error")
+    console_error = mocker.patch("reflex.utils.console.error")
+    queue: asyncio.Queue = asyncio.Queue()
+    queue.shutdown()
+
+    processor._queue = queue
+    await processor._process_queue()
+
+    processor._queue = None
+    processor._queue_task = asyncio.create_task(queue.get())
+    await asyncio.sleep(0)
+    await processor.stop()
+
+    send_error.assert_not_called()
+    console_error.assert_not_called()
+
+
 async def test_async_context_manager(processor: EventProcessor):
     """Entering/exiting via ``async with`` starts and stops the processor.