API Reference¶

class AsyncAroundClientInterceptor(AsyncClientInterceptor, abc.ABC):
    """A logical interceptor that wraps calls instead of re-issuing them.

    Subclasses implement `around_call` — an async generator that yields exactly once:

    - code before ``yield`` runs before the RPC is created, which is where the call details may
      still be rewritten;
    - the RPC happens at the ``yield``, including full consumption of a streaming response;
    - code after ``yield``, or in ``except`` / ``finally``, runs once the call has finished.

    A failure reaches the ``yield`` as `grpc.aio.AioRpcError`, mid-stream failures included, so an
    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.
    """

    @abc.abstractmethod
    def around_call(self, call: ClientCall) -> AsyncIterator[None]:
        """Wrap one RPC (async generator: setup before ``yield``, teardown after)."""

    @functools.cached_property
    def _around(self) -> Callable[[ClientCall], contextlib.AbstractAsyncContextManager[None]]:
        # Built once per instance; asynccontextmanager itself makes a fresh manager per call.
        return contextlib.asynccontextmanager(self.around_call)

    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:
            try:
                stream = await call.invoke_stream()
            except BaseException as error:
                await around.__aexit__(type(error), error, error.__traceback__)
                raise
            # 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)

        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

        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

class AsyncCircuitBreakerInterceptor(AsyncAroundClientInterceptor):
    """Async circuit breaker interceptor for gRPC clients.

    This interceptor implements the Circuit Breaker pattern on a per-method basis.
    It prevents cascading failures by "opening" the circuit for methods that fail
    repeatedly, immediately rejecting subsequent calls until a recovery period
    has passed.

    States:
    - CLOSED: Normal operation. Successes and failures are tracked.
    - OPEN: Service is failing. Calls are immediately rejected with CircuitBreakerOpenError.
    - HALF-OPEN: Limited trial calls are allowed to see if the service has recovered.

    Scope:
        An interceptor is bound to one channel, hence to one target. Give each target its own
        instance (``GrpcClientFactory`` does) and state is effectively per (target, method), so one
        failing backend cannot trip the breaker for its healthy peers.

    What counts as a failure:
        Only the statuses in `_should_record_failure` — the ones that describe the server rather
        than the request. A malformed request or a rejected permission is the caller's problem and
        must not be able to trip a circuit. A cancellation is nobody's failure: `asyncio.CancelledError`
        derives from `BaseException`, so it passes this interceptor's handlers untouched and only
        the trial slot it was holding is given back.

    Features:
    - Thread-safe and async-safe implementation.
    - Per-method state isolation.
    - LRU cache for method states to prevent memory leaks in long-running processes.
    - Observability via optional metrics registry.
    """

    def __init__(
        self,
        fail_threshold: int = 5,
        recovery_timeout: float = 60.0,
        half_open_max_calls: int = 1,
        max_methods: int = 1000,
        metrics: GrpcClientMetricsProtocol | None = None,
    ) -> None:
        """Initialize the circuit breaker interceptor.

        Args:
            fail_threshold: Number of failures before transitioning from CLOSED to OPEN.
            recovery_timeout: Seconds to wait in OPEN state before transitioning to HALF-OPEN.
            half_open_max_calls: Maximum number of concurrent trial calls allowed in HALF-OPEN.
            max_methods: Maximum number of distinct methods to track in LRU cache.
            metrics: Optional metrics registry for recording state transitions.

        Raises:
            ValueError: If any threshold or timeout is out of range.
        """
        if fail_threshold <= 0:
            raise ValueError("fail_threshold must be positive")
        if recovery_timeout < 0:
            raise ValueError("recovery_timeout must be non-negative")
        if half_open_max_calls <= 0:
            raise ValueError("half_open_max_calls must be positive")
        if max_methods <= 0:
            raise ValueError("max_methods must be positive")

        self._fail_threshold = fail_threshold
        self._recovery_timeout = recovery_timeout
        self._half_open_max_calls = half_open_max_calls
        self._max_methods = max_methods
        self._metrics = metrics
        # State transitions and rejections are an optional protocol extension of the registry,
        # checked once here rather than with hasattr magic on every transition.
        self._state_metrics: CircuitBreakerMetricsProtocol | None = (
            metrics if isinstance(metrics, CircuitBreakerMetricsProtocol) else None
        )

        # Map: method_name -> state data (using OrderedDict for LRU). No lock: every mutation of
        # the map happens in an await-free section, atomic with respect to the event loop.
        self._states: OrderedDict[str, MethodCircuitState] = OrderedDict()

    async def around_call(self, call: ClientCall) -> AsyncIterator[None]:
        """Refuse the call while the circuit is open, otherwise run it and record how it ended.

        Args:
            call: The call being guarded.

        Yields:
            Once, with the RPC in flight. For a streaming response the yield spans the whole stream,
            so a server that fails halfway through it still counts as a failure.

        Raises:
            CircuitBreakerOpenError: If the circuit for this method is open, or every half-open
                trial slot is taken. Raised before the ``yield``, so the RPC is never created and
                the rejected call never touches the network.
        """
        state = await self._get_method_state(call.method)
        trial = await self._admit(call.method, state)

        try:
            yield
        except grpc.aio.AioRpcError as error:
            if self._should_record_failure(_error_code(error)):
                await self._record_failure(call.method, state)
            raise
        # `asyncio.CancelledError` is a BaseException and so escapes both handlers on purpose: a
        # client that walked away has told us nothing about the health of the server.
        except Exception:
            await self._record_failure(call.method, state)
            raise
        else:
            await self._record_success(call.method, state)
        finally:
            if trial:
                await self._release_trial(state)

    async def get_states(self) -> dict[str, CircuitBreakerStatus]:
        """Get the current state of all monitored methods.

        Returns:
            A dictionary mapping method names to their status (state, failure_count, etc.).
        """
        return {
            method: CircuitBreakerStatus(
                state=state.state.value,
                failure_count=state.failure_count,
                last_failure_time=state.last_failure_time,
                half_open_calls=state.half_open_calls,
            )
            for method, state in self._states.items()
        }

    async def _get_method_state(self, method: str) -> MethodCircuitState:
        """Get or create state for a specific method, implementing LRU eviction.

        Args:
            method: Full gRPC method path.

        Returns:
            The state object for the method.
        """
        if method in self._states:
            # Move to end (mark as most recently used)
            self._states.move_to_end(method)
            return self._states[method]

        if len(self._states) >= self._max_methods:
            self._evict_one_state()

        state = MethodCircuitState()
        self._states[method] = state

        # Initial state metric
        self._update_state_metrics(method, CircuitState.CLOSED)

        return state

    def _evict_one_state(self) -> None:
        """Drop one state to stay under the limit, never a state that is still protecting.

        Evicting an OPEN circuit silently re-closes it: the next call to that method goes out to a
        backend the breaker had declared down, and it takes another `fail_threshold` real network
        failures to open it again — under method churn the breaker degrades into "let threshold
        calls through per LRU cycle". So the victim is the least recently used state that carries
        no signal: CLOSED with a clean failure count first, any CLOSED second, and when every state
        is protecting something the map grows past the limit instead — with a warning, since that
        is a memory bound consciously traded for correctness.
        """
        for candidate, state in self._states.items():
            if state.state is CircuitState.CLOSED and state.failure_count == 0:
                del self._states[candidate]
                return

        for candidate, state in self._states.items():
            if state.state is CircuitState.CLOSED:
                del self._states[candidate]
                return

        logger.warning(
            "Circuit breaker holds %d OPEN/HALF_OPEN states, exceeding max_methods=%d; "
            "growing past the limit instead of evicting a protecting circuit",
            len(self._states),
            self._max_methods,
        )

    def _update_state_metrics(self, method: str, state: CircuitState) -> None:
        """Report a state transition to a registry that opted into breaker metrics."""
        if self._state_metrics is None:
            return

        try:
            self._state_metrics.record_circuit_state(method, state.value)
        except Exception:
            logger.exception("Failed to record circuit breaker state for %s", method)

    def _record_rejection(self, method: str) -> None:
        """Report one locally refused call to a registry that opted into breaker metrics."""
        if self._state_metrics is None:
            return

        try:
            self._state_metrics.record_circuit_rejection(method)
        except Exception:
            logger.exception("Failed to record circuit breaker rejection for %s", method)

    async def _admit(self, method: str, state: MethodCircuitState) -> bool:
        """Decide whether the call may go out, claiming a half-open trial slot when it does.

        Args:
            method: Full gRPC method path, for the error and the logs.
            state: Circuit state of that method.

        Returns:
            True if the call was admitted as a half-open trial and owes the slot back.

        Raises:
            CircuitBreakerOpenError: If the circuit is open, or its trial slots are all taken.
        """
        # Await-free on purpose: atomic with respect to the event loop, no lock needed or useful.
        if state.state == CircuitState.OPEN:
            if time.time() - state.last_failure_time > self._recovery_timeout:
                logger.info("Circuit breaker for %s entering HALF-OPEN state", method)
                state.state = CircuitState.HALF_OPEN
                state.half_open_calls = 0
                self._update_state_metrics(method, CircuitState.HALF_OPEN)
            else:
                self._record_rejection(method)
                raise CircuitBreakerOpenError(method)

        if state.state != CircuitState.HALF_OPEN:
            return False

        if state.half_open_calls >= self._half_open_max_calls:
            logger.debug(
                "Circuit breaker for %s: half-open calls limit reached (%d >= %d)",
                method,
                state.half_open_calls,
                self._half_open_max_calls,
            )
            self._record_rejection(method)
            raise CircuitBreakerOpenError(method)

        state.half_open_calls += 1
        return True

    async def _release_trial(self, state: MethodCircuitState) -> None:
        """Give back the half-open slot a trial call was holding.

        A recorded outcome has already reset the counter on its way to CLOSED or OPEN, so the
        decrement only applies while the circuit is still half-open — which is the case when the
        trial ended in something that was not recorded at all, a cancellation above all.
        """
        if state.state == CircuitState.HALF_OPEN:
            state.half_open_calls = max(0, state.half_open_calls - 1)

    async def _record_success(self, method: str, state: MethodCircuitState) -> None:
        """Record a successful call and potentially transition to CLOSED."""
        if state.state == CircuitState.HALF_OPEN:
            logger.info("Circuit breaker for %s CLOSED after successful half-open test", method)
            state.state = CircuitState.CLOSED
            state.failure_count = 0
            state.half_open_calls = 0
            self._update_state_metrics(method, CircuitState.CLOSED)
        elif state.state == CircuitState.CLOSED:
            # Reset failure count to prevent gradual accumulation of old failures
            state.failure_count = 0

    async def _record_failure(self, method: str, state: MethodCircuitState) -> None:
        """Record a failed call and potentially transition to OPEN."""
        state.failure_count += 1
        state.last_failure_time = time.time()

        if state.state == CircuitState.HALF_OPEN:
            logger.warning("Circuit breaker for %s OPEN after failure in half-open state", method)
            state.state = CircuitState.OPEN
            state.half_open_calls = 0
            self._update_state_metrics(method, CircuitState.OPEN)
        elif state.state == CircuitState.CLOSED and state.failure_count >= self._fail_threshold:
            logger.warning("Circuit breaker for %s OPEN after %d failures", method, state.failure_count)
            state.state = CircuitState.OPEN
            self._update_state_metrics(method, CircuitState.OPEN)

    def _should_record_failure(self, code: grpc.StatusCode | None) -> bool:
        """Determine if a gRPC error code should count as a failure.

        Args:
            code: The status code of the failed call, or None when it carries none.

        Returns:
            True if the failure says something about the server's health.
        """
        return code in (
            grpc.StatusCode.UNAVAILABLE,
            grpc.StatusCode.DEADLINE_EXCEEDED,
            grpc.StatusCode.INTERNAL,
            grpc.StatusCode.RESOURCE_EXHAUSTED,
            grpc.StatusCode.ABORTED,
            grpc.StatusCode.UNKNOWN,
            grpc.StatusCode.DATA_LOSS,
        )

