check_repeaters() task iterates repeaters

corehq/apps/app_manager/tests/test_repeater.py

@@ -3,8 +3,6 @@
 from django.test import TestCase
 from django.test.client import Client
 
-from unittest.mock import patch
-
 from corehq.apps.accounting.models import SoftwarePlanEdition
 from corehq.apps.accounting.tests.utils import DomainSubscriptionMixin
 from corehq.apps.accounting.utils import clear_plan_version_cache
@@ -64,18 +62,3 @@ def test_repeat_record_created(self):
         later = datetime.utcnow() + timedelta(hours=48 + 1)
         repeat_records = RepeatRecord.objects.filter(domain=self.domain, next_check__lt=later)
         self.assertEqual(len(repeat_records), 1)
-
-    def test_repeat_record_forwarded(self):
-        """
-        When an application with a repeater is saved, then HQ should try to forward the repeat record
-        """
-        self.app_structure_repeater = AppStructureRepeater(domain=self.domain, connection_settings=self.conn)
-        self.app_structure_repeater.save()
-        self.addCleanup(self.app_structure_repeater.delete)
-
-        with patch('corehq.motech.repeaters.models.simple_request') as mock_fire:
-            self.application = Application(domain=self.domain)
-            self.application.save()
-            self.addCleanup(self.application.delete)
-
-            self.assertEqual(mock_fire.call_count, 1)

corehq/motech/repeaters/const.py

@@ -8,12 +8,18 @@
 MIN_RETRY_WAIT = timedelta(minutes=60)
 RATE_LIMITER_DELAY_RANGE = (timedelta(minutes=0), timedelta(minutes=15))
 CHECK_REPEATERS_INTERVAL = timedelta(minutes=5)
-CHECK_REPEATERS_PARTITION_COUNT = settings.CHECK_REPEATERS_PARTITION_COUNT
+CHECK_REPEATERS_PARTITION_COUNT = settings.CHECK_REPEATERS_PARTITION_COUNT  # TODO: Drop
 CHECK_REPEATERS_KEY = 'check-repeaters-key'
 # Number of attempts to an online endpoint before cancelling payload
 MAX_ATTEMPTS = 3
 # Number of exponential backoff attempts to an offline endpoint
 MAX_BACKOFF_ATTEMPTS = 6
+# The maximum number of workers that one repeater can use to send repeat
+# records at the same time. (In other words, HQ's capacity to DDOS
+# attack a remote API endpoint.) This is a guardrail to prevent one
+# domain from hogging repeat record queue workers and to ensure that
+# repeaters are iterated fairly.
+MAX_REPEATER_WORKERS = 7
 
 
 class State(IntegerChoices):

corehq/motech/repeaters/management/commands/find_missing_repeat_records.py

@@ -527,7 +527,8 @@ def create_case_repeater_register(repeater, domain, payload):
         'domain': domain,
         'doc_type': repeater.repeater_type
     })
-    repeat_record.attempt_forward_now()
+    # TODO: No, the repeater sends the repeat record.
+    # repeat_record.attempt_forward_now()
     return repeat_record
 
 

corehq/motech/repeaters/models.py

