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, billing, org management) and Region silos (project data, events, issues).

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
Region silo (projects, events, issues, org data)	`SiloMode.REGION`	`@regional_rpc_method(resolve=...)`	`OrganizationService`
Control silo (users, auth, billing, org mappings)	`SiloMode.CONTROL`	`@rpc_method`	`OrganizationMemberMappingService`

Decision rule: If the Django models you need to query live in the region database, use

SiloMode.REGION

. If they live in the control database, use

SiloMode.CONTROL

Region-silo services require a

RegionResolutionStrategy

on every RPC method so the framework knows which region 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

from __future__ import annotations

service.py

model.py

Step 4: Add or Update Methods

For REGION silo services

Load

references/resolvers.md

for resolver details.

python

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

Key rules:

```
@regional_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

python

@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:

python

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:

python

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.

python

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, REGION, CONTROL). This ensures the delegation layer works for both local and remote dispatch paths.

python

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

@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 regions (e.g., testing region resolution):

python

@all_silo_test(regions=create_test_regions("us", "eu"))
class MyServiceRegionTest(TransactionTestCase):
    ...

Use

assume_test_silo_mode

assume_test_silo_mode_of

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

python

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.REGION):
        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:

python

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:

python

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:

python

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:

python

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:

python

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):

python

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.REGION):
        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:

python

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:

python

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:

python

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:

python

from sentry.hybridcloud.rpc.service import RpcRemoteException

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

Test that failed operations produce no side effects:

python

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.REGION):
        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

python

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    region_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_regions,
)
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)
Region service methods have
```
@regional_rpc_method
```
with appropriate resolver
Control service methods have
```
@rpc_method
```

@regional_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
```

hybrid-cloud-rpc

NPX Install

Tags

SKILL.md Content

Hybrid Cloud RPC Services

Critical Constraints

Step 1: Determine Operation

Step 2: Determine Silo Mode

Step 3: Create a New Service

Directory structure

Registration

Checklist for new services

Step 4: Add or Update Methods

For REGION silo services

For CONTROL silo services

Non-abstract convenience methods

Implementation in impl.py

Error propagation

RPC Models

Step 5: Update Method Signatures

Safe changes (backwards compatible)

Breaking changes (require coordination)

Step 6: Deprecate or Remove

Step 7: Test

7.1 Silo mode compatibility with
`@all_silo_test`

7.2 Serialization round-trip with
`dispatch_to_local_service`

7.3 RPC model data accuracy

7.4 Cross-silo resource creation

7.5 Error handling

7.6 Key imports for testing

Step 8: Verify (Pre-flight Checklist)