From 17c3a2fb25c2bd04d978824377ca9713becc975e Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:27:39 -0400
Subject: [PATCH 1/9] Create +drain-oversized-polling-post.bugfix.md

---
 news/+drain-oversized-polling-post.bugfix.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 news/+drain-oversized-polling-post.bugfix.md

diff --git a/news/+drain-oversized-polling-post.bugfix.md b/news/+drain-oversized-polling-post.bugfix.md
new file mode 100644
index 00000000000..e129df6a050
--- /dev/null
+++ b/news/+drain-oversized-polling-post.bugfix.md
@@ -0,0 +1 @@
+Fixed a polling-transport stall where a page whose connection event exceeded `REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE` never hydrated and never retried: the backend now drains the oversized request body before rejecting it, so the client finishes sending and can reconnect.

From a68625f0917b5f5c17bf8c7ffe04d5f77923ef4b Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:28:16 -0400
Subject: [PATCH 2/9] Update app.py

---
 reflex/app.py | 78 ++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 77 insertions(+), 1 deletion(-)

diff --git a/reflex/app.py b/reflex/app.py
index 12837ff60a2..b474a91ad9e 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -368,6 +368,76 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         await self.app(scope, receive, send)
 
 
+class _DrainOversizedBodyMiddleware:
+    """Drain oversized polling POST bodies before engine.io rejects the request.
+
+    engine.io compares ``Content-Length`` against ``max_http_buffer_size`` and
+    answers 400 without reading the request body. The client is still sending
+    at that point, and the unsent remainder can wedge its connection pool so
+    the reconnect handshake is never served. Reading the body to completion
+    first lets the rejection reach a client that has finished sending, so it
+    can record the failure and retry.
+    """
+
+    def __init__(self, app: ASGIApp, max_size: int) -> None:
+        """Wrap an ASGI app, draining oversized POST bodies before forwarding.
+
+        Args:
+            app: The next ASGI app in the middleware stack.
+            max_size: The maximum accepted body size; larger POST bodies are drained.
+        """
+        self.app = app
+        self.max_size = max_size
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """Drain an oversized POST body, then forward the request downstream.
+
+        Args:
+            scope: The ASGI connection scope.
+            receive: The ASGI receive callable.
+            send: The ASGI send callable.
+        """
+        if scope["type"] != "http" or scope["method"] != "POST":
+            await self.app(scope, receive, send)
+            return
+
+        content_length: int | None = None
+        for name, value in scope.get("headers", []):
+            if name.lower() == b"content-length":
+                try:
+                    content_length = int(value)
+                except ValueError:
+                    content_length = None
+                break
+
+        if content_length is None or content_length <= self.max_size:
+            await self.app(scope, receive, send)
+            return
+
+        # Drain the declared body so the client finishes sending before
+        # engine.io answers 400. The bytes are discarded; engine.io rejects on
+        # Content-Length without reading them.
+        remaining = content_length
+        while remaining > 0:
+            message = await receive()
+            if message["type"] != "http.request":
+                break
+            remaining -= len(message.get("body", b""))
+            if not message.get("more_body"):
+                break
+
+        drained = False
+
+        async def drained_receive() -> Message:
+            nonlocal drained
+            if not drained:
+                drained = True
+                return {"type": "http.request", "body": b"", "more_body": False}
+            return await receive()
+
+        await self.app(scope, drained_receive, send)
+
+
 def _is_location_specifier(specifier: str) -> bool:
     """Check whether a dependency specifier points at a location.
 
@@ -655,6 +725,12 @@ def _setup_state(self) -> None:
 
         # Create the socket app. Note event endpoint constant replaces the default 'socket.io' path.
         socket_app = EngineIOApp(self.sio, socketio_path="")
+        # Drain oversized polling POST bodies before engine.io rejects them, so
+        # the 400 response does not strand a half-sent request on the client.
+        drained_socket_app = _DrainOversizedBodyMiddleware(
+            socket_app,
+            max_size=self.sio.eio.max_http_buffer_size,
+        )
         namespace = config.get_event_namespace()
 
         # Create the event namespace and attach the main app. Not related to any paths.
@@ -690,7 +766,7 @@ async def modified_send(message: Message):
 
                     return await self.app(scope, receive, modified_send)
 
-            socket_app_with_headers = HeaderMiddleware(socket_app)
+            socket_app_with_headers = HeaderMiddleware(drained_socket_app)
             self._api.mount(
                 config.prepend_backend_path(str(constants.Endpoint.EVENT)),
                 socket_app_with_headers,

From e922e3ea973817285343bdad1bef4d4644361c63 Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:28:41 -0400
Subject: [PATCH 3/9] Update test_app.py

---
 tests/units/test_app.py | 179 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 179 insertions(+)

diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index feda7d3b7ef..99fdf977cc5 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -67,6 +67,7 @@
     EventNamespace,
     _ContextMiddleware,
     _decode_asgi_headers,
+    _DrainOversizedBodyMiddleware,
     _sio_dumps,
     _sio_loads,
     default_overlay_component,
@@ -5695,3 +5696,181 @@ def reader():
         for future in readers:
             future.result()
     assert json.loads(marker.read_text()) == routes
+
+
+def _polling_engine_app(max_size: int):
+    """Build the socket ASGI app the way ``App._setup_state`` mounts it.
+
+    Args:
+        max_size: Value for engine.io's ``max_http_buffer_size``.
+
+    Returns:
+        A socket.io ASGI app wrapping a fresh engine.io server.
+    """
+    from socketio import ASGIApp as EngineIOApp
+    from socketio import AsyncServer
+
+    sio = AsyncServer(
+        async_mode="asgi",
+        max_http_buffer_size=max_size,
+        async_handlers=False,
+    )
+    return EngineIOApp(sio, socketio_path="")
+
+
+async def _asgi_http_call(
+    app, method: str, query_string: str, body: bytes = b"", chunk_size: int = 65536
+):
+    """Drive one HTTP request through an ASGI app with a chunked body.
+
+    Args:
+        app: The ASGI app to call.
+        method: The HTTP method.
+        query_string: The raw query string.
+        body: The request body.
+        chunk_size: Size of each http.request message body chunk.
+
+    Returns:
+        A tuple of (status, response body, request bytes consumed from receive).
+    """
+    chunks = [body[i : i + chunk_size] for i in range(0, len(body), chunk_size)] or [
+        b""
+    ]
+    sent = []
+    consumed = 0
+    index = 0
+
+    async def receive():  # noqa: RUF029
+        nonlocal consumed, index
+        if index < len(chunks):
+            chunk = chunks[index]
+            index += 1
+            consumed += len(chunk)
+            return {
+                "type": "http.request",
+                "body": chunk,
+                "more_body": index < len(chunks),
+            }
+        return {"type": "http.disconnect"}
+
+    async def send(message):  # noqa: RUF029
+        sent.append(message)
+
+    headers = []
+    if method == "POST":
+        headers.append((b"content-length", str(len(body)).encode()))
+    scope = {
+        "type": "http",
+        "asgi": {"version": "3.0"},
+        "http_version": "1.1",
+        "method": method,
+        "scheme": "http",
+        "path": "/",
+        "raw_path": b"/",
+        "query_string": query_string.encode(),
+        "root_path": "",
+        "headers": headers,
+        "client": ("127.0.0.1", 12345),
+        "server": ("127.0.0.1", 8000),
+    }
+    await app(scope, receive, send)
+    status = next(m["status"] for m in sent if m["type"] == "http.response.start")
+    response_body = b"".join(
+        m.get("body", b"") for m in sent if m["type"] == "http.response.body"
+    )
+    return status, response_body, consumed
+
+
+async def test_engineio_rejects_oversized_polling_post_without_draining_body():
+    """Document the engine.io behavior _DrainOversizedBodyMiddleware works around.
+
+    engine.io rejects a polling POST larger than max_http_buffer_size with 400
+    without reading the request body, stranding the client mid-send.
+    """
+    app = _polling_engine_app(max_size=100)
+    status, response_body, _ = await _asgi_http_call(
+        app, "GET", "EIO=4&transport=polling"
+    )
+    assert status == 200
+    sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
+    assert sid is not None
+
+    # A real polling client always has a poll in flight; park one so the
+    # server can deliver its CLOSE packet when it rejects the POST.
+    poll_task = asyncio.create_task(
+        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}")
+    )
+    await asyncio.sleep(0.1)
+
+    big_body = b"x" * 1000
+    status, _, consumed = await asyncio.wait_for(
+        _asgi_http_call(
+            app,
+            "POST",
+            f"EIO=4&transport=polling&sid={sid.group(1)}",
+            big_body,
+            chunk_size=100,
+        ),
+        timeout=10,
+    )
+    poll_status, _, _ = await asyncio.wait_for(poll_task, timeout=5)
+    assert status == 400
+    assert poll_status == 200
+    assert consumed < len(big_body)
+
+
+async def test_drain_oversized_body_middleware_drains_before_engineio_rejects():
+    """Oversized polling POST bodies are drained so the client can finish sending."""
+    app = _DrainOversizedBodyMiddleware(_polling_engine_app(max_size=100), max_size=100)
+    status, response_body, _ = await _asgi_http_call(
+        app, "GET", "EIO=4&transport=polling"
+    )
+    assert status == 200
+    sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
+    assert sid is not None
+
+    # A real polling client always has a poll in flight; park one so the
+    # server can deliver its CLOSE packet when it rejects the POST.
+    poll_task = asyncio.create_task(
+        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}")
+    )
+    await asyncio.sleep(0.1)
+
+    big_body = b"x" * 1000
+    status, _, consumed = await asyncio.wait_for(
+        _asgi_http_call(
+            app,
+            "POST",
+            f"EIO=4&transport=polling&sid={sid.group(1)}",
+            big_body,
+            chunk_size=100,
+        ),
+        timeout=10,
+    )
+    poll_status, _, _ = await asyncio.wait_for(poll_task, timeout=5)
+    assert status == 400
+    assert poll_status == 200
+    assert consumed == len(big_body)
+
+    # The reconnect handshake is served normally after the rejected POST.
+    status, _, _ = await _asgi_http_call(app, "GET", "EIO=4&transport=polling")
+    assert status == 200
+
+
+async def test_drain_oversized_body_middleware_passes_regular_requests_through():
+    """Requests within the size limit reach the downstream app untouched."""
+    received = []
+
+    async def echo_app(scope, receive, send):
+        received.append(await receive())
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+    app = _DrainOversizedBodyMiddleware(echo_app, max_size=100)
+    status, response_body, _ = await _asgi_http_call(app, "POST", "", b"hello")
+    assert status == 200
+    assert response_body == b"ok"
+    assert received[0]["body"] == b"hello"
+
+    status, _, _ = await _asgi_http_call(app, "GET", "", b"")
+    assert status == 200

From 6ec7af77d62e6e5f707bb246b1e7f64b35f45f7a Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:56:59 -0400
Subject: [PATCH 4/9] Update app.py

---
 reflex/app.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/reflex/app.py b/reflex/app.py
index b474a91ad9e..dbb5b957ecc 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -429,6 +429,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         drained = False
 
         async def drained_receive() -> Message:
+            """Replay the drained request as a single empty body, then delegate.
+
+            Returns:
+                A one-time empty http.request message, then downstream messages.
+            """
             nonlocal drained
             if not drained:
                 drained = True

From e108c5b6c7adf9125ea6733d9c7025b42d997fde Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:57:24 -0400
Subject: [PATCH 5/9] Update test_app.py

---
 tests/units/test_app.py | 22 +++++++++++++++++++---
 1 file changed, 19 insertions(+), 3 deletions(-)

diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 99fdf977cc5..bddfc193313 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -50,6 +50,8 @@
 from reflex_components_core.core.upload import selected_files
 from reflex_components_radix.themes.typography.text import Text
 from reflex_otel import ReflexInstrumentor
+from socketio import ASGIApp as EngineIOApp
+from socketio import AsyncServer
 from starlette.applications import Starlette
 from starlette.datastructures import FormData, Headers, UploadFile
 from starlette.exceptions import HTTPException
@@ -5707,9 +5709,6 @@ def _polling_engine_app(max_size: int):
     Returns:
         A socket.io ASGI app wrapping a fresh engine.io server.
     """
