ref(db): Refactor EnvironmentProject to reduce rollbacks (#89265)

Similar to how it was done in:
#88885

Refactor of `add_project` method of `Environment` model, preparing it
for gradual rollout and to decrease the number of rollbacks it does
since almost every attempt of insertions for this model results in a
rollback.

In this method `EnvironmentProject` was doing overly optimistic inserts
leading to us having almost 100 rollbacks/second coming just from this
model


## Why are we doing this

Currently we are doing around 300 rollbacks per second, mostly caused by
overly optimistic writes - almost all of the writes result in the
rollback because the data already exists in the table, and for those
occasions `get_or_create` is more suitable since SELECT statement is
more performant than ROLLBACK when they happen most of the times.

[Datadog notebok with
investagtion](https://app.datadoghq.com/notebook/12067672/postgres-rollback-investigation?range=604800000&start=1743184480708&live=true),
where 3 problematic models where detected:
- ~GroupRelease~
- `Commit`
- `EnvironmentProject`

Models will be refactored one at the time, and the refactor will be
rolled out gradually: 10% - 50% - 100%

Author: vgrozdanic

src/sentry/models/environment.py

@@ -15,6 +15,7 @@
     sane_repr,
 )
 from sentry.db.models.manager.base import BaseManager
+from sentry.options.rollout import in_random_rollout
 from sentry.utils import metrics
 from sentry.utils.cache import cache
 from sentry.utils.hashlib import md5_text
@@ -111,16 +112,23 @@ def add_project(self, project, is_hidden=None):
         cache_key = f"envproj:c:{self.id}:{project.id}"
 
         if cache.get(cache_key) is None:
-            try:
-                with transaction.atomic(router.db_for_write(EnvironmentProject)):
-                    EnvironmentProject.objects.create(
-                        project=project, environment=self, is_hidden=is_hidden
-                    )
-                cache.set(cache_key, 1, 3600)
-            except IntegrityError:
-                incr_rollback_metrics(EnvironmentProject)
-                # We've already created the object, should still cache the action.
+            if in_random_rollout("environmentproject.new_add_project.rollout"):
+                EnvironmentProject.objects.get_or_create(
+                    project=project, environment=self, defaults={"is_hidden": is_hidden}
+                )
+                # The object already exists, we cache the action to reduce the load on the database.
                 cache.set(cache_key, 1, 3600)
+            else:
+                try:
+                    with transaction.atomic(router.db_for_write(EnvironmentProject)):
+                        EnvironmentProject.objects.create(
+                            project=project, environment=self, is_hidden=is_hidden
+                        )
+                    cache.set(cache_key, 1, 3600)
+                except IntegrityError:
+                    incr_rollback_metrics(EnvironmentProject)
+                    # We've already created the object, should still cache the action.
+                    cache.set(cache_key, 1, 3600)
 
     @staticmethod
     def get_name_from_path_segment(segment):

src/sentry/options/defaults.py

@@ -2809,6 +2809,14 @@
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+# Killswitch for EnvironmentProject.get_or_create refactor
+register(
+    "environmentproject.new_add_project.rollout",
+    default=0.0,
+    type=Float,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # Killswitch for Postgres query timeout error handling
 register(
     "api.postgres-query-timeout-error-handling.enabled",

tests/sentry/models/test_environment.py

@@ -2,6 +2,7 @@
 
 from sentry.models.environment import Environment
 from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers import override_options
 
 
 class GetOrCreateTest(TestCase):
@@ -23,6 +24,30 @@ def test_simple(self):
         with self.assertNumQueries(0):
             assert Environment.get_for_organization_id(project.organization_id, "prod").id == env.id
 
+    @override_options({"environmentproject.new_add_project.rollout": 1.0})
+    def test_simple_with_new_option(self):
+        """
+        Same test as test_simple, but with the new option enabled which has refactored
+        the get_or_create method.
+        Note (vgrozdanic): this test will be removed after the rollout is complete.
+        """
+        project = self.create_project()
+
+        with pytest.raises(Environment.DoesNotExist):
+            Environment.get_for_organization_id(project.organization_id, "prod")
+
+        env = Environment.get_or_create(project=project, name="prod")
+
+        assert env.name == "prod"
+        assert env.projects.first().id == project.id
+
+        env2 = Environment.get_or_create(project=project, name="prod")
+
+        assert env2.id == env.id
+
+        with self.assertNumQueries(0):
+            assert Environment.get_for_organization_id(project.organization_id, "prod").id == env.id
+
 
 @pytest.mark.parametrize(
     "val,expected",