hybrid-cloud-rpc

작성자: sentry

Sentry에서 하이브리드 클라우드 RPC 서비스를 생성, 업데이트, 폐기하는 가이드입니다. "RPC 메서드 추가", "RPC 서비스 생성", "하이브리드 클라우드…"와 같은 요청 시 사용하세요.

npx skills add https://github.com/getsentry/sentry --skill hybrid-cloud-rpc

Hybrid Cloud RPC Services

This skill guides you through creating, modifying, and deprecating RPC services in Sentry's hybrid cloud architecture. RPC services enable cross-silo communication between the Control silo (user auth, org management) and Cell silos (project data, events, issues, billing).

Critical Constraints

NEVER use from __future__ import annotations in service.py or model.py files. The RPC framework reflects on type annotations at import time. Forward references break serialization silently.

ALL RPC method parameters must be keyword-only (use * in the signature).

ALL parameters and return types must have full type annotations — no string forward references.

ONLY serializable types are allowed: int, str, bool, float, None, Optional[T], list[T], dict[str, T], RpcModel subclasses, Enum subclasses, datetime.datetime.

The service MUST live in one of the 12 registered discovery packages (see Step 3).

Use Field(repr=False) on sensitive fields (tokens, secrets, keys, config blobs, metadata dicts) to prevent them from leaking into logs and error reports. See references/rpc-models.md for the full guide.

Step 1: Determine Operation

Classify what the developer needs:

Intent	Go to
Create a brand-new RPC service	Step 2, then Step 3
Add a method to an existing service	Step 2, then Step 4
Update an existing method's signature	Step 5
Deprecate or remove a method/service	Step 6

Step 2: Determine Silo Mode

The service's local_mode determines where the database-backed implementation runs:

Data lives in...	`local_mode`	Decorator on methods	Example
Cell silo (projects, events, issues, org data, billing)	`SiloMode.CELL`	`@cell_rpc_method(resolve=...)`	`OrganizationService`
Control silo (users, auth, org mappings)	`SiloMode.CONTROL`	`@rpc_method`	`OrganizationMemberMappingService`

Decision rule: If the Django models you need to query live in the cell database, use SiloMode.CELL. If they live in the control database, use SiloMode.CONTROL.

Cell-silo services require a CellResolutionStrategy on every RPC method so the framework knows which cell to route remote calls to. Load references/resolvers.md for the full resolver table.

Step 3: Create a New Service

Load references/service-template.md for copy-paste file templates.

Directory structure

src/sentry/{domain}/services/{service_name}/
├── __init__.py      # Re-exports model and service
├── model.py         # RpcModel subclasses (NO future annotations)
├── serial.py        # ORM → RpcModel conversion functions
├── service.py       # Abstract service class (NO future annotations)
└── impl.py          # DatabaseBacked implementation

Registration

The service package MUST be a sub-package of one of these 12 registered discovery packages:

sentry.auth.services
sentry.audit_log.services
sentry.backup.services
sentry.hybridcloud.services
sentry.identity.services
sentry.integrations.services
sentry.issues.services
sentry.notifications.services
sentry.organizations.services
sentry.projects.services
sentry.sentry_apps.services
sentry.users.services

If your service doesn't fit any of these, add a new entry to the service_packages tuple in src/sentry/hybridcloud/rpc/service.py:list_all_service_method_signatures().

Checklist for new services

key is unique across all services (check existing keys with grep -r 'key = "' src/sentry/*/services/*/service.py)
local_mode matches where the data lives
get_local_implementation() returns the DatabaseBacked subclass
Module-level my_service = MyService.create_delegation() at bottom of service.py
__init__.py re-exports models and service
No from __future__ import annotations in service.py or model.py

Step 4: Add or Update Methods

For CELL silo services

Load references/resolvers.md for resolver details.

@cell_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def my_method(
    self,
    *,
    organization_id: int,
    name: str,
    options: RpcMyOptions | None = None,
) -> RpcMyResult | None:
    pass

Key rules:

@cell_rpc_method MUST come before @abstractmethod
The resolver parameter (e.g., organization_id) MUST be in the method signature
Use return_none_if_mapping_not_found=True when the return type is Optional and a missing org mapping means "not found" rather than an error

For CONTROL silo services

@rpc_method
@abstractmethod
def my_method(
    self,
    *,
    user_id: int,
    data: RpcMyData,
) -> RpcMyResult:
    pass

Non-abstract convenience methods

You can also add non-abstract methods that compose other RPC calls. These run locally and are NOT exposed as RPC endpoints:

def get_by_slug_or_id(self, *, slug: str | None = None, id: int | None = None) -> RpcThing | None:
    if slug:
        return self.get_by_slug(slug=slug)
    if id:
        return self.get_by_id(id=id)
    return None

Implementation in impl.py

The DatabaseBacked subclass must implement every @abstractmethod with the exact same parameter names:

class DatabaseBackedMyService(MyService):
    def my_method(self, *, organization_id: int, name: str, options: RpcMyOptions | None = None) -> RpcMyResult | None:
        # ORM queries here
        obj = MyModel.objects.filter(organization_id=organization_id, name=name).first()
        if obj is None:
            return None
        return serialize_my_model(obj)

Error propagation

All errors an RPC method propagates must be done via the return type. Errors are rewrapped and returned as generic Invalid service request to external callers.

class RpcTentativeResult(RpcModel):
    success: bool
    error_str: str | None
    result: str | None

class DatabaseBackedMyService(MyService):
    def foobar(self, *, organization_id: int) -> RpcTentativeResult
        try:
            some_function_call()
        except e:
            return RpcTentativeResult(success=False, error_str = str(e))

        return RpcTentativeResult(success=True, result="foobar")

RPC Models

Load references/rpc-models.md for supported types, default values, and serialization patterns.

Step 5: Update Method Signatures

Safe changes (backwards compatible)

Adding a new optional parameter with a default value
Widening a return type (e.g., RpcFoo → RpcFoo | None) on a Control RPC service
Adding fields with defaults to an RpcModel

Breaking changes (require coordination)

Removing or renaming a parameter
Changing a parameter's type
Narrowing a return type
Removing fields from an RpcModel

For breaking changes, use a two-phase approach:

Add the new method alongside the old one
Migrate all callers to the new method
Remove the old method (see Step 6)

Step 6: Deprecate or Remove

Load references/deprecation.md for the full 3-phase workflow.

Quick summary: Disable at runtime → migrate callers → remove code.

Step 7: Test

Every RPC service needs three categories of tests: silo mode compatibility, data accuracy, and error handling. Use TransactionTestCase (not TestCase) when tests need outbox processing or on_commit hooks.

7.1 Silo mode compatibility with `@all_silo_test`

Every service test class MUST use @all_silo_test so tests run in all three modes (MONOLITH, CELL, CONTROL). This ensures the delegation layer works for both local and remote dispatch paths.

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, create_test_cells

@all_silo_test
class MyServiceTest(TestCase):
    def test_get_by_id(self):
        org = self.create_organization()
        result = my_service.get_by_id(organization_id=org.id, id=thing.id)
        assert result is not None

For tests that need named cells (e.g., testing cell resolution):

@all_silo_test(cells=create_test_cells("us", "eu"))
class MyServiceCellTest(TransactionTestCase):
    ...

Use assume_test_silo_mode or assume_test_silo_mode_of to switch modes within a test when accessing ORM models that live in a different silo:

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.CELL):
        org = self.create_organization()
    result = my_service.get_by_id(organization_id=org.id, id=thing.id)
    assert result is not None

7.2 Serialization round-trip with `dispatch_to_local_service`

Test that arguments and return values survive serialization/deserialization:

from sentry.hybridcloud.rpc.service import dispatch_to_local_service

def test_serialization_round_trip(self):
    result = dispatch_to_local_service(
        "my_service_key",
        "my_method",
        {"organization_id": org.id, "name": "test"},
    )
    assert result["value"] is not None

7.3 RPC model data accuracy

