From da2eeff7770cf151f66520bbf011103eed9f7d07 Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 16 Oct 2026 19:48:14 +0800 Subject: [PATCH] fix(grpc): mask outbound RPC errors (#86) Closes #86 --- docs/adapters/grpc.md | 8 ++- servicewright/adapters/grpc/errors.py | 11 +-- tests/integration/test_grpc_integration.py | 78 ++++++++++++++++++++++ tests/unit/test_grpc.py | 35 ++++++---- 4 files changed, 112 insertions(+), 20 deletions(-) diff --git a/docs/adapters/grpc.md b/docs/adapters/grpc.md index 31ed094..7467cbf 100644 --- a/docs/adapters/grpc.md +++ b/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 diff --git a/servicewright/adapters/grpc/errors.py b/servicewright/adapters/grpc/errors.py index d2a1277..501f3d6 100644 --- a/servicewright/adapters/grpc/errors.py +++ b/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}) diff --git a/tests/integration/test_grpc_integration.py b/tests/integration/test_grpc_integration.py index b82fcc0..f845ce2 100644 --- a/tests/integration/test_grpc_integration.py +++ b/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) diff --git a/tests/unit/test_grpc.py b/tests/unit/test_grpc.py index 1e15eba..98ce255 100644 --- a/tests/unit/test_grpc.py +++ b/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