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 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: