feat(hybridcloud): Bucket webhook mailboxes on the key, not on volume

src/sentry/integrations/middleware/hybrid_cloud/parser.py

@@ -2,17 +2,16 @@
 
 import logging
 from abc import ABC
+from collections.abc import Mapping
 from concurrent.futures import as_completed
 from typing import TYPE_CHECKING, Any, ClassVar
 
 import orjson
-from django.core.cache import cache
 from django.http import HttpRequest, HttpResponse
 from django.http.response import HttpResponseBase
 from django.urls import ResolverMatch, resolve
 from rest_framework import status
 
-from sentry.api.base import ONE_DAY
 from sentry.constants import ObjectStatus
 from sentry.hybridcloud.models.webhookpayload import DestinationType, WebhookPayload
 from sentry.hybridcloud.outbox.category import WebhookProviderIdentifier
@@ -30,12 +29,12 @@
 from sentry.integrations.services.integration.model import RpcIntegration
 from sentry.killswitches import get_killswitch_value, value_matches
 from sentry.logging.handlers import SamplingFilter
-from sentry.ratelimits import backend as ratelimiter
 from sentry.silo.base import SiloLimit, SiloMode
 from sentry.silo.client import CellSiloClient, SiloClientError
 from sentry.types.cell import Cell, find_cells_for_orgs, get_cell_by_name
 from sentry.utils import metrics
 from sentry.utils.concurrent import ContextPropagatingThreadPoolExecutor
+from sentry.utils.safe import get_path
 
 logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
@@ -81,13 +80,14 @@ class BaseRequestParser(ABC):
     mailbox_bucket_count: ClassVar[int] = 100
     """How many sub-mailboxes `mailbox_bucket_id` is spread over.
 
-    Every mailbox costs a scheduler row and a dispatch slot, so splitting past what
-    the volume needs buys queue rows rather than parallelism.
-    """
+    Set it from the key's cardinality. A repeating key (a repository) coalesces
+    payloads onto one mailbox per value, so a high count is free. A key that barely
+    repeats (an issue) fills a bucket once and never returns, and every mailbox costs
+    a scheduler row and a dispatch slot.
 
-    always_bucket: ClassVar[bool] = False
-    """Split every integration's mailbox by `mailbox_bucket_id` instead of waiting
-    for it to exceed the hourly rate limit first."""
+    A static count is the interim answer; CW-1987 sizes it from a rolling per-
+    integration rate instead. https://linear.app/getsentry/issue/CW-1987
+    """
 
     def __init__(self, request: HttpRequest, response_handler: ResponseHandler):
         self.request = request
@@ -298,12 +298,8 @@ def _bucketed_mailbox_identifier(
     ) -> str:
         """The mailbox identifier up to the bucket, before any event-type suffix.
 
-        Falls back to the integration-level mailbox when the integration is below
-        the volume that warrants buckets, or when no bucket ID is available."""
-        if not self.always_bucket and not self._exceeds_bucketing_volume(integration):
-            self._record_mailbox_routing(bucketed=False, reason="under_volume_gate")
-            return str(integration.id)
-
+        A payload carrying a bucket key is bucketed; one without falls back to the
+        integration-level mailbox."""
         mailbox_bucket_id = self.mailbox_bucket_id(data)
         if mailbox_bucket_id is None:
             self._record_mailbox_routing(bucketed=False, reason="no_bucket_key")
