fix(hc): Fixing snuba tests in split db mode (#53403)

Get snuba tests passing in split db mode, and add stable test
decoration. These are mostly small, superficial adjustments to comply
with hybrid cloud.

src/sentry/api/helpers/group_index/update.py

@@ -532,7 +532,9 @@ def update_groups(
                     # TODO(dcramer): we need a solution for activity rollups
                     # before sending notifications on bulk changes
                     if not is_bulk:
-                        activity.send_notification()
+                        transaction.on_commit(
+                            lambda: activity.send_notification(), router.db_for_write(Group)
+                        )
 
             issue_resolved.send_robust(
                 organization_id=organization_id,

src/sentry/notifications/notifications/activity/base.py

@@ -96,7 +96,7 @@ class GroupActivityNotification(ActivityNotification, abc.ABC):
 
     def get_participants_with_group_subscription_reason(self) -> ParticipantMap:
         """This is overridden by the activity subclasses."""
-        return get_participants_for_group(self.group, self.user)
+        return get_participants_for_group(self.group, self.activity.user_id)
 
     def get_unsubscribe_key(self) -> tuple[str, int, str | None] | None:
         return "issue", self.group.id, None

src/sentry/notifications/utils/participants.py

@@ -126,7 +126,7 @@ class ParticipantMap:
 
 
 def get_providers_from_which_to_remove_user(
-    user: RpcUser,
+    user_id: int,
     participants_by_provider: ParticipantMap,
 ) -> set[ExternalProviders]:
     """
@@ -138,13 +138,13 @@ def get_providers_from_which_to_remove_user(
     providers = {
         provider
         for provider, participants in participants_by_provider.get_participant_sets()
-        if user.id in map(lambda p: int(p.id), participants)
+        if user_id in map(lambda p: int(p.id), participants)
     }
 
     if (
         get_option_from_list(
             user_option_service.get_many(
-                filter={"user_ids": [user.id], "keys": ["self_notifications"]}
+                filter={"user_ids": [user_id], "keys": ["self_notifications"]}
             ),
             key="self_notifications",
             default="0",
@@ -155,13 +155,13 @@ def get_providers_from_which_to_remove_user(
     return set()
 
 
-def get_participants_for_group(group: Group, user: RpcUser | None = None) -> ParticipantMap:
+def get_participants_for_group(group: Group, user_id: int | None = None) -> ParticipantMap:
     participants_by_provider: ParticipantMap = GroupSubscription.objects.get_participants(group)
-    if user:
+    if user_id:
         # Optionally remove the actor that created the activity from the recipients list.
-        providers = get_providers_from_which_to_remove_user(user, participants_by_provider)
+        providers = get_providers_from_which_to_remove_user(user_id, participants_by_provider)
         for provider in providers:
-            participants_by_provider.delete_participant_by_id(provider, ActorType.USER, user.id)
+            participants_by_provider.delete_participant_by_id(provider, ActorType.USER, user_id)
 
     return participants_by_provider
 

src/sentry/services/hybrid_cloud/__init__.py

@@ -347,6 +347,17 @@ def silo_mode_delegation(
     In split database mode, it will also inject DelegatedByOpenTransaction in for the monolith mode implementation.
     """
 
+    return cast(ServiceInterface, DelegatedBySiloMode(get_delegated_constructors(mapping)))
+
+
+def get_delegated_constructors(
+    mapping: Mapping[SiloMode, Callable[[], ServiceInterface]]
+) -> Mapping[SiloMode, Callable[[], ServiceInterface]]:
+    """
+    Creates a new constructor mapping by replacing the monolith constructor with a DelegatedByOpenTransaction
+    that intelligently selects the correct service implementation based on the call site.
+    """
+
     def delegator() -> ServiceInterface:
         from sentry.models import Organization, User
 
@@ -366,8 +377,7 @@ def silo_mode_delegation(
         SiloMode.MONOLITH: delegator,
         **({k: v for k, v in mapping.items() if k != SiloMode.MONOLITH}),
     }
-
-    return cast(ServiceInterface, DelegatedBySiloMode(final_mapping))
+    return final_mapping
 
 
 def coerce_id_from(m: object | int | None) -> int | None:

tests/sentry/event_manager/test_save_aggregate.py

@@ -3,13 +3,17 @@ import time
 from threading import Thread
 
 import pytest
+from django.db import router, transaction
 
 from sentry.event_manager import _save_aggregate
 from sentry.eventstore.models import Event
 from sentry.grouping.result import CalculatedHashes
+from sentry.models import GroupHash
+from sentry.testutils.silo import region_silo_test
+from sentry.utils.pytest.fixtures import django_db_all
 
 
-@pytest.mark.django_db(transaction=True)
+@django_db_all(transaction=True)
 @pytest.mark.parametrize(
     "is_race_free",
     [
@@ -27,6 +31,7 @@ from sentry.grouping.result import CalculatedHashes
         False,
     ],
 )
+@region_silo_test(stable=True)
 def test_group_creation_race(monkeypatch, default_project, is_race_free):
     CONCURRENCY = 2
 
@@ -48,27 +53,30 @@ def test_group_creation_race(monkeypatch, default_project, is_race_free):
     return_values = []
 
     def save_event():
-        data = {"timestamp": time.time()}
-        evt = Event(
-            default_project.id,
-            "89aeed6a472e4c5fb992d14df4d7e1b6",
-            data=data,
-        )
-        ret = _save_aggregate(
-            evt,
-            hashes=CalculatedHashes(
-                hashes=["a" * 32, "b" * 32],
-                hierarchical_hashes=[],
-                tree_labels=[],
-            ),
-            release=None,
-            metadata={},
-            received_timestamp=0,
-            level=10,
-            culprit="",
-        )
-        assert ret is not None
-        return_values.append(ret)
+        try:
+            data = {"timestamp": time.time()}
+            evt = Event(
+                default_project.id,
+                "89aeed6a472e4c5fb992d14df4d7e1b6",
+                data=data,
+            )
+            ret = _save_aggregate(
+                evt,
+                hashes=CalculatedHashes(
+                    hashes=["a" * 32, "b" * 32],
+                    hierarchical_hashes=[],
+                    tree_labels=[],
+                ),
+                release=None,
+                metadata={},
+                received_timestamp=0,
+                level=10,
+                culprit="",
+            )
+            assert ret is not None
+            return_values.append(ret)
+        finally:
+            transaction.get_connection(router.db_for_write(GroupHash)).close()
 
     threads = []
     for _ in range(CONCURRENCY):

tests/snuba/api/endpoints/test_organization_group_index.py

@@ -2031,7 +2031,7 @@ class GroupListTest(APITestCase, SnubaTestCase):
         )
 
 
-@region_silo_test
+@region_silo_test(stable=True)
 class GroupUpdateTest(APITestCase, SnubaTestCase):
     endpoint = "sentry-api-0-organization-group-index"
     method = "put"
@@ -2165,24 +2165,26 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
 
         org = self.organization
 
-        integration = Integration.objects.create(provider="example", name="Example")
-        integration.add_organization(org, self.user)
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            integration = Integration.objects.create(provider="example", name="Example")
+            integration.add_organization(org, self.user)
         event = self.store_event(
             data={"timestamp": iso_format(self.min_ago)}, project_id=self.project.id
         )
         group = event.group
 
-        OrganizationIntegration.objects.filter(
-            integration_id=integration.id, organization_id=group.organization.id
-        ).update(
-            config={
-                "sync_comments": True,
-                "sync_status_outbound": True,
-                "sync_status_inbound": True,
-                "sync_assignee_outbound": True,
-                "sync_assignee_inbound": True,
-            }
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            OrganizationIntegration.objects.filter(
+                integration_id=integration.id, organization_id=group.organization.id
+            ).update(
+                config={
+                    "sync_comments": True,
+                    "sync_status_outbound": True,
+                    "sync_status_inbound": True,
+                    "sync_assignee_outbound": True,
+                    "sync_assignee_inbound": True,
+                }
+            )
         external_issue = ExternalIssue.objects.get_or_create(
             organization_id=org.id, integration_id=integration.id, key="APP-%s" % group.id
         )[0]
@@ -2218,20 +2220,21 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
     def test_set_unresolved_with_integration(self, mock_sync_status_outbound):
         release = self.create_release(project=self.project, version="abc")
         group = self.create_group(status=GroupStatus.RESOLVED)
-        org = self.organization
-        integration = Integration.objects.create(provider="example", name="Example")
-        integration.add_organization(org, self.user)
-        OrganizationIntegration.objects.filter(
-            integration_id=integration.id, organization_id=group.organization.id
-        ).update(
-            config={
-                "sync_comments": True,
-                "sync_status_outbound": True,
-                "sync_status_inbound": True,
-                "sync_assignee_outbound": True,
-                "sync_assignee_inbound": True,
-            }
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            org = self.organization
+            integration = Integration.objects.create(provider="example", name="Example")
+            integration.add_organization(org, self.user)
+            OrganizationIntegration.objects.filter(
+                integration_id=integration.id, organization_id=group.organization.id
+            ).update(
+                config={
+                    "sync_comments": True,
+                    "sync_status_outbound": True,
+                    "sync_status_inbound": True,
+                    "sync_assignee_outbound": True,
+                    "sync_assignee_inbound": True,
+                }
+            )
         GroupResolution.objects.create(group=group, release=release)
         external_issue = ExternalIssue.objects.get_or_create(
             organization_id=org.id, integration_id=integration.id, key="APP-%s" % group.id
@@ -2271,9 +2274,10 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
         group = self.create_group(status=GroupStatus.UNRESOLVED)
         user = self.user
 
-        uo1 = UserOption.objects.create(
-            key="self_assign_issue", value="1", project_id=None, user=user
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1 = UserOption.objects.create(
+                key="self_assign_issue", value="1", project_id=None, user=user
+            )
 
         self.login_as(user=user)
         response = self.get_success_response(qs_params={"id": group.id}, status="resolved")
@@ -2287,7 +2291,8 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
             user_id=user.id, group=group, is_active=True
         ).exists()
 
-        uo1.delete()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1.delete()
 
     def test_self_assign_issue_next_release(self):
         release = Release.objects.create(organization_id=self.project.organization_id, version="a")
@@ -2295,9 +2300,10 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
 
         group = self.create_group(status=GroupStatus.UNRESOLVED)
 
-        uo1 = UserOption.objects.create(
-            key="self_assign_issue", value="1", project_id=None, user=self.user
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1 = UserOption.objects.create(
+                key="self_assign_issue", value="1", project_id=None, user=self.user
+            )
 
         self.login_as(user=self.user)
 
@@ -2322,7 +2328,8 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
             group=group, type=ActivityType.SET_RESOLVED_IN_RELEASE.value
         )
         assert activity.data["version"] == ""
-        uo1.delete()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1.delete()
 
     def test_in_semver_projects_group_resolution_stores_current_release_version(self):
         """
@@ -3361,7 +3368,7 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
         ).exists()
 
 
-@region_silo_test
+@region_silo_test(stable=True)
 class GroupDeleteTest(APITestCase, SnubaTestCase):
     endpoint = "sentry-api-0-organization-group-index"
     method = "delete"

tests/snuba/api/endpoints/test_project_group_index.py

@@ -31,15 +31,16 @@ from sentry.models import (
     UserOption,
 )
 from sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox
+from sentry.silo import SiloMode
 from sentry.testutils import APITestCase, SnubaTestCase
 from sentry.testutils.helpers import parse_link_header
 from sentry.testutils.helpers.datetime import before_now, iso_format
-from sentry.testutils.silo import region_silo_test
+from sentry.testutils.silo import assume_test_silo_mode, region_silo_test
 from sentry.types.activity import ActivityType
 from sentry.utils import json
 
 
-@region_silo_test
+@region_silo_test(stable=True)
 class GroupListTest(APITestCase, SnubaTestCase):
     def setUp(self):
         super().setUp()
@@ -338,7 +339,8 @@ class GroupListTest(APITestCase, SnubaTestCase):
         assert len(response.data) == 0
 
     def test_token_auth(self):
-        token = ApiToken.objects.create(user=self.user, scopes=256)
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            token = ApiToken.objects.create(user=self.user, scopes=256)
         response = self.client.get(
             self.path, format="json", HTTP_AUTHORIZATION=f"Bearer {token.token}"
         )
@@ -356,7 +358,7 @@ class GroupListTest(APITestCase, SnubaTestCase):
         assert [int(r["id"]) for r in response.data] == [event.group.id]
 
 
-@region_silo_test
+@region_silo_test(stable=True)
 class GroupUpdateTest(APITestCase, SnubaTestCase):
     def setUp(self):
         super().setUp()
@@ -440,21 +442,23 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
 
         org = self.organization
 
-        integration = Integration.objects.create(provider="example", name="Example")
-        integration.add_organization(org, self.user)
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            integration = Integration.objects.create(provider="example", name="Example")
+            integration.add_organization(org, self.user)
         group = self.create_group(status=GroupStatus.UNRESOLVED)
 
-        OrganizationIntegration.objects.filter(
-            integration_id=integration.id, organization_id=group.organization.id
-        ).update(
-            config={
-                "sync_comments": True,
-                "sync_status_outbound": True,
-                "sync_status_inbound": True,
-                "sync_assignee_outbound": True,
-                "sync_assignee_inbound": True,
-            }
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            OrganizationIntegration.objects.filter(
+                integration_id=integration.id, organization_id=group.organization.id
+            ).update(
+                config={
+                    "sync_comments": True,
+                    "sync_status_outbound": True,
+                    "sync_status_inbound": True,
+                    "sync_assignee_outbound": True,
+                    "sync_assignee_inbound": True,
+                }
+            )
         external_issue = ExternalIssue.objects.get_or_create(
             organization_id=org.id, integration_id=integration.id, key="APP-%s" % group.id
         )[0]
@@ -496,19 +500,20 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
         release = self.create_release(project=self.project, version="abc")
         group = self.create_group(status=GroupStatus.RESOLVED)
         org = self.organization
-        integration = Integration.objects.create(provider="example", name="Example")
-        integration.add_organization(org, self.user)
-        OrganizationIntegration.objects.filter(
-            integration_id=integration.id, organization_id=group.organization.id
-        ).update(
-            config={
-                "sync_comments": True,
-                "sync_status_outbound": True,
-                "sync_status_inbound": True,
-                "sync_assignee_outbound": True,
-                "sync_assignee_inbound": True,
-            }
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            integration = Integration.objects.create(provider="example", name="Example")
+            integration.add_organization(org, self.user)
+            OrganizationIntegration.objects.filter(
+                integration_id=integration.id, organization_id=group.organization.id
+            ).update(
+                config={
+                    "sync_comments": True,
+                    "sync_status_outbound": True,
+                    "sync_status_inbound": True,
+                    "sync_assignee_outbound": True,
+                    "sync_assignee_inbound": True,
+                }
+            )
         GroupResolution.objects.create(group=group, release=release)
         external_issue = ExternalIssue.objects.get_or_create(
             organization_id=org.id, integration_id=integration.id, key="APP-%s" % group.id
@@ -548,9 +553,10 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
         group = self.create_group(status=GroupStatus.UNRESOLVED)
         user = self.user
 
-        uo1 = UserOption.objects.create(
-            key="self_assign_issue", value="1", project_id=None, user=user
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1 = UserOption.objects.create(
+                key="self_assign_issue", value="1", project_id=None, user=user
+            )
 
         self.login_as(user=user)
         url = f"{self.path}?id={group.id}"
@@ -567,7 +573,8 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
             user_id=user.id, group=group, is_active=True
         ).exists()
 
-        uo1.delete()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1.delete()
 
     def test_self_assign_issue_next_release(self):
         release = Release.objects.create(organization_id=self.project.organization_id, version="a")
@@ -575,9 +582,10 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
 
         group = self.create_group(status=GroupStatus.UNRESOLVED)
 
-        uo1 = UserOption.objects.create(
-            key="self_assign_issue", value="1", project_id=None, user=self.user
-        )
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1 = UserOption.objects.create(
+                key="self_assign_issue", value="1", project_id=None, user=self.user
+            )
 
         self.login_as(user=self.user)
 
@@ -602,7 +610,8 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
             group=group, type=ActivityType.SET_RESOLVED_IN_RELEASE.value
         )
         assert activity.data["version"] == ""
-        uo1.delete()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            uo1.delete()
 
     def test_selective_status_update(self):
         group1 = self.create_group(status=GroupStatus.RESOLVED)
@@ -1347,7 +1356,7 @@ class GroupUpdateTest(APITestCase, SnubaTestCase):
         assert Group.objects.filter(id=group1.id).exists()
 
 
-@region_silo_test
+@region_silo_test(stable=True)
 class GroupDeleteTest(APITestCase, SnubaTestCase):
     @cached_property
     def path(self):