Only stop receiving stream on body_stream if body is empty

[Kludex]

• Closes middleware causes exceptions to not be raised/handled silently #1977

This solves the issue reported by @MrSalman333 on #1609 (comment).

Test 1 (#1609 (comment))

import logging

import anyio
from fastapi import Depends, FastAPI, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware import Middleware


class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        return await call_next(request)


app = FastAPI(middleware=[Middleware(CustomMiddleware)])


def dependency_a(response: Response):
    try:
        yield "A"
    finally:
        logging.warning("A")


def dependency_b(response: Response):
    try:
        yield "B"
    finally:
        logging.warning("B")


def dependency_c(response: Response):
    try:
        yield "C"
    finally:
        logging.warning("C")


@app.get(
    "/issue",
    dependencies=[
        Depends(dependency_a),
    ],
)
def show_dependencies_issue(
    b=Depends(dependency_b),
    c=Depends(dependency_c),
):
    print(b, c)
    return {"status": "ok"}

Test 2 (#1579 (comment))

from typing import Awaitable, Callable

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.routing import Route
from starlette.middleware import Middleware
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.requests import Request

class NoopMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable]):
        return await call_next(request)

def endpoint(request: Request):
    return Response(status_code=304)


app2 = Starlette(
    routes=[Route("/", endpoint=endpoint)],
    middleware=[Middleware(GZipMiddleware), Middleware(NoopMiddleware)]
)

[Kludex]

Now I need to understand what's happening 🤣

[Kludex]

Removing those two lines also solves the issue.

[Kludex]

I've removed them instead.

[Kludex]

I don't know why we have the issue... It would be cool if we could find out the reason, but this change solves the issue and doesn't bring any regression.

[adriangb]

I don't see how this change can hurt, but I'd like to get some more input and maybe do a little post mortem on how we got here since quickly merging a "this can't hurt" PR last time is what hurt us in the end. BaseHTTMiddleware is a complex beast!

[adriangb]

@jhominal I would really appreciate your thoughts on this if you can take a look

[MrSalman333]

This is what I found originally ,,
I didn't open a PR though since the PR that added this line was done to solve a bug since it was alternative to #1579

[Kludex]

Some findings: this only happens with more than one sync dependency.

That's it, no more findings.

[Kludex]

Ok, if we add a await asyncio.sleep after the yield on the async dependency, I'm able to see the cancellation.

[Kludex]

I know why we are able to see the logs after some more HTTP calls: it's because the garbage collector is acting there.

[Kludex]

Got it. The contextmanager_in_threadpool fails completely, but it's the contextmanager itself that is able to conclude when the garbage collector acts, see:

async def solve_generator(
    *, call: Callable[..., Any], stack: AsyncExitStack, sub_values: Dict[str, Any]
) -> Any:
    if is_gen_callable(call):
        cm = contextmanager_in_threadpool(contextmanager(call)(**sub_values))
    elif is_async_gen_callable(call):
        cm = asynccontextmanager(call)(**sub_values)
    return await stack.enter_async_context(cm)

[Kludex]

I'm tired. I'm going to write a report tomorrow morning, but the last messages show what is happening (maybe not in a clear way).

[Kludex]

This happens in FastAPI because they change the middleware order on the build_stack_middleware, adding the AsyncExitMiddleware. That's not actually a problem, just pointing out why FastAPI is able to experience this issue. 😅

We are wrong here in breaking the receive stream.

[Kludex]

To test this, we can add a middleware wrapped by the BaseHTTPMiddleware and verify that it runs until the end.

I'll add this test tomorrow.

(I think this makes sense... 👀)

[adriangb]

Thank you for digging @Kludex !

[MrSalman333]

To test this, we can add a middleware wrapped by the BaseHTTPMiddleware and verify that it runs until the end.

I'll add this test tomorrow.

(I think this makes sense... 👀)

thanks @Kludex !!

keep in mind. this behavior is not exposed in the TestClient.

[Kludex]

One more finding... Actually, the issue doesn't exist in the latest Starlette, so FastAPI just needs to bump Starlette to 0.21.0.

