Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Raise exception from background task on BaseHTTPMiddleware #2812

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Raise exception from background task on BaseHTTPMiddleware #2812

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c7f1629
Raise exception from background task on BaseHTTPMiddleware
Kludex Dec 26, 2024
7ef64c9
Use test client factory
Kludex Dec 26, 2024
ebe0a95
Merge branch 'master' into raise-exception-on-bkg-task
Kludex Dec 28, 2024
b4caa4e
Merge branch 'master' into raise-exception-on-bkg-task
graingert Dec 29, 2024
a48bcda
Update starlette/middleware/base.py
graingert Dec 29, 2024
069ad2e
lint
graingert Dec 29, 2024
558b389
collapse exceptions groups from streaming response
graingert Dec 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions starlette/middleware/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,9 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
request = _CachedRequest(scope, receive)
wrapped_receive = request.wrapped_receive
response_sent = anyio.Event()
app_exc: Exception | None = None

async def call_next(request: Request) -> Response:
app_exc: Exception | None = None

async def receive_or_disconnect() -> Message:
if response_sent.is_set():
return {"type": "http.disconnect"}
Expand Down Expand Up @@ -165,9 +164,6 @@ async def body_stream() -> typing.AsyncGenerator[bytes, None]:
if not message.get("more_body", False):
break

if app_exc is not None:
raise app_exc

response = _StreamingResponse(status_code=message["status"], content=body_stream(), info=info)
response.raw_headers = message["headers"]
return response
Expand All @@ -180,6 +176,9 @@ async def body_stream() -> typing.AsyncGenerator[bytes, None]:
response_sent.set()
recv_stream.close()

if app_exc is not None:
raise app_exc

async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
raise NotImplementedError() # pragma: no cover

Expand Down
14 changes: 8 additions & 6 deletions starlette/responses.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import anyio
import anyio.to_thread

from starlette._utils import collapse_excgroups
from starlette.background import BackgroundTask
from starlette.concurrency import iterate_in_threadpool
from starlette.datastructures import URL, Headers, MutableHeaders
Expand Down Expand Up @@ -258,14 +259,15 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
except OSError:
raise ClientDisconnect()
else:
async with anyio.create_task_group() as task_group:
with collapse_excgroups():
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why was this added?

Copy link
Member

@graingert graingert Dec 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's subtle but it's needed to get test_custom_middleware to pass

the reason is app_exc shouldn't be subject to collapse_excgroups (it's outside of with recv_stream, send_stream, collapse_excgroups():)- eg it's from the user's code so might have use a TG or Nursery and therefore legitimately wrapped in a EG

however that means if something raises an exception from StreamingResponse they're not really expecting an (optional!) background task to watch for disconnections, so they expect the EG to be unwrapped

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mention that it's an optional background task in StreamingResponse, because it should not be the case that StreamingResponse raises a different kind of exception based on asgi version

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't want to change anything outside the base http middleware because of it

Copy link
Member

@graingert graingert Dec 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can nest the raise app_exc inside the collapse_excgroups() in base.py - but it's not really correct

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have a think

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you have an MRE to share of a code that would be a problem with the raise app_exc inside the collapse_excgroups?

The PR you are working in seems to also change the StreamingResponse, so it's not really avoiding changes in it.

Copy link
Member

@graingert graingert Dec 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so the raise app_exc can't be put inside the async with of the task group because the app_exc might not be set until after the task group has finished, and you should not put anything in between a task group and collapse_excgroups because then you're collapsing exc groups that you've not created. Thus it has to be this way.

If you want to avoid touching StreamingResponse in this PR, I've separated the changes out into another PR that's more like the old anyio 3 behaviour and strict_exception_groups=False

Copy link
Member

@graingert graingert Dec 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's a discussion ongoing in the trio gitter to make strict_exception_groups not deprecated and expose it in anyio, then I'd make a PR to use that where available

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you have an MRE to share of a code that would be a problem with the raise app_exc inside the collapse_excgroups?

The PR you are working in seems to also change the StreamingResponse, so it's not really avoiding changes in it.

MRE converted to test cases and in #2830

async with anyio.create_task_group() as task_group:

async def wrap(func: typing.Callable[[], typing.Awaitable[None]]) -> None:
await func()
task_group.cancel_scope.cancel()
async def wrap(func: typing.Callable[[], typing.Awaitable[None]]) -> None:
await func()
task_group.cancel_scope.cancel()

task_group.start_soon(wrap, partial(self.stream_response, send))
await wrap(partial(self.listen_for_disconnect, receive))
task_group.start_soon(wrap, partial(self.stream_response, send))
await wrap(partial(self.listen_for_disconnect, receive))

if self.background is not None:
await self.background()
Expand Down
23 changes: 23 additions & 0 deletions tests/middleware/test_base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,29 @@ async def send(message: Message) -> None:
assert background_task_run.is_set()


def test_run_background_tasks_raise_exceptions(test_client_factory: TestClientFactory) -> None:
# test for https://github.com/encode/starlette/issues/2625

async def sleep_and_set() -> None:
await anyio.sleep(0.1)
raise ValueError("TEST")

async def endpoint_with_background_task(_: Request) -> PlainTextResponse:
return PlainTextResponse(background=BackgroundTask(sleep_and_set))

async def passthrough(request: Request, call_next: RequestResponseEndpoint) -> Response:
return await call_next(request)

app = Starlette(
middleware=[Middleware(BaseHTTPMiddleware, dispatch=passthrough)],
routes=[Route("/", endpoint_with_background_task)],
)

client = test_client_factory(app)
with pytest.raises(ValueError, match="TEST"):
client.get("/")


@pytest.mark.anyio
async def test_do_not_block_on_background_tasks() -> None:
response_complete = anyio.Event()
Expand Down
Loading