ray-project--ray/python/ray/serve/_private/proxy.py

import asyncio
import gc
import json
import logging
import os
import pickle
import time
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, Generator, Optional, Set, Tuple

import grpc
import starlette
import starlette.routing
from packaging import version
from starlette.types import Receive

import ray
from ray._common.filters import CoreContextFilter
from ray._common.utils import get_or_create_event_loop
from ray.serve._private.common import (
    DeploymentID,
    EndpointInfo,
    NodeId,
    ReplicaID,
    RequestMetadata,
    RequestProtocol,
    gRPCStreamingRequest,
)
from ray.serve._private.constants import (
    HEALTHY_MESSAGE,
    PROXY_MIN_DRAINING_PERIOD_S,
    RAY_SERVE_ENABLE_HA_PROXY,
    RAY_SERVE_ENABLE_PROXY_GC_OPTIMIZATIONS,
    RAY_SERVE_HAPROXY_METRICS_ENABLED,
    RAY_SERVE_PROXY_GC_THRESHOLD,
    RAY_SERVE_REQUEST_PATH_LOG_BUFFER_SIZE,
    RAY_SERVE_WORKER_PROXY_GRPC_PORT,
    RAY_SERVE_WORKER_PROXY_HTTP_PORT,
    SERVE_CONTROLLER_NAME,
    SERVE_HTTP_REQUEST_ID_HEADER,
    SERVE_LOG_COMPONENT,
    SERVE_LOG_COMPONENT_ID,
    SERVE_LOG_REQUEST_ID,
    SERVE_LOG_ROUTE,
    SERVE_LOGGER_NAME,
    SERVE_MULTIPLEXED_MODEL_ID,
    SERVE_NAMESPACE,
)
from ray.serve._private.default_impl import get_proxy_handle
from ray.serve._private.event_loop_monitoring import EventLoopMonitor
from ray.serve._private.grpc_util import (
    get_grpc_response_status,
    set_grpc_code_and_details,
    start_grpc_server,
)
from ray.serve._private.http_util import (
    MessageQueue,
    _matches_session_id_header,
    configure_http_middlewares,
    convert_object_to_asgi_messages,
    get_http_response_status,
    parse_disconnect_disabled_header,
    parse_request_timeout_header,
    receive_http_body,
    send_http_response_on_exception,
    start_asgi_http_server,
)
from ray.serve._private.logging_utils import (
    access_log_msg,
    configure_component_logger,
    configure_component_memory_profiler,
    format_client_address,
    get_component_logger_file_path,
)
from ray.serve._private.long_poll import LongPollClient, LongPollNamespace
from ray.serve._private.proxy_request_response import (
    ASGIProxyRequest,
    HandlerMetadata,
    ProxyRequest,
    ResponseGenerator,
    ResponseHandlerInfo,
    ResponseStatus,
    gRPCProxyRequest,
    gRPCStreamingType,
)
from ray.serve._private.proxy_response_generator import ProxyResponseGenerator
from ray.serve._private.proxy_router import ProxyRouter
from ray.serve._private.request_ingress_metrics import RequestIngressMetrics
from ray.serve._private.tracing_utils import (
    is_span_recording,
    set_http_span_attributes,
    set_rpc_span_attributes,
    set_span_attributes,
    set_span_exception,
    set_span_name,
    set_trace_status,
    setup_tracing,
    tracing_decorator_factory,
)
from ray.serve._private.usage import ServeUsageTag
from ray.serve._private.utils import (
    asyncio_grpc_exception_handler,
    generate_request_id,
    get_head_node_id,
    is_grpc_enabled,
)
from ray.serve.config import HTTPOptions, gRPCOptions
from ray.serve.generated.serve_pb2 import HealthzResponse, ListApplicationsResponse
from ray.serve.handle import DeploymentHandle
from ray.serve.schema import EncodingType, LoggingConfig

logger = logging.getLogger(SERVE_LOGGER_NAME)

SOCKET_REUSE_PORT_ENABLED = (
    os.environ.get("SERVE_SOCKET_REUSE_PORT_ENABLED", "1") == "1"
)

if os.environ.get("SERVE_REQUEST_PROCESSING_TIMEOUT_S") is not None:
    logger.warning(
        "The `SERVE_REQUEST_PROCESSING_TIMEOUT_S` environment variable has "
        "been deprecated. Please set `request_timeout_s` in your Serve config's "
        "`http_options` or `grpc_options` field instead. `SERVE_REQUEST_PROCESSING_TIMEOUT_S` will be "
        "ignored in future versions. See: https://docs.ray.io/en/releases-2.5.1/serve/a"
        "pi/doc/ray.serve.schema.HTTPOptionsSchema.html#ray.serve.schema.HTTPOptionsSch"
        "ema.request_timeout_s and https://docs.ray.io/en/latest/serve/api/"
        "doc/ray.serve.config.gRPCOptions.request_timeout_s.html#"
        "ray.serve.config.gRPCOptions.request_timeout_s"
    )


INITIAL_BACKOFF_PERIOD_SEC = 0.05
MAX_BACKOFF_PERIOD_SEC = 5

DRAINING_MESSAGE = "This node is being drained."


