From 62a20003dcdd5383f19fff0dc8db3fd91fa37496 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Mon, 7 Sep 2026 14:00:08 +0300 Subject: [PATCH] fix: an around_call teardown cannot change a call's outcome, and shares its setup's context MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One `around_call` generator was stepped from four different places, and both defects follow from that. `contextvars` compares contexts by identity, so the teardown of every kind but unary-unary could not reset a token the setup had minted — the pair that scopes a request id, a correlation id or an OpenTelemetry context to a single call. And an exception raised after the `yield` had three different fates depending only on the kind of call it wrapped: it replaced a response the server had already sent, it truncated a response stream, or it disappeared into the event loop's exception handler. `_AroundScope` now owns the generator. It closes it exactly once, logs at ERROR and drops whatever the teardown raises — cancellation excepted, since a task being torn down is not a teardown failing — and, for the three kinds whose teardown finishes in another task, steps the setup, the RPC creation and the teardown in one `contextvars.Context` of its own, which `asyncio.Task(context=...)` enters rather than copies. Unary-unary pins nothing: its three phases are already one coroutine, and a task per call per layer is a real cost on the busiest path in the kit. The tracing interceptor's reason for avoiding the seam ("detaching the OpenTelemetry context there fails") no longer holds; its docstring now says what does keep it on `intercept`, which is the cost of an around scope on a layer that does nothing at all when no SDK is configured. --- README.md | 10 +- docs/agents.md | 28 ++- docs/guide/interceptors.md | 44 ++++- grpc_client_kit/interceptors/base.py | 252 ++++++++++++++++++------ grpc_client_kit/interceptors/tracing.py | 14 +- tests/helpers.py | 59 ++++++ tests/integration/test_around_call.py | 68 +++++++ tests/unit/interceptors/conftest.py | 54 +++++ tests/unit/interceptors/test_base.py | 84 +++++++- 9 files changed, 526 insertions(+), 87 deletions(-) create mode 100644 tests/integration/test_around_call.py diff --git a/README.md b/README.md index 3e4bed9..12f254e 100644 --- a/README.md +++ b/README.md @@ -222,9 +222,13 @@ class FailureCounter(AsyncAroundClientInterceptor): Code before the `yield` runs before the RPC exists, so that is where call details are rewritten and where raising refuses a call outright. Code after it -runs once the call is over, whichever of the four kinds it was. Layers that -re-issue a call — retries — subclass `AsyncClientInterceptor` and issue it -themselves; see the +runs once the call is over, whichever of the four kinds it was, and cannot +change how it ended: what the teardown raises is logged and dropped. Both sides +of the `yield` run in one `contextvars.Context`, so `token = VAR.set(...)` +before it and `VAR.reset(token)` after it is a supported way to scope a request +id — or an OpenTelemetry context — to a single call. Layers that re-issue a +call — retries — subclass `AsyncClientInterceptor` and issue it themselves; see +the [advanced guide](https://bedrock-python.github.io/grpc-client-kit/guide/advanced/). **`INTERNAL` is not retryable by default.** `DEFAULT_RETRYABLE_CODES` holds diff --git a/docs/agents.md b/docs/agents.md index 90448a2..6fa5b66 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -235,15 +235,15 @@ Layer classes, all exported and all usable on their own: | Name | Use | |---|---| -| `AsyncAroundClientInterceptor` | subclass and write one `async def around_call(self, call) -> AsyncIterator[None]` that yields exactly once — the whole RPC happens at the `yield`, a response stream to its last item included | +| `AsyncAroundClientInterceptor` | subclass and write one `async def around_call(self, call) -> AsyncIterator[None]` that yields exactly once — the whole RPC happens at the `yield`, a response stream to its last item included. Both sides of the `yield` share one `contextvars.Context`; nothing raised after it reaches the caller | | `AsyncClientInterceptor` | subclass and implement `async def intercept(self, call)` when the call must be issued by hand, re-issued, or not issued at all | | `ClientCall` | `method` (already decoded `str`), `rpc_type`, `details`, `request`, `request_streaming`, `response_streaming`, `response`, `underlying_call`, `await invoke_unary()`, `await invoke_stream()` | | `flatten_interceptors(interceptors)` | expands a mixed chain into what a channel accepts | | `logical_interceptor(entry)` | an adapter back to its interceptor; anything else unchanged | Rewrite the call before the `yield` with `call.details = call.details._replace(timeout=...)`; -raise before the `yield` to refuse the call outright. Swallowing an exception after it is not -supported. +raise before the `yield` to refuse the call outright. After the `yield` the outcome is already +decided: swallowing an exception is not supported, and neither is raising one — see rule 14. ### Balancing @@ -403,29 +403,37 @@ Per-method budgets are the one thing settings cannot express: `timeout` carries interceptor you write against gRPC's own `intercept_*` methods that inherits all four base classes is registered for **unary-unary only**, silently — the channel files each entry into the first list it matches. -14. **`create_client(interceptors=...)` puts your layers in the outer slot**, above logging, +14. **`around_call` cannot change a call's outcome after the `yield`, and both sides of it + share one context.** Whatever the teardown raises — a metrics push to a collector that + went away, a `ContextVar.reset` that was refused — is logged at ERROR against the method + and dropped, identically for all four RPC kinds; only cancellation still propagates. And + a token minted before the `yield` may be reset after it on every kind, because the kit + pins a `contextvars.Context` per call for the three whose teardown finishes in another + task. That costs an `asyncio.Task` or two per call, per layer; unary-unary pins nothing + and pays nothing. +15. **`create_client(interceptors=...)` puts your layers in the outer slot**, above logging, tracing, metrics and the timeout. That is right for metadata injection and wrong for anything that has to read or reshape the deadline; those belong in a hand-built chain given to `GrpcClient(interceptors=...)`. -15. **`interceptors` and `interceptor_factory` are mutually exclusive, and so are `balancer` +16. **`interceptors` and `interceptor_factory` are mutually exclusive, and so are `balancer` and `config.target`.** Both pairs raise `ValueError` at construction rather than picking a winner at run time. A shared `interceptors` list also means one circuit breaker shared across every target of that client; the factory always passes a factory instead. -16. **An unchecked target is not a healthy target.** Every target reads unhealthy until the +17. **An unchecked target is not a healthy target.** Every target reads unhealthy until the first health pass lands, so enter the factory's `async with` (or `await checker.wait_until_ready()`) before the first RPC, or every fresh pod fails its first call with `NoHealthyTargetsError`. A checker that was never started raises `HealthCheckerNotRunningError` from `is_healthy`, and balancers gather health with `return_exceptions=True`, so that mistake otherwise looks exactly like a cluster that is entirely down. -17. **A port is always required.** Targets are validated before a channel exists, more +18. **A port is always required.** Targets are validated before a channel exists, more strictly than gRPC — which silently falls back to 443 for a portless target. `[::1]:50051` must be bracketed; `http://` is rejected by name. -18. **Never stack kit retries on a native `retryPolicy`.** Service-config retries run inside +19. **Never stack kit retries on a native `retryPolicy`.** Service-config retries run inside the channel, below every interceptor, so the two multiply: 3 × 3 = 9 requests reach the server, invisibly to the kit's logs and metrics. `GrpcClient` warns once when it sees both. Native retries *without* kit retries are fully supported. -19. **Batteries are opt-in, and a missing one is a warning, not an error.** +20. **Batteries are opt-in, and a missing one is a warning, not an error.** `import grpc_client_kit` never reaches for an extra. `HealthChecker` resolves on first attribute access and raises `ImportError` naming `[health]` — an `ImportError` and not an `AttributeError`, so a broken install says so rather than looking like a name that never @@ -434,7 +442,7 @@ Per-method budgets are the one thing settings cannot express: `timeout` carries `importlib.util.find_spec("grpc_health")` or catch the ImportError. Tracing, metrics and the deadline budget layers are left out of the chain, with a log line, when their extra is absent — the chain still builds and the calls still run. -20. **There is no sync API and no thread safety.** Everything here assumes one event loop. +21. **There is no sync API and no thread safety.** Everything here assumes one event loop. ## Common mistakes diff --git a/docs/guide/interceptors.md b/docs/guide/interceptors.md index 693d284..36866b0 100644 --- a/docs/guide/interceptors.md +++ b/docs/guide/interceptors.md @@ -156,15 +156,53 @@ class TimingInterceptor(AsyncAroundClientInterceptor): to refuse the call outright: nothing is sent. - **At the `yield`** the call runs, start to finish. A failure arrives as `grpc.aio.AioRpcError`, one halfway through a response stream included. -- **After it** — `except`, `else`, `finally` — the outcome is known. - Swallowing the exception is not supported: there is no response to put in - its place. +- **After it** — `except`, `else`, `finally` — the outcome is known and + nothing done here can change it. Swallowing the exception is not supported: + there is no response to put in its place. Raising is not either: whatever + the teardown raises is logged at `ERROR` against the method and dropped, so + a metrics push to a collector that has gone away cannot take a response the + server already sent with it. `call` carries what a layer needs: `method` (already decoded to `str`), `rpc_type`, `request_streaming` / `response_streaming`, the mutable `details`, `response` once a unary one has arrived, and `underlying_call` for the `grpc.aio.Call` itself. +### Scoping a value to one call + +Both sides of the `yield` run in one `contextvars.Context`, so the pair that +scopes something for the length of a call is written the obvious way and works +on all four RPC kinds: + +```python +from contextvars import ContextVar + +REQUEST_ID: ContextVar[str | None] = ContextVar("request_id", default=None) + + +class RequestId(AsyncAroundClientInterceptor): + """Tag every outgoing call with an id the layers below it can read.""" + + async def around_call(self, call: ClientCall) -> AsyncIterator[None]: + token = REQUEST_ID.set(new_request_id()) + try: + yield + finally: + REQUEST_ID.reset(token) +``` + +What the setup sets is visible to every layer below and to the RPC itself. It +is not visible to your own calling code: `grpc.aio` runs a chain in a task of +its own, so a chain has never been able to write into the caller's context. +OpenTelemetry's `attach` and `detach` work across the `yield` for the same +reason the token pair does. + +The kit pays for that by pinning a context per call on the three kinds whose +teardown finishes somewhere else — after the last item of a response stream, +or once a streaming request's outcome arrives — which costs an `asyncio.Task` +or two per call, per layer. A unary-unary call pins nothing and pays nothing: +its setup, RPC and teardown are one coroutine already. + ## When `intercept` is the right seam A layer that re-issues a call rather than merely wrapping it — a retry — diff --git a/grpc_client_kit/interceptors/base.py b/grpc_client_kit/interceptors/base.py index 45aa7a0..ecf803a 100644 --- a/grpc_client_kit/interceptors/base.py +++ b/grpc_client_kit/interceptors/base.py @@ -33,6 +33,15 @@ async def around_call(self, call: ClientCall) -> AsyncIterator[None]: That one generator covers all four RPC kinds: for a streaming response the base class holds it open until the last item has been delivered, so the ``except`` branch sees a mid-stream failure too. + +Two guarantees the base class gives it, both of them measured rather than assumed. Whatever the +generator raises *after* the ``yield`` is logged and dropped: an interceptor is observability, and a +metrics push or a log write that fails must not turn a call the server answered ``OK`` into an +exception for the application — nor vanish into the event loop's exception handler, which is where +it used to go for a streaming request. And the code before the ``yield`` and the code after it run +in one `contextvars.Context`, so a token minted in the setup can be reset in the teardown, on every +one of the four kinds and not only on the one whose teardown happens to run in the task that started +the call. `_AroundScope` is where both live. """ from __future__ import annotations @@ -40,24 +49,41 @@ async def around_call(self, call: ClientCall) -> AsyncIterator[None]: import abc import asyncio import contextlib +import contextvars import functools -from collections.abc import AsyncIterator, Awaitable, Callable, Iterable +import logging +from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine, Iterable from dataclasses import dataclass, field from typing import Any, Literal import grpc.aio +logger = logging.getLogger(__name__) + # Outcome observers for deferred calls; referenced here so the loop cannot collect them mid-flight. _BACKGROUND_TASKS: set[asyncio.Task[None]] = set() -def _spawn_background(coro: Awaitable[None]) -> None: - """Run an outcome observer as a task the event loop cannot garbage-collect early.""" - task = asyncio.ensure_future(coro) +def _spawn_background(coro: Coroutine[Any, Any, None], context: contextvars.Context | None = None) -> None: + """Run an outcome observer as a task the event loop cannot garbage-collect early. + + Args: + coro: The observer to run. + context: The context to step it in. `None` copies the current one, which is what a task + does by default; an around scope hands over the context its setup ran in, so that the + teardown it eventually reaches is allowed to reset what the setup set. + """ + task = asyncio.get_running_loop().create_task(coro, context=context) _BACKGROUND_TASKS.add(task) task.add_done_callback(_BACKGROUND_TASKS.discard) +def _running_context() -> contextvars.Context | None: + """The context the task running right now was started in, or None outside a task.""" + task = asyncio.current_task() + return task.get_context() if task is not None else None + + type RpcType = Literal["unary_unary", "unary_stream", "stream_unary", "stream_stream"] # What a logical interceptor is handed as a continuation: ``(call_details, request) -> Call``. @@ -236,6 +262,16 @@ class AsyncAroundClientInterceptor(AsyncClientInterceptor, abc.ABC): ordinary ``try/except`` around it covers the whole call. Refusing a call is a matter of raising before the ``yield``: the RPC is then never created. Swallowing an exception is not supported — the call has already failed by then, and there is no response to return in its place. + + Raising *after* the ``yield`` is not a way to fail a call either. The outcome is decided by + then, so whatever the teardown raises is logged at ERROR against the method and dropped — + identically for all four RPC kinds, and whether the call had succeeded or failed. + + The two sides of the ``yield`` share one `contextvars.Context`. The pair that scopes something + for the length of a call is therefore legal here on every kind — ``token = VAR.set(...)`` before + the yield and ``VAR.reset(token)`` after it, or OpenTelemetry's ``attach`` and ``detach`` — and + what the setup sets is visible to the layers below. `_AroundScope` is where that is arranged, + and what it costs. """ @abc.abstractmethod @@ -249,113 +285,199 @@ def _around(self) -> Callable[[ClientCall], contextlib.AbstractAsyncContextManag async def intercept(self, call: ClientCall) -> Any: """Issue the call inside `around_call` (base plumbing; subclasses override the generator).""" - around = self._around(call) - await around.__aenter__() - - if call.response_streaming: + deferred_teardown = call.request_streaming or call.response_streaming + scope = _AroundScope(self._around(call), call.method, pin_context=deferred_teardown) + + if not deferred_teardown: + # Setup, call and teardown are this one coroutine, so the generator already runs start + # to finish in a single context and there is nothing to pin. The kind that dominates + # every chain pays no task for a guarantee it satisfies for free. + await scope.enter() try: - stream = await call.invoke_stream() + response = await call.invoke_unary() except BaseException as error: - await around.__aexit__(type(error), error, error.__traceback__) + await scope.close(error) raise + await scope.close(None) + return response + + # The RPC is created inside the scope's context, not only the setup: what the setup sets + # has to be visible to the layers below, or scoping a value for the length of a call would + # be legal and pointless at the same time. + started = await scope.in_context(_open_call(scope, call)) + + if call.response_streaming: # The teardown has to outlive this method: the call is not over until the last item is. # It must also not depend on the consumer finishing the iteration — a cancelled or # abandoned call fires its done callback, and that closes the teardown deterministically # instead of waiting for garbage collection. - finalizer = _AroundFinalizer(around) - _finalize_when_done(call.underlying_call, finalizer) - return _closing_stream(finalizer, stream) + _finalize_when_done(call.underlying_call, scope) + return _closing_stream(scope, started) - if call.request_streaming: - try: - deferred = await call.invoke_unary() - except BaseException as error: - await around.__aexit__(type(error), error, error.__traceback__) - raise - # The outcome of a streaming-request call arrives after this task has returned (see - # `invoke_unary`), so the teardown runs from an observer instead of from here. - finalizer = _AroundFinalizer(around) - _spawn_background(_finalize_unary_outcome(call, deferred, finalizer)) - return deferred + # The outcome of a streaming-request call arrives after this task has returned (see + # `invoke_unary`), so the teardown runs from an observer instead of from here. + _spawn_background(_finalize_unary_outcome(call, started, scope), context=scope.context) + return started - try: - response = await call.invoke_unary() - except BaseException as error: - await around.__aexit__(type(error), error, error.__traceback__) - raise - await around.__aexit__(None, None, None) - return response +async def _open_call(scope: _AroundScope, call: ClientCall) -> Any: + """Run the setup and issue the RPC, both inside the scope's context. + Args: + scope: The around scope wrapping this call. + call: The call to issue. -class _AroundFinalizer: - """Closes one around context exactly once, from whichever side finishes first. + Returns: + The response iterator of a streaming response, or the `Call` of a streaming request. - The iterator path and the call's done callback both race to report the outcome; whichever - arrives second must find the work already done rather than throw a second exception into a - generator that has already stopped. + Raises: + BaseException: Whatever the setup raised — refusing the call, with no teardown, which is + what raising before the ``yield`` has always meant — or whatever issuing the call + raised, after the teardown has been given it. """ + await scope.enter() + try: + if call.response_streaming: + return await call.invoke_stream() + return await call.invoke_unary() + except BaseException as error: + await scope.close(error) + raise + + +class _AroundScope: + """One `around_call` generator: closed exactly once, in the context its setup ran in. + + Two things a bare pair of ``__aenter__`` / ``__aexit__`` calls does not give the seam, both + measured against a live server rather than inferred. + + **The two sides of the ``yield`` have to share a context.** For three of the four RPC kinds the + teardown is stepped from a different task than the setup — whichever one drains the response + stream, or the observer that awaits a deferred outcome — and `contextvars` compares contexts by + identity, so a token minted before the ``yield`` cannot be reset after it. `asyncio.Task` takes + a ``context=`` and *enters* that object instead of copying it, so one context of this scope's + own, entered by both sides, is what makes the set/reset pair legal on every kind. Unary-unary + pins nothing: its setup, call and teardown are already one coroutine, and a task per call per + layer is a real cost on the busiest path in the kit for a guarantee already held there. + + **A failing teardown is not an outcome.** An interceptor is observability. Something that fails + after the call has finished must not replace the response the server sent, must not truncate a + stream, and must not vanish into the event loop's exception handler either — which were the + three different fates one raising teardown had, depending only on the kind of call it wrapped. + """ + + __slots__ = ("_around", "_closed", "_context", "_method") - __slots__ = ("_around", "_closed") + def __init__( + self, + around: contextlib.AbstractAsyncContextManager[None], + method: str, + *, + pin_context: bool, + ) -> None: + """Wrap one generator, pinning a context for it when its teardown will run elsewhere. - def __init__(self, around: contextlib.AbstractAsyncContextManager[None]) -> None: + Args: + around: The `around_call` generator, already wrapped as an async context manager. + method: Full method name of the call, for the log record a failing teardown produces. + pin_context: Whether to give the generator a context of its own, which every kind but + unary-unary needs. + """ self._around = around + self._method = method + self._context = contextvars.copy_context() if pin_context else None self._closed = False @property def closed(self) -> bool: - """Whether the context has already been closed.""" + """Whether the generator has already been closed.""" return self._closed + @property + def context(self) -> contextvars.Context | None: + """The pinned context, or None when the generator runs wherever it is stepped from.""" + return self._context + + def in_context[T](self, coro: Coroutine[Any, Any, T]) -> asyncio.Task[T]: + """Step one coroutine in the pinned context, as a task of its own.""" + return asyncio.get_running_loop().create_task(coro, context=self._context) + + async def enter(self) -> None: + """Run the setup — the code before the ``yield``. Raising here refuses the call.""" + await self._around.__aenter__() + async def close(self, error: BaseException | None) -> None: - """Close the context with the call's outcome, once.""" + """Close the generator with the call's outcome, once, in the context the setup ran in. + + The iterator path and the call's done callback both race to report the outcome; whichever + arrives second must find the work already done rather than throw a second exception into a + generator that has already stopped. + + Args: + error: How the call ended, or None if it succeeded. + """ if self._closed: return self._closed = True - if error is None: - await self._around.__aexit__(None, None, None) - else: - await self._around.__aexit__(type(error), error, error.__traceback__) + # Already the right context on the unary-unary path, which pins none, and on every caller + # that the scope itself started — the teardown then costs no second task. + if self._context is None or self._context is _running_context(): + await self._teardown(error) + return + await self.in_context(self._teardown(error)) + + async def _teardown(self, error: BaseException | None) -> None: + """Run the code after the ``yield`` and let nothing it raises reach the call.""" + try: + if error is None: + await self._around.__aexit__(None, None, None) + else: + await self._around.__aexit__(type(error), error, error.__traceback__) + except Exception: + # Dropped deliberately, and logged so that it is not lost: see the class docstring. + # Cancellation is not caught — a task torn down around the teardown is not the teardown + # failing, and swallowing it would strand a cancel that the caller asked for. + logger.exception("around_call teardown failed for %s", self._method) -async def _finalize_unary_outcome(call: ClientCall, deferred: Any, finalizer: _AroundFinalizer) -> None: - """Observe a deferred unary outcome and close the around context with it.""" +async def _finalize_unary_outcome(call: ClientCall, deferred: Any, scope: _AroundScope) -> None: + """Observe a deferred unary outcome and close the around scope with it.""" try: response = await deferred if hasattr(deferred, "__await__") else deferred except BaseException as error: - await finalizer.close(error) + await scope.close(error) else: call._response = response - await finalizer.close(None) + await scope.close(None) -def _finalize_when_done(underlying: Any, finalizer: _AroundFinalizer) -> None: - """Arrange for the around context to close when the call finishes, however it finishes.""" +def _finalize_when_done(underlying: Any, scope: _AroundScope) -> None: + """Arrange for the around scope to close when the call finishes, however it finishes.""" add_done_callback = getattr(underlying, "add_done_callback", None) if add_done_callback is None: return def _on_done(done_call: Any) -> None: - if not finalizer.closed: - _spawn_background(_finalize_stream_outcome(done_call, finalizer)) + if not scope.closed: + _spawn_background(_finalize_stream_outcome(done_call, scope), context=scope.context) add_done_callback(_on_done) -async def _finalize_stream_outcome(done_call: Any, finalizer: _AroundFinalizer) -> None: - """Close the around context with the status of an already finished streaming call.""" - if finalizer.closed: +async def _finalize_stream_outcome(done_call: Any, scope: _AroundScope) -> None: + """Close the around scope with the status of an already finished streaming call.""" + if scope.closed: return try: code = await done_call.code() except BaseException as error: - await finalizer.close(error) + await scope.close(error) return if code == grpc.StatusCode.OK: - await finalizer.close(None) + await scope.close(None) elif done_call.cancelled(): - await finalizer.close(asyncio.CancelledError()) + await scope.close(asyncio.CancelledError()) else: failure = grpc.aio.AioRpcError( code, @@ -363,14 +485,14 @@ async def _finalize_stream_outcome(done_call: Any, finalizer: _AroundFinalizer) await done_call.trailing_metadata(), await done_call.details(), ) - await finalizer.close(failure) + await scope.close(failure) async def _closing_stream( - finalizer: _AroundFinalizer, + scope: _AroundScope, stream: AsyncIterator[Any], ) -> AsyncIterator[Any]: - """Yield a whole response stream, then close the around context with however it ended.""" + """Yield a whole response stream, then close the around scope with however it ended.""" try: async for item in stream: yield item @@ -379,14 +501,14 @@ async def _closing_stream( # view the call did not finish; closing with a cancellation says exactly that, while # throwing GeneratorExit into the around generators would only produce "generator didn't # stop after athrow" noise. The done callback of the underlying call, when there is one, - # may report the real outcome first — the finalizer keeps whichever arrived first. - await finalizer.close(asyncio.CancelledError()) + # may report the real outcome first — the scope keeps whichever arrived first. + await scope.close(asyncio.CancelledError()) raise except BaseException as error: - await finalizer.close(error) + await scope.close(error) raise - await finalizer.close(None) + await scope.close(None) class _Adapter: diff --git a/grpc_client_kit/interceptors/tracing.py b/grpc_client_kit/interceptors/tracing.py index 75919f4..90682e6 100644 --- a/grpc_client_kit/interceptors/tracing.py +++ b/grpc_client_kit/interceptors/tracing.py @@ -113,11 +113,15 @@ class AsyncClientTracingInterceptor(AsyncClientInterceptor): Note: This layer implements `base.AsyncClientInterceptor.intercept` rather than the simpler - ``around_call`` seam, because the span must be made current only while the RPC is being - created. ``around_call`` resumes when the *whole* call is over, and for a streaming response - that happens in whichever task drains the stream, while gRPC runs the interceptor chain in a - task of its own: detaching the OpenTelemetry context there fails ("Failed to detach - context") and leaves the client span current in the task that started the call. + ``around_call`` seam. It once had no choice: the teardown of an ``around_call`` resumed in + whichever task drained the stream, and detaching the OpenTelemetry context there failed + ("Failed to detach context"), leaving the client span current in the task that started the + call. `base._AroundScope` steps both sides of the ``yield`` in one context now, so an + ``attach`` / ``detach`` pair across it is supported and that reason is gone. What keeps + this layer on `intercept` is cost: with ``opentelemetry-api`` installed and no SDK + configured — the default state of the ``tracing`` extra — every span is non-recording and + this interceptor does nothing at all, while an around scope would be opened for every call + regardless, and on a streaming call that is a task per RPC. Note: Without the ``tracing`` extra (``opentelemetry-api``) the interceptor is a documented diff --git a/tests/helpers.py b/tests/helpers.py index 1dfdf6f..2c2365e 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -15,6 +15,7 @@ import asyncio from collections.abc import AsyncIterator, Callable, Generator +from contextvars import ContextVar from typing import Any, NamedTuple from unittest.mock import AsyncMock, MagicMock @@ -23,6 +24,7 @@ from grpc_client_kit.channel import ChannelPool from grpc_client_kit.deadline import DeadlineExceededError +from grpc_client_kit.interceptors.base import AsyncAroundClientInterceptor, ClientCall # The method path calls are issued against unless a test needs to tell two methods apart. METHOD = "/pkg.Service/Method" @@ -329,6 +331,63 @@ async def requests(*items: Any) -> AsyncIterator[Any]: yield item +# -------------------------------------------------------------------------------------------- +# Around-call layers, for the two suites that measure the seam itself. +# -------------------------------------------------------------------------------------------- + +# The logger the seam reports a teardown that failed on. +AROUND_LOGGER = "grpc_client_kit.interceptors.base" + +# A value an `around_call` scopes to one call, as a request id or a correlation id would be. +REQUEST_ID: ContextVar[str | None] = ContextVar("request_id", default=None) + + +class TokenAcrossYield(AsyncAroundClientInterceptor): + """Scopes a value to one call: the shortest useful thing this seam can be written for. + + Attributes: + teardowns: How many calls have reached the code after the ``yield``. + reset_error: Whatever resetting the token raised, or None when the reset was allowed. + """ + + def __init__(self) -> None: + """Start with nothing torn down and no failure.""" + self.teardowns = 0 + self.reset_error: BaseException | None = None + + async def around_call(self, call: ClientCall) -> AsyncIterator[None]: + token = REQUEST_ID.set("req-42") + try: + yield + finally: + try: + REQUEST_ID.reset(token) + except ValueError as error: + # Kept rather than raised: a refused reset reaches the caller differently on every + # RPC kind, and what these tests assert is that the reset itself was legal. + self.reset_error = error + self.teardowns += 1 + + +class RaisingTeardown(AsyncAroundClientInterceptor): + """A layer whose teardown fails, as a metrics push to a collector that has gone away would. + + Attributes: + teardowns: How many calls have reached the code after the ``yield``. + """ + + def __init__(self) -> None: + """Start with nothing torn down.""" + self.teardowns = 0 + + async def around_call(self, call: ClientCall) -> AsyncIterator[None]: + try: + yield + finally: + self.teardowns += 1 + raise RuntimeError("the teardown itself failed") + + # -------------------------------------------------------------------------------------------- # Channels, pools and recorders. # -------------------------------------------------------------------------------------------- diff --git a/tests/integration/test_around_call.py b/tests/integration/test_around_call.py new file mode 100644 index 0000000..93fadde --- /dev/null +++ b/tests/integration/test_around_call.py @@ -0,0 +1,68 @@ +"""The `around_call` contract on a live channel: one context across the yield, one outcome per call. + +Neither property exists inside a single task, and a mocked channel has only one. `grpc.aio` runs an +interceptor chain in a task of its own, drains a response stream in whichever task the caller +iterates from, and resolves a streaming request's outcome after the chain has already returned — so +the setup of one generator and its teardown really do run in three different places, and what +`contextvars` and a failing teardown make of that is only visible from here. +""" + +from __future__ import annotations + +import logging + +import pytest + +from grpc_client_kit import flatten_interceptors +from tests.helpers import AROUND_LOGGER, RaisingTeardown, TokenAcrossYield + +from .calls import EVERY_KIND, RpcKind +from .echo_bench import ClientFactory +from .waiting import until + + +@pytest.mark.parametrize("kind", EVERY_KIND) +async def test__around_call__contextvar_token__is_reset_after_the_yield_on_every_kind( + make_client: ClientFactory, + kind: RpcKind, +) -> None: + # Arrange + # A value scoped to one call — a request id, a correlation id, an OpenTelemetry context — is + # the shortest useful thing this seam is written for, and the pair that scopes it is only legal + # when both sides of the yield are stepped in one and the same context. + interceptor = TokenAcrossYield() + stub = await make_client(flatten_interceptors([interceptor])).connect() + + # Act + response = await kind.invoke(stub) + await until(lambda: interceptor.teardowns == 1, message="the teardown never ran") + + # Assert + assert response == kind.expected + assert interceptor.reset_error is None, "the teardown was stepped in a different context than the setup" + + +@pytest.mark.parametrize("kind", EVERY_KIND) +async def test__around_call__teardown_raises__the_call_keeps_the_response_it_had( + make_client: ClientFactory, + kind: RpcKind, + caplog: pytest.LogCaptureFixture, +) -> None: + # Arrange + # One mistake, one treatment. Before this it had three: the exception replaced the response, it + # truncated a response stream, or it landed in the event loop's exception handler where nobody + # was looking — decided by nothing but which of the four kinds the call happened to be. + caplog.set_level(logging.ERROR, logger=AROUND_LOGGER) + interceptor = RaisingTeardown() + stub = await make_client(flatten_interceptors([interceptor])).connect() + + # Act + response = await kind.invoke(stub) + await until(lambda: interceptor.teardowns == 1, message="the teardown never ran") + + # Assert + assert response == kind.expected + reported = [record for record in caplog.records if record.name == AROUND_LOGGER] + assert len(reported) == 1, [record.getMessage() for record in reported] + assert reported[0].exc_info is not None, "the failure was reported without the traceback that explains it" + assert reported[0].exc_info[0] is RuntimeError diff --git a/tests/unit/interceptors/conftest.py b/tests/unit/interceptors/conftest.py index 0973ab7..f314cf4 100644 --- a/tests/unit/interceptors/conftest.py +++ b/tests/unit/interceptors/conftest.py @@ -37,11 +37,15 @@ from grpc_client_kit.interceptors.metrics import AsyncClientMetricsInterceptor from tests.helpers import ( METHOD, + REQUEST_ID, RPC_KINDS, + STREAMING_RESPONSE, + FakeStreamCall, FakeUnaryCall, Wire, adapter_for, await_result, + collect, make_call_details, make_rpc_error, ) @@ -99,6 +103,56 @@ async def run_call( await settle() +async def drive_call( + interceptor: AsyncClientInterceptor, + wire: Continuation, + rpc_type: str = "unary_unary", +) -> Any: + """Drive one call of any kind to its end, with the chain in a task of its own. + + That task is the point. `grpc.aio` runs an interceptor chain in a task it creates and hands the + caller back a `Call` or an iterator to consume in whichever task the caller likes, so the setup + of an `around_call` and its teardown routinely happen on opposite sides of a task boundary. + Driving both halves from one task hides everything that only goes wrong across that boundary. + + Args: + interceptor: The layer under test. + wire: The continuation the call is issued through. + rpc_type: Which of the four kinds to issue. + + Returns: + What the caller received: the response, or a response stream drained into a list. + """ + started = await asyncio.create_task(start_call(interceptor, wire, rpc_type)) + try: + if rpc_type in STREAMING_RESPONSE: + return await collect(started) + return await await_result(started) + finally: + # The assertions that follow must see the completed teardown of deferred outcomes. + await settle() + + +def wire_for(rpc_type: str) -> Wire: + """A continuation answering with the shape of `Call` that RPC kind's response side has.""" + return Wire(FakeStreamCall("a", "b") if rpc_type in STREAMING_RESPONSE else FakeUnaryCall("ok")) + + +def response_for(rpc_type: str) -> Any: + """What a caller receives from `wire_for` when the call succeeds.""" + return ["a", "b"] if rpc_type in STREAMING_RESPONSE else "ok" + + +def context_reading_wire(wire: Continuation, seen: list[str | None]) -> Continuation: + """A continuation noting what `REQUEST_ID` held below the layer, at the moment it issued.""" + + async def _read(details: Any, request: Any) -> Any: + seen.append(REQUEST_ID.get()) + return await wire(details, request) + + return _read + + def nested_wire( interceptor: AsyncClientInterceptor, wire: Continuation, diff --git a/tests/unit/interceptors/test_base.py b/tests/unit/interceptors/test_base.py index 4ff7b0e..1cdfcaa 100644 --- a/tests/unit/interceptors/test_base.py +++ b/tests/unit/interceptors/test_base.py @@ -7,6 +7,9 @@ from __future__ import annotations +import logging + +import grpc import grpc.aio import pytest @@ -14,10 +17,14 @@ from grpc_client_kit.interceptors.circuit_breaker import CircuitBreakerOpenError from grpc_client_kit.interceptors.retry import AsyncRetryInterceptor from tests.helpers import ( + AROUND_LOGGER, METHOD, + REQUEST_ID, RPC_KINDS, FakeStreamCall, FakeUnaryCall, + RaisingTeardown, + TokenAcrossYield, Wire, collect, make_call_details, @@ -25,7 +32,17 @@ refusing_wire, ) -from .conftest import OldStyleInterceptor, Probe, Recorder, run_call, start_call +from .conftest import ( + OldStyleInterceptor, + Probe, + Recorder, + context_reading_wire, + drive_call, + response_for, + run_call, + start_call, + wire_for, +) pytestmark = pytest.mark.unit @@ -210,6 +227,71 @@ async def test__around_call__stream_refused_by_an_inner_layer__still_runs_the_te ] +# -------------------------------------------------------------------------------------------- +# What the two sides of the yield are promised: one context, and no say in the outcome. +# -------------------------------------------------------------------------------------------- + + +@pytest.mark.parametrize("rpc_type", list(RPC_KINDS)) +async def test__around_call__contextvar_token__is_reset_in_the_context_the_setup_ran_in(rpc_type: str) -> None: + """Scoping a value to one call is the shortest thing this seam is for, and it has to work. + + `contextvars` compares contexts by identity, so a teardown stepped from another task than the + setup cannot reset what the setup set — which is where three of the four kinds finish. + """ + # Arrange + interceptor = TokenAcrossYield() + seen: list[str | None] = [] + + # Act + received = await drive_call(interceptor, context_reading_wire(wire_for(rpc_type), seen), rpc_type) + + # Assert + assert interceptor.reset_error is None, "the teardown ran in a different context than the setup" + assert seen == ["req-42"], "what the setup set was invisible to the layers below it" + assert received == response_for(rpc_type) + assert REQUEST_ID.get() is None, "the value escaped the call and reached the caller's context" + + +@pytest.mark.parametrize("rpc_type", list(RPC_KINDS)) +async def test__around_call__teardown_raises__the_response_still_reaches_the_caller( + rpc_type: str, + caplog: pytest.LogCaptureFixture, +) -> None: + """An interceptor is observability: a metrics push that fails must not take the response with it.""" + # Arrange + caplog.set_level(logging.ERROR, logger=AROUND_LOGGER) + interceptor = RaisingTeardown() + + # Act + received = await drive_call(interceptor, wire_for(rpc_type), rpc_type) + + # Assert + assert received == response_for(rpc_type) + assert interceptor.teardowns == 1 + reported = [record for record in caplog.records if record.name == AROUND_LOGGER] + assert [record.getMessage() for record in reported] == [f"around_call teardown failed for {METHOD}"] + assert reported[0].exc_info is not None, "the failure was reported without the traceback that explains it" + assert reported[0].exc_info[0] is RuntimeError + + +async def test__around_call__teardown_raises_on_a_failed_call__the_server_status_survives( + caplog: pytest.LogCaptureFixture, +) -> None: + """A status is an outcome too: the teardown must not replace what the server answered either.""" + # Arrange + caplog.set_level(logging.ERROR, logger=AROUND_LOGGER) + wire = Wire(FakeUnaryCall(error=make_rpc_error(grpc.StatusCode.PERMISSION_DENIED))) + + # Act + with pytest.raises(grpc.aio.AioRpcError) as raised: + await drive_call(RaisingTeardown(), wire) + + # Assert + assert raised.value.code() == grpc.StatusCode.PERMISSION_DENIED + assert [record.name for record in caplog.records] == [AROUND_LOGGER] + + # -------------------------------------------------------------------------------------------- # Assembling a chain out of both interceptor generations. # --------------------------------------------------------------------------------------------