class AsyncClientContextInterceptor(AsyncClientInterceptor):
    """Async client interceptor that injects metadata from a provider.

    This interceptor allows for dynamic metadata injection (e.g., authorization
    tokens, request IDs, tenant IDs) into every outgoing gRPC call.

    Features:
    - Support for both synchronous and asynchronous metadata providers.
    - Automatic validation and normalization of gRPC metadata keys.
    - Automatic encoding of binary metadata values.
    - The same injection on all four RPC kinds, streams included. Injection used to reach
      unary-unary calls only: a channel files an interceptor by class, and an interceptor deriving
      from all four gRPC base classes lands in the first list alone (see `base`).

    Note:
        The metadata is written onto the call details before the RPC exists, so this layer
        implements `base.AsyncClientInterceptor.intercept` and hands the call straight on instead of
        using the ``around_call`` seam. Wrapping a response stream to observe an outcome it has no
        use for would cost a generator hop per item, and that wrapper is not free in another way
        either: when a caller abandons a stream, closing it drives the ``GeneratorExit`` back into
        the wrapper, which the event loop reports as an error if it has closed the generator first
        (measured at loop shutdown as ``RuntimeError: generator didn't stop after athrow()``).
    """

    def __init__(
        self, metadata_provider: Callable[[], dict[str, str | bytes] | Awaitable[dict[str, str | bytes]]]
    ) -> None:
        """Initialize the context interceptor.

        Args:
            metadata_provider: A callable that returns a dictionary of metadata.
                               Can be a regular function or an async function.
        """
        self._provider = metadata_provider

    def _validate_metadata_key(self, key: str) -> str:
        """Validate and normalize metadata key.

        gRPC metadata keys must be:
        - ASCII lowercase
        - No uppercase, no unicode
        - Binary keys must end with '-bin'
        """
        key_lower = key.lower()
        try:
            key_lower.encode("ascii")
        except UnicodeEncodeError as e:
            raise ValueError(f"Metadata key must be ASCII: {key}") from e
        return key_lower

    def _encode_metadata_value(self, key: str, value: str | bytes) -> tuple[str, str | bytes]:
        """Encode metadata value based on key suffix.

        Binary metadata (keys ending with '-bin') can contain arbitrary bytes.
        Text metadata must be ASCII-only.
        """
        if key.endswith("-bin"):
            # Binary metadata - encode as bytes
            if isinstance(value, bytes):
                return (key, value)
            return (key, value.encode("utf-8"))
        else:
            # Text metadata - must be ASCII
            str_value = str(value)
            try:
                str_value.encode("ascii")
            except UnicodeEncodeError as e:
                raise ValueError(f"Non-binary metadata value must be ASCII (key={key}): {str_value}") from e
            return (key, str_value)

    async def _context_metadata(self) -> dict[str, str | bytes]:
        """Ask the provider for this call's metadata, tolerating a provider that fails.

        Returns:
            The provider's metadata, or an empty mapping if it raised. Injection is an enrichment:
            a broken provider must not decide the fate of the call it was supposed to describe.
        """
        try:
            provided = self._provider()
            metadata: dict[str, str | bytes] = await provided if inspect.isawaitable(provided) else provided
        except Exception:
            logger.exception("Metadata provider failed")
            return {}
        else:
            return metadata

    async def intercept(self, call: ClientCall) -> Any:
        """Attach the provider's metadata to the call and issue it.

        Args:
            call: The call to issue; its details are rewritten before the RPC exists.

        Returns:
            The response of a unary call, or the iterator of a streaming one, untouched.

        Raises:
            ValueError: If the provider returned a key or a value gRPC cannot carry. Raising before
                anything is issued means no call ever leaves with metadata that would be rejected
                further down.
        """
        context_metadata = await self._context_metadata()

        if context_metadata:
            metadata = list(call.details.metadata or [])
            for key, value in context_metadata.items():
                if value is not None:
                    key_normalized = self._validate_metadata_key(key)
                    encoded_key, encoded_value = self._encode_metadata_value(key_normalized, value)
                    metadata.append((encoded_key, encoded_value))

            call.details = call.details._replace(metadata=metadata)

        if call.response_streaming:
            return await call.invoke_stream()

        return await call.invoke_unary()

