Checking if a request is disconnected fails when using BaseHTTPMiddleware.

[dominik-schwabe]

I use FastAPI where I have some route that does some long work which becomes useless when the client disconnects. I want to cancel the request when the client disconnects. I used some code similar to the code below which worked for me in the past. I tried to configured some middleware which broke the code. The following code works fine when using starlette before version 0.21.0 or when removing the middleware.

import asyncio
import logging

import uvicorn
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

logger = logging.getLogger("uvicorn")


async def check_disconnected(request):
    while not await request.is_disconnected(): # check every second if the client is disconnected
        logger.info("sleeping")
        await asyncio.sleep(1)


def cancel_on_disconnect(func):
    async def wrapper(request):
        futures = [
            asyncio.ensure_future(c) for c in [check_disconnected(request), func()]
        ]
        # wait until the request is done or the client disconnects
        done, _ = await asyncio.wait(futures, return_when=asyncio.FIRST_COMPLETED)
        done_coro = done.pop()
        if done_coro is futures[0]: # when the client disconnected
            logger.info("disconnected")
            raise HTTPException(503)
        logger.info("request end")
        return done_coro.result()

    return wrapper


@cancel_on_disconnect
async def homepage():
    await asyncio.sleep(5) # the request does some work
    return JSONResponse({})


async def some_middleware(request: Request, call_next):
    return await call_next(request)


app = Starlette(
    debug=True,
    routes=[Route("/", homepage, methods=["GET"])],
    middleware=[Middleware(BaseHTTPMiddleware, dispatch=some_middleware)],
)

uvicorn.run(app, host="0.0.0.0", port=5000)

The following client code can be used to check the behaviour.

import asyncio
import aiohttp


async def make_request():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:5000/"):
            pass


async def main(): # make the request in parallel and cancel it after 2.5 seconds
    request = asyncio.ensure_future(make_request())
    await asyncio.sleep(2.5)
    request.cancel()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

The output when using starlette before version 0.21.0 or when removing the middleware should be similar to the following (expected behaviour):

INFO:     Started server process [76409]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:5000 (Press CTRL+C to quit)
INFO:     sleeping
INFO:     sleeping
INFO:     sleeping
INFO:     disconnected

The output when using starlette after version 0.21.0 with the middleware should be similar to the following (unexpected behaviour):

INFO:     Started server process [75813]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:5000 (Press CTRL+C to quit)
INFO:     sleeping
INFO:     sleeping
INFO:     sleeping
INFO:     sleeping
INFO:     sleeping
INFO:     request end

[Kludex]

Maybe related to #2093

[Krismix1]

We've stumbled upon the same problem. Downgrading to 0.20.4 fixes the problem.

Not sure if this would be useful at all, but before finding this thread, I have downgraded to 0.14.2 (the version when it worked for us), which is the last version of Starlette before the migration to anyio. On 0.14.2 it worked both when a middleware was used and without it.

[adriangb]

I tried translating your example to use anyio (with the asyncio backend) and still see the issue.

[Kludex]

I think it was fixed from 0.20.1 to 0.23.1, according to my previous investigation: #2093 (comment)

[weditor]

same issue on the latest version: 0.31.1 .

[adriangb]

@Kludex mind looking into this since you’ve been following related issues for a while?

[weditor]

This simpler code can also reproduce this problem #2271 , hope it helps.

[Kludex]

Can someone create a table of affected versions for every Starlette version from 0.14.0 to current, and considering AnyIO versions 3 and 4 as well? Also, was this reintroduced?

I don't have bandwidth to investigate this for the time being, but I appreciate help. If confirmed, I can change the priority in my backlog.

[weditor]

I'm not sure if this's what you need:

