feat(hybridcloud): Propagate ViewerContext through cross-silo RPC

src/sentry/api/endpoints/internal/rpc.py

@@ -17,6 +17,7 @@
 )
 from sentry.hybridcloud.rpc.sig import SerializableFunctionValueException
 from sentry.utils.env import in_test_environment
+from sentry.viewer_context import ViewerContext, viewer_context_scope
 
 
 @internal_all_silo_endpoint
@@ -61,8 +62,18 @@ def post(self, request: Request, service_name: str, method_name: str) -> Respons
                 sentry_sdk.capture_exception()
                 raise ParseError from e
 
+        meta = request.data.get("meta") or {}
+        vc_data = meta.get("viewer_context")
+        vc = ViewerContext()
+        if vc_data:
+            try:
+                vc = ViewerContext.deserialize(vc_data)
+            except Exception as e:
+                sentry_sdk.capture_exception()
+                raise ParseError from e
+
         try:
-            with auth_context.applied_to_request(request):
+            with viewer_context_scope(vc), auth_context.applied_to_request(request):
                 result = dispatch_to_local_service(service_name, method_name, arguments)
         except RpcValidationException as e:
             return Response(

src/sentry/hybridcloud/rpc/service.py

@@ -35,6 +35,7 @@
 from sentry.types.cell import Cell, CellMappingNotFound
 from sentry.utils import json, metrics
 from sentry.utils.env import in_test_environment
+from sentry.viewer_context import get_viewer_context
 
 if TYPE_CHECKING:
     from sentry.hybridcloud.rpc.resolvers import CellResolutionStrategy
@@ -571,8 +572,13 @@ def get_method_timeout(self) -> float:
         return settings.RPC_TIMEOUT
 
     def _send_to_remote_silo(self, use_test_client: bool) -> Any:
+        vc = get_viewer_context()
+        meta: dict[str, Any] = {}
+        if vc is not None:
+            meta["viewer_context"] = vc.serialize()
+
         request_body = {
-            "meta": {},  # reserved for future use
+            "meta": meta,
             "args": self.serial_arguments,
         }
         data = json.dumps(request_body).encode(_RPC_CONTENT_CHARSET)

src/sentry/viewer_context.py

@@ -62,6 +62,19 @@ def serialize(self) -> dict[str, Any]:
             result["token"] = {"kind": self.token.kind, "scopes": list(self.token.get_scopes())}
         return result
 
+    @classmethod
+    def deserialize(cls, data: dict[str, Any]) -> ViewerContext:
+        """Reconstruct from a serialized dict. Token is not deserialized."""
+        try:
+            actor_type = ActorType(data.get("actor_type", "unknown"))
+        except ValueError:
+            actor_type = ActorType.UNKNOWN
+        return cls(
+            organization_id=data.get("organization_id"),
+            user_id=data.get("user_id"),
+            actor_type=actor_type,
+        )
+
 
 @contextlib.contextmanager
 def viewer_context_scope(ctx: ViewerContext) -> Generator[None]:

tests/sentry/api/endpoints/test_rpc.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from typing import Any
+from unittest.mock import patch
 
 import orjson
 from django.test import override_settings
@@ -10,6 +11,7 @@
 from sentry.hybridcloud.rpc.service import generate_request_signature
 from sentry.organizations.services.organization import RpcUserOrganizationContext
 from sentry.testutils.cases import APITestCase
+from sentry.viewer_context import ActorType, ViewerContext, get_viewer_context
 
 
 @override_settings(RPC_SHARED_SECRET=["a-long-value-that-is-hard-to-guess"])
@@ -142,3 +144,107 @@ def test_with_invalid_arguments(self) -> None:
         assert response.data == {
             "detail": ErrorDetail(string="Malformed request.", code="parse_error")
         }
+
+    def test_viewer_context_propagated_from_meta(self) -> None:
+        """ViewerContext in meta is set as the contextvar during dispatch."""
+        organization = self.create_organization()
+        captured_contexts: list[ViewerContext | None] = []
+
+        original_dispatch = __import__(
+            "sentry.hybridcloud.rpc.service", fromlist=["dispatch_to_local_service"]
+        ).dispatch_to_local_service
+
+        def capturing_dispatch(*args, **kwargs):
+            captured_contexts.append(get_viewer_context())
+            return original_dispatch(*args, **kwargs)
+
+        path = self._get_path("organization", "get_organization_by_id")
+        data = {
+            "args": {"id": organization.id},
+            "meta": {
+                "viewer_context": {
+                    "organization_id": organization.id,
+                    "user_id": 42,
+                    "actor_type": "user",
+                }
+            },
+        }
+
+        with patch(
+            "sentry.api.endpoints.internal.rpc.dispatch_to_local_service",
+            side_effect=capturing_dispatch,
+        ):
+            response = self._send_post_request(path, data)
+
+        assert response.status_code == 200
+        assert len(captured_contexts) == 1
+        ctx = captured_contexts[0]
+        assert ctx is not None
+        assert ctx.organization_id == organization.id
+        assert ctx.user_id == 42
+        assert ctx.actor_type == ActorType.USER
+
+    def test_viewer_context_unknown_when_meta_empty(self) -> None:
+        """Empty ViewerContext with UNKNOWN actor type when meta has no viewer_context."""
+        organization = self.create_organization()
+        captured_contexts: list[ViewerContext | None] = []
+
+        original_dispatch = __import__(
+            "sentry.hybridcloud.rpc.service", fromlist=["dispatch_to_local_service"]
+        ).dispatch_to_local_service
+
+        def capturing_dispatch(*args, **kwargs):
+            captured_contexts.append(get_viewer_context())
+            return original_dispatch(*args, **kwargs)
+
+        path = self._get_path("organization", "get_organization_by_id")
+        data = {"args": {"id": organization.id}, "meta": {}}
+
+        with patch(
+            "sentry.api.endpoints.internal.rpc.dispatch_to_local_service",
+            side_effect=capturing_dispatch,
+        ):
+            response = self._send_post_request(path, data)
+
+        assert response.status_code == 200
+        assert len(captured_contexts) == 1
+        ctx = captured_contexts[0]
+        assert ctx is not None
+        assert ctx.user_id is None
+        assert ctx.organization_id is None
+        assert ctx.actor_type == ActorType.UNKNOWN
+
+    def test_viewer_context_roundtrip_through_meta(self) -> None:
+        """ViewerContext set on the sending side arrives on the receiving side."""
+        organization = self.create_organization()
+        captured_contexts: list[ViewerContext | None] = []
+
+        original_dispatch = __import__(
+            "sentry.hybridcloud.rpc.service", fromlist=["dispatch_to_local_service"]
+        ).dispatch_to_local_service
+
+        def capturing_dispatch(*args, **kwargs):
+            captured_contexts.append(get_viewer_context())
+            return original_dispatch(*args, **kwargs)
+
+        # Simulate what _send_to_remote_silo builds when ViewerContext is set
+        ctx = ViewerContext(organization_id=organization.id, user_id=42, actor_type=ActorType.USER)
+        path = self._get_path("organization", "get_organization_by_id")
+        data = {
+            "args": {"id": organization.id},
+            "meta": {"viewer_context": ctx.serialize()},
+        }
+
+        with patch(
+            "sentry.api.endpoints.internal.rpc.dispatch_to_local_service",
+            side_effect=capturing_dispatch,
+        ):
+            response = self._send_post_request(path, data)
+
+        assert response.status_code == 200
+        assert len(captured_contexts) == 1
+        restored = captured_contexts[0]
+        assert restored is not None
+        assert restored.organization_id == ctx.organization_id
+        assert restored.user_id == ctx.user_id
+        assert restored.actor_type == ctx.actor_type