ref(hc): Validate that we don't have non snowflake hcfks (#51467)

Two adjustments that tighten up deletions:
1. Validate in tests that no HybridCloudForeignKey pointing to a region
model is a not a snowflake id (since otherwise it won't be globally
unique to handle)
2. Ensure that the OrganizationMember objects are only being deleted in
codepaths that generate outboxes

Fortunately both conditions are met and there are no bugs, but this
tightens things up in terms of preventing future problems.

src/sentry/services/hybrid_cloud/organizationmember_mapping/impl.py

@@ -7,6 +7,7 @@ from typing import Optional
 
 from django.db import IntegrityError, transaction
 
+from sentry.db.postgres.roles import in_test_psql_role_override
 from sentry.models import outbox_context
 from sentry.models.organizationmembermapping import OrganizationMemberMapping
 from sentry.models.user import User
@@ -100,4 +101,5 @@ class DatabaseBackedOrganizationMemberMappingService(OrganizationMemberMappingSe
             organizationmember_id=organizationmember_id,
         )
         if org_member_map:
-            org_member_map.delete()
+            with in_test_psql_role_override("postgres"):
+                org_member_map.delete()

src/sentry/testutils/silo.py

@@ -15,10 +15,13 @@ from django.test import override_settings
 
 from sentry import deletions
 from sentry.db.models.base import ModelSiloLimit
+from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.deletions.base import BaseDeletionTask
+from sentry.models import Actor, NotificationSetting
 from sentry.silo import SiloMode
 from sentry.testutils.region import override_regions
 from sentry.types.region import Region, RegionCategory
+from sentry.utils.snowflake import SnowflakeIdMixin
 
 TestMethod = Callable[..., None]
 
@@ -283,6 +286,22 @@ def validate_relation_does_not_cross_silo_foreign_keys(
             )
 
 
+def validate_hcfk_has_global_id(model: Type[Model], related_model: Type[Model]):
+    # HybridCloudForeignKey can point to region models if they have snowflake ids
+    if issubclass(related_model, SnowflakeIdMixin):
+        return
+
+    # This particular relation is being removed before we go multi region.
+    if related_model is Actor and model is NotificationSetting:
+        return
+
+    # but they cannot point to region models otherwise.
+    if SiloMode.REGION in _model_silo_limit(related_model).modes:
+        raise ValueError(
+            f"{related_model!r} runs in {SiloMode.REGION}, but is related to {model!r} via a HybridCloudForeignKey! Region model ids are not global, unless you use a snowflake id."
+        )
+
+
 def validate_model_no_cross_silo_foreign_keys(
     model: Type[Model],
     exemptions: Set[Tuple[Type[Model], Type[Model]]],
@@ -301,4 +320,6 @@ def validate_model_no_cross_silo_foreign_keys(
 
             validate_relation_does_not_cross_silo_foreign_keys(model, field.related_model)
             validate_relation_does_not_cross_silo_foreign_keys(field.related_model, model)
+        if isinstance(field, HybridCloudForeignKey):
+            validate_hcfk_has_global_id(model, field.foreign_model)
     return seen

tests/conftest.py

@@ -205,6 +205,9 @@ def protect_hybrid_cloud_writes_and_deletes(request):
     # outboxes in a transaction, and cover that transaction with `in_test_psql_role_override`
     restrict_role(role="postgres_unprivileged", model=OrganizationMember, revocation_type="INSERT")
     restrict_role(role="postgres_unprivileged", model=OrganizationMember, revocation_type="UPDATE")
+    # OrganizationMember objects need to cascade, but they can't use the standard hybrid cloud foreign key because the
+    # identifiers are not snowflake ids.
+    restrict_role(role="postgres_unprivileged", model=OrganizationMember, revocation_type="DELETE")
 
     restrict_role(
         role="postgres_unprivileged", model=OrganizationMemberMapping, revocation_type="INSERT"

tests/sentry/receivers/test_transactions.py

@@ -1,6 +1,7 @@
 from functools import cached_property
 from unittest.mock import patch
 
+from sentry.db.postgres.roles import in_test_psql_role_override
 from sentry.models import OrganizationMember, Project
 from sentry.signals import event_processed, transaction_processed
 from sentry.testutils import TestCase
@@ -83,7 +84,8 @@ class RecordFirstTransactionTest(TestCase):
         )
 
     def test_analytics_event_no_owner(self):
-        OrganizationMember.objects.filter(organization=self.organization, role="owner").delete()
+        with in_test_psql_role_override("postgres"):
+            OrganizationMember.objects.filter(organization=self.organization, role="owner").delete()
         assert not self.project.flags.has_transactions
         event = self.store_event(
             data={