class GenericProxy(ABC):
    """This class is served as the base class for different types of proxies.
    It contains all the common setup and methods required for running a proxy.

    The proxy subclass need to implement the following methods:
      - `protocol()`
      - `not_found_response()`
      - `routes_response()`
      - `health_response()`
      - `setup_request_context_and_handle()`
      - `send_request_to_replica()`
    """

    def __init__(
        self,
        node_id: NodeId,
        node_ip_address: str,
        is_head: bool,
        proxy_router: ProxyRouter,
        request_timeout_s: Optional[float] = None,
        access_log_context: Dict[str, Any] = None,
    ):
        self.request_timeout_s = request_timeout_s
        if self.request_timeout_s is not None and self.request_timeout_s < 0:
            self.request_timeout_s = None

        self._node_id = node_id
        self._is_head = is_head

        self.proxy_router = proxy_router

        self._proxy_metrics = RequestIngressMetrics(
            self.protocol,
            source="proxy",
            node_id=node_id,
            node_ip_address=node_ip_address,
        )

        # `self._ongoing_requests` is used to count the number of ongoing requests
        self._ongoing_requests = 0
        # The time when the node starts to drain.
        # The node is not draining if it's None.
        self._draining_start_time: Optional[float] = None

        self._access_log_context = access_log_context or {}

        getattr(ServeUsageTag, f"{self.protocol.upper()}_PROXY_USED").record("1")

    @property
    @abstractmethod
    def protocol(self) -> RequestProtocol:
        """Protocol used in the proxy.

        Each proxy needs to implement its own logic for setting up the protocol.
        """
        raise NotImplementedError

    def _is_draining(self) -> bool:
        """Whether is proxy actor is in the draining status or not."""
        return self._draining_start_time is not None

    def is_drained(self):
        """Check whether the proxy actor is drained or not.

        A proxy actor is drained if it has no ongoing requests
        AND it has been draining for more than
        `PROXY_MIN_DRAINING_PERIOD_S` seconds.
        """
        if not self._is_draining():
            return False

        return (not self._ongoing_requests) and (
            (time.time() - self._draining_start_time) > PROXY_MIN_DRAINING_PERIOD_S
        )

    def update_draining(self, draining: bool):
        """Update the draining status of the proxy.

        This is called by the proxy state manager
        to drain or un-drain the proxy actor.
        """

        if draining and (not self._is_draining()):
            logger.info(
                f"Start to drain the proxy actor on node {self._node_id}.",
                extra={"log_to_stderr": False},
            )
            self._draining_start_time = time.time()
        if (not draining) and self._is_draining():
            logger.info(
                f"Stop draining the proxy actor on node {self._node_id}.",
                extra={"log_to_stderr": False},
            )
            self._draining_start_time = None

    @abstractmethod
    async def not_found_response(
        self, proxy_request: ProxyRequest
    ) -> ResponseGenerator:
        raise NotImplementedError

    @abstractmethod
    async def routes_response(
        self, *, healthy: bool, message: str
    ) -> ResponseGenerator:
        raise NotImplementedError

    @abstractmethod
    async def health_response(
        self, *, healthy: bool, message: str
    ) -> ResponseGenerator:
        raise NotImplementedError

    def _should_emit_request_ingress_metrics(self) -> bool:
        """Whether this proxy emits the RequestIngressMetrics family itself.

        In HAProxy mode (with HAProxy metrics enabled), this proxy runs as the
        head-node fallback behind HAProxy. HAProxy already counts every request
        it forwards -- including the ones it routes to this fallback -- from its
        per-request log datagrams, so emitting here too would double-count HTTP
        ingress metrics. gRPC is not proxied by HAProxy, so the fallback proxy
        still emits gRPC.
        """
        return not (
            RAY_SERVE_ENABLE_HA_PROXY
            and RAY_SERVE_HAPROXY_METRICS_ENABLED
            and self.protocol == RequestProtocol.HTTP
        )

    def _ongoing_requests_start(self):
        """Ongoing requests start.

        The current autoscale logic can downscale nodes with ongoing requests if the
        node doesn't have replicas and has no primary copies of objects in the object
        store. The counter and the dummy object reference will help to keep the node
        alive while draining requests, so they are not dropped unintentionally.
        """
        self._ongoing_requests += 1
        if self._should_emit_request_ingress_metrics():
            self._proxy_metrics.set_num_ongoing_requests(self._ongoing_requests)

    def _ongoing_requests_end(self):
        """Ongoing requests end.

        Decrement the ongoing request counter and drop the dummy object reference
        signaling that the node can be downscaled safely.
        """
        self._ongoing_requests -= 1
        if self._should_emit_request_ingress_metrics():
            self._proxy_metrics.set_num_ongoing_requests(self._ongoing_requests)

    def _setup_proxy_tracing(
        self,
        request_id: str,
        handle: DeploymentHandle,
        proxy_request: ProxyRequest,
        additional_attributes: Optional[Dict[str, Any]] = None,
    ):
        """Set up tracing attributes for proxy requests.

        Args:
            request_id: The unique request ID.
            handle: The deployment handle.
            proxy_request: The proxy request object.
            additional_attributes: Optional additional tracing attributes to set.
        """
        trace_attributes = {
            "request_id": request_id,
            "deployment": handle.deployment_name,
            "app": handle.app_name,
            "request_type": proxy_request.request_type,
        }
        if additional_attributes:
            trace_attributes.update(additional_attributes)
        set_span_attributes(trace_attributes)

    def _finalize_proxy_tracing(
        self,
        status: Optional[ResponseStatus],
        exc: Optional[BaseException],
    ):
        """Finalize tracing for proxy requests.

        Set exception and trace status. This is a common helper used by
        both HTTP and gRPC proxies. Protocol-specific attributes should be
        set separately (e.g., via set_http_span_attributes or set_rpc_span_attributes).

        Args:
            status: The response status, if available.
            exc: The exception that occurred, if any.
        """
        if exc:
            set_span_exception(exc, escaped=True)
            if status is not None:
                set_trace_status(status.is_error, str(exc))
        elif status is not None:
            set_trace_status(status.is_error)

    def _get_health_or_routes_reponse(
        self, proxy_request: ProxyRequest
    ) -> ResponseHandlerInfo:
        """Get the response handler for system health and route endpoints.

        If the proxy is draining or has not yet received a route table update from the
        controller, both will return a non-OK status.
        """
        router_ready_for_traffic, router_msg = self.proxy_router.ready_for_traffic(
            self._is_head
        )
        if self._is_draining():
            healthy = False
            message = DRAINING_MESSAGE
        elif not router_ready_for_traffic:
            healthy = False
            message = router_msg
        else:
            healthy = True
            message = HEALTHY_MESSAGE

        if proxy_request.is_health_request:
            response_generator = self.health_response(healthy=healthy, message=message)
        else:
            assert proxy_request.is_route_request
            response_generator = self.routes_response(healthy=healthy, message=message)

        return ResponseHandlerInfo(
            response_generator=response_generator,
            metadata=HandlerMetadata(
                route=proxy_request.route_path,
            ),
            should_record_access_log=False,
            should_increment_ongoing_requests=False,
        )

    def _get_response_handler_info(
        self, proxy_request: ProxyRequest
    ) -> ResponseHandlerInfo:
        if proxy_request.is_health_request or proxy_request.is_route_request:
            return self._get_health_or_routes_reponse(proxy_request)

        proxy_request.populate_tracing_context()
        matched_route = None
        if self.protocol == RequestProtocol.HTTP:
            matched_route = self.proxy_router.match_route(proxy_request.route_path)
        elif self.protocol == RequestProtocol.GRPC:
            matched_route = self.proxy_router.get_handle_for_endpoint(
                proxy_request.route_path
            )

        if matched_route is None:
            return ResponseHandlerInfo(
                response_generator=self.not_found_response(proxy_request),
                metadata=HandlerMetadata(
                    # Don't include the invalid route prefix because it can blow up our
                    # metrics' cardinality.
                    # See: https://github.com/ray-project/ray/issues/47999
                    route="",
                ),
                should_record_access_log=True,
                should_increment_ongoing_requests=False,
            )
        else:
            route_prefix, handle, app_is_cross_language = matched_route
            # Modify the path and root path so that reverse lookups and redirection
            # work as expected. We do this here instead of in replicas so it can be
            # changed without restarting the replicas.
            route_path = proxy_request.route_path
            if route_prefix != "/" and self.protocol == RequestProtocol.HTTP:
                assert not route_prefix.endswith("/")
                proxy_request.set_root_path(proxy_request.root_path + route_prefix)
                # NOTE(edoakes): starlette<0.33.0 expected the ASGI 'root_prefix'
                # to be stripped from the 'path', which wasn't technically following
                # the standard. See https://github.com/encode/starlette/pull/2352.
                if version.parse(starlette.__version__) < version.parse("0.33.0"):
                    proxy_request.set_path(route_path.replace(route_prefix, "", 1))

            # NOTE(abrar): we try to match to a specific route pattern (e.g., /api/{user_id})
            # for logs & metrics when available. If no pattern matches, we fall back to the
            # route_prefix to avoid high cardinality.
            # See: https://github.com/ray-project/ray/issues/47999 and
            # https://github.com/ray-project/ray/issues/52212
            if self.protocol == RequestProtocol.HTTP:
                logs_and_metrics_route = self.proxy_router.match_route_pattern(
                    route_prefix, proxy_request.scope
                )
            else:
                logs_and_metrics_route = handle.deployment_id.app_name
            internal_request_id = generate_request_id()
            handle, request_id = self.setup_request_context_and_handle(
                app_name=handle.deployment_id.app_name,
                handle=handle,
                route=logs_and_metrics_route,
                proxy_request=proxy_request,
                internal_request_id=internal_request_id,
            )

            response_generator = self.send_request_to_replica(
                request_id=request_id,
                internal_request_id=internal_request_id,
                handle=handle,
                proxy_request=proxy_request,
                app_is_cross_language=app_is_cross_language,
            )

            return ResponseHandlerInfo(
                response_generator=response_generator,
                metadata=HandlerMetadata(
                    application_name=handle.deployment_id.app_name,
                    deployment_name=handle.deployment_id.name,
                    route=logs_and_metrics_route,
                ),
                should_record_access_log=True,
                should_increment_ongoing_requests=True,
            )

    async def proxy_request(self, proxy_request: ProxyRequest) -> ResponseGenerator:
        """Wrapper for proxy request.

        This method is served as common entry point by the proxy. It handles the
        routing, including routes and health checks, ongoing request counter,
        and metrics.
        """
        assert proxy_request.request_type in {"http", "websocket", "grpc"}

        response_handler_info = self._get_response_handler_info(proxy_request)

        start_time = time.time()
        if response_handler_info.should_increment_ongoing_requests:
            self._ongoing_requests_start()

        try:
            # The final message yielded must always be the `ResponseStatus`.
            status: Optional[ResponseStatus] = None
            async for message in response_handler_info.response_generator:
                if isinstance(message, ResponseStatus):
                    status = message

                yield message

            assert status is not None and isinstance(status, ResponseStatus)
        finally:
            # If anything during the request failed, we still want to ensure the ongoing
            # request counter is decremented.
            if response_handler_info.should_increment_ongoing_requests:
                self._ongoing_requests_end()

        latency_ms = (time.time() - start_time) * 1000.0
        status_code = (
            status.code.name
            if self.protocol == RequestProtocol.GRPC
            else str(status.code)
        )
        if response_handler_info.should_record_access_log:
            request_context = ray.serve.context._get_serve_request_context()
            self._access_log_context[SERVE_LOG_ROUTE] = request_context.route
            self._access_log_context[SERVE_LOG_REQUEST_ID] = request_context.request_id
            logger.info(
                access_log_msg(
                    method=proxy_request.method,
                    route=request_context.route,
                    status=status_code,
                    latency_ms=latency_ms,
                    client=format_client_address(proxy_request.client),
                ),
                extra=self._access_log_context,
            )

        if self._should_emit_request_ingress_metrics():
            self._proxy_metrics.record_request(
                route=response_handler_info.metadata.route,
                method=proxy_request.method,
                application=response_handler_info.metadata.application_name,
                status_code=status_code,
                latency_ms=latency_ms,
                is_error=status.is_error,
                deployment_name=response_handler_info.metadata.deployment_name,
            )

    @abstractmethod
    def setup_request_context_and_handle(
        self,
        app_name: str,
        handle: DeploymentHandle,
        route: str,
        proxy_request: ProxyRequest,
        internal_request_id: str,
    ) -> Tuple[DeploymentHandle, str]:
        """Setup the request context and handle for the request.

        Each proxy needs to implement its own logic for setting up the request context
        and handle.
        """
        raise NotImplementedError

    @abstractmethod
    async def send_request_to_replica(
        self,
        request_id: str,
        internal_request_id: str,
        handle: DeploymentHandle,
        proxy_request: ProxyRequest,
        app_is_cross_language: bool = False,
    ) -> ResponseGenerator:
        """Send the request to the replica and handle streaming response.

        Each proxy needs to implement its own logic for sending the request and
        handling the streaming response.
        """
        raise NotImplementedError


