Hybrid Cloud Rpc

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:

Step 2: Determine Silo Mode

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

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:

1. Add the new method alongside the old one
2. Migrate all callers to the new method
3. 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,
)

Step 8: Verify (Pre-flight Checklist)

Before submitting your PR, verify:

• No from __future__ import annotations in service.py or model.py
• All RPC method parameters are keyword-only (* separator)
• All parameters have explicit type annotations
• All types are serializable (primitives, RpcModel, list, Optional, dict, Enum, datetime)
• Cell service methods have @cell_rpc_method with appropriate resolver
• Control service methods have @rpc_method
• @cell_rpc_method / @rpc_method comes BEFORE @abstractmethod
• create_delegation() is called at module level at the bottom of service.py
• Service package is under one of the 12 registered discovery packages
• impl.py implements every abstract method with matching parameter names
• serial.py correctly converts ORM models to RPC models
• Sensitive fields use Field(repr=False) (tokens, secrets, config, metadata)
• Tests use @all_silo_test for full silo mode coverage
• Tests validate RPC model field accuracy against ORM objects
• Tests verify cross-silo resources (mappings, replicas) are created with correct data
• Tests cover error cases (not found, disabled methods, failed operations)
• Tests cover serialization round-trip via dispatch_to_local_service