-    from socketio import ASGIApp as EngineIOApp
-    from socketio import AsyncServer
-
     sio = AsyncServer(
         async_mode="asgi",
         max_http_buffer_size=max_size,
@@ -5741,6 +5740,11 @@ async def _asgi_http_call(
     index = 0
 
     async def receive():  # noqa: RUF029
+        """Yield the request body one chunk at a time, then disconnect.
+
+        Returns:
+            http.request messages for each chunk, then http.disconnect.
+        """
         nonlocal consumed, index
         if index < len(chunks):
             chunk = chunks[index]
@@ -5754,6 +5758,11 @@ async def receive():  # noqa: RUF029
         return {"type": "http.disconnect"}
 
     async def send(message):  # noqa: RUF029
+        """Record an outgoing ASGI message.
+
+        Args:
+            message: The ASGI message sent by the app.
+        """
         sent.append(message)
 
     headers = []
@@ -5862,6 +5871,13 @@ async def test_drain_oversized_body_middleware_passes_regular_requests_through()
     received = []
 
     async def echo_app(scope, receive, send):
+        """Echo app that records the first request body and replies 200.
+
+        Args:
+            scope: The ASGI connection scope.
+            receive: The ASGI receive callable.
+            send: The ASGI send callable.
+        """
         received.append(await receive())
         await send({"type": "http.response.start", "status": 200, "headers": []})
         await send({"type": "http.response.body", "body": b"ok"})

From c7f7f1aa6d561c499ffa1d1e203502e28cee6825 Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Thu, 22 Oct 2026 21:16:42 -0400
Subject: [PATCH 6/9] Answer oversized polling POSTs from the drain middleware

After the drain, answer the 400 from _DrainOversizedBodyMiddleware itself
and close the engine.io session with socket.close(wait=False) instead of
delegating to engine.io. engine.io rejects by disconnecting the session,
whose close(wait=True) blocks in queue.join() until a poll GET consumes
the CLOSE packet; the engine.io client races the CONNECT POST against the
first poll GET, so with no poll parked the POST response was held forever
and later polls got 'Session is disconnected'. The CLOSE packet still
reaches a parked poll, but a missing poll no longer holds the response.

Add a no-parked-poll regression test under asyncio.wait_for; it times out
on the unpatched middleware and passes with the fix.
---
 reflex/app.py           | 69 ++++++++++++++++++++++++++++-------------
 tests/units/test_app.py | 52 ++++++++++++++++++++++++++++---
 2 files changed, 95 insertions(+), 26 deletions(-)

diff --git a/reflex/app.py b/reflex/app.py
index dbb5b957ecc..4d01809384f 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -33,6 +33,7 @@
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, overload
 
+import engineio
 from reflex_base import constants, otel
 from reflex_base.components.component import Component, ComponentStyle
 from reflex_base.config import get_config, reload_config
@@ -369,7 +370,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
 
 class _DrainOversizedBodyMiddleware:
-    """Drain oversized polling POST bodies before engine.io rejects the request.
+    """Drain oversized polling POST bodies, then reject them with a 400.
 
     engine.io compares ``Content-Length`` against ``max_http_buffer_size`` and
     answers 400 without reading the request body. The client is still sending
@@ -377,17 +378,32 @@ class _DrainOversizedBodyMiddleware:
     the reconnect handshake is never served. Reading the body to completion
     first lets the rejection reach a client that has finished sending, so it
     can record the failure and retry.
+
+    The rejection is answered by this middleware rather than engine.io because
+    engine.io rejects by disconnecting the session, and its disconnect waits
+    for a poll GET to consume the session's CLOSE packet before the POST
+    response is sent. When no poll is parked - common, since the engine.io
+    client sends the CONNECT POST and the first poll GET at the same time -
+    the POST would never be answered. The middleware answers 400 itself and
+    closes the session without waiting for the send queue.
     """
 
-    def __init__(self, app: ASGIApp, max_size: int) -> None:
-        """Wrap an ASGI app, draining oversized POST bodies before forwarding.
+    def __init__(
+        self, app: ASGIApp, max_size: int, eio: engineio.AsyncServer | None = None
+    ) -> None:
+        """Wrap an ASGI app, rejecting oversized POST bodies after draining them.
 
         Args:
             app: The next ASGI app in the middleware stack.
-            max_size: The maximum accepted body size; larger POST bodies are drained.
+            max_size: The maximum accepted body size; larger POST bodies are drained
+                and rejected.
+            eio: The engine.io server, used to close the rejected session without
+                waiting for its send queue. Optional; when omitted the session is
+                left to engine.io's own cleanup.
         """
         self.app = app
         self.max_size = max_size
+        self.eio = eio
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Drain an oversized POST body, then forward the request downstream.
@@ -414,9 +430,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             await self.app(scope, receive, send)
             return
 
-        # Drain the declared body so the client finishes sending before
-        # engine.io answers 400. The bytes are discarded; engine.io rejects on
-        # Content-Length without reading them.
+        # Drain the declared body so the client finishes sending before the
+        # rejection reaches it. The bytes are discarded.
         remaining = content_length
         while remaining > 0:
             message = await receive()
@@ -426,21 +441,32 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             if not message.get("more_body"):
                 break
 
-        drained = False
-
-        async def drained_receive() -> Message:
-            """Replay the drained request as a single empty body, then delegate.
-
-            Returns:
-                A one-time empty http.request message, then downstream messages.
-            """
-            nonlocal drained
-            if not drained:
-                drained = True
-                return {"type": "http.request", "body": b"", "more_body": False}
-            return await receive()
+        # Answer the oversized POST from here instead of delegating to
+        # engine.io. engine.io rejects on Content-Length by disconnecting the
+        # session, and disconnect waits in queue.join() for a poll GET to
+        # consume the CLOSE packet before the POST response is sent. The
+        # engine.io client sends the CONNECT POST and the first poll GET at
+        # the same time, so no poll may be parked when the POST arrives, and
+        # its response would be held forever. Answer directly, then close the
+        # session without waiting: a parked poll still receives the CLOSE
+        # packet, but no poll is required for this response.
+        await send({
+            "type": "http.response.start",
+            "status": 400,
+            "headers": [(b"content-type", b"text/plain")],
+        })
+        await send({"type": "http.response.body", "body": b'"Bad Request"'})
 
-        await self.app(scope, drained_receive, send)
+        if self.eio is not None:
+            sid = urllib.parse.parse_qs(
+                scope.get("query_string", b"").decode("latin-1")
+            ).get("sid", [None])[0]
+            socket = self.eio.sockets.get(sid) if sid else None
+            if socket is not None:
+                with contextlib.suppress(Exception):
+                    # Best-effort; the 400 above is already sent.
+                    await socket.close(wait=False)
+        return
 
 
 def _is_location_specifier(specifier: str) -> bool:
@@ -735,6 +761,7 @@ def _setup_state(self) -> None:
         drained_socket_app = _DrainOversizedBodyMiddleware(
             socket_app,
             max_size=self.sio.eio.max_http_buffer_size,
+            eio=self.sio.eio,
         )
         namespace = config.get_event_namespace()
 
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index bddfc193313..00c8b90bbbd 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -5707,14 +5707,15 @@ def _polling_engine_app(max_size: int):
         max_size: Value for engine.io's ``max_http_buffer_size``.
 
     Returns:
-        A socket.io ASGI app wrapping a fresh engine.io server.
+        A (socket.io ASGI app, engine.io server) pair for a fresh engine.io
+        server.
     """
     sio = AsyncServer(
         async_mode="asgi",
         max_http_buffer_size=max_size,
         async_handlers=False,
     )
-    return EngineIOApp(sio, socketio_path="")
+    return EngineIOApp(sio, socketio_path=""), sio.eio
 
 
 async def _asgi_http_call(
@@ -5796,7 +5797,7 @@ async def test_engineio_rejects_oversized_polling_post_without_draining_body():
     engine.io rejects a polling POST larger than max_http_buffer_size with 400
     without reading the request body, stranding the client mid-send.
     """
-    app = _polling_engine_app(max_size=100)
+    app, _ = _polling_engine_app(max_size=100)
     status, response_body, _ = await _asgi_http_call(
         app, "GET", "EIO=4&transport=polling"
     )
@@ -5828,9 +5829,10 @@ async def test_engineio_rejects_oversized_polling_post_without_draining_body():
     assert consumed < len(big_body)
 
 
-async def test_drain_oversized_body_middleware_drains_before_engineio_rejects():
+async def test_drain_oversized_body_middleware_drains_and_rejects_oversized_post():
     """Oversized polling POST bodies are drained so the client can finish sending."""
-    app = _DrainOversizedBodyMiddleware(_polling_engine_app(max_size=100), max_size=100)
+    socket_app, eio = _polling_engine_app(max_size=100)
+    app = _DrainOversizedBodyMiddleware(socket_app, max_size=100, eio=eio)
     status, response_body, _ = await _asgi_http_call(
         app, "GET", "EIO=4&transport=polling"
     )
@@ -5866,6 +5868,46 @@ async def test_drain_oversized_body_middleware_drains_before_engineio_rejects():
     assert status == 200
 
 
+async def test_drain_oversized_body_middleware_answers_post_without_parked_poll():
+    """The oversized POST is answered even when no poll is in flight.
+
+    The engine.io client sends the CONNECT POST and the first poll GET at the
+    same time, so the POST can be rejected before any poll is parked. The
+    rejection must not wait for a poll to consume the session CLOSE packet.
+    """
+    socket_app, eio = _polling_engine_app(max_size=100)
+    app = _DrainOversizedBodyMiddleware(socket_app, max_size=100, eio=eio)
+    status, response_body, _ = await _asgi_http_call(
+        app, "GET", "EIO=4&transport=polling"
+    )
+    assert status == 200
+    sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
+    assert sid is not None
+
+    # No poll is parked: the engine.io client races the CONNECT POST against
+    # the first poll GET, so the POST can arrive first.
+    big_body = b"x" * 1000
+    status, _, consumed = await asyncio.wait_for(
+        _asgi_http_call(
+            app,
+            "POST",
+            f"EIO=4&transport=polling&sid={sid.group(1)}",
+            big_body,
+            chunk_size=100,
+        ),
+        timeout=10,
+    )
+    assert status == 400
+    assert consumed == len(big_body)
+
+    # A later poll learns the session is gone instead of the POST hanging.
+    poll_status, _, _ = await asyncio.wait_for(
+        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}"),
+        timeout=5,
+    )
+    assert poll_status == 400
+
+
 async def test_drain_oversized_body_middleware_passes_regular_requests_through():
     """Requests within the size limit reach the downstream app untouched."""
     received = []

From 762788c748e32f9f8b9ecd0d62c849a355953da8 Mon Sep 17 00:00:00 2001
From: Anish Gurjar <50738829+anishgurjar@users.noreply.github.com>
Date: Thu, 22 Oct 2026 21:22:41 -0400
Subject: [PATCH 7/9] Correct eio docstring, assert 400 body in no-poll test

---
 reflex/app.py           | 5 +++--
 tests/units/test_app.py | 3 ++-
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/reflex/app.py b/reflex/app.py
index 4d01809384f..5ce0f5ab11d 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -398,8 +398,9 @@ def __init__(
             max_size: The maximum accepted body size; larger POST bodies are drained
                 and rejected.
             eio: The engine.io server, used to close the rejected session without
-                waiting for its send queue. Optional; when omitted the session is
-                left to engine.io's own cleanup.
+                waiting for its send queue. Optional; the rejected request never
+                reaches engine.io, so when omitted the session is left open and
+                expires on engine.io's own timeout.
         """
         self.app = app
         self.max_size = max_size
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 00c8b90bbbd..03699e36bb5 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -5887,7 +5887,7 @@ async def test_drain_oversized_body_middleware_answers_post_without_parked_poll(
     # No poll is parked: the engine.io client races the CONNECT POST against
     # the first poll GET, so the POST can arrive first.
     big_body = b"x" * 1000
-    status, _, consumed = await asyncio.wait_for(
+    status, response_body, consumed = await asyncio.wait_for(
         _asgi_http_call(
             app,
             "POST",
@@ -5898,6 +5898,7 @@ async def test_drain_oversized_body_middleware_answers_post_without_parked_poll(
         timeout=10,
     )
     assert status == 400
+    assert response_body == b'"Bad Request"'
     assert consumed == len(big_body)
 
     # A later poll learns the session is gone instead of the POST hanging.

From 21918f7c1b07247afbd09f35ab20c60eb56d3e18 Mon Sep 17 00:00:00 2001
From: Farhan <www.mfarvirus@gmail.com>
Date: Tue, 27 Oct 2026 22:44:24 +0500
Subject: [PATCH 8/9] Simplify oversized POST drain middleware and test its
 mount

- Take the engine.io server as a required argument and read its buffer
  limit, instead of a separate max_size and an optional eio.
- Drain with a single more_body loop, read Content-Length through
  starlette Headers, and drop the ValueError and broad exception guards
  (servers validate Content-Length; engine.io logs handler errors).
- Import engineio only for type checking.
- Extract the handshake and oversized POST test helpers, and add a test
  that drives an oversized POST through the App's event endpoint.
- Describe the remaining user-facing case in the news fragment: #7551
  stops oversized hydrate packets in the client, so the stall now comes
  from oversized events.
---
 news/+drain-oversized-polling-post.bugfix.md |   2 +-
 reflex/app.py                                | 106 +++----------
 tests/units/test_app.py                      | 155 ++++++++++---------
 3 files changed, 105 insertions(+), 158 deletions(-)

diff --git a/news/+drain-oversized-polling-post.bugfix.md b/news/+drain-oversized-polling-post.bugfix.md
index e129df6a050..12ddebbaea2 100644
--- a/news/+drain-oversized-polling-post.bugfix.md
+++ b/news/+drain-oversized-polling-post.bugfix.md
@@ -1 +1 @@
-Fixed a polling-transport stall where a page whose connection event exceeded `REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE` never hydrated and never retried: the backend now drains the oversized request body before rejecting it, so the client finishes sending and can reconnect.
+Fixed a polling-transport stall where an event larger than `REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE` left the page unable to reconnect: the backend now reads the oversized request body before it rejects it, so the browser reconnects at once.
diff --git a/reflex/app.py b/reflex/app.py
index 5ce0f5ab11d..022a8082be0 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -33,7 +33,6 @@
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, overload
 
-import engineio
 from reflex_base import constants, otel
 from reflex_base.components.component import Component, ComponentStyle
 from reflex_base.config import get_config, reload_config
@@ -75,6 +74,7 @@
 from socketio import ASGIApp as EngineIOApp
 from socketio import AsyncNamespace, AsyncServer
 from starlette.applications import Starlette
+from starlette.datastructures import Headers
 from starlette.middleware import cors
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
@@ -122,6 +122,7 @@
     from warnings import deprecated
 
 if TYPE_CHECKING:
+    from engineio import AsyncServer as AsyncEngineIOServer
     from reflex_base.plugins import Plugin
     from reflex_base.plugins.base import AddPageProtocol
     from reflex_base.vars import Var
@@ -370,104 +371,51 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
 
 class _DrainOversizedBodyMiddleware:
-    """Drain oversized polling POST bodies, then reject them with a 400.
-
-    engine.io compares ``Content-Length`` against ``max_http_buffer_size`` and
-    answers 400 without reading the request body. The client is still sending
-    at that point, and the unsent remainder can wedge its connection pool so
-    the reconnect handshake is never served. Reading the body to completion
-    first lets the rejection reach a client that has finished sending, so it
-    can record the failure and retry.
-
-    The rejection is answered by this middleware rather than engine.io because
-    engine.io rejects by disconnecting the session, and its disconnect waits
-    for a poll GET to consume the session's CLOSE packet before the POST
-    response is sent. When no poll is parked - common, since the engine.io
-    client sends the CONNECT POST and the first poll GET at the same time -
-    the POST would never be answered. The middleware answers 400 itself and
-    closes the session without waiting for the send queue.
+    """Drain an oversized polling POST, answer 400, and close its session.
+
+    engine.io rejects a POST over ``max_http_buffer_size`` without reading its
+    body, which strands the browser mid-send, and holds the 400 until a poll
+    takes the session's CLOSE packet. This middleware reads the body first,
+    answers itself, and closes the session without waiting for a poll.
     """
 
-    def __init__(
-        self, app: ASGIApp, max_size: int, eio: engineio.AsyncServer | None = None
-    ) -> None:
-        """Wrap an ASGI app, rejecting oversized POST bodies after draining them.
+    def __init__(self, app: ASGIApp, eio: AsyncEngineIOServer) -> None:
+        """Wrap the socket app.
 
         Args:
-            app: The next ASGI app in the middleware stack.
-            max_size: The maximum accepted body size; larger POST bodies are drained
-                and rejected.
-            eio: The engine.io server, used to close the rejected session without
-                waiting for its send queue. Optional; the rejected request never
-                reaches engine.io, so when omitted the session is left open and
-                expires on engine.io's own timeout.
+            app: The socket ASGI app.
+            eio: The engine.io server whose buffer limit and sessions apply.
         """
         self.app = app
-        self.max_size = max_size
         self.eio = eio
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        """Drain an oversized POST body, then forward the request downstream.
+        """Drain and reject an oversized POST; pass other requests through.
 
         Args:
             scope: The ASGI connection scope.
             receive: The ASGI receive callable.
             send: The ASGI send callable.
         """
-        if scope["type"] != "http" or scope["method"] != "POST":
-            await self.app(scope, receive, send)
-            return
-
-        content_length: int | None = None
-        for name, value in scope.get("headers", []):
-            if name.lower() == b"content-length":
-                try:
-                    content_length = int(value)
-                except ValueError:
-                    content_length = None
-                break
-
-        if content_length is None or content_length <= self.max_size:
+        if (
+            scope["type"] != "http"
+            or scope["method"] != "POST"
+            or int(Headers(scope=scope).get("content-length", 0))
+            <= self.eio.max_http_buffer_size
+        ):
             await self.app(scope, receive, send)
             return
-
-        # Drain the declared body so the client finishes sending before the
-        # rejection reaches it. The bytes are discarded.
-        remaining = content_length
-        while remaining > 0:
-            message = await receive()
-            if message["type"] != "http.request":
-                break
-            remaining -= len(message.get("body", b""))
-            if not message.get("more_body"):
-                break
-
-        # Answer the oversized POST from here instead of delegating to
-        # engine.io. engine.io rejects on Content-Length by disconnecting the
-        # session, and disconnect waits in queue.join() for a poll GET to
-        # consume the CLOSE packet before the POST response is sent. The
-        # engine.io client sends the CONNECT POST and the first poll GET at
-        # the same time, so no poll may be parked when the POST arrives, and
-        # its response would be held forever. Answer directly, then close the
-        # session without waiting: a parked poll still receives the CLOSE
-        # packet, but no poll is required for this response.
+        while (await receive()).get("more_body"):
+            pass
         await send({
             "type": "http.response.start",
             "status": 400,
             "headers": [(b"content-type", b"text/plain")],
         })
         await send({"type": "http.response.body", "body": b'"Bad Request"'})
-
-        if self.eio is not None:
-            sid = urllib.parse.parse_qs(
-                scope.get("query_string", b"").decode("latin-1")
-            ).get("sid", [None])[0]
-            socket = self.eio.sockets.get(sid) if sid else None
-            if socket is not None:
-                with contextlib.suppress(Exception):
-                    # Best-effort; the 400 above is already sent.
-                    await socket.close(wait=False)
-        return
+        sid = urllib.parse.parse_qs(scope["query_string"].decode()).get("sid", [""])[0]
+        if socket := self.eio.sockets.get(sid):
+            await socket.close(wait=False)
 
 
 def _is_location_specifier(specifier: str) -> bool:
@@ -759,11 +707,7 @@ def _setup_state(self) -> None:
         socket_app = EngineIOApp(self.sio, socketio_path="")
         # Drain oversized polling POST bodies before engine.io rejects them, so
         # the 400 response does not strand a half-sent request on the client.
-        drained_socket_app = _DrainOversizedBodyMiddleware(
-            socket_app,
-            max_size=self.sio.eio.max_http_buffer_size,
-            eio=self.sio.eio,
-        )
+        drained_socket_app = _DrainOversizedBodyMiddleware(socket_app, self.sio.eio)
         namespace = config.get_event_namespace()
 
         # Create the event namespace and attach the main app. Not related to any paths.
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 03699e36bb5..1295b79099f 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -5719,7 +5719,12 @@ def _polling_engine_app(max_size: int):
 
 
 async def _asgi_http_call(
-    app, method: str, query_string: str, body: bytes = b"", chunk_size: int = 65536
+    app,
+    method: str,
+    query_string: str,
+    body: bytes = b"",
+    chunk_size: int = 65536,
+    path: str = "/",
 ):
     """Drive one HTTP request through an ASGI app with a chunked body.
 
@@ -5729,6 +5734,7 @@ async def _asgi_http_call(
         query_string: The raw query string.
         body: The request body.
         chunk_size: Size of each http.request message body chunk.
+        path: The request path.
 
     Returns:
         A tuple of (status, response body, request bytes consumed from receive).
@@ -5775,8 +5781,8 @@ async def send(message):  # noqa: RUF029
         "http_version": "1.1",
         "method": method,
         "scheme": "http",
-        "path": "/",
-        "raw_path": b"/",
+        "path": path,
+        "raw_path": path.encode(),
         "query_string": query_string.encode(),
         "root_path": "",
         "headers": headers,
@@ -5791,124 +5797,121 @@ async def send(message):  # noqa: RUF029
     return status, response_body, consumed
 
 
-async def test_engineio_rejects_oversized_polling_post_without_draining_body():
-    """Document the engine.io behavior _DrainOversizedBodyMiddleware works around.
+async def _polling_handshake(app, path: str = "/") -> str:
+    """Open an engine.io polling session.
 
-    engine.io rejects a polling POST larger than max_http_buffer_size with 400
-    without reading the request body, stranding the client mid-send.
+    Args:
+        app: The ASGI app serving engine.io.
+        path: The request path.
+
+    Returns:
+        The query string that addresses the new session.
     """
-    app, _ = _polling_engine_app(max_size=100)
     status, response_body, _ = await _asgi_http_call(
-        app, "GET", "EIO=4&transport=polling"
+        app, "GET", "EIO=4&transport=polling", path=path
     )
     assert status == 200
     sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
     assert sid is not None
+    return f"EIO=4&transport=polling&sid={sid.group(1)}"
 
-    # A real polling client always has a poll in flight; park one so the
-    # server can deliver its CLOSE packet when it rejects the POST.
-    poll_task = asyncio.create_task(
-        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}")
-    )
-    await asyncio.sleep(0.1)
 
-    big_body = b"x" * 1000
-    status, _, consumed = await asyncio.wait_for(
+async def _oversized_post(app, query_string: str, path: str = "/"):
+    """POST a 1000-byte body in 100-byte chunks, failing after 10 seconds.
+
+    Args:
+        app: The ASGI app serving engine.io.
+        query_string: The query string that addresses the session.
+        path: The request path.
+
+    Returns:
+        A tuple of (status, response body, request bytes consumed from receive).
+    """
+    return await asyncio.wait_for(
         _asgi_http_call(
-            app,
-            "POST",
-            f"EIO=4&transport=polling&sid={sid.group(1)}",
-            big_body,
-            chunk_size=100,
+            app, "POST", query_string, b"x" * 1000, chunk_size=100, path=path
         ),
         timeout=10,
     )
+
+
+async def test_engineio_rejects_oversized_polling_post_without_draining_body():
+    """Document the engine.io behavior _DrainOversizedBodyMiddleware works around.
+
+    engine.io rejects a polling POST larger than max_http_buffer_size with 400
+    without reading the request body, stranding the client mid-send.
+    """
+    app, _ = _polling_engine_app(max_size=100)
+    query = await _polling_handshake(app)
+
+    # Park a poll so engine.io can deliver its CLOSE packet and answer the POST.
+    poll_task = asyncio.create_task(_asgi_http_call(app, "GET", query))
+    await asyncio.sleep(0.1)
+
+    status, _, consumed = await _oversized_post(app, query)
     poll_status, _, _ = await asyncio.wait_for(poll_task, timeout=5)
     assert status == 400
     assert poll_status == 200
-    assert consumed < len(big_body)
+    assert consumed < 1000
 
 
 async def test_drain_oversized_body_middleware_drains_and_rejects_oversized_post():
     """Oversized polling POST bodies are drained so the client can finish sending."""
     socket_app, eio = _polling_engine_app(max_size=100)
-    app = _DrainOversizedBodyMiddleware(socket_app, max_size=100, eio=eio)
-    status, response_body, _ = await _asgi_http_call(
-        app, "GET", "EIO=4&transport=polling"
-    )
-    assert status == 200
-    sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
-    assert sid is not None
+    app = _DrainOversizedBodyMiddleware(socket_app, eio)
+    query = await _polling_handshake(app)
 
-    # A real polling client always has a poll in flight; park one so the
-    # server can deliver its CLOSE packet when it rejects the POST.
-    poll_task = asyncio.create_task(
-        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}")
-    )
+    # A parked poll still receives the CLOSE packet of the rejected session.
+    poll_task = asyncio.create_task(_asgi_http_call(app, "GET", query))
     await asyncio.sleep(0.1)
 
-    big_body = b"x" * 1000
-    status, _, consumed = await asyncio.wait_for(
-        _asgi_http_call(
-            app,
-            "POST",
-            f"EIO=4&transport=polling&sid={sid.group(1)}",
-            big_body,
-            chunk_size=100,
-        ),
-        timeout=10,
-    )
+    status, _, consumed = await _oversized_post(app, query)
     poll_status, _, _ = await asyncio.wait_for(poll_task, timeout=5)
     assert status == 400
     assert poll_status == 200
-    assert consumed == len(big_body)
+    assert consumed == 1000
 
     # The reconnect handshake is served normally after the rejected POST.
-    status, _, _ = await _asgi_http_call(app, "GET", "EIO=4&transport=polling")
-    assert status == 200
+    await _polling_handshake(app)
 
 
 async def test_drain_oversized_body_middleware_answers_post_without_parked_poll():
     """The oversized POST is answered even when no poll is in flight.
 
-    The engine.io client sends the CONNECT POST and the first poll GET at the
-    same time, so the POST can be rejected before any poll is parked. The
-    rejection must not wait for a poll to consume the session CLOSE packet.
+    The rejection must not wait for a poll to consume the session CLOSE packet.
     """
     socket_app, eio = _polling_engine_app(max_size=100)
-    app = _DrainOversizedBodyMiddleware(socket_app, max_size=100, eio=eio)
-    status, response_body, _ = await _asgi_http_call(
-        app, "GET", "EIO=4&transport=polling"
-    )
-    assert status == 200
-    sid = re.search(r'"sid":"([^"]+)"', response_body.decode())
-    assert sid is not None
+    app = _DrainOversizedBodyMiddleware(socket_app, eio)
+    query = await _polling_handshake(app)
 
-    # No poll is parked: the engine.io client races the CONNECT POST against
-    # the first poll GET, so the POST can arrive first.
-    big_body = b"x" * 1000
-    status, response_body, consumed = await asyncio.wait_for(
-        _asgi_http_call(
-            app,
-            "POST",
-            f"EIO=4&transport=polling&sid={sid.group(1)}",
-            big_body,
-            chunk_size=100,
-        ),
-        timeout=10,
-    )
+    status, response_body, consumed = await _oversized_post(app, query)
     assert status == 400
     assert response_body == b'"Bad Request"'
-    assert consumed == len(big_body)
+    assert consumed == 1000
 
     # A later poll learns the session is gone instead of the POST hanging.
     poll_status, _, _ = await asyncio.wait_for(
-        _asgi_http_call(app, "GET", f"EIO=4&transport=polling&sid={sid.group(1)}"),
-        timeout=5,
+        _asgi_http_call(app, "GET", query), timeout=5
     )
     assert poll_status == 400
 
 
+async def test_app_event_endpoint_drains_oversized_polling_post():
+    """The App's event endpoint drains and answers an oversized polling POST."""
+    app = App(
+        enable_state=False,
+        sio=AsyncServer(async_mode="asgi", max_http_buffer_size=100),
+    )
+    app._enable_state()
+    assert app._api is not None
+    path = f"{constants.Endpoint.EVENT}/"
+    query = await _polling_handshake(app._api, path)
+
+    status, _, consumed = await _oversized_post(app._api, query, path)
+    assert status == 400
+    assert consumed == 1000
+
+
 async def test_drain_oversized_body_middleware_passes_regular_requests_through():
     """Requests within the size limit reach the downstream app untouched."""
     received = []
@@ -5925,7 +5928,7 @@ async def echo_app(scope, receive, send):
         await send({"type": "http.response.start", "status": 200, "headers": []})
         await send({"type": "http.response.body", "body": b"ok"})
 
-    app = _DrainOversizedBodyMiddleware(echo_app, max_size=100)
+    app = _DrainOversizedBodyMiddleware(echo_app, _polling_engine_app(100)[1])
     status, response_body, _ = await _asgi_http_call(app, "POST", "", b"hello")
     assert status == 200
     assert response_body == b"ok"

From 304bdab77a1344234ee2d74bc9c050054a378d10 Mon Sep 17 00:00:00 2001
From: Farhan <www.mfarvirus@gmail.com>
Date: Tue, 27 Oct 2026 23:37:19 +0500
Subject: [PATCH 9/9] Decode the drained POST's query string as Latin-1

engine.io treats a query string that is not UTF-8 as empty. A strict
decode raised after the 400 was sent, so the session was not closed.
Also word the news fragment as 'can reconnect': the client's reconnect
policy controls the timing.
---
 news/+drain-oversized-polling-post.bugfix.md |  2 +-
 reflex/app.py                                |  4 +++-
 tests/units/test_app.py                      | 13 ++++++++++++-
 3 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/news/+drain-oversized-polling-post.bugfix.md b/news/+drain-oversized-polling-post.bugfix.md
index 12ddebbaea2..ab4e39df1a4 100644
--- a/news/+drain-oversized-polling-post.bugfix.md
+++ b/news/+drain-oversized-polling-post.bugfix.md
@@ -1 +1 @@
-Fixed a polling-transport stall where an event larger than `REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE` left the page unable to reconnect: the backend now reads the oversized request body before it rejects it, so the browser reconnects at once.
+Fixed a polling-transport stall where an event larger than `REFLEX_SOCKET_MAX_HTTP_BUFFER_SIZE` left the page unable to reconnect: the backend now reads the oversized request body before it rejects it, so the browser can reconnect.
diff --git a/reflex/app.py b/reflex/app.py
index 022a8082be0..234ca0fc2bc 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -413,7 +413,9 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             "headers": [(b"content-type", b"text/plain")],
         })
         await send({"type": "http.response.body", "body": b'"Bad Request"'})
