From 0abf79f8991a7395715e5a4ce33936d00bf8dff4 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:31:49 +0400 Subject: [PATCH] fix: map a failed outbound RPC instead of passing its status to the caller AsyncExceptionHandlerInterceptor re-raised every grpc.RpcError untouched. grpc.aio.AbortError is not an RpcError, so that clause only ever let through grpc.aio.AioRpcError from calls the handler made, and grpc.aio answered the caller with UNKNOWN and the downstream error's repr as details. Such an error now goes through the status map like any other exception: INTERNAL with the safe detail by default, logged with its traceback. Mapping grpc.RpcError in error_status_map picks another status. Fixes #48 --- docs/agents.md | 12 ++-- docs/guide/interceptors.md | 25 ++++++- docs/guide/observability.md | 4 +- .../aio/interceptors/exception_handler.py | 12 ++-- tests/integration/test_interceptor_chain.py | 52 ++++++++++++++- .../interceptors/test_exception_handler.py | 65 ++++++++++++++++--- 6 files changed, 149 insertions(+), 21 deletions(-) diff --git a/docs/agents.md b/docs/agents.md index a8fb090..5628932 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -356,7 +356,11 @@ The kit's defaults live in `grpc_server_kit.constants` as `DEFAULT_*` names, alo 4. **`context.abort()` raises `grpc.aio.AbortError`, which is not a `grpc.RpcError`.** `except grpc.RpcError:` does not catch a deliberate abort; `except Exception:` does. An interceptor that logs and re-raises everything will fire on every intentional - abort unless it filters `AbortError` out first. + abort unless it filters `AbortError` out first. The reverse holds too: a + `grpc.RpcError` in a server interceptor is a failed *outbound* call, not a status this + RPC chose. `AsyncExceptionHandlerInterceptor` maps it to `INTERNAL` (or whatever + `grpc.RpcError` maps to in your `error_status_map`); re-raising it untouched sends the + caller `UNKNOWN` with the downstream error's details. 5. **`around_call` is an async generator, not a coroutine.** It must contain exactly one `yield`; the RPC — including the complete consumption of a response stream — happens there. A body with no `yield` anywhere is a plain coroutine, and every RPC through @@ -485,9 +489,7 @@ except (asyncio.CancelledError, KeyboardInterrupt, SystemExit): raise except grpc.aio.AbortError: raise # deliberate; the context carries the real status -except grpc.RpcError: - raise -except Exception: +except Exception: # includes a failed outbound call (grpc.RpcError) metrics.failure() raise ``` @@ -547,7 +549,7 @@ builder.with_servicers(register) | `PermissionError` | A private key readable beyond its owner, or a TLS file that cannot be read | | `OSError` | Any other I/O failure reading a TLS file | | `grpc.aio.AbortError` | Raised by `context.abort()` — by your handler, by `AsyncExceptionHandlerInterceptor`, by `AsyncContextInterceptor` on a missing required header, or by the health servicer aborting a broken `Watch` | -| `grpc.RpcError` | gRPC's own transport-level errors, passed through untouched | +| `grpc.RpcError` | Raised by a *client* call (`grpc.aio.AioRpcError`) when another service fails. One that escapes your handler is mapped by `AsyncExceptionHandlerInterceptor` like any other exception — `INTERNAL` by default, never the downstream status or details | Pydantic's `ValidationError` replaces `ValueError` when the settings object is a `BaseGrpcServerSettings`. diff --git a/docs/guide/interceptors.md b/docs/guide/interceptors.md index ca68518..73e851f 100644 --- a/docs/guide/interceptors.md +++ b/docs/guide/interceptors.md @@ -108,11 +108,34 @@ except (asyncio.CancelledError, KeyboardInterrupt, SystemExit): except grpc.aio.AbortError: raise # deliberate — the context already carries the real status except grpc.RpcError: - raise # already a proper gRPC error + raise # an outbound call failed — see below except Exception as exc: ... # genuinely unexpected — map it, log it, capture it ``` +A `grpc.RpcError` reaching an interceptor is never a status this RPC chose. +The only subclass grpcio defines is `grpc.aio.AioRpcError`, which a *client* +call raises: the handler called another service and that call failed. Its +`code()` and `details()` belong to the other service. The logging, tracing +and metrics interceptors read its `code()` when no exception handler sits +inside them; `AsyncExceptionHandlerInterceptor` does not pass it through, it +maps it like any other exception. Left alone, `grpc.aio` would answer the +caller with `UNKNOWN` and the downstream error's `repr` as details. + +With the default map that means `INTERNAL` and "Internal server error". The +downstream status describes this server's call, not the caller's request: a +downstream `INVALID_ARGUMENT` or `UNAUTHENTICATED` means this server sent a +bad request or holds bad credentials. To report every failed outbound call +with another status, map `grpc.RpcError` like any exception type: + +```python +AsyncExceptionHandlerInterceptor({grpc.RpcError: grpc.StatusCode.UNAVAILABLE}) +``` + +To forward a particular downstream status (say, a `NOT_FOUND` that really is +the caller's), catch the `AioRpcError` in the handler and `context.abort()` +with the status you mean. + ## The canonical chain The six shipped interceptors, outermost first: diff --git a/docs/guide/observability.md b/docs/guide/observability.md index 91690db..e5c8ba4 100644 --- a/docs/guide/observability.md +++ b/docs/guide/observability.md @@ -105,7 +105,9 @@ process-global SDK state: without a fresh scope per RPC, tags and breadcrumbs from concurrent calls would bleed into each other's error reports. It only captures raw `Exception`s that reach it — `grpc.aio.AbortError` and -`grpc.RpcError` are re-raised untouched, never captured. Combined with the +`grpc.RpcError` are re-raised untouched, never captured. A `grpc.RpcError` +here is a failed outbound call; the exception handler logs it at `ERROR` with +its traceback and answers the caller with a masked status. Combined with the [canonical chain order](interceptors.md#the-canonical-chain), this is why Sentry's position matters: placed inside (closer to the handler than) `AsyncExceptionHandlerInterceptor`, it sees the exact exception the handler diff --git a/grpc_server_kit/aio/interceptors/exception_handler.py b/grpc_server_kit/aio/interceptors/exception_handler.py index e1ec4a5..71fe82c 100644 --- a/grpc_server_kit/aio/interceptors/exception_handler.py +++ b/grpc_server_kit/aio/interceptors/exception_handler.py @@ -97,13 +97,19 @@ class AsyncExceptionHandlerInterceptor(AsyncServerInterceptor): """Interceptor that catches exceptions and converts them to gRPC errors. Rules: - - ``grpc.aio.AbortError`` (the handler already aborted with its own status) and - ``grpc.RpcError`` are re-raised as-is. + - ``grpc.aio.AbortError`` (the handler already aborted with its own status) is + re-raised as-is. - ``CancelledError`` / ``KeyboardInterrupt`` / ``SystemExit`` are re-raised. - Any other exception is mapped to a ``grpc.StatusCode`` (via the MRO) and the request is aborted with details from the ``detail_factory``. Unmapped exceptions default to ``INTERNAL`` with a safe message (no internal details leaked). + A ``grpc.RpcError`` raised by an outbound call the handler made (``grpc.aio.AioRpcError``) + is one of those "other" exceptions: its status and details belong to the downstream + service, so it is mapped like anything else (``INTERNAL`` by default) instead of reaching + the caller as ``UNKNOWN`` with the downstream details. Map ``grpc.RpcError`` in + ``error_status_map`` to report downstream failures with another status. + Covers streaming RPCs: exceptions raised while producing a response stream are mapped and aborted the same way as unary failures. """ @@ -142,8 +148,6 @@ async def around_call(self, call: RpcCall) -> AsyncIterator[None]: # The handler (or an inner interceptor) already aborted this RPC with # its own status — pass it through untouched. raise - except grpc.RpcError: - raise except Exception as exc: await self._abort_with_mapped_status(exc, call) diff --git a/tests/integration/test_interceptor_chain.py b/tests/integration/test_interceptor_chain.py index 14f76a4..939677a 100644 --- a/tests/integration/test_interceptor_chain.py +++ b/tests/integration/test_interceptor_chain.py @@ -8,7 +8,10 @@ import grpc import pytest -from .conftest import SERVICE, RunningServer +from grpc_server_kit import GrpcApp +from grpc_server_kit.aio.interceptors import AsyncExceptionHandlerInterceptor, AsyncMetricsInterceptor + +from .conftest import SERVICE, RecordingMetrics, RunningServer pytestmark = pytest.mark.integration @@ -152,3 +155,50 @@ async def test__interceptor_chain__unary_stream_error_mid_stream__maps_status_af record = running_server.metrics.for_method(f"/{SERVICE}/StreamFailMid") assert record["status"] == "error" assert record["grpc_code"] == "INVALID_ARGUMENT" + + +async def test__interceptor_chain__outbound_call_fails__caller_gets_internal_not_downstream_status( + running_server: RunningServer, +) -> None: + # Arrange + # A second service whose handler calls running_server, which aborts with + # PERMISSION_DENIED / "handler-abort". That status describes this service's + # call, not its caller's request, so it must not reach the caller. + proxy_service = "test.Proxy" + + async def call_downstream(request: bytes, _context: grpc.aio.ServicerContext[bytes, bytes]) -> bytes: + async with grpc.aio.insecure_channel(running_server.target()) as downstream: + response: bytes = await downstream.unary_unary(f"/{SERVICE}/AbortDirectly")(request, timeout=5) + return response + + metrics = RecordingMetrics() + proxy = GrpcApp( + host="127.0.0.1", + port=0, + interceptors=[ + AsyncMetricsInterceptor(metrics=metrics, service_name=proxy_service), + AsyncExceptionHandlerInterceptor(), + ], + ) + proxy.register( + lambda raw: raw.add_generic_rpc_handlers( + ( + grpc.method_handlers_generic_handler( + proxy_service, {"CallDownstream": grpc.unary_unary_rpc_method_handler(call_downstream)} + ), + ) + ) + ) + + # Act + async with proxy, grpc.aio.insecure_channel(f"127.0.0.1:{proxy.bound_port}") as channel: + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + await channel.unary_unary(f"/{proxy_service}/CallDownstream")(b"x", timeout=5) + + # Assert + assert exc_info.value.code() == grpc.StatusCode.INTERNAL + assert exc_info.value.details() == "Internal server error" + + record = metrics.for_method(f"/{proxy_service}/CallDownstream") + assert record["status"] == "error" + assert record["grpc_code"] == "INTERNAL" diff --git a/tests/unit/aio/interceptors/test_exception_handler.py b/tests/unit/aio/interceptors/test_exception_handler.py index f487200..a8e42a4 100644 --- a/tests/unit/aio/interceptors/test_exception_handler.py +++ b/tests/unit/aio/interceptors/test_exception_handler.py @@ -3,11 +3,11 @@ from __future__ import annotations import asyncio +import logging from unittest.mock import MagicMock import grpc import pytest -from pytest_lazy_fixtures import lf from grpc_server_kit.aio.interceptors.exception_handler import ( GRPC_SAFE_ERROR_MESSAGES, @@ -15,17 +15,11 @@ default_error_detail, ) -from .conftest import RESPONSE, FailingMethodFactory, MakeContext, MakeRpcError, OkMethod, RunUnary +from .conftest import RESPONSE, FailingMethodFactory, MakeContext, OkMethod, RunUnary pytestmark = pytest.mark.unit -@pytest.fixture -def rpc_error(make_rpc_error: MakeRpcError) -> grpc.RpcError: - """A bare RpcError, e.g. one surfaced by an upstream call — not this interceptor's concern.""" - return make_rpc_error() - - async def test__exception_handler__no_exception__returns_response( make_context: MakeContext, ok_method: OkMethod, @@ -44,7 +38,6 @@ async def test__exception_handler__no_exception__returns_response( @pytest.mark.parametrize( "exc", [ - pytest.param(lf("rpc_error"), id="rpc_error"), # The handler already aborted with its own status: no re-mapping, no second abort. pytest.param(grpc.aio.AbortError(), id="abort_error"), pytest.param(asyncio.CancelledError(), id="cancelled"), @@ -94,6 +87,60 @@ async def test__exception_handler__default_map__aborts_with_mapped_status( assert ctx.abort.call_args.args[1] == GRPC_SAFE_ERROR_MESSAGES[expected_status] +def _outbound_rpc_error(code: grpc.StatusCode, details: str) -> grpc.aio.AioRpcError: + """The error an outbound call made by the handler raises when the other service fails.""" + return grpc.aio.AioRpcError(code, grpc.aio.Metadata(), grpc.aio.Metadata(), details=details) + + +@pytest.mark.parametrize( + "downstream_code", + [ + pytest.param(grpc.StatusCode.INTERNAL, id="downstream_internal"), + # The downstream status describes this server's call, not the caller's request: + # forwarding PERMISSION_DENIED would blame the caller for this server's credentials. + pytest.param(grpc.StatusCode.PERMISSION_DENIED, id="downstream_permission_denied"), + ], +) +async def test__exception_handler__outbound_rpc_error__aborts_internal_without_downstream_details( + make_context: MakeContext, + failing_method: FailingMethodFactory, + run_unary: RunUnary, + caplog: pytest.LogCaptureFixture, + downstream_code: grpc.StatusCode, +) -> None: + # Arrange + interceptor = AsyncExceptionHandlerInterceptor() + ctx = make_context() + error = _outbound_rpc_error(downstream_code, "ledger dsn=postgres://ledger:s3cret@10.0.3.7/ledger") + + # Act + with caplog.at_level(logging.ERROR), pytest.raises(grpc.aio.AbortError): + await run_unary(interceptor, failing_method(error), MagicMock(), ctx) + + # Assert + ctx.abort.assert_awaited_once_with(grpc.StatusCode.INTERNAL, "Internal server error") + # The real failure goes to the log with its traceback, as for any other server error. + (record,) = caplog.records + assert record.exc_info is not None + assert record.exc_info[1] is error + + +async def test__exception_handler__rpc_error_in_status_map__aborts_with_mapped_status( + make_context: MakeContext, + failing_method: FailingMethodFactory, + run_unary: RunUnary, +) -> None: + # Arrange + interceptor = AsyncExceptionHandlerInterceptor({grpc.RpcError: grpc.StatusCode.UNAVAILABLE}) + ctx = make_context() + error = _outbound_rpc_error(grpc.StatusCode.UNAVAILABLE, "connection refused") + + # Act & Assert + with pytest.raises(grpc.aio.AbortError): + await run_unary(interceptor, failing_method(error), MagicMock(), ctx) + ctx.abort.assert_awaited_once_with(grpc.StatusCode.UNAVAILABLE, "Service unavailable") + + async def test__exception_handler__custom_status_map__overrides_default_mapping( make_context: MakeContext, failing_method: FailingMethodFactory,