@@ -314,20 +310,6 @@ def _bucketed_mailbox_identifier(
 
         return f"{integration.id}:{bucket_number}"
 
-    def _exceeds_bucketing_volume(self, integration: RpcIntegration | Integration) -> bool:
-        # If we get fewer than 3000 in 1 hour we don't need to split into buckets
-        ratelimit_key = f"webhookpayload:{self.provider}:{integration.id}"
-        use_buckets_key = f"{ratelimit_key}:use_buckets"
-
-        if cache.get(use_buckets_key):
-            return True
-        if ratelimiter.is_limited(key=ratelimit_key, window=60 * 60, limit=3000):
-            # Once we have gone over the rate limit in a day, we use smaller
-            # buckets for the next day.
-            cache.set(use_buckets_key, 1, timeout=ONE_DAY)
-            return True
-        return False
-
     def _record_mailbox_routing(self, bucketed: bool, reason: str) -> None:
         """`reason` is the full breakdown; `bucketed` stays for the dashboards on it."""
         metrics.incr(
@@ -344,6 +326,18 @@ def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None:
             "You must implement mailbox_bucket_id to use bucketed identifiers"
         )
 
+    @staticmethod
+    def bucket_key_at(data: Mapping[str, Any], *path: str) -> int | None:
+        """Read a bucket key out of `data`, or None if it is missing or not numeric.
+
+        Shared so every provider degrades the same way rather than raising out of the
+        parser on a body it did not expect.
+        """
+        try:
+            return int(get_path(data, *path))
+        except (TypeError, ValueError):
+            return None
+
     def _mailbox_event_type(self, data: dict[str, Any]) -> str | None:
         """Validation lives here, not in the subclass: the discriminator comes out of
         a body control has not verified — gitlab and bitbucket resolve their handlers

src/sentry/middleware/integrations/parsers/github.py

@@ -66,24 +66,15 @@ class GithubRequestParser(BaseRequestParser):
     webhook_identifier = WebhookProviderIdentifier.GITHUB
     webhook_endpoint: Any = GitHubIntegrationsWebhookEndpoint
     """Overridden in GithubEnterpriseRequestParser"""
-    always_bucket = True
 
     def _get_external_id(self, event: Mapping[str, Any]) -> str | None:
         """Overridden in GithubEnterpriseRequestParser"""
         return get_github_external_id(event)
 
     def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None:
-        """Hash on repository ID to distribute webhooks across sub-mailboxes.
-
-        GitHub webhook payloads include repository.id for most event types.
-        Installation events are routed to control silo and don't reach this path.
-        """
-        repository = data.get("repository")
-        if isinstance(repository, dict):
-            repo_id = repository.get("id")
-            if isinstance(repo_id, int):
-                return repo_id
-        return None
+        """Every event type that reaches a cell carries `repository.id`; installation
+        events are handled on control and never get here."""
+        return self.bucket_key_at(data, "repository", "id")
 
     def mailbox_event_type(self, data: Mapping[str, Any]) -> str | None:
         return self.request.META.get(GITHUB_WEBHOOK_TYPE_HEADER)

src/sentry/middleware/integrations/parsers/gitlab.py

@@ -83,15 +83,8 @@ def get_response_from_gitlab_webhook(self) -> HttpResponseBase:
         )
 
     def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None:
-        """
-        Used by get_mailbox_identifier to find the project.id a payload is for.
-        In high volume gitlab instances we shard messages by project for greater
-        delivery throughput.
-        """
-        project_id = data.get("project", {}).get("id", None)
-        if not project_id:
-            return None
-        return project_id
+        """Every event kind a cell processes names the project it belongs to."""
+        return self.bucket_key_at(data, "project", "id")
 
     def mailbox_event_type(self, data: Mapping[str, Any]) -> str | None:
         """Reads the body's `object_kind`, not the `X-Gitlab-Event` header the

src/sentry/middleware/integrations/parsers/jira.py

@@ -28,7 +28,6 @@
     parse_integration_from_request,
 )
 from sentry.shared_integrations.exceptions import ApiError
-from sentry.utils.safe import get_path
 
 logger = logging.getLogger(__name__)
 
@@ -37,8 +36,6 @@ class JiraRequestParser(BaseRequestParser):
     provider = IntegrationProviderSlug.JIRA.value
     webhook_identifier = WebhookProviderIdentifier.JIRA
 
