feat(iris): add actor proxy service for external access to cluster actors (#4126)

Adds an actor proxy route on the controller HTTP server that forwards
ActorService RPCs to actors on worker VMs. External clients use
ProxyResolver to route calls through the controller instead of
connecting directly to internal IPs.

Closes #4109

Generated with [Claude Code](https://claude.ai/code)

---------

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Rafal Wojdyla <ravwojdyla@users.noreply.github.com>

Author: 2 people

lib/iris/src/iris/actor/__init__.py

@@ -10,21 +10,25 @@
 from iris.actor.client import ActorClient
 from iris.actor.pool import ActorPool, BroadcastFuture, CallResult
 from iris.actor.resolver import (
+    ACTOR_ENDPOINT_HEADER,
     FixedResolver,
+    ProxyResolver,
     ResolvedEndpoint,
     ResolveResult,
     Resolver,
 )
 from iris.actor.server import ActorId, ActorServer
 
 __all__ = [
+    "ACTOR_ENDPOINT_HEADER",
     "ActorClient",
     "ActorId",
     "ActorPool",
     "ActorServer",
     "BroadcastFuture",
     "CallResult",
     "FixedResolver",
+    "ProxyResolver",
     "ResolveResult",
     "ResolvedEndpoint",
     "Resolver",

lib/iris/src/iris/actor/client.py

@@ -80,6 +80,7 @@ def __init__(
         self._backoff = backoff
 
         self._rpc_client: ActorServiceClientSync | None = None
+        self._rpc_headers: dict[str, str] = {}
 
     def rpc_client(self) -> ActorServiceClientSync:
         """Resolve actor name to an RPC client (single attempt).
@@ -111,17 +112,19 @@ def rpc_client(self) -> ActorServiceClientSync:
             self._name,
             len(result.endpoints),
         )
-        logger.info("First endpoint: %s", result.first())
-        url = result.first().url
+        endpoint = result.first()
+        logger.info("First endpoint: %s", endpoint)
+        self._rpc_headers = dict(endpoint.metadata)
         self._rpc_client = ActorServiceClientSync(
-            address=url,
+            address=endpoint.url,
             timeout_ms=None if self._call_timeout is None else int(self._call_timeout * 1000),
             accept_compression=[],
         )
         return self._rpc_client
 
     def _clear_connection(self, _exc: Exception) -> None:
         self._rpc_client = None
+        self._rpc_headers = {}
 
     def start_operation(self, method_name: str, *args: Any, **kwargs: Any) -> str:
         """Start a long-running operation. Returns the operation ID."""
@@ -134,7 +137,7 @@ def start_operation(self, method_name: str, *args: Any, **kwargs: Any) -> str:
 
         def do_call():
             client = self.rpc_client()
-            return client.start_operation(call)
+            return client.start_operation(call, headers=self._rpc_headers)
 
         op = call_with_retry(
             f"{self._name}.start_operation({method_name})",
@@ -150,7 +153,7 @@ def poll_operation_status(self, operation_id: str) -> actor_pb2.Operation:
         req = actor_pb2.OperationId(operation_id=operation_id)
 
         def do_call():
-            return self.rpc_client().get_operation(req)
+            return self.rpc_client().get_operation(req, headers=self._rpc_headers)
 
         return call_with_retry(
             f"{self._name}.poll_operation_status({operation_id[:8]})",
@@ -179,7 +182,7 @@ def cancel_operation(self, operation_id: str) -> actor_pb2.Operation:
         req = actor_pb2.OperationId(operation_id=operation_id)
 
         def do_call():
-            return self.rpc_client().cancel_operation(req)
+            return self.rpc_client().cancel_operation(req, headers=self._rpc_headers)
 
         return call_with_retry(
             f"{self._name}.cancel_operation({operation_id[:8]})",
@@ -208,7 +211,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any:
 
         def do_call():
             client = self._client.rpc_client()
-            resp = client.call(call)
+            resp = client.call(call, headers=self._client._rpc_headers)
             return unwrap_actor_response(resp)
 
         return call_with_retry(

lib/iris/src/iris/actor/resolver.py

@@ -6,6 +6,10 @@
 from dataclasses import dataclass, field
 from typing import Protocol
 
+# Header used by ActorProxy to route requests to the correct actor endpoint.
+# Shared constant between ProxyResolver (client-side) and ActorProxy (server-side).
+ACTOR_ENDPOINT_HEADER = "x-iris-actor-endpoint"
+
 
 @dataclass
 class ResolvedEndpoint:
@@ -72,3 +76,33 @@ def resolve(self, name: str) -> ResolveResult:
         urls = self._endpoints.get(name, [])
         endpoints = [ResolvedEndpoint(url=url, actor_id=f"fixed-{name}-{i}") for i, url in enumerate(urls)]
         return ResolveResult(name=name, endpoints=endpoints)
+
+
+class ProxyResolver:
+    """Resolver that routes actor calls through the controller's actor proxy.
+
+    Instead of resolving to the actor's direct address, returns the controller
+    URL so all RPCs go through the proxy. The proxy uses the
+    ``X-Iris-Actor-Endpoint`` header to resolve the actual actor endpoint.
+
+    Args:
+        controller_url: Controller URL (e.g., ``http://localhost:8080``)
+        namespace: Namespace prefix for endpoint resolution
+    """
+
+    def __init__(self, controller_url: str, namespace: str):
+        self._controller_url = controller_url.rstrip("/")
+        self._namespace = namespace
+
+    def resolve(self, name: str) -> ResolveResult:
+        endpoint_name = f"{self._namespace}/{name}"
+        return ResolveResult(
+            name=name,
+            endpoints=[
+                ResolvedEndpoint(
+                    url=self._controller_url,
+                    actor_id=f"proxy-{endpoint_name}",
+                    metadata={ACTOR_ENDPOINT_HEADER: endpoint_name},
+                )
+            ],
+        )

lib/iris/src/iris/cluster/controller/actor_proxy.py

@@ -0,0 +1,108 @@
+# Copyright The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Actor proxy for forwarding ActorService RPCs to actors within the cluster.
+
+External clients send actor calls to the controller; the proxy resolves the
+target endpoint from the controller's DB and forwards the raw request to the
+actor server on the worker VM. All ActorService methods are proxied
+transparently (raw byte forwarding, no deserialization).
+
+Route pattern::
+
+    POST /iris.actor.ActorService/{method}
+    X-Iris-Actor-Endpoint: namespace/actor-name
+"""
+
+import logging
+
+import httpx
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+from iris.cluster.controller.db import ControllerDB, EndpointQuery, endpoint_query_predicate
+
+logger = logging.getLogger(__name__)
+
+# Header used by ProxyResolver to tell the proxy which endpoint to forward to.
+# Duplicated from iris.actor.resolver.ACTOR_ENDPOINT_HEADER to avoid a
+# cluster → actor import dependency.
+ACTOR_ENDPOINT_HEADER = "x-iris-actor-endpoint"
+
+PROXY_ROUTE = "/iris.actor.ActorService/{method}"
+PROXY_TIMEOUT_SECONDS = 60.0
+
+# Headers that should not be forwarded to upstream (hop-by-hop or routing-specific).
+_HOP_BY_HOP_HEADERS = frozenset(
+    {
+        "host",
+        "transfer-encoding",
+        "connection",
+        "keep-alive",
+        "upgrade",
+        ACTOR_ENDPOINT_HEADER,
+    }
+)
+
+
+class ActorProxy:
+    """Forwards ActorService RPCs to actors resolved from the endpoint registry."""
+
+    def __init__(self, db: ControllerDB):
+        self._db = db
+        self._client = httpx.AsyncClient(timeout=PROXY_TIMEOUT_SECONDS)
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    async def handle(self, request: Request) -> Response:
+        """Proxy an ActorService RPC to the resolved actor endpoint."""
+        method = request.path_params["method"]
+        endpoint_name = request.headers.get(ACTOR_ENDPOINT_HEADER)
+        if not endpoint_name:
+            return JSONResponse(
+                {"error": f"Missing {ACTOR_ENDPOINT_HEADER} header"},
+                status_code=400,
+            )
+
+        address = self._resolve_endpoint(endpoint_name)
+        if address is None:
+            return JSONResponse(
+                {"error": f"No endpoint found for '{endpoint_name}'"},
+                status_code=404,
+            )
+
+        upstream_url = f"http://{address}/iris.actor.ActorService/{method}"
+        body = await request.body()
+        forward_headers = {k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP_HEADERS}
+
+        try:
+            upstream_resp = await self._client.post(
+                upstream_url,
+                content=body,
+                headers=forward_headers,
+            )
+        except httpx.HTTPError as exc:
+            logger.warning("Proxy upstream error for %s: %s", endpoint_name, exc)
+            return JSONResponse(
+                {"error": f"Upstream error: {exc}"},
+                status_code=502,
+            )
+
+        return Response(
+            content=upstream_resp.content,
+            status_code=upstream_resp.status_code,
+            media_type=upstream_resp.headers.get("content-type"),
+        )
+
+    def _resolve_endpoint(self, name: str) -> str | None:
+        """Resolve an endpoint name to an address via the controller DB."""
+        query = EndpointQuery(exact_name=name)
+        joins, where = endpoint_query_predicate(query)
+        from iris.cluster.controller.service import ENDPOINTS
+
+        with self._db.read_snapshot() as q:
+            endpoints = q.select(ENDPOINTS, where=where, joins=joins)
+        if not endpoints:
+            return None
+        return endpoints[0].address

lib/iris/src/iris/cluster/controller/dashboard.py

@@ -37,6 +37,7 @@
 from starlette.routing import Mount, Route
 from starlette.types import ASGIApp, Receive, Scope, Send
 
+from iris.cluster.controller.actor_proxy import PROXY_ROUTE, ActorProxy
 from iris.cluster.controller.service import ControllerServiceImpl
 from iris.cluster.dashboard_common import html_shell, static_files_mount
 from iris.rpc.auth import SESSION_COOKIE, NullAuthInterceptor, TokenVerifier, extract_bearer_token, resolve_auth
@@ -260,6 +261,12 @@ def _create_app(self) -> ASGIApp:
         rpc_wsgi_app = ControllerServiceWSGIApplication(service=self._service, interceptors=interceptors)
         rpc_app = WSGIMiddleware(rpc_wsgi_app)
 
+        self._actor_proxy = ActorProxy(self._service._db)
+
+        @requires_auth
+        async def _proxy_actor_rpc(request: Request) -> Response:
+            return await self._actor_proxy.handle(request)
+
         routes = [
             Route("/", self._dashboard),
             Route("/auth/session_bootstrap", self._session_bootstrap),
@@ -270,10 +277,14 @@ def _create_app(self) -> ASGIApp:
             Route("/worker/{worker_id:path}", self._worker_detail_page),
             Route("/bundles/{bundle_id:str}.zip", self._bundle_download),
             Route("/health", self._health),
+            Route(PROXY_ROUTE, _proxy_actor_rpc, methods=["POST"]),
             Mount(rpc_wsgi_app.path, app=rpc_app),
             static_files_mount(),
         ]
-        app: Starlette | _RouteAuthMiddleware = Starlette(routes=routes)
+        app: Starlette | _RouteAuthMiddleware = Starlette(
+            routes=routes,
+            on_shutdown=[self._actor_proxy.close],
+        )
         if self._auth_verifier is not None and self._auth_provider is not None:
             app = _RouteAuthMiddleware(app, self._auth_verifier, optional=self._auth_optional)
         return app