class AsyncClientInterceptor(abc.ABC):
    """A logical client interceptor: one implementation covering all four RPC kinds.

    Subclasses implement `intercept`, which is handed a `ClientCall` and owns the call from there:
    it decides whether to issue it at all, how many times, and with which details. That is the hook
    for layers that re-issue calls — retries — or refuse them outright. Everything that only needs
    to *wrap* a call should subclass `AsyncAroundClientInterceptor` instead and write one generator.

    Deliberately not a `grpc.aio.ClientInterceptor`: passing one straight to a channel raises
    ``ValueError`` from grpc instead of silently registering it for unary-unary calls only. Chains
    reach a channel through `flatten_interceptors`, which turns each logical interceptor into the
    four adapters a channel can file correctly.
    """

    @abc.abstractmethod
    async def intercept(self, call: ClientCall) -> Any:
        """Run one RPC and return its outcome.

        Args:
            call: The call to run, and the handle used to issue it.

        Returns:
            For a unary response, the response message (`ClientCall.invoke_unary` returns it). For a
            streaming response, the response iterator — which must be built over an *already issued*
            call, i.e. over the result of `ClientCall.invoke_stream`.

        Raises:
            grpc.aio.AioRpcError: To fail the call, whether the error came off the wire or was
                raised by this interceptor without issuing anything.
        """

    @functools.cached_property
    def adapters(self) -> tuple[grpc.aio.ClientInterceptor, ...]:
        """The four single-ABC objects through which a channel registers this interceptor.

        Built once per instance and kept: a chain's identity — and with it the channel pool's idea
        of which channels are interchangeable — is the identity of the objects in it.
        """
        return (
            _UnaryUnaryAdapter(self),
            _UnaryStreamAdapter(self),
            _StreamUnaryAdapter(self),
            _StreamStreamAdapter(self),
        )

    async def _run_rpc(self, rpc_type: RpcType, continuation: Continuation, details: Any, request: Any) -> Any:
        """Turn one grpc ``intercept_*`` invocation into an `intercept` call (adapter plumbing)."""
        method = details.method
        call = ClientCall(
            method=method.decode("utf-8", errors="replace") if isinstance(method, bytes) else method,
            rpc_type=rpc_type,
            details=details,
            request=request,
            _continuation=continuation,
        )
        result = await self.intercept(call)

        if call.response_streaming:
            return result
        # grpc.aio wraps a bare response into a Call for unary-unary only; stream-unary passes it
        # to the caller untouched, where a message without ``__await__`` breaks the call and its
        # finalizer. Handing back the Call is correct for both, and it keeps the status and the
        # trailing metadata that a bare response drops.
        return call.underlying_call if call.underlying_call is not None else result

