Merge pull request #34981 from dimagi/gh/rate-limiting/repeater-attempts

Rate limit repeaters based on number of attempts (Take 2)

corehq/motech/rate_limiter.py

@@ -6,7 +6,7 @@
     PerUserRateDefinition,
     RateLimiter,
 )
-from corehq.toggles import RATE_LIMIT_REPEATERS, NAMESPACE_DOMAIN
+from corehq.toggles import RATE_LIMIT_REPEATERS, RATE_LIMIT_REPEATER_ATTEMPTS, NAMESPACE_DOMAIN
 from corehq.util.decorators import silence_and_report_error, run_only_when
 from corehq.util.metrics import metrics_gauge, metrics_counter
 from corehq.util.quickcache import quickcache
@@ -55,14 +55,33 @@ def _get_per_user_repeater_wait_milliseconds_rate_definition(domain):
 )
 
 
+repeater_attempts_rate_limiter = RateLimiter(
+    feature_key='repeater_attempts',
+    get_rate_limits=lambda scope: get_dynamic_rate_definition(
+        "repeater_attempts",
+        default=RateDefinition(
+            per_week=None,
+            per_day=None,
+            per_hour=360000,
+            per_minute=6000,
+            per_second=100,
+        ),
+    ).get_rate_limits(),
+)
+
+
 SHOULD_RATE_LIMIT_REPEATERS = not settings.UNIT_TESTING
 
 
-@run_only_when(lambda: SHOULD_RATE_LIMIT_REPEATERS)
+@run_only_when(SHOULD_RATE_LIMIT_REPEATERS)
 @silence_and_report_error("Exception raised in the repeater rate limiter",
                           'commcare.repeaters.rate_limiter_errors')
 def rate_limit_repeater(domain):
-    if global_repeater_rate_limiter.allow_usage() or repeater_rate_limiter.allow_usage(domain):
+    is_domain_allowed_usage = repeater_rate_limiter.allow_usage(domain)
+    if RATE_LIMIT_REPEATER_ATTEMPTS.enabled(domain, namespace=NAMESPACE_DOMAIN):
+        is_domain_allowed_usage = is_domain_allowed_usage and repeater_attempts_rate_limiter.allow_usage(domain)
+
+    if global_repeater_rate_limiter.allow_usage() or is_domain_allowed_usage:
         allow_usage = True
     elif not RATE_LIMIT_REPEATERS.enabled(domain, namespace=NAMESPACE_DOMAIN):
         allow_usage = True
@@ -87,6 +106,13 @@ def report_repeater_usage(domain, milliseconds):
     _report_current_global_repeater_thresholds()
 
 
+@run_only_when(SHOULD_RATE_LIMIT_REPEATERS)
+@silence_and_report_error("Exception raised reporting usage to the repeater attempt rate limiter",
+                          'commcare.repeaters.report_usage_errors')
+def report_repeater_attempt(domain):
+    repeater_attempts_rate_limiter.report_usage(domain)
+
+
 @quickcache([], timeout=60)  # Only report up to once a minute
 def _report_current_global_repeater_thresholds():
     for scope, limits in global_repeater_rate_limiter.iter_rates():

corehq/motech/repeaters/tasks.py

@@ -33,7 +33,11 @@
 )
 from .models import RepeatRecord, domain_can_forward
 
-from ..rate_limiter import report_repeater_usage, rate_limit_repeater
+from ..rate_limiter import (
+    rate_limit_repeater,
+    report_repeater_attempt,
+    report_repeater_usage,
+)
 
 _check_repeaters_buckets = make_buckets_from_timedeltas(
     timedelta(seconds=10),
@@ -176,6 +180,7 @@ def _process_repeat_record(repeat_record):
                 repeat_record.postpone_by(random.uniform(*RATE_LIMITER_DELAY_RANGE))
                 action = 'rate_limited'
             elif repeat_record.is_queued():
+                report_repeater_attempt(repeat_record.domain)
                 with timer('fire_timing') as fire_timer:
                     repeat_record.fire(timing_context=fire_timer)
                 # round up to the nearest millisecond, meaning always at least 1ms

corehq/toggles/__init__.py

@@ -1982,6 +1982,17 @@ def _commtrackify(domain_name, toggle_is_enabled):
     """
 )
 
+RATE_LIMIT_REPEATER_ATTEMPTS = DynamicallyPredictablyRandomToggle(
+    'rate_limit_repeater_attempts',
+    'Apply rate limiting to attempts for data forwarding (repeaters)',
+    TAG_INTERNAL,
+    [NAMESPACE_DOMAIN],
+    description="""
+    In addition to the rate limits based on time spent waiting, these rate limits ensure a project
+    is limited to how many records they can attempt to forward in a given time window.
+    """
+)
+
 TEST_FORM_SUBMISSION_RATE_LIMIT_RESPONSE = StaticToggle(
     'test_form_submission_rate_limit_response',
     "Respond to all form submissions with a 429 response",