Expected behavior of Response.aclose?

[james-mchugh]

Hello everyone!

I am using the httpx.AsyncClient to send a request to an endpoint that responds with an infinite stream of data. Specifically, I am using it to proxy a request from a client to the API with the streaming response, like so:

CLI Client --> Proxy --> Streaming API

I have followed the docs here to return the streaming response from the proxy, resulting in code that looks similar to (modified for brevity)

client = httpx.AsyncClient(base_url=API_URL)

def proxy(request):
    request = client.build_request(
        "GET",
        request.path,
    )
    resp = await client.send(request, stream=True)
    return fastapi.responses.StreamingResponse(
        resp.aiter_lines(),
        background=starlette.background.BackgroundTask(resp.aclose),
    )

As instructed in the link above:

When using this "manual streaming mode", it is your duty as a developer to make sure that Response.aclose() is called eventually. Failing to do so would leave connections open, most likely resulting in resource leaks down the line.

We have a BackgroundTask to ensure the response is closed. However, this does not appear to actually close any connections. After running the Proxy and Streaming API through debuggers, I can see that the proxy calls resp.aclose as expected, but the Streaming API continues to attempt to stream the response back, ignorant of the resp.aclose call.

On closer inspection with strace, I saw that the Proxy was not making only making a single close call to close the connection with the Client, but it was not called to close the connection with the Streaming API. Furthermore, tcpdump does not show any Fin TCP packets sent between the Proxy and Streaming API, so the Streaming API is never notified that it should stop streaming the response. It seems that the Response.aclose method is just doing some client-side housekeeping.

I'm assuming that this is due to connection pooling done by the httpx.Client and HTTP 1.1 persistent connections. I was able to resolve the issue by rewriting some of the proxy code to close the client instead of the response, which got me the behavior I was looking for.

My question is, is this right? Or am I just doing something wrong? Given the way the documentation explains how to handle streaming responses when proxying requests, it seems like the server would always be left in a state where it streaming responses back to a client that isn't acting on them. Does the documentation have to be updated to avoid this situation? I felt a bit misled that it says "Failing to do so [call Response.aclose()] would leave connections open", which seemed to imply that calling Response.aclose would close the connection. But maybe it is actually referring to the client being told that the response is closed so it can eventually clean up the connection if no other requests are made.

[lovelydinosaur]

Hi. There's two different things mixed in here - how httpx is behaving, and whatever subtleties backgrounds tasks and streaming responses have in fastapi/starlette.

Assume I don't know anything about fastapi/starlette - can you present the issue just with an httpx code example?

(This should also help you debug the problem)

[james-mchugh]

Hello @tomchristie . Thank you for responding! Good call with separating out the code and simplifying the example. I'm not sure why I didn't think to do that. I'm assuming it was because I was trying to follow the example in the httpx docs as closely as possible. Based on the behavior I was seeing (observing Response.aclose being called when debugging, no close calls in strace, and no TCP Fin packets when running tcpdump), I was sure something was wrong with the aclose method or the documentation. But now when testing, I'm not able to reproduce the same situation anymore. It was undoubtedly a byproduct of trying to test too much at once. Sorry for the misunderstanding, and thank you for the help

[james-mchugh]

Okay, I looked into this before and I think my response was a little premature. I was able to reproduce the issue. To your original point, I do not encounter this issue when using a simple example without Starlette/FastAPI. Naturally, this indicates that httpx is operating fine and there is something weird going on with Starlette/FastAPI. However, it also indicates that the httpx streaming response documentation might need to be updated. Anybody reading that example would likely expect for the connection to be properly closed (especially given the explicit BackgroundTask calling Response.aclose), but that does not appear to be working.

I did trace this down a bit to better understand why aclose wasn't actually closing the connection, but I hit a bit of a wall. It seems that httpcore._async.http11.HTTP11ConnectionByteStream.__aiter__ will call httpcore._async.http11.HTTP11ConnectionByteStream.aclose when an exception is raised (such as a task being cancelled in the case of FastAPI/Starlette). httpcore._async.http11.HTTP11ConnectionByteStream.aclose marks httpcore._async.http11.HTTP11ConnectionByteStream()._closed as True and then calls httpcore._async.http11.AsyncHTTP11Connection._response_closed, which waits for a lock. However, the task never finishes running. When the background task finally executes, it sees that the HTTP11ConnectionByteStream instance already has _closed set to True and takes no further action to close the stream.

I'm not really sure why the original HTTP11ConnectionByteStream.aclose task never finishes running. I tried to reproduce it outside of Starlette, but it worked as expected.

[karpetrosyan]

Hi. At this moment we have concrete problems with our connection pool; if a request is cancelled in the middle of execution, it will leave the connection open; we are working on it and hope to have it fixed in the next httpcore release.

Also see: encode/httpcore#658 and encode/httpcore#704

If you're looking for a solution to that problem, this PR is probably what you're looking for.

[lovelydinosaur]

However, it also indicates that the httpx streaming response documentation might need to be updated.

Ah that's interesting, thanks @james-mchugh. I'd very much like us to not have examples like that.

I wonder how we could reframe that into a standalond code-example?

[james-mchugh]

That's a fair point @tomchristie. I think the example above the starlette one is sufficient for showing how the to iterate over a streaming response. I'm not sure what a good example would be for using the streaming response in "manual mode" though. I'll think about it a bit. I created a few examples when attempting to reproduce this error without starlette, but I don't think they would be super useful for readers

[karpetrosyan]

Something like this can reproduce that issue for me

import httpcore
import logging
import trio

logging.getLogger("httpcore.http11").setLevel(1)

async def log(p, e):
    if p.startswith('http11'):
        print(p, e)

pool = httpcore.AsyncConnectionPool()

async def main():
    with trio.move_on_after(0.3):
        await pool.request("GET", "https://www.youtube.com", extensions={"trace": log})
    print(pool.connections)

trio.run(main)

OUT

http11.send_request_headers.started {'request': <Request [b'GET']>}
http11.send_request_headers.complete {'return_value': None}
http11.send_request_body.started {'request': <Request [b'GET']>}
http11.send_request_body.complete {'return_value': None}
http11.receive_response_headers.started {'request': <Request [b'GET']>}
http11.receive_response_headers.failed {'exception': Cancelled()}
http11.response_closed.started {}
http11.response_closed.failed {'exception': Cancelled()}
[<AsyncHTTPConnection ['https://youtube.com:443', HTTP/1.1, ACTIVE, Request Count: 1]>]

[james-mchugh]

This is great. Thank you! I didn't know how to produce it without using starlette and I haven't used trio directly.