-    # Far lower volume than GitHub: enough to unserialize a burst without thinning
-    # mailboxes into scheduler rows that each carry a handful of payloads.
     mailbox_bucket_count = 10
 
     control_classes = [
@@ -97,7 +94,4 @@ def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None:
         """The Connect descriptor registers only `jira:issue_updated`, so the issue is
         the only axis a Jira mailbox can be split on.
         """
-        try:
-            return int(get_path(data, "issue", "id"))
-        except (TypeError, ValueError):
-            return None
+        return self.bucket_key_at(data, "issue", "id")

src/sentry/middleware/integrations/parsers/jira_server.py

@@ -23,6 +23,8 @@ class JiraServerRequestParser(BaseRequestParser):
     provider = IntegrationProviderSlug.JIRA_SERVER.value
     webhook_identifier = WebhookProviderIdentifier.JIRA_SERVER
 
+    mailbox_bucket_count = 10
+
     def get_response_from_issue_update_webhook(self) -> HttpResponseBase:
         token = self.match.kwargs.get("token")
         try:
@@ -52,18 +54,8 @@ def get_response_from_issue_update_webhook(self) -> HttpResponseBase:
         )
 
     def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None:
-        """
-        Used by get_mailbox_identifier to find the issue.id a payload is for.
-        In high volume jira_server instances we shard messages by issue for greater
-        delivery throughput.
-        """
-        issue_id = data.get("issue", {}).get("id", None)
-        if not issue_id:
-            return None
-        try:
-            return int(issue_id)
-        except ValueError:
-            return None
+        """Only changelog webhooks reach a cell, and each names its issue."""
+        return self.bucket_key_at(data, "issue", "id")
 
     def get_response(self) -> HttpResponseBase:
         if self.view_class == JiraServerIssueUpdatedWebhook:

src/sentry/middleware/integrations/parsers/vsts.py

@@ -14,7 +14,6 @@
 from sentry.integrations.types import IntegrationProviderSlug
 from sentry.integrations.vsts.webhooks import WorkItemWebhook, get_vsts_external_id
 from sentry.silo.base import control_silo_function
-from sentry.utils.safe import get_path
 
 logger = logging.getLogger(__name__)
 
@@ -23,8 +22,6 @@ class VstsRequestParser(BaseRequestParser):
     provider = IntegrationProviderSlug.AZURE_DEVOPS.value
     webhook_identifier = WebhookProviderIdentifier.VSTS
 
-    # Far lower volume than GitHub: enough to unserialize a burst without thinning
-    # mailboxes into scheduler rows that each carry a handful of payloads.
     mailbox_bucket_count = 10
 
     cell_view_classes = [WorkItemWebhook]
@@ -65,7 +62,4 @@ def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None:
         """The subscription is created for `workitem.updated` only, so the work item
         is the only axis a VSTS mailbox can be split on.
         """
-        try:
-            return int(get_path(data, "resource", "workItemId"))
-        except (TypeError, ValueError):
-            return None
+        return self.bucket_key_at(data, "resource", "workItemId")

tests/sentry/integrations/middleware/hybrid_cloud/test_base.py

@@ -3,7 +3,6 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
-from django.core.cache import cache
 from django.http import HttpResponse
 from django.test import RequestFactory, override_settings
 from pytest import raises
@@ -143,7 +142,7 @@ class MockParser(BaseRequestParser):
             assert payload.request_method
             assert payload.destination_type == DestinationType.SENTRY_CELL
 
-    def test_get_mailbox_identifier_buckets_only_above_volume(self) -> None:
+    def test_get_mailbox_identifier_buckets_whenever_a_key_exists(self) -> None:
         class BucketedParser(ExampleRequestParser):
             def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None:
                 return 177
@@ -153,34 +152,21 @@ def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None:
         )
         parser = BucketedParser(self.request, self.response_handler)
 
-        with patch(
-            "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited",
-            return_value=False,
-        ):
-            assert parser.get_mailbox_identifier(integration, {}) == str(integration.id)
-        with patch(
-            "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited",
-            return_value=True,
-        ):
-            assert parser.get_mailbox_identifier(integration, {}) == f"{integration.id}:77"
-
-    def test_get_mailbox_identifier_always_bucket_skips_volume_check(self) -> None:
-        class AlwaysBucketedParser(ExampleRequestParser):
-            always_bucket = True
+        assert parser.get_mailbox_identifier(integration, {}) == f"{integration.id}:77"
 
-            def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None:
-                return 177
+    def test_bucket_key_at_coerces_or_falls_back(self) -> None:
+        at = BaseRequestParser.bucket_key_at
 
