---
name: hybrid-cloud-rpc
source: https://app.decimal.ai/s/hybrid-cloud-rpc@1/SKILL.md
source_sha256: 70d070b93af4
---

# 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.

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

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

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.

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

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

```python
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:

```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.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:

```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.CELL:
        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.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

```python
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`