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,