@jhominal Solved this on #1715 as well... Specifically mentioned on this sentence:

[Kludex]

On my tests, I was going back in time to the PR that introduced, but once I used the latest Starlette I was not able to reproduce it... 👀

[Kludex]

We even have a test for this behavior:

starlette/tests/middleware/test_base.py

Lines 266 to 325 in bf5d735

	async def test_run_context_manager_exit_even_if_client_disconnects():
	# test for https://github.com/encode/starlette/issues/1678#issuecomment-1172916042
	request_body_sent = False
	response_complete = anyio.Event()
	context_manager_exited = anyio.Event()
	async def sleep_and_set():
	# small delay to give BaseHTTPMiddleware a chance to cancel us
	# this is required to make the test fail prior to fixing the issue
	# so do not be surprised if you remove it and the test still passes
	await anyio.sleep(0.1)
	context_manager_exited.set()
	class ContextManagerMiddleware:
	def __init__(self, app):
	self.app = app
	async def __call__(self, scope: Scope, receive: Receive, send: Send):
	async with AsyncExitStack() as stack:
	stack.push_async_callback(sleep_and_set)
	await self.app(scope, receive, send)
	async def simple_endpoint(_):
	return PlainTextResponse(background=BackgroundTask(sleep_and_set))
	async def passthrough(request, call_next):
	return await call_next(request)
	app = Starlette(
	middleware=[
	Middleware(BaseHTTPMiddleware, dispatch=passthrough),
	Middleware(ContextManagerMiddleware),
	],
	routes=[Route("/", simple_endpoint)],
	)
	scope = {
	"type": "http",
	"version": "3",
	"method": "GET",
	"path": "/",
	}
	async def receive():
	nonlocal request_body_sent
	if not request_body_sent:
	request_body_sent = True
	return {"type": "http.request", "body": b"", "more_body": False}
	# We simulate a client that disconnects immediately after receiving the response
	await response_complete.wait()
	return {"type": "http.disconnect"}
	async def send(message):
	if message["type"] == "http.response.body":
	if not message.get("more_body", False):
	response_complete.set()
	await app(scope, receive, send)
	assert context_manager_exited.is_set()

[Kludex]

In any case, I'll write a report because there was a mistake.

Some months ago, there was a PR (#1579) that tried to fix an issue reported in FastAPI (tiangolo/fastapi#4050). I was able to reproduce that issue, but I concluded that the issue was not in the GZipMiddleware (as the PR proposed), but actually in the BaseHTTPMiddleware. For that reason, I've created a PR that fixed the issue on the BaseHTTPMiddleware (#1609).

The PR introduced two changes:

1. It didn't send empty data to the StreamingResponse;
2. and it closed the receive stream.

I overstepped there, as the 2 was not needed, and actually caused an issue: inner middlewares wouldn't be able to run as expected, as there was a CancellationError. On the other hand, the 1 is fine.

On the process of merging this PR, what we lacked on the author (me) side was to create a test for the behavior change to avoid regressions. Specifically, he needed to come up with a test case like what was created on #1715. As for the reviewers side, they thought the changes were fine based on the amount of changes, and considering that there was no drop of coverage, I think it made sense to have that consideration. What the reviewers could have done differently was to insist on a test case, maybe.

We took a lot to notice this, and it's not even relevant right now because FastAPI pins each version of Starlette. Currently, latest FastAPI runs with Starlette 0.20.4, which contains the issue. The thing is, this behavior was fixed on #1715, which was released on Starlette 0.21.0. On the description of that PR, we can already see the intention of solving the case of the receive stream being cancelled.

To conclude:

• This issue doesn't exist on the latest Starlette.
• This PR is valid, but not urgent.
• There's already a test for it.
• FastAPI needs to bump Starlette to 0.21.0.

[Kludex]

This is the solution for #1977. It looks like my report above is not quite precise.

And it looks like we could benefit from another test case here... Ideas?

[tomchristie]

Given that we're not regressing any test cases, I think code removal without additional tests is okay.

[Kludex]

I still want a test to cover this, but I'll merge this as is.