@@ -65,10 +65,12 @@
 """
 import inspect
 import json
+import random
 import traceback
 import uuid
 from collections import defaultdict
 from datetime import datetime, timedelta
+from http import HTTPStatus
 from typing import Any
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
@@ -92,6 +94,7 @@
 
 from corehq import toggles
 from corehq.apps.accounting.utils import domain_has_privilege
+from corehq.apps.domain.models import Domain
 from corehq.apps.locations.models import SQLLocation
 from corehq.apps.users.models import CommCareUser
 from corehq.form_processor.exceptions import XFormNotFound
@@ -122,8 +125,10 @@
 from .const import (
     MAX_ATTEMPTS,
     MAX_BACKOFF_ATTEMPTS,
+    MAX_REPEATER_WORKERS,
     MAX_RETRY_WAIT,
     MIN_RETRY_WAIT,
+    RATE_LIMITER_DELAY_RANGE,
     State,
 )
 from .exceptions import RequestConnectionError, UnknownRepeater
@@ -220,6 +225,7 @@ def all_ready(self):
         """
         Return all Repeaters ready to be forwarded.
         """
+        domains = get_domains_forwarding_enabled()
         not_paused = models.Q(is_paused=False)
         next_attempt_not_in_the_future = (
             models.Q(next_attempt_at__isnull=True)
@@ -228,10 +234,13 @@ def all_ready(self):
         repeat_records_ready_to_send = models.Q(
             repeat_records__state__in=(State.Pending, State.Fail)
         )
-        return (self.get_queryset()
-                .filter(not_paused)
-                .filter(next_attempt_not_in_the_future)
-                .filter(repeat_records_ready_to_send))
+        return (
+            self.get_queryset()
+            .filter(domain__in=domains)
+            .filter(not_paused)
+            .filter(next_attempt_not_in_the_future)
+            .filter(repeat_records_ready_to_send)
+        )
 
     def get_queryset(self):
         repeater_obj = self.model()
@@ -258,6 +267,7 @@ class Repeater(RepeaterSuperProxy):
     is_paused = models.BooleanField(default=False)
     next_attempt_at = models.DateTimeField(null=True, blank=True)
     last_attempt_at = models.DateTimeField(null=True, blank=True)
+    # TODO: max_workers = models.IntegerField(default=1)
     options = JSONField(default=dict)
     connection_settings_id = models.IntegerField(db_index=True)
     is_deleted = models.BooleanField(default=False, db_index=True)
@@ -348,21 +358,19 @@ def _repeater_type(cls):
 
     @property
     def repeat_records_ready(self):
-        return self.repeat_records.filter(state__in=(State.Pending, State.Fail))
+        return (
+            self.repeat_records
+            .filter(state__in=(State.Pending, State.Fail))
+            .order_by('registered_at')
+        )
 
-    @property
-    def is_ready(self):
-        """
-        Returns True if there are repeat records to be sent.
-        """
-        if self.is_paused or toggles.PAUSE_DATA_FORWARDING.enabled(self.domain):
-            return False
-        if not (self.next_attempt_at is None
-                or self.next_attempt_at < timezone.now()):
-            return False
-        return self.repeat_records_ready.exists()
+    def rate_limit(self):
+        interval = random.uniform(*RATE_LIMITER_DELAY_RANGE)
+        Repeater.objects.filter(id=self.repeater_id).update(
+            next_attempt_at=datetime.utcnow() + interval,
+        )
 
-    def set_next_attempt(self):
+    def set_backoff(self):
         now = datetime.utcnow()
         interval = _get_retry_interval(self.last_attempt_at, now)
         self.last_attempt_at = now
@@ -375,7 +383,7 @@ def set_next_attempt(self):
             next_attempt_at=now + interval,
         )
 
-    def reset_next_attempt(self):
+    def reset_backoff(self):
         if self.last_attempt_at or self.next_attempt_at:
             self.last_attempt_at = None
             self.next_attempt_at = None
@@ -411,7 +419,8 @@ def register(self, payload, fire_synchronously=False):
             # Prime the cache to prevent unnecessary lookup. Only do this for synchronous repeaters
             # to prevent serializing the repeater in the celery task payload
             repeat_record.__dict__["repeater"] = self
-        repeat_record.attempt_forward_now(fire_synchronously=fire_synchronously)
+        # TODO: No, send the repeat record when it's its turn.
+        # repeat_record.attempt_forward_now(fire_synchronously=fire_synchronously)
         return repeat_record
 
     def allowed_to_forward(self, payload):
@@ -432,6 +441,11 @@ def retire(self):
         self.is_deleted = True
         Repeater.objects.filter(id=self.repeater_id).update(is_deleted=True)
 
+    @property
+    def num_workers(self):
+        # TODO: return min(self.max_workers, MAX_REPEATER_WORKERS)
+        return MAX_REPEATER_WORKERS
+
     def fire_for_record(self, repeat_record):
         payload = self.get_payload(repeat_record)
         try:
@@ -473,10 +487,36 @@ def handle_response(self, result, repeat_record):
 
         result may be either a response object or an exception
         """
