From 5281cf171edb5b8699fba6c7c1300703ae627e1a Mon Sep 17 00:00:00 2001
From: axelray-dev <110029405+axelray-dev@users.noreply.github.com>
Date: Wed, 15 Jul 2026 18:49:22 +0800
Subject: [PATCH] fix(event): suppress native asyncio.QueueShutDown on Python
 3.13+

---
 .../event/processor/event_processor.py        | 12 +++++--
 .../event/processor/test_event_processor.py   | 31 +++++++++++++++++++
 2 files changed, 41 insertions(+), 2 deletions(-)

diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
index 7d9296fe4dd..a2eff1c64a7 100644
--- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
+++ b/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 (
diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py
index bcc1108be98..d1787c76f11 100644
--- a/tests/units/reflex_base/event/processor/test_event_processor.py
+++ b/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.