class AsyncLoggingInterceptor(AsyncAroundClientInterceptor):
    """Async logging interceptor for gRPC clients with sensitive data filtering.

    Provides comprehensive structured logging for every gRPC call, including
    start time, duration, status, and metadata.

    Features:
    - **Real Outcomes**: The terminal record carries the status the caller observed, whether it
      arrived at the end of a unary call or in the middle of a response stream.
    - **Sensitive Data Redaction**: Automatically masks common authentication headers.
    - **Method-Level Sensitivity**: Entire methods or regex patterns can be marked
      as sensitive to prevent payload logging.
    - **Request ID Correlation**: Automatically extracts and includes `request-id`
      from metadata in all log entries.
    - **Payload Logging**: Optional logging of request and response bodies (truncated to 1KB).
    - **LRU Cache**: Efficiently caches sensitivity check results for method names.
    - **Structured Logging**: Uses the standard Python `logging` module with `extra` dict.

    Note:
        Every log record gets a freshly built `extra` mapping derived from the call-wide fields
        (service, method, request id, sensitivity). Records therefore never inherit fields from an
        earlier stage of the same call, and handlers can retain the mapping they were given.
    """

    def __init__(
        self,
        service_name: str,
        sensitive_methods: set[str] | None = None,
        sensitive_patterns: list[str] | None = None,
        sensitive_headers: set[str] | None = None,
        log_request_payload: bool = False,
        log_response_payload: bool = False,
        log_metadata: bool = True,
        max_cache_size: int = 1000,
        success_log_level: int = logging.INFO,
    ) -> None:
        """Initialize the logging interceptor.

        Args:
            service_name: Name of the service for log categorization.
            sensitive_methods: Set of full method names to treat as sensitive.
            sensitive_patterns: List of regex patterns for sensitive method names.
            sensitive_headers: Set of header names (case-insensitive) to redact.
            log_request_payload: Whether to log the request object.
            log_response_payload: Whether to log the response object.
            log_metadata: Whether to log gRPC metadata (redacted).
            max_cache_size: Size of the sensitivity check LRU cache.
            success_log_level: Level of the record a successful call emits. One INFO line per
                successful RPC is a flood at high QPS — drop this to ``logging.DEBUG`` there, or
                keep the default and set the logger's own level instead.
        """
        self._service_name = service_name
        self._sensitive_methods = sensitive_methods or set()
        self._sensitive_patterns = [re.compile(pattern) for pattern in (sensitive_patterns or [])]
        self._sensitive_headers = {h.lower() for h in (sensitive_headers or DEFAULT_SENSITIVE_HEADERS)}
        self._log_request_payload = log_request_payload
        self._log_response_payload = log_response_payload
        self._log_metadata = log_metadata
        self._success_log_level = success_log_level

        # The service name never changes, so neither does the logger: resolving it per call costs
        # an f-string plus logging's module lock, on the hot path, for nothing.
        self._log = logging.getLogger(f"grpc.client.{service_name}")

        # LRU cache for sensitivity checks. No lock on purpose: the critical section below contains
        # no await, so it is atomic with respect to the event loop — and asyncio.Lock offers no
        # cross-thread protection anyway. Keep it await-free or bring the lock back.
        self._sensitivity_cache: OrderedDict[str, bool] = OrderedDict()
        self._max_cache_size = max_cache_size

    def _is_sensitive(self, method: str) -> bool:
        """Check if method is sensitive, with LRU caching."""
        if method in self._sensitivity_cache:
            self._sensitivity_cache.move_to_end(method)
            return self._sensitivity_cache[method]

        is_sensitive = method in self._sensitive_methods or any(
            pattern.match(method) for pattern in self._sensitive_patterns
        )

        if len(self._sensitivity_cache) >= self._max_cache_size:
            self._sensitivity_cache.popitem(last=False)

        self._sensitivity_cache[method] = is_sensitive
        return is_sensitive

    def _extract_request_id(self, metadata: Mapping[str, str]) -> str | None:
        """Extract the request ID used for log correlation.

        Args:
            metadata: Metadata already normalized by
                :func:`~grpc_client_kit.utils.metadata_to_dict`.

        Returns:
            The correlation id, or None when the call carries none.
        """
        for key, value in metadata.items():
            if key.lower() in ("request-id", "x-request-id"):
                return value
        return None

    def _redact(self, metadata: Mapping[str, str]) -> dict[str, str]:
        """Return a copy of the metadata with sensitive header values masked."""
        return {key: ("***" if key.lower() in self._sensitive_headers else value) for key, value in metadata.items()}

    def _finish_extra(self, base_extra: _LogExtra, start_time: float, status: str) -> _LogExtra:
        """Build the `extra` mapping of a terminal record: call-wide fields plus duration/status."""
        return {
            **base_extra,
            "grpc.duration_ms": (time.perf_counter() - start_time) * 1000,
            "grpc.status": status,
        }

    def _failure_extra(
        self,
        base_extra: _LogExtra,
        start_time: float,
        error: BaseException,
        is_sensitive: bool,
    ) -> tuple[_LogExtra, bool, str]:
        """Describe a failed call.

        Args:
            base_extra: Call-wide log fields.
            start_time: Value of `time.perf_counter()` taken when the call started.
            error: The gRPC error raised by the call.
            is_sensitive: Whether the method is marked sensitive (suppresses error details).

        Returns:
            A tuple of the record's `extra` mapping, whether the failure deserves a stack
            trace, and the gRPC error details.
        """
        code_func = getattr(error, "code", None)
        code: grpc.StatusCode | None = code_func() if callable(code_func) else None
        details_func = getattr(error, "details", None)
        details = str(details_func() if callable(details_func) else error)

        extra = self._finish_extra(base_extra, start_time, code.name if code else "UNKNOWN")
        if not is_sensitive:
            extra["grpc.error"] = details

        return extra, code in _CRITICAL_STATUS_CODES or code is None, details

    def _log_start(
        self,
        log: logging.Logger,
        base_extra: _LogExtra,
        metadata: Mapping[str, str],
        call: ClientCall,
        is_sensitive: bool,
    ) -> None:
        """Emit the call-start debug record."""
        if is_sensitive:
            log.debug("gRPC call started (SENSITIVE)", extra=dict(base_extra))
            return

        extra = dict(base_extra)
        if self._log_metadata and metadata:
            extra["grpc.metadata"] = str(self._redact(metadata))

        if self._log_request_payload:
            if call.request_streaming:
                extra["grpc.request"] = "<stream_request>"
            else:
                extra["grpc.request"] = str(call.request)[:_MAX_PAYLOAD_CHARS]

        log.debug("gRPC call started", extra=extra)

    def _log_rpc_failure(
        self,
        log: logging.Logger,
        extra: _LogExtra,
        is_critical: bool,
        details: str,
        subject: str,
    ) -> None:
        """Emit the terminal record of a call that came back with a gRPC status."""
        if is_critical:
            log.exception("gRPC %s failed with critical error", subject, extra=extra)
        else:
            # error() instead of exception() to avoid long tracebacks for expected errors
            log.error("gRPC %s failed with status %s: %s", subject, extra["grpc.status"], details, extra=extra)

    def _log_success(
        self,
        log: logging.Logger,
        extra: _LogExtra,
        call: ClientCall,
        is_sensitive: bool,
        subject: str,
    ) -> None:
        """Emit the terminal record of a call that ended with OK."""
        if is_sensitive:
            log.log(self._success_log_level, "gRPC %s successful (SENSITIVE)", subject, extra=extra)
            return

        # A streaming response has no single payload to show: `call.response` stays None for it.
        if self._log_response_payload and not call.response_streaming:
            extra["grpc.response"] = str(call.response)[:_MAX_PAYLOAD_CHARS]
        log.log(self._success_log_level, "gRPC %s successful", subject, extra=extra)

    async def around_call(self, call: ClientCall) -> AsyncIterator[None]:
        """Log one RPC: a record when it starts, and exactly one saying how it ended."""
        log = self._log

        # A logger that will emit nothing must cost nothing: no metadata normalization, no
        # sensitivity lookup, no extra dicts. WARNING is the lowest level any record of this
        # interceptor is emitted at when things go wrong, so it is the gate.
        if not log.isEnabledFor(logging.WARNING):
            yield
            return

        is_sensitive = self._is_sensitive(call.method)

        # Metadata is normalized exactly once per call: both the correlation id and the redacted
        # metadata field read from this dict instead of walking the raw pairs again.
        metadata = metadata_to_dict(call.details.metadata)
        request_id = self._extract_request_id(metadata)

        base_extra: _LogExtra = {
            "grpc.service": self._service_name,
            "grpc.method": call.method,
        }
        if request_id:
            base_extra["request_id"] = request_id
        if is_sensitive:
            base_extra["grpc.sensitive"] = True

        # Formatting metadata and payloads is pure waste when DEBUG is off, and this is a hot path.
        if log.isEnabledFor(logging.DEBUG):
            self._log_start(log, base_extra, metadata, call, is_sensitive)

        # Streaming responses are reported as streams: they end when their last item is delivered.
        subject = "stream" if call.response_streaming else "call"
        start_time = time.perf_counter()

        try:
            yield
        except (asyncio.CancelledError, GeneratorExit):
            # GeneratorExit is a caller walking away from a stream, which ends the call as surely as
            # a cancellation does.
            if log.isEnabledFor(logging.INFO):
                log.info("gRPC %s cancelled", subject, extra=self._finish_extra(base_extra, start_time, "CANCELLED"))
            raise
        except CircuitBreakerOpenError as error:
            # A rejection by the local breaker is expected behaviour while the circuit recovers,
            # not a fresh server failure: one WARNING per rejection, never an ERROR flood that
            # drowns the handful of genuine transition records.
            extra = self._finish_extra(base_extra, start_time, "UNAVAILABLE")
            log.warning("gRPC %s refused by the open circuit breaker: %s", subject, error.details(), extra=extra)
            raise
        except grpc.aio.AioRpcError as error:
            extra, is_critical, details = self._failure_extra(base_extra, start_time, error, is_sensitive)
            self._log_rpc_failure(log, extra, is_critical, details, subject)
            raise
        except Exception as error:
            extra = self._finish_extra(base_extra, start_time, "INTERNAL")
            if not is_sensitive:
                extra["grpc.error"] = str(error)
            log.exception("gRPC %s failed with unexpected error", subject, extra=extra)
            raise
        else:
            if log.isEnabledFor(self._success_log_level):
                extra = self._finish_extra(base_extra, start_time, "OK")
                self._log_success(log, extra, call, is_sensitive, subject)