fix: map a failed outbound RPC instead of passing its status to the caller

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`.

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:

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

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)
 

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"

tests/unit/aio/interceptors/test_exception_handler.py

@@ -3,29 +3,23 @@
 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,
     AsyncExceptionHandlerInterceptor,
     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,