Starlette: Detect/handle closed client connections

[blueyed]

It appears like starlette does not cancel inner processing for when a client disconnects.

With the following code it will keep on fetching from the remote, although the
client has disconnected already, e.g. when cancelling a curl call.

@app.route(…)
async def foo(request):
    …
    if content_type.startswith("video/"):
        bytes_total = req.headers["Content-Length"]
        logger.info("Streaming {} bytes...", bytes_total)

        async def generator(bytes_total):
            import aiohttp
            async with aiohttp.ClientSession() as session:
                chunk = 0
                chunk_size = 102400
                bytes_read = 0
                async with session.request(method, str(url), chunked=chunk_size) as resp:
                    async for data, end_of_http_chunk in resp.content.iter_chunks():
                        chunk += 1
                        bytes_read += chunk_size
                        logger.info("chunk {} {}/{}", chunk, bytes_read, bytes_total)
                        yield data
        return StreamingResponse(generator(bytes_total), media_type=content_type)

It might be useful if the handler could be notified of this, in case it wants
to handle this, but in general I think a late middleware, responsible for
sending could handle this maybe?!

[blueyed]

btw: I've looked a bit into requests-threads etc, but it appears to not be possible really to use requests inside of a handler asynchronously, is it?

[blueyed]

I've found that uvicorn just ignores it in send: https://github.com/encode/uvicorn/blob/83442255299f49a6e9b0a0ea3935992bc3e3dc81/uvicorn/protocols/http/httptools_impl.py#L432-L433
Should it or rather Starlette maybe raise ClientDisconnect like with reading the request? https://github.com/encode/uvicorn/blob/83442255299f49a6e9b0a0ea3935992bc3e3dc81/uvicorn/protocols/http/httptools_impl.py#L432-L433

Anyway, the following example app works with uvicorn:

import asyncio

import logging
logger = logging.getLogger(__name__)


class App():
    def __init__(self, scope):
        self.scope = scope

    async def __call__(self, receive, send):
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [
                [b'content-type', b'text/plain'],
            ],
        })
        for i in range(10):
            await send({
                'type': 'http.response.body',
                'body': str(i).encode(),
                'more_body': True,
            })
            logger.info("sent: %d", i)
            await asyncio.sleep(1)

            if receive.__self__.disconnected:
                logger.info("disconnected")
                break

        await send({
            'type': 'http.response.body',
            'body': '',
            'more_body': False,
        })

[florimondmanca]

@blueyed I believe this is a limitation of the ASGI spec itself. The Error handling section reads:

Note messages received by a server after the connection has been closed are not considered errors. In this case the send awaitable callable should act as a no-op.

And that's why send() just returns when the client has disconnected.

I noticed that @tomchristie already started the discussion on this https://github.com/django/asgiref/issues/66, though.

That said, I second the need to know when a client has disconnected, especially when dealing with StreamingResponses. I'm trying to implement SSE manually by returning a StreamingResponse, but the generator cannot know when the client disconnects and will keep sending events (although to a no-op) indefinitely.

[blueyed]

@florimondmanca
Thanks for the pointer(s).

I think checking receive.__self__.disconnected would also work for you, wouldn't it?

(I've initially tried await receive(), where a disconnect message would appear, but that might block without any message(s))

[florimondmanca]

Yup, the __self__ trick does work for me. Also tried hacking StreamingResponse to check for http.disconnect on each iteration of the body iterator, but that won't work because it blocks as you've just said. :) The only way I've found was raising a disconnect exception and wrapping the await response(receive, send) call around a try/except block. I understand this is not the philosophy of ASGI for now, but it works. That said, I'm not sure about whether this skips necessary cleanup on the uvicorn side (e.g. releasing connections).

[tomchristie]

I'm trying to implement SSE manually by returning a StreamingResponse, but the generator cannot know when the client disconnects and will keep sending events (although to a no-op) indefinitely.

Right. The thing is that the disconnect needs to be handled in an event driven way, rather than raising an exception on send().

I've initially tried await receive(), where a disconnect message would appear, but that might block without any message(s)

Sure, so I think what you'd need is something like:

done, pending = await asyncio.wait([receive, response.content.read], return_when=FIRST_COMPLETED)
task = done[0]
if task is receive:
    message = task.result()
else:
    data = task.result()

Not wild about how akward it is to do a await either this thing or that thing and switch on either,
but there we are.

[tomchristie]

I think the following would also do the trick, for a non-blocking receive: message = await asyncio.wait_for(receive, timeout=0).

[tomchristie]

What do we think to #320?