asgi: Send http.disconnect message on receive() after response complete

[JayH5]

I've been trying to track down what broke for me when using httpx with starlette==0.13.3 with ASGI dispatch, see Kludex/starlette#908.

There are kind of two different problems:

1. Starlette calls both the receive() (to check for disconnects) and send() (to send the response) concurrently in its StreamingResponse type. It tries to wait for either of those two options to complete (using asyncio.wait({tasks}, return_when=asyncio.FIRST_COMPLETED)). Because the ASGI dispatch in httpx is generally effectively synchronous, both send() and receive() may never return control to the event loop when called which means that if either task that is waited on does not complete, then asyncio.wait() will not return. I've made Fix infinite loop when using StreamingResponse with ASGI client Kludex/starlette#912 to attempt to work around that.
2. httpx's ASGI dispatch never sends an http.disconnect message. The ASGI docs say this should happen if receive() is called after a response is complete. This is what this PR does.

Starlette's test client does this already, although it also checks that the request is complete before sending http.disconnect--I'm not really sure why that check is necessary, though.

[florimondmanca]

This sounds good to me - right now we’re indeed missing the http.disconnect part from the ASGI spec so its worth fixing in itself. Thanks!

[florimondmanca]

Does this PR and Kludex/starlette#912 fix the issue you were seeing with StreamingResponse?

[JayH5]

@florimondmanca unfortunately, it looks like this only partially fixes my problem there 😞

I have a custom Starlette middleware library (unfortunately closed-source) that I'm testing using httpx. Before this change, any test that called the app with middleware installed would hang. After this change, one of the tests will pass, but then subsequent tests still hang. Some callback somewhere or something is obviously not completing. The tests are all very simple and similar to the one in Kludex/starlette#908.

My tests all pass with Kludex/starlette#912 (even without this change) but it's obviously not the nicest fix so it'd be good to get to the bottom of why this PR doesn't fix things. I'm still digging...just somewhat stuck in async callback complexity 😰

[JayH5]

OK so I think I have some idea of what's happening. My point number 1 in the description above is still accurate, but also it's worse than that.

When Starlette calls asyncio.wait(), the tasks it waits on are put into a set and thus may be executed in any order. Sometimes we get lucky, the stream_response happens first, the response is completely consumed, control is passed to the listen_for_disconnect function, it calls receive() and gets the http.disconnect message (thanks to this PR) and is able to complete. But sometimes we get unlucky and listen_for_disconnect gets called first and it gets stuck in a loop again and control is never passed to the event loop or even the stream_response method.

So I think this PR is a valid fix but doesn't solve Kludex/starlette#908.

Starlette's test client doesn't suffer from this problem because it doesyield to the event loop. httpx is not really able to do that because we can't just call asyncio.sleep() for async runtime compatibility reasons.

[florimondmanca]

@JayH5 I think this is a valid fix in itself. It's very likely that we need a Starlette fix as well - most probably via Kludex/starlette#912. Are we okay to merge this?

[lovelydinosaur]

Seems valid to me too, yup.

[JayH5]

I think we're ok to merge this one, the Starlette one I'm less sure about 👍

[lovelydinosaur]

httpx is not really able to do that because we can't just call asyncio.sleep() for async runtime compatibility reasons.

Well, maybe. It's possible that actually we do want the fix to that to be at the httpx level, and ensure that await receive() is always going to yield control.

HTTPX really shouldn't just keep sending {'type': 'http.request', 'body': b'', 'more_body': False} once there's no more body to send. Instead, once that's occurred, if receive is called again, then we really ought to wait for a response_completeevent to be set, and then return {"type": "http.disconnect"}.

(And either error if we're called once again after that, or just block forever.)

[lovelydinosaur]

if receive is called again, then we really ought to wait for a response_complete event to be set, and then return {"type": "http.disconnect"}.

(Which would resolve the issue, since waiting for the event really would be an async checkpoint)

[JayH5]

@tomchristie yeah I'm in agreement, I just don't know how to implement an "event"?

[lovelydinosaur]

Sketching out some of it...

defget_event():
ifsniffio.current_async_library() =="trio":
returntrio.Event()
else:
returnasyncio.Event()
...
asyncdefrequest(...):
# ...response_complete_event=get_event()
# To set the event.response_complete_event.set()
# To wait for the event.awaitresponse_complete_event.wait()

[lovelydinosaur]

We've got similar stuff in the httpcore backends, but given the limited scope of what we need in the ASGI dispatcher, I think we could perfectly well just use a little utility function like that, here.

[JayH5]

Cool, yeah, I was uncertain about where that event abstraction should live since now most of the async-runtime-specific stuff is in httpcore. I can do what you're suggesting.

One concern is that sniffio is not a direct dependency anymore (side note that the dependency list in the README needs to be updated).

[lovelydinosaur]

Well noticed, yes. I'm okay with sniffio being a direct dependancy (as well as a dependancy via httpcore).

[JayH5]

OK, should get to another PR a bit later today.

[JayH5]

@tomchristie this discussion resulted in #919