starlette ver.	anyio ver.	result
0.14.0	/	ok
0.14.1	/	ok
0.14.2	/	ok
0.15.0	3.7.1	ok
0.16.0	3.7.1	ok
0.17.0	3.7.1	ok
0.17.1	3.7.1	ok
0.18.0	3.7.1	ok
0.19.0	4.0.0	ok
0.19.1	4.0.0	ok
0.20.0	4.0.0	ok
0.20.1	4.0.0	ok
0.20.2	4.0.0	ok
0.20.3	4.0.0	ok
0.20.4	4.0.0	ok
0.21.0	4.0.0	fail
0.22.0	4.0.0	fail
0.23.0	4.0.0	fail
0.23.1	4.0.0	fail
0.24.0	4.0.0	fail
0.25.0	4.0.0	fail
0.26.0	4.0.0	fail
0.26.1	4.0.0	fail
0.27.0	4.0.0	fail
0.28.0	4.0.0	fail
0.29.0	4.0.0	fail
0.30.0	4.0.0	fail
0.31.0	4.0.0	fail
0.31.1	4.0.0	fail

[marcbouissa]

I think the issue is related to this section of code in the is_disconnected(self) method of the request.

  with anyio.CancelScope() as cs:
          cs.cancel()
          message = await self._receive()

I am using python 3.10.11 and starlette 0.27.0 with fastapi 0.97.0

the issue is that the cs.cancel() is getting triggered immediately without giving any time for the message to be read using self._receive(). the message would always remain an empty dictionary.

changing the code to

    with anyio.move_on_after(0.1) as scope:
        message = await self.request.receive()

resolved the issue.

[jakkdl]

I independently found these lines to be the cause of pgjones/hypercorn#210

I can't find any documentation on what starlette demands of the receive callable. The way this code is written it appears that it demands the receive callable not to checkpoint before checking for new data (because if it does, the cancelscope will immediately cancel and stop execution of the callable), that the callable does not block while checking if there is a message in the queue, and after doing so either exits or checkpoints.

Hypercorn passes asyncio.Queue.get when running with asyncio as backend, which ends up calling asyncio.Queue.get_nowait, which checks if there is a message in the queue and otherwise immediately exits. It never checkpoints, so the CancelScope never comes into effect.
When hypercorn runs with trio as a backend it passes MemoryChannel.receive, which immediately calls await trio.lowlevel.checkpoint_if_cancelled() - cancelling the CancelScope before it ever tries to check for new messages.

Checking git blame this was changed in https://github.com/encode/starlette/pull/1157/files#diff-7d5cdd0c41aca64786e140201f4b95f1983e2ceb09d47f52de587669e8df776f, with these lines previously being
https://github.com/uSpike/starlette/blob/5ee04ef9b1bc11dc14d299e6c855c9a3f7d5ff16/starlette/requests.py#L254-L257
which is pretty much identical to what @dominik-schwabe suggested.
I'm completely sure why the PR settled on this behaviour with its niche demands on the receive callback. Traversing git blame within the PR they did try different timeouts but ultimately abandoned it.