class gRPCProxy(GenericProxy):
    """This class is meant to be instantiated and run by an gRPC server.

    This is the servicer class for the gRPC server. It implements `unary_unary`
    as the entry point for unary gRPC request and `unary_stream` as the entry
    point for streaming gRPC request. It also implements `stream_unary` and
    `stream_stream` for client streaming and bidirectional streaming RPCs.
    """

    def __init__(
        self,
        node_id: NodeId,
        node_ip_address: str,
        is_head: bool,
        proxy_router: "ProxyRouter",
        request_timeout_s: Optional[float] = None,
        access_log_context: Dict[str, Any] = None,
    ):
        super().__init__(
            node_id,
            node_ip_address,
            is_head,
            proxy_router,
            request_timeout_s=request_timeout_s,
            access_log_context=access_log_context,
        )
        # Dictionary to store active streaming sessions for client/bidi streaming.
        # Maps session_id -> (request_iterator, cancel_event)
        self._streaming_sessions: Dict[str, Tuple[Any, asyncio.Event]] = {}

    @property
    def protocol(self) -> RequestProtocol:
        return RequestProtocol.GRPC

    async def not_found_response(
        self, proxy_request: ProxyRequest
    ) -> ResponseGenerator:
        if not proxy_request.app_name:
            application_message = "Application metadata not set."
        else:
            application_message = f"Application '{proxy_request.app_name}' not found."
        not_found_message = (
            f"{application_message} Ping "
            "/ray.serve.RayServeAPIService/ListApplications for available applications."
        )

        yield ResponseStatus(
            code=grpc.StatusCode.NOT_FOUND,
            message=not_found_message,
            is_error=True,
        )

    async def routes_response(
        self, *, healthy: bool, message: str
    ) -> ResponseGenerator:
        yield ListApplicationsResponse(
            application_names=[
                endpoint.app_name for endpoint in self.proxy_router.endpoints
            ],
        ).SerializeToString()

        yield ResponseStatus(
            code=grpc.StatusCode.OK if healthy else grpc.StatusCode.UNAVAILABLE,
            message=message,
            is_error=not healthy,
        )

    async def health_response(self, *, healthy: bool, message) -> ResponseGenerator:
        yield HealthzResponse(message=message).SerializeToString()
        yield ResponseStatus(
            code=grpc.StatusCode.OK if healthy else grpc.StatusCode.UNAVAILABLE,
            message=message,
            is_error=not healthy,
        )

    def service_handler_factory(
        self, service_method: str, streaming_type: gRPCStreamingType
    ) -> Callable:
        async def unary_unary(
            request_proto: Any, context: grpc._cython.cygrpc._ServicerContext
        ) -> bytes:
            """Entry point of the gRPC proxy unary request.

            This method is called by the gRPC server when a unary request is received.
            It wraps the request in a ProxyRequest object and calls proxy_request.
            The return value is serialized user defined protobuf bytes.
            """
            proxy_request = gRPCProxyRequest(
                request_proto=request_proto,
                context=context,
                service_method=service_method,
                stream=False,
                streaming_type=gRPCStreamingType.UNARY_UNARY,
            )

            status = None
            response = None
            async for message in self.proxy_request(proxy_request=proxy_request):
                if isinstance(message, ResponseStatus):
                    status = message
                else:
                    response = message

            set_grpc_code_and_details(context, status)

            # When only ResponseStatus is yielded (not-found, errors), response stays
            # None. Returning None to gRPC causes serialization/type errors; return
            # empty bytes so the error status is sent cleanly.
            return response if response is not None else b""

        async def unary_stream(
            request_proto: Any, context: grpc._cython.cygrpc._ServicerContext
        ) -> Generator[bytes, None, None]:
            """Entry point of the gRPC proxy streaming request.

            This method is called by the gRPC server when a streaming request is
            received. It wraps the request in a ProxyRequest object and calls
            proxy_request. The return value is a generator of serialized user defined
            protobuf bytes.
            """
            proxy_request = gRPCProxyRequest(
                request_proto=request_proto,
                context=context,
                service_method=service_method,
                stream=True,
                streaming_type=gRPCStreamingType.UNARY_STREAM,
            )

            status = None
            async for message in self.proxy_request(proxy_request=proxy_request):
                if isinstance(message, ResponseStatus):
                    status = message
                else:
                    yield message

            set_grpc_code_and_details(context, status)

        async def stream_unary(
            request_iterator: Any, context: grpc._cython.cygrpc._ServicerContext
        ) -> bytes:
            """Entry point of the gRPC proxy client streaming request.

            This method is called by the gRPC server when a client streaming request
            is received. It wraps the request iterator and calls proxy_request.
            The return value is serialized user defined protobuf bytes.
            """
            # Create async iterator wrapper for the request stream
            async def async_request_iterator():
                async for request in request_iterator:
                    yield request

            proxy_request = gRPCProxyRequest(
                request_proto=None,
                context=context,
                service_method=service_method,
                stream=False,
                streaming_type=gRPCStreamingType.STREAM_UNARY,
                request_iterator=async_request_iterator(),
            )

            status = None
            response = None
            async for message in self.proxy_request(proxy_request=proxy_request):
                if isinstance(message, ResponseStatus):
                    status = message
                else:
                    response = message

            set_grpc_code_and_details(context, status)

            # When only ResponseStatus is yielded (not-found, errors), response stays
            # None. Returning None to gRPC causes serialization/type errors; return
            # empty bytes so the error status is sent cleanly.
            return response if response is not None else b""

        async def stream_stream(
            request_iterator: Any, context: grpc._cython.cygrpc._ServicerContext
        ) -> Generator[bytes, None, None]:
            """Entry point of the gRPC proxy bidirectional streaming request.

            This method is called by the gRPC server when a bidirectional streaming
            request is received. It wraps the request iterator and calls proxy_request.
            The return value is a generator of serialized user defined protobuf bytes.
            """
            # Create async iterator wrapper for the request stream
            async def async_request_iterator():
                async for request in request_iterator:
                    yield request

            proxy_request = gRPCProxyRequest(
                request_proto=None,
                context=context,
                service_method=service_method,
                stream=True,
                streaming_type=gRPCStreamingType.STREAM_STREAM,
                request_iterator=async_request_iterator(),
            )

            status = None
            async for message in self.proxy_request(proxy_request=proxy_request):
                if isinstance(message, ResponseStatus):
                    status = message
                else:
                    yield message

            set_grpc_code_and_details(context, status)

        handler_map = {
            gRPCStreamingType.UNARY_UNARY: unary_unary,
            gRPCStreamingType.UNARY_STREAM: unary_stream,
            gRPCStreamingType.STREAM_UNARY: stream_unary,
            gRPCStreamingType.STREAM_STREAM: stream_stream,
        }
        return handler_map[streaming_type]

    def setup_request_context_and_handle(
        self,
        app_name: str,
        handle: DeploymentHandle,
        route: str,
        proxy_request: ProxyRequest,
        internal_request_id: str,
    ) -> Tuple[DeploymentHandle, str]:
        """Setup request context and handle for the request.

        Unpack gRPC request metadata and extract info to set up request context and
        handle.
        """
        multiplexed_model_id = proxy_request.multiplexed_model_id
        session_id = proxy_request.session_id
        request_id = proxy_request.request_id
        if not request_id:
            request_id = generate_request_id()
            proxy_request.request_id = request_id

        handle = handle.options(
            stream=proxy_request.stream,
            multiplexed_model_id=multiplexed_model_id,
            session_id=session_id,
            method_name=proxy_request.method_name,
        )

        request_context_info = {
            "route": route,
            "request_id": request_id,
            "_internal_request_id": internal_request_id,
            "app_name": app_name,
            "multiplexed_model_id": multiplexed_model_id,
            "session_id": session_id,
            "grpc_context": proxy_request.ray_serve_grpc_context,
            "_client": proxy_request.client,
        }
        ray.serve.context._serve_request_context.set(
            ray.serve.context._RequestContext(**request_context_info)
        )
        proxy_request.send_request_id(request_id=request_id)
        return handle, request_id

    async def receive_grpc_messages(
        self, session_id: str
    ) -> Tuple[bool, Optional[Any], bool]:
        """Receive the next message from a gRPC streaming session.

        This method is called by replicas to receive messages from
        client/bidirectional streaming sessions.

        Args:
            session_id: The session ID of the streaming session.

        Returns:
            A tuple of (has_more, message, is_cancelled).
            - has_more: True if there are more messages, False if stream is done.
            - message: The protobuf message object, or None if stream is done.
            - is_cancelled: True if the stream was cancelled by client or error.

        Note:
            If the session ID is not found (e.g., after cleanup due to timeout,
            error, or completion), returns (False, None, True) for graceful
            termination instead of raising an exception.
        """
        if session_id not in self._streaming_sessions:
            # Session was already cleaned up - return graceful termination
            # This is consistent with the behavior when cancel_event.is_set()
            return (False, None, True)

        request_iterator, cancel_event = self._streaming_sessions[session_id]

        if cancel_event.is_set():
            return (False, None, True)

        try:
            message = await request_iterator.__anext__()
            # Return message directly - let Ray handle serialization
            return (True, message, False)
        except StopAsyncIteration:
            return (False, None, False)
        except Exception as e:
            logger.warning(
                f"Error receiving gRPC message for session {session_id}: {e}"
            )
            cancel_event.set()
            return (False, None, True)

    def _cleanup_streaming_session(self, session_id: str):
        """Clean up a streaming session."""
        session = self._streaming_sessions.pop(session_id, None)
        if session is not None:
            _, cancel_event = session
            cancel_event.set()

    def _setup_grpc_tracing(
        self,
        request_id: str,
        handle: DeploymentHandle,
        proxy_request: "gRPCProxyRequest",
    ):
        """Set up tracing for gRPC requests.

        This helper function sets up span attributes and span name for gRPC requests,
        used by both standard and streaming request paths.
        """
        self._setup_proxy_tracing(
            request_id=request_id,
            handle=handle,
            proxy_request=proxy_request,
        )
        set_span_name(
            f"proxy_{proxy_request.request_type}_request {handle.deployment_name} {proxy_request.method}"
        )

    def _finalize_grpc_tracing(
        self,
        proxy_request: "gRPCProxyRequest",
        status: Optional[ResponseStatus],
        exc: Optional[BaseException],
    ):
        """Finalize tracing for gRPC requests.

        This helper function sets RPC span attributes, exception tracking, and trace status
        in the finally block, used by both standard and streaming request paths.
        """
        if status is not None:
            set_rpc_span_attributes(
                system=proxy_request.request_type,
                method=proxy_request.method,
                status_code=status.code.name
                if isinstance(status.code, grpc.StatusCode)
                else grpc.StatusCode.UNKNOWN.name,
            )
        self._finalize_proxy_tracing(status=status, exc=exc)

    @tracing_decorator_factory(
        trace_name="proxy_grpc_request",
    )
    async def send_request_to_replica(
        self,
        request_id: str,
        internal_request_id: str,
        handle: DeploymentHandle,
        proxy_request: ProxyRequest,
        app_is_cross_language: bool = False,
    ) -> ResponseGenerator:
        # handle the streaming input that exists in client-streaming and bidi-streaming RPC types.
        if (
            isinstance(proxy_request, gRPCProxyRequest)
            and proxy_request.has_input_stream
        ):
            async for message in self._send_streaming_request_to_replica(
                request_id=request_id,
                internal_request_id=internal_request_id,
                handle=handle,
                proxy_request=proxy_request,
            ):
                yield message
            return

        # Standard server-unary/server-streaming path
        self._setup_grpc_tracing(request_id, handle, proxy_request)

        response_generator = ProxyResponseGenerator(
            handle.remote(proxy_request.serialized_replica_arg()),
            timeout_s=self.request_timeout_s,
        )
        status = None
        exc = None
        try:
            async for context, result in response_generator:
                context._set_on_grpc_context(proxy_request.context)
                yield result

            status = ResponseStatus(code=grpc.StatusCode.OK)
        except BaseException as e:
            status = get_grpc_response_status(e, self.request_timeout_s, request_id)
            exc = e
        finally:
            self._finalize_grpc_tracing(proxy_request, status, exc)

        # The status code should always be set.
        assert status is not None
        yield status

    async def _send_streaming_request_to_replica(
        self,
        request_id: str,
        internal_request_id: str,
        handle: DeploymentHandle,
        proxy_request: "gRPCProxyRequest",
    ) -> ResponseGenerator:
        """Handle sending a streaming request (client/bidi) to replica.

        For client streaming (stream_unary), we create a streaming session that
        the replica can use to receive messages from the client.

        For bidirectional streaming (stream_stream), we do the same but the
        response is also a stream.
        """
        # Set up tracing attributes for streaming
        self._setup_grpc_tracing(request_id, handle, proxy_request)

        # Create a streaming session
        session_id = internal_request_id
        cancel_event = asyncio.Event()
        self._streaming_sessions[session_id] = (
            proxy_request.request_iterator,
            cancel_event,
        )

        status = None
        exc = None
        try:
            # Get the proxy actor name for callback
            proxy_actor_name = ray.get_runtime_context().get_actor_name()

            # Create the streaming request
            streaming_request = gRPCStreamingRequest(
                session_id=session_id,
                proxy_actor_name=proxy_actor_name,
            )

            # Serialize the streaming request
            serialized_arg = pickle.dumps(streaming_request)

            response_generator = ProxyResponseGenerator(
                handle.remote(serialized_arg),
                timeout_s=self.request_timeout_s,
            )

            try:
                async for context, result in response_generator:
                    context._set_on_grpc_context(proxy_request.context)
                    yield result

                status = ResponseStatus(code=grpc.StatusCode.OK)
            except BaseException as e:
                status = get_grpc_response_status(e, self.request_timeout_s, request_id)
                exc = e
                # Yield the error status and return immediately to avoid falling through
                # to the outer except block, which would produce a duplicate ResponseStatus.
                yield status
                return

            # The status code should always be set.
            assert status is not None
            yield status
        except Exception as e:
            # Handle exceptions that occur before response_generator is created
            # (e.g., in get_actor_name, pickle.dumps, ProxyResponseGenerator)
            status = get_grpc_response_status(e, self.request_timeout_s, request_id)
            exc = e
            yield status
        finally:
            # Clean up the streaming session first to ensure no resource leaks
            self._cleanup_streaming_session(session_id)

            # Finalize tracing for streaming requests (best-effort)
            try:
                self._finalize_grpc_tracing(proxy_request, status, exc)
            except Exception as e:
                logger.warning(
                    f"Failed to finalize tracing for session {session_id}: {e}"
                )