-        sid = urllib.parse.parse_qs(scope["query_string"].decode()).get("sid", [""])[0]
+        sid = urllib.parse.parse_qs(scope["query_string"].decode("latin-1")).get(
+            "sid", [""]
+        )[0]
         if socket := self.eio.sockets.get(sid):
             await socket.close(wait=False)
 
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 1295b79099f..08281a54293 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -5783,7 +5783,7 @@ async def send(message):  # noqa: RUF029
         "scheme": "http",
         "path": path,
         "raw_path": path.encode(),
-        "query_string": query_string.encode(),
+        "query_string": query_string.encode("latin-1"),
         "root_path": "",
         "headers": headers,
         "client": ("127.0.0.1", 12345),
@@ -5896,6 +5896,17 @@ async def test_drain_oversized_body_middleware_answers_post_without_parked_poll(
     assert poll_status == 400
 
 
+async def test_drain_oversized_body_middleware_closes_session_for_non_utf8_query():
+    """A query string that is not UTF-8 still closes the rejected session."""
+    socket_app, eio = _polling_engine_app(max_size=100)
+    app = _DrainOversizedBodyMiddleware(socket_app, eio)
+    query = await _polling_handshake(app)
+
+    status, _, _ = await _oversized_post(app, f"{query}&x=\xff")
+    assert status == 400
+    assert eio.sockets[query.rpartition("=")[2]].closed
+
+
 async def test_app_event_endpoint_drains_oversized_polling_post():
     """The App's event endpoint drains and answers an oversized polling POST."""
     app = App(