feat(hybridcloud): Size webhook mailboxes from a rolling rate, not a volume gate

src/sentry/hybridcloud/webhook_mailbox_sizing.py

@@ -0,0 +1,157 @@
+"""
+Sizing an integration's webhook mailbox split from the rate it is currently sending.
+
+Only where the delivery side may reorder: the divisor is the key-to-mailbox map, and
+a provider delivered in order cannot have that map move under it.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import replace
+from time import time
+
+from django.conf import settings
+from redis.exceptions import RedisError
+
+from sentry import options
+from sentry.hybridcloud.mailbox import MailboxName
+from sentry.utils import redis
+
+logger = logging.getLogger(__name__)
+
+WINDOW_SECONDS = 15 * 60
+"""Long enough that a lull between bursts does not read as a quiet integration,
+short enough that a burst raises the count within minutes."""
+
+SHARD_SECONDS = 3 * 60
+"""Width of one counter: how coarsely the window rolls, how far the estimate
+undercounts while the current shard fills, and how many keys one read touches.
+
+Nothing derives three minutes -- it is a fifth of the window. Minute shards would
+read fifteen keys to be accurate within a fifteenth; five-minute shards three keys
+to be accurate within a third.
+"""
+
+SHARD_COUNT = WINDOW_SECONDS // SHARD_SECONDS
+
+SHARD_TTL_SECONDS = WINDOW_SECONDS + SHARD_SECONDS
+"""A real bound despite being reset on every write: a shard takes writes only while
+it is the current one."""
+
+STRICT_BUCKET_COUNT = 10
+"""Width for a provider delivered in order. Every change of divisor re-maps keys,
+which would leave one issue's backlog draining concurrently with its next
+payloads."""
+
+
+def mailbox_bucket_count(mailbox: MailboxName) -> int:
+    """How many sub-mailboxes to spread `mailbox`'s bucket keys over.
+
+    Counts this payload against the window, so call it once per payload queued. A
+    strictly ordered provider is not counted: nothing would read the result.
+    """
+    if not _tolerates_reordering(mailbox.provider):
+        return STRICT_BUCKET_COUNT
+    return _count_for_payloads(_record_and_read_window(_rate_counter_key(mailbox)))
+
+
+def _payloads_per_mailbox() -> int:
+    """Depth a mailbox reaches before its split widens.
+
+    A drain delivers `worker_threads` payloads at once, so the depth is a whole
+    number of those, and raising delivery concurrency narrows the split rather than
+    leaving it where it was.
+
+    Floored at one because both options are automator-modifiable and neither is
+    validated: the drain reads a zero `worker_threads` as one rather than rejecting
+    it, so a zero must not reach the division here either.
+    """
+    return max(
+        1,
+        options.get("hybridcloud.webhookpayload.payloads_per_thread")
+        * options.get("hybridcloud.webhookpayload.worker_threads"),
+    )
+
+
+def _max_buckets() -> int:
+    """Widest split allowed, floored to a power of two.
+
+    The ladder `_count_for_payloads` climbs is doublings, and a cap off that ladder
+    would make a resize into it re-map nearly every key instead of half -- so the set
+    value is floored rather than trusted.
+    """
+    configured = options.get("hybridcloud.webhookpayload.max_mailbox_buckets")
+    return 1 << (configured.bit_length() - 1) if configured >= 1 else 1
+
+
+def _tolerates_reordering(provider: str) -> bool:
+    """Read from the option the drain itself reads, so a provider earns a
+    rate-derived width only once it tolerates the re-mapping that width costs."""
+    return provider in (options.get("hybridcloud.webhookpayload.skip_on_failure_providers") or ())
+
+
+def _rate_counter_key(mailbox: MailboxName) -> str:
+    """The mailbox name without the parts one split varies: the bucket, which the
+    split chooses, and the cell, since one mailbox is built for the whole fanout.
+
+    Event type stays -- it separates mailboxes ahead of bucketing, so dropping it
+    would size one split for every event type at once.
+    """
+    return str(replace(mailbox, cell=None, bucket=None))
+
+
+def _count_for_payloads(payloads: int | None) -> int:
+    """The widest split whose mailboxes would each still fill to the target depth,
+    given the payloads counted over the window.
+
+    Powers of two because `key % 2n` puts a key where `key % n` did or n along, so a
+    doubling moves half the keys. Rounding down rather than to nearest is the
+    hysteresis: the count has to double to add a bucket.
+
+    A window we could not read sizes to the cap -- an outage must not re-serialize
+    the integrations this exists to unserialize.
+    """
+    if payloads is None:
+        return _max_buckets()
+
+    mailboxes = payloads // _payloads_per_mailbox()
+    if mailboxes < 2:
+        return 1
+    return min(1 << (mailboxes.bit_length() - 1), _max_buckets())
+
+
+def _record_and_read_window(counter_key: str) -> int | None:
+    """Count this payload and return the payloads over the window, or None when Redis
+    could not answer.
+
+    A reply that does not destructure or coerce counts as not answering: this runs
+    before the payload row is written, so an exception escaping here would turn a
+    webhook we could still have queued into a 500.
+
+    The current shard is still filling, so the sum runs low by up to one shard and
+    catches up -- the right direction, delaying a widening rather than forcing one.
+    """
+    shard = int(time() // SHARD_SECONDS)
+    current_key = _shard_key(counter_key, shard)
+    older_keys = [_shard_key(counter_key, shard - i) for i in range(1, SHARD_COUNT)]
+
+    try:
+        pipe = redis.redis_clusters.get(settings.SENTRY_RATE_LIMIT_REDIS_CLUSTER).pipeline()
+        pipe.incr(current_key)
+        pipe.expire(current_key, SHARD_TTL_SECONDS)
+        pipe.mget(older_keys)
+        current, _, older = pipe.execute()
+        return int(current) + sum(int(count) for count in older if count is not None)
+    except (RedisError, TypeError, ValueError, IndexError):
+        logger.exception(
+            "hybridcloud.webhook_mailbox_sizing.unavailable",
+            extra={"counter_key": counter_key},
+        )
+        return None
+
+
+def _shard_key(counter_key: str, shard: int) -> str:
+    """Hash-tagged so one counter's shards share a slot and the window reads as a
+    single-node pipeline."""
+    return f"whrate:{{{counter_key}}}:{shard}"

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

@@ -6,13 +6,11 @@
 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.mailbox import MailboxName
 from sentry.hybridcloud.models.webhookpayload import DestinationType, WebhookPayload
@@ -21,6 +19,7 @@
 from sentry.hybridcloud.services.organization_mapping.model import RpcOrganizationMapping
 from sentry.hybridcloud.tasks.deliver_webhooks import maybe_trigger_drain
 from sentry.hybridcloud.webhook_event_types import MAILBOX_EVENT_TYPES
+from sentry.hybridcloud.webhook_mailbox_sizing import mailbox_bucket_count
 from sentry.integrations.middleware.metrics import (
     MiddlewareHaltReason,
     MiddlewareOperationEvent,
@@ -31,7 +30,6 @@
 from sentry.integrations.services.integration.model import RpcIntegration
 from sentry.killswitches import KillswitchConfig, 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_org_mappings, get_cell_by_name
@@ -79,17 +77,6 @@ class BaseRequestParser(ABC):
     webhook_identifier: ClassVar[WebhookProviderIdentifier]
     """The webhook provider identifier"""
 
-    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.
-    """
-
-    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)
@@ -326,64 +313,59 @@ def get_mailbox(
         that can be delivered in parallel. Requires the integration to implement
         `mailbox_bucket_id`
 
-        The cell is left for the fanout to add -- one mailbox is built for all of
-        them.
+        The event type is resolved before the bucket because the split is sized per
+        event type, and only the validated value may reach that: it is read out of a
+        body control has not verified.
         """
-        return self._bucketed(
-            MailboxName(
-                provider=self.provider,
-                subject=str(integration.id),
-                event_type=self._mailbox_event_type(data),
-            ),
-            integration,
-            data,
+        mailbox = MailboxName(
+            provider=self.provider,
+            subject=str(integration.id),
+            event_type=self._mailbox_event_type(data),
         )
-
-    def _bucketed(
-        self,
-        mailbox: MailboxName,
-        integration: RpcIntegration | Integration,
-        data: dict[str, Any],
-    ) -> MailboxName:
-        """`mailbox` in a bucket, or unchanged where the integration is below the
-        volume that warrants buckets, or the payload carries no key to bucket it on."""
-        if not self.always_bucket and not self._exceeds_bucketing_volume(integration):
-            self._record_mailbox_routing(bucketed=False, reason="under_volume_gate")
+        # Callers build the mailbox as an argument, so this runs before the shed check
+        # in get_response_from_webhookpayload. A shed payload is never queued, so it
+        # must not raise the rate that sizes the split -- nor pay the Redis write that
+        # shedding exists to avoid.
+        if self._should_shed(integration.id):
             return mailbox
+        return self._bucketed(mailbox, data)
 
+    def _bucketed(self, mailbox: MailboxName, data: dict[str, Any]) -> MailboxName:
+        """`mailbox` in a bucket, or unchanged where the payload does not get one.
+
+        A keyless payload lands on the unsplit mailbox however wide the split is, so
+        it is left out of the rate that sizes it.
+        """
         mailbox_bucket_id = self.mailbox_bucket_id(data)
         if mailbox_bucket_id is None:
             self._record_mailbox_routing(bucketed=False, reason="no_bucket_key")
             return mailbox
 
-        self._record_mailbox_routing(bucketed=True, reason="bucketed")
+        bucket_count = mailbox_bucket_count(mailbox)
+        if bucket_count == 1:
+            self._record_mailbox_routing(bucketed=False, reason="under_rate", buckets=1)
+            return mailbox
 
-        return mailbox.in_bucket(mailbox_bucket_id % self.mailbox_bucket_count)
+        self._record_mailbox_routing(bucketed=True, reason="bucketed", buckets=bucket_count)
 
-    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"
+        return mailbox.in_bucket(mailbox_bucket_id % bucket_count)
 
-        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, buckets: int | None = None
+    ) -> None:
+        """`reason` is the full breakdown; `bucketed` stays for the dashboards on it.
 
-    def _record_mailbox_routing(self, bucketed: bool, reason: str) -> None:
-        """`reason` is the full breakdown; `bucketed` stays for the dashboards on it."""
-        metrics.incr(
-            "hybridcloud.webhookpayload.mailbox_routing",
-            tags={
-                "provider": self.provider,
-                "bucketed": "true" if bucketed else "false",
-                "reason": reason,
-            },
-        )
+        `buckets` is left off the path that never consults a count, so a query for
+        split width does not average in routing that has no width.
+        """
+        tags = {
+            "provider": self.provider,
+            "bucketed": "true" if bucketed else "false",
+            "reason": reason,
+        }
+        if buckets is not None:
+            tags["buckets"] = str(buckets)
+        metrics.incr("hybridcloud.webhookpayload.mailbox_routing", tags=tags)
 
     def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None:
         raise NotImplementedError(

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

@@ -66,7 +66,6 @@ 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"""

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

@@ -37,10 +37,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 = [
         JiraDescriptorEndpoint,
         JiraSentryInstallationView,

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

@@ -23,10 +23,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]
 
     @control_silo_function

src/sentry/options/defaults.py

@@ -2619,6 +2619,26 @@
     default=4,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+# How many payloads over the rate window one delivery thread should be worth. Times
+# `worker_threads`, this is the depth a mailbox reaches before its split widens, so
+# lowering it splits sooner and wider. Tunable because the right value is not known:
+# the `buckets` tag on `hybridcloud.webhookpayload.mailbox_routing` is what would
+# settle it.
+register(
+    "hybridcloud.webhookpayload.payloads_per_thread",
+    default=4,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+# Most mailboxes one integration's split may occupy; past it they simply grow deeper.
+# A safety valve on how many scheduler rows and dispatch slots one sender can take.
+# Rounded down to a power of two when read: the split climbs a ladder of doublings,
+# and a cap off that ladder makes a resize into it re-map nearly every key instead of
+# half.
+register(
+    "hybridcloud.webhookpayload.max_mailbox_buckets",
+    default=64,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 # Remove the rows a claim-bounded drain finishes with — delivered, attempts
 # exhausted, or stale — in batches instead of one DELETE per row. Such a drain
 # stays inside a claim reserved for its whole run, so deferring deletes cannot

src/sentry/testutils/outbox.py

@@ -2,7 +2,9 @@
 
 import contextlib
 import functools
+from collections.abc import Generator
 from typing import Any
+from unittest import mock
 
 from django.conf import settings
 from django.core.handlers.wsgi import WSGIRequest
@@ -64,6 +66,21 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
             raise OutboxRecursionLimitError
 
 
+@contextlib.contextmanager
+def override_mailbox_bucket_count(count: int) -> Generator[None]:
+    """Pin how wide a parser splits an integration's mailbox.
+
+    The split is sized from the integration's recent webhook rate, so a test that
+    asserts on a bucket number would otherwise have to send enough webhooks to earn
+    one first.
+    """
+    with mock.patch(
+        "sentry.integrations.middleware.hybrid_cloud.parser.mailbox_bucket_count",
+        return_value=count,
+    ):
+        yield
+
+
 def assert_no_webhook_payloads() -> None:
     messages = WebhookPayload.objects.filter().count()
     assert messages == 0, "No webhookpayload messages should be created"