class HTTPProxy(GenericProxy):
    """This class is meant to be instantiated and run by an ASGI HTTP server."""

    def __init__(
        self,
        node_id: NodeId,
        node_ip_address: str,
        is_head: bool,
        proxy_router: ProxyRouter,
        self_actor_name: str,
        request_timeout_s: Optional[float] = None,
        access_log_context: Dict[str, Any] = None,
    ):
        super().__init__(
            node_id,
            node_ip_address,
            is_head,
            proxy_router,
            request_timeout_s=request_timeout_s,
            access_log_context=access_log_context,
        )
        self.self_actor_name = self_actor_name
        self.asgi_receive_queues: Dict[str, MessageQueue] = dict()

    @property
    def protocol(self) -> RequestProtocol:
        return RequestProtocol.HTTP

    async def not_found_response(
        self, proxy_request: ProxyRequest
    ) -> ResponseGenerator:
        status_code = 404
        for message in convert_object_to_asgi_messages(
            f"Path '{proxy_request.path}' not found. "
            "Ping http://.../-/routes for available routes.",
            status_code=status_code,
        ):
            yield message

        yield ResponseStatus(code=status_code, is_error=True)

    async def routes_response(
        self, *, healthy: bool, message: str
    ) -> ResponseGenerator:
        status_code = 200 if healthy else 503
        if healthy:
            response = dict()
            for endpoint, info in self.proxy_router.endpoints.items():
                # For 2.x deployments, return {route -> app name}
                if endpoint.app_name:
                    response[info.route] = endpoint.app_name
                # Keep compatibility with 1.x deployments.
                else:
                    response[info.route] = endpoint.name
        else:
            response = message

        for asgi_message in convert_object_to_asgi_messages(
            response,
            status_code=status_code,
        ):
            yield asgi_message

        yield ResponseStatus(
            code=status_code,
            message=message,
            is_error=not healthy,
        )

    async def health_response(
        self, *, healthy: bool, message: str = ""
    ) -> ResponseGenerator:
        status_code = 200 if healthy else 503
        for asgi_message in convert_object_to_asgi_messages(
            message,
            status_code=status_code,
        ):
            yield asgi_message

        yield ResponseStatus(
            code=status_code,
            is_error=not healthy,
            message=message,
        )

    async def receive_asgi_messages(
        self, request_metadata: RequestMetadata
    ) -> ResponseGenerator:
        queue = self.asgi_receive_queues.get(request_metadata.internal_request_id, None)
        if queue is None:
            raise KeyError(f"Request ID {request_metadata.request_id} not found.")

        await queue.wait_for_message()
        return queue.get_messages_nowait()

    async def __call__(self, scope, receive, send):
        """Implements the ASGI protocol.

        See details at:
            https://asgi.readthedocs.io/en/latest/specs/index.html.
        """
        proxy_request = ASGIProxyRequest(scope=scope, receive=receive, send=send)
        async for message in self.proxy_request(proxy_request):
            if not isinstance(message, ResponseStatus):
                await send(message)

    async def proxy_asgi_receive(
        self, receive: Receive, queue: MessageQueue
    ) -> Optional[int]:
        """Proxies the `receive` interface, placing its messages into the queue.

        Once a disconnect message is received, the call exits and `receive` is no longer
        called.

        For HTTP messages, `None` is always returned.
        For websocket messages, the disconnect code is returned if a disconnect code is
        received.
        """
        try:
            while True:
                msg = await receive()
                await queue(msg)

                if msg["type"] == "http.disconnect":
                    return None

                if msg["type"] == "websocket.disconnect":
                    return msg["code"]
        finally:
            # Close the queue so any subsequent calls to fetch messages return
            # immediately: https://github.com/ray-project/ray/issues/38368.
            queue.close()

    def setup_request_context_and_handle(
        self,
        app_name: str,
        handle: DeploymentHandle,
        route: str,
        proxy_request: ProxyRequest,
        internal_request_id: str,
    ) -> Tuple[DeploymentHandle, str]:
        """Setup request context and handle for the request.

        Unpack HTTP request headers and extract info to set up request context and
        handle.
        """
        request_context_info = {
            "route": route,
            "app_name": app_name,
            "_internal_request_id": internal_request_id,
            "is_http_request": True,
            "_client": format_client_address(proxy_request.client),
        }
        for key, value in proxy_request.headers:
            # Normalize the header key: lowercase and replace hyphens with
            # underscores so that both "serve_multiplexed_model_id" and
            # "serve-multiplexed-model-id" (the form produced by proxies such
            # as nginx / AWS API Gateway that convert underscores to hyphens)
            # are recognised correctly.
            normalized_key = key.decode().lower().replace("-", "_")
            if normalized_key == SERVE_MULTIPLEXED_MODEL_ID:
                multiplexed_model_id = value.decode()
                handle = handle.options(multiplexed_model_id=multiplexed_model_id)
                request_context_info["multiplexed_model_id"] = multiplexed_model_id
            elif _matches_session_id_header(key.decode()):
                session_id = value.decode()
                handle = handle.options(session_id=session_id)
                request_context_info["session_id"] = session_id
            if key.decode() == SERVE_HTTP_REQUEST_ID_HEADER:
                request_context_info["request_id"] = value.decode()
        ray.serve.context._serve_request_context.set(
            ray.serve.context._RequestContext(**request_context_info)
        )
        return handle, request_context_info["request_id"]

    def _setup_http_tracing(
        self,
        request_id: str,
        handle: DeploymentHandle,
        proxy_request: ProxyRequest,
    ):
        """Set up tracing for HTTP requests.

        This helper function sets up span attributes and span name for HTTP requests.
        """
        self._setup_proxy_tracing(
            request_id=request_id,
            handle=handle,
            proxy_request=proxy_request,
            additional_attributes={
                "request_method": proxy_request.method,
                "request_route_path": proxy_request.route_path,
            },
        )
        set_span_name(
            f"proxy_{proxy_request.request_type}_request {handle.deployment_name} {proxy_request.method} {proxy_request.route_path}"
        )

    def _finalize_http_tracing(
        self,
        proxy_request: ProxyRequest,
        status_code: str,
        is_error: bool,
        exc: Optional[BaseException],
    ):
        """Finalize tracing for HTTP requests.

        This helper function sets HTTP span attributes, exception tracking, and trace status
        in finally block.
        """
        set_http_span_attributes(
            method=proxy_request.method,
            status_code=int(status_code),
            route=proxy_request.route_path,
        )
        self._finalize_proxy_tracing(
            status=ResponseStatus(code=status_code, is_error=is_error),
            exc=exc,
        )

    async def _format_handle_arg_for_java(
        self,
        proxy_request: ProxyRequest,
    ) -> bytes:
        """Convert an HTTP request to the Java-accepted format (single byte string)."""
        query_string = proxy_request.scope.get("query_string")
        http_body_bytes = await receive_http_body(
            proxy_request.scope, proxy_request.receive, proxy_request.send
        )
        if query_string:
            arg = query_string.decode().split("=", 1)[1]
        else:
            arg = http_body_bytes.decode()

        return arg

    @tracing_decorator_factory(
        trace_name="proxy_http_request",
    )
    async def send_request_to_replica(
        self,
        request_id: str,
        internal_request_id: str,
        handle: DeploymentHandle,
        proxy_request: ProxyRequest,
        app_is_cross_language: bool = False,
    ) -> ResponseGenerator:
        """Send the request to the replica and yield its response messages.

        The yielded values will be ASGI messages until the final one, which will be
        the status code.
        """
        if is_span_recording():
            self._setup_http_tracing(request_id, handle, proxy_request)

        if app_is_cross_language:
            handle_arg_bytes = await self._format_handle_arg_for_java(proxy_request)
            # Response is returned as raw bytes, convert it to ASGI messages.
            result_callback = convert_object_to_asgi_messages
        else:
            handle_arg_bytes = proxy_request.serialized_replica_arg(
                proxy_actor_name=self.self_actor_name,
            )
            # Messages are returned as pickled dictionaries.
            result_callback = pickle.loads

        # Proxy the receive interface by placing the received messages on a queue.
        # The downstream replica must call back into `receive_asgi_messages` on this
        # actor to receive the messages.
        receive_queue = MessageQueue()
        self.asgi_receive_queues[internal_request_id] = receive_queue
        proxy_asgi_receive_task = get_or_create_event_loop().create_task(
            self.proxy_asgi_receive(proxy_request.receive, receive_queue)
        )

        # Per-request headers override the global HTTPOptions timeout and disconnect
        # policy, enabling HAProxy (or other callers) to pass per-request hints.
        request_headers = dict(proxy_request.headers)
        request_timeout_s = parse_request_timeout_header(
            request_headers, self.request_timeout_s
        )
        request_disconnect_disabled = parse_disconnect_disabled_header(request_headers)

        response_generator = ProxyResponseGenerator(
            handle.remote(handle_arg_bytes),
            timeout_s=request_timeout_s,
            disconnected_task=(
                None if request_disconnect_disabled else proxy_asgi_receive_task
            ),
            result_callback=result_callback,
        )

        status: Optional[ResponseStatus] = None
        response_started = False
        expecting_trailers = False
        exc = None
        status_code = None
        try:
            async for asgi_message_batch in response_generator:
                # See the ASGI spec for message details:
                # https://asgi.readthedocs.io/en/latest/specs/www.html.
                for asgi_message in asgi_message_batch:
                    if asgi_message["type"] == "http.response.start":
                        # HTTP responses begin with exactly one
                        # "http.response.start" message containing the "status"
                        # field. Other response types (e.g., WebSockets) may not.
                        status_code = str(asgi_message["status"])
                        status = ResponseStatus(
                            code=status_code,
                            is_error=status_code.startswith(("4", "5")),
                        )
                        expecting_trailers = asgi_message.get("trailers", False)
                    elif asgi_message["type"] == "websocket.accept":
                        # Websocket code explicitly handles client disconnects,
                        # so let the ASGI disconnect message propagate instead of
                        # cancelling the handler.
                        response_generator.stop_checking_for_disconnect()
                    elif (
                        asgi_message["type"] == "http.response.body"
                        and not asgi_message.get("more_body", False)
                        and not expecting_trailers
                    ):
                        # If the body is completed and we aren't expecting trailers, the
                        # response is done so we should stop listening for disconnects.
                        response_generator.stop_checking_for_disconnect()
                    elif asgi_message["type"] == "http.response.trailers":
                        # If we are expecting trailers, the response is only done when
                        # the trailers message has been sent.
                        if not asgi_message.get("more_trailers", False):
                            response_generator.stop_checking_for_disconnect()
                    elif asgi_message["type"] in [
                        "websocket.close",
                        "websocket.disconnect",
                    ]:
                        status_code = str(asgi_message["code"])
                        status = ResponseStatus(
                            code=status_code,
                            # All status codes are considered errors aside from:
                            # 1000 (CLOSE_NORMAL), 1001 (CLOSE_GOING_AWAY).
                            is_error=status_code not in ["1000", "1001"],
                        )
                        response_generator.stop_checking_for_disconnect()

                    yield asgi_message
                    response_started = True
        except BaseException as e:
            error_status = get_http_response_status(e, request_timeout_s, request_id)
            if status is None:
                status = error_status
            for asgi_message in send_http_response_on_exception(
                error_status, response_started
            ):
                yield asgi_message
            exc = e

        finally:
            # For websocket connection, queue receive task is done when receiving
            # disconnect message from client.
            receive_client_disconnect_msg = False
            if not proxy_asgi_receive_task.done():
                proxy_asgi_receive_task.cancel()
            else:
                receive_client_disconnect_msg = True

            # If the server disconnects, status_code can be set above from the
            # disconnect message.
            # If client disconnects, the disconnect code comes from
            # a client message via the receive interface.
            if status is None and proxy_request.request_type == "websocket":
                if receive_client_disconnect_msg:
                    # The disconnect message is sent from the client.
                    status_code = str(proxy_asgi_receive_task.result())
                    status = ResponseStatus(
                        code=status_code,
                        is_error=True,
                    )
                else:
                    status_code = "1000"
                    # The server disconnect without sending a disconnect message
                    # (otherwise the `status` would be set).
                    status = ResponseStatus(
                        code="1000",
                        is_error=True,
                    )
            else:
                status_code = status.code

            del self.asgi_receive_queues[internal_request_id]

            if is_span_recording():
                self._finalize_http_tracing(
                    proxy_request, status_code, status.is_error, exc
                )

        # The status code should always be set.
        assert status is not None
        yield status