+        never_gonna_work = (
+            HTTPStatus.BAD_REQUEST,
+            HTTPStatus.UNAUTHORIZED,
+            HTTPStatus.PAYMENT_REQUIRED,
+            HTTPStatus.FORBIDDEN,
+            HTTPStatus.NOT_FOUND,
+            HTTPStatus.METHOD_NOT_ALLOWED,
+            HTTPStatus.NOT_ACCEPTABLE,
+            HTTPStatus.PROXY_AUTHENTICATION_REQUIRED,
+            HTTPStatus.GONE,
+            HTTPStatus.LENGTH_REQUIRED,
+            HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
+            HTTPStatus.REQUEST_URI_TOO_LONG,
+            HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
+            HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE,
+            HTTPStatus.EXPECTATION_FAILED,
+            HTTPStatus.IM_A_TEAPOT,  # For completeness :)
+            HTTPStatus.MISDIRECTED_REQUEST,
+            HTTPStatus.UNPROCESSABLE_ENTITY,
+            HTTPStatus.REQUEST_HEADER_FIELDS_TOO_LARGE,
+            HTTPStatus.UNAVAILABLE_FOR_LEGAL_REASONS,
+        )
+
         if isinstance(result, Exception):
             repeat_record.handle_exception(result)
         elif is_response(result) and 200 <= result.status_code < 300 or result is True:
             repeat_record.handle_success(result)
+        elif is_response(result) and result.status_code in never_gonna_work:
+            message = format_response(result)
+            repeat_record.handle_payload_error(message)
         else:
             repeat_record.handle_failure(result)
 
@@ -916,10 +956,21 @@ def count_by_repeater_and_state(self, domain):
         return result
 
     def count_overdue(self, threshold=timedelta(minutes=10)):
-        return self.filter(
-            next_check__isnull=False,
-            next_check__lt=datetime.utcnow() - threshold
-        ).count()
+        overdue = datetime.utcnow() - threshold
+        domains = get_domains_forwarding_enabled()
+        repeater_not_paused = models.Q(repeater__is_paused=False)
+        repeater_next_attempt_overdue = models.Q(repeater__next_attempt_at__lt=overdue)
+        ready_to_send = models.Q(
+            state__in=(State.Pending, State.Fail)
+        )
+        return (
+            self.get_queryset()
+            .filter(domain__in=domains)
+            .filter(repeater_not_paused)
+            .filter(repeater_next_attempt_overdue)
+            .filter(ready_to_send)
+            .count()
+        )
 
     def iterate(self, domain, repeater_id=None, state=None, chunk_size=1000):
         db = router.db_for_read(self.model)
@@ -1154,9 +1205,12 @@ def fire(self, force_send=False):
             except Exception as e:
                 log_repeater_error_in_datadog(self.domain, status_code=None,
                                               repeater_type=self.repeater_type)
-                self.handle_payload_exception(e)
-                raise
+                self.handle_payload_error(str(e))
+            finally:
+                return self.state
+        return None
 
+    # TODO: Drop: `process_repeater` task will call `process_repeat_record` tasks directly
     def attempt_forward_now(self, *, is_retry=False, fire_synchronously=False):
         from corehq.motech.repeaters.tasks import (
             process_repeat_record,
@@ -1206,8 +1260,8 @@ def handle_failure(self, response):
     def handle_exception(self, exception):
         self.add_client_failure_attempt(str(exception))
 
-    def handle_payload_exception(self, exception):
-        self.add_client_failure_attempt(str(exception), retry=False)
+    def handle_payload_error(self, message):
+        self.add_client_failure_attempt(message, retry=False)
 
     def cancel(self):
         self.state = State.Cancelled
@@ -1370,7 +1424,29 @@ def is_response(duck):
 
 
 def domain_can_forward(domain):
+    """
+    Checks whether ``domain`` has the privilege to forward data. Ignores
+    the status of the (temporary) ``PAUSE_DATA_FORWARDING`` toggle.
+
+    Used for registering repeat records.
+    """
     return domain and (
         domain_has_privilege(domain, ZAPIER_INTEGRATION)
         or domain_has_privilege(domain, DATA_FORWARDING)
     )
+
+
+def get_domains_forwarding_enabled():
+    """
+    Returns a set of domains that are *currently* able to forward data.
+    Considers the status of the (temporary) ``PAUSE_DATA_FORWARDING``
+    toggle.
+
+    Used for iterating repeaters and counting overdue repeat records.
+    """
+    domains_can_forward = {
+        domain for domain in Domain.get_all_names()
+        if domain_can_forward(domain)
+    }
+    domains_paused = set(toggles.PAUSE_DATA_FORWARDING.get_enabled_domains())
+    return domains_can_forward - domains_paused