Propagate FastAPI exceptions to framework handlers

microbootstrap/bootstrappers/faststream.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
+import functools
 import json
 import typing
 
 import prometheus_client
+import sentry_sdk
 import structlog
 import typing_extensions
 from faststream._internal.logger.logger_proxy import RealLoggerObject
@@ -26,6 +28,29 @@
 
 
 tracer: typing.Final = trace.get_tracer(__name__)
+MessageT = typing.TypeVar("MessageT")
+ResponseT = typing.TypeVar("ResponseT")
+
+
+def _with_sentry_isolation_scope(
+    process_message: typing.Callable[[MessageT], typing.Awaitable[ResponseT]],
+) -> typing.Callable[[MessageT], typing.Awaitable[ResponseT]]:
+    @functools.wraps(process_message)
+    async def isolated_process_message(message: MessageT) -> ResponseT:
+        with sentry_sdk.isolation_scope():
+            return await process_message(message)
+
+    return isolated_process_message
+
+
+def _isolate_faststream_subscribers(application: AsgiFastStream) -> None:
+    for broker in application.brokers:
+        for subscriber in broker.subscribers:
+            object.__setattr__(
+                subscriber,
+                "process_message",
+                _with_sentry_isolation_scope(subscriber.process_message),
+            )
 
 
 class KwargsAsgiFastStream(AsgiFastStream):
@@ -51,7 +76,18 @@ def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]:
         }
 
 
-FastStreamBootstrapper.use_instrument()(SentryInstrument)
+@FastStreamBootstrapper.use_instrument()
+class FastStreamSentryInstrument(SentryInstrument):
+    def bootstrap_after(self, application: AsgiFastStream) -> AsgiFastStream:  # type: ignore[override]
+        # FastStream logs handler errors after custom broker middlewares exit, so the isolation scope
+        # must enclose the subscriber's complete processing lifecycle.
+        if application.brokers:
+            _isolate_faststream_subscribers(application)
+        else:
+            application.on_startup(functools.partial(_isolate_faststream_subscribers, application))
+        return application
+
+
 FastStreamBootstrapper.use_instrument()(PyroscopeInstrument)
 
 

microbootstrap/middlewares/fastapi.py