def apply_per_node_port_overrides(
    http_options: HTTPOptions, grpc_options: gRPCOptions, is_head: bool
) -> None:
    """Override this proxy's HTTP and gRPC bind ports from the per-node env knobs.

    Worker proxies bind RAY_SERVE_WORKER_PROXY_HTTP_PORT and
    RAY_SERVE_WORKER_PROXY_GRPC_PORT when set. The head node is exempt so its
    configured ports and the fallback proxy stay intact.
    """
    if is_head:
        return
    if RAY_SERVE_WORKER_PROXY_HTTP_PORT is not None:
        http_options.port = RAY_SERVE_WORKER_PROXY_HTTP_PORT
    if RAY_SERVE_WORKER_PROXY_GRPC_PORT is not None:
        grpc_options.port = RAY_SERVE_WORKER_PROXY_GRPC_PORT


class ProxyActorInterface(ABC):
    """Abstract interface for proxy actors in Ray Serve.

    This interface defines the contract that all proxy actor implementations must follow,
    allowing for different proxy backends (Ray HTTP/gRPC proxies, HAProxy, etc.).
    """

    def __init__(
        self,
        *,
        node_id: NodeId,
        node_ip_address: str,
        logging_config: LoggingConfig,
        log_buffer_size: int = RAY_SERVE_REQUEST_PATH_LOG_BUFFER_SIZE,
    ):
        """Initialize the proxy actor.

        Args:
            node_id: ID of the node this proxy is running on
            node_ip_address: IP address of the node
            logging_config: Logging configuration
            log_buffer_size: Size of the log buffer
        """
        self._node_id = node_id
        self._node_ip_address = node_ip_address
        self._logging_config = logging_config
        self._log_buffer_size = log_buffer_size

        self._update_logging_config(logging_config)

    @abstractmethod
    async def ready(self) -> str:
        """Blocks until the proxy is ready to serve requests.

        Returns:
            JSON-serialized metadata containing proxy information (worker ID, log file path, etc.)
        """
        pass

    @abstractmethod
    async def serving(self, wait_for_applications_running: bool = True) -> None:
        """Wait for the proxy to be ready to serve requests.

        Args:
            wait_for_applications_running: Whether to wait for the applications to be running

        Returns:
            None
        """
        pass

    @abstractmethod
    async def update_draining(
        self, draining: bool, _after: Optional[Any] = None
    ) -> None:
        """Update the draining status of the proxy.

        Args:
            draining: Whether the proxy should be draining
            _after: Optional ObjectRef for scheduling dependency
        """
        pass

    @abstractmethod
    async def is_drained(self, _after: Optional[Any] = None) -> bool:
        """Check whether the proxy is drained.

        Args:
            _after: Optional ObjectRef for scheduling dependency

        Returns:
            True if the proxy is drained, False otherwise
        """
        pass

    @abstractmethod
    async def check_health(self) -> bool:
        """Check the health of the proxy.

        Returns:
            True if the proxy is healthy, False otherwise
        """
        pass

    @abstractmethod
    def pong(self) -> str:
        """Respond to ping from replicas.

        Returns:
            A response string
        """
        pass

    @abstractmethod
    async def receive_asgi_messages(self, request_metadata: RequestMetadata) -> bytes:
        """Handle ASGI messages for HTTP requests.

        Args:
            request_metadata: Metadata about the request

        Returns:
            Serialized ASGI messages
        """
        pass

    @abstractmethod
    async def receive_grpc_messages(
        self, session_id: str
    ) -> Tuple[bool, Optional[Any], bool]:
        """Get the next gRPC message for a streaming session.

        Args:
            session_id: The session ID of the streaming session

        Returns:
            Tuple of (has_more, message, is_cancelled)
        """
        pass

    # Testing and debugging methods
    @abstractmethod
    def _get_http_options(self) -> HTTPOptions:
        """Get HTTP options used by the proxy."""
        pass

    @abstractmethod
    def _get_logging_config(self) -> Optional[str]:
        """Get the file path for the logger (for testing purposes)."""
        pass

    @abstractmethod
    def _dump_ingress_replicas_for_testing(self, route: str) -> Set:
        """Get replicas for a route (for testing)."""
        pass

    @abstractmethod
    def shutdown(self) -> None:
        """Shuts down proxy."""
        pass

    def _update_logging_config(self, logging_config: LoggingConfig):
        configure_component_logger(
            component_name="proxy",
            component_id=self._node_ip_address,
            logging_config=logging_config,
            buffer_size=self._log_buffer_size,
        )


