Process Repeaters, Part 1

corehq/ex-submodules/dimagi/utils/couch/tests/test_redis_lock.py

@@ -0,0 +1,31 @@
+import uuid
+
+from redis.lock import Lock as RedisLock
+
+from dimagi.utils.couch import get_redis_lock
+
+from corehq.tests.noseplugins.redislocks import TestLock
+from corehq.util.metrics.lockmeter import MeteredLock
+
+
+def test_get_redis_lock_with_token():
+    lock_name = 'test-1'
+    metered_lock = get_redis_lock(key=lock_name, name=lock_name, timeout=1)
+    assert isinstance(metered_lock, MeteredLock)
+    # metered_lock.lock is a TestLock instance because of
+    # corehq.tests.noseplugins.redislocks.RedisLockTimeoutPlugin
+    test_lock = metered_lock.lock
+    assert isinstance(test_lock, TestLock)
+    redis_lock = test_lock.lock
+    assert isinstance(redis_lock, RedisLock)
+
+    token = uuid.uuid1().hex
+    acquired = redis_lock.acquire(blocking=False, token=token)
+    assert acquired
+
+    # What we want to be able to do in a separate process:
+    metered_lock_2 = get_redis_lock(key=lock_name, name=lock_name, timeout=1)
+    redis_lock_2 = metered_lock_2.lock.lock
+    redis_lock_2.local.token = token
+    # Does not raise LockNotOwnedError:
+    redis_lock_2.release()

corehq/motech/repeaters/const.py

@@ -13,6 +13,8 @@
 CHECK_REPEATERS_INTERVAL = timedelta(minutes=5)
 CHECK_REPEATERS_PARTITION_COUNT = settings.CHECK_REPEATERS_PARTITION_COUNT
 CHECK_REPEATERS_KEY = 'check-repeaters-key'
+PROCESS_REPEATERS_INTERVAL = timedelta(minutes=1)
+PROCESS_REPEATERS_KEY = 'process-repeaters-key'
 ENDPOINT_TIMER = 'endpoint_timer'
 # Number of attempts to an online endpoint before cancelling payload
 MAX_ATTEMPTS = 3

corehq/motech/repeaters/models.py

@@ -73,6 +73,7 @@
 from http import HTTPStatus
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
+from django.conf import settings
 from django.db import models, router
 from django.db.models.base import Deferred
 from django.dispatch import receiver
@@ -237,18 +238,43 @@ def all_ready(self):
         """
         Return all Repeaters ready to be forwarded.
         """
