From a165788df59c88307c3b12f29e357591b1a095aa Mon Sep 17 00:00:00 2001 From: Ivan Dlugos Date: Wed, 2 Sep 2026 15:39:33 +0200 Subject: [PATCH 1/3] feat(hybridcloud): Bucket webhook mailboxes on the key, not on volume An integration was only split into sub-mailboxes once it exceeded 3,000 payloads in an hour, measured by a fixed-window Redis counter whose windows align to the clock hour. That counter resets on the hour, so the decision needed a second piece of state -- a `use_buckets` cache key with a one-day TTL -- to survive the reset, and a burst straddling the boundary evades it entirely: 2,900 payloads at 11:59 and 2,900 at 12:01 never exceed the limit in either window. It also fails open, routing unbucketed whenever Redis errors. GitHub already bypassed all of it with `always_bucket`, so unconditional bucketing is what the highest-volume forwarding provider has been doing all along. The bucket key now decides on its own: a payload carrying one is bucketed, one without falls back to the integration-level mailbox. That leaves the bucket count as the only knob, and it belongs to the key's cardinality rather than the provider's volume. A key that repeats across payloads -- a repository, a project -- already coalesces them onto one mailbox per distinct value, so 100 costs nothing. A key that barely repeats -- an issue, a work item -- puts one payload in a bucket and never returns to it, so a wide split buys shallow mailboxes that each still cost a scheduler row and a dispatch slot, and dispatch is the binding constraint rather than throughput. jira, jira_server and vsts take 10; github and gitlab keep 100. The five hand-rolled key readers disagreed about failure. github required an int and rejected a numeric string, gitlab returned the value uncoerced so a string id raised TypeError at the modulo, and jira_server caught ValueError but not TypeError. All five now read through `BaseRequestParser.bucket_key_at`, so a key that is missing, nested under a non-object, or not numeric falls back to the integration-level mailbox instead of raising out of the parser. Two routing changes follow. github buckets payloads whose `repository.id` arrives as a JSON string, which the isinstance check used to reject. gitlab and jira_server bucket every payload rather than only those past the gate. Refs CW-1887 --- .../middleware/hybrid_cloud/parser.py | 51 ++++++-------- .../middleware/integrations/parsers/github.py | 14 +--- .../middleware/integrations/parsers/gitlab.py | 11 +-- .../middleware/integrations/parsers/jira.py | 9 +-- .../integrations/parsers/jira_server.py | 17 ++--- .../middleware/integrations/parsers/vsts.py | 9 +-- .../middleware/hybrid_cloud/test_base.py | 58 ++++----------- .../integrations/parsers/test_github.py | 2 +- .../integrations/parsers/test_gitlab.py | 38 ++++------ .../integrations/parsers/test_jira.py | 4 -- .../integrations/parsers/test_jira_server.py | 70 +------------------ .../integrations/parsers/test_vsts.py | 17 ++--- 12 files changed, 74 insertions(+), 226 deletions(-) diff --git a/src/sentry/integrations/middleware/hybrid_cloud/parser.py b/src/sentry/integrations/middleware/hybrid_cloud/parser.py index c3c2d6a90f26..903532faf3ee 100644 --- a/src/sentry/integrations/middleware/hybrid_cloud/parser.py +++ b/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,14 +80,13 @@ 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. + Choose it from the bucket key's cardinality. A key that repeats across payloads -- + a repository, a project -- already coalesces them onto one mailbox per distinct + value, so a high count costs nothing. A key that barely repeats -- an issue, a + work item -- puts one payload in a bucket and never returns to it, so a high count + buys shallow mailboxes that each still cost 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.""" - def __init__(self, request: HttpRequest, response_handler: ResponseHandler): self.request = request self.match: ResolverMatch = resolve(self.request.path) @@ -298,12 +296,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) - + The bucket key is the gate: a payload that carries one is bucketed, and one + that does not 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 +308,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 +324,19 @@ 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 when it is absent or not numeric. + + Every provider coerces its key through here so they fail the same way: a key + that is missing, nested under a non-object, or not an integer falls back to + the integration-level mailbox instead of raising out of the parser. + """ + 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 diff --git a/src/sentry/middleware/integrations/parsers/github.py b/src/sentry/middleware/integrations/parsers/github.py index 2dd03b176912..ce9fd6e22e06 100644 --- a/src/sentry/middleware/integrations/parsers/github.py +++ b/src/sentry/middleware/integrations/parsers/github.py @@ -66,24 +66,16 @@ 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. + """Payloads carry `repository.id` for every event type that reaches a cell; + installation events are handled on control and never get here. """ - repository = data.get("repository") - if isinstance(repository, dict): - repo_id = repository.get("id") - if isinstance(repo_id, int): - return repo_id - return None + 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) diff --git a/src/sentry/middleware/integrations/parsers/gitlab.py b/src/sentry/middleware/integrations/parsers/gitlab.py index bd6ce4346697..f5b4e01a24df 100644 --- a/src/sentry/middleware/integrations/parsers/gitlab.py +++ b/src/sentry/middleware/integrations/parsers/gitlab.py @@ -83,15 +83,10 @@ def get_response_from_gitlab_webhook(self) -> HttpResponseBase: ) def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None: + """Every event kind a cell processes names the project it belongs to, so the + project is the axis a GitLab mailbox splits on. """ - 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 + 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 diff --git a/src/sentry/middleware/integrations/parsers/jira.py b/src/sentry/middleware/integrations/parsers/jira.py index 9bf8a5fc5a85..4cd486365d8b 100644 --- a/src/sentry/middleware/integrations/parsers/jira.py +++ b/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,7 @@ 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. + # `issue.id` barely repeats between payloads; see `mailbox_bucket_count`. mailbox_bucket_count = 10 control_classes = [ @@ -97,7 +95,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") diff --git a/src/sentry/middleware/integrations/parsers/jira_server.py b/src/sentry/middleware/integrations/parsers/jira_server.py index f484e8f03927..87aa925f07bb 100644 --- a/src/sentry/middleware/integrations/parsers/jira_server.py +++ b/src/sentry/middleware/integrations/parsers/jira_server.py @@ -23,6 +23,9 @@ class JiraServerRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.JIRA_SERVER.value webhook_identifier = WebhookProviderIdentifier.JIRA_SERVER + # `issue.id` barely repeats between payloads; see `mailbox_bucket_count`. + mailbox_bucket_count = 10 + def get_response_from_issue_update_webhook(self) -> HttpResponseBase: token = self.match.kwargs.get("token") try: @@ -52,18 +55,10 @@ def get_response_from_issue_update_webhook(self) -> HttpResponseBase: ) def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None: + """Only changelog webhooks reach a cell, and each names its issue, so the + issue is the axis a Jira Server mailbox splits on. """ - 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 + return self.bucket_key_at(data, "issue", "id") def get_response(self) -> HttpResponseBase: if self.view_class == JiraServerIssueUpdatedWebhook: diff --git a/src/sentry/middleware/integrations/parsers/vsts.py b/src/sentry/middleware/integrations/parsers/vsts.py index 6f1949b0ce26..4e259a9e1d53 100644 --- a/src/sentry/middleware/integrations/parsers/vsts.py +++ b/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,7 @@ 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. + # `resource.workItemId` barely repeats between payloads; see `mailbox_bucket_count`. mailbox_bucket_count = 10 cell_view_classes = [WorkItemWebhook] @@ -65,7 +63,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") diff --git a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py b/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py index 0c9b6e8add46..0e9f23ddd51b 100644 --- a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py +++ b/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,22 @@ 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() + # A key that is missing, nested under a non-object, or not a number leaves the + # payload on the integration-level mailbox 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 +374,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 +395,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 ( diff --git a/tests/sentry/middleware/integrations/parsers/test_github.py b/tests/sentry/middleware/integrations/parsers/test_github.py index 850f9994d71c..fa7099bbc66b 100644 --- a/tests/sentry/middleware/integrations/parsers/test_github.py +++ b/tests/sentry/middleware/integrations/parsers/test_github.py @@ -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}, ) diff --git a/tests/sentry/middleware/integrations/parsers/test_gitlab.py b/tests/sentry/middleware/integrations/parsers/test_gitlab.py index 92f5440fadeb..850a8681f1ba 100644 --- a/tests/sentry/middleware/integrations/parsers/test_gitlab.py +++ b/tests/sentry/middleware/integrations/parsers/test_gitlab.py @@ -1,5 +1,3 @@ -from unittest import mock - import responses from django.db import router, transaction from django.http import HttpRequest, HttpResponse @@ -133,7 +131,7 @@ def test_routing_webhook_properly_with_cells(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}:push", + mailbox_name=f"gitlab:{integration.id}:15:push", cell_names=[cell.name], ) @@ -157,7 +155,7 @@ def test_routing_webhook_ignores_an_unhandled_event_type(self) -> None: # An unvalidated suffix would put an arbitrary body value in the mailbox name. assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}", + mailbox_name=f"gitlab:{integration.id}:15", cell_names=[cell.name], ) @@ -204,15 +202,11 @@ def test_routing_webhook_properly_with_multiple_orgs(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}:push", + mailbox_name=f"gitlab:{integration.id}:15:push", cell_names=[cell.name], ) - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_buckets(self) -> None: - integration = self.get_integration() + def test_mailbox_bucket_id(self) -> None: request = self.factory.post( self.path, data=PUSH_EVENT, @@ -220,22 +214,14 @@ def test_routing_webhook_with_mailbox_buckets(self) -> None: HTTP_X_GITLAB_TOKEN=WEBHOOK_TOKEN, HTTP_X_GITLAB_EVENT="Push Hook", ) - with mock.patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited" - ) as mock_is_limited: - mock_is_limited.return_value = True - parser = GitlabRequestParser(request=request, response_handler=self.get_response) - response = parser.get_response() + parser = GitlabRequestParser(request=request, response_handler=self.get_response) - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - mailbox_name=f"gitlab:{integration.id}:15:push", - cell_names=[cell.name], - ) + assert parser.mailbox_bucket_id({"project": {"id": 15}}) == 15 + assert parser.mailbox_bucket_id({"project": {"id": "15"}}) == 15 + assert parser.mailbox_bucket_id({}) is None + assert parser.mailbox_bucket_id({"project": {}}) is None + assert parser.mailbox_bucket_id({"project": "sentry"}) is None + assert parser.mailbox_bucket_id({"project": {"id": "sentry"}}) is None @override_settings(SILO_MODE=SiloMode.CONTROL) @override_cells(cell_config) @@ -297,6 +283,6 @@ def test_webhook_outbox_creation(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}:push", + mailbox_name=f"gitlab:{integration.id}:15:push", cell_names=[cell.name], ) diff --git a/tests/sentry/middleware/integrations/parsers/test_jira.py b/tests/sentry/middleware/integrations/parsers/test_jira.py index 7d4fbb8c1908..aba9421a5204 100644 --- a/tests/sentry/middleware/integrations/parsers/test_jira.py +++ b/tests/sentry/middleware/integrations/parsers/test_jira.py @@ -3,7 +3,6 @@ from unittest.mock import patch import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory, override_settings from rest_framework import status @@ -148,8 +147,6 @@ def test_get_response_routing_to_cell_async(self) -> None: @override_cells(cell_config) def test_get_response_routing_to_cell_async_bucketed(self) -> None: integration = self.get_integration() - use_buckets_key = f"webhookpayload:jira:{integration.id}:use_buckets" - cache.set(use_buckets_key, 1) request = self.factory.post( path=f"{self.path_base}/issue-updated/", data={"issue": {"id": "10425"}}, @@ -161,7 +158,6 @@ def test_get_response_routing_to_cell_async_bucketed(self) -> None: method.return_value = integration response = parser.get_response() - cache.delete(use_buckets_key) assert isinstance(response, HttpResponse) assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( diff --git a/tests/sentry/middleware/integrations/parsers/test_jira_server.py b/tests/sentry/middleware/integrations/parsers/test_jira_server.py index 2e8167728777..796c5eb1a5f9 100644 --- a/tests/sentry/middleware/integrations/parsers/test_jira_server.py +++ b/tests/sentry/middleware/integrations/parsers/test_jira_server.py @@ -2,7 +2,6 @@ from unittest import mock import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory, override_settings from django.urls import reverse @@ -79,7 +78,7 @@ def test_routing_endpoint_with_integration(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"jira_server:{self.integration.id}", + mailbox_name=f"jira_server:{self.integration.id}:1", cell_names=[cell.name], ) @@ -112,7 +111,7 @@ def test_routing_endpoint_with_integration_no_organization_integration(self) -> @override_cells(cell_config) @override_settings(SILO_MODE=SiloMode.CONTROL) @responses.activate - def test_routing_webhook_with_mailbox_buckets_low_volume(self) -> None: + def test_routing_webhook_buckets_on_issue_id(self) -> None: route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) request = self.factory.post( @@ -131,71 +130,6 @@ def test_routing_webhook_with_mailbox_buckets_low_volume(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"jira_server:{self.integration.id}", - cell_names=[cell.name], - ) - - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_buckets_high_volume(self) -> None: - route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) - - request = self.factory.post( - route, data=issue_updated_payload, content_type="application/json" - ) - parser = JiraServerRequestParser(request=request, response_handler=self.get_response) - - with ( - mock.patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited" - ) as mock_is_limited, - mock.patch( - "sentry.middleware.integrations.parsers.jira_server.get_integration_from_token" - ) as mock_get_token, - ): - mock_is_limited.return_value = True - mock_get_token.return_value = self.integration - response = parser.get_response() - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - # Mailbox name should have an extra segment - mailbox_name=f"jira_server:{self.integration.id}:1", - cell_names=[cell.name], - ) - - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_bucket_mode_active(self) -> None: - route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) - - request = self.factory.post( - route, data=issue_updated_payload, content_type="application/json" - ) - parser = JiraServerRequestParser(request=request, response_handler=self.get_response) - - use_bucket_key = f"webhookpayload:jira_server:{self.integration.id}:use_buckets" - cache.set(use_bucket_key, 1) - - with mock.patch( - "sentry.middleware.integrations.parsers.jira_server.get_integration_from_token" - ) as mock_get_token: - mock_get_token.return_value = self.integration - response = parser.get_response() - - cache.delete(use_bucket_key) - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - # Mailbox name should have an extra segment mailbox_name=f"jira_server:{self.integration.id}:1", cell_names=[cell.name], ) diff --git a/tests/sentry/middleware/integrations/parsers/test_vsts.py b/tests/sentry/middleware/integrations/parsers/test_vsts.py index 7b858b86e06b..1a08a78a6543 100644 --- a/tests/sentry/middleware/integrations/parsers/test_vsts.py +++ b/tests/sentry/middleware/integrations/parsers/test_vsts.py @@ -1,7 +1,6 @@ from copy import deepcopy import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory from django.urls import reverse @@ -71,7 +70,7 @@ def test_routing_work_item_webhook(self) -> None: assert response.status_code == 202 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"vsts:{self.integration.id}", + mailbox_name=f"vsts:{self.integration.id}:1", cell_names=["us"], ) @@ -139,16 +138,16 @@ def test_webhook_outbox_creation(self) -> None: parser.get_response() assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"vsts:{self.integration.id}", + mailbox_name=f"vsts:{self.integration.id}:1", cell_names=["us"], ) - def test_webhook_outbox_creation_bucketed(self) -> None: - use_buckets_key = f"webhookpayload:vsts:{self.integration.id}:use_buckets" - cache.set(use_buckets_key, 1) + def test_webhook_outbox_creation_without_a_work_item(self) -> None: + data = deepcopy(WORK_ITEM_UPDATED) + del data["resource"]["workItemId"] request = self.factory.post( self.path, - data=WORK_ITEM_UPDATED, + data=data, content_type="application/json", HTTP_SHARED_SECRET=self.shared_secret, ) @@ -157,11 +156,9 @@ def test_webhook_outbox_creation_bucketed(self) -> None: assert_no_webhook_payloads() parser.get_response() - cache.delete(use_buckets_key) assert_webhook_payloads_for_mailbox( request=request, - # workItemId 31 % 10 - mailbox_name=f"vsts:{self.integration.id}:1", + mailbox_name=f"vsts:{self.integration.id}", cell_names=["us"], ) From d1a4033a46231db6c16bcfe8e792b031a4705fe9 Mon Sep 17 00:00:00 2001 From: Ivan Dlugos Date: Wed, 2 Sep 2026 15:44:58 +0200 Subject: [PATCH 2/3] ref(hybridcloud): Tighten the mailbox bucketing comments The per-parser comments restated the cardinality rule that the `mailbox_bucket_count` docstring already carries, and that docstring now records that a static count is the interim answer and points at the issue replacing it. Refs CW-1887 --- .../middleware/hybrid_cloud/parser.py | 23 ++++++++++--------- .../middleware/integrations/parsers/github.py | 5 ++-- .../middleware/integrations/parsers/gitlab.py | 4 +--- .../middleware/integrations/parsers/jira.py | 1 - .../integrations/parsers/jira_server.py | 5 +--- .../middleware/integrations/parsers/vsts.py | 1 - .../middleware/hybrid_cloud/test_base.py | 3 +-- 7 files changed, 17 insertions(+), 25 deletions(-) diff --git a/src/sentry/integrations/middleware/hybrid_cloud/parser.py b/src/sentry/integrations/middleware/hybrid_cloud/parser.py index 903532faf3ee..14980202fd35 100644 --- a/src/sentry/integrations/middleware/hybrid_cloud/parser.py +++ b/src/sentry/integrations/middleware/hybrid_cloud/parser.py @@ -80,11 +80,13 @@ class BaseRequestParser(ABC): mailbox_bucket_count: ClassVar[int] = 100 """How many sub-mailboxes `mailbox_bucket_id` is spread over. - Choose it from the bucket key's cardinality. A key that repeats across payloads -- - a repository, a project -- already coalesces them onto one mailbox per distinct - value, so a high count costs nothing. A key that barely repeats -- an issue, a - work item -- puts one payload in a bucket and never returns to it, so a high count - buys shallow mailboxes that each still cost a scheduler row and a dispatch slot. + 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. + + 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): @@ -296,8 +298,8 @@ def _bucketed_mailbox_identifier( ) -> str: """The mailbox identifier up to the bucket, before any event-type suffix. - The bucket key is the gate: a payload that carries one is bucketed, and one - that does not falls back to the integration-level mailbox.""" + 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") @@ -326,11 +328,10 @@ def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None: @staticmethod def bucket_key_at(data: Mapping[str, Any], *path: str) -> int | None: - """Read a bucket key out of `data`, or None when it is absent or not numeric. + """Read a bucket key out of `data`, or None if it is missing or not numeric. - Every provider coerces its key through here so they fail the same way: a key - that is missing, nested under a non-object, or not an integer falls back to - the integration-level mailbox instead of raising out of the parser. + 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)) diff --git a/src/sentry/middleware/integrations/parsers/github.py b/src/sentry/middleware/integrations/parsers/github.py index ce9fd6e22e06..e438039e80f5 100644 --- a/src/sentry/middleware/integrations/parsers/github.py +++ b/src/sentry/middleware/integrations/parsers/github.py @@ -72,9 +72,8 @@ def _get_external_id(self, event: Mapping[str, Any]) -> str | None: return get_github_external_id(event) def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None: - """Payloads carry `repository.id` for every event type that reaches a cell; - installation events are handled on control and never get here. - """ + """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: diff --git a/src/sentry/middleware/integrations/parsers/gitlab.py b/src/sentry/middleware/integrations/parsers/gitlab.py index f5b4e01a24df..eeec41d94f43 100644 --- a/src/sentry/middleware/integrations/parsers/gitlab.py +++ b/src/sentry/middleware/integrations/parsers/gitlab.py @@ -83,9 +83,7 @@ def get_response_from_gitlab_webhook(self) -> HttpResponseBase: ) def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None: - """Every event kind a cell processes names the project it belongs to, so the - project is the axis a GitLab mailbox splits on. - """ + """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: diff --git a/src/sentry/middleware/integrations/parsers/jira.py b/src/sentry/middleware/integrations/parsers/jira.py index 4cd486365d8b..dedf2307b4c2 100644 --- a/src/sentry/middleware/integrations/parsers/jira.py +++ b/src/sentry/middleware/integrations/parsers/jira.py @@ -36,7 +36,6 @@ class JiraRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.JIRA.value webhook_identifier = WebhookProviderIdentifier.JIRA - # `issue.id` barely repeats between payloads; see `mailbox_bucket_count`. mailbox_bucket_count = 10 control_classes = [ diff --git a/src/sentry/middleware/integrations/parsers/jira_server.py b/src/sentry/middleware/integrations/parsers/jira_server.py index 87aa925f07bb..8e0b549bd22e 100644 --- a/src/sentry/middleware/integrations/parsers/jira_server.py +++ b/src/sentry/middleware/integrations/parsers/jira_server.py @@ -23,7 +23,6 @@ class JiraServerRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.JIRA_SERVER.value webhook_identifier = WebhookProviderIdentifier.JIRA_SERVER - # `issue.id` barely repeats between payloads; see `mailbox_bucket_count`. mailbox_bucket_count = 10 def get_response_from_issue_update_webhook(self) -> HttpResponseBase: @@ -55,9 +54,7 @@ def get_response_from_issue_update_webhook(self) -> HttpResponseBase: ) def mailbox_bucket_id(self, data: Mapping[str, Any]) -> int | None: - """Only changelog webhooks reach a cell, and each names its issue, so the - issue is the axis a Jira Server mailbox splits on. - """ + """Only changelog webhooks reach a cell, and each names its issue.""" return self.bucket_key_at(data, "issue", "id") def get_response(self) -> HttpResponseBase: diff --git a/src/sentry/middleware/integrations/parsers/vsts.py b/src/sentry/middleware/integrations/parsers/vsts.py index 4e259a9e1d53..00af1c7f225c 100644 --- a/src/sentry/middleware/integrations/parsers/vsts.py +++ b/src/sentry/middleware/integrations/parsers/vsts.py @@ -22,7 +22,6 @@ class VstsRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.AZURE_DEVOPS.value webhook_identifier = WebhookProviderIdentifier.VSTS - # `resource.workItemId` barely repeats between payloads; see `mailbox_bucket_count`. mailbox_bucket_count = 10 cell_view_classes = [WorkItemWebhook] diff --git a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py b/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py index 0e9f23ddd51b..1fac8cbb879e 100644 --- a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py +++ b/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py @@ -160,8 +160,7 @@ def test_bucket_key_at_coerces_or_falls_back(self) -> None: assert at({"issue": {"id": 10237}}, "issue", "id") == 10237 assert at({"issue": {"id": "10237"}}, "issue", "id") == 10237 - # A key that is missing, nested under a non-object, or not a number leaves the - # payload on the integration-level mailbox rather than raising at the modulo. + # 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 From de966a21060c5a2d050d6b2e149363b552d67ec7 Mon Sep 17 00:00:00 2001 From: Ivan Dlugos Date: Wed, 2 Sep 2026 15:46:55 +0200 Subject: [PATCH 3/3] fix(hybridcloud): Use a realistic repository id in the github parser fixture `test_issue_deleted_routing` posted `"repository": {"id": "1"}` as a string. GitHub sends numeric ids -- the cell handler reads `str(event["repository"]["id"])` and every other test in the file uses an int -- so the fixture, not the parser, was wrong. Under the previous isinstance check the string silently produced no bucket, which made the earlier commit look like it changed GitHub's routing. It does not: GitHub already bucketed every payload. Refs CW-1887 --- tests/sentry/middleware/integrations/parsers/test_github.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/sentry/middleware/integrations/parsers/test_github.py b/tests/sentry/middleware/integrations/parsers/test_github.py index fa7099bbc66b..eb23b3731b6a 100644 --- a/tests/sentry/middleware/integrations/parsers/test_github.py +++ b/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},