@ray.remote(num_cpus=0)
class ProxyActor(ProxyActorInterface):
    def __init__(
        self,
        http_options: HTTPOptions,
        grpc_options: gRPCOptions,
        *,
        node_id: NodeId,
        node_ip_address: str,
        logging_config: LoggingConfig,
        long_poll_client: Optional[LongPollClient] = None,
    ):  # noqa: F821
        super().__init__(
            node_id=node_id,
            node_ip_address=node_ip_address,
            logging_config=logging_config,
        )

        is_head = self._node_id == get_head_node_id()
        apply_per_node_port_overrides(http_options, grpc_options, is_head)

        self._grpc_options = grpc_options
        self._http_options = configure_http_middlewares(http_options)
        grpc_enabled = is_grpc_enabled(self._grpc_options)
        event_loop = get_or_create_event_loop()
        self.long_poll_client = long_poll_client or LongPollClient(
            ray.get_actor(SERVE_CONTROLLER_NAME, namespace=SERVE_NAMESPACE),
            {
                LongPollNamespace.GLOBAL_LOGGING_CONFIG: self._update_logging_config,
                LongPollNamespace.ROUTE_TABLE: self._update_routes_in_proxies,
            },
            call_in_event_loop=event_loop,
            client_id=f"{type(self).__name__}:{ray.get_runtime_context().get_actor_id()}",
        )

        try:
            is_tracing_setup_successful = setup_tracing(
                component_name="proxy", component_id=node_ip_address
            )
            if is_tracing_setup_successful:
                logger.info("Successfully set up tracing for proxy")
        except Exception as e:
            logger.warning(
                f"Failed to set up tracing: {e}. "
                "The proxy will continue running, but traces will not be exported."
            )

        startup_msg = f"Proxy starting on node {self._node_id} (HTTP port: {self._http_options.port}"
        if grpc_enabled:
            startup_msg += f", gRPC port: {self._grpc_options.port})."
        else:
            startup_msg += ")."
        logger.info(startup_msg)
        logger.debug(
            f"Configure Proxy actor {ray.get_runtime_context().get_actor_id()} "
            f"logger with logging config: {logging_config}"
        )

        configure_component_memory_profiler(
            component_name="proxy", component_id=node_ip_address
        )
        if logging_config.encoding == EncodingType.JSON:
            # Create logging context for access logs as a performance optimization.
            # While logging_utils can automatically add Ray core and Serve access log context,
            # we pre-compute it here since context evaluation is expensive and this context
            # will be reused for multiple access log entries.
            ray_core_logging_context = CoreContextFilter.get_ray_core_logging_context()
            # remove task level log keys from ray core logging context, it would be nice
            # to have task level log keys here but we are letting those go in favor of
            # performance optimization. Also we cannot include task level log keys here because
            # they would referance the current task (__init__) and not the task that is logging.
            for key in CoreContextFilter.TASK_LEVEL_LOG_KEYS:
                ray_core_logging_context.pop(key, None)
            access_log_context = {
                **ray_core_logging_context,
                SERVE_LOG_COMPONENT: "proxy",
                SERVE_LOG_COMPONENT_ID: self._node_ip_address,
                "log_to_stderr": False,
                "skip_context_filter": True,
                "serve_access_log": True,
            }
        else:
            access_log_context = {
                "log_to_stderr": False,
                "skip_context_filter": True,
                "serve_access_log": True,
            }

        self.proxy_router = ProxyRouter(get_proxy_handle)
        self.http_proxy = HTTPProxy(
            node_id=self._node_id,
            node_ip_address=self._node_ip_address,
            is_head=is_head,
            self_actor_name=ray.get_runtime_context().get_actor_name(),
            proxy_router=self.proxy_router,
            request_timeout_s=self._http_options.request_timeout_s,
            access_log_context=access_log_context,
        )
        self.grpc_proxy = (
            gRPCProxy(
                node_id=self._node_id,
                node_ip_address=self._node_ip_address,
                is_head=is_head,
                proxy_router=self.proxy_router,
                request_timeout_s=self._grpc_options.request_timeout_s,
                access_log_context=access_log_context,
            )
            if grpc_enabled
            else None
        )
        if self.grpc_proxy:
            get_or_create_event_loop().set_exception_handler(
                asyncio_grpc_exception_handler
            )

        # Start a task to initialize the HTTP server.
        # The result of this task is checked in the `ready` method.
        self._start_http_server_task = event_loop.create_task(
            start_asgi_http_server(
                self.http_proxy,
                self._http_options,
                event_loop=event_loop,
                enable_so_reuseport=SOCKET_REUSE_PORT_ENABLED,
            )
        )
        # A task that runs the HTTP server until it exits (currently runs forever).
        # Populated with the result of self._start_http_server_task.
        self._running_http_server_task: Optional[asyncio.Task] = None

        # Start a task to initialize the gRPC server.
        # The result of this task is checked in the `ready` method.
        self._start_grpc_server_task: Optional[asyncio.Task] = None
        if grpc_enabled:
            self._start_grpc_server_task = event_loop.create_task(
                start_grpc_server(
                    self.grpc_proxy.service_handler_factory,
                    self._grpc_options,
                    event_loop=event_loop,
                    enable_so_reuseport=SOCKET_REUSE_PORT_ENABLED,
                ),
            )
        # A task that runs the gRPC server until it exits (currently runs forever).
        # Populated with the result of self._start_grpc_server_task.
        self._running_grpc_server_task: Optional[asyncio.Task] = None

        _configure_gc_options()

        # Start event loop monitoring for the proxy's main event loop.
        self._event_loop_monitor = EventLoopMonitor(
            component=EventLoopMonitor.COMPONENT_PROXY,
            loop_type=EventLoopMonitor.LOOP_TYPE_MAIN,
            actor_id=ray.get_runtime_context().get_actor_id(),
        )
        self._event_loop_monitor.start(event_loop)

    def _update_routes_in_proxies(self, endpoints: Dict[DeploymentID, EndpointInfo]):
        self.proxy_router.update_routes(endpoints)

    def _get_logging_config(self) -> Tuple:
        """Get the logging configuration (for testing purposes)."""
        log_file_path = None
        for handler in logger.handlers:
            if isinstance(handler, logging.handlers.MemoryHandler):
                log_file_path = handler.target.baseFilename
        return log_file_path

    def _dump_ingress_replicas_for_testing(self, route: str) -> Set[ReplicaID]:
        _, handle, _ = self.http_proxy.proxy_router.match_route(route)
        return handle._router._asyncio_router._request_router._replica_id_set

    def _dump_ingress_cache_for_testing(self, route: str) -> Set[ReplicaID]:
        """Get replica IDs that have entries in the queue length cache (for testing)."""
        _, handle, _ = self.http_proxy.proxy_router.match_route(route)
        request_router = handle._router._asyncio_router._request_router
        cache = request_router.replica_queue_len_cache
        return {
            replica_id
            for replica_id in request_router._replica_id_set
            if cache.get(replica_id) is not None
        }

    async def ready(self) -> str:
        """Blocks until the proxy HTTP (and optionally gRPC) servers are running.

        Returns JSON-serialized metadata containing the proxy's worker ID and log
        file path.

        Raises any exceptions that occur setting up the HTTP or gRPC server.
        """
        try:
            # The proxy has its own draining; the returned server object is
            # unused.
            self._running_http_server_task, _ = await self._start_http_server_task
        except Exception as e:
            logger.exception("Failed to start proxy HTTP server.")
            raise e from None

        try:
            if self._start_grpc_server_task is not None:
                self._running_grpc_server_task, _ = await self._start_grpc_server_task
        except Exception as e:
            logger.exception("Failed to start proxy gRPC server.")
            raise e from None

        # Return proxy metadata used by the controller.
        # NOTE(zcin): We need to convert the metadata to a json string because
        # of cross-language scenarios. Java can't deserialize a Python tuple.
        return json.dumps(
            [
                ray.get_runtime_context().get_worker_id(),
                get_component_logger_file_path(),
            ]
        )

    async def serving(self, wait_for_applications_running: bool = True) -> None:
        """Wait for the proxy to be ready to serve requests."""
        return

    def shutdown(self) -> None:
        return

    async def update_draining(self, draining: bool, _after: Optional[Any] = None):
        """Update the draining status of the HTTP and gRPC proxies.

        Unused `_after` argument is for scheduling: passing an ObjectRef
        allows delaying this call until after the `_after` call has returned.
        """

        self.http_proxy.update_draining(draining)
        if self.grpc_proxy:
            self.grpc_proxy.update_draining(draining)

    async def is_drained(self, _after: Optional[Any] = None):
        """Check whether both HTTP and gRPC proxies are drained or not.

        Unused `_after` argument is for scheduling: passing an ObjectRef
        allows delaying this call until after the `_after` call has returned.
        """

        return self.http_proxy.is_drained() and (
            self.grpc_proxy is None or self.grpc_proxy.is_drained()
        )

    async def check_health(self) -> bool:
        """No-op method to check on the health of the HTTP Proxy.

        Make sure the async event loop is not blocked.
        """
        logger.debug("Received health check.", extra={"log_to_stderr": False})
        return True

    def pong(self):
        """Called by the replica to initialize its handle to the proxy."""
        pass

    async def receive_asgi_messages(self, request_metadata: RequestMetadata) -> bytes:
        """Get ASGI messages for the provided `request_metadata`.

        After the proxy has stopped receiving messages for this `request_metadata`,
        this will always return immediately.

        Raises `KeyError` if this request ID is not found. This will happen when the
        request is no longer being handled (e.g., the user disconnects).
        """
        return pickle.dumps(
            await self.http_proxy.receive_asgi_messages(request_metadata)
        )

    async def receive_grpc_messages(
        self, session_id: str
    ) -> Tuple[bool, Optional[Any], bool]:
        """Get the next gRPC message for a streaming session.

        This method is called by replicas to receive messages from
        client/bidirectional streaming sessions.

        Args:
            session_id: The session ID of the streaming session.

        Returns:
            Tuple of (has_more: bool, message: Optional[Any], is_cancelled: bool).
            - has_more: True if there are more messages, False if stream is done.
            - message: The protobuf message object, or None if stream is done.
            - is_cancelled: True if the stream was cancelled by client or error.
              Also returned when the session ID is not found (e.g., after
              cleanup due to timeout, error, or completion).
        """
        if self.grpc_proxy is None:
            raise RuntimeError("gRPC proxy is not enabled.")

        # Return tuple directly - Ray handles serialization
        return await self.grpc_proxy.receive_grpc_messages(session_id)

    def _get_http_options(self) -> HTTPOptions:
        """Internal method to get HTTP options used by the proxy."""
        return self._http_options


def _configure_gc_options():
    if not RAY_SERVE_ENABLE_PROXY_GC_OPTIMIZATIONS:
        return

    # Collect any objects that exist already and exclude them from future GC.
    gc.collect(2)
    gc.freeze()

    # Tune the GC threshold to run less frequently (default is 700).
    gc.set_threshold(RAY_SERVE_PROXY_GC_THRESHOLD)