-        integration = self.create_integration(
-            organization=self.organization, external_id="1", provider="test_provider"
-        )
-        parser = AlwaysBucketedParser(self.request, self.response_handler)
+        assert at({"issue": {"id": 10237}}, "issue", "id") == 10237
+        assert at({"issue": {"id": "10237"}}, "issue", "id") == 10237
 
-        with patch(
-            "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited"
-        ) as mock_is_limited:
-            assert parser.get_mailbox_identifier(integration, {}) == f"{integration.id}:77"
-        mock_is_limited.assert_not_called()
+        # Anything unusable falls back rather than raising at the modulo.
+        assert at({}, "issue", "id") is None
+        assert at({"issue": {}}, "issue", "id") is None
+        assert at({"issue": "PROJ-1"}, "issue", "id") is None
+        assert at({"issue": {"id": None}}, "issue", "id") is None
+        assert at({"issue": {"id": "not-a-number"}}, "issue", "id") is None
+        assert at({"issue": {"id": ["10237"]}}, "issue", "id") is None
 
     @override_settings(SILO_MODE=SiloMode.CONTROL)
     @patch("sentry.integrations.middleware.hybrid_cloud.parser.maybe_trigger_drain")
@@ -387,28 +373,12 @@ def test_get_organizations_from_integration_missing_org_integration(
         assert mock_record.call_count == 2
         assert_halt_metric(mock_record, MiddlewareHaltReason.ORG_INTEGRATION_DOES_NOT_EXIST)
 
-    @override_settings(SILO_MODE=SiloMode.CONTROL)
-    @patch("sentry.integrations.middleware.hybrid_cloud.parser.metrics.incr")
-    def test_mailbox_identifier_under_volume_gate(self, mock_incr: MagicMock) -> None:
-        integration = self.create_integration(
-            organization=self.organization, provider="test_provider", external_id="test_external_id"
-        )
-        parser = BucketingRequestParser(self.request, self.response_handler)
-
-        assert parser.get_mailbox_identifier(integration, {"bucket_id": 101}) == str(integration.id)
-
-        mock_incr.assert_any_call(
-            "hybridcloud.webhookpayload.mailbox_routing",
-            tags={"provider": "test_provider", "bucketed": "false", "reason": "under_volume_gate"},
-        )
-
     @override_settings(SILO_MODE=SiloMode.CONTROL)
     @patch("sentry.integrations.middleware.hybrid_cloud.parser.metrics.incr")
     def test_mailbox_identifier_without_a_bucket_key(self, mock_incr: MagicMock) -> None:
         integration = self.create_integration(
             organization=self.organization, provider="test_provider", external_id="test_external_id"
         )
-        cache.set(f"webhookpayload:test_provider:{integration.id}:use_buckets", 1)
         parser = BucketingRequestParser(self.request, self.response_handler)
 
         assert parser.get_mailbox_identifier(integration, {}) == str(integration.id)
@@ -424,7 +394,6 @@ def test_mailbox_identifier_bucketed(self, mock_incr: MagicMock) -> None:
         integration = self.create_integration(
             organization=self.organization, provider="test_provider", external_id="test_external_id"
         )
-        cache.set(f"webhookpayload:test_provider:{integration.id}:use_buckets", 1)
         parser = BucketingRequestParser(self.request, self.response_handler)
 
         assert (

tests/sentry/middleware/integrations/parsers/test_github.py

@@ -256,7 +256,7 @@ def test_issue_deleted_routing(self) -> None:
                 "installation": {"id": "1"},
                 "issue": {"id": "1"},
                 "action": "deleted",
-                "repository": {"id": "1"},
+                "repository": {"id": 1},
             },
             content_type="application/json",
             headers={"X-GITHUB-EVENT": GithubWebhookType.ISSUE.value},
@@ -270,7 +270,7 @@ def test_issue_deleted_routing(self) -> None:
         assert len(responses.calls) == 0
         assert_webhook_payloads_for_mailbox(
             request=request,
-            mailbox_name=f"github:{integration.id}:issues",
+            mailbox_name=f"github:{integration.id}:1:issues",
             cell_names=[cell.name],
             destination_types={DestinationType.SENTRY_CELL: 1},
         )