From a3f06cc645b9affbf201684084530332b8a593f5 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Mon, 7 Sep 2026 15:17:46 +0300 Subject: [PATCH 1/2] fix: an unhandled exception over gRPC is masked instead of sent as UNKNOWN plus its repr MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ServiceErrorInterceptor claimed ServiceError and everything else fell through to grpc.aio, whose answer for an exception it was never told about is UNKNOWN with repr() of it. The HTTP side masks the same exception to a 500 with code=internal_error and keeps the message in the log, so the promise of the shared error model held for the errors a service declared and broke for the ones it did not — and those are the ones whose wording nobody reviewed. In the service the report came from, that text carried a DSN with a password. UnhandledErrorInterceptor is the gRPC counterpart of the HTTP stack's UnhandledErrorMiddleware and sits at the same depth: inside UnitScopeInterceptor so the traceback is logged with the RPC's correlation ids, and outside the metrics interceptor and the caller's own, so an exception type you map yourself still reaches your interceptor first. It aborts with INTERNAL, detail internal_error and x-error-code: internal_error, which is byte for byte what a public=False ServiceError already produced, and re-raises AbortError and RpcError — a status somebody chose deliberately. CancelledError is a BaseException and never reaches it. It is installed unconditionally, exactly as its HTTP counterpart is. map_service_errors=False still turns off only the mapping of declared errors; an unmapped ServiceError then arrives as a masked INTERNAL rather than as its own message. ErrorKind.CONFLICT keeps mapping to ALREADY_EXISTS. ABORTED also renders as 409 but tells clients to retry at a higher level, which a state conflict will not survive, and FAILED_PRECONDITION is spoken for by PRECONDITION_FAILED; the reason now sits next to the table instead of nowhere. Refs #50 --- servicewright/adapters/grpc/__init__.py | 8 +- servicewright/adapters/grpc/entrypoint.py | 23 ++- servicewright/adapters/grpc/errors.py | 94 ++++++++--- tests/integration/test_grpc_integration.py | 57 ++++++- tests/unit/test_error_parity.py | 185 +++++++++++++++++++++ tests/unit/test_grpc.py | 157 ++++++++++++++++- 6 files changed, 488 insertions(+), 36 deletions(-) create mode 100644 tests/unit/test_error_parity.py diff --git a/servicewright/adapters/grpc/__init__.py b/servicewright/adapters/grpc/__init__.py index 4a7d62c..d2d75fe 100644 --- a/servicewright/adapters/grpc/__init__.py +++ b/servicewright/adapters/grpc/__init__.py @@ -4,7 +4,12 @@ from .config import GrpcConfig from .entrypoint import GrpcEntrypoint, GrpcPlugin, InterceptorFactory, ServicerRegisterer -from .errors import ERROR_CODE_TRAILING_METADATA, GRPC_STATUS_BY_KIND, ServiceErrorInterceptor +from .errors import ( + ERROR_CODE_TRAILING_METADATA, + GRPC_STATUS_BY_KIND, + ServiceErrorInterceptor, + UnhandledErrorInterceptor, +) from .health import GrpcHealthBridge from .interceptors import UnitScopeInterceptor, current_unit_scope from .metadata import ( @@ -28,6 +33,7 @@ "InterceptorFactory", "ServiceErrorInterceptor", "ServicerRegisterer", + "UnhandledErrorInterceptor", "UnitScopeInterceptor", "current_unit_scope", "get_client_context", diff --git a/servicewright/adapters/grpc/entrypoint.py b/servicewright/adapters/grpc/entrypoint.py index 592641a..bd80ac7 100644 --- a/servicewright/adapters/grpc/entrypoint.py +++ b/servicewright/adapters/grpc/entrypoint.py @@ -27,7 +27,7 @@ from ._imports import AsyncMetricsInterceptor, AsyncServer, bind_server_port, create_async_grpc_server, grpc from .config import DEFAULT_HEALTH_SERVICE_NAME, GrpcConfig from .context import get_default_context_setters -from .errors import ServiceErrorInterceptor +from .errors import ServiceErrorInterceptor, UnhandledErrorInterceptor from .health import GrpcHealthBridge from .interceptors import UnitScopeInterceptor from .metrics import GrpcServerMetricsRecorder @@ -56,10 +56,11 @@ class GrpcEntrypoint(ServerEntrypoint): config: Self-contained server configuration (NOT read from settings). servicers: Callback registering servicers on the gRPC server. interceptors: Static interceptors. They wrap the servicer *inside* the - unit-scope and metrics interceptors but *outside* the service-error - mapper, so a generic exception handler of yours (e.g. the kit's - ``AsyncExceptionHandlerInterceptor``) sees an already-mapped - ``AbortError`` instead of swallowing the domain error. + unit-scope, unhandled-error and metrics interceptors but *outside* + the service-error mapper, so a generic exception handler of yours + (e.g. the kit's ``AsyncExceptionHandlerInterceptor``) sees an + already-mapped ``AbortError`` instead of swallowing the domain + error, and still gets first refusal on everything else. interceptors_factory: Optional callback returning extra interceptors, resolved at ``bind`` time with the :class:`ServiceContext`. context_setters: Bridges that push the per-RPC context (request id, user @@ -69,7 +70,10 @@ class GrpcEntrypoint(ServerEntrypoint): map_service_errors: Convert raised :class:`~servicewright.core.errors.ServiceError` into the mapped ``grpc.StatusCode`` abort (non-public errors masked). Default - ``True``. + ``True``. Turning it off does not uninstall + :class:`~servicewright.adapters.grpc.UnhandledErrorInterceptor`: + a ``ServiceError`` then reaches the client as a masked ``INTERNAL`` + like any other unhandled exception, never as its own message. enable_metrics: Add the RPC metrics interceptor, recording through the app's configured metrics sink (``ObsConfig(metrics=...)`` + the matching extra, e.g. servicewright[metrics] for prometheus). @@ -201,6 +205,13 @@ async def _collect_interceptors(self, ctx: ServiceContext[Any, Any]) -> list[grp setters = get_default_context_setters() if self._context_setters is None else self._context_setters interceptors: list[grpc.aio.ServerInterceptor] = [UnitScopeInterceptor(ctx.container, context_setters=setters)] + # Inside the unit scope (so the masked abort is logged with the RPC's + # correlation ids) but outside everything else: nothing below may leave + # without a status, and grpc.aio's own answer for an exception it was + # not told about is UNKNOWN with repr() of it. Not optional, for the + # same reason UnhandledErrorMiddleware is not optional over HTTP. + interceptors.append(UnhandledErrorInterceptor()) + if self._enable_metrics: recorder = GrpcServerMetricsRecorder(ctx.observability.metrics, prefix=self._metrics_prefix) interceptors.append(AsyncMetricsInterceptor(recorder, service_name=ctx.service_name)) diff --git a/servicewright/adapters/grpc/errors.py b/servicewright/adapters/grpc/errors.py index 39fe834..d2a1277 100644 --- a/servicewright/adapters/grpc/errors.py +++ b/servicewright/adapters/grpc/errors.py @@ -6,6 +6,12 @@ like over HTTP — the client sees a generic ``INTERNAL`` and the real code is only logged. The machine-readable code travels in the ``x-error-code`` trailing metadata so clients can branch without parsing messages. + +An exception that is *not* a ``ServiceError`` is masked the same way by +:class:`UnhandledErrorInterceptor`, the counterpart of the HTTP stack's +``UnhandledErrorMiddleware``. Without it ``grpc.aio`` answers such an exception +with ``UNKNOWN`` and ``repr`` of it, so the one error class nobody vetted the +wording of is the one that reaches the caller verbatim. """ from __future__ import annotations @@ -13,7 +19,7 @@ import logging from typing import TYPE_CHECKING -from ...core.errors import ErrorInfo, ErrorKind, ServiceError, mask_private_error +from ...core.errors import INTERNAL_ERROR_CODE, ErrorInfo, ErrorKind, ServiceError, mask_private_error from ._imports import AsyncServerInterceptor, grpc if TYPE_CHECKING: @@ -24,6 +30,13 @@ logger = logging.getLogger(__name__) # Transport mapping: the neutral failure category -> the gRPC status. +# +# Injective on purpose, so a status maps back to exactly one kind. CONFLICT is +# the only entry with a real alternative: ABORTED also renders as 409, but the +# gRPC contract tells clients to retry it at a higher level, which is wrong for +# a state conflict that will never succeed on a retry. FAILED_PRECONDITION is +# spoken for by PRECONDITION_FAILED. Clients needing to tell a duplicate key +# from a state conflict branch on the x-error-code metadata, not on the status. GRPC_STATUS_BY_KIND: dict[ErrorKind, grpc.StatusCode] = { ErrorKind.INVALID: grpc.StatusCode.INVALID_ARGUMENT, ErrorKind.UNAUTHENTICATED: grpc.StatusCode.UNAUTHENTICATED, @@ -40,14 +53,27 @@ ERROR_CODE_TRAILING_METADATA = "x-error-code" +_MASKED_INTERNAL = ErrorInfo(kind=ErrorKind.INTERNAL, code=INTERNAL_ERROR_CODE) + + +async def _abort(call: RpcCall, info: ErrorInfo) -> None: + """Abort the RPC with the kind's status and the code in trailing metadata.""" + # abort() raises grpc.aio.AbortError and never returns. + await call.context.abort( + GRPC_STATUS_BY_KIND[info.kind], + info.detail or info.code, + trailing_metadata=((ERROR_CODE_TRAILING_METADATA, info.code),), + ) + class ServiceErrorInterceptor(AsyncServerInterceptor): """Abort RPCs failing with :class:`ServiceError` using the mapped status. - Added automatically by :class:`GrpcEntrypoint` (inside the metrics - interceptor, so aborts are recorded with their real status). Any other - exception passes through untouched — compose grpc-server-kit's - ``AsyncExceptionHandlerInterceptor`` for generic exception mapping. + Added automatically by :class:`GrpcEntrypoint` as the innermost + interceptor, so a domain error becomes a mapped abort before any + interceptor of yours sees it. Any other exception passes through untouched, + to your own interceptors first and to :class:`UnhandledErrorInterceptor` + last. """ async def around_call(self, call: RpcCall) -> AsyncIterator[None]: @@ -55,28 +81,50 @@ async def around_call(self, call: RpcCall) -> AsyncIterator[None]: try: yield except ServiceError as exc: - await self._abort(call, exc) - - async def _abort(self, call: RpcCall, exc: ServiceError) -> None: - if not exc.public: - logger.warning( - "Private service error occurred: %s", - exc.code, - extra={"error_code": exc.code, "error_kind": exc.kind, "params": exc.params}, - ) - info = mask_private_error(ErrorInfo.from_service_error(exc)) - status = GRPC_STATUS_BY_KIND[info.kind] - - # abort() raises grpc.aio.AbortError and never returns. - await call.context.abort( - status, - info.detail or info.code, - trailing_metadata=((ERROR_CODE_TRAILING_METADATA, info.code),), - ) + if not exc.public: + logger.warning( + "Private service error occurred: %s", + exc.code, + extra={"error_code": exc.code, "error_kind": exc.kind, "params": exc.params}, + ) + await _abort(call, mask_private_error(ErrorInfo.from_service_error(exc))) + + +class UnhandledErrorInterceptor(AsyncServerInterceptor): + """Abort with a masked ``INTERNAL`` when an RPC fails with anything else. + + The last resort, and the gRPC counterpart of the HTTP stack's + ``UnhandledErrorMiddleware``: ``grpc.aio`` answers an exception it was never + told about with ``UNKNOWN`` and ``repr`` of the exception, which puts a + message nobody wrote for a client — a DSN, a row of a query — on the wire. + Here the client gets exactly what a ``public=False`` + :class:`~servicewright.core.errors.ServiceError` produces (``INTERNAL``, + detail ``internal_error``, ``x-error-code: internal_error``) and the real + exception goes to the log with its traceback. + + :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. + """ + + 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): + raise + except Exception: + logger.exception("Unhandled exception while serving RPC", extra={"grpc_method": call.method_name}) + await _abort(call, mask_private_error(_MASKED_INTERNAL)) __all__ = [ "ERROR_CODE_TRAILING_METADATA", "GRPC_STATUS_BY_KIND", "ServiceErrorInterceptor", + "UnhandledErrorInterceptor", ] diff --git a/tests/integration/test_grpc_integration.py b/tests/integration/test_grpc_integration.py index e3f6faf..b82fcc0 100644 --- a/tests/integration/test_grpc_integration.py +++ b/tests/integration/test_grpc_integration.py @@ -1,9 +1,10 @@ -"""Integration test driving a real in-process gRPC server. +"""Integration tests driving a real in-process gRPC server. Marked ``integration`` so it does NOT run under ``-m unit``. It exercises the actual grpc-server-kit primitives (no mocking): a real ``grpc.aio`` server is -created, bound to an ephemeral port, started, the standard health service is -queried over a real channel, then the entrypoint is drained and stopped. +created, bound to an ephemeral port and started, then queried over a real +channel — the standard health service in one test, a failing RPC in the other, +which is the only way to see the status grpc.aio itself would have chosen. """ from __future__ import annotations @@ -12,6 +13,7 @@ from typing import Any import grpc +import grpc.aio import pytest from grpc_health.v1 import health_pb2, health_pb2_grpc @@ -56,3 +58,52 @@ async def probe_then_stop() -> None: # After shutdown, readiness is flipped off and the port was actually bound. assert spec.health.ready is False assert ep.bound_port is not None and ep.bound_port > 0 + + +_SERVICE = "lab.Orders" +_SECRET = "dsn=postgres://user:pw@db/ledger" + + +def _register_raising_servicer(server: Any, _ctx: Any) -> None: + """Register one RPC that fails with a plain exception, no proto needed.""" + + async def pay(request: bytes, context: Any) -> bytes: + raise RuntimeError(_SECRET) + + server.add_generic_rpc_handlers( + (grpc.method_handlers_generic_handler(_SERVICE, {"Pay": grpc.unary_unary_rpc_method_handler(pay)}),) + ) + + +async def test__grpc_entrypoint__servicer_raises_an_unexpected_exception__client_gets_a_masked_internal() -> None: + # Arrange + spec: AppSpec[Any, Any] = AppSpec(service_name="grpc-errors-it", create_container=lambda _s: FakeContainer()) + config = GrpcConfig(host="127.0.0.1", port=0, enable_reflection=False, enable_channelz=False) + ep = GrpcEntrypoint(config=config, servicers=_register_raising_servicer) + service = Service(spec, entrypoints=[ep]) + + stop = asyncio.Event() + failures: list[grpc.aio.AioRpcError] = [] + + async def call_then_stop() -> None: + while not (spec.health.ready and ep.bound_port): + await asyncio.sleep(0.01) + await asyncio.sleep(0.05) + + async with grpc.aio.insecure_channel(f"127.0.0.1:{ep.bound_port}") as channel: + try: + await channel.unary_unary(f"/{_SERVICE}/Pay")(b"42", timeout=5) + except grpc.aio.AioRpcError as error: + failures.append(error) + + stop.set() + + # Act + await asyncio.gather(service.run(FakeSettings(), stop=stop), call_then_stop()) + + # Assert — what grpc.aio would have sent unaided is UNKNOWN plus repr() of the exception. + 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" + assert _SECRET not in str(error.details()) diff --git a/tests/unit/test_error_parity.py b/tests/unit/test_error_parity.py new file mode 100644 index 0000000..0809e6e --- /dev/null +++ b/tests/unit/test_error_parity.py @@ -0,0 +1,185 @@ +"""Both transports owe a caller the same conclusion about the same failure. + +The masking rule is transport-neutral by design, so it is asserted here as one +property over both adapters rather than twice in their own modules. The way it +broke last time was a gap on one side only: an exception that was not a +``ServiceError`` reached a gRPC caller with its message while HTTP masked it. +""" + +from __future__ import annotations + +import contextlib +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import Any + +import grpc +import grpc.aio +import pytest +from fastapi import APIRouter +from fastapi.testclient import TestClient +from grpc_server_kit.aio.interceptors import RpcCall +from pytest_lazy_fixtures import lf + +from servicewright import ErrorKind, ServiceError +from servicewright.adapters.fastapi import FastApiEntrypoint, HttpConfig +from servicewright.adapters.grpc import ( + ERROR_CODE_TRAILING_METADATA, + GRPC_STATUS_BY_KIND, + ServiceErrorInterceptor, + UnhandledErrorInterceptor, +) +from servicewright.core.errors import HTTP_STATUS_BY_KIND, INTERNAL_ERROR_CODE +from servicewright.core.health import HealthRegistry +from servicewright.core.spec import BootstrapContext, ServiceContext +from servicewright.testing import FakeContainer, FakeScope, FakeSettings + +pytestmark = pytest.mark.unit + +SECRET = "dsn=postgres://user:pw@db/ledger" + + +class OrderNotFoundError(ServiceError): + kind = ErrorKind.NOT_FOUND + + +@dataclass(frozen=True, slots=True) +class Conclusion: + """What one caller learns from one failed call, in transport-neutral terms. + + Attributes: + kind: The failure category the received status maps back to. + code: The machine-readable code the caller can branch on. + detail: The human-readable message, if any. + wire_text: Every byte the caller can read, for leak assertions. + """ + + kind: ErrorKind + code: str + detail: str + wire_text: str + + +Probe = Callable[[Exception], Awaitable[Conclusion]] + + +def _kind_of(status: Any, table: Mapping[ErrorKind, Any]) -> ErrorKind: + """Reverse the transport's own mapping table; both are injective.""" + return next(kind for kind, mapped in table.items() if mapped == status) + + +def _service_ctx() -> ServiceContext: + bootstrap = BootstrapContext( + settings=FakeSettings(), + service_name="parity", + container=FakeContainer(), + lifecycle=object(), # type: ignore[arg-type] + ) + return ServiceContext(bootstrap=bootstrap, app_scope=FakeScope(), health=HealthRegistry()) + + +class _AbortRecordingContext: + """Servicer-context double recording abort() and raising what gRPC raises.""" + + def __init__(self) -> None: + self.aborts: list[tuple[Any, str, Any]] = [] + + async def abort(self, code: Any, details: str = "", trailing_metadata: Any = None) -> None: + self.aborts.append((code, details, trailing_metadata)) + raise grpc.aio.AbortError(details) + + +@pytest.fixture +def over_http() -> Probe: + """Ask a route that raises, through the entrypoint's default handlers.""" + + async def call(exc: Exception) -> Conclusion: + router = APIRouter() + + @router.get("/pay") + async def pay() -> None: + raise exc + + ep = FastApiEntrypoint(config=HttpConfig(port=0), routers=(router,)) + app = await ep.build_app(_service_ctx()) + response = TestClient(app, raise_server_exceptions=False).get("/pay") + + body = response.json() + return Conclusion( + kind=_kind_of(response.status_code, HTTP_STATUS_BY_KIND), + code=body["code"], + detail=body.get("detail", ""), + wire_text=response.text, + ) + + return call + + +@pytest.fixture +def over_grpc() -> Probe: + """Ask a servicer that raises, through the interceptors the entrypoint installs.""" + + async def call(exc: Exception) -> Conclusion: + context = _AbortRecordingContext() + rpc = RpcCall( + method_name="/lab.Orders/Pay", + request=b"", + context=context, + request_streaming=False, + response_streaming=False, + ) + + # The order GrpcEntrypoint composes them in; test_grpc.py asserts it. + with contextlib.suppress(grpc.aio.AbortError): + async with UnhandledErrorInterceptor().around(rpc), ServiceErrorInterceptor().around(rpc): + raise exc + + # Exactly one status: the net must not re-abort a mapped domain error. + assert len(context.aborts) == 1 + (status, details, trailing) = context.aborts[0] + return Conclusion( + kind=_kind_of(status, GRPC_STATUS_BY_KIND), + code=dict(trailing)[ERROR_CODE_TRAILING_METADATA], + detail=details, + wire_text=f"{details} {trailing}", + ) + + return call + + +@pytest.mark.parametrize("probe", [lf("over_http"), lf("over_grpc")], ids=["http", "grpc"]) +@pytest.mark.parametrize( + "failure", + [ + pytest.param( + ServiceError(SECRET, code="ledger_corrupted", kind=ErrorKind.INTERNAL, public=False), + id="declared-private", + ), + pytest.param(RuntimeError(SECRET), id="never-declared"), + ], +) +async def test__masked_failure__raised_over_either_transport__tells_the_caller_only_internal_error( + probe: Probe, + failure: Exception, +) -> None: + # Act + conclusion = await probe(failure) + + # Assert + assert conclusion.kind is ErrorKind.INTERNAL + assert conclusion.code == INTERNAL_ERROR_CODE + assert SECRET not in conclusion.wire_text + assert "ledger_corrupted" not in conclusion.wire_text + + +@pytest.mark.parametrize("probe", [lf("over_http"), lf("over_grpc")], ids=["http", "grpc"]) +async def test__public_service_error__raised_over_either_transport__reaches_the_caller_intact( + probe: Probe, +) -> None: + # Act + conclusion = await probe(OrderNotFoundError("no order with id 42")) + + # Assert + assert conclusion.kind is ErrorKind.NOT_FOUND + assert conclusion.code == "order_not_found" + assert conclusion.detail == "no order with id 42" diff --git a/tests/unit/test_grpc.py b/tests/unit/test_grpc.py index 451ab86..f6d9d58 100644 --- a/tests/unit/test_grpc.py +++ b/tests/unit/test_grpc.py @@ -8,10 +8,13 @@ from __future__ import annotations import asyncio +import logging import uuid from typing import Any from unittest.mock import MagicMock +import grpc +import grpc.aio import pytest from grpc_server_kit.aio.interceptors import AsyncMetricsInterceptor, RpcCall @@ -272,6 +275,29 @@ async def test__grpc_bind__error_mapping_disabled__omits_the_mapper(patched_serv assert not any(isinstance(i, ServiceErrorInterceptor) for i in interceptors) +async def test__grpc_bind__called__installs_the_unhandled_error_net_inside_the_unit_scope( + patched_server: _FakeAsyncServer, +) -> None: + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + ep = GrpcEntrypoint(config=GrpcConfig(), servicers=lambda _s, _c: None) + await ep.bind(_make_service_ctx(FakeContainer())) + interceptors = patched_server.create_calls[0]["interceptors"] # type: ignore[attr-defined] + # Second: the correlation ids the unit scope binds are on the masked abort's log line. + assert isinstance(interceptors[1], UnhandledErrorInterceptor) + + +async def test__grpc_bind__error_mapping_disabled__still_installs_the_unhandled_error_net( + patched_server: _FakeAsyncServer, +) -> None: + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + ep = GrpcEntrypoint(config=GrpcConfig(), servicers=lambda _s, _c: None, map_service_errors=False) + await ep.bind(_make_service_ctx(FakeContainer())) + interceptors = patched_server.create_calls[0]["interceptors"] # type: ignore[attr-defined] + assert any(isinstance(i, UnhandledErrorInterceptor) for i in interceptors) + + async def test__grpc_bind__async_servicer_registerer__awaits_it(patched_server: _FakeAsyncServer) -> None: called: list[str] = [] @@ -369,9 +395,9 @@ async def test__grpc_bind__metrics_enabled__adds_the_metrics_interceptor(patched await ep.bind(_make_service_ctx(FakeContainer(), service_name="metered", observability=manager)) interceptors = patched_server.create_calls[0]["interceptors"] # type: ignore[attr-defined] - # UnitScope first, the kit's metrics interceptor second. + # UnitScope first, the unhandled-error net second, the kit's metrics interceptor third. assert isinstance(interceptors[0], UnitScopeInterceptor) - assert isinstance(interceptors[1], AsyncMetricsInterceptor) + assert isinstance(interceptors[2], AsyncMetricsInterceptor) # The recorder minted the frozen instruments with the prefix applied. assert sink.counters == ["myprefix_grpc_requests_total"] assert sink.histograms == ["myprefix_grpc_request_duration_seconds"] @@ -393,7 +419,7 @@ async def test__grpc_bind__metrics_enabled_without_a_sink__records_into_null_ins await ep.bind(_make_service_ctx(FakeContainer())) interceptors = patched_server.create_calls[0]["interceptors"] # type: ignore[attr-defined] - assert isinstance(interceptors[1], AsyncMetricsInterceptor) + assert isinstance(interceptors[2], AsyncMetricsInterceptor) class _RecordingRecorder: @@ -746,6 +772,109 @@ async def test__service_error_interceptor__other_exception__passes_it_through() assert context.aborts == [] +# --------------------------------------------------------------------------- # +# UnhandledErrorInterceptor: the last-resort mask +# --------------------------------------------------------------------------- # +async def test__unhandled_error_interceptor__unexpected_exception__aborts_with_a_masked_internal() -> 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 RuntimeError("dsn=postgres://user:pw@db/ledger") + + # 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"),) + # Without the mask grpc.aio answers UNKNOWN with repr() of the exception. + assert "postgres" not in details + + +async def test__unhandled_error_interceptor__unexpected_exception__logs_it_with_the_traceback( + caplog: pytest.LogCaptureFixture, +) -> None: + # Arrange + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + context = _AbortRecordingContext() + + # Act + with caplog.at_level(logging.ERROR), pytest.raises(_AbortedSentinelError): + async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")): + raise RuntimeError("dsn=postgres://user:pw@db/ledger") + + # Assert + 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 record.__dict__["grpc_method"] == "/pkg.Svc/Rpc" + + +async def test__unhandled_error_interceptor__service_error__masks_it_too() -> None: + # Arrange + import grpc as grpc_lib + + from servicewright import ErrorKind, ServiceError + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + context = _AbortRecordingContext() + + # Act — with map_service_errors=False nothing maps it first. + with pytest.raises(_AbortedSentinelError): + async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")): + raise ServiceError("no such user", code="user_missing", kind=ErrorKind.NOT_FOUND) + + # Assert + (code, details, _trailing) = context.aborts[0] + assert code == grpc_lib.StatusCode.INTERNAL + 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: + # Arrange + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + context = _AbortRecordingContext() + + # Act + with pytest.raises(type(deliberate)): + async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")): + raise deliberate + + # Assert + assert context.aborts == [] + + +async def test__unhandled_error_interceptor__caller_cancelled__passes_it_through() -> None: + # Arrange + from servicewright.adapters.grpc import UnhandledErrorInterceptor + + context = _AbortRecordingContext() + + # Act + with pytest.raises(asyncio.CancelledError): + async with UnhandledErrorInterceptor().around(_make_call(context, "/pkg.Svc/Rpc")): + raise asyncio.CancelledError + + # Assert + assert context.aborts == [] + + # --------------------------------------------------------------------------- # # GrpcHealthBridge # --------------------------------------------------------------------------- # @@ -1153,6 +1282,28 @@ async def test__grpc_entrypoint_bind__user_interceptors_supplied__error_mapper_s assert chain.index(user_static) < chain.index(user_factory) < len(chain) - 1 +async def test__grpc_entrypoint_bind__user_interceptors_supplied__unhandled_error_net_stays_outside_them( + patched_server: _FakeAsyncServer, +) -> None: + # Arrange + from servicewright.adapters.grpc.errors import UnhandledErrorInterceptor + + user_static = MagicMock(name="static") + ep = GrpcEntrypoint( + config=GrpcConfig(port=0), + servicers=lambda _s, _c: None, + interceptors=[user_static], + ) + + # Act + await ep.bind(_make_service_ctx(FakeContainer())) + + # Assert — an exception type you map yourself reaches your interceptor first. + chain = patched_server.create_calls[0]["interceptors"] + net = next(i for i in chain if isinstance(i, UnhandledErrorInterceptor)) + assert chain.index(net) < chain.index(user_static) + + async def test__grpc_entrypoint_bind__error_mapping_disabled__omits_the_mapper( patched_server: _FakeAsyncServer, ) -> None: From 429466c5803113dc1298bd5f2bafc44e0d96c8fb Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Mon, 7 Sep 2026 15:17:56 +0300 Subject: [PATCH 2/2] docs: what a caller learns from an error nobody declared, on both transports The errors page said public=False masks at every transport and left the other half unsaid, so the pages a reader lands on described the gap as if it were the design. They now carry the masked-by-default table for an undeclared exception, the note that neither default_exception_handlers=False nor map_service_errors=False removes that layer, and the reason CONFLICT maps to ALREADY_EXISTS rather than to ABORTED. The gRPC adapter page gains the section on anything that is not a ServiceError and the interceptor chain grows its new row, as do the blueprint's two renderings of it. The runbook entry for a domain error becoming INTERNAL says what map_service_errors=False now leaves in place, the checklist gains the line about catching Exception and returning its str(), and agents.md carries the new name, the chain and a widened rule 16. Refs #50 --- docs/adapters/grpc.md | 46 +++++++++++++++++++++++++---- docs/agents.md | 21 ++++++++++--- docs/blueprints/grpc-service.md | 16 ++++++---- docs/concepts/errors.md | 52 +++++++++++++++++++++++++++++++++ docs/operations/checklist.md | 3 ++ docs/operations/runbooks.md | 4 +++ 6 files changed, 126 insertions(+), 16 deletions(-) diff --git a/docs/adapters/grpc.md b/docs/adapters/grpc.md index f9e95b0..c8a5ad1 100644 --- a/docs/adapters/grpc.md +++ b/docs/adapters/grpc.md @@ -90,18 +90,43 @@ real code only reaches your logs. The full mapping table is in Turn it off with `map_service_errors=False`. +## Anything that is not a `ServiceError` + +`UnhandledErrorInterceptor` is the last resort, and it is always installed: + +```python +raise RuntimeError("dividing by the number of retries, which is zero") +# → the RPC aborts with INTERNAL +# → details "internal_error", trailing metadata "x-error-code: internal_error" +# → the exception and its traceback go to the log, with the RPC's correlation ids +``` + +Byte for byte what a `public=False` `ServiceError` produces, which is the point: the two are +indistinguishable to a caller. Without it `grpc.aio` answers `UNKNOWN` with `repr()` of the +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. + +`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 +unhandled exception, never as its own message. + ## Interceptor ordering `grpc.aio` hands control in list order, so the first entry is the **outermost** wrapper. The entrypoint assembles the chain like this: ``` -UnitScopeInterceptor ← outermost: everything below sees a live scope - metrics interceptor ← records the status the client actually receives - your static interceptors - your interceptors_factory results - ServiceErrorInterceptor ← innermost: closest to the servicer - your servicer +UnitScopeInterceptor ← outermost: everything below sees a live scope + UnhandledErrorInterceptor ← the net: inside the scope, so the log is correlated + metrics interceptor ← records the status the client actually receives + your static interceptors + your interceptors_factory results + ServiceErrorInterceptor ← innermost: closest to the servicer + your servicer ``` !!! warning "Why `ServiceErrorInterceptor` is innermost" @@ -113,6 +138,15 @@ UnitScopeInterceptor ← outermost: everything below sees a live scope It still sits *inside* the metrics interceptor, so aborts are recorded with their real status. +!!! note "Why `UnhandledErrorInterceptor` is not" + + It is the mirror of the HTTP stack's `UnhandledErrorMiddleware`, and it sits at the same + depth: inside the layer that binds the correlation ids, so the masked abort is logged with + the request id, and outside everything else, so it only ever sees what nobody below it + claimed. An interceptor of yours that maps an exception type of its own still gets it first — + including grpc-server-kit's `AsyncExceptionHandlerInterceptor`, if that is the mapping you + want. Compose exactly as before. + ```python GrpcEntrypoint( config=config, diff --git a/docs/agents.md b/docs/agents.md index 34ffdf5..a7aa091 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -251,12 +251,17 @@ once per run inside the application scope and appended to `spec.warmers`. | `ErrorKind` | `INVALID`, `UNAUTHENTICATED`, `FORBIDDEN`, `NOT_FOUND`, `CONFLICT`, `PRECONDITION_FAILED`, `TOO_MANY_REQUESTS`, `DEADLINE_EXCEEDED`, `UNAVAILABLE`, `NOT_IMPLEMENTED`, `INTERNAL` | 400/401/403/404/409/412/429/504/503/501/500 over HTTP | | `ErrorInfo` | `kind`, `code`, `detail=None`, `params={}`, `public=True`, `status_override=None`, `headers=None`; `.http_status`; `ErrorInfo.from_service_error(exc)` | the normalized view every renderer takes | | `mask_private_error` | `mask_private_error(info) -> ErrorInfo` | `public=False` collapses to a generic `internal_error` 500 | +| `INTERNAL_ERROR_CODE` | `"internal_error"` | the code every masked error renders as, on both transports | | `RenderedError` | `status_code`, `body`, `media_type="application/problem+json"`, `headers=None` | | | `ProblemDetailsRenderer` | `ProblemDetailsRenderer(*, type_base=None)` | RFC 9457; the default renderer | -`HTTP_STATUS_BY_KIND`, `status_title` and `to_json_safe` live in +`HTTP_STATUS_BY_KIND`, `INTERNAL_ERROR_CODE`, `status_title` and `to_json_safe` live in `servicewright.core.errors` and are not re-exported at the top level. +`ErrorKind.CONFLICT` maps to `ALREADY_EXISTS` and not to `ABORTED`: both mean 409, but `ABORTED` +tells the client to retry at a higher level, which a state conflict will not survive. Both status +tables are injective, so a status maps back to exactly one kind. + ### Request context | Name | Signature | Notes | @@ -308,7 +313,7 @@ installed and raise (or degrade) at construction instead. |---|---| | `servicewright.adapters.fastapi` | `FastApiEntrypoint`, `FastApiPlugin`, `HttpConfig`, `MiddlewareConfig`, `HealthConfig`, `CORSMiddlewareConfig`, `LoggingMiddlewareConfig`, `GZipMiddlewareConfig`, `CorrelationIdMiddlewareConfig`, `MetricsInstrumentatorConfig`, `UnitScopeDep`, `UnitScopeMiddleware`, `get_unit_scope`, `current_unit_scope`, `setup_default_exception_handlers`, `setup_metrics_instrumentator`, `LivenessResponse`, `ReadinessResponse`, `ProblemDetails`, `XUserId`, `IdempotencyKey`, `AuthorizationHeader`, `XFingerprintHeader`, `OtelBaggageSetter`, `StructlogSetter`, `get_default_context_setters`, `RoutesRegisterer`, `ConfigureApp` | | `servicewright.adapters.litestar` | `LitestarEntrypoint`, `LitestarPlugin`, `LitestarConfig`, `HealthConfig`, `build_health_routes`, `UnitScopeMiddleware`, `get_unit_scope`, `current_unit_scope`, `RouteRegisterer`, `ConfigureApp` | -| `servicewright.adapters.grpc` | `GrpcEntrypoint`, `GrpcPlugin`, `GrpcConfig`, `ServicerRegisterer`, `InterceptorFactory`, `ServiceErrorInterceptor`, `GRPC_STATUS_BY_KIND`, `ERROR_CODE_TRAILING_METADATA`, `GrpcHealthBridge`, `UnitScopeInterceptor`, `current_unit_scope`, `GrpcServerMetricsRecorder`, `IDEMPOTENCY_KEY_METADATA`, `get_idempotency_key`, `get_client_ip`, `get_user_agent`, `get_client_context` | +| `servicewright.adapters.grpc` | `GrpcEntrypoint`, `GrpcPlugin`, `GrpcConfig`, `ServicerRegisterer`, `InterceptorFactory`, `ServiceErrorInterceptor`, `UnhandledErrorInterceptor`, `GRPC_STATUS_BY_KIND`, `ERROR_CODE_TRAILING_METADATA`, `GrpcHealthBridge`, `UnitScopeInterceptor`, `current_unit_scope`, `GrpcServerMetricsRecorder`, `IDEMPOTENCY_KEY_METADATA`, `get_idempotency_key`, `get_client_ip`, `get_user_agent`, `get_client_context` | | `servicewright.adapters.apscheduler4` (and `.apscheduler3`) | `SchedulerEntrypoint`, `SchedulerPlugin`, `ScheduledJob`, `ScheduledJobFunc`, `SchedulerJobMetricsRecorder`, `SchedulerError`, `DuplicateScheduleError` | | `servicewright.adapters.dishka` | `DishkaContainer`, `DishkaScope` | | `servicewright.adapters.settings` | `BaseServiceSettings`, `LoggingSettings`, `MetricsSettings`, `TracingSettings`, `ErrorTrackingSettings` | @@ -333,7 +338,9 @@ Entrypoint constructors, all keyword-only: context_setters=None, map_service_errors=True, enable_metrics=False, metrics_prefix=None, kind="grpc", essential=True)`. `GrpcConfig` defaults: `port=50051`, `grace_period=30.0`, `enable_reflection=False`, `enable_channelz=False`, - `health_service_names=()`, `health_refresh_interval=5.0`. + `health_service_names=()`, `health_refresh_interval=5.0`. Interceptor chain, outermost first: + `UnitScopeInterceptor`, `UnhandledErrorInterceptor`, metrics, yours, + `ServiceErrorInterceptor`. * `SchedulerEntrypoint(jobs, enable_metrics=False, metrics_prefix=None, kind="scheduler", essential=True)`; `ScheduledJob(id, func, trigger, args=(), kwargs={}, max_instances=None, misfire_grace_time=None, coalesce=None)`. @@ -420,7 +427,13 @@ Each `*Plugin` takes exactly the same arguments as its entrypoint and exposes `. 16. **`ServiceError(public=False)` masks everything at the transport**: the client gets a generic `internal_error` 500 (or `INTERNAL` over gRPC) and the real code only reaches the log. A subclass's `code` is derived from its class name, so renaming the class is a wire - change. + change. **An exception that is not a `ServiceError` is masked identically**, by + `UnhandledErrorMiddleware` over HTTP and `UnhandledErrorInterceptor` over gRPC — both + installed unconditionally, neither removed by `default_exception_handlers=False` or + `map_service_errors=False`, which only stop the mapping of the errors you declared. So a + caller cannot tell an error you hid from one you never knew about, and no exception text + reaches the wire. Do not catch-and-return an exception in a handler or servicer to "make the + error nicer": that is the one way to get its message back onto the wire. 17. **One unit scope per request, not two.** If the DI framework's own integration owns the request scope (dishka's `setup_dishka`), switch servicewright's off — `MiddlewareConfig(unit_scope=False)` or `LitestarConfig(unit_scope=False)` — otherwise diff --git a/docs/blueprints/grpc-service.md b/docs/blueprints/grpc-service.md index cfffda7..43bf283 100644 --- a/docs/blueprints/grpc-service.md +++ b/docs/blueprints/grpc-service.md @@ -13,7 +13,8 @@ pip install "servicewright[grpc,dishka,postgres,metrics,observability]" flowchart LR C["gRPC clients"] --> EP["GrpcEntrypoint
:50051"] EP --> I1["UnitScopeInterceptor"] - I1 --> I2["metrics"] + I1 --> I0["UnhandledErrorInterceptor"] + I0 --> I2["metrics"] I2 --> I3["your interceptors"] I3 --> I4["ServiceErrorInterceptor"] I4 --> S["Servicers"] @@ -97,15 +98,18 @@ Your interceptors land between the framework's: ``` UnitScopeInterceptor ← scope is live for everything below - metrics ← records the status the client receives - AuthInterceptor ← yours - LoggingInterceptor ← yours - ServiceErrorInterceptor - OrdersServicer + UnhandledErrorInterceptor ← nothing below leaves without a status + metrics ← records the status the client receives + AuthInterceptor ← yours + LoggingInterceptor ← yours + ServiceErrorInterceptor + OrdersServicer ``` That is why a generic "map everything to INTERNAL" interceptor of yours cannot swallow a deliberate `NOT_FOUND`: the domain error has already become an abort by the time it reaches you. +And it is why anything your interceptors do *not* map still reaches the client as `INTERNAL` with +`x-error-code: internal_error` rather than as `UNKNOWN` plus the exception's own text. ## 4. main diff --git a/docs/concepts/errors.md b/docs/concepts/errors.md index f0dfcb7..497c6ed 100644 --- a/docs/concepts/errors.md +++ b/docs/concepts/errors.md @@ -13,6 +13,7 @@ servicewright solves that with three replaceable pieces: ```mermaid flowchart LR E["raise OrderNotFoundError"] --> N["ErrorInfo
kind · code · detail · params · public"] + X["raise anything else"] --> N N --> M["mask_private_error()"] M --> R["HttpErrorRenderer"] M --> K["ErrorKind → grpc.StatusCode"] @@ -66,6 +67,17 @@ except ServiceError as exc: | `DEADLINE_EXCEEDED` | 504 | `DEADLINE_EXCEEDED` | | `INTERNAL` | 500 | `INTERNAL` | +Both columns are injective, so a status maps back to exactly one kind — which is what lets a +gateway translate between the two without guessing. + +One row is worth stating out loud, because it has a plausible-looking alternative: `CONFLICT` +renders as `ALREADY_EXISTS` and not as `ABORTED`, although both mean 409 to a gRPC gateway. +`ABORTED` is the transaction-conflict code, and the gRPC contract tells clients to retry it at a +higher level — the wrong advice for "this order was already paid", which will never succeed on a +retry. `FAILED_PRECONDITION` is spoken for by `PRECONDITION_FAILED`. A client that has to tell a +duplicate key from a state conflict branches on `x-error-code`, which carries the exact code +either way, rather than on the status. + ## What the client sees Over HTTP, an RFC 9457 problem document with `Content-Type: application/problem+json`: @@ -111,6 +123,34 @@ leak through a custom renderer by accident. public = False ``` +## Errors you never declared + +An exception that is not a `ServiceError` at all — a `KeyError` off a dict, a `RuntimeError` from +a driver — is masked the same way, and by the same rule, on both transports: + +| | HTTP | gRPC | +| --- | --- | --- | +| status | 500 | `INTERNAL` | +| machine code | `"code": "internal_error"` | `x-error-code: internal_error` | +| message | none | `internal_error` | +| the exception itself | logged with its traceback | logged with its traceback | + +So a caller cannot tell an error you hid from an error you never knew about, and neither one +carries a sentence you did not write. That matters most for the second kind, because its wording +is whatever a library chose: an exception text is where a DSN with a password, a failing row or a +file path ends up. + +Over HTTP this is `UnhandledErrorMiddleware` plus the `Exception` handler; over gRPC it is +`UnhandledErrorInterceptor`. Both are installed unconditionally, and neither one is what +`map_service_errors=False` or `default_exception_handlers=False` turns off — those switch off the +mapping of the errors you *did* declare, and an undeclared exception then reaches the client as a +masked internal error rather than as its own message. + +!!! warning "The gRPC default without it" + + `grpc.aio` answers an exception it was not told about with `UNKNOWN` and `repr()` of it — + status code and message both chosen by the transport, out of material nobody reviewed. + ## Own the wire format If your product has its own error envelope, implement one renderer and every default handler @@ -173,6 +213,18 @@ from servicewright.core.errors import status_title, to_json_safe | Deadline exceeded | 504, `code="deadline_exceeded"` | | Anything unhandled | masked 500, logged with the request id — and the response carries that id too | +## The gRPC equivalents + +| Situation | Result | +| --- | --- | +| `ServiceError` | its kind's status; masked when `public=False` | +| Anything unhandled | `INTERNAL`, `x-error-code: internal_error`, logged with the correlation ids | +| A handler that called `context.abort()` itself | left alone — the status was already chosen | +| The caller went away (`CancelledError`) | left alone — not an error to report | + +The two interceptors doing that sit at different depths of the chain, and +[the adapter page](../adapters/grpc.md#interceptor-ordering) explains why. + Add handlers for your own exception types, or switch the defaults off entirely: ```python diff --git a/docs/operations/checklist.md b/docs/operations/checklist.md index 9dd48ff..891fb5a 100644 --- a/docs/operations/checklist.md +++ b/docs/operations/checklist.md @@ -44,6 +44,9 @@ to a failure someone has already had. `to_json_safe`. - [ ] A masked 500 still carries the request id (it does by default — do not move the unhandled-error layer). +- [ ] Nothing catches a broad `Exception` in a handler or servicer and returns its `str()` to + the client. The unhandled-error layer masks what reaches it; a message you re-raise + yourself as a public error is yours to have read. ## Observability diff --git a/docs/operations/runbooks.md b/docs/operations/runbooks.md index 58685ae..c1d66b5 100644 --- a/docs/operations/runbooks.md +++ b/docs/operations/runbooks.md @@ -192,6 +192,10 @@ Check, in order: `ServiceErrorInterceptor` is installed innermost precisely so your own interceptors cannot intercept the domain error first — but code inside the servicer still can. +With the mapper off, `UnhandledErrorInterceptor` stays: an unmapped `ServiceError` then arrives as +a masked `INTERNAL` with `x-error-code: internal_error`, exactly like an exception nobody +declared. The log line next to it carries the real one. + ### Validation errors have a different shape than the rest They should not — every default handler renders through the same renderer. A 422 with a foreign