@Kludex @adriangb I can try to write a regression test (that doesn't rely on hypercorn) and/or a PR if that would be welcomed.

[jakkdl]

Oh there was a bunch of discussion on these lines in the PR:
#1157 (comment)
#1157 (comment)

I think the cleanest way of rewriting this is to have a fairly long timeout, and specify in the documentation that it should be a non-blocking receive callback or the call to is_disconnect might be very slow. Which means that e.g. hypercorn should pass MemoryChannel.receive_nowait (wrapped to be an async method, and suppressing trio.WouldBlock)

[ener-jay]

I am experiencing the same issue. Client disconnect is not detected for an app with some middleware. It works without middleware enabled. It also works with middleware enabled for older versions as @weditor pointed out.

starlette 0.27.0
anyio 3.7.1

How can this be escalated as an issue (from potential issue)?

[adriangb]

Can you rewrite it to be a pure ASGI middleware?

[jakkdl]

End-users wanting a quick until starlette fixes it upstream can use this helper instead of calling Request.is_disconnected()

async def my_is_disconnected(request: Request) -> bool:
    assert hasattr(request, '_is_disconnected'), "private API in starlette changed"
    assert isinstance(request._is_disconnected, bool), "private API in starlette changed"

    if request._is_disconnected:
        return True

    message: Message = {}

    # this timeout may need to be tweaked.
    # Ideally request.receive() is non-blocking, in which case the timeout doesn't matter.
    # But the ASGI spec seems to imply it should be a blocking callable, as it doesn't specify
    # what should be returned in case of no new messages. In this situation you can return an
    # empty dict, but given that it's against the spec it seems inadvisable.
    with anyio.move_on_after(0.01):
        message = await request.receive()

    if message.get("type") == "http.disconnect":
        request._is_disconnected = True

    return request._is_disconnected

Alternatively if you control the receive callable you can also edit/wrap it such that it first tries to return a receivable, then checkpoints, and only then does an actual blocking wait. That way you will never have a slowdown from the timeout in the above workaround.

From my reading of the ASGI spec it implies that the callable should be a blocking wait*, so the only way I could see starlette never needing to have a timeout at all would be to rewrite the code in starlette/requests.py to have an internal nursery that watches for disconnect events and sets Request._is_disconnected, and Request.is_disconnected() solely checks the status of Request._is_disconnected. But maybe there's other reasons not to do that

* https://github.com/django/asgiref/blob/main/specs/asgi.rst says

receive: an awaitable callable that will yield a new event dictionary when one is available

and the types only specifies valid messages as return objects.

[jakkdl]

Quickly bashed together some tests that demonstrate the issue (and one way that ASGI servers could wrap memory object streams to work with starlette.. but that clearly should not be required of them): https://github.com/jakkdl/starlette/blob/disconnect_not_detected/tests/test_disconnect.py

[jakkdl]

Uh, tangentially related, isn't it also a problem that is_disconnect() may eat messages without handling them properly? Added another test to the above fork.

[ihnokim]

@jakkdl Thank you, with your help, I was able to resolve the issue. It seems that the cs.cancel() part was indeed the cause of the problem.

[jakkdl]

After trying to write a fixed is_disconnected() I'm starting to believe that maybe that method shouldn't be relied upon at all, and users should instead rely on .stream()/.body()/.json() to raise ClientDisconnect - and move logic pertaining to streaming a response or similar to a generator passed to StreamingResponse. It has a TaskGroup internally with one task listening for client disconnects, and will cancel the group (and your generator) when the client disconneccts. Although be wary of AsyncGenerators: python-trio/flake8-async#211

[jakkdl]

I think maybe is_disconnected should be deprecated and be replaced with a method that will wait until the client disconnects (i.e. never return False).
That way users can recreate the behavior from inside StreamingResponse with creating a TaskGroup that waits for disconnects in one task, or do it inside a anyio.move_on_after with a timeout of their own choice, etc.

It should also raise an error in case it is called when _stream_is_consumed is False, or errors if it receives a message that's anything other than a disconnect, to avoid the problem of it eating message data.

[ihnokim]

@jakkdl 's solution offered a clever workaround that effectively addresses the current issue. However, I'll also retain the temporary solution I've been using for documentation purposes.

class MyMiddleware(BaseHTTPMiddleware):
  async def dispatch(
    self, request: Request, call_next: RequestResponseEndpoint
  ) -> Response:
    request.state.is_disconnected = request.is_disconnected
    response = await call_next(request)
    return response

# app.add_middleware(... other middlewares ...)
app.add_middleware(MyMiddleware)

@app.get("/")
async def hello(request: Request) -> Optional[Dict[str, Any]]:
  for _ in range(10):
    if await request.state.is_disconnected():
      print("Client is disconnected.")
      return None
    else:
      print("Client is connected.")
    await asyncio.sleep(1)
  return {"Hello": "World"}

In summary, it involves using the state as specified in the ASGI spec to share the is_disconnected method among request objects. Remember to be mindful of the order in which middleware is added, as it should contain the method of the very first middleware."

For reference, the ASGI doc describes the state as follows:

An empty namespace where the application can persist state to be used when handling subsequent requests.

Applications often want to persist data from the lifespan cycle to request/response handling. For example, a database connection can be established in the lifespan cycle and persisted to the request/response cycle.

[ihnokim]

I don't know why... but the first request object that the first middleware receives can address the cs.cancel() issue 😅