-        not_paused = models.Q(is_paused=False)
-        next_attempt_not_in_the_future = (
-            models.Q(next_attempt_at__isnull=True)
-            | models.Q(next_attempt_at__lte=timezone.now())
+        return self._all_ready_next_attempt_null().union(
+            self._all_ready_next_attempt_now(),
+            all=True,
         )
-        repeat_records_ready_to_send = models.Q(
-            repeat_records__state__in=(State.Pending, State.Fail)
+
+    def get_all_ready_ids_by_domain(self):
+        next_attempt_null = self._all_ready_next_attempt_null().values_list('domain', 'id')
+        next_attempt_now = self._all_ready_next_attempt_now().values_list('domain', 'id')
+        query = next_attempt_null.union(next_attempt_now, all=True)
+        results = defaultdict(list)
+        for (domain, id_uuid) in query.all():
+            results[domain].append(id_uuid.hex)
+        return results
+
+    def all_ready_count(self):
+        return (
+            self._all_ready_next_attempt_null().count()
+            + self._all_ready_next_attempt_now().count()
+        )
+
+    def _all_ready_next_attempt_null(self):
+        # Slower query. Uses deleted_partial_idx
+        return (
+            self.get_queryset()
+            .filter(is_paused=False)
+            .filter(next_attempt_at__isnull=True)
+            .filter(repeat_records__state__in=(State.Pending, State.Fail))
+        )
+
+    def _all_ready_next_attempt_now(self):
+        # Fast query. Uses deleted_paused_partial_idx and state_partial_idx
+        return (
+            self.get_queryset()
+            .filter(is_paused=False)
+            .filter(next_attempt_at__lte=timezone.now())
+            .filter(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))
 
     def get_queryset(self):
         repeater_obj = self.model()
@@ -378,9 +404,24 @@ def _repeater_type(cls):
 
     @property
     def repeat_records_ready(self):
-        return self.repeat_records.filter(state__in=(State.Pending, State.Fail))
+        """
+        A QuerySet of repeat records in the Pending or Fail state in the
+        order in which they were registered
+        """
+        return (
+            self.repeat_records
+            .filter(state__in=(State.Pending, State.Fail))
+            .order_by('registered_at')
+        )
+
+    @property
+    def num_workers(self):
+        # If num_workers is 1, repeat records are sent in the order in
+        # which they were registered.
+        num_workers = self.max_workers or settings.DEFAULT_REPEATER_WORKERS
+        return min(num_workers, settings.MAX_REPEATER_WORKERS)
 
-    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
@@ -393,8 +434,12 @@ 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:
+            # `_get_retry_interval()` implements exponential backoff by
+            # multiplying the previous interval by 3. Set last_attempt_at
+            # to None so that the next time we need to back off, we
+            # know it is the first interval.
             self.last_attempt_at = None
             self.next_attempt_at = None
             # Avoid a possible race condition with self.pause(), etc.
@@ -1004,11 +1049,16 @@ def get_repeat_record_ids(self, domain, repeater_id=None, state=None, payload_id
 class RepeatRecord(models.Model):
     domain = models.CharField(max_length=126)
     payload_id = models.CharField(max_length=255)
-    repeater = models.ForeignKey(Repeater,
-                                 on_delete=DB_CASCADE,
-                                 db_column="repeater_id_",
-                                 related_name='repeat_records')
-    state = models.PositiveSmallIntegerField(choices=State.choices, default=State.Pending)
+    repeater = models.ForeignKey(
+        Repeater,
+        on_delete=DB_CASCADE,
+        db_column="repeater_id_",
+        related_name='repeat_records',
+    )
+    state = models.PositiveSmallIntegerField(
+        choices=State.choices,
+        default=State.Pending,
+    )
     registered_at = models.DateTimeField()
     next_check = models.DateTimeField(null=True, default=None)
     max_possible_tries = models.IntegerField(default=MAX_BACKOFF_ATTEMPTS)
@@ -1196,7 +1246,21 @@ def fire(self, force_send=False, timing_context=None):
                 self.repeater.fire_for_record(self, timing_context=timing_context)
             except Exception as e:
                 self.handle_payload_error(str(e), traceback_str=traceback.format_exc())
-                raise
+                # Repeat records with State.Fail are retried, and repeat
+                # records with State.InvalidPayload are not.
+                #
+                # But a repeat record can have State.InvalidPayload
+                # because it was sent and rejected, so we know that the
+                # remote endpoint is healthy and responding, or because
+                # this exception occurred and it was not sent, so we
+                # don't know anything about the remote endpoint.
+                #
+                # Return None so that `tasks.update_repeater()` treats
+                # the repeat record as unsent, and does not apply or
+                # reset a backoff.
+                return None
+            return self.state
+        return None
 
     def attempt_forward_now(self, *, is_retry=False, fire_synchronously=False):
         from corehq.motech.repeaters.tasks import (
@@ -1206,6 +1270,22 @@ def attempt_forward_now(self, *, is_retry=False, fire_synchronously=False):
             retry_process_datasource_repeat_record,
         )
 
+        def is_new_synchronous_case_repeater_record():
+            """
+            Repeat record is a new record for a synchronous case repeater
+            See corehq.motech.repeaters.signals.fire_synchronous_case_repeaters
+            """
+            # This will also return True if a user clicked "Resend" on a
+            # Pending repeat record in the Repeat Records Report. This
+            # is not intended, but it's also not harmful.
+            return fire_synchronously and self.state == State.Pending
+
+        if (
+            toggles.PROCESS_REPEATERS.enabled(self.domain, toggles.NAMESPACE_DOMAIN)
+            and not is_new_synchronous_case_repeater_record()
+        ):
+            return
+
         if self.next_check is None or self.next_check > datetime.utcnow():
             return
 
@@ -1361,7 +1441,26 @@ def is_response(duck):
 
 
 def domain_can_forward(domain):
+    """
+    Returns whether ``domain`` has data forwarding or Zapier integration
+    privileges.
+
+    Used for determining whether to register a repeat record.
+    """
     return domain and (
         domain_has_privilege(domain, ZAPIER_INTEGRATION)
         or domain_has_privilege(domain, DATA_FORWARDING)
     )
+
+
+def domain_can_forward_now(domain):
+    """
+    Returns ``True`` if ``domain`` has the requisite privileges and data
+    forwarding is not paused.
+
+    Used for determining whether to send a repeat record now.
+    """
+    return (
+        domain_can_forward(domain)
+        and not toggles.PAUSE_DATA_FORWARDING.enabled(domain)
+    )