chore(hybrid-cloud): Break ProjectIntegration fk (#44992)

Accomplishes two goals:

1. Setups Integration.delete to properly cascade through
HybridCloudForeignKey @leeandher
2.  Uses outboxes to perform the syncing @dashed 
3. Incidentally breaks the ProjectIntegration.integration foreign key,
which needs to be broken anyways.

It is understood that ProjectIntegration isn't currently being used and
is a deprecated model, but the reality is that it is a model and table
that exists, and thus its foreign key must be broken as part of our
hybrid cloud work. It also is a good target for 1 and 2 above because
breaking it can't have any production impact otherwise.

migrations_lockfile.txt

@@ -6,5 +6,5 @@ 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.
 
 nodestore: 0002_nodestore_no_dictfield
-sentry: 0361_monitor_environment
+sentry: 0362_break_project_integration_fk
 social_auth: 0001_initial

src/sentry/db/postgres/roles.py

@@ -8,7 +8,7 @@ from django.db.transaction import get_connection
 
 
 @contextlib.contextmanager
-def test_psql_role_override(role_name: str, using: str | None = None):
+def in_test_psql_role_override(role_name: str, using: str | None = None):
     """
     During test runs, the role of the current connection will be swapped with role_name, and then swapped
     back to its original.  Has no effect in production.

src/sentry/migrations/0362_break_project_integration_fk.py

@@ -0,0 +1,61 @@
+# Generated by Django 2.2.28 on 2023-02-22 21:05
+
+from django.db import migrations
+
+import sentry.db.models.fields.hybrid_cloud_foreign_key
+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. For
+    # the most part, 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 dangerous:
+    # - Large data migrations. Typically we want these to be run manually by ops 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
+    #   have ops run this and not block the deploy. Note that while adding an index is a schema
+    #   change, it's completely safe to run the operation after the code has deployed.
+    is_dangerous = False
+
+    dependencies = [
+        ("sentry", "0361_monitor_environment"),
+    ]
+
+    database_operations = [
+        migrations.AlterField(
+            model_name="projectintegration",
+            name="integration",
+            field=sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                to="sentry.Integration", db_constraint=False, db_index=True
+            ),
+        ),
+    ]
+
+    state_operations = [
+        migrations.AlterField(
+            model_name="projectintegration",
+            name="integration",
+            field=sentry.db.models.fields.hybrid_cloud_foreign_key.HybridCloudForeignKey(
+                "sentry.Integration", db_index=True, on_delete="CASCADE"
+            ),
+        ),
+        migrations.RenameField(
+            model_name="projectintegration",
+            old_name="integration",
+            new_name="integration_id",
+        ),
+        migrations.AlterUniqueTogether(
+            name="projectintegration",
+            unique_together={("project", "integration_id")},
+        ),
+        migrations.RemoveField(
+            model_name="integration",
+            name="projects",
+        ),
+    ]
+
+    operations = database_operations + [
+        migrations.SeparateDatabaseAndState(state_operations=state_operations)
+    ]

src/sentry/models/counter.py

@@ -13,7 +13,7 @@ from sentry.db.models import (
     region_silo_only_model,
     sane_repr,
 )
-from sentry.db.postgres.roles import test_psql_role_override
+from sentry.db.postgres.roles import in_test_psql_role_override
 
 
 @region_silo_only_model
@@ -99,7 +99,7 @@ def create_counter_function(app_config, using, **kwargs):
     if not get_model_if_available(app_config, "Counter"):
         return
 
-    with test_psql_role_override("postgres", using), connections[using].cursor() as cursor:
+    with in_test_psql_role_override("postgres", using), connections[using].cursor() as cursor:
         cursor.execute(
             """
             create or replace function sentry_increment_project_counter(

src/sentry/models/integrations/integration.py

@@ -1,7 +1,7 @@
 import logging
-from typing import Any, Sequence
+from typing import Any, List, Sequence
 
-from django.db import IntegrityError, models
+from django.db import IntegrityError, models, transaction
 
 from sentry.constants import ObjectStatus
 from sentry.db.models import (
@@ -11,8 +11,9 @@ from sentry.db.models import (
 )
 from sentry.db.models.fields.jsonfield import JSONField
 from sentry.db.models.manager import BaseManager
+from sentry.db.postgres.roles import in_test_psql_role_override
 from sentry.models.integrations.organization_integration import OrganizationIntegration
-from sentry.models.integrations.project_integration import ProjectIntegration
+from sentry.models.outbox import ControlOutbox, OutboxCategory, OutboxScope, find_regions_for_orgs
 from sentry.signals import integration_added
 
 logger = logging.getLogger(__name__)
@@ -34,9 +35,6 @@ class Integration(DefaultFieldsModel):
     organizations = models.ManyToManyField(
         "sentry.Organization", related_name="integrations", through=OrganizationIntegration
     )
-    projects = models.ManyToManyField(
-        "sentry.Project", related_name="integrations", through=ProjectIntegration
-    )
     provider = models.CharField(max_length=64)
     external_id = models.CharField(max_length=64)
     name = models.CharField(max_length=200)
@@ -60,6 +58,28 @@ class Integration(DefaultFieldsModel):
 
         return integrations.get(self.provider)
 
+    def delete(self, *args, **kwds):
+        with transaction.atomic(), in_test_psql_role_override("postgres"):
+            for outbox in Integration.outboxes_for_update(self.id):
+                outbox.save()
+            return super().delete(*args, **kwds)
+
+    @staticmethod
+    def outboxes_for_update(identifier: int) -> List[ControlOutbox]:
+        org_ids: List[int] = OrganizationIntegration.objects.filter(
+            integration_id=identifier
+        ).values_list("organization_id", flat=True)
+        return [
+            ControlOutbox(
+                shard_scope=OutboxScope.INTEGRATION_SCOPE,
+                shard_identifier=identifier,
+                object_identifier=identifier,
+                category=OutboxCategory.INTEGRATION_UPDATE,
+                region_name=region_name,
+            )
+            for region_name in find_regions_for_orgs(org_ids)
+        ]
+
     def get_installation(self, organization_id: int, **kwargs: Any) -> Any:
         return self.get_provider().get_installation(self, organization_id, **kwargs)
 

src/sentry/models/integrations/project_integration.py

@@ -1,4 +1,5 @@
 from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.db.models.fields.jsonfield import JSONField
 
 
@@ -12,10 +13,10 @@ class ProjectIntegration(Model):
     __include_in_export__ = False
 
     project = FlexibleForeignKey("sentry.Project")
-    integration = FlexibleForeignKey("sentry.Integration")
+    integration_id = HybridCloudForeignKey("sentry.Integration", on_delete="CASCADE")
     config = JSONField(default=dict)
 
     class Meta:
         app_label = "sentry"
         db_table = "sentry_projectintegration"
-        unique_together = (("project", "integration"),)
+        unique_together = (("project", "integration_id"),)

src/sentry/models/outbox.py

@@ -35,6 +35,7 @@ class OutboxScope(IntEnum):
     WEBHOOK_SCOPE = 2
     AUDIT_LOG_SCOPE = 3
     USER_IP_SCOPE = 4
+    INTEGRATION_SCOPE = 5
 
     def __str__(self):
         return self.name
@@ -52,6 +53,7 @@ class OutboxCategory(IntEnum):
     VERIFY_ORGANIZATION_MAPPING = 4
     AUDIT_LOG_EVENT = 5
     USER_IP_EVENT = 6
+    INTEGRATION_UPDATE = 7
 
     @classmethod
     def as_choices(cls):
@@ -342,18 +344,9 @@ def _find_orgs_for_user(user_id: int) -> Set[int]:
     }
 
 
-def find_regions_for_user(user_id: int) -> Set[str]:
+def find_regions_for_orgs(org_ids: Iterable[int]) -> Set[str]:
     from sentry.models import OrganizationMapping
 
-    org_ids: Set[int]
-    if "pytest" in sys.modules:
-        from sentry.testutils.silo import exempt_from_silo_limits
-
-        with exempt_from_silo_limits():
-            org_ids = _find_orgs_for_user(user_id)
-    else:
-        org_ids = _find_orgs_for_user(user_id)
-
     if SiloMode.get_current_mode() == SiloMode.MONOLITH:
         return {
             MONOLITH_REGION_NAME,
@@ -367,6 +360,19 @@ def find_regions_for_user(user_id: int) -> Set[str]:
         }
 
 
+def find_regions_for_user(user_id: int) -> Set[str]:
+    org_ids: Set[int]
+    if "pytest" in sys.modules:
+        from sentry.testutils.silo import exempt_from_silo_limits
+
+        with exempt_from_silo_limits():
+            org_ids = _find_orgs_for_user(user_id)
+    else:
+        org_ids = _find_orgs_for_user(user_id)
+
+    return find_regions_for_orgs(org_ids)
+
+
 def outbox_silo_modes() -> List[SiloMode]:
     cur = SiloMode.get_current_mode()
     result: List[SiloMode] = []

src/sentry/models/user.py

@@ -23,7 +23,7 @@ from sentry.db.models import (
     control_silo_only_model,
     sane_repr,
 )
-from sentry.db.postgres.roles import test_psql_role_override
+from sentry.db.postgres.roles import in_test_psql_role_override
 from sentry.models import LostPasswordHash
 from sentry.models.authenticator import Authenticator
 from sentry.models.outbox import ControlOutbox, OutboxCategory, OutboxScope, find_regions_for_user
@@ -200,7 +200,7 @@ class User(BaseModel, AbstractBaseUser):
     def delete(self):
         if self.username == "sentry":
             raise Exception('You cannot delete the "sentry" user as it is required by Sentry.')
-        with transaction.atomic(), test_psql_role_override("postgres"):
+        with transaction.atomic(), in_test_psql_role_override("postgres"):
             avatar = self.avatar.first()
             if avatar:
                 avatar.delete()

src/sentry/receivers/outbox/control.py

@@ -4,7 +4,7 @@ from typing import Any
 
 from django.dispatch import receiver
 
-from sentry.models import OutboxCategory, User, process_control_outbox
+from sentry.models import Integration, OutboxCategory, User, process_control_outbox
 from sentry.receivers.outbox import maybe_process_tombstone
 
 
@@ -13,3 +13,10 @@ def process_user_updates(object_identifier: int, **kwds: Any):
     if (user := maybe_process_tombstone(User, object_identifier)) is None:
         return
     user  # Currently we do not sync any other user changes, but if we did, you can use this variable.
+
+
+@receiver(process_control_outbox, sender=OutboxCategory.INTEGRATION_UPDATE)
+def process_integration_updates(object_identifier: int, **kwds: Any):
+    if (integration := maybe_process_tombstone(Integration, object_identifier)) is None:
+        return
+    integration  # Currently we do not sync any other integration changes, but if we did, you can use this variable.

tests/conftest.py

@@ -163,15 +163,19 @@ def protect_hybrid_cloud_deletions(request):
     logic to see how to escalate the connection's role in tests.  Make absolutely sure that you
     create Outbox objects in the same transaction that matches what you delete.
     """
-    if "django_db_setup" not in request.fixturenames:
-        yield
-        return
-
     from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
     from sentry.testutils.silo import iter_models, reset_test_role, restrict_role
 
-    with get_connection().cursor() as conn:
-        conn.execute("SET ROLE 'postgres'")
+    try:
+        with get_connection().cursor() as conn:
+            conn.execute("SET ROLE 'postgres'")
+    except (RuntimeError, AssertionError) as e:
+        # Tests that do not have access to the database should pass through.
+        # Ideally we'd use request.fixture names to infer this, but there didn't seem to be a single stable
+        # fixture name that fully covered all cases of database access, so this approach is "try and then fail".
+        if "Database access not allowed" in str(e) or "Database queries to" in str(e):
+            yield
+            return
 
     reset_test_role(role="postgres_unprivileged")