"Task was destroyed but it is pending!" error when middleware dispatch func discards response from call_next

[itayperl]

Here's a small test case based on FastAPI

from fastapi import FastAPI, Request, Response

app = FastAPI()

@app.middleware('http')
async def foo(request: Request, call_next):
    resp = await call_next(request)
    if resp.status_code == 404:
        return Response('Oh no!')
    return resp

@app.get('/meow')
def meow():
    return 'Meow!'

Calling a nonexistent endpoint on this server will result in this error log at some point:

ERROR:    Task was destroyed but it is pending!
task: <Task pending name='Task-4' coro=<BaseHTTPMiddleware.call_next.<locals>.coro() done, defined at /usr/lib/python3.8/site-packages/starlette/middleware/base.py:36> wait_for=<Future cancelled>>

The problem is that the response returned from call_next is not used, and the coroutine started in BaseHTTPMiddleware.call_next keeps blocking on queue.put until the Task object is GC'd.

This seems to be a new issue in 0.13.7 as in previous versions the queue was not limited and didn't block.

[erewok]

Thanks for reporting this issue and thanks for putting together a straightforward example. If you had middleware that completely discarded the request on every call, then it looks like it could accumulate pending tasks in a more predictable manner. A simpler example of that would look like this:

@app.middleware('http')
async def foo(request: Request, call_next):
    resp = await call_next(request)
    return Response('Oh no!')

The change to limit the queue size was to benefit those using this middleware class with their own StreamingResponses but in this case, the code is assuming that the response will be fully evaluated. We'll have to figure out how to accommodate both cases.

[erewok]

@itayperl, After discussing this issue with encode maintainers, we've decided we should revert this change and issue a new release, under the suspicion that your issue is likely to be more common and potentially more problematic than the issue we fixed for users who are returning StreamingResponses from their handlers.

On the whole, this particular middleware class has some issues that need to be worked through, and it will need to be reevaluated in the future. I suggest creating Middleware similar to other examples in the Starlette repo and in this particular case (if you are using middleware for particular status codes as in your example), to switch to using something like the exception handlers example from the docs.

[itayperl]

@erewok, thanks for looking into this.

I didn't mention this in my report, but I actually upgraded to 0.13.7 because I needed the backpressure fix. I have an event-stream endpoint that streams data for as long as the connection is open. In some cases a client would keep a connection open but stop reading from it. Whenever that happened, the server would gradually leak all of the system's memory, until it got killed by the kernel.

In the meantime I've also changed my middleware to the following, which works for me on 0.13.7:

@app.middleware('http')
async def foo(request: Request, call_next):
    for route in app.router.routes:
        match, _ = route.matches(request.scope)
        if match != starlette.routing.Match.NONE:
            return await call_next(request)

    return Response('...')

[erewok]

I have opened a separate PR on reverting the maxsize fix here for discussion: #1028

It's good to know that the fix in version 0.13.7 resolved your problem.

[erewok]

We have reverted the fix from 0.13.7 that lead to this problem. I am going to close this issue as a result, because the behavior from 0.13.6 should be restored. Thanks again for reporting the issue.