@@ -26,10 +26,7 @@ async def dispatch(
                 return await call_next(request)
 
             start_time: typing.Final = time.perf_counter_ns()
-            try:
-                response = await call_next(request)
-            except Exception:  # noqa: BLE001
-                response = fastapi.Response(status_code=500)
+            response = await call_next(request)
 
             fill_log_message(
                 "exception" if response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR else "info",

tests/bootstrappers/test_faststream.py

@@ -1,9 +1,12 @@
+import asyncio
+import logging
 import typing
 from unittest import mock
 from unittest.mock import MagicMock
 
 import faker
 import pytest
+import sentry_sdk
 from fastapi import status
 from fastapi.testclient import TestClient
 from faststream.redis import RedisBroker, TestRedisBroker
@@ -16,6 +19,7 @@
 from microbootstrap.instruments.logging_instrument import LoggingConfig
 from microbootstrap.instruments.opentelemetry_instrument import FastStreamOpentelemetryConfig, OpentelemetryConfig
 from microbootstrap.instruments.prometheus_instrument import FastStreamPrometheusConfig
+from microbootstrap.instruments.sentry_instrument import SentryConfig
 from microbootstrap.settings import FastStreamSettings
 
 
@@ -124,3 +128,93 @@ async def test_faststream_logging(broker: RedisBroker, minimal_logging_config: L
     FastStreamBootstrapper(FastStreamSettings()).configure_application(
         FastStreamConfig(broker=broker)
     ).configure_instruments(minimal_logging_config).bootstrap()
+
+
+async def test_faststream_sentry_isolates_concurrent_messages(
+    broker: RedisBroker,
+    minimal_sentry_config: SentryConfig,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    channel: typing.Final = "test-channel"
+    conversation_id_tag: typing.Final = "conversation_id"
+    first_started = asyncio.Event()
+    second_started = asyncio.Event()
+    second_logged = asyncio.Event()
+    captured_tags: dict[str, str | None] = {}
+    monkeypatch.setattr(sentry_sdk, "init", mock.Mock())
+    minimal_sentry_config.sentry_tags = None
+
+    @broker.subscriber(channel)
+    async def handler(conversation_id: str) -> None:
+        sentry_sdk.get_isolation_scope().set_tag(conversation_id_tag, conversation_id)
+        if conversation_id == "first":
+            first_started.set()
+            await second_started.wait()
+            await second_logged.wait()
+        else:
+            second_started.set()
+            await first_started.wait()
+        raise ValueError(conversation_id)
+
+    FastStreamBootstrapper(FastStreamSettings()).configure_application(
+        FastStreamConfig(broker=broker)
+    ).configure_instruments(minimal_sentry_config).bootstrap()
+
+    original_log = broker.config.logger.log
+
+    def record_error_tag(*args: typing.Any, **kwargs: typing.Any) -> None:  # noqa: ANN401
+        if kwargs.get("log_level") == logging.ERROR:
+            error_message = typing.cast("str", kwargs["message"])
+            captured_tags[error_message] = sentry_sdk.get_isolation_scope()._tags.get(conversation_id_tag)  # noqa: SLF001
+            if error_message.endswith("second"):
+                second_logged.set()
+        original_log(*args, **kwargs)
+
+    monkeypatch.setattr(broker.config.logger, "log", record_error_tag)
+    event_loop = asyncio.get_running_loop()
+    previous_exception_handler = event_loop.get_exception_handler()
+    event_loop.set_exception_handler(lambda *_: None)
+    try:
+        async with TestRedisBroker(broker):
+            errors: typing.Final = await asyncio.gather(
+                broker.publish("first", channel),
+                broker.publish("second", channel),
+                return_exceptions=True,
+            )
+    finally:
+        event_loop.set_exception_handler(previous_exception_handler)
+
+    assert all(isinstance(error, ValueError) for error in errors)
+    assert captured_tags == {
+        "ValueError: first": "first",
+        "ValueError: second": "second",
+    }
+    assert conversation_id_tag not in sentry_sdk.get_isolation_scope()._tags  # noqa: SLF001
+
+
+async def test_faststream_sentry_isolates_broker_configured_on_startup(
+    broker: RedisBroker,
+    minimal_sentry_config: SentryConfig,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    application: typing.Any
+
+    @broker.subscriber("test-channel")
+    async def handler(message: str) -> None:
+        pass
+
+    def set_broker() -> None:
+        application.set_broker(broker)
+
+    monkeypatch.setattr(sentry_sdk, "init", mock.Mock())
+    monkeypatch.setattr(broker, "start", mock.AsyncMock())
+    application = (
+        FastStreamBootstrapper(FastStreamSettings())
+        .configure_application(FastStreamConfig(on_startup=[set_broker]))
+        .configure_instruments(minimal_sentry_config)
+        .bootstrap()
+    )
+
+    await application.start()
+
+    assert hasattr(broker.subscribers[0].process_message, "__wrapped__")

tests/instruments/test_logging.py

@@ -6,6 +6,7 @@
 import fastapi
 import litestar
 import pytest
+from fastapi.responses import JSONResponse
 from fastapi.testclient import TestClient as FastAPITestClient
 from faststream.redis import RedisBroker, TestRedisBroker
 from litestar.testing import TestClient as LitestarTestClient
@@ -181,6 +182,33 @@ async def test_handler() -> str:
     assert fill_log_mock.call_count == 2  # noqa: PLR2004
 
 
+def test_fastapi_logging_bootstrap_propagates_unhandled_exception(
+    monkeypatch: pytest.MonkeyPatch,
+    minimal_logging_config: LoggingConfig,
+) -> None:
+    fastapi_application: typing.Final = fastapi.FastAPI()
+
+    @fastapi_application.get("/test-error-handler")
+    async def test_error_handler() -> None:
+        raise RuntimeError("test error")
+
+    @fastapi_application.exception_handler(Exception)
+    async def handle_exception(_request: fastapi.Request, exception: Exception) -> JSONResponse:
+        return JSONResponse(status_code=418, content={"detail": str(exception)})
+
+    logging_instrument: typing.Final = FastApiLoggingInstrument(minimal_logging_config)
+    logging_instrument.bootstrap()
+    logging_instrument.bootstrap_after(fastapi_application)
+    monkeypatch.setattr("microbootstrap.middlewares.fastapi.fill_log_message", fill_log_mock := mock.Mock())
+
+    with FastAPITestClient(app=fastapi_application, raise_server_exceptions=False) as test_client:
+        response: typing.Final = test_client.get("/test-error-handler")
+
+    assert response.status_code == 418  # noqa: PLR2004
+    assert response.json() == {"detail": "test error"}
+    fill_log_mock.assert_not_called()
+
+
 def test_fastapi_logging_bootstrap_ignores_health(
     monkeypatch: pytest.MonkeyPatch, minimal_logging_config: LoggingConfig
 ) -> None:

tests/instruments/test_sentry.py

@@ -4,13 +4,16 @@
 import typing
 from unittest import mock
 
+import fastapi
 import litestar
 import pytest
 import structlog
+from fastapi.testclient import TestClient as FastAPITestClient
 from litestar.testing import TestClient as LitestarTestClient
 from opentelemetry import baggage
 from opentelemetry.context import Context, attach, detach
 
+from microbootstrap.bootstrappers.fastapi import FastApiLoggingInstrument
 from microbootstrap.bootstrappers.litestar import LitestarSentryInstrument
 from microbootstrap.instruments.logging_instrument import LoggingConfig, LoggingInstrument
 from microbootstrap.instruments.sentry_instrument import (
@@ -80,6 +83,37 @@ async def error_handler() -> None:
         assert mock_capture_event.called
 
 
+def test_fastapi_sentry_captures_unhandled_exception_with_traceback(
+    minimal_logging_config: LoggingConfig,
+    minimal_sentry_config: SentryConfig,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr("sentry_sdk.Scope.capture_event", capture_event := mock.Mock())
+    SentryInstrument(minimal_sentry_config).bootstrap()
+    fastapi_application: typing.Final = fastapi.FastAPI()
+
+    @fastapi_application.get("/test-error-handler")
+    async def error_handler() -> None:
+        raise RuntimeError("test error")
+
+    logging_instrument: typing.Final = FastApiLoggingInstrument(minimal_logging_config)
+    logging_instrument.bootstrap()
+    logging_instrument.bootstrap_after(fastapi_application)
+
+    with FastAPITestClient(app=fastapi_application, raise_server_exceptions=False) as test_client:
+        response: typing.Final = test_client.get("/test-error-handler")
+
+    assert response.status_code == fastapi.status.HTTP_500_INTERNAL_SERVER_ERROR
+    captured_events: typing.Final = [call.args[0] for call in capture_event.mock_calls]
+    exception_events: typing.Final = [event for event in captured_events if event.get("exception")]
+    assert len(exception_events) == 1
+    exception_event: typing.Final = exception_events[0]
+    exception_value: typing.Final = exception_event["exception"]["values"][-1]
+    assert exception_value["type"] == "RuntimeError"
+    assert exception_value["value"] == "test error"
+    assert exception_value["stacktrace"]["frames"]
+
+
 class TestSentryEnrichEventFromStructlog:
     @pytest.mark.parametrize(
         "event",