feat(hybridcloud) Add control generational caching table (#68465)

Add a control silo version of the cross-region generational caching
model. I'd like to start high-frequency RPC calls coming from control to
the regions.

Author: markstory

fixtures/backup/model_dependencies/detailed.json

@@ -87,6 +87,22 @@
     "table_name": "hybridcloud_apitokenreplica",
     "uniques": []
   },
+  "hybridcloud.controlcacheversion": {
+    "dangling": false,
+    "foreign_keys": {},
+    "model": "hybridcloud.controlcacheversion",
+    "relocation_dependencies": [],
+    "relocation_scope": "Excluded",
+    "silos": [
+      "Control"
+    ],
+    "table_name": "hybridcloud_controlcacheversion",
+    "uniques": [
+      [
+        "key"
+      ]
+    ]
+  },
   "hybridcloud.externalactorreplica": {
     "dangling": false,
     "foreign_keys": {

fixtures/backup/model_dependencies/flat.json

@@ -14,6 +14,7 @@
     "sentry.organization",
     "sentry.user"
   ],
+  "hybridcloud.controlcacheversion": [],
   "hybridcloud.externalactorreplica": [
     "sentry.externalactor",
     "sentry.integration",

fixtures/backup/model_dependencies/sorted.json

@@ -1,4 +1,5 @@
 [
+  "hybridcloud.controlcacheversion",
   "hybridcloud.regioncacheversion",
   "hybridcloud.webhookpayload",
   "nodestore.node",

fixtures/backup/model_dependencies/truncate.json

@@ -1,4 +1,5 @@
 [
+  "hybridcloud_controlcacheversion",
   "hybridcloud_regioncacheversion",
   "hybridcloud_webhookpayload",
   "nodestore_node",

migrations_lockfile.txt

@@ -6,7 +6,7 @@ To resolve this, rebase against latest master and regenerate your migration. Thi
 will then be regenerated, and you should be able to merge without conflicts.
 
 feedback: 0004_index_together
-hybridcloud: 0015_apitokenreplica_hashed_token_index
+hybridcloud: 0016_add_control_cacheversion
 nodestore: 0002_nodestore_no_dictfield
 replays: 0004_index_together
 sentry: 0693_add_monitors_ownership_actor_id

src/sentry/hybridcloud/migrations/0016_add_control_cacheversion.py

@@ -0,0 +1,45 @@
+# Generated by Django 5.0.3 on 2024-04-08 20:58
+
+from django.db import migrations, models
+
+import sentry.db.models.fields.bounded
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production.
+    # This should only be used for operations where it's safe to run the migration after your
+    # code has deployed. So this should not be used for most operations that alter the schema
+    # of a table.
+    # Here are some things that make sense to mark as post deployment:
+    # - Large data migrations. Typically we want these to be run manually so that they can be
+    #   monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   run this outside deployments so that we don't block them. Note that while adding an index
+    #   is a schema change, it's completely safe to run the operation after the code has deployed.
+    # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment
+
+    is_post_deployment = False
+
+    dependencies = [
+        ("hybridcloud", "0015_apitokenreplica_hashed_token_index"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ControlCacheVersion",
+            fields=[
+                (
+                    "id",
+                    sentry.db.models.fields.bounded.BoundedBigAutoField(
+                        primary_key=True, serialize=False
+                    ),
+                ),
+                ("key", models.CharField(max_length=64, unique=True)),
+                ("version", models.PositiveBigIntegerField(default=0)),
+            ],
+            options={
+                "db_table": "hybridcloud_controlcacheversion",
+            },
+        ),
+    ]

src/sentry/hybridcloud/models/cacheversion.py

@@ -2,7 +2,7 @@
 from django.db.models import F
 
 from sentry.backup.scopes import RelocationScope
-from sentry.db.models import Model, region_silo_only_model
+from sentry.db.models import Model, control_silo_only_model, region_silo_only_model
 from sentry.db.postgres.transactions import enforce_constraints
 
 
@@ -33,3 +33,12 @@ class RegionCacheVersion(CacheVersionBase):
     class Meta:
         app_label = "hybridcloud"
         db_table = "hybridcloud_regioncacheversion"
+
+
+@control_silo_only_model
+class ControlCacheVersion(CacheVersionBase):
+    __relocation_scope__ = RelocationScope.Excluded
+
+    class Meta:
+        app_label = "hybridcloud"
+        db_table = "hybridcloud_controlcacheversion"

src/sentry/hybridcloud/rpc/services/caching/impl.py

@@ -3,7 +3,11 @@
 
 from django.core.cache import cache
 
-from sentry.hybridcloud.models.cacheversion import CacheVersionBase, RegionCacheVersion
+from sentry.hybridcloud.models.cacheversion import (
+    CacheVersionBase,
+    ControlCacheVersion,
+    RegionCacheVersion,
+)
 from sentry.hybridcloud.rpc.services.caching import ControlCachingService, RegionCachingService
 from sentry.silo import SiloMode
 
@@ -21,7 +25,7 @@ def _consume_generator(g: Generator[None, None, _V]) -> _V:
             return e.value
 
 
-def _set_cache(key: str, value: str, version: int) -> Generator[None, None, bool]:
+def _set_cache(key: str, value: str | None, version: int) -> Generator[None, None, bool]:
     result = cache.add(_versioned_key(key, version), value)
     yield
     return result
@@ -34,6 +38,8 @@ def _versioned_key(key: str, version: int) -> str:
 def _version_model(mode: SiloMode) -> type[CacheVersionBase]:
     if mode == SiloMode.REGION:
         return RegionCacheVersion
+    if mode == SiloMode.CONTROL:
+        return ControlCacheVersion
     raise ValueError
 
 

src/sentry/hybridcloud/rpc/services/caching/service.py

@@ -4,7 +4,7 @@
 # defined, because we want to reflect on type annotations and avoid forward references.
 import abc
 from collections.abc import Callable, Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Generic, TypeVar
+from typing import TYPE_CHECKING, Generic, TypeVar, Union
 
 import pydantic
 
@@ -35,21 +35,24 @@ def clear_key(self, *, region_name: str, key: str) -> int:
 
 
 _R = TypeVar("_R", bound=pydantic.BaseModel)
+_OptionalR = Union[_R, None]
 
 
 class SiloCacheBackedCallable(Generic[_R]):
     silo_mode: SiloMode
     base_key: str
-    cb: Callable[[int], _R]
+    cb: Callable[[int], _R | None]
     type_: type[_R]
 
-    def __init__(self, base_key: str, silo_mode: SiloMode, cb: Callable[[int], _R], t: type[_R]):
+    def __init__(
+        self, base_key: str, silo_mode: SiloMode, cb: Callable[[int], _OptionalR], t: type[_R]
+    ):
         self.base_key = base_key
         self.silo_mode = silo_mode
         self.cb = cb
         self.type_ = t
 
-    def __call__(self, args: int) -> _R:
+    def __call__(self, args: int) -> _OptionalR:
         if (
             SiloMode.get_current_mode() != self.silo_mode
             and SiloMode.get_current_mode() != SiloMode.MONOLITH
@@ -60,7 +63,9 @@ def __call__(self, args: int) -> _R:
     def key_from(self, args: int) -> str:
         return f"{self.base_key}:{args}"
 
-    def resolve_from(self, i: int, values: Mapping[str, int | str]) -> Generator[None, None, _R]:
+    def resolve_from(
+        self, i: int, values: Mapping[str, int | str]
+    ) -> Generator[None, None, _OptionalR]:
         from .impl import _consume_generator, _delete_cache, _set_cache
 
         key = self.key_from(i)
@@ -75,10 +80,11 @@ def resolve_from(self, i: int, values: Mapping[str, int | str]) -> Generator[Non
             version = value
 
         r = self.cb(i)
-        _consume_generator(_set_cache(key, r.json(), version))
+        if r is not None:
+            _consume_generator(_set_cache(key, r.json(), version))
         return r
 
-    def get_many(self, ids: Sequence[int]) -> list[_R]:
+    def get_many(self, ids: Sequence[int]) -> list[_OptionalR]:
         from .impl import _consume_generator, _get_cache
 
         keys = [self.key_from(i) for i in ids]
@@ -88,8 +94,8 @@ def get_many(self, ids: Sequence[int]) -> list[_R]:
 
 def back_with_silo_cache(
     base_key: str, silo_mode: SiloMode, t: type[_R]
-) -> Callable[[Callable[[int], _R]], "SiloCacheBackedCallable[_R]"]:
-    def wrapper(cb: Callable[[int], _R]) -> "SiloCacheBackedCallable[_R]":
+) -> Callable[[Callable[[int], _OptionalR]], "SiloCacheBackedCallable[_R]"]:
+    def wrapper(cb: Callable[[int], _OptionalR]) -> "SiloCacheBackedCallable[_R]":
         return SiloCacheBackedCallable(base_key, silo_mode, cb, t)
 
     return wrapper

src/sentry/services/hybrid_cloud/user/service.py

@@ -121,10 +121,7 @@ def flush_nonce(self, *, user_id: int) -> None:
         pass
 
     def get_user(self, user_id: int) -> RpcUser | None:
-        user = get_user(user_id)
-        if user.is_anonymous:
-            return None
-        return user
+        return get_user(user_id)
 
     @rpc_method
     @abstractmethod
@@ -195,12 +192,11 @@ def get_user_avatar(self, *, user_id: int) -> RpcAvatar | None:
 
 
 @back_with_silo_cache("user_service.get_user", SiloMode.REGION, RpcUser)
-def get_user(user_id: int) -> RpcUser:
+def get_user(user_id: int) -> RpcUser | None:
     users = user_service.get_many(filter=dict(user_ids=[user_id]))
     if len(users) > 0:
         return users[0]
-    else:
-        return RpcUser(is_anonymous=True)
+    return None
 
 
 user_service = UserService.create_delegation()

tests/sentry/backup/snapshots/test_comparators/test_default_comparators.pysnap

@@ -1,5 +1,5 @@
 ---
-created: '2024-04-10T14:45:05.473571+00:00'
+created: '2024-04-10T15:02:31.213310+00:00'
 creator: sentry
 source: tests/sentry/backup/test_comparators.py
 ---
@@ -24,6 +24,10 @@ source: tests/sentry/backup/test_comparators.py
     - organization
     - user_id
   model_name: hybridcloud.apitokenreplica
+- comparators:
+  - class: ForeignKeyComparator
+    fields: []
+  model_name: hybridcloud.controlcacheversion
 - comparators:
   - class: ForeignKeyComparator
     fields:

tests/sentry/hybridcloud/test_caching.py

@@ -2,14 +2,20 @@
 
 from django.core.cache import cache
 
-from sentry.hybridcloud.rpc.services.caching import back_with_silo_cache, region_caching_service
+from sentry.hybridcloud.rpc.services.caching import (
+    back_with_silo_cache,
+    control_caching_service,
+    region_caching_service,
+)
 from sentry.hybridcloud.rpc.services.caching.impl import CacheBackend
+from sentry.services.hybrid_cloud.organization.model import RpcOrganizationSummary
+from sentry.services.hybrid_cloud.organization.service import organization_service
 from sentry.services.hybrid_cloud.user import RpcUser
 from sentry.services.hybrid_cloud.user.service import user_service
 from sentry.silo import SiloMode
 from sentry.testutils.factories import Factories
 from sentry.testutils.pytest.fixtures import django_db_all
-from sentry.testutils.silo import assume_test_silo_mode, no_silo_test
+from sentry.testutils.silo import assume_test_silo_mode, control_silo_test, no_silo_test
 from sentry.types.region import get_local_region
 
 
@@ -26,6 +32,7 @@ def get_user(user_id: int) -> RpcUser:
 
     for u in users:
         next_user = get_user(u.id)
+        assert next_user
         assert next_user == get_user.cb(u.id)
         old.append(next_user)
 
@@ -43,9 +50,65 @@ def get_user(user_id: int) -> RpcUser:
         )
 
     for u, cached in zip(users, get_user.get_many([u.id for u in users])):
+        assert cached
         assert cached.username == u.username
 
 
+@control_silo_test
+@django_db_all(transaction=True)
+def test_caching_function_control():
+    cache.clear()
+
+    @back_with_silo_cache(
+        base_key="my-test-key", silo_mode=SiloMode.CONTROL, t=RpcOrganizationSummary
+    )
+    def get_org(org_id: int) -> RpcOrganizationSummary | None:
+        return organization_service.get_org_by_id(id=org_id)
+
+    user = Factories.create_user()
+    orgs = [Factories.create_organization(owner=user) for _ in range(2)]
+    old: list[RpcOrganizationSummary] = []
+
+    for org in orgs:
+        next_org = get_org(org.id)
+        assert next_org
+        assert next_org == get_org.cb(org.id)
+        old.append(next_org)
+
+    for org in orgs:
+        with assume_test_silo_mode(SiloMode.REGION):
+            org.update(name=org.name + " updated")
+
+    # Does not include updates
+    for org in old:
+        next_org = get_org(org.id)
+        assert next_org == org
+
+        control_caching_service.clear_key(key=get_org.key_from(org.id))
+
+    for o, cached in zip(orgs, get_org.get_many([o.id for o in orgs])):
+        assert cached
+        assert cached.name == o.name
+
+
+@control_silo_test
+@django_db_all(transaction=True)
+def test_caching_function_none_value():
+    cache.clear()
+
+    @back_with_silo_cache(
+        base_key="my-test-key", silo_mode=SiloMode.CONTROL, t=RpcOrganizationSummary
+    )
+    def get_org(org_id: int) -> RpcOrganizationSummary | None:
+        return organization_service.get_org_by_id(id=org_id)
+
+    result = get_org(9999)
+    assert result is None
+
+    result = get_org(9999)
+    assert result is None
+
+
 @django_db_all(transaction=True)
 @no_silo_test
 def test_cache_versioning():