Validate that RPC models faithfully represent the ORM data. Compare every field of the RPC model against the source ORM object:

def test_rpc_model_accuracy(self):
    orm_obj = MyModel.objects.get(id=thing.id)
    rpc_obj = my_service.get_by_id(organization_id=org.id, id=thing.id)

    assert rpc_obj.id == orm_obj.id
    assert rpc_obj.name == orm_obj.name
    assert rpc_obj.organization_id == orm_obj.organization_id
    assert rpc_obj.is_active == orm_obj.is_active
    assert rpc_obj.date_added == orm_obj.date_added

For models with flags or nested objects, iterate all field names:

def test_flags_accuracy(self):
    rpc_org = organization_service.get(id=org.id)
    for field_name in rpc_org.flags.get_field_names():
        assert getattr(rpc_org.flags, field_name) == getattr(orm_org.flags, field_name)

For list results, sort both sides by ID before comparing:

def test_list_accuracy(self):
    rpc_items = my_service.list_things(organization_id=org.id)
    orm_items = list(MyModel.objects.filter(organization_id=org.id).order_by("id"))
    assert len(rpc_items) == len(orm_items)
    for rpc_item, orm_item in zip(sorted(rpc_items, key=lambda x: x.id), orm_items):
        assert rpc_item.id == orm_item.id
        assert rpc_item.name == orm_item.name

7.4 Cross-silo resource creation

If your service creates or updates resources that propagate across silos (via outboxes or mappings), verify the cross-silo effects.

Use outbox_runner() to flush outboxes synchronously during tests:

from sentry.testutils.outbox import outbox_runner

def test_cross_silo_mapping_created(self):
    with outbox_runner():
        my_service.create_thing(organization_id=org.id, name="test")

    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
        assert mapping.name == "test"

For triple-equality assertions (RPC result = source ORM = cross-silo replica):

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.CELL):
        orm_obj = MyModel.objects.get(id=rpc_result.id)
    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
    assert rpc_result.slug == orm_obj.slug == mapping.slug

Use HybridCloudTestMixin for common cross-silo assertions:

from sentry.testutils.hybrid_cloud import HybridCloudTestMixin

class MyServiceTest(HybridCloudTestMixin, TransactionTestCase):
    def test_member_mapping_synced(self):
        self.assert_org_member_mapping(org_member=org_member)

7.5 Error handling

Test that the service handles errors correctly in all silo modes:

def test_not_found_returns_none(self):
    result = my_service.get_by_id(organization_id=org.id, id=99999)
    assert result is None

def test_missing_org_returns_none(self):
    # For methods with return_none_if_mapping_not_found=True
    result = my_service.get_by_id(organization_id=99999, id=1)
    assert result is None

Test disabled methods:

from sentry.hybridcloud.rpc.service import RpcDisabledException
from sentry.testutils.helpers.options import override_options

def test_disabled_method_raises(self):
    with override_options({"hybrid_cloud.rpc.disabled-service-methods": ["MyService.my_method"]}):
        with pytest.raises(RpcDisabledException):
            dispatch_remote_call(None, "my_service_key", "my_method", {"id": 1})

Test that remote exceptions are properly wrapped:

from sentry.hybridcloud.rpc.service import RpcRemoteException

def test_remote_error_wrapping(self):
    if SiloMode.get_current_mode() == SiloMode.CELL:
        with pytest.raises(RpcRemoteException):
            my_control_service.do_thing_that_fails(...)

Test that failed operations produce no side effects:

def test_no_side_effects_on_failure(self):
    result = my_service.create_conflicting_thing(organization_id=org.id)
    assert not result
    with assume_test_silo_mode(SiloMode.CELL):
        assert not MyModel.objects.filter(organization_id=org.id).exists()

Test that any calling code (both direct and indirect) is also appropriately tested with the correct silo decorators.

7.6 Key imports for testing

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    cell_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.hybridcloud.rpc.service import (
    dispatch_to_local_service,
    dispatch_remote_call,
    RpcDisabledException,
    RpcRemoteException,
)