perf(relay): Avoid N+1 project queries when debouncing config invalidations

src/sentry/tasks/relay.py

@@ -426,6 +426,20 @@ def _schedule_invalidate_project_config(
             check_debounce_keys["project_id"] = proj_id
             check_debounce_keys["organization_id"] = org_id
     elif project_id:
+        # Check the project-level debounce key first: if an invalidation for this
+        # project is already queued we can skip the organization lookup entirely.
+        # This avoids an N+1 query when many project options are written in a
+        # single transaction (e.g. during project creation).
+        with quiet_redis_noise():
+            if projectconfig_debounce_cache.invalidation.is_debounced(
+                public_key=None, project_id=project_id, organization_id=None
+            ):
+                metrics.incr(
+                    "relay.projectconfig_cache.skipped",
+                    tags={"reason": "debounce", "update_reason": trigger, "task": "invalidation"},
+                )
+                return
+
         try:
             (org_id,) = Project.objects.values_list("organization__id").get(id=project_id)
         except Project.DoesNotExist:

tests/sentry/tasks/test_relay.py

@@ -460,6 +460,38 @@ def apply_async(
             },
         ]
 
+    def test_debounced_project_skips_organization_lookup(
+        self,
+        default_project: Project,
+        invalidation_debounce_cache: RedisProjectConfigDebounceCache,
+        django_cache: object,
+    ) -> None:
+        tasks = []
+
+        def apply_async(
+            args: object = None, kwargs: object = None, countdown: object = None
+        ) -> None:
+            tasks.append(kwargs)
+
+        invalidation_debounce_cache.mark_task_done(
+            public_key=None, project_id=default_project.id, organization_id=None
+        )
+        with mock.patch("sentry.tasks.relay.invalidate_project_config.apply_async", apply_async):
+            # First call looks up the organization and schedules the task.
+            _schedule_invalidate_project_config(project_id=default_project.id, trigger="test")
+
+            # Subsequent calls for the same project are debounced at the project
+            # level before hitting the database.
+            with mock.patch.object(
+                Project.objects, "values_list", side_effect=AssertionError("unexpected query")
+            ):
+                for _ in range(5):
+                    _schedule_invalidate_project_config(
+                        project_id=default_project.id, trigger="test"
+                    )
+
+        assert len(tasks) == 1
+
     def test_invalidate(
         self,
         default_project: Project,