fix(grpc): mask outbound RPC errors

docs/adapters/grpc.md

@@ -106,9 +106,11 @@ indistinguishable to a caller. Without it `grpc.aio` answers `UNKNOWN` with `rep
 exception, so the one error class whose wording nobody reviewed is the one that reaches the wire
 verbatim.
 
-Three things are left alone, because each already carries a decision:
-`grpc.aio.AbortError` and `grpc.RpcError` (a status a handler chose itself) are re-raised, and
-`asyncio.CancelledError` never reaches it — a caller that walked away is not a failure to report.
+An explicit `context.abort()` raises `grpc.aio.AbortError`, so its chosen status passes through.
+A client-side `grpc.aio.AioRpcError` from an outbound call is a failure at this server boundary;
+it is logged and masked like any other unhandled exception, keeping the downstream status and
+details off the caller-facing response. `asyncio.CancelledError` never reaches the interceptor;
+a caller that walked away is not a failure to report.
 
 `map_service_errors=False` does not remove it. That flag turns off the mapping of the errors you
 declared; with it off a `ServiceError` reaches the client as a masked `INTERNAL` like any other

servicewright/adapters/grpc/errors.py

@@ -105,17 +105,18 @@ class UnhandledErrorInterceptor(AsyncServerInterceptor):
     :class:`GrpcEntrypoint` adds it inside :class:`UnitScopeInterceptor` — so
     the log record still carries the correlation ids — and outside your own
     interceptors, so an exception type you map yourself reaches your
-    interceptor first and never gets here. A deliberate status passes through:
-    ``grpc.aio.AbortError`` and ``grpc.RpcError`` are re-raised, and
-    ``asyncio.CancelledError`` is a ``BaseException``, so a caller that walks
-    away is not an error to report.
+    interceptor first and never gets here. ``grpc.aio.AbortError`` is re-raised
+    because the servicer deliberately chose its status. A client-side
+    ``grpc.aio.AioRpcError`` from an outbound call is masked like any other
+    unhandled exception. ``asyncio.CancelledError`` is a ``BaseException``, so
+    a caller that walks away is not an error to report.
     """
 
     async def around_call(self, call: RpcCall) -> AsyncIterator[None]:
         """Run the RPC; mask anything that comes out of it without a status."""
         try:
             yield
-        except (grpc.aio.AbortError, grpc.RpcError):
+        except grpc.aio.AbortError:
             raise
         except Exception:
             logger.exception("Unhandled exception while serving RPC", extra={"grpc_method": call.method_name})

tests/integration/test_grpc_integration.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from typing import Any
 
 import grpc
@@ -107,3 +108,80 @@ async def call_then_stop() -> None:
     assert error.details() == "internal_error"
     assert dict(error.trailing_metadata() or ())["x-error-code"] == "internal_error"
     assert _SECRET not in str(error.details())
+
+
+_DOWNSTREAM_SERVICE = "lab.Ledger"
+_DOWNSTREAM_DETAIL = "private downstream detail"
+
+
+def _register_aborting_servicer(server: Any, _ctx: Any) -> None:
+    async def lookup(request: bytes, context: Any) -> bytes:
+        await context.abort(grpc.StatusCode.INTERNAL, _DOWNSTREAM_DETAIL)
+
+    server.add_generic_rpc_handlers(
+        (
+            grpc.method_handlers_generic_handler(
+                _DOWNSTREAM_SERVICE,
+                {"Lookup": grpc.unary_unary_rpc_method_handler(lookup)},
+            ),
+        )
+    )
+
+
+def _register_outbound_calling_servicer(server: Any, _ctx: Any, downstream: GrpcEntrypoint) -> None:
+    async def pay(request: bytes, context: Any) -> bytes:
+        port = downstream.bound_port
+        if port is None:
+            raise RuntimeError("downstream gRPC server has not bound")
+        async with grpc.aio.insecure_channel(f"127.0.0.1:{port}") as channel:
+            return await channel.unary_unary(f"/{_DOWNSTREAM_SERVICE}/Lookup")(request, timeout=2)
+
+    server.add_generic_rpc_handlers(
+        (grpc.method_handlers_generic_handler(_SERVICE, {"Pay": grpc.unary_unary_rpc_method_handler(pay)}),)
+    )
+
+
+async def test__grpc_entrypoint__outbound_rpc_fails__client_gets_a_masked_internal(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # Arrange
+    spec: AppSpec[Any, Any] = AppSpec(
+        service_name="grpc-outbound-errors-it", create_container=lambda _s: FakeContainer()
+    )
+    config = GrpcConfig(host="127.0.0.1", port=0, enable_reflection=False, enable_channelz=False)
+    downstream = GrpcEntrypoint(config=config, servicers=_register_aborting_servicer)
+
+    def register_proxy(server: Any, ctx: Any) -> None:
+        _register_outbound_calling_servicer(server, ctx, downstream)
+
+    proxy = GrpcEntrypoint(config=config, servicers=register_proxy)
+    service = Service(spec, entrypoints=[downstream, proxy])
+    stop = asyncio.Event()
+    failures: list[grpc.aio.AioRpcError] = []
+
+    async def call_then_stop() -> None:
+        while not (spec.health.ready and downstream.bound_port and proxy.bound_port):
+            await asyncio.sleep(0.01)
+        await asyncio.sleep(0.05)
+
+        try:
+            async with grpc.aio.insecure_channel(f"127.0.0.1:{proxy.bound_port}") as channel:
+                await channel.unary_unary(f"/{_SERVICE}/Pay")(b"42", timeout=5)
+        except grpc.aio.AioRpcError as error:
+            failures.append(error)
+        finally:
+            stop.set()
+
+    # Act
+    with caplog.at_level(logging.ERROR, logger="servicewright.adapters.grpc.errors"):
+        await asyncio.gather(service.run(FakeSettings(), stop=stop), call_then_stop())
+
+    # Assert — the downstream status and details stay inside this service boundary.
+    assert len(failures) == 1
+    error = failures[0]
+    assert error.code() is grpc.StatusCode.INTERNAL
+    assert error.details() == "internal_error"
+    assert dict(error.trailing_metadata() or ()) == {"x-error-code": "internal_error"}
+    record = next(r for r in caplog.records if "Unhandled exception while serving RPC" in r.message)
+    assert record.exc_info is not None
+    assert isinstance(record.exc_info[1], grpc.aio.AioRpcError)

tests/unit/test_grpc.py

@@ -930,30 +930,41 @@ async def test__unhandled_error_interceptor__service_error__masks_it_too() -> No
     assert "user_missing" not in details
 
 
-@pytest.mark.parametrize(
-    "deliberate",
-    [
-        pytest.param(grpc.aio.AbortError("already aborted"), id="abort-error"),
-        pytest.param(grpc.RpcError("already failed"), id="rpc-error"),
-    ],
-)
-async def test__unhandled_error_interceptor__status_already_chosen__passes_it_through(
-    deliberate: Exception,
-) -> None:
+async def test__unhandled_error_interceptor__abort_error__passes_it_through() -> None:
     # Arrange
     from servicewright.adapters.grpc import UnhandledErrorInterceptor
 
     context = _AbortRecordingContext()
 
     # Act
-    with pytest.raises(type(deliberate)):
+    with pytest.raises(grpc.aio.AbortError):
         async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")):
-            raise deliberate
+            raise grpc.aio.AbortError("already aborted")
 
     # Assert
     assert context.aborts == []
 
 
+async def test__unhandled_error_interceptor__rpc_error__masks_it() -> None:
+    # Arrange
+    import grpc as grpc_lib
+
+    from servicewright.adapters.grpc import ERROR_CODE_TRAILING_METADATA, UnhandledErrorInterceptor
+
+    context = _AbortRecordingContext()
+
+    # Act
+    with pytest.raises(_AbortedSentinelError):
+        async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")):
+            raise grpc_lib.RpcError("downstream failure")
+
+    # Assert
+    (code, details, trailing) = context.aborts[0]
+    assert code == grpc_lib.StatusCode.INTERNAL
+    assert details == "internal_error"
+    assert trailing == ((ERROR_CODE_TRAILING_METADATA, "internal_error"),)
+
+
 async def test__unhandled_error_interceptor__caller_cancelled__passes_it_through() -> None:
     # Arrange
     from servicewright.adapters.grpc import UnhandledErrorInterceptor